From d55eb1353342062f9dfe54d37fc08e5022052727 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Thu, 30 Jul 2026 20:14:34 +0000 Subject: [PATCH] feat(chain): assorbita la raccolta catena opzioni, cerbero-bite dismesso MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cerbero-bite viene eliminato. L'unica sua parte irreversibile e' il DATO: una catena opzioni non si ricostruisce a posteriori (Deribit non serve book storici, non c'e' un secondo venue). Il codice si riscrive; le ore non raccolte no. ASSORBITO - scripts/live/collect_chain.py + scripts/cron_chain.sh (cron 25 * * * *): raccolta propria, ~570 strumenti/giro, ~3 min. - scripts/analysis/import_cb_archive.py: archivio 1.23M righe (2026-05-01+) + market_snapshots 17.402 righe (2026-03-26+: dealer gamma, gamma flip, rischio liquidazioni, funding cross — dati che non abbiamo altrove). - snapshot sqlite integrale in /opt/docker/backups/manual/ (SHA256). NON ASSORBITO, con motivo: motore credit-spread ETH (regola "niente short-vol da modello in deploy", conto a $52 contro minimo $720), GUI, kill switch/dead-man/audit (abbiamo venue_watch/edge_watch/monitor_health/ fee_watch), dvol_history (fetch_dvol.py ha storia PIU' LUNGA: 2020+ contro 2026-05), decisions/positions (0 posizioni). TRE DIFETTI DI BITE NON REPLICATI, tutti misurati il 30/07: 1. una chiamata per strumento invece di due (get_order_book?depth=3 da' gia' quote+greche+IV+OI+book+underlying) + prefiltro OI in una chiamata sola: 551 -> ~290 chiamate per asset; 2. pacing invece di raffica. Il carico non e' mai stato il problema: 570 chiamate/ora = 0.16/s DISTRIBUITE; bite le sparava in 26s (~44/s) e si auto-saturava il rate limit per-IP (12.186 risposte 429 in 26h, 96% al minuto :00). Primo giro reale: 574 chiamate, 0 risposte 429. Il minuto :25 e' scelto: :00 era la raffica, :07 e' cron_book (feed 5m di SKH01). 3. quote_status esplicito {ok, no_quote, error} e book_depth NULL su errore mai 0. "Book vuoto" e "chiamata fallita" sono cose diverse: e' per questo che il guasto del 29/07 (50% di quote perse, 38 ore) non produsse alcun segnale. Le righe ereditate restano 'unknown': bite non lo registrava e a posteriori non e' ricostruibile. Battuta di cuore in data/chain_collect/runs.jsonl anche a giro fallito, sorvegliata da monitor_health (1h, max_age 3h): un collettore fermo non produce niente, e il niente si legge come "nessun dato quel giorno". Difetto trovato per strada: due formati ISO nella stessa colonna (92 righe di backfill senza microsecondi). pd.to_datetime senza `format` ne inferisce uno solo e manda gli altri a NaT -> il dropna a valle li toglieva in silenzio, e la serie di contesto perdeva 5 settimane slittando dal 26/03 al 01/05. Corretto con format="ISO8601" e scarto RUMOROSO. Book, pesi, config, strategia INVARIATI. 537 test verdi. Co-Authored-By: Claude Opus 5 (1M context) --- CLAUDE.md | 42 ++- data/chain_collect/runs.jsonl | 1 + .../2026-07-30-assorbimento-cerbero-bite.md | 91 ++++++ ...{fetch_cb_chain.py => certify_cb_chain.py} | 87 +++--- scripts/analysis/import_cb_archive.py | 132 +++++++++ scripts/cron_chain.sh | 24 ++ scripts/live/collect_chain.py | 261 ++++++++++++++++++ scripts/research/cblib.py | 22 +- src/live/monitor_health.py | 5 + tests/test_cb_chain_vrp.py | 2 +- tests/test_collect_chain.py | 139 ++++++++++ tests/test_monitor_health.py | 5 +- 12 files changed, 755 insertions(+), 56 deletions(-) create mode 100644 data/chain_collect/runs.jsonl create mode 100644 docs/diary/2026-07-30-assorbimento-cerbero-bite.md rename scripts/analysis/{fetch_cb_chain.py => certify_cb_chain.py} (67%) create mode 100644 scripts/analysis/import_cb_archive.py create mode 100755 scripts/cron_chain.sh create mode 100644 scripts/live/collect_chain.py create mode 100644 tests/test_collect_chain.py diff --git a/CLAUDE.md b/CLAUDE.md index 7065a7b..974eed9 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -1667,12 +1667,38 @@ df = load_data("BTC", "1h") # OK. load_data("SOL", ...) -> FileNotFoundError ( < 365g reali (es. **AXS 83% sintetico → fuori**), (3) gata i gap vol=0 interni. Universo certificato = **51** (era 52). I **19 major di XS01 hanno 0 backfill → invariati** (strategia live non toccata). Verificato direttamente su cerbero MCP. Diario `2026-06-20-cerbero-backfill-fix.md`. -- **CATENA OPZIONI REALE (cerbero-bite mainnet) — integrata e certificata 2026-07-30.** - `/opt/docker/cerbero-bite` accumula dal 2026-06-09 la catena Deribit **mainnet** (BTC+ETH, - entrambe le ali, scadenze 1g→3 mesi, ORARIA, con `book_depth_top3`). Estrazione+certificazione: - `scripts/analysis/fetch_cb_chain.py` → `data/raw/cb_chain.parquet` (gitignored, ~1.23M righe); - harness `scripts/research/cblib.py`. **È l'unica fonte di prezzi opzioni VERI del progetto** e - serve a sostituire il premio modellato BS-su-DVOL (vedi il f di VRP01, misurato il 30/07). +- **CATENA OPZIONI REALE — RACCOLTA PROPRIA dal 2026-07-30 (cerbero-bite ASSORBITO e dismesso).** + `/opt/docker/cerbero-bite` (progetto separato) accumulava dal 2026-06-09 la catena Deribit + **mainnet** BTC+ETH; **viene eliminato, e la raccolta è passata dentro PythagorasGoal**: + - **raccolta:** `scripts/live/collect_chain.py`, cron **`25 * * * *`** (`scripts/cron_chain.sh`) → + `data/raw/cb_chain/YYYY-MM-DD.parquet`. Entrambe le ali, scadenze ≤95g, OI≥100, ~570 strumenti + a giro, ~3 min. **Minuto :25 scelto, non arbitrario:** :00 era la raffica di bite, :07 è + `cron_book` (feed 5m di SKH01, la cosa che non deve trovare l'IP occupato). + - **archivio ereditato:** `scripts/analysis/import_cb_archive.py` (una-tantum) → + `cb_chain/bite_archive.parquet` (1.23M righe, 2026-05-01+) e `cb_market_snapshots.parquet` + (17.402 righe, 2026-03-26+: DVOL, RV30, funding perp e cross, **dealer net gamma**, gamma flip, + **rischio liquidazioni**, giorni all'evento macro — dati che il progetto non ha altrove). + Snapshot sqlite integrale in `/opt/docker/backups/manual/cerbero-bite-20260730/` (SHA256). + - **certificazione:** `scripts/analysis/certify_cb_chain.py`; **harness** `scripts/research/cblib.py` + (`load_chain()` unisce archivio + raccolta, dedup su `(ts, strumento)`). + - **battuta di cuore:** ogni giro scrive `data/chain_collect/runs.jsonl`, sorvegliato da + `monitor_health` (cadenza 1h, `max_age_h=3`) — **un collettore fermo non produce niente, e il + niente si legge come "nessun dato quel giorno"**. + **TRE DIFETTI DI BITE NON REPLICATI** (misurati il 30/07): (a) **una chiamata per strumento** — + `get_order_book?depth=3` dà già quote+greche+IV+OI+book+underlying, bite ne faceva due con + rischio di disallineamento; (b) **pacing** (token bucket 4/s + backoff) invece della raffica — + il carico non è mai stato il problema (~570 chiamate/ora = 0.16/s **distribuite**), bite le + sparava in ~26s (~44/s) auto-saturandosi il rate limit per-IP; misurato sul nostro giro: + **574 chiamate, 0 risposte 429**; (c) **`quote_status` esplicito** in {`ok`, `no_quote`, `error`} + — "book vuoto" (fatto di mercato) e "chiamata fallita" (fatto di infrastruttura) sono cose + diverse, e `book_depth_top3` è **NULL su errore, mai 0**. ⚠️ Le righe ereditate da bite hanno + `quote_status='unknown'`: bite non registrava il perché, e si dichiara l'ignoranza invece di + inventare uno stato. + **NON assorbito, e perché:** il motore credit-spread ETH (il progetto ha già la regola "niente + short-vol da modello in deploy"), la GUI, kill switch/dead-man/audit (PythagorasGoal ha + `venue_watch`/`edge_watch`/`monitor_health`/`fee_watch`), `dvol_history` (`fetch_dvol.py` ha + storia **più lunga**: 2020+ contro 2026-05), `decisions`/`positions` (59 righe a capitale $52, + 0 posizioni). Test `tests/test_collect_chain.py` (11) + `tests/test_cb_chain_vrp.py` (13). **Perché si MEMORIZZA invece di interrogarla:** una catena opzioni non è ricostruibile a posteriori — Deribit non serve book storici, un'ora non raccolta è persa per sempre, e non c'è un secondo venue da cui recuperarla. @@ -1684,8 +1710,8 @@ df = load_data("BTC", "1h") # OK. load_data("SOL", ...) -> FileNotFoundError ( ogni giro, **96% al minuto :00**, generate dalla spazzata full-chain che si auto-satura) → `bid`/`ask`/`iv`/`delta` NULL e **conteggio righe INVARIATO** (13k/giorno prima e dopo). Tasso di quote vuote per settimana: 0.4·0.7·2.2·1.5·0.5·0.4·0.3 → **22%**; giorno peggiore **51.7% BTC / - 30.6% ETH**. Il fix lato bite (cron `:00`→`:20`) è **scritto ma non attivo** (`strategy.yaml` è - copiato nell'immagine, serve rebuild). **REGOLA: una riga presente non è un dato presente** — + 30.6% ETH**. ✅ **Risolto dal cambio di collettore** (30/07): la raccolta propria è paced e ha + fatto 574 chiamate con **0 risposte 429**. **REGOLA: una riga presente non è un dato presente** — contare ciò che è QUOTATO, non ciò che è SCRITTO (3ª occorrenza dopo `paper_dvolspread` e `fresh_5m`). **REGOLA: un guasto IN CORSO si misura al GIORNO PEGGIORE, non in media** — la prima stesura confrontava "tutto" con "ultimi 7g" e diluiva un guasto di 2 giorni al 13.5%, commettendo diff --git a/data/chain_collect/runs.jsonl b/data/chain_collect/runs.jsonl new file mode 100644 index 0000000..d513534 --- /dev/null +++ b/data/chain_collect/runs.jsonl @@ -0,0 +1 @@ +{"ts": 1785442011000, "righe": 570, "ok": 570, "no_quote": 0, "error": 0, "chiamate": 574, "rate_limited": 0} diff --git a/docs/diary/2026-07-30-assorbimento-cerbero-bite.md b/docs/diary/2026-07-30-assorbimento-cerbero-bite.md new file mode 100644 index 0000000..8661ab5 --- /dev/null +++ b/docs/diary/2026-07-30-assorbimento-cerbero-bite.md @@ -0,0 +1,91 @@ +# 2026-07-30 (2º filone) — cerbero-bite viene eliminato: assorbita la raccolta, non il motore + +**Decisione dell'operatore:** `/opt/docker/cerbero-bite` viene smantellato e il suo lavoro passa a +PythagorasGoal. **Esito:** book, pesi, config e strategia INVARIATI. Cambia solo *chi* raccoglie +la catena opzioni — e cambia in meglio su tre assi misurati. + +## Cosa di bite valeva la pena, e cosa no + +L'unica parte **irreversibile** è il dato: una catena opzioni non si ricostruisce a posteriori +(Deribit non serve book storici, non esiste un secondo venue). Tutto il resto è codice, e il codice +si riscrive. + +| parte | assorbita? | perché | +|---|---|---| +| `option_chain_snapshots` (1.23M righe) | **sì** | ha appena falsificato il *f* di VRP01; irrecuperabile | +| `market_snapshots` (17.402 righe, 26/03+) | **sì** | dealer net gamma, gamma flip, rischio liquidazioni, funding cross: dati che il progetto non ha altrove | +| raccolta continua | **sì, riscritta** | senza, l'archivio si congela e "aspettare il crash" muore | +| `dvol_history` | no | `fetch_dvol.py` ha storia **più lunga** (2020+ vs 2026-05): copiarla sarebbe una seconda copia peggiore | +| motore credit-spread ETH | no | il progetto ha già la regola *niente short-vol da modello in deploy*; e il conto era a $52 contro un minimo di $720 | +| GUI, kill switch, dead-man, audit chain | no | PythagorasGoal ha già `venue_watch`, `edge_watch`, `monitor_health`, `fee_watch` | +| `decisions` / `positions` | no | 59 valutazioni d'ingresso, 0 posizioni | + +Prima di toccare qualsiasi cosa: snapshot `VACUUM INTO` del DB (461 MB) + `audit.log` + config in +`/opt/docker/backups/manual/cerbero-bite-20260730/`, con SHA256. + +## Il collettore nuovo — tre difetti di bite non replicati + +`scripts/live/collect_chain.py`, cron **`25 * * * *`**. + +**1. Una chiamata per strumento invece di due.** `public/get_order_book?depth=3` restituisce già +quote, greche, IV, open interest, volume, book **e** `underlying_price`. Bite chiamava ticker e +orderbook separatamente: doppio costo, e i due potevano riferirsi a istanti diversi. +Un `get_book_summary_by_currency` iniziale dà l'OI di tutta la catena in **una** chiamata e +prefiltra sotto soglia: 551 → **~290 chiamate per asset**. + +**2. Pacing invece di raffica.** ⚠️ **Il carico non è mai stato il problema.** ~570 chiamate/ora +sono **0.16/s** se distribuite; bite le sparava in ~26 secondi (**~44/s**) e si auto-saturava il +rate limit per-IP — 12.186 risposte 429 in 26 ore, il 96% nel minuto `:00`. Token bucket a 4/s + +backoff: primo giro reale **574 chiamate, 0 risposte 429**. + +Il minuto `:25` è una scelta, non un default: `:00` era la raffica di bite, `:07` è `cron_book` +(da lì passa il feed 5m di SKH01, la cosa che non deve trovare l'IP occupato). + +**3. Stato esplicito della quota.** `quote_status` ∈ {`ok`, `no_quote`, `error`}: +- `no_quote` = il venue ha risposto, il book è vuoto → **fatto di mercato** +- `error` = la chiamata è fallita → **fatto di infrastruttura** + +Bite li faceva collassare entrambi su "riga con bid/ask NULL", ed è per questo che il guasto del +29/07 (50% di quote perse per 38 ore) non ha prodotto **nessun** segnale: il conteggio righe non +cambiava. Per lo stesso motivo `book_depth_top3` ora è **NULL su errore, mai 0** — bite scriveva 0, +e "chiamata fallita" diventava indistinguibile da "book vuoto". + +⚠️ Le righe ereditate portano `quote_status='unknown'`. Bite non registrava il perché e **a +posteriori non è ricostruibile**: si dichiara l'ignoranza invece di inventare uno stato. + +## Il silenzio, di nuovo + +Un collettore fermo non produce niente, e il niente si legge come "nessun dato quel giorno" invece +che come "raccolta rotta" — su una serie irrecuperabile è il modo più caro di sbagliare. Ogni giro +scrive una riga in `data/chain_collect/runs.jsonl`, **anche quando fallisce**, e +`monitor_health` la sorveglia con cadenza 1h e `max_age_h=3` (due giri persi). È la stessa lezione +di `paper_dvolspread` e di `fresh_5m`, applicata prima che serva invece che dopo. + +## Un difetto trovato per strada: due formati ISO nella stessa colonna + +L'import del contesto restituiva **17.310 righe dal 2026-05-01** contro le 17.402 dal **2026-03-26** +del database. 92 righe di backfill sono scritte senza microsecondi +(`2026-03-26T12:00:00+00:00`), le altre con (`2026-05-01T15:45:00.062918+00:00`). +`pd.to_datetime` senza `format` **inferisce un formato solo** dal primo elemento e manda gli altri +a `NaT`, che il `dropna` a valle rimuoveva **in silenzio**: la serie perdeva i suoi 5 settimane più +vecchi e la data d'inizio slittava di un mese senza un messaggio. + +Corretto con `format="ISO8601"` **e** con uno scarto rumoroso (`_drop_unparsed` stampa quante righe +cadono e perché). **REGOLA: `dropna` dopo un parsing è un rilevatore di difetti travestito da +pulizia — se toglie righe deve dirlo.** Un import silenzioso non è un import pulito, è un import +che non sai se ha funzionato. + +## Cosa resta da fare fuori dal repo + +Il container di bite è **ancora acceso** mentre scrivo: la sovrapposizione fra le due raccolte è +voluta (nessun buco alla consegna). Quando verrà spento, l'unica cosa che si ferma è la raffica +al minuto `:00`. + +## File + +- `scripts/live/collect_chain.py` + `scripts/cron_chain.sh` — la raccolta (cron `25 * * * *`) +- `scripts/analysis/import_cb_archive.py` — import una-tantum dell'archivio +- `scripts/analysis/certify_cb_chain.py` — certificazione dello store (era `fetch_cb_chain.py`) +- `scripts/research/cblib.py` — `load_chain()` ora unisce archivio + raccolta +- `src/live/monitor_health.py` — nuovo spec `collect_chain` +- `tests/test_collect_chain.py` (11 casi) diff --git a/scripts/analysis/fetch_cb_chain.py b/scripts/analysis/certify_cb_chain.py similarity index 67% rename from scripts/analysis/fetch_cb_chain.py rename to scripts/analysis/certify_cb_chain.py index 8f1049a..1630489 100644 --- a/scripts/analysis/fetch_cb_chain.py +++ b/scripts/analysis/certify_cb_chain.py @@ -1,27 +1,25 @@ -"""FETCH + CERTIFY della catena opzioni REALE accumulata da cerbero-bite. +"""CERTIFICAZIONE della catena opzioni in `data/raw/cb_chain/` (archivio bite + raccolta propria). -Perche' esiste: VRP01 prezza le sue gambe con BS su DVOL ATM (`src/portfolio/sleeves.VRP_CFG`, -f=1.0). L'unico modo per sapere se quel prezzo e' quello del mercato e' confrontarlo con quote -vere; cerbero-bite (/opt/docker/cerbero-bite) accumula la catena Deribit mainnet ora per ora, -entrambe le ali, scadenze 1g..3mesi. Una catena opzioni NON e' ricostruibile a posteriori -(Deribit non serve book storici) -> il dato va estratto e conservato, non richiesto quando serve. - -Come `fetch_hyperliquid.py`: estrae, CERTIFICA, e scrive solo se la certificazione ha un esito -dichiarato. Le soglie sono qui sotto e sono decisioni, non default. +Storia: nato il 30/07 come `fetch_cb_chain.py` per estrarre la catena dal container cerbero-bite. +Con lo smantellamento di bite l'estrazione e' diventata un import una-tantum +(`import_cb_archive.py`) e la raccolta e' passata a `scripts/live/collect_chain.py`; qui resta +cio' che serve per sempre: **il giudizio sul dato**. CERTIFICAZIONE — quattro difetti, ognuno con la sua diagnostica: - 1. QUOTE VUOTE bid/ask NULL con lo strumento presente: il collettore persiste la riga anche - quando il ticker fallisce (rate-limit) -> il CONTEGGIO RIGHE resta identico e qualunque - controllo di copertura basato sul numero di righe dice "tutto bene". E' il difetto che il - 29/07 ha portato il tasso da ~0.4% a ~50% senza che nulla lo segnalasse. + 1. QUOTE VUOTE la riga c'e' ma la quota no. E' il difetto che il 29/07 ha portato il tasso da + ~0.4% a ~50% per 38 ore senza un segnale: il CONTEGGIO RIGHE resta identico, quindi ogni + controllo di copertura basato sulle righe dice "tutto bene". Sulle righe raccolte da + `collect_chain.py` si legge anche il PERCHE' (`quote_status`: no_quote vs error); sulle righe + ereditate da bite no (`unknown`), perche' bite non lo registrava. 2. BOOK INCROCIATO bid > ask (quote incoerenti). - 3. MONOTONIA a parita' di scadenza/istante il premio di una put deve essere non-decrescente - nello strike; una violazione e' una quota stantia o corrotta. - 4. PROFONDITA' ZERO book_depth_top3 == 0 e' AMBIGUO by design (chiamata fallita e book - davvero vuoto danno lo stesso valore) -> si riporta, non si ripara. + 3. MONOTONIA a parita' di scadenza/istante il premio di una put e' non-decrescente nello + strike: e' il payoff, non un modello -> una violazione e' un difetto di dato. + 4. PROFONDITA' ZERO nell'archivio bite `book_depth_top3 == 0` e' AMBIGUO by design (chiamata + fallita e book vuoto danno lo stesso valore) -> si riporta, non si ripara. Nelle righe nuove + l'ambiguita' non esiste: su errore la profondita' e' NULL. - uv run python scripts/analysis/fetch_cb_chain.py # docker cp dal container - uv run python scripts/analysis/fetch_cb_chain.py --db /percorso/state.sqlite + uv run python scripts/analysis/certify_cb_chain.py + uv run python scripts/analysis/certify_cb_chain.py --db /percorso/state.sqlite # legacy """ from __future__ import annotations @@ -38,8 +36,8 @@ PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) RAW = PROJECT_ROOT / "data" / "raw" -OUT = RAW / "cb_chain.parquet" -CONTAINER = "cerbero-bite-cerbero-bite-1" +STORE = RAW / "cb_chain" +CONTAINER = "cerbero-bite-cerbero-bite-1" # legacy: il container non esiste piu' DB_IN_CONTAINER = "/app/data/state.sqlite" # --- soglie di certificazione (decise, non default) --- @@ -70,10 +68,16 @@ def read_chain(db: Path) -> pd.DataFrame: con.close() for c in ("strike", "bid", "ask", "mid", "iv", "delta", "gamma", "theta", "vega"): df[c] = pd.to_numeric(df[c], errors="coerce") - df["ts"] = pd.to_datetime(df["timestamp"], utc=True, errors="coerce") - df["exp"] = pd.to_datetime(df["expiry"], utc=True, errors="coerce") + # formati ISO MISTI nella stessa colonna (con e senza microsecondi): senza `format` pandas + # ne inferisce uno solo e manda gli altri a NaT, che il dropna sotto toglierebbe in SILENZIO. + df["ts"] = pd.to_datetime(df["timestamp"], utc=True, format="ISO8601", errors="coerce") + df["exp"] = pd.to_datetime(df["expiry"], utc=True, format="ISO8601", errors="coerce") df["dte"] = (df["exp"] - df["ts"]).dt.total_seconds() / 86400.0 - return df.dropna(subset=["ts", "exp", "strike"]).drop(columns=["timestamp", "expiry"]) + n0 = len(df) + df = df.dropna(subset=["ts", "exp", "strike"]).drop(columns=["timestamp", "expiry"]) + if len(df) != n0: + print(f" ATTENZIONE: {n0 - len(df)} righe scartate (timestamp illeggibile) su {n0}") + return df # ------------------------------------------------------------------ certificazione @@ -161,20 +165,21 @@ def certify(df: pd.DataFrame) -> list[dict]: def main() -> int: ap = argparse.ArgumentParser() - ap.add_argument("--db", type=Path, default=None, help="state.sqlite (default: docker cp)") - ap.add_argument("--out", type=Path, default=OUT) + ap.add_argument("--db", type=Path, default=None, + help="legacy: certifica direttamente uno state.sqlite di bite invece dello store") args = ap.parse_args() - tmp = None - db = args.db - if db is None: - tmp = Path(tempfile.mkdtemp()) / "state.sqlite" - print(f" docker cp {CONTAINER}:{DB_IN_CONTAINER} ...") - db = _pull_db(tmp) - df = read_chain(db) + if args.db is not None: + df = read_chain(args.db) + titolo = f"catena da {args.db}" + else: + sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research")) + from cblib import load_chain + df = load_chain() + titolo = f"store {STORE}" print("=" * 100) - print(" CATENA OPZIONI cerbero-bite — certificazione") + print(f" CATENA OPZIONI — certificazione ({titolo})") print("=" * 100) rep = certify(df) for r in rep: @@ -184,17 +189,17 @@ def main() -> int: f"| ultimi {RECENT_DAYS}g {100*r[f'vuote_{RECENT_DAYS}g']:.1f}% " f"| GIORNO PEGGIORE {100*r['vuote_giorno_peggiore']:.1f}% ({r['giorno_peggiore']})") print(f" book incrociato {100*r['incrociate']:.2f}% | premio non monotono nello strike " - f"{100*r['non_monotone']:.2f}% | depth==0 {100*r['depth_zero']:.1f}% (ambiguo by design)") + f"{100*r['non_monotone']:.2f}% | depth==0 {100*r['depth_zero']:.1f}% (ambiguo solo " + f"nell'archivio bite)") print(f" STATUS: {r['status']}") - args.out.parent.mkdir(parents=True, exist_ok=True) - df.to_parquet(args.out, index=False) - print(f"\n scritto {args.out} ({len(df):,} righe)") + if "quote_status" in df.columns: + print("\n stato dichiarato della quota (solo le righe raccolte da noi lo hanno):") + for k, v in df["quote_status"].value_counts().items(): + print(f" {k:9s} {v:>9,}") if any(r["status"] != "OK" for r in rep): - print(" ATTENZIONE: almeno un asset non e' OK — le quote vuote NON sono righe mancanti,") + print("\n ATTENZIONE: almeno un asset non e' OK — le quote vuote NON sono righe mancanti,") print(" il conteggio righe resta identico. Ogni misura su questa finestra va pesata.") - if tmp is not None: - tmp.unlink(missing_ok=True) return 0 diff --git a/scripts/analysis/import_cb_archive.py b/scripts/analysis/import_cb_archive.py new file mode 100644 index 0000000..fcf41c5 --- /dev/null +++ b/scripts/analysis/import_cb_archive.py @@ -0,0 +1,132 @@ +"""IMPORT dell'archivio cerbero-bite dentro PythagorasGoal — una volta sola, prima della sua +eliminazione (2026-07-30). + +cerbero-bite viene smantellato; la sua raccolta continua in `scripts/live/collect_chain.py`. Questo +script prende cio' che NON e' ricostruibile e lo porta dentro: + + * `option_chain_snapshots` -> data/raw/cb_chain/bite_archive.parquet (1.23M righe, 2026-05-01+) + * `market_snapshots` -> data/raw/cb_market_snapshots.parquet (17k righe, 2026-03-26+: + spot, DVOL, RV30, IV-RV, funding perp e cross-exchange, dealer net gamma, gamma flip level, + OI delta 4h, rischio liquidazioni long/short, giorni all'evento macro) + +NON importato e perche': + * `dvol_history` — il progetto ha gia' `fetch_dvol.py` con storia PIU' LUNGA (2020+ contro + 2026-05): reimportarlo aggiungerebbe una seconda copia peggiore della prima. + * `decisions` / `positions` — 59 righe di valutazioni d'ingresso a capitale $52, 0 posizioni: + e' il log di una strategia che il progetto ha gia' deciso di non deployare (regola + "niente short-vol da modello in deploy"). + * `audit.log` — conservato come file nel backup, non e' una serie storica. + +⚠️ LO STATO DELLA QUOTA NELL'ARCHIVIO E' `unknown`, NON `ok`. Bite scriveva la riga anche quando +la chiamata falliva, con bid/ask NULL, senza registrare il perche': "book vuoto" e "chiamata +fallita" sono indistinguibili a posteriori. Il collettore nuovo li separa (`no_quote` / `error`); +per lo storico si dichiara l'ignoranza invece di inventare uno stato. Chi misura sull'archivio +deve sapere che le righe senza quota sono ~1.5% fino al 28/07 e ~50% dal 29/07 (guasto di +rate-limit, vedi diario 2026-07-30). + + uv run python scripts/analysis/import_cb_archive.py --db /opt/docker/backups/manual/cerbero-bite-20260730/state.sqlite +""" +from __future__ import annotations + +import argparse +import sqlite3 +import sys +from pathlib import Path + +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +RAW = PROJECT_ROOT / "data" / "raw" +STORE = RAW / "cb_chain" +ARCHIVE = STORE / "bite_archive.parquet" +MARKET = RAW / "cb_market_snapshots.parquet" +DEFAULT_DB = Path("/opt/docker/backups/manual/cerbero-bite-20260730/state.sqlite") + + +def parse_ts(s: pd.Series) -> pd.Series: + """Timestamp ISO con formati MISTI nella stessa colonna. + + bite scrive sia '2026-03-26T12:00:00+00:00' (backfill) sia + '2026-05-01T15:45:00.062918+00:00' (runtime). `pd.to_datetime` senza `format` inferisce UN + formato dal primo elemento e manda gli altri a NaT: con un `dropna` a valle spariscono in + silenzio (qui: 92 righe, e la data d'inizio della serie slittava dal 26/03 al 01/05). + `format="ISO8601"` accetta entrambe le varianti. + """ + return pd.to_datetime(s, utc=True, format="ISO8601", errors="coerce") + + +def _drop_unparsed(df: pd.DataFrame, cols: list[str], etichetta: str) -> pd.DataFrame: + """Scarta le righe con timestamp illeggibile DICHIARANDOLO. Uno scarto silenzioso in un + import e' indistinguibile da un dato che non e' mai esistito.""" + before = len(df) + out = df.dropna(subset=cols) + if len(out) != before: + print(f" ATTENZIONE [{etichetta}]: {before - len(out)} righe scartate " + f"(timestamp illeggibile) su {before}") + return out + + +def import_chain(db: Path) -> pd.DataFrame: + con = sqlite3.connect(f"file:{db}?mode=ro", uri=True) + try: + df = pd.read_sql_query( + """select timestamp, asset, instrument_name, strike, expiry, option_type, + bid, ask, mid, iv, delta, gamma, theta, vega, + open_interest, volume_24h, book_depth_top3, source + from option_chain_snapshots""", con) + finally: + con.close() + for c in ("strike", "bid", "ask", "mid", "iv", "delta", "gamma", "theta", "vega", + "open_interest", "volume_24h", "book_depth_top3"): + df[c] = pd.to_numeric(df[c], errors="coerce") + df["ts"] = parse_ts(df["timestamp"]) + df["exp"] = parse_ts(df["expiry"]) + df = _drop_unparsed(df.drop(columns=["timestamp", "expiry"]), ["ts", "exp", "strike"], "catena") + # bite non distingue "book vuoto" da "chiamata fallita": si dichiara, non si indovina. + df["quote_status"] = df["bid"].notna().map({True: "ok", False: "unknown"}) + df["underlying_price"] = pd.NA + df["index_price"] = pd.NA + df["source"] = "bite:" + df["source"].astype(str) + return df + + +def import_market(db: Path) -> pd.DataFrame: + con = sqlite3.connect(f"file:{db}?mode=ro", uri=True) + try: + df = pd.read_sql_query("select * from market_snapshots", con) + finally: + con.close() + df["ts"] = parse_ts(df["timestamp"]) + return _drop_unparsed(df.drop(columns=["timestamp"]), ["ts"], "contesto") + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--db", type=Path, default=DEFAULT_DB) + args = ap.parse_args() + if not args.db.exists(): + print(f" DB assente: {args.db}") + return 1 + + STORE.mkdir(parents=True, exist_ok=True) + ch = import_chain(args.db) + ch.to_parquet(ARCHIVE, index=False) + st = ch["quote_status"].value_counts().to_dict() + print(f" catena -> {ARCHIVE}") + print(f" {len(ch):,} righe {ch['ts'].min():%Y-%m-%d} -> {ch['ts'].max():%Y-%m-%d} " + f"| ok {st.get('ok', 0):,} unknown {st.get('unknown', 0):,} " + f"({100*st.get('unknown', 0)/len(ch):.1f}%)") + + mk = import_market(args.db) + mk.to_parquet(MARKET, index=False) + print(f" contesto -> {MARKET}") + print(f" {len(mk):,} righe {mk['ts'].min():%Y-%m-%d} -> {mk['ts'].max():%Y-%m-%d} " + f"| colonne: {', '.join(c for c in mk.columns if c not in ('ts', 'asset'))[:110]}...") + print("\n NON importati (motivo nel docstring): dvol_history, decisions, positions, audit.log") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/cron_chain.sh b/scripts/cron_chain.sh new file mode 100755 index 0000000..48b304d --- /dev/null +++ b/scripts/cron_chain.sh @@ -0,0 +1,24 @@ +#!/bin/bash +# RACCOLTA CATENA OPZIONI Deribit — oraria. Successore di cerbero-bite (smantellato il 2026-07-30). +# +# Perche' e' un cron e non un container: la raccolta e' un giro di ~3 minuti ogni ora, senza stato +# fra un giro e l'altro. cerbero-bite girava come servizio con scheduler interno; l'unica parte che +# serviva davvero era questa. +# +# MINUTO :25, e la scelta NON e' arbitraria: +# :00 cerbero-bite sparava la sua spazzata full-chain (~44 chiamate/s) e saturava il rate limit +# Deribit per-IP — 12.186 risposte 429 in 26 ore, il 96% nel minuto tondo. +# :07 cron_book.sh (esecuzione del book live). Il feed 5m di SKH01 passa da li': e' la cosa che +# NON deve trovare l'IP occupato. +# :25 nessun altro job. Il giro dura ~3 minuti a 4 chiamate/s, quindi finisce ben prima del :30. +# +# Il pacing e' nello script (token bucket + backoff sul 429): un giro lento non costa niente, +# una raffica costa il dato di tutti gli altri. +export PATH="/home/adriano/.local/bin:$PATH" +cd /opt/docker/PythagorasGoal || exit 1 +mkdir -p logs +{ + echo "===== $(date -u '+%Y-%m-%dT%H:%M:%SZ') cron_chain =====" + uv run python scripts/live/collect_chain.py + echo "===== done $(date -u '+%H:%M:%SZ') =====" +} >> logs/cron_chain.log 2>&1 diff --git a/scripts/live/collect_chain.py b/scripts/live/collect_chain.py new file mode 100644 index 0000000..a8ed99e --- /dev/null +++ b/scripts/live/collect_chain.py @@ -0,0 +1,261 @@ +"""COLLETTORE CATENA OPZIONI — successore di cerbero-bite, dentro PythagorasGoal (2026-07-30). + +Perche' esiste: cerbero-bite viene eliminato, e con esso si fermerebbe l'unica raccolta di prezzi +opzioni REALI del progetto. Una catena non e' ricostruibile a posteriori (Deribit non serve book +storici, non c'e' un secondo venue) -> l'ora non raccolta e' persa per sempre. Il valore +dell'archivio sta negli eventi RARI: e' la rete stesa in attesa del regime di vol alta che +promuovera' o uccidera' VRP01 (criterio dichiarato il 19/06, gate IV-rank>0.30 mai attivo nella +finestra raccolta finora). + +TRE DIFFERENZE DA BITE, tutte misurate il 30/07 e tutte deliberate: + +1. UNA CHIAMATA PER STRUMENTO, non due. `public/get_order_book?depth=3` restituisce gia' quote, + greche, IV, open interest, volume, book E underlying_price. Bite chiamava ticker + orderbook + separatamente: doppio costo e possibilita' di disallineamento fra i due (quote di un istante, + book di un altro). + +2. PACING, non raffica. Il carico non e' mai stato il problema: ~570 chiamate/ora = 0.16/s se + distribuite. Bite le sparava in ~26 secondi (~44/s) e si auto-saturava il rate limit per-IP + (12.186 risposte 429 in 26 ore, 96% nel minuto :00), con l'effetto collaterale di disturbare + il feed 5m del book live sulla stessa VPS. Qui: token bucket a `--rps` (default 4/s, ~2.5 + minuti per giro) + backoff sul 429. Un giro lento non costa nulla; una raffica costa il dato. + +3. STATO ESPLICITO DELLA QUOTA. Bite persisteva la riga anche quando la chiamata falliva, con + bid/ask NULL: il conteggio righe restava identico e nessun controllo di copertura se ne + accorgeva (il 29/07 il 50% delle quote e' diventato vuoto per 38 ore senza un segnale). Qui + ogni riga porta `quote_status` in {ok, no_quote, error}: + ok = il venue ha risposto e c'e' almeno un lato del book + no_quote = il venue ha risposto e il book e' vuoto da entrambi i lati (fatto di mercato) + error = la chiamata e' fallita (fatto di infrastruttura) + Sono cose diverse e non vanno mai confuse. Per lo stesso motivo `book_depth_top3` e' NULL su + errore, MAI 0: bite scriveva 0 e "chiamata fallita" diventava indistinguibile da "book vuoto". + + uv run python scripts/live/collect_chain.py # un giro, entrambi gli asset + uv run python scripts/live/collect_chain.py --assets ETH --rps 8 + uv run python scripts/live/collect_chain.py --dry-run # non scrive +""" +from __future__ import annotations + +import argparse +import json +import sys +import time +from dataclasses import dataclass, field +from datetime import UTC, datetime +from pathlib import Path + +import pandas as pd +import requests + +PROJECT_ROOT = Path(__file__).resolve().parents[2] +sys.path.insert(0, str(PROJECT_ROOT)) + +API = "https://www.deribit.com/api/v2/public" +STORE = PROJECT_ROOT / "data" / "raw" / "cb_chain" +ASSETS = ("BTC", "ETH") +EXPIRY_MAX_DAYS = 95 # 1g..3mesi, come la finestra di bite (continuita' della serie) +OI_MIN = 100.0 # come bite: sotto questa soglia lo strumento e' rumore +DEFAULT_RPS = 4.0 +TIMEOUT = 15 + + +@dataclass +class Budget: + """Token bucket + contabilita' del giro. Il 429 non e' un dettaglio: e' la cosa da non fare.""" + rps: float + _next: float = 0.0 + calls: int = 0 + errors: int = 0 + rate_limited: int = 0 + waited_s: float = 0.0 + err_samples: list[str] = field(default_factory=list) + + def wait(self) -> None: + now = time.monotonic() + if now < self._next: + time.sleep(self._next - now) + self.waited_s += self._next - now + self._next = max(now, self._next) + 1.0 / self.rps + + def note_error(self, msg: str) -> None: + self.errors += 1 + if len(self.err_samples) < 5: + self.err_samples.append(msg[:160]) + + +def _get(path: str, params: dict, budget: Budget, tries: int = 3) -> dict | None: + """GET con pacing e backoff. Ritorna None se la chiamata non e' andata a buon fine.""" + for k in range(tries): + budget.wait() + budget.calls += 1 + try: + r = requests.get(f"{API}/{path}", params=params, timeout=TIMEOUT) + except Exception as exc: # rete: si registra QUI, non dopo + budget.note_error(f"{path}: {type(exc).__name__}: {exc}") + time.sleep(1.5 * (k + 1)) + continue + if r.status_code == 429: + budget.rate_limited += 1 + time.sleep(2.0 * (k + 1)) # backoff: il venue ha detto di rallentare + continue + if r.status_code != 200: + budget.note_error(f"{path}: HTTP {r.status_code}") + time.sleep(1.0 * (k + 1)) + continue + try: + return r.json()["result"] + except Exception as exc: + budget.note_error(f"{path}: payload illeggibile: {exc}") + return None + return None + + +def instruments(asset: str, budget: Budget, now: datetime) -> list[dict]: + res = _get("get_instruments", {"currency": asset, "kind": "option", "expired": "false"}, budget) + if not res: + return [] + horizon = now.timestamp() * 1000 + EXPIRY_MAX_DAYS * 86400_000 + return [i for i in res if i.get("expiration_timestamp", 0) <= horizon] + + +def _depth_top3(side: list) -> float | None: + if side is None: + return None + return float(sum(row[1] for row in side[:3] if isinstance(row, (list, tuple)) and len(row) >= 2)) + + +def snapshot_row(inst: dict, ob: dict | None, ts: datetime) -> dict: + """Una riga per strumento — SEMPRE, ma con lo stato della quota dichiarato.""" + name = inst["instrument_name"] + base = { + "ts": ts, "asset": inst["base_currency"], "instrument_name": name, + "strike": float(inst["strike"]), "option_type": "P" if inst["option_type"] == "put" else "C", + "exp": pd.Timestamp(inst["expiration_timestamp"], unit="ms", tz="UTC"), + "bid": None, "ask": None, "mid": None, "iv": None, + "delta": None, "gamma": None, "theta": None, "vega": None, + "open_interest": None, "volume_24h": None, "book_depth_top3": None, + "underlying_price": None, "index_price": None, + "quote_status": "error", "source": "pyg", + } + if ob is None: + return base # errore: depth resta NULL, mai 0 + g = ob.get("greeks") or {} + stats = ob.get("stats") or {} + bid, ask = ob.get("best_bid_price"), ob.get("best_ask_price") + bid = float(bid) if bid else None # Deribit manda 0.0 per "nessun lato" + ask = float(ask) if ask else None + db, da = _depth_top3(ob.get("bids")), _depth_top3(ob.get("asks")) + base.update({ + "bid": bid, "ask": ask, + "mid": (bid + ask) / 2 if (bid is not None and ask is not None) else None, + "iv": float(ob["mark_iv"]) if ob.get("mark_iv") is not None else None, + "delta": g.get("delta"), "gamma": g.get("gamma"), + "theta": g.get("theta"), "vega": g.get("vega"), + "open_interest": ob.get("open_interest"), "volume_24h": stats.get("volume"), + "book_depth_top3": (db or 0.0) + (da or 0.0), + "underlying_price": ob.get("underlying_price"), "index_price": ob.get("index_price"), + # il venue ha risposto: se non c'e' nessun lato e' un fatto di MERCATO, non un guasto + "quote_status": "ok" if (bid is not None or ask is not None) else "no_quote", + }) + return base + + +def open_interest_map(asset: str, budget: Budget) -> dict[str, float] | None: + """OI di TUTTA la catena in UNA chiamata (`get_book_summary_by_currency`). + + Serve a non spendere una chiamata per scoprire che uno strumento e' sotto soglia: il prefiltro + dimezza il giro (551 -> ~300 chiamate su ETH). None = la chiamata e' fallita, e allora si + raccoglie TUTTO invece di filtrare su un dato che non si ha: un filtro su dati mancanti + scarterebbe strumenti buoni fingendo che fossero illiquidi. + """ + res = _get("get_book_summary_by_currency", {"currency": asset, "kind": "option"}, budget) + if not res: + return None + return {r["instrument_name"]: float(r.get("open_interest") or 0.0) for r in res} + + +def sweep(asset: str, budget: Budget, now: datetime) -> pd.DataFrame: + insts = [i for i in instruments(asset, budget, now) if float(i.get("strike") or 0) > 0] + oi = open_interest_map(asset, budget) + if oi is not None: + insts = [i for i in insts if oi.get(i["instrument_name"], 0.0) >= OI_MIN] + rows = [] + for inst in insts: + ob = _get("get_order_book", {"instrument_name": inst["instrument_name"], "depth": 3}, budget) + rows.append(snapshot_row(inst, ob, now)) + return pd.DataFrame(rows) + + +def heartbeat(now: datetime, df: pd.DataFrame, budget: Budget) -> None: + """Una riga per giro in `data/chain_collect/runs.jsonl`, letta da `monitor_health`. + + Serve perche' un collettore fermo non produce NIENTE, e il niente si legge come "nessun dato + quel giorno" invece che come "raccolta rotta" — con una serie irrecuperabile e' il modo piu' + caro di sbagliare. La battuta di cuore esiste anche quando il giro fallisce. + """ + d = PROJECT_ROOT / "data" / "chain_collect" + d.mkdir(parents=True, exist_ok=True) + st = df["quote_status"].value_counts().to_dict() if not df.empty else {} + riga = {"ts": int(now.timestamp() * 1000), "righe": int(len(df)), + "ok": int(st.get("ok", 0)), "no_quote": int(st.get("no_quote", 0)), + "error": int(st.get("error", 0)), "chiamate": budget.calls, + "rate_limited": budget.rate_limited} + with (d / "runs.jsonl").open("a") as fh: + fh.write(json.dumps(riga) + "\n") + + +def write(df: pd.DataFrame, day: datetime) -> Path: + """Un parquet per giorno: append-friendly e nessun file che cresce senza fine.""" + STORE.mkdir(parents=True, exist_ok=True) + p = STORE / f"{day:%Y-%m-%d}.parquet" + if p.exists(): + df = pd.concat([pd.read_parquet(p), df], ignore_index=True) + df = df.drop_duplicates(subset=["ts", "instrument_name"], keep="last") + df.to_parquet(p, index=False) + return p + + +def main() -> int: + ap = argparse.ArgumentParser() + ap.add_argument("--assets", nargs="+", default=list(ASSETS)) + ap.add_argument("--rps", type=float, default=DEFAULT_RPS, help="chiamate/secondo (pacing)") + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + now = datetime.now(UTC).replace(microsecond=0) + budget = Budget(rps=args.rps) + t0 = time.monotonic() + frames = [] + for a in args.assets: + df = sweep(a, budget, now) + frames.append(df) + if df.empty: + print(f" {a}: NESSUNA riga — il giro e' fallito, non e' un mercato vuoto") + continue + st = df["quote_status"].value_counts().to_dict() + print(f" {a}: {len(df):4d} strumenti | ok {st.get('ok', 0)} | " + f"no_quote {st.get('no_quote', 0)} | error {st.get('error', 0)}") + out = pd.concat(frames, ignore_index=True) if frames else pd.DataFrame(columns=["quote_status"]) + dur = time.monotonic() - t0 + print(f" {budget.calls} chiamate in {dur:.0f}s ({budget.calls/max(dur,1):.1f}/s) | " + f"429: {budget.rate_limited} | errori: {budget.errors}") + for e in budget.err_samples: + print(f" errore: {e}") + if not args.dry_run: + heartbeat(now, out, budget) # anche a giro fallito: il silenzio non e' un dato + if out.empty: + print(" NIENTE DA SCRIVERE — giro fallito") + return 1 + bad = float((out["quote_status"] == "error").mean()) + if bad > 0.20: + print(f" ATTENZIONE: {100*bad:.0f}% di quote in ERRORE — la riga c'e' ma il dato no.") + if args.dry_run: + print(" --dry-run: non scrivo") + return 0 + p = write(out, now) + print(f" scritto {p} ({len(out)} righe)") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/research/cblib.py b/scripts/research/cblib.py index 4a09d5d..b9b6d47 100644 --- a/scripts/research/cblib.py +++ b/scripts/research/cblib.py @@ -27,7 +27,7 @@ from scipy.stats import norm ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) RAW = ROOT / "data" / "raw" -CHAIN = RAW / "cb_chain.parquet" +STORE = RAW / "cb_chain" # bite_archive.parquet (una-tantum) + YYYY-MM-DD.parquet (nostri) # finestra "settimanale" di VRP01 (tenor_d=7): si accettano 4..10 DTE come in options_vrp_calibrate DTE_LO, DTE_HI = 4.0, 10.0 @@ -38,14 +38,26 @@ SHORT_DELTA, LONG_DELTA = -0.28, -0.10 @lru_cache(maxsize=4) def load_chain() -> pd.DataFrame: - if not CHAIN.exists(): + """Tutta la catena: archivio ereditato da cerbero-bite + raccolta propria, in una serie sola. + + Le due fonti restano distinguibili dalla colonna `source` (`bite:live`, `bite:research`, `pyg`) + e da `quote_status` (`unknown` per l'archivio, che non registrava il perche' di una quota + assente). La deduplica su (ts, instrument_name) tiene l'ULTIMA occorrenza: se un giorno e' + coperto da entrambe le fonti vince la piu' recente, cioe' la nostra. + """ + if not STORE.exists(): raise FileNotFoundError( - f"{CHAIN} assente — estrai prima con scripts/analysis/fetch_cb_chain.py" + f"{STORE} assente — importa l'archivio con scripts/analysis/import_cb_archive.py " + f"e raccogli con scripts/live/collect_chain.py" ) - df = pd.read_parquet(CHAIN) + parts = sorted(STORE.glob("*.parquet")) + if not parts: + raise FileNotFoundError(f"{STORE} vuoto") + df = pd.concat([pd.read_parquet(p) for p in parts], ignore_index=True) df["ts"] = pd.to_datetime(df["ts"], utc=True) df["exp"] = pd.to_datetime(df["exp"], utc=True) - return df + df["dte"] = (df["exp"] - df["ts"]).dt.total_seconds() / 86400.0 + return df.drop_duplicates(subset=["ts", "instrument_name"], keep="last").reset_index(drop=True) def puts(asset: str, df: pd.DataFrame | None = None) -> pd.DataFrame: diff --git a/src/live/monitor_health.py b/src/live/monitor_health.py index bf42b25..947c42b 100644 --- a/src/live/monitor_health.py +++ b/src/live/monitor_health.py @@ -74,6 +74,11 @@ MONITORS: tuple[MonitorSpec, ...] = ( # 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"), ) diff --git a/tests/test_cb_chain_vrp.py b/tests/test_cb_chain_vrp.py index 85e11f6..f3c77f6 100644 --- a/tests/test_cb_chain_vrp.py +++ b/tests/test_cb_chain_vrp.py @@ -24,7 +24,7 @@ sys.path.insert(0, str(ROOT / "scripts" / "research")) sys.path.insert(0, str(ROOT / "scripts" / "analysis")) from cblib import bs_put, f_factors, pick_legs # noqa: E402 -from fetch_cb_chain import ( # noqa: E402 +from certify_cb_chain import ( # noqa: E402 crossed_rate, hollow_rate, monotonicity_violations, worst_day_hollow, ) diff --git a/tests/test_collect_chain.py b/tests/test_collect_chain.py new file mode 100644 index 0000000..5384f4b --- /dev/null +++ b/tests/test_collect_chain.py @@ -0,0 +1,139 @@ +"""Test del collettore catena opzioni (successore di cerbero-bite). + +Il collettore esiste per NON ripetere tre difetti misurati su bite il 2026-07-30. I test +sorvegliano esattamente quei tre, perche' sono difetti SILENZIOSI: nessuno di loro fa fallire un +giro, tutti e tre corrompono una serie irrecuperabile. + + 1. una chiamata fallita non deve diventare una riga che sembra un dato; + 2. "book vuoto" e "chiamata fallita" non devono collassare sullo stesso valore; + 3. il pacing deve esserci davvero (bite si auto-saturava il rate limit per-IP). + +Nessun test tocca la rete. +""" +from __future__ import annotations + +import sys +import time +from datetime import UTC, datetime +from pathlib import Path + +import pandas as pd +import pytest + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT)) +sys.path.insert(0, str(ROOT / "scripts" / "live")) + +from collect_chain import Budget, _depth_top3, snapshot_row, write # noqa: E402 + +TS = datetime(2026, 7, 30, 20, 0, tzinfo=UTC) +INST = { + "instrument_name": "ETH-31JUL26-1750-P", "base_currency": "ETH", + "strike": 1750.0, "option_type": "put", + "expiration_timestamp": int(pd.Timestamp("2026-07-31T08:00:00Z").timestamp() * 1000), +} + + +def _ob(bid=0.016, ask=0.0175, bids=((0.016, 10.0),), asks=((0.0175, 12.0),)): + return { + "best_bid_price": bid, "best_ask_price": ask, "mark_iv": 52.3, + "greeks": {"delta": -0.28, "gamma": 0.001, "theta": -1.2, "vega": 0.9}, + "open_interest": 500.0, "stats": {"volume": 31.0}, + "bids": [list(b) for b in bids], "asks": [list(a) for a in asks], + "underlying_price": 1923.4, "index_price": 1922.7, + } + + +# ------------------------------------------------- 1. una chiamata fallita non e' un dato + +def test_chiamata_fallita_marcata_error_e_senza_quota(): + r = snapshot_row(INST, None, TS) + assert r["quote_status"] == "error" + assert r["bid"] is None and r["ask"] is None and r["iv"] is None + assert r["instrument_name"] == INST["instrument_name"], ( + "la riga si scrive lo stesso: cio' che cambia e' che dichiara di non avere il dato" + ) + + +def test_profondita_e_NULL_su_errore_mai_zero(): + """Il difetto di bite: depth 0 su chiamata fallita = indistinguibile da 'book vuoto'. + Qui su errore la profondita' non esiste, e un book davvero vuoto vale 0.""" + assert snapshot_row(INST, None, TS)["book_depth_top3"] is None + vuoto = snapshot_row(INST, _ob(bid=0.0, ask=0.0, bids=(), asks=()), TS) + assert vuoto["book_depth_top3"] == 0.0 + assert vuoto["quote_status"] == "no_quote" + + +def test_book_vuoto_e_chiamata_fallita_sono_stati_DIVERSI(): + """Controllo positivo dell'idea stessa del collettore: se questi due collassassero, la + diagnostica del 29/07 (50% di quote perse per rate-limit) sarebbe di nuovo impossibile.""" + fallita = snapshot_row(INST, None, TS)["quote_status"] + vuoto = snapshot_row(INST, _ob(bid=0.0, ask=0.0, bids=(), asks=()), TS)["quote_status"] + assert fallita == "error" and vuoto == "no_quote" and fallita != vuoto + + +def test_un_solo_lato_quotato_e_comunque_ok(): + """Un book con solo il bid e' un mercato reale (illiquido), non un guasto.""" + r = snapshot_row(INST, _ob(ask=0.0, asks=()), TS) + assert r["quote_status"] == "ok" + assert r["bid"] is not None and r["ask"] is None + assert r["mid"] is None, "senza un lato il mid non esiste: non si inventa" + + +def test_campi_valorizzati_quando_la_quota_c_e(): + r = snapshot_row(INST, _ob(), TS) + assert r["quote_status"] == "ok" + assert r["mid"] == pytest.approx((0.016 + 0.0175) / 2) + assert r["delta"] == pytest.approx(-0.28) and r["iv"] == pytest.approx(52.3) + assert r["book_depth_top3"] == pytest.approx(22.0) + assert r["underlying_price"] == pytest.approx(1923.4) + assert r["option_type"] == "P" + + +def test_depth_top3_somma_solo_i_primi_tre_livelli(): + assert _depth_top3([[1, 5], [2, 5], [3, 5], [4, 100]]) == pytest.approx(15.0) + assert _depth_top3([]) == 0.0 + assert _depth_top3(None) is None + + +# ------------------------------------------------- 3. il pacing deve esistere + +def test_il_budget_impone_davvero_una_cadenza(): + """Bite non aveva pacing: ~44 chiamate/s per 26s. Qui la cadenza e' una proprieta' misurabile.""" + b = Budget(rps=20.0) + t0 = time.monotonic() + for _ in range(6): + b.wait() + dur = time.monotonic() - t0 + assert dur >= 5 / 20.0 * 0.8, f"6 chiamate a 20/s non possono durare {dur:.3f}s" + + +def test_il_budget_conta_i_rate_limit_e_gli_errori(): + b = Budget(rps=100.0) + b.note_error("boom") + b.rate_limited += 2 + assert b.errors == 1 and b.rate_limited == 2 and b.err_samples == ["boom"] + + +# ------------------------------------------------- scrittura + +def test_write_deduplica_e_tiene_l_ultima_osservazione(tmp_path, monkeypatch): + import collect_chain as CC + monkeypatch.setattr(CC, "STORE", tmp_path) + a = pd.DataFrame([snapshot_row(INST, _ob(bid=0.010), TS)]) + b = pd.DataFrame([snapshot_row(INST, _ob(bid=0.020), TS)]) # stesso (ts, strumento) + CC.write(a, TS) + p = CC.write(b, TS) + out = pd.read_parquet(p) + assert len(out) == 1, "stesso istante e stesso strumento = una riga sola" + assert out.iloc[0]["bid"] == pytest.approx(0.020), "vince l'osservazione piu' recente" + + +def test_write_separa_i_giorni(tmp_path, monkeypatch): + import collect_chain as CC + monkeypatch.setattr(CC, "STORE", tmp_path) + CC.write(pd.DataFrame([snapshot_row(INST, _ob(), TS)]), TS) + dopo = TS.replace(day=31) + CC.write(pd.DataFrame([snapshot_row(INST, _ob(), dopo)]), dopo) + assert sorted(p.name for p in tmp_path.glob("*.parquet")) == \ + ["2026-07-30.parquet", "2026-07-31.parquet"] diff --git a/tests/test_monitor_health.py b/tests/test_monitor_health.py index 1086750..6f55046 100644 --- a/tests/test_monitor_health.py +++ b/tests/test_monitor_health.py @@ -21,6 +21,7 @@ sys.path.insert(0, str(ROOT)) from src.live.monitor_health import ( # noqa: E402 MIN_COVERAGE, + MONITORS, MonitorSpec, alerts_from, assess, @@ -170,6 +171,8 @@ def test_i_monitor_reali_sono_leggibili_e_giudicati(): """Non asserisce che siano sani (dipende da quando gira il test): asserisce che la guardia li trova e si pronuncia su ognuno — il fallimento silenzioso sarebbe una lista vuota.""" rows = check_all(ROOT, now=NOW) - assert len(rows) == 6 + # legato al REGISTRO, non a una costante: aggiungere un monitor e' normale (30/07: + # collect_chain), dimenticarne uno per strada no. + assert len(rows) == len(MONITORS) >= 6 assert all(r["status"] in ("OK", "NUOVO", "FERMO", "BUCATO", "ASSENTE") for r in rows) assert {"paper_xsr", "paper_statarb", "paper_dvolspread"} <= {r["name"] for r in rows}