feat(live): alert Telegram su esecuzione ed errori

src/live/notifier.py (stdlib, no-op se non configurato): legge TELEGRAM_BOT_TOKEN/CHAT_ID da env o
.env(.mainnet) gitignored. live_execute.py invia alert su: ordine eseguito (), ordine non
verificato (⚠️), disaster-SL piazzato/fallito (🛡️/⚠️), conto offline, e qualsiasi eccezione (🛑).
Nessun alert nei giorni flat/HOLD (no rumore). Config gia' presente in .env -> alert attivi.

Test config: uv run python -m src.live.notifier "msg". Test 28/28.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-20 16:09:09 +00:00
parent 3cba5bb9d0
commit bf84bc91e2
2 changed files with 94 additions and 1 deletions
+71
View File
@@ -0,0 +1,71 @@
"""Notifiche Telegram per l'esecuzione live (ordini + errori). Stdlib only, NO-OP se non configurato.
Config (gitignored): TELEGRAM_BOT_TOKEN + TELEGRAM_CHAT_ID da env, oppure da .env.mainnet / .env.
Se mancano, notify() ritorna False e non rompe nulla (l'esecuzione non dipende dagli alert).
Test della config: uv run python -m src.live.notifier "messaggio di prova"
"""
from __future__ import annotations
import os
import urllib.parse
import urllib.request
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[2]
def _cfg() -> tuple[str, str]:
tok = os.environ.get("TELEGRAM_BOT_TOKEN", "")
chat = os.environ.get("TELEGRAM_CHAT_ID", "")
if tok and chat:
return tok, chat
for fn in (".env.mainnet", ".env"):
p = PROJECT_ROOT / fn
if not p.exists():
continue
for ln in p.read_text().splitlines():
ln = ln.strip()
if ln.startswith("TELEGRAM_BOT_TOKEN=") and not tok:
tok = ln.split("=", 1)[1].strip()
elif ln.startswith("TELEGRAM_CHAT_ID=") and not chat:
chat = ln.split("=", 1)[1].strip()
return tok, chat
def send(text: str) -> bool:
tok, chat = _cfg()
if not tok or not chat:
return False
try:
url = f"https://api.telegram.org/bot{tok}/sendMessage"
data = urllib.parse.urlencode({"chat_id": chat, "text": text, "parse_mode": "HTML"}).encode()
urllib.request.urlopen(url, data, timeout=10)
return True
except Exception:
return False
def notify(title: str, data: dict | None = None) -> bool:
"""Invia un alert formattato. Ritorna True se inviato (config presente + rete ok)."""
try:
from src.version import APP_VERSION
ver = f" <code>v{APP_VERSION}</code>"
except Exception:
ver = ""
lines = [f"<b>{title}</b>{ver}"]
for k, v in (data or {}).items():
lines.append(f" {k}: {v}")
return send("\n".join(lines))
def is_configured() -> bool:
tok, chat = _cfg()
return bool(tok and chat)
if __name__ == "__main__":
import sys
msg = sys.argv[1] if len(sys.argv) > 1 else "TP01 — test alert"
ok = notify("🔔 " + msg, {"configurato": is_configured()})
print("inviato" if ok else "NON inviato (TELEGRAM_BOT_TOKEN/CHAT_ID assenti in env o .env.mainnet)")