PythagorasGoal/scripts/analysis/fetch_cb_chain.py

"""FETCH + CERTIFY della catena opzioni REALE accumulata da cerbero-bite.

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.

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.
  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.

    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
"""
from __future__ import annotations

import argparse
import sqlite3
import subprocess
import sys
import tempfile
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"
OUT = RAW / "cb_chain.parquet"
CONTAINER = "cerbero-bite-cerbero-bite-1"
DB_IN_CONTAINER = "/app/data/state.sqlite"

# --- soglie di certificazione (decise, non default) ---
HOLLOW_WARN = 0.05      # >5% di quote vuote su una finestra = DEGRADATO
HOLLOW_FAIL = 0.30      # >30% = il dato non descrive piu' il mercato
RECENT_DAYS = 7         # la finestra recente conta a parte: un guasto in corso non va diluito
                        # in 3 mesi di storia sana (media su tutto il campione = guasto invisibile)


def _pull_db(dest: Path) -> Path:
    """Copia il DB dal container. Il volume docker non e' leggibile senza root."""
    cmd = ["docker", "cp", f"{CONTAINER}:{DB_IN_CONTAINER}", str(dest)]
    subprocess.run(cmd, check=True, capture_output=True)
    return dest


def read_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"):
        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")
    df["dte"] = (df["exp"] - df["ts"]).dt.total_seconds() / 86400.0
    return df.dropna(subset=["ts", "exp", "strike"]).drop(columns=["timestamp", "expiry"])


# ------------------------------------------------------------------ certificazione

def hollow_rate(df: pd.DataFrame) -> float:
    """Frazione di righe presenti ma SENZA quota. Non e' 'dati mancanti': la riga c'e'."""
    if df.empty:
        return float("nan")
    return float(df["bid"].isna().mean())


def worst_day_hollow(df: pd.DataFrame, days: int) -> tuple[float, object]:
    """Tasso di quote vuote del GIORNO PEGGIORE nella finestra recente.

    La media sulla finestra non basta: un guasto iniziato 2 giorni fa, mediato su 7, si legge
    13% invece del 50% reale — la stessa diluizione che questo modulo dichiara di voler evitare
    quando confronta 'ultimi 7g' con 'tutto'. Un guasto IN CORSO si misura al suo giorno peggiore.
    """
    if df.empty:
        return float("nan"), None
    recent = df[df["ts"] >= df["ts"].max() - pd.Timedelta(days=days)]
    if recent.empty:
        return float("nan"), None
    per_day = recent.assign(d=recent["ts"].dt.date).groupby("d")["bid"].apply(lambda s: s.isna().mean())
    return float(per_day.max()), per_day.idxmax()


def crossed_rate(df: pd.DataFrame) -> float:
    q = df.dropna(subset=["bid", "ask"])
    if q.empty:
        return float("nan")
    return float((q["bid"] > q["ask"]).mean())


def monotonicity_violations(df: pd.DataFrame) -> float:
    """Frazione di coppie strike-adiacenti in cui il premio put DIMINUISCE al salire dello strike.

    Su un singolo (istante, scadenza) il premio di una put e' non-decrescente nello strike: e' una
    proprieta' del payoff, non del modello -> una violazione e' un difetto di dato, non un regime.
    """
    puts = df[(df["option_type"] == "P")].dropna(subset=["mid"])
    if puts.empty:
        return float("nan")
    bad = tot = 0
    for _, g in puts.groupby(["ts", "asset", "exp"], sort=False):
        if len(g) < 2:
            continue
        m = g.sort_values("strike")["mid"].to_numpy()
        d = m[1:] - m[:-1]
        bad += int((d < 0).sum())
        tot += len(d)
    return float(bad / tot) if tot else float("nan")


def certify(df: pd.DataFrame) -> list[dict]:
    """Un verdetto per asset. La finestra recente e' valutata a parte (vedi RECENT_DAYS)."""
    out = []
    if df.empty:
        return out
    now = df["ts"].max()
    for asset, g in df.groupby("asset"):
        rec = g[g["ts"] >= now - pd.Timedelta(days=RECENT_DAYS)]
        h_all, h_rec = hollow_rate(g), hollow_rate(rec)
        h_day, day_worst = worst_day_hollow(g, RECENT_DAYS)
        worst = max(h_all, h_rec, h_day)  # il verdetto segue il giorno peggiore, non la media
        status = "OK" if worst <= HOLLOW_WARN else ("DEGRADATO" if worst <= HOLLOW_FAIL else "QUOTE-VUOTE")
        out.append({
            "asset": asset,
            "righe": len(g),
            "da": g["ts"].min(),
            "a": g["ts"].max(),
            "scadenze": g["exp"].nunique(),
            "vuote_tutto": h_all,
            f"vuote_{RECENT_DAYS}g": h_rec,
            "vuote_giorno_peggiore": h_day,
            "giorno_peggiore": day_worst,
            "incrociate": crossed_rate(g),
            "non_monotone": monotonicity_violations(g),
            "depth_zero": float((g["book_depth_top3"] == 0).mean()),
            "settimanali_4_10dte": int(((g["dte"] >= 4) & (g["dte"] <= 10)).sum()),
            "status": status,
        })
    return out


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

    print("=" * 100)
    print("  CATENA OPZIONI cerbero-bite — certificazione")
    print("=" * 100)
    rep = certify(df)
    for r in rep:
        print(f"\n  {r['asset']}  {r['righe']:,} righe  {r['da']:%Y-%m-%d} -> {r['a']:%Y-%m-%d}  "
              f"({r['scadenze']} scadenze, {r['settimanali_4_10dte']:,} righe a 4-10 DTE)")
        print(f"     quote VUOTE (riga presente, bid/ask NULL): tutto {100*r['vuote_tutto']:.1f}%  "
              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)")
        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 any(r["status"] != "OK" for r in rep):
        print("  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


if __name__ == "__main__":
    raise SystemExit(main())