PythagorasGoal/src/live/venue_watch.py

"""VENUE WATCH — allerta precoce sul fallimento dell'exchange (Deribit).

PERCHE' ESISTE. TP01+SKH01+VRP01 stanno tutti sullo stesso conto Deribit: quasi-ortogonali sui
ritorni, **perfettamente correlati sul fallimento del venue** (`r0726_venue_risk.py`). L'operatore
ha deciso il 26/07 di restare concentrato al 100% fino a $20k, quindi ridurre l'ESPOSIZIONE e'
fuori discussione. Resta una sola leva: **il tempo**. I fallimenti reali non sono istantanei —
Mt.Gox gato' i prelievi per mesi, FTX ebbe ~72h, Bitfinex nel 2018-19 resto' dislocato per
**2.324 ore consecutive**. Se il saldo esce dentro quella finestra, la perdita non e' totale.

IL SEGNALE (misurato in `scripts/research/r0726_venue_tripwire.py`).
Quando un exchange gata i prelievi, **l'arbitraggio si rompe** e il suo prezzo si stacca dal
consenso in modo persistente: non si puo' piu' comprare qui e vendere la' per chiudere lo scarto.
Il segnale e' |scarto|, NON il suo segno: su Mt.Gox il BTC andava a PREMIO (si comprava BTC per far
uscire valore), su un venue in fuga si vede lo sconto. Entrambi dicono la stessa cosa.

TARATURA (dichiarata, non tarata a occhio):
  * zero falsi allarmi su Deribit in 8 anni (65.043 ore BTC + 64.541 ETH), inclusi il crash
    COVID 2020-03, il maggio 2021, LUNA e il novembre 2022;
  * margine >= 3x sul caso storico piu' DEBOLE (FTX ~300 bps) -> soglia <= 100 bps;
  * a quei due vincoli, minima latenza -> **100 bps persistenti 4 ore a segno costante**.
  Margine: 3x su FTX, 5x su QuadrigaCX, 10-20x su Mt.Gox.

CONTROLLO POSITIVO (un rilevatore che non segnala mai puo' essere semplicemente rotto): puntato su
BITFINEX 2018-19 (problemi bancari/Tether) scatta **22 volte**, episodio piu' lungo 2.324h a +447
bps di picco. Specificita', non cecita'.

TRE STATI, e il terzo non e' il primo:
  OK     — misurato, sotto soglia;
  ALERT  — misurato, sopra soglia e persistente -> Telegram;
  BLIND  — NON misurato (referenze irraggiungibili o in disaccordo fra loro). **"Non vedo" non e'
           "va tutto bene"**: se dura, e' un allarme suo (piu' morbido). Stessa lezione della
           contabilita' a 3 stati di `paper_dvolspread` (25/07).

PERCHE' ALLERTA E NON BLOCCA. L'azione giusta a un vero positivo e' **prelevare**, che richiede
comunque un intervento manuale: una chiave API con permesso di prelievo sarebbe essa stessa un
rischio (se la chiave esce, escono i fondi). Bloccare l'esecuzione invece non protegge il saldo —
il saldo e' a rischio anche stando flat. Quindi: allerta forte, azione umana.

RUNBOOK a un ALERT (pre-deciso ORA per non doverlo decidere nel momento sbagliato):
  1. verificare che non sia un guasto delle referenze (il campo `n_refs` e `ref_spread_bps`);
  2. controllare `public/status` (campo `platform_locked`) e i canali ufficiali Deribit;
  3. **tentare subito un prelievo di prova piccolo**: e' l'unica evidenza DIRETTA. Se non passa in
     tempi normali, l'allarme e' vero indipendentemente da ogni altra spiegazione;
  4. se il prelievo di prova non passa: flat del book e prelievo totale. Costo atteso di sbagliarsi
     = **0.248% dell'equity** (uscita 3 giorni, misurato in `r0726_venue_response.py`), contro il
     100% che si evita se e' vero.
"""
from __future__ import annotations

import json
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
STATE_PATH = PROJECT_ROOT / "data" / "venue_watch" / "state.json"

# --- taratura CONGELATA (vedi docstring; cambiarla invalida "zero falsi allarmi in 8 anni") ---
THRESHOLD_BPS = 100.0        # scarto minimo per contare come anomalia
PERSIST_HOURS = 4            # ore consecutive a segno costante prima di allertare
REF_DISAGREE_BPS = 100.0     # oltre questo le referenze litigano fra loro -> BLIND
BLIND_ALERT_HOURS = 12       # "non vedo" per tanto tempo e' anch'esso una notizia
# Una manutenzione ANNUNCIATA e' attesa: vale un avviso morbido, non il runbook del prelievo.
# Ma il 18/08 Deribit aveva annunciato 15-30 minuti e la piattaforma e' rimasta bloccata per ore:
# quindi la manutenzione declassa l'allarme SOLO dentro una finestra di tolleranza, e oltre quella
# RIALZA — perche' una manutenzione che dura sei volte l'annuncio e' di nuovo una notizia.
# 2 osservazioni orarie: sopra ogni finestra annunciata plausibile, sotto l'evento vero di ieri.
MAINT_GRACE_HOURS = 2

ASSETS = ("BTC", "ETH")
# Referenze USD indipendenti da Deribit. NIENTE USDT: il depeg 2022 sposta BTC/USDT fino al 3%
# dal dollaro e produrrebbe falsi allarmi giganti (regola del progetto sul DATO).
# ⚠️⚠️ KRAKEN AGGIUNTA IL 2026-08-19, e la ragione non e' la robustezza generica: **Coinbase ha
# comprato Deribit**. Le comunicazioni del 12-18/08 parlano di "Deribit by Coinbase", del Coinbase
# Index come riferimento per una parte dei perpetual lineari, e dello spot Deribit instradato su
# Coinbase Exchange. Questo rilevatore misura "Deribit sta scollando dal mondo?" contro un consenso
# indipendente: se una delle due referenze e' la CASA MADRE, uno shock di Coinbase muove Deribit e
# la referenza INSIEME, e lo scarto in bps resta piccolo proprio nell'ora in cui dovrebbe aprirsi.
# Con due sole referenze, il consenso indipendente si riduceva di fatto a Bitstamp.
# NB: BTC ed ETH restano sul Deribit Index (hanno opzioni quotate), quindi la contaminazione oggi
# e' di proprieta', non ancora di calcolo. Si aggiunge finche' la cosa e' teorica.
# ⚠️ ONESTA' SULLA TARATURA: "zero falsi allarmi in 8 anni" e' stato misurato con l'insieme di
# referenze di allora. THRESHOLD_BPS e PERSIST_HOURS non sono stati toccati, ma il CONSENSO si
# calcola su tre serie invece che due, e quel numero non e' stato ri-misurato. La direzione pero'
# e' verificabile a mente sul codice: con 3 referenze il consenso e' la MEDIANA (robusta a un
# outlier) invece della media di due, e lo spread max-min si allarga -> si va piu' spesso in BLIND,
# che e' lo stato MORBIDO. Cioe' il modo in cui questa modifica puo' sbagliare e' "allerta di
# meno", non "grida al lupo". Se un giorno serve il numero vero, si rilancia
# scripts/research/r0726_venue_tripwire.py con la terza serie.
REF_VENUES = [("coinbase", {"BTC": "BTC/USD", "ETH": "ETH/USD"}),
              ("bitstamp", {"BTC": "BTC/USD", "ETH": "ETH/USD"}),
              ("kraken", {"BTC": "BTC/USD", "ETH": "ETH/USD"})]
DERIBIT_SYMBOL = {"BTC": "BTC/USD:BTC", "ETH": "ETH/USD:ETH"}


# ===========================================================================
# nucleo PURO (nessuna rete, nessun file) — e' qui che vivono i test
# ===========================================================================
@dataclass
class AssetState:
    """Stato per-asset dello streak. `sign` 0 = nessuno streak in corso."""
    streak_hours: int = 0
    sign: int = 0
    blind_hours: int = 0
    last_bps: float | None = None
    alerted: bool = False        # gia' allertato per QUESTO streak (evita spam orario)


@dataclass
class LockState:
    """Stato del blocco piattaforma. Prima non esisteva: l'allarme era un `if` secco senza
    memoria, quindi ripartiva a ogni giro dell'ora. Il 18/08 sono usciti quattro 🚨 identici
    con scritto 'PRIMO PASSO: prelievo di prova' per una manutenzione annunciata — e tutti e
    quattro DOPO che il book aveva gia' ripreso a eseguire."""
    hours: int = 0
    alerted_soft: bool = False
    alerted_hard: bool = False


@dataclass
class WatchState:
    assets: dict[str, AssetState] = field(default_factory=dict)
    lock: LockState = field(default_factory=LockState)
    last_ts: int = 0

    def get(self, a: str) -> AssetState:
        return self.assets.setdefault(a, AssetState())


def lock_step(st: LockState, locked: bool | None, maintenance: bool,
              grace: int = MAINT_GRACE_HOURS) -> tuple[LockState, str]:
    """Avanza lo stato del lock di UNA osservazione oraria. PURA.

    Livelli: "OK" | "MAINT" (avviso morbido) | "ALERT" (runbook prelievo) | "RIENTRATO" | "MUTO".
    "MUTO" = la condizione dura ma e' gia' stata detta: si tace invece di ripetersi ogni ora.

    ⚠️ `locked is None` (public/status illeggibile) NON e' un rientro: le ore restano dove sono e
    non si annuncia niente. Dichiarare "e' rientrato" perche' non si e' riusciti a guardare
    sarebbe la bugia peggiore di tutte in questo file.
    """
    new = LockState(**asdict(st))
    if locked is None:
        return new, "MUTO" if st.hours else "OK"
    if not locked:
        if st.hours:
            return LockState(), "RIENTRATO"
        return LockState(), "OK"
    new.hours = st.hours + 1
    if maintenance and new.hours <= grace:
        if st.alerted_soft:
            return new, "MUTO"
        new.alerted_soft = True
        return new, "MAINT"
    # niente manutenzione dichiarata -> un blocco inspiegato e' subito il caso serio;
    # manutenzione oltre la tolleranza -> ha sforato, e lo sforamento e' esso stesso la notizia.
    if st.alerted_hard:
        return new, "MUTO"
    new.alerted_soft, new.alerted_hard = True, True
    return new, "ALERT"


def dislocation_bps(deribit: float, refs: list[float]) -> tuple[float | None, int, float | None]:
    """(scarto firmato in bps, n referenze, spread fra referenze in bps).

    Ritorna scarto None se non c'e' un consenso utilizzabile: servono >= 2 referenze concordi.
    Con UNA sola referenza non si distingue "Deribit e' fuori" da "la referenza e' rotta" — e' il
    controllo che rende il sistema usabile invece che rumoroso.
    """
    vals = sorted(float(x) for x in refs if x and x > 0)
    if len(vals) < 2:
        return None, len(vals), None
    n = len(vals)
    consensus = vals[n // 2] if n % 2 else 0.5 * (vals[n // 2 - 1] + vals[n // 2])
    if consensus <= 0:
        return None, n, None
    spread = (vals[-1] - vals[0]) / consensus * 1e4
    if spread > REF_DISAGREE_BPS:
        return None, n, spread
    return (deribit - consensus) / consensus * 1e4, n, spread


def step(st: AssetState, bps: float | None, threshold: float = THRESHOLD_BPS,
         persist: int = PERSIST_HOURS, blind_alert: int = BLIND_ALERT_HOURS) -> tuple[AssetState, str]:
    """Avanza lo stato di UN asset di una osservazione oraria. PURA.

    Ritorna (nuovo stato, livello) con livello in {"OK", "BLIND", "WATCH", "ALERT"}.
      - bps None            -> BLIND; lo streak si AZZERA (non si accumula evidenza su dati che
                               non parlano) e si conta da quanto non si vede;
      - |bps| > soglia      -> lo streak cresce se il segno e' lo stesso, altrimenti riparte;
      - streak >= persist   -> ALERT (una volta sola per streak, poi resta WATCH senza rispammare).
    """
    new = AssetState(**asdict(st))
    if bps is None:
        new.blind_hours = st.blind_hours + 1
        new.streak_hours, new.sign, new.alerted = 0, 0, False
        return new, ("BLIND" if new.blind_hours >= blind_alert else "OK")
    new.blind_hours = 0
    new.last_bps = float(bps)
    sign = 1 if bps > 0 else (-1 if bps < 0 else 0)
    if abs(bps) > threshold and sign != 0:
        new.streak_hours = st.streak_hours + 1 if sign == st.sign else 1
        new.sign = sign
        if new.streak_hours < persist:
            new.alerted = False
            return new, "WATCH"
        already = st.alerted and st.sign == sign
        new.alerted = True
        return new, ("WATCH" if already else "ALERT")
    new.streak_hours, new.sign, new.alerted = 0, 0, False
    return new, "OK"


# ===========================================================================
# I/O: stato su disco, rete, report
# ===========================================================================
def load_state(path: Path = STATE_PATH) -> WatchState:
    if not path.exists():
        return WatchState()
    try:
        raw = json.loads(path.read_text())
        return WatchState(assets={k: AssetState(**v) for k, v in raw.get("assets", {}).items()},
                          lock=LockState(**raw.get("lock", {})),
                          last_ts=int(raw.get("last_ts", 0)))
    except Exception:
        return WatchState()      # stato illeggibile -> si riparte pulito, mai un crash del cron


def save_state(st: WatchState, path: Path = STATE_PATH) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(
        {"assets": {k: asdict(v) for k, v in st.assets.items()},
         "lock": asdict(st.lock), "last_ts": st.last_ts}, indent=2))


def platform_status() -> tuple[bool | None, str]:
    """`public/status` di Deribit: (bloccata?, valore GREZZO). None se non leggibile.

    ⚠️ Il valore grezzo si porta dietro apposta: Deribit risponde anche `"partial"`, e fino al
    2026-08-19 il messaggio diceva comunque «locked=true» — cioe' DICHIARAVA un valore che non
    aveva letto. Il runbook al passo 2 manda a controllare proprio questo campo: mandarci qualcuno
    con in testa la stringa sbagliata e' peggio che non dirgliela. E lo stato non lo salvava
    nessuno, per cui il valore del 18/08 oggi non e' piu' ricostruibile.
    """
    try:
        import ccxt
        r = ccxt.deribit({"enableRateLimit": True}).publicGetStatus()
        grezzo = str(r.get("result", {}).get("locked", "false")).lower()
        return grezzo not in ("false", "none"), grezzo
    except Exception:
        return None, "non leggibile"


def platform_locked() -> bool | None:
    """Compatibilita': solo il booleano. Il valore vero sta in `platform_status()`."""
    try:
        return platform_status()[0]
    except Exception:
        return None


def _last_price(exchange_id: str, symbol: str, errori: list[str] | None = None) -> float | None:
    """⚠️ Raccoglie il MOTIVO del fallimento invece di inghiottirlo. E' cosi' che si distingue
    «Deribit e' in manutenzione annunciata» da «Deribit non risponde e non si sa perche'»: ccxt
    solleva `OnMaintenance` col codice 11051, e quell'informazione il 18/08 c'era gia' — ce
    l'aveva lo strato book, nello stesso identico minuto, e non arrivava a chi decide la
    gravita' dell'allarme."""
    try:
        import ccxt
        ex = getattr(ccxt, exchange_id)({"enableRateLimit": True})
        t = ex.fetch_ticker(symbol)
        p = t.get("last") or t.get("close")
        return float(p) if p else None
    except Exception as e:  # noqa: BLE001
        if errori is not None:
            errori.append(f"{type(e).__name__}: {e}")
        return None


def is_maintenance(errori: list[str]) -> bool:
    """PURA. La firma della manutenzione Deribit negli errori raccolti."""
    return any("onmaintenance" in e.lower() or "system_maintenance" in e.lower() or "11051" in e
               for e in errori)


def observe() -> dict:
    """Una osservazione: prezzo Deribit e referenze, per asset. Solo letture pubbliche."""
    locked, grezzo = platform_status()
    err_deribit: list[str] = []
    out: dict = {"ts": int(time.time()), "platform_locked": locked, "locked_raw": grezzo,
                 "assets": {}}
    for a in ASSETS:
        der = _last_price("deribit", DERIBIT_SYMBOL[a], err_deribit)
        refs = [_last_price(eid, syms[a]) for eid, syms in REF_VENUES]
        refs = [r for r in refs if r]
        bps, n, spread = (None, 0, None) if der is None else dislocation_bps(der, refs)
        out["assets"][a] = dict(deribit=der, n_refs=n, ref_spread_bps=spread, bps=bps)
    out["maintenance"] = is_maintenance(err_deribit)
    out["deribit_errors"] = err_deribit
    return out


def run_once(state_path: Path = STATE_PATH, sender=None) -> dict:
    """Un giro completo: osserva, avanza lo stato, INVIA (se gli si da' un `sender`), salva.

    PERCHE' L'INVIO E' ENTRATO QUI. Fino al 2026-08-28 questa funzione salvava lo stato — coi
    marcatori `alerted` gia' a True — e l'invio lo faceva il chiamante DOPO. Con un tasso di
    invii falliti misurato del 6,9% (2 su 29), un 🚨 perso restava perso per l'EPISODIO INTERO:
    l'ora dopo lo stato diceva «gia' detto» e usciva WATCH/MUTO. Gli episodi storici durano
    200-2.324 ore, quindi il buco non e' teorico. Il marcatore «gia' detto» adesso si scrive
    solo se qualcuno l'ha davvero sentito.

    `sender(report) -> (ok: bool, motivo: str)`. Iniettato, non importato: e' cio' che tiene
    questa funzione testabile senza rete d'uscita, che era la ragione del disegno precedente.
    Senza `sender` il comportamento resta quello di prima e il report lo DICE (`invio`), invece
    di lasciar credere che qualcosa sia stato spedito.

    COSA SI DISFA SU UN INVIO FALLITO — solo i marcatori «gia' detto», non le misure:
      · asset in ALERT      -> `alerted=False`: l'ora dopo ri-allerta;
      · lock MAINT / ALERT  -> `alerted_soft`/`alerted_hard`=False ma le ORE restano a correre,
        cosi' una manutenzione che sfora la grazia sale ad ALERT anche se il trasporto e' giu'
        (disfare anche le ore congelerebbe l'escalation proprio mentre non si riesce a parlare);
      · lock RIENTRATO      -> si ripristina l'intero LockState precedente, perche' il rientro
        si annuncia una volta sola e senza le ore a zero non ci sarebbe piu' un rientro da dire.
    Restano fuori BLIND (si ripete ogni ora da solo) e il livello WATCH (non consuma nulla).
    """
    st = load_state(state_path)
    prima_lock = LockState(**asdict(st.lock))     # per disfare il rientro, che si dice una volta
    asset_alertati: list[str] = []
    obs = observe()
    report = {"ts": obs["ts"], "platform_locked": obs["platform_locked"],
              "locked_raw": obs.get("locked_raw"), "maintenance": obs.get("maintenance", False),
              "levels": {}, "detail": obs["assets"], "alerts": [], "severity": None}
    for a in ASSETS:
        o = obs["assets"][a]
        new, level = step(st.get(a), o["bps"])
        st.assets[a] = new
        report["levels"][a] = level
        if level == "ALERT":
            asset_alertati.append(a)
            report["alerts"].append(
                f"{a}: scarto {o['bps']:+.0f} bps vs consenso ({o['n_refs']} referenze) "
                f"persistente da {new.streak_hours}h a segno costante")
        elif level == "BLIND":
            report["alerts"].append(
                f"{a}: consenso NON misurabile da {new.blind_hours}h "
                f"({o['n_refs']} referenze, spread {o['ref_spread_bps']})")
    if report["levels"] and any(l == "ALERT" for l in report["levels"].values()):
        report["severity"] = "alert"

    # --- blocco piattaforma: adesso passa dalla stessa disciplina degli altri stati
    lock_new, lock_lvl = lock_step(st.lock, obs["platform_locked"], obs.get("maintenance", False))
    st.lock = lock_new
    report["lock_level"], report["lock_hours"] = lock_lvl, lock_new.hours
    grezzo = obs.get("locked_raw") or "?"
    if lock_lvl == "MAINT":
        report["alerts"].append(
            f"Deribit bloccata (locked={grezzo}) e dichiara MANUTENZIONE: attesa, non allarme. "
            f"Il book si astiene da solo. Se supera {MAINT_GRACE_HOURS}h te lo ridico piu' forte.")
        report["severity"] = report["severity"] or "warn"
    elif lock_lvl == "ALERT":
        motivo = (f"la manutenzione ha SFORATO {MAINT_GRACE_HOURS}h" if obs.get("maintenance")
                  else "nessuna manutenzione dichiarata")
        report["alerts"].append(
            f"Deribit public/status: PIATTAFORMA BLOCCATA (locked={grezzo}) da {lock_new.hours}h "
            f"— {motivo}")
        report["severity"] = "alert"
    elif lock_lvl == "RIENTRATO":
        report["alerts"].append(f"Deribit public/status: rientrata (locked={grezzo})")
        report["severity"] = report["severity"] or "warn"

    st.last_ts = obs["ts"]

    # --- l'invio, e il commit dei marcatori SOLO se e' andato ---------------------------------
    if not report["alerts"]:
        report["invio"] = "niente da dire"
    elif sender is None:
        report["invio"] = ("non tentato (nessun sender): i marcatori 'gia' detto' restano "
                           "impostati, l'invio e' responsabilita' del chiamante")
    else:
        ok, motivo = sender(report)
        report["invio"] = "inviato" if ok else f"FALLITO — {motivo}"
        if not ok:
            for a in asset_alertati:
                st.assets[a].alerted = False
            if lock_lvl == "MAINT":
                st.lock.alerted_soft = False
            elif lock_lvl == "ALERT":
                st.lock.alerted_soft = st.lock.alerted_hard = False
            elif lock_lvl == "RIENTRATO":
                st.lock = prima_lock

    save_state(st, state_path)
    return report