Build a Telegram Bot That Sends Buy/Sell Signals in Python

Telegram is the notification layer that most retail traders already use. Building a bot that pushes BUY/SELL/HOLD alerts directly to your phone — with reasoning, confidence scores, and price targets — takes about 150 lines of Python. This guide covers everything from creating the bot to running scheduled analysis with the TA-Lib engine covered in our previous post.

1. Create Your Bot via BotFather

1. Open Telegram and search for @BotFather
2. Send /newbot and follow the prompts
3. Copy the API token — it looks like 123456789:ABCdefGHIjklMNOpqrstUVWxyz
4. Send /setprivacy → disable (so the bot can read group messages)

To get your personal chat ID, send a message to your bot and then visit:
https://api.telegram.org/bot<YOUR_TOKEN>/getUpdates
Find "chat":{"id":123456789} — that's your chat ID.

2. Install Dependencies

pip install python-telegram-bot==20.7 yfinance TA-Lib APScheduler python-dotenv

Store credentials in a .env file — never hardcode tokens:

TELEGRAM_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrstUVWxyz
TELEGRAM_CHAT_ID=123456789

3. The Signal Engine

This builds directly on the TA-Lib screener from the previous guide. Here's the core signal function:

# signal_engine.py
import yfinance as yf
import numpy as np
import talib
from dataclasses import dataclass

@dataclass
class Signal:
    symbol: str
    price: float
    signal: str          # BUY / SELL / HOLD
    confidence: int      # 0-100
    rsi: float
    above_sma200: bool
    macd_cross: str
    adx: float
    atr: float
    stop_loss: float
    take_profit: float
    rationale: str

def generate_signal(symbol: str) -> Signal | None:
    try:
        df = yf.Ticker(symbol).history(period="1y")
        if len(df) < 200:
            return None

        c = df["Close"].values.astype(np.float64)
        h = df["High"].values.astype(np.float64)
        l = df["Low"].values.astype(np.float64)

        rsi       = talib.RSI(c, 14)[-1]
        sma50     = talib.SMA(c, 50)[-1]
        sma200    = talib.SMA(c, 200)[-1]
        adx       = talib.ADX(h, l, c, 14)[-1]
        atr       = talib.ATR(h, l, c, 14)[-1]
        upper, _, lower = talib.BBANDS(c, 20)
        macd, sig, _    = talib.MACD(c, 12, 26, 9)

        score = 0
        notes = []

        # RSI
        if rsi < 35:
            score += 2; notes.append(f"RSI oversold at {rsi:.0f}")
        elif rsi < 50:
            score += 1; notes.append(f"RSI neutral-bullish at {rsi:.0f}")
        elif rsi > 70:
            score -= 2; notes.append(f"RSI overbought at {rsi:.0f}")

        # Trend
        above_200 = c[-1] > sma200
        if c[-1] > sma50 > sma200:
            score += 2; notes.append("Price above 50 and 200 SMA — strong uptrend")
        elif c[-1] < sma50 < sma200:
            score -= 2; notes.append("Price below 50 and 200 SMA — downtrend")

        # MACD crossover
        macd_cross = "none"
        if macd[-1] > sig[-1] and macd[-2] <= sig[-2]:
            score += 2; macd_cross = "bullish"; notes.append("Fresh MACD bullish crossover")
        elif macd[-1] < sig[-1] and macd[-2] >= sig[-2]:
            score -= 2; macd_cross = "bearish"; notes.append("Fresh MACD bearish crossover")

        # Bollinger Band squeeze breakout
        if c[-1] > upper[-1]:
            score += 1; notes.append("Breaking above upper Bollinger Band")
        elif c[-1] < lower[-1]:
            score -= 1; notes.append("Breaking below lower Bollinger Band")

        # ADX trend strength
        if adx > 30:
            notes.append(f"Trend is strong (ADX {adx:.0f})")

        # Map score to signal
        if score >= 3:
            signal = "BUY"
        elif score <= -3:
            signal = "SELL"
        else:
            signal = "HOLD"

        confidence = min(100, abs(score) * 15 + 30)
        stop_loss   = round(c[-1] - 1.5 * atr, 2)
        take_profit = round(c[-1] + 2.5 * atr, 2)
        rationale   = ". ".join(notes) + "."

        return Signal(
            symbol=symbol, price=round(c[-1], 2), signal=signal,
            confidence=confidence, rsi=round(rsi, 1), above_sma200=above_200,
            macd_cross=macd_cross, adx=round(adx, 1), atr=round(atr, 2),
            stop_loss=stop_loss, take_profit=take_profit, rationale=rationale,
        )
    except Exception as e:
        print(f"Signal error for {symbol}: {e}")
        return None

4. Format the Telegram Message

Telegram supports HTML and Markdown formatting. Use HTML for clean, readable alerts:

# formatter.py
def format_signal_message(s, include_details=True):
    icons = {"BUY": "🟢", "SELL": "🔴", "HOLD": "🟡"}
    icon = icons.get(s.signal, "⚪")

    conf_bar = "█" * (s.confidence // 10) + "░" * (10 - s.confidence // 10)

    msg = (
        f"{icon} {s.symbol} — {s.signal}\n"
        f"💰 Price: ${s.price:,.2f}\n"
        f"📊 Confidence: {conf_bar} {s.confidence}%\n"
    )

    if include_details:
        msg += (
            f"\n📉 Stop loss:   ${s.stop_loss:,.2f}\n"
            f"📈 Take profit: ${s.take_profit:,.2f}\n"
            f"\nRationale:\n{s.rationale}\n"
            f"\nRSI {s.rsi} · ADX {s.adx} · "
            f"{'Above' if s.above_sma200 else 'Below'} 200 SMA"
        )

    return msg

5. The Bot

# bot.py
import os
import asyncio
from dotenv import load_dotenv
from telegram import Update
from telegram.ext import (
    Application, CommandHandler, ContextTypes
)
from signal_engine import generate_signal
from formatter import format_signal_message

load_dotenv()
TOKEN   = os.environ["TELEGRAM_BOT_TOKEN"]
CHAT_ID = int(os.environ["TELEGRAM_CHAT_ID"])

WATCHLIST = ["NVDA", "AAPL", "TSLA", "MSFT", "BTC-USD", "ETH-USD"]

# /start
async def start(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_html(
        "👋 Signal Bot active.\n\n"
        "Commands:\n"
        "/signal NVDA — get signal for a symbol\n"
        "/watchlist — scan all tracked symbols\n"
        "/buys — show only BUY signals\n"
        "/help — this message"
    )

# /signal NVDA
async def cmd_signal(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    if not ctx.args:
        await update.message.reply_text("Usage: /signal SYMBOL (e.g. /signal NVDA)")
        return
    symbol = ctx.args[0].upper()
    await update.message.reply_text(f"Analysing {symbol}…")
    s = generate_signal(symbol)
    if s:
        await update.message.reply_html(format_signal_message(s))
    else:
        await update.message.reply_text(f"Could not fetch data for {symbol}.")

# /watchlist — scan all tracked symbols
async def cmd_watchlist(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text(f"Scanning {len(WATCHLIST)} symbols…")
    results = []
    for sym in WATCHLIST:
        s = generate_signal(sym)
        if s:
            results.append(s)

    if not results:
        await update.message.reply_text("No results.")
        return

    results.sort(key=lambda x: (x.signal != "BUY", x.signal != "HOLD", -x.confidence))
    summary = "\n".join(
        f"{'🟢' if r.signal=='BUY' else '🔴' if r.signal=='SELL' else '🟡'} "
        f"{r.symbol} {r.signal} {r.confidence}% — ${r.price:,.2f}"
        for r in results
    )
    await update.message.reply_html(f"Watchlist scan:\n\n{summary}")

# /buys — only BUY signals
async def cmd_buys(update: Update, ctx: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text("Finding BUY signals…")
    buys = [s for sym in WATCHLIST if (s := generate_signal(sym)) and s.signal == "BUY"]
    if not buys:
        await update.message.reply_text("No BUY signals right now.")
        return
    for s in sorted(buys, key=lambda x: -x.confidence):
        await update.message.reply_html(format_signal_message(s))

def main():
    app = Application.builder().token(TOKEN).build()
    app.add_handler(CommandHandler("start",     start))
    app.add_handler(CommandHandler("help",      start))
    app.add_handler(CommandHandler("signal",    cmd_signal))
    app.add_handler(CommandHandler("watchlist", cmd_watchlist))
    app.add_handler(CommandHandler("buys",      cmd_buys))

    print("Bot running…")
    app.run_polling()

if __name__ == "__main__":
    main()

6. Scheduled Daily Alerts

Use APScheduler to push a daily digest automatically at market close (4 PM ET):

# Add to bot.py, inside main(), before app.run_polling()
from apscheduler.schedulers.asyncio import AsyncIOScheduler
import pytz

async def daily_digest(app):
    # Push a full watchlist scan to your chat every day after market close.
    results = [s for sym in WATCHLIST if (s := generate_signal(sym))]
    if not results:
        return

    buys  = [r for r in results if r.signal == "BUY"]
    sells = [r for r in results if r.signal == "SELL"]
    holds = [r for r in results if r.signal == "HOLD"]

    lines = ["📊 Daily Signal Digest\n"]
    for label, group in [("🟢 BUY", buys), ("🔴 SELL", sells), ("🟡 HOLD", holds)]:
        if group:
            lines.append(f"{label}")
            for r in sorted(group, key=lambda x: -x.confidence):
                lines.append(f"  {r.symbol} {r.confidence}% — ${r.price:,.2f}")
            lines.append("")

    await app.bot.send_message(chat_id=CHAT_ID, text="\n".join(lines), parse_mode="HTML")

scheduler = AsyncIOScheduler(timezone=pytz.timezone("America/New_York"))
scheduler.add_job(
    lambda: asyncio.create_task(daily_digest(app)),
    trigger="cron",
    hour=16, minute=5,   # 4:05 PM ET — just after US market close
    day_of_week="mon-fri"
)
scheduler.start()

7. Price Alerts

Trigger alerts only when a signal changes, to avoid spamming yourself with repeated HOLDs:

# alert_tracker.py
import json, os

CACHE_FILE = "signal_cache.json"

def load_cache():
    if os.path.exists(CACHE_FILE):
        with open(CACHE_FILE) as f:
            return json.load(f)
    return {}

def save_cache(cache):
    with open(CACHE_FILE, "w") as f:
        json.dump(cache, f)

def has_signal_changed(symbol: str, new_signal: str) -> bool:
    cache = load_cache()
    old = cache.get(symbol)
    if old != new_signal:
        cache[symbol] = new_signal
        save_cache(cache)
        return True
    return False

# In your scheduler job:
async def check_for_changes(app):
    for sym in WATCHLIST:
        s = generate_signal(sym)
        if s and has_signal_changed(sym, s.signal):
            msg = f"🔔 Signal change: {sym}\n" + format_signal_message(s)
            await app.bot.send_message(chat_id=CHAT_ID, text=msg, parse_mode="HTML")

8. Production Deployment

The bot needs to run continuously. Options:

Raspberry Pi (zero cost after hardware)

# /etc/systemd/system/signal-bot.service
[Unit]
Description=Trading Signal Bot
After=network.target

[Service]
User=pi
WorkingDirectory=/home/pi/signal-bot
ExecStart=/home/pi/signal-bot/venv/bin/python bot.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

sudo systemctl enable signal-bot
sudo systemctl start signal-bot

Cloud (cheapest reliable option)

A £3/month VPS (Oracle Free Tier is genuinely free) or a Railway/Render free plan runs the bot persistently. Docker makes it portable:

FROM python:3.11-slim
RUN apt-get update && apt-get install -y build-essential wget \
    && wget http://prdownloads.sourceforge.net/ta-lib/ta-lib-0.4.0-src.tar.gz \
    && tar -xzf ta-lib-0.4.0-src.tar.gz && cd ta-lib \
    && ./configure --prefix=/usr && make && make install
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["python", "bot.py"]

Going Further: Add AI Analysis

The TA-Lib signals above are a solid foundation, but the reasoning can feel mechanical. The next step is to pass the TA-Lib output to an AI model (Gemini, DeepSeek, or GPT-4) to generate a human-readable rationale that weighs the technicals against current news and macro context — which is exactly what Indikators does under the hood. The architecture is:

1. TA-Lib computes all indicator values → structured dict
2. Fundamentals fetched from FMP API → added to the dict
3. Dict formatted into a prompt and sent to the AI model
4. AI returns: signal, confidence, rationale, key risks, price targets
5. Bot formats and delivers to Telegram

The advantage over pure TA-Lib rules is that the AI can say "RSI is oversold and MACD just crossed bullish, but there's an earnings report in 3 days and the sector is under pressure from rising rates — HOLD rather than BUY." That contextual reasoning is hard to code with rules; it's natural for a language model.