PythagorasGoal/scripts/live/collect_chain.py

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