Blame for telegram_bot.py

"""
telegram_bot.py — Telegram integration for notifications and user commands.

Sends formatted alert messages and handles interactive commands:
    /add    <ASSET> <MIN> <MAX>  — set a price range alert
    /remove <ASSET>              — stop tracking an asset
    /list                        — show all tracked assets
    /price  <ASSET>              — 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 <ASSET> <MIN> <MAX>` — Track an asset\n"
            "`/remove <ASSET>` — Stop tracking\n"
            "`/list` — Show all tracked assets\n"
            "`/price <ASSET>` — 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 <ASSET> <MIN> <MAX>."""
        args = context.args
        if not args or len(args) != 3:
            await update.message.reply_text(
                "⚠️ Usage: `/add <ASSET> <MIN_PRICE> <MAX_PRICE>`\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 <ASSET>."""
        args = context.args
        if not args or len(args) != 1:
            await update.message.reply_text("⚠️ Usage: `/remove <ASSET>`", 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 <ASSET> — show last known price."""
        args = context.args
        if not args or len(args) != 1:
            await update.message.reply_text(
                "⚠️ Usage: `/price <ASSET>`\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)