PythagorasGoal/scripts/research/cblib.py

"""CBLIB — harness della catena opzioni REALE (cerbero-bite), come altlib/eqlib per gli altri lati.

Legge la cache su disco `data/raw/cb_chain.parquet` (scritta una volta da
`scripts/analysis/fetch_cb_chain.py`), MAI dal container.

Serve a rispondere a una domanda che il progetto pone dal 19/06 e non aveva mai potuto misurare:
il premio che VRP01 incassa nel backtest e' quello che il mercato paga? Il sleeve prezza ENTRAMBE
le gambe con BS su DVOL ATM (`sleeves.VRP_CFG`, f=1.0 applicato al credito NETTO); qui si misura
    f = credito reale / credito modellato
sugli STESSI strike, cosi' il rapporto isola l'errore di PREZZO e non la scelta degli strike.

Regola che vale per ogni misura fatta con questo modulo: il f della sola gamba corta NON e' il f
dello spread. Il modello sbaglia soprattutto sull'ala che si COMPRA (piu' OTM, IV piu' alta per
skew, prezzata dal modello a vol ATM) -> misurare una gamba sola da' la risposta sbagliata con
segno rassicurante.

Seconda regola (09/09, debito §5.18): spot e DVOL escono da qui gia' CAUSALI — `asof(ts)` da'
l'ultima chiusura nota a ts. Chi costruisce una serie di prezzi altrove usi `causale()`, non
l'indice del feed com'e' (etichettato all'apertura: D6).
"""
from __future__ import annotations

import sys
from functools import lru_cache
from pathlib import Path

import numpy as np
import pandas as pd
from scipy.stats import norm

ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
RAW = ROOT / "data" / "raw"
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
SHORT_DELTA, LONG_DELTA = -0.28, -0.10


# ------------------------------------------------------------------ dati

@lru_cache(maxsize=4)
def load_chain() -> pd.DataFrame:
    """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"{STORE} assente — importa l'archivio con scripts/analysis/import_cb_archive.py "
            f"e raccogli con scripts/live/collect_chain.py"
        )
    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)
    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:
    d = load_chain() if df is None else df
    return d[(d["asset"] == asset) & (d["option_type"] == "P")]


def causale(s: pd.Series, cadenza: str) -> pd.Series:
    """Rietichetta una serie di CHIUSURE dall'APERTURA della barra all'istante in cui la chiusura
    e' nota (apertura + cadenza), cosi' `s.asof(ts)` restituisce l'ultima chiusura CONOSCIUTA a ts
    e mai una futura. I valori non si toccano, solo le etichette.

    🚨 Perche' esiste (debito §5.18, 09/09): i feed del progetto sono etichettati all'apertura —
    la barra 1h `20:00` chiude col 5m delle 20:55 (verificato: chiusura 1h a T == chiusura 5m a
    T+55 nel 100% delle barre) e la riga DVOL del giorno D e' la chiusura delle 23:00 di D
    (verificato contro la risoluzione 1h dell'API pubblica). Senza questo spostamento `asof(ts)`
    dava lo spot di UN'ORA dopo e il DVOL di FINO A 24 ORE dopo l'ingresso (D6, quinta occorrenza)."""
    out = s.copy()
    out.index = out.index + pd.Timedelta(cadenza)
    return out


@lru_cache(maxsize=8)
def spot_series(asset: str) -> pd.Series:
    """Chiusure 1h del feed certificato, etichettate all'istante in cui sono NOTE (apertura + 1h).

    `spot_series(a).asof(ts)` = ultima chiusura oraria conosciuta a `ts`. Conseguenza sul
    regolamento: `asof(exp)` alle 08:00 e' la chiusura della barra 07:00, cioe' il prezzo delle
    08:00 — prima era il prezzo delle 09:00, un'ora DOPO la scadenza."""
    from scripts.analysis.research_lab import load_tf
    px = load_tf(asset, "1h")
    s = pd.Series(px["close"].values.astype(float),
                  index=pd.to_datetime(px["timestamp"], unit="ms", utc=True)).sort_index()
    s.index = pd.DatetimeIndex(s.index).as_unit("ns")   # le quote bite hanno microsecondi
    return causale(s, "1h")


@lru_cache(maxsize=8)
def dvol_series(asset: str) -> pd.Series:
    """DVOL giornaliero (chiusura), etichettato all'istante in cui e' NOTO (giorno + 1D).

    `dvol_series(a).asof(ts)` = ultima chiusura giornaliera del DVOL conosciuta a `ts`: dentro il
    giorno D e' la chiusura di D−1, quindi vecchia fino a 23 ore. E' il prezzo della causalita'
    con un feed giornaliero; l'alternativa (DVOL orario dall'API pubblica) non e' nel feed."""
    d = pd.read_parquet(RAW / f"dvol_{asset.lower()}.parquet")
    s = pd.Series(d["close"].values.astype(float),
                  index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)).sort_index()
    s.index = pd.DatetimeIndex(s.index).as_unit("ns")
    return causale(s, "1D")


# ------------------------------------------------------------------ prezzo

def bs_put(S: float, K: float, T: float, sig: float) -> float:
    """Identica a `sleeves._bs_put` (r=0): il confronto dev'essere col prezzatore del sleeve."""
    if T <= 0 or sig <= 0:
        return max(K - S, 0.0)
    d1 = (np.log(S / K) + 0.5 * sig**2 * T) / (sig * np.sqrt(T))
    return K * norm.cdf(-(d1 - sig * np.sqrt(T))) - S * norm.cdf(-d1)


# ------------------------------------------------------------------ struttura

def pick_legs(snap: pd.DataFrame, short_delta: float = SHORT_DELTA,
              long_delta: float = LONG_DELTA) -> dict | None:
    """Dalle put di UNO snapshot+scadenza: gamba corta ~short_delta e lunga ~long_delta.

    Ritorna None se non e' costruibile un credit spread: serve che ENTRAMBE siano quotate
    (bid>0, ask>0, non incrociate) e che lo strike corto sia sopra quello lungo. Una gamba sola
    quotata non e' "meta' struttura": e' un'altra strategia, e va scartata.
    """
    ok = snap.dropna(subset=["bid", "ask", "delta"])
    ok = ok[(ok["bid"] > 0) & (ok["ask"] > 0) & (ok["ask"] >= ok["bid"])]
    if len(ok) < 2:
        return None
    s = ok.iloc[[int((ok["delta"] - short_delta).abs().to_numpy().argmin())]].iloc[0]
    l = ok.iloc[[int((ok["delta"] - long_delta).abs().to_numpy().argmin())]].iloc[0]
    if s["strike"] <= l["strike"]:
        return None
    return {
        "inst_short": s["instrument_name"], "inst_long": l["instrument_name"],
        "k_short": float(s["strike"]), "k_long": float(l["strike"]),
        "d_short": float(s["delta"]), "d_long": float(l["delta"]),
        "iv_short": float(s["iv"]) if pd.notna(s["iv"]) else np.nan,
        "iv_long": float(l["iv"]) if pd.notna(l["iv"]) else np.nan,
        "bid_short": float(s["bid"]), "ask_short": float(s["ask"]),
        "bid_long": float(l["bid"]), "ask_long": float(l["ask"]),
        "mid_short": float(s["mid"]) if pd.notna(s["mid"]) else np.nan,
        "mid_long": float(l["mid"]) if pd.notna(l["mid"]) else np.nan,
    }


def f_factors(legs: dict, spot: float, dvol_frac: float, dte_days: float) -> dict:
    """f = reale/modellato per gamba e per credito NETTO, agli STESSI strike.

    Convenzione di fill CONSERVATIVA: si vende al bid e si compra all'ask (si attraversa lo
    spread in entrambe le direzioni). Il `_mid` e' la stessa cosa a meta' spread.
    I premi Deribit sono quotati nel sottostante -> moltiplicati per lo spot passano a USD.
    """
    T = dte_days / 365.25
    mod_s = bs_put(spot, legs["k_short"], T, dvol_frac)
    mod_l = bs_put(spot, legs["k_long"], T, dvol_frac)
    cred_mod = mod_s - mod_l
    cred_real = (legs["bid_short"] - legs["ask_long"]) * spot
    cred_mid = (legs["mid_short"] - legs["mid_long"]) * spot
    return {
        "mod_short": mod_s, "mod_long": mod_l, "cred_mod": cred_mod,
        "cred_real": cred_real, "cred_mid": cred_mid,
        "f_short": legs["bid_short"] * spot / mod_s if mod_s > 0 else np.nan,
        "f_long": legs["ask_long"] * spot / mod_l if mod_l > 0 else np.nan,
        "f_net": cred_real / cred_mod if cred_mod > 0 else np.nan,
        "f_net_mid": cred_mid / cred_mod if cred_mod > 0 else np.nan,
        "width": legs["k_short"] - legs["k_long"],
    }


def weekly_entries(asset: str, df: pd.DataFrame | None = None, target_dte: float = 7.0) -> pd.DataFrame:
    """Un ingresso per scadenza: lo snapshot con DTE piu' vicino a `target_dte` in cui ENTRAMBE
    le gambe sono quotate. Ritorna anche f_* e il contesto (spot, DVOL, IV-rank causale)."""
    p = puts(asset, df)
    wk = p[(p["dte"] >= DTE_LO) & (p["dte"] <= DTE_HI)]
    S, V = spot_series(asset), dvol_series(asset)
    rows = []
    for exp, g in wk.groupby("exp"):
        cand = sorted(g["ts"].unique(),
                      key=lambda t: abs((exp - pd.Timestamp(t)).total_seconds() / 86400.0 - target_dte))
        for ts in cand:
            legs = pick_legs(g[g["ts"] == ts])
            if legs is None:
                continue
            t = pd.Timestamp(ts).as_unit("ns")
            dte = (exp - pd.Timestamp(ts)).total_seconds() / 86400.0
            spot = float(S.asof(t))
            dvol = float(V.asof(t))
            hist = V[V.index < t]
            ivr = float((hist < dvol).mean()) if len(hist) else np.nan
            rows.append({"asset": asset, "ts": t, "exp": exp, "dte": dte,
                         "spot": spot, "dvol_pct": dvol, "ivrank": ivr,
                         **legs, **f_factors(legs, spot, dvol / 100.0, dte)})
            break
    return pd.DataFrame(rows).sort_values("ts").reset_index(drop=True)


def close_cost_path(entry: pd.Series, df: pd.DataFrame | None = None) -> pd.DataFrame:
    """Costo di CHIUDERE lo spread, ora per ora, dalle standing quotes delle stesse due gambe.

    Chiudere = ricomprare la corta all'ask e rivendere la lunga al bid (di nuovo conservativo).
    E' cio' che il backtest modellato non ha: tiene a scadenza, quindi non vede ne' il
    50%-profit-take ne' il drawdown dentro la settimana.
    """
    d = load_chain() if df is None else df
    seg = d[(d["asset"] == entry["asset"]) & (d["ts"] > entry["ts"]) & (d["ts"] <= entry["exp"])
            & (d["instrument_name"].isin([entry["inst_short"], entry["inst_long"]]))]
    seg = seg.dropna(subset=["bid", "ask"])
    S = spot_series(entry["asset"])
    out = []
    for ts2, g2 in seg.groupby("ts"):
        s2 = g2[g2["instrument_name"] == entry["inst_short"]]
        l2 = g2[g2["instrument_name"] == entry["inst_long"]]
        if s2.empty or l2.empty:
            continue                                  # una gamba sola non prezza lo spread
        sp2 = float(S.asof(pd.Timestamp(ts2).as_unit("ns")))
        costo = (float(s2["ask"].iloc[0]) - float(l2["bid"].iloc[0])) * sp2
        out.append({"ts": ts2, "costo_chiusura": costo,
                    "pnl_aperto": entry["cred_real"] - costo})
    return pd.DataFrame(out)