""" telegram_bot.py — Telegram integration for notifications and user commands. Sends formatted alert messages and handles interactive commands: /add — set a price range alert /remove — stop tracking an asset /list — show all tracked assets /price — get last known price for an asset /alerts — show last 10 triggered alerts /pause — silence all alert notifications /resume — re-enable alert notifications /status — bot health check /help — this message """ import asyncio from datetime import datetime, timezone from telegram import Update from telegram.constants import ParseMode from telegram.ext import ( Application, CommandHandler, ContextTypes, ) import config from alert_engine import AlertEngine, AlertEvent from logger import get_logger log = get_logger("telegram_bot") # Rate-limit: minimum seconds between Telegram messages _MIN_SEND_INTERVAL = 1.0 _last_send_time: float = 0.0 # ── Markdown V2 escape helper ───────────────────────────────────────────────── _MD2_SPECIAL = r"\_*[]()~`>#+-=|{}.!" def _esc(text: str) -> str: """Escape special characters for Telegram MarkdownV2.""" for ch in _MD2_SPECIAL: text = text.replace(ch, f"\\{ch}") return text class TelegramBot: """Wraps python-telegram-bot to provide both commands and alert sending.""" def __init__(self, alert_engine: AlertEngine, dry_run: bool = False) -> None: self.engine = alert_engine self.dry_run = dry_run self._app: Application | None = None self._start_time = datetime.now(timezone.utc) # ── Initialise the bot application ─────────────────────────────────── def build_app(self) -> Application: """Create and configure the Telegram Application.""" if not config.TELEGRAM_BOT_TOKEN: raise ValueError( "TELEGRAM_BOT_TOKEN is not set. " "Add it to your .env file (see .env.example)." ) builder = Application.builder().token(config.TELEGRAM_BOT_TOKEN) self._app = builder.build() # Register command handlers self._app.add_handler(CommandHandler("start", self._cmd_help)) self._app.add_handler(CommandHandler("help", self._cmd_help)) self._app.add_handler(CommandHandler("add", self._cmd_add)) self._app.add_handler(CommandHandler("remove", self._cmd_remove)) self._app.add_handler(CommandHandler("list", self._cmd_list)) self._app.add_handler(CommandHandler("price", self._cmd_price)) self._app.add_handler(CommandHandler("alerts", self._cmd_alerts)) self._app.add_handler(CommandHandler("pause", self._cmd_pause)) self._app.add_handler(CommandHandler("resume", self._cmd_resume)) self._app.add_handler(CommandHandler("status", self._cmd_status)) return self._app # ── Send alert notification ────────────────────────────────────────── async def send_alert(self, event: AlertEvent) -> None: """Send a price-alert message to the configured Telegram chat.""" asset = _esc(event.asset_name) price = _esc(f"${event.price:,.4f}") rng = _esc(f"${event.min_price:,.4f} – ${event.max_price:,.4f}") ts = _esc(event.timestamp.strftime("%Y-%m-%d %H:%M:%S")) text = ( f"🔔 *Price Alert — {asset}*\n" f"━━━━━━━━━━━━━━━━━━━━━━━\n" f"💰 *Current Price:* `{price}`\n" f"📊 *Target Range:* `{rng}`\n" f"🕐 *Time \\(UTC\\):* `{ts}`\n" f"━━━━━━━━━━━━━━━━━━━━━━━\n" f"_Price has entered your defined range\\!_" ) if self.dry_run: log.info("[DRY-RUN] Would send alert:\n%s", text) return await self._send_message(text) async def _send_message(self, text: str, chat_id: str | None = None) -> None: """Send a MarkdownV2 message with simple rate-limiting.""" global _last_send_time target_chat = chat_id or config.TELEGRAM_CHAT_ID if not target_chat: log.warning("No TELEGRAM_CHAT_ID configured, skipping message send") return # Simple rate limiting using running loop time loop = asyncio.get_running_loop() now = loop.time() elapsed = now - _last_send_time if elapsed < _MIN_SEND_INTERVAL: await asyncio.sleep(_MIN_SEND_INTERVAL - elapsed) try: if self._app and self._app.bot: await self._app.bot.send_message( chat_id=target_chat, text=text, parse_mode=ParseMode.MARKDOWN_V2, ) _last_send_time = asyncio.get_running_loop().time() log.debug("Telegram message sent to chat %s", target_chat) except Exception as exc: log.error("Failed to send Telegram message: %s", exc) # ── Command handlers ───────────────────────────────────────────────── async def _cmd_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /help and /start.""" commodities = sorted(k for k in config.YFINANCE_TICKERS if k in ("GOLD","SILVER","CRUDE_OIL")) stocks = sorted(k for k in config.YFINANCE_TICKERS if k not in commodities) crypto = sorted(config.BINANCE_SYMBOLS.keys()) text = ( "📈 *Multi\\-Asset Price Alert Bot*\n\n" "*Commands:*\n" "`/add ` — Track an asset\n" "`/remove ` — Stop tracking\n" "`/list` — Show all tracked assets\n" "`/price ` — Get current price\n" "`/alerts` — Last 10 triggered alerts\n" "`/pause` — Mute all notifications\n" "`/resume` — Unmute notifications\n" "`/status` — Bot health check\n" "`/help` — This message\n\n" f"*📦 Commodities:* `{', '.join(commodities)}`\n" f"*₿ Crypto:* `{', '.join(crypto)}`\n" f"*📊 Stocks/Indices:* `{', '.join(stocks)}`\n\n" "_Example:_ `/add BTC 80000 90000`" ) await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN_V2) async def _cmd_add(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /add .""" args = context.args if not args or len(args) != 3: await update.message.reply_text( "⚠️ Usage: `/add `\n" "Example: `/add BTC 80000 90000`", parse_mode=ParseMode.MARKDOWN_V2, ) return asset_name = args[0].upper() try: min_price = float(args[1]) max_price = float(args[2]) except ValueError: await update.message.reply_text("⚠️ MIN and MAX must be numbers\\.") return if min_price >= max_price: await update.message.reply_text("⚠️ MIN must be less than MAX\\.", parse_mode=ParseMode.MARKDOWN_V2) return if asset_name not in config.ALL_ASSETS: supported = _esc(", ".join(sorted(config.ALL_ASSETS))) await update.message.reply_text( f"⚠️ Unknown asset `{_esc(asset_name)}`\\.\n\nSupported: `{supported}`", parse_mode=ParseMode.MARKDOWN_V2, ) return self.engine.add_asset(asset_name, min_price, max_price) min_str = _esc(f"${min_price:,.4f}") max_str = _esc(f"${max_price:,.4f}") await update.message.reply_text( f"✅ Tracking *{_esc(asset_name)}*\n" f"Range: `{min_str}` \u2013 `{max_str}`", parse_mode=ParseMode.MARKDOWN_V2, ) async def _cmd_remove(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /remove .""" args = context.args if not args or len(args) != 1: await update.message.reply_text("⚠️ Usage: `/remove `", parse_mode=ParseMode.MARKDOWN_V2) return asset_name = args[0].upper() if self.engine.remove_asset(asset_name): await update.message.reply_text( f"🗑️ Stopped tracking *{_esc(asset_name)}*\\.", parse_mode=ParseMode.MARKDOWN_V2, ) else: await update.message.reply_text( f"⚠️ `{_esc(asset_name)}` is not being tracked\\.", parse_mode=ParseMode.MARKDOWN_V2, ) async def _cmd_list(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /list — show all tracked assets.""" details = self.engine.get_asset_details() if not details: await update.message.reply_text( "📭 No assets tracked\\. Use `/add` to start\\.", parse_mode=ParseMode.MARKDOWN_V2, ) return lines = ["📋 *Tracked Assets:*\n"] for d in sorted(details, key=lambda x: x["name"]): state_emoji = "🟢" if d["state"] == "INSIDE" else "⚪" pause_tag = " ⏸" if d.get("paused") else "" price_str = _esc(f"${d['last_price']:,.4f}") if d["last_price"] else "N/A" rng_min = _esc(f"${d['min']:,.4f}") rng_max = _esc(f"${d['max']:,.4f}") lines.append( f"{state_emoji} *{_esc(d['name'])}*{_esc(pause_tag)}\n" f" Range: `{rng_min}` \u2013 `{rng_max}`\n" f" Last: `{price_str}` \\| State: `{_esc(d['state'])}`" ) await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN_V2) async def _cmd_price(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /price — show last known price.""" args = context.args if not args or len(args) != 1: await update.message.reply_text( "⚠️ Usage: `/price `\nExample: `/price BTC`", parse_mode=ParseMode.MARKDOWN_V2, ) return asset_name = args[0].upper() price = self.engine.get_asset_price(asset_name) if price is None: await update.message.reply_text( f"❓ No price data for `{_esc(asset_name)}` yet\\.\n" "_Make sure it's tracked with_ `/add`\\.", parse_mode=ParseMode.MARKDOWN_V2, ) else: await update.message.reply_text( f"💰 *{_esc(asset_name)}* latest price: `{_esc(f'${price:,.4f}')}`", parse_mode=ParseMode.MARKDOWN_V2, ) async def _cmd_alerts(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /alerts — show last 10 triggered alerts.""" history = self.engine.get_alert_history(limit=10) if not history: await update.message.reply_text( "📭 No alerts have been triggered yet\\.", parse_mode=ParseMode.MARKDOWN_V2, ) return lines = ["🔔 *Recent Alerts \\(newest first\\):*\n"] for event in history: ts = _esc(event.timestamp.strftime("%m/%d %H:%M UTC")) name = _esc(event.asset_name) px = _esc(f"${event.price:,.4f}") lines.append(f"• *{name}* — `{px}` at {ts}") await update.message.reply_text("\n".join(lines), parse_mode=ParseMode.MARKDOWN_V2) async def _cmd_pause(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /pause — silence all alert notifications.""" self.engine.pause_all() await update.message.reply_text( "⏸️ *Alerts paused\\.* You won't receive notifications until you send `/resume`\\.", parse_mode=ParseMode.MARKDOWN_V2, ) async def _cmd_resume(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /resume — re-enable alert notifications.""" self.engine.resume_all() await update.message.reply_text( "▶️ *Alerts resumed\\.* You will now receive notifications again\\.", parse_mode=ParseMode.MARKDOWN_V2, ) async def _cmd_status(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Handle /status — bot health check.""" now = datetime.now(timezone.utc) uptime = now - self._start_time hours, remainder = divmod(int(uptime.total_seconds()), 3600) minutes, seconds = divmod(remainder, 60) tracked = self.engine.get_tracked_assets() crypto_cnt = sum(1 for n in tracked if n in config.BINANCE_SYMBOLS) stock_cnt = len(tracked) - crypto_cnt paused_flag = "⏸️ YES" if self.engine.is_paused() else "▶️ NO" alert_cnt = len(self.engine.get_alert_history(limit=config.ALERT_HISTORY_MAX)) text = ( "🤖 *Bot Status*\n" f"━━━━━━━━━━━━━━━━━━━━━━━\n" f"⏱️ Uptime: `{_esc(f'{hours}h {minutes}m {seconds}s')}`\n" f"📊 Tracking: `{len(tracked)}` assets " f"\\(`{crypto_cnt}` crypto, `{stock_cnt}` stocks/commodities\\)\n" f"🔔 Alerts fired: `{alert_cnt}`\n" f"⏸️ Paused: {_esc(paused_flag)}\n" f"🔄 Stock poll: `{config.STOCK_POLL_INTERVAL}s`\n" f"⏳ Cooldown: `{config.ALERT_COOLDOWN_SECONDS}s`\n" f"📡 Crypto: WebSocket streaming\n" f"━━━━━━━━━━━━━━━━━━━━━━━" ) await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN_V2)