"""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())