Adriano Dal Pastro
bf84bc91e2
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>
72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
"""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)")
|