"""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. """ 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")] @lru_cache(maxsize=8) def spot_series(asset: str) -> pd.Series: 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 s @lru_cache(maxsize=8) def dvol_series(asset: str) -> pd.Series: 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 s # ------------------------------------------------------------------ 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)