File size: 2,739 Bytes
d53dc44 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 | """
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}")
|