monitoring/alerts.py

"""
monitoring/alerts.py — Telegram + email alert system.
Fires on circuit breakers, safe mode, orphans, WebSocket disconnects, etc.
"""
from __future__ import annotations

import logging
import threading

import requests

import config

logger = logging.getLogger("trading_system.alerts")

_send_lock = threading.Lock()


def send_telegram(message: str) -> bool:
    """Send a message via Telegram bot API.

    Returns True if sent successfully.
    """
    if not config.TELEGRAM_BOT_TOKEN or not config.TELEGRAM_CHAT_ID:
        logger.debug("Telegram not configured, skipping alert")
        return False

    url = f"https://api.telegram.org/bot{config.TELEGRAM_BOT_TOKEN}/sendMessage"
    payload = {
        "chat_id": config.TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "HTML",
    }

    with _send_lock:
        try:
            resp = requests.post(url, json=payload, timeout=10)
            if resp.status_code == 200:
                logger.debug("Telegram alert sent: %s", message[:80])
                return True
            else:
                logger.error(
                    "Telegram API error %d: %s", resp.status_code, resp.text[:200]
                )
                return False
        except Exception as e:
            logger.error("Failed to send Telegram alert: %s", e)
            return False


def alert(message: str):
    """Send an alert via all configured channels."""
    logger.info("ALERT: %s", message)
    send_telegram(message)


def alert_circuit_breaker(reason: str):
    alert(f"🛑 <b>CIRCUIT BREAKER</b>\n{reason}")


def alert_safe_mode(reason: str):
    alert(f"🚨 <b>SAFE MODE ACTIVATED</b>\n{reason}")


def alert_orphaned_order(symbol: str, entry_id: str):
    alert(f"🚨 <b>ORPHANED ORDER</b>\n{symbol} entry {entry_id} — missing stop/TP legs")


def alert_websocket_disconnect(duration_sec: float):
    alert(f"⚠️ <b>WebSocket Disconnected</b>\nDuration: {duration_sec:.0f}s")


def alert_corporate_action(symbol: str, action_type: str, ratio: float, date: str):
    alert(f"🔔 <b>CORPORATE ACTION</b>\n{symbol} {action_type} ratio={ratio:.4f} on {date}")


def alert_sentiment_stale(symbol: str, minutes: float):
    alert(f"⚠️ <b>Sentiment Stale</b>\n{symbol} last refresh {minutes:.0f} min ago")


def alert_end_of_day_close(positions: list[dict]):
    lines = ["📊 <b>End-of-Day Close</b>"]
    for p in positions:
        pnl = p.get("pnl", 0)
        lines.append(f"  {p['symbol']}: ${pnl:.2f}")
    alert("\n".join(lines))


def alert_shutdown(final_pnl: float):
    alert(f"🔌 <b>Bot Shut Down Gracefully</b>\nFinal P&L: ${final_pnl:.2f}")