How to Build a Telegram Crypto Alert & Trading Bot (Python Tutorial)

Why Telegram for Crypto Bots?

Telegram is the communication platform of crypto. It is:

• Available on every device
• Fast and reliable
• Free to build bots on (no hosting fees for the bot API)
• Already where your community is

A Telegram bot lets you receive real-time alerts, query your bot's status, and even execute trades from your phone — from anywhere in the world.

What We Are Building

By the end of this tutorial you will have a bot that:

1. Sends price alerts when BTC moves ±5%
2. Reports your portfolio balance on command
3. Sends trading signals from your bot
4. Can place trades directly from Telegram

Step 1: Create a Telegram Bot

1. Open Telegram and search for @BotFather
2. Send /newbot
3. Choose a name: My Crypto Bot
4. Choose a username: mycryptotrading_bot (must end in bot)
5. Save the token — looks like 7123456789:AAF...

Step 2: Get Your Chat ID

Start a conversation with your new bot. Then run:

curl https://api.telegram.org/bot<YOUR_TOKEN>/getUpdates

Find "chat": {"id": 123456789} — that is your chat ID.

Step 3: Install Dependencies

pip install python-telegram-bot ccxt python-dotenv aiohttp

# .env
TELEGRAM_TOKEN=7123456789:AAF...
TELEGRAM_CHAT_ID=123456789
BINANCE_API_KEY=your_key
BINANCE_SECRET=your_secret

Step 4: Build the Bot

# crypto_telegram_bot.py
import os
import asyncio
import logging
from dotenv import load_dotenv
from telegram import Update, Bot
from telegram.ext import Application, CommandHandler, ContextTypes
import ccxt

load_dotenv()

TOKEN = os.getenv('TELEGRAM_TOKEN')
CHAT_ID = int(os.getenv('TELEGRAM_CHAT_ID'))

# Exchange setup
exchange = ccxt.binance({
    'apiKey': os.getenv('BINANCE_API_KEY'),
    'secret': os.getenv('BINANCE_SECRET'),
})

# --- COMMAND HANDLERS ---

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Welcome message."""
    await update.message.reply_text(
        "🤖 Crypto Bot Online!\n\n"
        "Commands:\n"
        "/price BTC — Get current price\n"
        "/balance — Show portfolio\n"
        "/alert BTC 5 — Alert on 5% move\n"
        "/buy BTC 100 — Buy $100 of BTC\n"
        "/sell BTC 50 — Sell $50 of BTC\n"
        "/stop — Stop active alerts"
    )

async def get_price(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Get current price for a coin."""
    if not context.args:
        await update.message.reply_text("Usage: /price BTC")
        return
    
    coin = context.args[0].upper()
    symbol = f"{coin}/USDT"
    
    try:
        ticker = exchange.fetch_ticker(symbol)
        price = ticker['last']
        change = ticker['percentage'] or 0
        emoji = "🟢" if change >= 0 else "🔴"
        
        await update.message.reply_text(
            f"{emoji} {coin}/USDT\n"
            f"Price: ${price:,.2f}\n"
            f"24h Change: {change:+.2f}%\n"
            f"24h High: ${ticker['high']:,.2f}\n"
            f"24h Low: ${ticker['low']:,.2f}"
        )
    except Exception as e:
        await update.message.reply_text(f"Error: {e}")

async def get_balance(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Show portfolio balance."""
    try:
        balance = exchange.fetch_balance()
        
        lines = ["💼 Portfolio Balance\n"]
        total_usd = 0
        
        for coin, amounts in balance['total'].items():
            if amounts > 0.0001 and coin != 'USDT':
                try:
                    ticker = exchange.fetch_ticker(f"{coin}/USDT")
                    usd_value = amounts * ticker['last']
                    total_usd += usd_value
                    lines.append(f"{coin}: {amounts:.4f} (${usd_value:,.2f})")
                except:
                    lines.append(f"{coin}: {amounts:.4f}")
        
        usdt = balance['total'].get('USDT', 0)
        total_usd += usdt
        lines.append(f"USDT: ${usdt:,.2f}")
        lines.append(f"\nTotal: ${total_usd:,.2f}")
        
        await update.message.reply_text('\n'.join(lines))
    except Exception as e:
        await update.message.reply_text(f"Error fetching balance: {e}")

async def buy_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Execute a buy order."""
    if len(context.args) < 2:
        await update.message.reply_text("Usage: /buy BTC 100 (buys $100 of BTC)")
        return
    
    coin = context.args[0].upper()
    usd_amount = float(context.args[1])
    symbol = f"{coin}/USDT"
    
    try:
        ticker = exchange.fetch_ticker(symbol)
        amount = usd_amount / ticker['last']
        
        await update.message.reply_text(
            f"⚡ Buying ${usd_amount} of {coin}...\n"
            f"Amount: {amount:.6f} {coin}"
        )
        
        order = exchange.create_market_buy_order(symbol, amount)
        
        await update.message.reply_text(
            f"✅ Buy order filled!\n"
            f"Order ID: {order['id']}\n"
            f"Amount: {order['amount']:.6f} {coin}\n"
            f"Price: ${order.get('average', ticker['last']):,.2f}"
        )
    except Exception as e:
        await update.message.reply_text(f"❌ Order failed: {e}")

# --- PRICE ALERT SYSTEM ---

active_alerts: dict = {}

async def set_alert(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Set a price alert."""
    if len(context.args) < 2:
        await update.message.reply_text("Usage: /alert BTC 5 (alerts on 5% move)")
        return
    
    coin = context.args[0].upper()
    threshold = float(context.args[1])
    symbol = f"{coin}/USDT"
    
    ticker = exchange.fetch_ticker(symbol)
    base_price = ticker['last']
    
    active_alerts[coin] = {
        'symbol': symbol,
        'base_price': base_price,
        'threshold': threshold,
    }
    
    await update.message.reply_text(
        f"🔔 Alert set for {coin}\n"
        f"Base price: ${base_price:,.2f}\n"
        f"Alert on: ±{threshold}% move"
    )

async def monitor_alerts(bot: Bot):
    """Background task: check price alerts every minute."""
    while True:
        for coin, alert in list(active_alerts.items()):
            try:
                ticker = exchange.fetch_ticker(alert['symbol'])
                current = ticker['last']
                change_pct = ((current - alert['base_price']) / alert['base_price']) * 100
                
                if abs(change_pct) >= alert['threshold']:
                    emoji = "🚨🟢" if change_pct > 0 else "🚨🔴"
                    await bot.send_message(
                        chat_id=CHAT_ID,
                        text=(
                            f"{emoji} PRICE ALERT: {coin}\n"
                            f"Change: {change_pct:+.2f}%\n"
                            f"Current: ${current:,.2f}\n"
                            f"Base: ${alert['base_price']:,.2f}"
                        )
                    )
                    # Update base price after alert
                    active_alerts[coin]['base_price'] = current
            except Exception as e:
                print(f"Alert error for {coin}: {e}")
        
        await asyncio.sleep(60)  # Check every minute

# --- SIGNAL SENDER (call this from your trading bot) ---

async def send_signal(bot: Bot, coin: str, action: str, price: float, reason: str):
    """Send a trading signal notification."""
    emoji = "🟢 BUY" if action == "BUY" else "🔴 SELL"
    await bot.send_message(
        chat_id=CHAT_ID,
        text=(
            f"{emoji} SIGNAL: {coin}\n"
            f"Price: ${price:,.2f}\n"
            f"Reason: {reason}"
        )
    )

def main():
    app = Application.builder().token(TOKEN).build()
    
    app.add_handler(CommandHandler("start", start))
    app.add_handler(CommandHandler("price", get_price))
    app.add_handler(CommandHandler("balance", get_balance))
    app.add_handler(CommandHandler("buy", buy_command))
    app.add_handler(CommandHandler("alert", set_alert))
    
    # Start alert monitor as background job
    async def post_init(app):
        asyncio.create_task(monitor_alerts(app.bot))
    
    app.post_init = post_init
    
    print("🤖 Telegram crypto bot started!")
    app.run_polling()

if __name__ == '__main__':
    main()

Step 5: Run the Bot

python crypto_telegram_bot.py

Open Telegram, message your bot /start — and it should respond!

Step 6: Integrating With Your Trading Bot

The most powerful use case is having your trading bot send signals through Telegram:

# In your main trading bot, import and use:
import asyncio
from telegram import Bot

bot = Bot(token=os.getenv('TELEGRAM_TOKEN'))
CHAT_ID = int(os.getenv('TELEGRAM_CHAT_ID'))

async def notify(message: str):
    await bot.send_message(chat_id=CHAT_ID, text=message)

# Call before executing any trade:
asyncio.run(notify(f"🤖 Executing BUY order for BTC\nPrice: $45,230\nAmount: 0.022 BTC"))

Deploying 24/7

# On your VPS
pm2 start crypto_telegram_bot.py --name telegram-bot --interpreter python3
pm2 save

Security Tips

• Never share your bot token publicly
• Add a user ID check so only you can control the bot:

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    if update.effective_user.id != CHAT_ID:
        await update.message.reply_text("Unauthorized.")
        return
    # ... rest of handler

More Ideas to Extend This Bot

• /news BTC — Fetch latest Bitcoin news
• /feargreed — Crypto Fear & Greed Index
• Daily portfolio summary at 8am automatically
• Chart images using mplfinance library

Get the full code on our Tools page with additional features included.