PythagorasGoal/scripts/analysis/import_cb_archive.py

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