Telegram Bot for Trade Notifications

Telegram Bot for Trade Notifications

A trader can't watch the screen 24/7. Telegram bot for notifications solves this: instantly reports on executed orders, price alert levels reached, balance changes, and suspicious activity.

Types of Notifications

Trading:

• Order fully executed: ✅ BUY 0.1 BTC @ $42,150 — executed
• Order partially executed: ⚡ SELL 0.05/0.1 ETH @ $2,205 — partial
• Order cancelled: ❌ Order #12345 cancelled

Price alerts:

• 🔔 BTC/USDT reached $45,000 (your alert)

Account:

• Withdrawal confirmed: 💸 Withdrawal 500 USDT → 0x...a1b2 confirmed
• New login: 🔐 New login from IP 1.2.3.4 (Berlin)

Implementation

from telegram import Bot
from telegram.ext import Application, CommandHandler, CallbackQueryHandler

class TradeNotificationBot:
    def __init__(self, token: str):
        self.bot = Bot(token)
        self.app = Application.builder().token(token).build()
        self._setup_handlers()
    
    def _setup_handlers(self):
        self.app.add_handler(CommandHandler('alerts', self.list_alerts))
        self.app.add_handler(CommandHandler('addalert', self.add_price_alert))
        self.app.add_handler(CallbackQueryHandler(self.handle_callback))
    
    async def notify_fill(self, user_chat_id: int, trade: TradeEvent):
        emoji = '✅' if trade.side == 'buy' else '🔴'
        text = (
            f"{emoji} Order executed\n"
            f"Pair: {trade.pair}\n"
            f"Side: {'Buy' if trade.side == 'buy' else 'Sell'}\n"
            f"Volume: {trade.quantity} {trade.base_currency}\n"
            f"Price: ${trade.price:,.2f}\n"
            f"Total: ${trade.total:,.2f}\n"
            f"Time: {trade.timestamp.strftime('%H:%M:%S UTC')}"
        )
        await self.bot.send_message(chat_id=user_chat_id, text=text)
    
    async def notify_price_alert(self, user_chat_id: int, alert: PriceAlert):
        direction = '📈' if alert.direction == 'above' else '📉'
        await self.bot.send_message(
            chat_id=user_chat_id,
            text=f"{direction} {alert.pair} reached {alert.price}\n"
                 f"Current price: {alert.current_price}"
        )

Integration with Exchange Backend

Notifications sent through event-driven system. On every fill, matching engine publishes event to Redis Pub/Sub, notification worker subscribes and sends to Telegram:

import redis.asyncio as aioredis

async def notification_worker():
    r = aioredis.from_url('redis://localhost')
    pubsub = r.pubsub()
    await pubsub.subscribe('trade.fills', 'price.alerts', 'account.events')
    
    async for message in pubsub.listen():
        if message['type'] != 'message':
            continue
        
        event = json.loads(message['data'])
        chat_id = await get_user_chat_id(event['user_id'])
        
        if not chat_id:
            continue  # user hasn't linked Telegram
        
        match event['type']:
            case 'fill':
                await notification_bot.notify_fill(chat_id, TradeEvent(**event))
            case 'price_alert':
                await notification_bot.notify_price_alert(chat_id, PriceAlert(**event))

Telegram Account Linking

User links Telegram via unique token in exchange settings:

1. Exchange generates one-time token: LINK-a1b2c3d4
2. User sends /link LINK-a1b2c3d4 to bot
3. Bot saves (user_id, chat_id) connection
4. Notifications go to this chat_id

Development of Telegram notification bot with account linking and price alerts: 1–2 weeks.