"""monitor_health — i forward-monitor stanno ancora registrando? (puro, testabile) IL BUCO CHE CHIUDE. Il progetto ha **tre gate pre-registrati** che si decideranno leggendo una serie forward: STATARB **2026-09-27**, XSR01 **2026-10-23**, DVOLSPREAD kill **2026-10-24** (decisione piena 2027-01-24). Di quelle serie, **una sola** ha una guardia d'integrita' (`paper_dvolspread`, contabilita' a 3 stati + veto sotto l'80% di barre attive). Le altre no: se un monitor smettesse di avanzare in silenzio — feed rotto, eccezione ingoiata, cron che non gira — il gate verrebbe deciso su dati mancanti, e i dati mancanti **somigliano a dati tranquilli**. E' lo stesso schema gia' pagato due volte: `fresh_5m` che falliva in silenzio (26/07) e il feed-freeze del 14/07. La regola del progetto e' che *"non vedo" non e' "va tutto bene"* — qui diventa: **un monitor fermo non e' un monitor che dice zero.** DUE GUASTI, DUE MISURE (una sola non basta): * **coda** — l'ultima barra e' vecchia: il monitor si e' fermato *adesso*. Misurata come eta' dell'ultima registrazione, sulla cadenza dichiarata dal monitor. * **buchi interni** — il monitor gira ma perde barre: la serie e' piu' corta di quanto il suo stesso arco temporale richieda. Misurata come copertura fra la PRIMA e l'ULTIMA barra. Un monitor che ha perso il 30% delle barre di mezzo passerebbe qualunque controllo di freschezza. CADENZE, che NON sono tutte uguali (e sbagliarle vuol dire un falso allarme a settimana): * `paper_prevday` registra a barra ORARIA (864 barre in 36 giorni), avanzato dal cron daily; * `paper_combo` vive sul calendario di BORSA (gambe IB): il venerdi' e' l'ultima barra fino al lunedi', e tre giorni di eta' sono la norma, non un guasto; * gli altri sono crypto giornalieri, 365 barre l'anno. SOGLIA DI COPERTURA 0.80 — la stessa del veto d'integrita' gia' pre-registrato per DVOLSPREAD ("se le barre attive sono sotto l'80% si ESTENDE la finestra, non si decide"). Riusarla e' voluto: una soglia diversa per monitor renderebbe i gate non confrontabili. ⚠️ LIMITE DICHIARATO. `np.busday_count` non conosce le FESTIVITA' di borsa: sul calendario equity la copertura attesa e' sovrastimata di un giorno per festivita' nell'arco misurato (oggi `paper_combo` legge 96% per il 3 luglio, non per un guasto). E' un errore in direzione CONSERVATIVA — segnala di piu', non di meno — e resta ampiamente sopra la soglia 0.80 finche' le festivita' sono <20% dei giorni. Se un giorno servisse precisione qui, la risposta e' un calendario di borsa vero, non alzare la soglia. """ from __future__ import annotations import csv import json from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path import numpy as np MIN_COVERAGE = 0.80 # sotto: la serie ha buchi -> non ci si decide sopra un gate MIN_BARS_PER_JUDGE = 2 # con una barra sola non esiste ne' coda ne' copertura GRAZIA_PREMATURO_MIN = 5.0 # tolleranza fra chiusura della barra e mtime del file (clock/IO) SOGLIA_ORARI_H = 6.0 # (#30) max_age_h <= questa soglia = "sorvegliante orario": i tre # con max_age 3h (cuscino_watch, venue_watch, collect_chain) oggi, # derivato dalla tabella e non da una lista di nomi cablata @dataclass(frozen=True) class MonitorSpec: """Cosa ci si aspetta da un monitor. `gate` = la decisione che dipende da questa serie.""" name: str directory: str series: str # returns.jsonl | equity.csv cadence_h: float # passo nominale fra due barre calendar: str = "crypto" # crypto (365g) | equity (giorni di borsa) max_age_h: float = 48.0 # oltre: FERMO gate: str | None = None # True = i timestamp sono APERTURE di barra (convenzione resample_tf/ccxt/HL) -> vale il # controllo PREMATURO. False = timbri di esecuzione (es. collect_chain scrive l'ora del # giro): il timbro coincide con la scrittura per costruzione e il controllo mentirebbe # SEMPRE (P14: una guardia piu' stretta del contratto insegna a ignorare la sezione). open_labeled: bool = True MONITORS: tuple[MonitorSpec, ...] = ( MonitorSpec("paper_portfolio", "paper_portfolio", "equity.csv", 24.0), MonitorSpec("paper_prevday", "paper_prevday", "returns.jsonl", 1.0), MonitorSpec("paper_statarb", "paper_statarb", "returns.jsonl", 24.0, gate="2026-09-27 — gate di deploy STATARB-RESID"), MonitorSpec("paper_xsr", "paper_xsr", "returns.jsonl", 24.0, gate="2026-10-23 — gate di deploy XSR01 (Sharpe>=1.0 E haircut<=40%)"), MonitorSpec("paper_dvolspread", "paper_dvolspread", "returns.jsonl", 24.0, gate="2026-10-24 kill / 2027-01-24 decisione — DVOLSPREAD"), # calendario di BORSA: le gambe GTAA vengono da IB, il weekend non e' un guasto. MonitorSpec("paper_combo", "paper_combo", "equity.csv", 24.0, calendar="equity", max_age_h=120.0), # Raccolta catena opzioni (successore di cerbero-bite): ORARIA. Non alimenta un gate a data # fissa ma una serie IRRECUPERABILE — un'ora non raccolta non si recupera da nessuna fonte, # quindi la soglia d'eta' e' stretta (3h = due giri persi). MonitorSpec("collect_chain", "chain_collect", "runs.jsonl", 1.0, max_age_h=3.0, gate="serie irrecuperabile — la catena opzioni non si ricostruisce a posteriori", open_labeled=False), # Sorveglianza del collaterale USDE (cron 12:35 UTC): se si ferma, spariscono gli allarmi # depeg/quota e il silenzio si legge come "va tutto bene" (P5). max_age 30h = un giro # mancato + margine; timbri di esecuzione come collect_chain -> open_labeled=False. MonitorSpec("usde_watch", "live", "usde_watch.jsonl", 24.0, max_age_h=30.0, gate="allarmi depeg/quota USDE + verdetto eligibilita' 29/08", open_labeled=False), # Cuscino USDC di regolamento (cron :53, ORARIO, debito §5.17 / issue #3): se si ferma nessuno # vede il saldo USDC scendere sotto il cuscino, e la riconversione automatica non parte. # max_age 3h = due giri persi; timbri di esecuzione -> open_labeled=False. MonitorSpec("cuscino_watch", "live", "cuscino_watch.jsonl", 1.0, max_age_h=3.0, gate="cuscino USDC di regolamento: allerta + riconversione automatica", open_labeled=False), # Trigger di regime CC01 (cron_daily 00:30, GIORNALIERO, #12): dal 03/09 al 16/09 e' morto 14 # notti di fila senza che nessuno lo vedesse (#10), perche' non scriveva niente da sorvegliare. # Il percorso e' quello di `cc01_regime_watch.RUNS` (un test li incrocia, P1). max_age 30h = # un giro mancato + margine; timbri di esecuzione -> open_labeled=False. MonitorSpec("cc01_regime_watch", "cc01_regime", "runs.jsonl", 24.0, max_age_h=30.0, gate="trigger di regime CC01: WARN>=10% / ALERT>=15% funding 30g (riapre il dossier)", open_labeled=False), # I sei sorveglianti che generalizzano la riparazione di cc01 (#16): ognuno registra il # proprio giro in `data//runs.jsonl` (src.live.giri.registra_giro), timbri di # ESECUZIONE come cc01 -> open_labeled=False. cron_daily.sh (GIORNALIERO) per i primi # cinque: max_age 30h = un giro mancato + margine. MonitorSpec("edge_watch", "edge_watch", "runs.jsonl", 24.0, max_age_h=30.0, gate="kill del libro live (Sharpe 36m < -0,5 / protezione TP01)", open_labeled=False), MonitorSpec("fee_watch", "fee_watch", "runs.jsonl", 24.0, max_age_h=30.0, gate="regola fee §3 decisa in anticipo", open_labeled=False), MonitorSpec("scale_watch", "scale_watch", "runs.jsonl", 24.0, max_age_h=30.0, gate="GATE SCALA-01 criterio A2", open_labeled=False), MonitorSpec("vrp_f_watch", "vrp_f_watch", "runs.jsonl", 24.0, max_age_h=30.0, gate="criteri (a)(b) §5.18", open_labeled=False), MonitorSpec("venue_news", "venue_news", "runs.jsonl", 24.0, max_age_h=30.0, gate="annunci Deribit, debito 8", open_labeled=False), # cron_book.sh: gira OGNI ORA accanto al libro vivo. max_age 3h = due giri persi. MonitorSpec("venue_watch", "venue_watch", "runs.jsonl", 1.0, max_age_h=3.0, gate="tripwire di venue orario", open_labeled=False), ) def read_stamps(path: Path) -> list[datetime]: """Timestamp di ogni barra registrata, in UTC e ordinati. Formati: jsonl con `ts` in ms epoch, oppure csv con la data in prima colonna.""" if not path.exists(): return [] out: list[datetime] = [] if path.suffix == ".jsonl": for ln in path.read_text().splitlines(): ln = ln.strip() if not ln: continue try: ts = json.loads(ln).get("ts") except json.JSONDecodeError: continue if ts is not None: out.append(datetime.fromtimestamp(float(ts) / 1000.0, tz=timezone.utc)) else: with path.open() as fh: for row in csv.DictReader(fh): raw = next(iter(row.values())) try: d = datetime.fromisoformat(str(raw)) except ValueError: continue out.append(d if d.tzinfo else d.replace(tzinfo=timezone.utc)) return sorted(out) def expected_bars(first: datetime, last: datetime, cadence_h: float, calendar: str) -> int: """Quante barre DOVREBBERO esserci fra la prima e l'ultima, estremi inclusi.""" if calendar == "equity": n = int(np.busday_count(first.date(), last.date())) + 1 return max(n, 1) span_h = (last - first).total_seconds() / 3600.0 return int(round(span_h / cadence_h)) + 1 def age_hours(last: datetime, now: datetime, calendar: str) -> float: """Eta' dell'ultima barra. Sul calendario di borsa il weekend non invecchia la serie: contarlo produrrebbe un allarme ogni lunedi'.""" raw = (now - last).total_seconds() / 3600.0 if calendar == "equity": bd = int(np.busday_count(last.date(), now.date())) return min(raw, bd * 24.0 + (raw % 24.0)) return raw def assess(spec: MonitorSpec, stamps: list[datetime], now: datetime, mtime: datetime | None = None) -> dict: """Verdetto su un monitor. Gli stati NON sono due: 'assente' e 'troppo giovane per un giudizio' non sono 'OK', e non devono poter passare per tali. `mtime` = quando il file di serie e' stato scritto l'ultima volta. Serve al controllo PREMATURO (r0822_monitor_audit): se l'ultima barra registrata CHIUDE dopo che il file e' stato scritto, il monitor ha consumato una barra parziale — e' il difetto che ha fatto registrare 4/41/2 minuti di mercato al giorno a 3 monitor con gate. Freschezza e copertura NON possono vederlo: una serie fresca, completa e SBAGLIATA le passa entrambe.""" base = dict(name=spec.name, gate=spec.gate, n_bars=len(stamps), age_h=None, coverage=None, expected=None) if not stamps: return {**base, "status": "ASSENTE", "why": "nessuna serie leggibile: il monitor non ha mai scritto, o lo stato e' perso"} if mtime is not None and spec.open_labeled: chiude = stamps[-1] + timedelta(hours=spec.cadence_h) ritardo_min = (chiude - mtime).total_seconds() / 60.0 if ritardo_min > GRAZIA_PREMATURO_MIN: return {**base, "age_h": age_hours(stamps[-1], now, spec.calendar), "status": "PREMATURO", "why": (f"l'ultima barra chiude alle {chiude:%Y-%m-%d %H:%M}Z ma il file e' " f"stato scritto {ritardo_min/60:.1f}h PRIMA: registrata una barra " "parziale — la serie va rigenerata, non solo riparata")} age = age_hours(stamps[-1], now, spec.calendar) if len(stamps) < MIN_BARS_PER_JUDGE: return {**base, "age_h": age, "status": "NUOVO", "why": f"{len(stamps)} barra: non c'e' abbastanza serie per misurare la copertura"} exp = expected_bars(stamps[0], stamps[-1], spec.cadence_h, spec.calendar) cov = len(stamps) / exp if exp > 0 else float("nan") d = {**base, "age_h": age, "coverage": cov, "expected": exp} if age > spec.max_age_h: return {**d, "status": "FERMO", "why": f"ultima barra {age:.0f}h fa (limite {spec.max_age_h:.0f}h)"} if cov < MIN_COVERAGE: return {**d, "status": "BUCATO", "why": f"copertura {cov:.0%} < {MIN_COVERAGE:.0%}: {exp - len(stamps)} barre mancanti"} return {**d, "status": "OK", "why": ""} def check_all(root: Path, now: datetime | None = None, monitors: tuple[MonitorSpec, ...] = MONITORS) -> list[dict]: now = now or datetime.now(timezone.utc) out = [] for spec in monitors: f = root / "data" / spec.directory / spec.series stamps = read_stamps(f) mtime = (datetime.fromtimestamp(f.stat().st_mtime, tz=timezone.utc) if f.exists() else None) out.append(assess(spec, stamps, now, mtime=mtime)) return out def alerts_from(rows: list[dict]) -> list[str]: """Cosa vale un allarme. Un monitor NUOVO non lo e' (sta solo iniziando); un monitor che alimenta un gate pre-registrato e' peggio degli altri, e l'allarme lo dice.""" out = [] for r in rows: if r["status"] in ("OK", "NUOVO"): continue msg = f"{r['name']}: {r['status']} — {r['why']}" if r["gate"]: msg += f" | METTE A RISCHIO il gate {r['gate']}" out.append(msg) return out # =========================================================================== # GIRO ORARIO (#30). Il giro giornaliero (cron_daily) resta com'e': report completo, avviso # sempre. I sorveglianti a max_age <= 6h (oggi cuscino_watch, venue_watch, collect_chain) si # scoprono fermi fino a 24h dopo se nessuno li guarda fuori da quel giro: qui si guardano ogni # ora, ma si notifica solo cio' che e' NUOVO rispetto all'ultimo avviso (niente Telegram a # raffica sullo stesso guasto fermo). Tutto PURO: il marcatore lo legge/scrive lo script. # =========================================================================== def monitors_orari(monitors: tuple[MonitorSpec, ...] = MONITORS, soglia_h: float = SOGLIA_ORARI_H) -> tuple[MonitorSpec, ...]: """I monitor con cadenza di controllo ORARIA: max_age_h <= soglia_h. Derivato dalla tabella `MONITORS`, non da una lista di nomi cablata: aggiungere un monitor orario lo include qui senza toccare questa funzione.""" return tuple(m for m in monitors if m.max_age_h <= soglia_h) def chiave_alert(row: dict) -> str: """Chiave stabile di un alert per il marcatore 'gia' detto': nome + stato, NON il testo (`why` contiene l'eta' dell'ultima barra, che cambia a ogni giro anche se il guasto e' lo stesso).""" return f"{row['name']}:{row['status']}" def alert_nuovi(rows: list[dict], detti: set[str]) -> list[dict]: """Le righe-alert (non OK, non NUOVO — vedi `alerts_from`) la cui chiave NON e' nel marcatore `detti`: quelle davvero nuove, le sole da notificare.""" return [r for r in rows if r["status"] not in ("OK", "NUOVO") and chiave_alert(r) not in detti] def marcatore_aggiornato(rows: list[dict], detti: set[str]) -> set[str]: """Il marcatore dopo questo giro: le chiavi dei monitor ancora in alert restano/si aggiungono, quelle dei monitor tornati OK si TOLGONO — cosi' un nuovo guasto sullo stesso monitor riparte con un nuovo avviso invece di restare 'gia' detto' per sempre. Non scrive nulla: lo fa lo script, e solo se la notifica e' partita (stessa regola di fee_watch #29).""" out = set(detti) for r in rows: chiavi_monitor = {k for k in out if k.startswith(f"{r['name']}:")} if r["status"] == "OK": out -= chiavi_monitor elif r["status"] != "NUOVO": out.add(chiave_alert(r)) return out