diff --git a/scripts/live/live_execute.py b/scripts/live/live_execute.py
index 0cf9e25..c2f860a 100644
--- a/scripts/live/live_execute.py
+++ b/scripts/live/live_execute.py
@@ -25,6 +25,7 @@ sys.path.insert(0, str(PROJECT_ROOT))
from src.live.deribit import INSTRUMENT
from src.live.execution import DeribitTrader
+from src.live.notifier import notify
from src.live.shadow import ASSETS, WEIGHT, shadow_report
CONFIG = PROJECT_ROOT / "config" / "live.json"
@@ -47,7 +48,7 @@ def log_event(rec: dict):
f.write(json.dumps(rec) + "\n")
-def main():
+def _run():
cfg = load_config()
want_execute = "--execute" in sys.argv[1:]
enabled = bool(cfg["execution_enabled"])
@@ -72,6 +73,8 @@ def main():
if not r["online"]:
print(" conto non leggibile (offline) -> stop, non eseguo a cieco.")
+ if do_execute:
+ notify("⚠️ TP01 LIVE — conto offline", {"nota": "salto l'esecuzione, non opero a cieco"})
return
trader = DeribitTrader() if do_execute else None
@@ -100,9 +103,20 @@ def main():
log_event(dict(ts_utc=str(pd.Timestamp(r['last_data'])), asset=asset, action=act,
side=f.side, filled=f.filled, price=f.price, fee=f.fee_usdc,
verified=f.verified, notes=f.notes, pos_after=newpos))
+ det = dict(asset=asset, side=f.side, amount=round(f.filled, 4),
+ price=round(f.price or 0, 1), fee=round(f.fee_usdc, 5), pos_after=round(newpos, 0))
+ if f.verified:
+ notify(f"✅ TP01 {act}", det)
+ else:
+ notify("⚠️ TP01 ORDINE NON VERIFICATO", {**det, "notes": f.notes})
print(f" reconcile: pos ${newpos:,.0f}")
ds = trader.ensure_disaster_sl(INSTRUMENT[asset], sl_pct) # bracket: piazza se long, pulisce se flat
print(f" disaster-SL: {ds.get('state')}" + (f" @ ${ds['stop']:,.1f}" if ds.get("stop") else ""))
+ if ds.get("state") == "placed":
+ notify("🛡️ TP01 disaster-SL piazzato", {"asset": asset, "stop": round(ds.get("stop") or 0, 1),
+ "amount": round(ds.get("amount") or 0, 4)})
+ elif ds.get("state") == "place-failed":
+ notify("⚠️ TP01 disaster-SL FALLITO", {"asset": asset, "notes": ds.get("notes")})
actions.append(act)
print()
@@ -115,5 +129,13 @@ def main():
print(" => Esecuzione completata (vedi data/live/executions.jsonl).")
+def main():
+ try:
+ _run()
+ except Exception as e:
+ notify("🛑 TP01 LIVE — ERRORE", {"error": f"{type(e).__name__}: {e}"})
+ raise
+
+
if __name__ == "__main__":
main()
diff --git a/src/live/notifier.py b/src/live/notifier.py
new file mode 100644
index 0000000..460068c
--- /dev/null
+++ b/src/live/notifier.py
@@ -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" v{APP_VERSION}"
+ except Exception:
+ ver = ""
+ lines = [f"{title}{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)")