From 3af6e1f0c368a210eddf321c143a7c02363f0739 Mon Sep 17 00:00:00 2001 From: "Adriano (goldrake)" Date: Sun, 27 Sep 2026 15:18:25 +0000 Subject: [PATCH] telegram_daily: gate da monitor_health (non cablati), formato giorni a tre casi, header condizionale su flat (#35 C14) Co-Authored-By: Claude Sonnet 5 --- scripts/live/telegram_daily.py | 72 ++++++++++++++++++++++++++++------ tests/test_telegram_daily.py | 72 +++++++++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 14 deletions(-) diff --git a/scripts/live/telegram_daily.py b/scripts/live/telegram_daily.py index ac8238c..be42e6a 100644 --- a/scripts/live/telegram_daily.py +++ b/scripts/live/telegram_daily.py @@ -25,11 +25,32 @@ import pandas as pd ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) +from src.live.monitor_health import MONITORS from src.live.notifier import send HORIZONS = (30, 90, 180) -GATES = [("STATARB-RESID", date(2026, 9, 27), "data/paper_statarb/returns.jsonl", "net_modeled"), - ("XSR01", date(2026, 10, 23), "data/paper_xsr/returns.jsonl", "net_modeled")] + + +def _gate_date(nome_monitor: str, cablata: date) -> date: + """(#35 C14) la data del gate viene da monitor_health.MONITORS (`MonitorSpec.gate`), la + STESSA tabella che "METTE A RISCHIO il gate" negli alert dei sorveglianti: una fonte, non + una seconda dichiarazione duplicata di CLAUDE.md §4. Il campo e' prosa libera + ("YYYY-MM-DD — descrizione"): si legge la data solo se e' nella forma attesa. `cablata' e' + rete di sicurezza se il monitor sparisse o il campo cambiasse forma — non deve succedere per + i due gate qui sotto, oggi allineati (verificato a mano contro CLAUDE.md §4).""" + for m in MONITORS: + if m.name == nome_monitor and m.gate: + try: + return date.fromisoformat(m.gate[:10]) + except ValueError: + break + return cablata + + +GATES = [("STATARB-RESID", _gate_date("paper_statarb", date(2026, 9, 27)), + "data/paper_statarb/returns.jsonl", "net_modeled"), + ("XSR01", _gate_date("paper_xsr", date(2026, 10, 23)), + "data/paper_xsr/returns.jsonl", "net_modeled")] def _fmt_usd(x: float) -> str: @@ -94,22 +115,41 @@ def trend_state() -> list[str]: return out -def book_state() -> list[str]: - """Conto reale, posizioni e target netto correnti. Sola lettura.""" +def book_state() -> tuple[list[str], bool | None]: + """Conto reale, posizioni e target netto correnti. Sola lettura. + + (#35 C14) Ritorna anche se il libro e' DAVVERO flat (nessuna posizione, nessun target: stessa + condizione che questa funzione gia' usa riga per riga per stampare "flat" invece di "pos -> + target"): serve a `build()` per decidere se l'header "Perche' non opera" e' vero o + fuorviante, SENZA una lettura nuova — None se lo stato del libro non e' leggibile (in quel + caso non sappiamo dire se opera).""" try: from src.live.book import book_report r = book_report() eq = r.get("equity") lines = [f" equity {_fmt_usd(eq)}" if eq else " equity non leggibile"] + flat = True for a in r.get("assets", []): pos, net = a.get("position_usd", 0.0), a.get("net_target", 0.0) - stato = "flat" if abs(pos) < 1 and abs(net) < 1 else f"pos {_fmt_usd(pos)} → target {_fmt_usd(net)}" + is_flat = abs(pos) < 1 and abs(net) < 1 + flat = flat and is_flat + stato = "flat" if is_flat else f"pos {_fmt_usd(pos)} → target {_fmt_usd(net)}" lines.append(f" {a['asset']}: {stato} (TP {a.get('tp_frac', 0):+.2f} · SKH {a.get('skh_sign', 0):+d})") if r.get("skh_error"): lines.append(f" ⚠️ SKH feed KO: {r['skh_error']}") - return lines + return lines, flat except Exception as e: - return [f" ⚠️ stato libro non leggibile: {type(e).__name__}: {e}"] + return [f" ⚠️ stato libro non leggibile: {type(e).__name__}: {e}"], None + + +def _fmt_giorni(giorni: int) -> str: + """(#35 C14) tre casi distinti, non un segno "−" cablato davanti a un intero che puo' essere + negativo: con giorni=-1 quel cablaggio stampava "(−-1g)", un doppio segno illeggibile.""" + if giorni > 0: + return f"-{giorni}g" + if giorni == 0: + return "oggi" + return f"scaduto da {-giorni}g" def monitors() -> list[str]: @@ -117,20 +157,21 @@ def monitors() -> list[str]: for name, dec, path, key in GATES: p = ROOT / path giorni = (dec - date.today()).days + gate_txt = f"gate {dec} ({_fmt_giorni(giorni)})" if not p.exists(): - out.append(f" {name}: in attesa della prima barra · gate {dec} (−{giorni}g)") + out.append(f" {name}: in attesa della prima barra · {gate_txt}") continue try: rows = [json.loads(x) for x in p.read_text().splitlines() if x.strip()] if not rows: - out.append(f" {name}: 0 barre · gate {dec} (−{giorni}g)") + out.append(f" {name}: 0 barre · {gate_txt}") continue r = np.array([x[key] for x in rows], float) sh = float(r.mean() / r.std() * np.sqrt(365)) if r.std() > 0 else 0.0 tot = float(np.prod(1 + r) - 1) * 100 - out.append(f" {name}: {len(rows)}g · {tot:+.2f}% · Sh {sh:+.2f} · gate {dec} (−{giorni}g)") + out.append(f" {name}: {len(rows)}g · {tot:+.2f}% · Sh {sh:+.2f} · {gate_txt}") except Exception: - out.append(f" {name}: log illeggibile · gate {dec} (−{giorni}g)") + out.append(f" {name}: log illeggibile · {gate_txt}") return out @@ -154,8 +195,13 @@ def last_trade() -> str: def build() -> str: oggi = datetime.now(timezone.utc).strftime("%d/%m %H:%M UTC") L = [f"📊 Book Deribit — {oggi}", "", "Conto"] - L += book_state() - L += ["", "Perche' non opera"] + righe_conto, flat = book_state() + L += righe_conto + # (#35 C14) "Perche' non opera" e' vero solo se il libro e' DAVVERO flat (nessuna posizione, + # nessun target: gia' letto sopra da book_state, nessuna lettura nuova). Se non e' flat (ha + # posizioni o un target diverso da zero) o non si sa (book non leggibile), l'header fisso + # mentirebbe: si usa un titolo neutro. + L += ["", "Perche' non opera" if flat else "Stato trend"] L += trend_state() L += ["", "Ordini", last_trade()] L += ["", "Monitor in osservazione"] diff --git a/tests/test_telegram_daily.py b/tests/test_telegram_daily.py index d11d1b6..4cfc2b3 100644 --- a/tests/test_telegram_daily.py +++ b/tests/test_telegram_daily.py @@ -7,6 +7,7 @@ e non invia niente in dry-run. from __future__ import annotations import sys +from datetime import date from pathlib import Path import pytest @@ -21,8 +22,33 @@ telegram_daily = pytest.importorskip("telegram_daily") def test_build_non_solleva_e_ha_le_sezioni(): txt = telegram_daily.build() assert isinstance(txt, str) and len(txt) > 50 - for sezione in ("Conto", "Perche' non opera", "Ordini", "Monitor in osservazione"): + for sezione in ("Conto", "Ordini", "Monitor in osservazione"): assert sezione in txt, f"sezione mancante: {sezione}" + # (#35 C14) l'header sul trend e' condizionale: esattamente uno dei due, non piu' fisso. + assert ("Perche' non opera" in txt) != ("Stato trend" in txt) + + +def test_header_perche_non_opera_solo_se_il_libro_e_flat(monkeypatch): + """(#35 C14) L'header fisso mentiva quando il libro NON e' fermo (posizioni o target != 0): + con flat=True resta "Perche' non opera".""" + monkeypatch.setattr(telegram_daily, "book_state", lambda: ([" equity $1.00"], True)) + assert "Perche' non opera" in telegram_daily.build() + + +def test_header_neutro_se_il_libro_non_e_flat(monkeypatch): + monkeypatch.setattr(telegram_daily, "book_state", + lambda: ([" BTC: pos $300 → target $300"], False)) + txt = telegram_daily.build() + assert "Stato trend" in txt and "Perche' non opera" not in txt + + +def test_header_neutro_se_il_libro_non_e_leggibile(monkeypatch): + """Se book_state non sa dire se e' flat (eccezione -> None), niente header "non opera" + dichiarato senza saperlo davvero: si usa il neutro.""" + monkeypatch.setattr(telegram_daily, "book_state", + lambda: ([" ⚠️ stato libro non leggibile: Errore"], None)) + txt = telegram_daily.build() + assert "Stato trend" in txt and "Perche' non opera" not in txt def test_nessuna_scrittura_ne_ordine(): @@ -82,3 +108,47 @@ def test_soglia_accensione_servono_tutti_i_flip(): = il piu' alto dei due (l'ultimo a bastare), non il minimo.""" r = telegram_daily._soglia_accensione([-1, -1], [200.0, 100.0]) assert r == 200.0 + + +# ---------------------------------------------------------------- (#35 C14) gate: data, formato + + +def test_gate_date_legge_da_monitor_health(): + """La data non e' una seconda dichiarazione: viene da monitor_health.MONITORS (`gate`), + la stessa tabella che gli alert dei sorveglianti citano come "METTE A RISCHIO il gate".""" + from src.live.monitor_health import MONITORS + m = next(x for x in MONITORS if x.name == "paper_statarb") + assert m.gate[:10] == "2026-09-27" + assert telegram_daily._gate_date("paper_statarb", date(2030, 1, 1)) == date(2026, 9, 27) + m2 = next(x for x in MONITORS if x.name == "paper_xsr") + assert telegram_daily._gate_date("paper_xsr", date(2030, 1, 1)) == date.fromisoformat(m2.gate[:10]) + + +def test_gate_date_usa_la_cablata_se_il_monitor_non_esiste(): + """Rete di sicurezza: se il nome non e' in MONITORS (o cambia), non si inventa una data.""" + assert telegram_daily._gate_date("non_esiste_piu'", date(2030, 1, 1)) == date(2030, 1, 1) + + +def test_fmt_giorni_tre_casi_mai_il_doppio_segno(): + """(#35 C14) giorni>0 -> '-Ng'; 0 -> 'oggi'; <0 -> 'scaduto da Ng'. Mai '(−-1g)'.""" + assert telegram_daily._fmt_giorni(5) == "-5g" + assert telegram_daily._fmt_giorni(0) == "oggi" + assert telegram_daily._fmt_giorni(-1) == "scaduto da 1g" + assert "-1g" not in telegram_daily._fmt_giorni(-1).replace("scaduto da 1g", "") # niente doppio segno + + +def test_monitors_formatta_i_tre_casi_del_giorno(monkeypatch, tmp_path): + """I rami di monitors() (barra assente qui: file inesistenti in tmp_path) usano tutti + _fmt_giorni, non il vecchio '(−{giorni}g)' cablato: futuro, oggi e passato si distinguono.""" + from datetime import timedelta + oggi = date.today() + monkeypatch.setattr(telegram_daily, "ROOT", tmp_path) + monkeypatch.setattr(telegram_daily, "GATES", + [("FUTURO", oggi + timedelta(days=10), "data/nonexiste1/r.jsonl", "k"), + ("OGGI", oggi, "data/nonexiste2/r.jsonl", "k"), + ("PASSATO", oggi - timedelta(days=3), "data/nonexiste3/r.jsonl", "k")]) + righe = telegram_daily.monitors() + assert "(-10g)" in righe[0] + assert "(oggi)" in righe[1] + assert "(scaduto da 3g)" in righe[2] + assert "−-" not in "".join(righe) and "(−" not in "".join(righe) # mai il doppio segno