PythagorasGoal/scripts/live/telegram_daily.py

"""REPORT GIORNALIERO su Telegram — stato del conto reale, cosa sta facendo il sistema e perche'.

PERCHE' ESISTE (2026-07-25). Gli alert Telegram esistenti (`src/live/notifier`) scattano SOLO su
ordine eseguito o errore. Con il libro flat — che e' lo stato normale e corretto quando il trend e'
giu' — questo significa **silenzio per settimane**, indistinguibile da un sistema morto. Questo
report rompe il silenzio ogni giorno e dice tre cose: dove sta il conto, perche' il sistema non
opera, e quanto manca perche' operi.

NON invia ordini, non tocca posizioni, non legge segreti oltre a quelli gia' usati dall'esecutore.
E' sola lettura: se fallisce, l'esecuzione oraria del libro non ne risente in alcun modo.

    uv run python scripts/live/telegram_daily.py            # calcola e invia
    uv run python scripts/live/telegram_daily.py --dry-run  # stampa e basta, non invia
"""
from __future__ import annotations

import json
import sys
from datetime import date, datetime, timezone
from pathlib import Path

import numpy as np
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)


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:
    return f"${x:,.2f}"


def _soglia_accensione(segni: list[int], neg_refs: list[float]) -> float | None:
    """(#35 C3) Quale riferimento, superato, accende TP01 (mean(segni) > 0)?

    I segni sono sempre in {-1,+1}, mai 0 (trend_state li costruisce con `1 if px>refs[h]
    else -1`: nessun pareggio possibile). Con k segni negativi da far diventare positivi, la
    somma cresce di 2 per flip: serve il piu' piccolo k tale che sum(segni)+2k > 0. La soglia
    e' il k-esimo riferimento negativo piu' basso (k=1 -> il minimo, come prima del fix; k=2 ->
    il secondo, ecc.) — non sempre il minimo: con TUTTI i riferimenti negativi, superare solo il
    piu' basso lascia la media <= 0 e TP01 a 0 (bug segnalato in #35 C3).

    None se la media e' gia' positiva (nessuna soglia da dare) o se k supera il numero di
    riferimenti negativi disponibili (non dovrebbe accadere: con n segni, flippare tutti gli n
    negativi porta comunque la somma a +n > 0, quindi k <= len(neg_refs) sempre — resta una
    guardia difensiva, non un caso noto)."""
    s = sum(segni)
    if s > 0 or not neg_refs:
        return None
    k = 0
    while s + 2 * k <= 0:
        k += 1
    if k > len(neg_refs):
        return None
    return sorted(neg_refs)[k - 1]


def trend_state() -> list[str]:
    """Stato TSMOM per asset + quanto manca all'accensione (2 orizzonti su 3 positivi).

    NB convenzione TP01: la direzione e' la media dei SEGNI (-1/+1) sugli orizzonti, poi clippata
    a >=0 (long-flat). Con 1 orizzonte su 3 positivo la media e' -0.33 -> target 0. Serve quindi
    che DUE orizzonti siano positivi perche' il libro si accenda: non basta il piu' breve."""
    from src.data.downloader import load_data
    out = []
    for a in ("BTC", "ETH"):
        try:
            d = load_data(a, "1h")
            s = pd.Series(d["close"].astype(float).values,
                          index=pd.to_datetime(d["timestamp"], unit="ms", utc=True))
            c = s.resample("1D").last().dropna().values
            if len(c) < max(HORIZONS) + 2:
                continue
            px = float(c[-1])
            refs = {h: float(c[-1 - h]) for h in HORIZONS}
            sg = {h: (1 if px > refs[h] else -1) for h in HORIZONS}
            m = float(np.mean(list(sg.values())))
            tgt = max(0.0, m)
            line = f"  {a} {px:,.0f} · segni " + "/".join(f"{sg[h]:+d}" for h in HORIZONS) + \
                   f" → TP01 {tgt:.2f}"
            if tgt == 0.0:
                r = _soglia_accensione(list(sg.values()), [refs[h] for h in HORIZONS if sg[h] < 0])
                if r is not None:
                    line += f"\n     accensione a {r:,.0f} ({(r/px-1)*100:+.1f}%)"
            out.append(line)
        except Exception as e:
            out.append(f"  {a}: stato trend non calcolabile ({type(e).__name__})")
    return out


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)
            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, flat
    except Exception as 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]:
    out = []
    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_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_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_txt}")
        except Exception:
            out.append(f"  {name}: log illeggibile · {gate_txt}")
    return out


def last_trade() -> str:
    p = ROOT / "data" / "live" / "book_executions.jsonl"
    if not p.exists():
        return "  nessun ordine registrato"
    try:
        rows = [json.loads(x) for x in p.read_text().splitlines() if x.strip()]
        if not rows:
            return "  nessun ordine registrato"
        t = rows[-1]
        d = datetime.fromisoformat(t["ts_utc"]).replace(tzinfo=timezone.utc)
        giorni = (datetime.now(timezone.utc) - d).days
        return (f"  ultimo: {t['ts_utc'][:10]} ({giorni}g fa) {t['asset']} {t['action']} "
                f"@ {t.get('price')} · totale ordini: {len(rows)}")
    except Exception:
        return "  ledger ordini illeggibile"


def build() -> str:
    oggi = datetime.now(timezone.utc).strftime("%d/%m %H:%M UTC")
    L = [f"📊 <b>Book Deribit — {oggi}</b>", "", "<b>Conto</b>"]
    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 += ["", "<b>Perche' non opera</b>" if flat else "<b>Stato trend</b>"]
    L += trend_state()
    L += ["", "<b>Ordini</b>", last_trade()]
    L += ["", "<b>Monitor in osservazione</b>"]
    L += monitors()
    return "\n".join(L)


def main() -> None:
    txt = build()
    if "--dry-run" in sys.argv[1:]:
        print(txt.replace("<b>", "").replace("</b>", ""))
        return
    print("inviato" if send(txt) else "NON inviato (config Telegram assente o rete KO)")


USO = """uso: telegram_daily.py [--dry-run]

  uv run python scripts/live/telegram_daily.py   # calcola e invia
  uv run python scripts/live/telegram_daily.py --dry-run   # stampa e basta, non invia

Dettaglio nel docstring in testa al file."""

if __name__ == "__main__":
    from src.live.cli import valida
    valida("telegram_daily.py", USO, flag=("--dry-run",), con_valore=())
    main()