"""TERM-STRUCTURE della vol implicita BTC/ETH — pendenza, carry, gate di rischio (ondata 2026-08-22). DOMANDA. Tre, dichiarate prima di misurare: Q1 la PENDENZA (backwardation = front > back) PRECEDE i ritorni, o e' contemporanea? Q2 il CARRY di vol (roll-down lungo la curva) e' INCASSABILE al bid-ask reale? Q3 vale come GATE DI RISCHIO sopra TP01/SKH01? IPOTESI A PRIORI (registrate prima di guardare i numeri — servono a poterle smentire): Q1 contemporanea. Il front reagisce allo spot: un tuffo alza la vol a 7g piu' di quella a 30g. Se e' cosi', la pendenza e' un TERMOMETRO, non un segnale. Q2 no. Il roll-down fra 7g e 30g vale frazioni di punto-vol al giorno; il bid-ask ATM di Deribit vale punti-vol INTERI. L'aritmetica dovrebbe chiudere la questione senza backtest. Q3 ridondante col trend, come i 4 precedenti (DVOL-spike, macro, funding, breadth): un gate di de-risk lavora nei giorni in cui TP01 e' gia' flat. IL PRECEDENTE SCOMODO (03/07, `r0703_vrpimp_*`): un gate di term-structure VIX/VXV su SPX valeva +0.90 di Sharpe (DSR 0.992) ed era un **confound di modello al 100%** — la variabile del gate coincideva con l'errore di prezzatura BS-flat vs term-structure. Da li' la regola: *riprezzare term-structure-consistent prima di credere a un gate vol su strutture BS-flat*. QUI QUEL CONFOUND NON PUO' ESISTERE: non si prezza nessuna opzione. Il sottostante e' il perp BTC/ETH, i ritorni sono quelli del feed certificato, e la term structure entra solo come VARIABILE OSSERVATA. Quindi il 03/07 non e' una scusa per non guardare — ed e' esattamente percio' che serve cercare il confound *di questo* filone, che e' un altro: ⚠️ IL CONFOUND DI QUESTO FILONE: iv(7g) e' in larga parte una funzione della vol REALIZZATA recente. Se e' cosi', `slope = iv30 - iv7` non e' un'informazione nuova ma un RICONFEZIONAMENTO di RV — cioe' la stessa variabile di `dvol_directional.py` (VRP-Z), gia' giudicata **HEDGE, earns_slot=False** su 5 anni il 2026-06-29. Misurato in SEZIONE 2. IL DATO, e il muro che ci sta dentro. La term structure richiede >=2 scadenze nello STESSO istante. `bite_archive` prima del **2026-06-09** collezionava **UNA scadenza per giro**: la finestra utilizzabile NON e' "3,7 mesi di catena", e' **dal 2026-06-09** (~74 giorni). Con 74 giorni **non esiste un hold-out** e SE(Sharpe) ~ sqrt(365/74) ~ 2.2: qualunque Sharpe misurato qui e' a una deviazione standard da zero. Verdetto massimo: LEAD con gate e data. Si conta cio' che e' QUOTATO (bid>0, ask>0, iv presente), non le righe: il guasto 29-30/07 ha fino al 51% di quote vuote in un giorno. METODO. ATM IV per scadenza = interpolazione della IV sul DELTA a 0.5 (call) / -0.5 (put), mai estrapolata. Il delta referenzia il FORWARD, quindi la curva non eredita lo skew via la base (prendere "lo strike piu' vicino allo spot" lo farebbe: a 90g il forward e' lontano dallo spot). Interpolazione ai tenori fissi in VARIANZA TOTALE (w = iv^2 * T lineare in T), stile VIX. nice -n 19 timeout 900 uv run python scripts/research/r0822_term_structure.py """ from __future__ import annotations import os import sys import tempfile from pathlib import Path import numpy as np import pandas as pd ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(ROOT)) sys.path.insert(0, str(ROOT / "scripts" / "research" / "alt")) import altlib as A # noqa: E402 RAW = ROOT / "data" / "raw" STORE = RAW / "cb_chain" CACHE = Path(os.environ.get("TMPDIR", tempfile.gettempdir())) / "r0822_ts_cache" ASSETS = ("BTC", "ETH") # Muro del dato: prima di questa data l'archivio ha UNA scadenza per giro -> nessuna curva. TS_START = pd.Timestamp("2026-06-09", tz="UTC") TENORS = (7, 14, 30, 60, 90) FRONT, BACK = 7, 30 # la pendenza canonica di questo studio DTE_MIN = 0.5 # sotto mezza giornata l'ATM IV e' rumore di microstruttura # Vincoli di venue DICHIARATI (misurati il 2026-07-30 su `get_instrument`, non ri-letti qui: # questo script non tocca la rete). Servono alla sezione di eseguibilita'. MIN_LOT = {"BTC": 0.1, "ETH": 1.0} # contratti minimi opzioni Deribit OPT_FEE_UNDERLYING = 0.0003 # 0,03% del sottostante per gamba OPT_FEE_CAP_PREMIUM = 0.125 # cap: 12,5% del premio della singola opzione CAPITALE = 635.0 # il conto vero # Conteggio dei trial, AL RIALZO (sezione 6). Ogni voce e' una configurazione VALUTATA. TRIALS: list[tuple[str, int]] = [] def hr(t: str = "") -> None: print("\n" + "=" * 100) if t: print(t) print("=" * 100) def sub(t: str) -> None: print(f"\n--- {t} " + "-" * max(0, 96 - len(t))) # =========================================================================================== # SEZIONE 0 — IL DATO # =========================================================================================== def load_quoted_chain() -> pd.DataFrame: """Catena filtrata IN LETTURA (colonne + finestra), tenendo solo cio' che e' QUOTATO. "Quotato" = bid>0 AND ask>0 AND iv presente AND delta presente. Una riga con bid/ask NULL esiste (il collettore la persiste) ma non e' un prezzo: contarla come dato e' l'errore che il progetto ha gia' fatto tre volte (paper_dvolspread, fresh_5m, guasto 29/07). """ cols = ["asset", "option_type", "strike", "iv", "delta", "bid", "ask", "vega", "ts", "exp", "quote_status"] parts = [] arch = STORE / "bite_archive.parquet" if arch.exists(): d = pd.read_parquet(arch, columns=cols) parts.append(d[d["ts"] >= TS_START]) del d for p in sorted(STORE.glob("2026-*.parquet")): parts.append(pd.read_parquet(p, columns=cols)) if not parts: raise FileNotFoundError(f"{STORE}: nessuna catena") df = pd.concat(parts, ignore_index=True) del parts df = df.drop_duplicates(subset=["ts", "asset", "strike", "option_type", "exp"], keep="last") df["hr"] = df["ts"].dt.floor("h") df["dte"] = (df["exp"] - df["ts"]).dt.total_seconds() / 86400.0 df["quoted"] = (df["bid"] > 0) & (df["ask"] > 0) & df["iv"].notna() & df["delta"].notna() return df def sezione0(chain: pd.DataFrame) -> None: hr("SEZIONE 0 — IL DATO: quanto ce n'e' davvero, e da quando") print(f" finestra letta : {chain.ts.min()} -> {chain.ts.max()}") print(f" righe : {len(chain):,}") print(f" righe QUOTATE : {int(chain.quoted.sum()):,} ({chain.quoted.mean():.1%})") print("\n ⚠️ IL MURO: la term structure richiede >=2 scadenze nello STESSO istante.") arch = STORE / "bite_archive.parquet" if arch.exists(): pre = pd.read_parquet(arch, columns=["asset", "ts", "exp"]) pre["hr"] = pre["ts"].dt.floor("h") n_pre = pre[pre.ts < TS_START].groupby(["asset", "hr"])["exp"].nunique() n_post = pre[pre.ts >= TS_START].groupby(["asset", "hr"])["exp"].nunique() print(f" archivio PRIMA del {TS_START.date()}: scadenze/ora mediana = " f"{n_pre.median():.0f} (max {n_pre.max():.0f}) su {len(n_pre)} ore-asset") print(f" archivio DOPO : mediana = {n_post.median():.0f} " f"(max {n_post.max():.0f}) su {len(n_post)} ore-asset") print(" => la catena copre dal 2026-05-01, ma la CURVA esiste solo dal 2026-06-09.") print(" Non sono 3,7 mesi: sono ~74 giorni. Nessun hold-out e' costruibile.") del pre q = chain.groupby([chain.hr.dt.date, "asset"])["quoted"].mean().unstack() bad = q[(q < 0.70).any(axis=1)] print(f"\n giorni con quote quotate <70% su almeno un asset: {len(bad)}") if len(bad): print(bad.round(3).to_string().replace("\n", "\n ")) print(" (guasto noto 29-30/07: rate-limit per-IP saturato dal collettore di bite.)") print(f"\n minuto dello snapshot (mediana): archivio bite = " f"{chain[chain.ts < pd.Timestamp('2026-07-30', tz='UTC')].ts.dt.minute.median():.0f}', " f"raccolta propria = {chain[chain.ts >= pd.Timestamp('2026-07-31', tz='UTC')].ts.dt.minute.median():.0f}'") print(" -> lo snapshot dell'ora t cade DENTRO la barra oraria t. Ogni ritorno usato come") print(" bersaglio parte dalla CHIUSURA della barra t (>=35 min dopo): buffer causale.") # =========================================================================================== # SEZIONE 1 — RICOSTRUZIONE DELLA CURVA + CERTIFICAZIONE # =========================================================================================== def _interp_at(gid: np.ndarray, x: np.ndarray, y: np.ndarray, ngrp: int, target: float) -> np.ndarray: """Interpolazione lineare di y(x) a `target`, per ogni gruppo, VETTORIALE e senza estrapolazione (NaN se il gruppo non racchiude il target). `gid` dev'essere ordinato crescente e (gid, x) lessicograficamente ordinati. Equivalente a np.interp gruppo-per-gruppo: qui serve solo perche' i gruppi sono ~30.000 e un ciclo python li paga a caro prezzo su una VPS a 2 core condivisa. """ n_tot = np.bincount(gid, minlength=ngrp) start = np.concatenate([[0], np.cumsum(n_tot)[:-1]]) n_below = np.bincount(gid[x <= target], minlength=ngrp) ok = (n_below > 0) & (n_below < n_tot) out = np.full(ngrp, np.nan) if not ok.any(): return out lo = (start + n_below - 1)[ok] hi = lo + 1 x0, x1, y0, y1 = x[lo], x[hi], y[lo], y[hi] w = np.where(x1 != x0, (target - x0) / np.where(x1 != x0, x1 - x0, 1.0), 0.0) out[ok] = y0 + w * (y1 - y0) return out def build_atm_by_expiry(chain: pd.DataFrame) -> pd.DataFrame: """ATM IV per (asset, ora, scadenza) = IV interpolata sul DELTA a 0.5 (call) / -0.5 (put). Perche' il delta e non "lo strike piu' vicino allo spot": il delta referenzia il FORWARD. A 90 giorni il forward sta lontano dallo spot, quindi lo strike ATM-spot e' un'opzione OTM e la sua IV porta dentro lo SKEW — che e' un altro filone e falserebbe la pendenza in modo sistematico e crescente col tenore. Nessuna estrapolazione: se il delta 0.5 non e' racchiuso dai quotati, quella scadenza a quell'ora non ha ATM e viene scartata. """ f = CACHE / "atm_by_expiry.parquet" if f.exists(): return pd.read_parquet(f) q = chain[chain.quoted & (chain.dte > DTE_MIN)].copy() keys = ["asset", "hr", "exp"] q["gid"] = q.groupby(keys, sort=True).ngroup() ngrp = int(q["gid"].max()) + 1 ivs = [] for typ, tgt in (("C", 0.5), ("P", -0.5)): s = q[q.option_type == typ].sort_values(["gid", "delta"]) v = np.full(ngrp, np.nan) if len(s): g2 = s["gid"].to_numpy() uniq, inv = np.unique(g2, return_inverse=True) v[uniq] = _interp_at(inv, s["delta"].to_numpy(), s["iv"].to_numpy(), len(uniq), tgt) ivs.append(v) M = np.vstack(ivs) with np.errstate(invalid="ignore"): cnt = np.sum(np.isfinite(M), axis=0) iv = np.where(cnt > 0, np.nansum(np.where(np.isfinite(M), M, 0.0), axis=0) / np.maximum(cnt, 1), np.nan) out = (q.groupby(keys, sort=True)["dte"].median().reset_index()) out["iv"] = iv out = out.dropna(subset=["iv"]).reset_index(drop=True) CACHE.mkdir(parents=True, exist_ok=True) out.to_parquet(f) return out def _curve_row(g: pd.DataFrame) -> pd.Series: g = g.sort_values("dte") t = g["dte"].to_numpy() w = (g["iv"].to_numpy() / 100.0) ** 2 * t # varianza totale o = {} for T in TENORS: o[f"iv_{T}"] = (np.sqrt(np.interp(T, t, w) / T) * 100.0 if len(t) >= 2 and t.min() <= T <= t.max() else np.nan) o["n_exp"] = len(t) return pd.Series(o) def build_curve(atm: pd.DataFrame) -> pd.DataFrame: f = CACHE / "curve.parquet" if f.exists(): return pd.read_parquet(f) c = atm.groupby(["asset", "hr"]).apply(_curve_row, include_groups=False).reset_index() CACHE.mkdir(parents=True, exist_ok=True) c.to_parquet(f) return c def sezione1(curve: pd.DataFrame) -> dict: hr("SEZIONE 1 — RICOSTRUZIONE DELLA CURVA E SUA CERTIFICAZIONE") print(" ATM = IV interpolata sul delta 0.5/-0.5 (mai estrapolata); tenori fissi in varianza totale.") cov = curve[[f"iv_{T}" for T in TENORS]].notna().mean() print("\n copertura per tenore (frazione delle ore-asset ricostruibili):") for T in TENORS: print(f" iv_{T:<3d} {cov[f'iv_{T}']:.1%}") print(f" -> iv_90 e' sotto il 10%: SCARTATO dallo studio. La pendenza canonica e' " f"iv_{BACK} - iv_{FRONT}.") sub("controllo 1 — contro il logger indipendente `vol_term_*.parquet` (altro codice, altra fonte)") print(" (`log_vol_termstructure.py`: mark_iv da book_summary, ATM = strike piu' vicino allo") print(" spot, interpolazione diversa. Se le due strade coincidono, la mia non e' un artefatto.)") agree = {} for a in ASSETS: p = RAW / f"vol_term_{a.lower()}.parquet" if not p.exists(): print(f" {a}: {p.name} assente — controllo NON GIRATO") continue vt = pd.read_parquet(p).set_index("date") mine = curve[curve.asset == a].set_index("hr") j = vt.join(mine[[f"iv_{T}" for T in TENORS]], how="inner") for T in (7, 30, 60): x, y = j[f"iv_{T}d"], j[f"iv_{T}"] m = x.notna() & y.notna() if m.sum() >= 10: print(f" {a} iv_{T:<3d} n={m.sum():3d} corr={x[m].corr(y[m]):+.4f} " f"scarto medio={(y[m] - x[m]).mean():+.3f} pt-vol") agree[(a, T)] = float(x[m].corr(y[m])) sub("controllo 2 — contro il DVOL (indice Deribit a 30g, storia 2021+)") for a in ASSETS: dv = pd.read_parquet(RAW / f"dvol_{a.lower()}.parquet") s = pd.Series(dv["close"].astype(float).values, index=pd.to_datetime(dv["timestamp"], unit="ms", utc=True)).sort_index() mine = curve[curve.asset == a].set_index("hr")["iv_30"].dropna() j = pd.DataFrame({"mine": mine}) j["dvol"] = s.reindex(j.index, method="ffill") j = j.dropna() print(f" {a}: n={len(j)} corr={j.mine.corr(j.dvol):+.4f} " f"bias={(j.mine - j.dvol).mean():+.3f} pt-vol sd={(j.mine - j.dvol).std():.3f}") print(" Il bias NEGATIVO e' atteso e conferma la ricostruzione: il DVOL e' un indice tipo") print(" variance-swap (integra tutto lo smile) e sta STRUTTURALMENTE sopra l'ATM per lo skew.") print(" Un bias ~0 sarebbe stato il segnale d'allarme, non questo.") return agree # =========================================================================================== # SEZIONE 2 — Q1: LEAD-LAG. La pendenza precede o segue? # =========================================================================================== def slope_hourly(curve: pd.DataFrame, a: str) -> pd.Series: s = curve[curve.asset == a].set_index("hr") return (s[f"iv_{BACK}"] - s[f"iv_{FRONT}"]).dropna().sort_index() def px_hourly(a: str) -> pd.Series: d = A.get(a, "1h") return pd.Series(d["close"].astype(float).values, index=pd.DatetimeIndex(pd.to_datetime(d["datetime"], utc=True))) def _block_pctl(x: np.ndarray, y: np.ndarray, stat: float, L: int, n: int = 400, seed: int = 20260822) -> float: """Percentile della correlazione osservata contro il null a BLOCCHI (y ricampionata da blocchi non allineati a x): conserva l'autocorrelazione e rompe solo l'accoppiamento.""" rng = np.random.default_rng(seed) m = len(x) if m <= L + 5: return float("nan") nb = max(1, m // L) out = np.empty(n) for i in range(n): ii = np.concatenate([np.arange(s, s + L) for s in rng.integers(0, m - L, nb)]) jj = np.concatenate([np.arange(s, s + L) for s in rng.integers(0, m - L, nb)]) out[i] = np.corrcoef(x[ii], y[jj])[0, 1] return float((out < stat).mean()) def sezione2(curve: pd.DataFrame) -> dict: hr("SEZIONE 2 — Q1: la pendenza PRECEDE i ritorni o li SEGUE?") res = {} for a in ASSETS: sl = slope_hourly(curve, a) px = px_hourly(a) lp = np.log(px) j = pd.DataFrame({"slope": sl}).join(px.rename("px"), how="inner").dropna() idx = j.index r1 = lp.diff() sub(f"{a} — n ore = {len(j)} ({idx.min()} -> {idx.max()})") print(f" pendenza iv_{BACK}-iv_{FRONT}: media {j.slope.mean():+.2f} pt-vol, " f"sd {j.slope.std():.2f}, backwardation nel {100*(j.slope<0).mean():.1f}% delle ore") print("\n (a) CROSS-CORRELAZIONE fra la VARIAZIONE oraria di pendenza e il ritorno orario.") print(" lag<0 = ritorno PRIMA della variazione (il prezzo guida) | lag>0 = DOPO (la") print(" pendenza guiderebbe). La barra t va da t a t+1h e lo snapshot cade dentro:") print(" il lag -1 e' quindi in gran parte SOVRAPPOSTO, non un vero anticipo.") ds = j.slope.diff() row = [] for k in (-3, -2, -1, 0, 1, 2, 3, 6, 12, 24): row.append((k, float(ds.corr(r1.shift(-k).reindex(idx))))) print(" " + " ".join(f"{k:+d}h:{v:+.3f}" for k, v in row)) best = max(row, key=lambda t: abs(t[1])) print(f" picco |corr| al lag {best[0]:+d}h ({best[1]:+.3f})") print("\n (b) LIVELLO della pendenza vs ritorni STRETTAMENTE FUTURI (ingresso alla") print(" chiusura della barra t) e vs ritorni PASSATI.") for h in (1, 3, 6, 24, 72, 168): fwd = (lp.shift(-h) - lp).reindex(idx) pst = (lp - lp.shift(h)).reindex(idx) m = j.slope.notna() & fwd.notna() & pst.notna() print(f" h={h:4d}h corr(pendenza, FUTURO) {j.slope[m].corr(fwd[m]):+.3f} " f"corr(pendenza, PASSATO) {j.slope[m].corr(pst[m]):+.3f}") TRIALS.append((f"Q1 {a}: 6 orizzonti x (futuro,passato) x (livello,variazione)", 24)) print("\n (c) IL CONFOUND DI QUESTO FILONE — la pendenza e' informazione NUOVA o RV riciclata?") rvp = (r1.rolling(24).std() * np.sqrt(24 * 365) * 100).reindex(idx) rvf = (r1.shift(-24).rolling(24).std().shift(-23) * np.sqrt(24 * 365) * 100).reindex(idx) ivf = curve[curve.asset == a].set_index("hr")[f"iv_{FRONT}"].reindex(idx) c_pp = float(j.slope.corr(rvp)) c_pf = float(j.slope.corr(rvf)) print(f" corr(pendenza, RV 24h PASSATA) {c_pp:+.3f} <-- quanto e' termometro") print(f" corr(pendenza, RV 24h FUTURA) {c_pf:+.3f} <-- quanto e' previsione") print(f" corr(iv_{FRONT}, RV 24h passata) {float(ivf.corr(rvp)):+.3f} " f"corr(iv_{FRONT}, RV futura) {float(ivf.corr(rvf)):+.3f}") res[(a, "conf_past")], res[(a, "conf_fwd")] = c_pp, c_pf print("\n (d) La correlazione col futuro SOPRAVVIVE se si controlla per il PASSATO?") print(" (regressione del ritorno futuro su pendenza + ritorno passato + RV passata;") print(" se il coefficiente della pendenza evapora, il segnale era 'e' appena sceso')") for h in (24, 72, 120): fwd = (lp.shift(-h) - lp).reindex(idx) pst = (lp - lp.shift(h)).reindex(idx) m = j.slope.notna() & fwd.notna() & pst.notna() & rvp.notna() X = np.column_stack([np.ones(m.sum()), j.slope[m], pst[m], rvp[m]]) y = fwd[m].to_numpy() b, *_ = np.linalg.lstsq(X, y, rcond=None) # errori standard Newey-West (lag = h, finestre sovrapposte) e = y - X @ b XtXi = np.linalg.pinv(X.T @ X) S = (X * e[:, None]).T @ (X * e[:, None]) for L in range(1, h + 1): w = 1.0 - L / (h + 1.0) G = (X[L:] * e[L:, None]).T @ (X[:-L] * e[:-L, None]) S += w * (G + G.T) se = np.sqrt(np.diag(XtXi @ S @ XtXi)) simple = float(j.slope[m].corr(fwd[m])) print(f" h={h:4d}h corr semplice {simple:+.3f} | beta_pendenza={b[1]:+.5f} " f"t(NW)={b[1]/se[1]:+.2f} | beta_ret_passato t={b[2]/se[2]:+.2f}") TRIALS.append((f"Q1 {a}: 3 regressioni controllate", 3)) print("\n (e) Test giornaliero con null a BLOCCHI (la finestra e' di 74 giorni: la") print(" correlazione va confrontata con la sua dispersione, non con zero).") sld = j.slope.resample("1D").last().dropna() pxd = px.resample("1D").last() lpd = np.log(pxd) for h in (1, 3, 5): fwd = (lpd.shift(-h) - lpd).reindex(sld.index) m = sld.notna() & fwd.notna() x, y = sld[m].to_numpy(), fwd[m].to_numpy() c = float(np.corrcoef(x, y)[0, 1]) p = _block_pctl(x, y, c, L=max(3 * h, 7)) print(f" h={h}g n={m.sum():3d} corr={c:+.3f} pctl vs null a blocchi={p:.3f}") TRIALS.append((f"Q1 {a}: 3 orizzonti giornalieri", 3)) return res # =========================================================================================== # SEZIONE 3 — Q2: il carry di vol e' incassabile? # =========================================================================================== def build_atm_cost(chain: pd.DataFrame) -> pd.DataFrame: """Costo REALE di attraversare lo spread su un'opzione ATM, in PUNTI DI VOL. (ask-bid) e' quotato nel sottostante -> x spot = USD; / vega = punti di vol, che e' l'unita' in cui si misura il roll-down. Cosi' costo e carry sono confrontabili senza modello. """ f = CACHE / "atm_cost.parquet" if f.exists(): return pd.read_parquet(f) q = chain[chain.quoted & (chain.dte > DTE_MIN) & (chain.option_type == "C") & (chain.vega > 0)].copy() q["ad"] = (q["delta"] - 0.5).abs() atm = q.sort_values("ad").groupby(["asset", "hr", "exp"], as_index=False).first() atm = atm[atm.ad < 0.08] for a in ASSETS: p = px_hourly(a) m = atm.asset == a atm.loc[m, "spot"] = p.reindex(atm.loc[m, "hr"]).ffill().to_numpy() atm = atm.dropna(subset=["spot"]) atm["spr_usd"] = (atm.ask - atm.bid) * atm.spot atm["spr_vol"] = atm.spr_usd / atm.vega atm["prem_usd"] = (atm.bid + atm.ask) / 2.0 * atm.spot CACHE.mkdir(parents=True, exist_ok=True) atm.to_parquet(f) return atm def _roll_table(r: pd.DataFrame, H: int) -> pd.DataFrame: """Roll IMPLICITO vs REALIZZATO su orizzonte H ore, appaiato per (scadenza, ISTANTE D'INGRESSO). Appaiato sull'INGRESSO e non sull'uscita: e' l'uscita che le due letture spostano, e un inner-join sulla data d'uscita terrebbe solo i casi in cui non e' successo niente (trappola incontrata 3 volte nel progetto: venue-watch, GTAA, VRP profit-take). impl = iv_curva(T-H) al tempo t - iv(T) al tempo t (se la curva non si muovesse) real = iv(T-H) al tempo t+H - iv(T) al tempo t (cio' che e' successo davvero) """ r = r[["hr", "exp", "iv", "dte"]].dropna() nxt = r[["hr", "exp", "iv"]].copy() nxt["hr"] = nxt["hr"] - pd.Timedelta(hours=H) j = r.merge(nxt.rename(columns={"iv": "iv_fut"}), on=["hr", "exp"], how="inner") j["T2"] = j["dte"] - H / 24.0 j = j[j["T2"] > DTE_MIN] if j.empty: return pd.DataFrame(columns=["dte", "impl", "real"]) # curva dell'ora t, ordinata: chiave monotona globale gid*1000 + dte -> searchsorted unico c = r.sort_values(["hr", "dte"]).copy() c["gid"] = c.groupby("hr", sort=True).ngroup() key = c["gid"].to_numpy() * 1000.0 + c["dte"].to_numpy() w = (c["iv"].to_numpy() / 100.0) ** 2 * c["dte"].to_numpy() # varianza totale gmap = dict(zip(c["hr"].to_numpy(), c["gid"].to_numpy())) n_tot = np.bincount(c["gid"].to_numpy()) start = np.concatenate([[0], np.cumsum(n_tot)[:-1]]) gq = np.array([gmap[h] for h in j["hr"].to_numpy()]) pos = np.searchsorted(key, gq * 1000.0 + j["T2"].to_numpy(), side="left") lo, hi = pos - 1, pos ok = (lo >= start[gq]) & (hi < start[gq] + n_tot[gq]) ivT2 = np.full(len(j), np.nan) if ok.any(): l, h2 = lo[ok], hi[ok] x0 = key[l] - gq[ok] * 1000.0 x1 = key[h2] - gq[ok] * 1000.0 t2 = j["T2"].to_numpy()[ok] ww = np.where(x1 != x0, (t2 - x0) / np.where(x1 != x0, x1 - x0, 1.0), 0.0) var = w[l] + ww * (w[h2] - w[l]) ivT2[ok] = np.sqrt(np.maximum(var, 0.0) / t2) * 100.0 j = j.assign(ivT2=ivT2).dropna(subset=["ivT2"]) return pd.DataFrame({"dte": j["dte"].to_numpy(), "impl": j["ivT2"].to_numpy() - j["iv"].to_numpy(), "real": j["iv_fut"].to_numpy() - j["iv"].to_numpy()}) def sezione3(chain: pd.DataFrame, atm_exp: pd.DataFrame, curve: pd.DataFrame) -> None: hr("SEZIONE 3 — Q2: il CARRY di vol (roll-down) e' incassabile al bid-ask reale?") print(" Definizione senza modello: un'opzione a tenore T, dopo dt, sta a T-dt. Se la curva non") print(" si muove la sua IV diventa iv(T-dt). Chi e' CORTO guadagna iv(T)-iv(T-dt) punti di vol.") print(" Il costo di entrare e uscire e' lo spread bid-ask, misurabile negli STESSI punti di vol.") sub("(a) quanto vale il roll-down, e viene DAVVERO incassato?") print(" Confronto appaiato per SCADENZA e per ISTANTE D'INGRESSO (mai allineato sull'uscita):") print(" roll IMPLICITO = iv_curva(T-dt) - iv(T) al tempo t ; roll REALIZZATO = iv(T-dt) a t+dt - iv(T) a t") roll_day = {} for H in (24, 168): for a in ASSETS: d = _roll_table(atm_exp[atm_exp.asset == a], H) d = d[(d.dte >= 4) & (d.dte <= 45)] if len(d) < 50: print(f" {a} dt={H}h: campione insufficiente ({len(d)})") continue b = np.polyfit(d.impl, d.real, 1)[0] diff = d.real - d.impl nblk = max(1, len(d) // max(1, H)) # osservazioni ~indipendenti t = diff.mean() / diff.std() * np.sqrt(nblk) if diff.std() > 0 else 0.0 print(f" {a} dt={H:3d}h n={len(d):5d} roll implicito medio {d.impl.mean():+.4f} pt-vol " f"realizzato {d.real.mean():+.4f} beta(real~impl)={b:+.2f} " f"corr={d.impl.corr(d.real):+.3f} (real-impl) t={t:+.2f}") if H == 24: roll_day[a] = -float(d.impl.mean()) # guadagno giornaliero di CHI E' CORTO print("\n Lettura: se il roll-down fosse incassato, beta(real~impl) sarebbe ~+1. E' NEGATIVO") print(" (~-1 a una settimana): sul campione la FORMA della curva torna indietro piu' di") print(" quanto la curva rotoli. Il segno del bias da errore-di-misura sarebbe POSITIVO") print(" (iv(T) compare in entrambi con segno opposto), quindi non spiega questo.") TRIALS.append(("Q2: 2 orizzonti x 2 asset (roll implicito vs realizzato)", 4)) sub("(b) il costo reale: bid-ask ATM in PUNTI DI VOL, per tenore") cost = build_atm_cost(chain) cost["bucket"] = pd.cut(cost.dte, [DTE_MIN, 3, 10, 20, 45, 95], labels=["1-3g", "3-10g", "10-20g", "20-45g", "45-95g"]) tab = cost.groupby(["asset", "bucket"], observed=True).agg( n=("spr_vol", "size"), spread_ptvol=("spr_vol", "median"), spread_pct_premio=("spr_usd", lambda s: np.nan), premio_usd=("prem_usd", "median"), vega_usd_per_ptvol=("vega", "median")) tab["spread_pct_premio"] = (cost.assign(x=cost.spr_usd / cost.prem_usd * 100) .groupby(["asset", "bucket"], observed=True)["x"].median()) print(tab.round(3).to_string().replace("\n", "\n ")) sub("(c) l'aritmetica che chiude la domanda") med = cost.groupby(["asset", "bucket"], observed=True)["spr_vol"].median() for a in ASSETS: rd = roll_day.get(a, np.nan) sp = float(med.loc[(a, "10-20g")]) print(f" {a}: roll-down incassabile da CORTO = {rd:+.4f} pt-vol/giorno | " f"un round-trip ATM 10-20g costa {sp:.3f} pt-vol") print(f" => servono {sp/rd:.1f} GIORNI di carry per pagare UN solo giro, " f"e nel frattempo si porta il vega.") TRIALS.append(("Q2: 5 secchi di tenore x 2 asset (costo)", 10)) sub("(d) e a $635 la domanda non e' nemmeno questa: il LOTTO MINIMO") for a in ASSETS: c = cost[(cost.asset == a) & (cost.bucket == "10-20g")] if c.empty: continue prem = float(c.prem_usd.median()) * MIN_LOT[a] vega = float(c.vega.median()) * MIN_LOT[a] spot = float(c.spot.median()) fee = min(OPT_FEE_UNDERLYING * spot, OPT_FEE_CAP_PREMIUM * float(c.prem_usd.median())) * MIN_LOT[a] rd = roll_day.get(a, np.nan) carry_g = rd * vega friz = float(c.spr_vol.median()) * vega + 4 * fee # 1 spread pieno + 4 gambe di fee print(f" {a}: lotto minimo {MIN_LOT[a]} -> premio ATM 10-20g = ${prem:,.2f} " f"({prem/CAPITALE:.0%} del conto), vega ${vega:.2f}/pt-vol, fee ${fee:.2f}/gamba") print(f" carry lordo ${carry_g:+.3f}/giorno per struttura; attrito per giro " f"${friz:.2f} => {friz/carry_g:.0f} giorni per andare in pari") print(f"\n Un calendar spread e' DUE gambe: a $635 una sola struttura ETH impegna il") print(f" ~15-30% del conto e rende centesimi al giorno. BTC e' fuori dal lotto minimo.") print(f" Il bersaglio dichiarato e' 50 EUR/giorno: qui si parla di due ordini di grandezza sotto.") TRIALS.append(("Q2: 2 asset x eseguibilita' al lotto minimo", 2)) # =========================================================================================== # SEZIONE 4 — Q3: gate di rischio sopra TP01. RIDONDANZA PRIMA DELL'UPLIFT. # =========================================================================================== def tp01_positions() -> dict: from src.strategies.trend_portfolio import CANONICAL, TrendPortfolio tp = TrendPortfolio(**CANONICAL) out = {} for a in ASSETS: df = A.get(a, "1d") out[a] = pd.Series(tp.target_series(df), index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))) return out def _anchored(sl_h: pd.Series, px_h: pd.Series, off: int): """Serie giornaliere (pendenza, ritorno) ancorate all'ora `off`.""" m = sl_h.index.hour == off s = sl_h[m] p = px_h.reindex(s.index).ffill() r = p.pct_change() return s, r def sezione4(curve: pd.DataFrame) -> None: hr("SEZIONE 4 — Q3: la pendenza come GATE DI RISCHIO sopra TP01. RIDONDANZA PRIMA DELL'UPLIFT") print(" Regola del progetto: 4 gate di de-risk su 4 (DVOL-spike, macro, funding, breadth) sono") print(" risultati RIDONDANTI col trend. Percio' la ridondanza si misura PRIMA di rivendicare") print(" qualunque uplift: quanti dei giorni che il gate spegne sono giorni in cui TP01 e' gia' flat?") pos = tp01_positions() sub("(a) il fatto che decide da solo la sezione") W = pd.concat(pos, axis=1).loc[TS_START:] flat = (W.abs() < 0.05).mean() print(f" Nella finestra {TS_START.date()} -> oggi ({len(W)} giorni), TP01 e' FLAT nel " f"{flat['BTC']:.1%} (BTC) / {flat['ETH']:.1%} (ETH) dei giorni.") print(f" Esposizione media: BTC {W['BTC'].mean():.3f}, ETH {W['ETH'].mean():.3f} " f"(target TSMOM risk-off, coerente col libro live dichiarato flat).") print(" ⚠️ Un gate di de-risk su un libro gia' flat il 93% del tempo NON E' MISURABILE su") print(" questa finestra: non c'e' rischio da togliere. Questo non e' un difetto del") print(" gate, e' un difetto del CAMPIONE — e va detto prima di qualunque numero.") sub("(b) ridondanza, per quel che il campione consente") for a in ASSETS: sl = slope_hourly(curve, a).resample("1D").last().dropna() j = pd.concat({"slope": sl, "tp": pos[a]}, axis=1, join="inner").dropna() for qq in (0.10, 0.25, 0.40): th = j.slope.quantile(qq) off = j.slope < th already = float((j.tp[off].abs() < 0.05).mean()) if off.any() else np.nan print(f" {a} gate OFF sotto p{int(qq*100)} ({th:+.2f} pt-vol): spegne il " f"{off.mean():.1%} dei giorni; di questi TP01 era GIA' flat nel {already:.1%} " f"| corr(pendenza, esposizione TP01) = {j.slope.corr(j.tp):+.3f}") TRIALS.append((f"Q3 {a}: 3 soglie di gate", 3)) print("\n Il segno di corr(pendenza, esposizione) e' NEGATIVO: la backwardation arriva") print(" quando TP01 e' gia' uscito. E' la firma esatta dei 4 precedenti.") sub("(c) uplift sulla finestra, su TUTTE le 24 ancore (mai su una sola)") print(" Candidato DIREZIONALE (non gate): contango -> long, backwardation -> flat; vol-target") print(" 14g, cap 2x, fee 5 bps/lato. Soglia a ZERO = a priori, nessun parametro tarato.") for a in ASSETS: sl_h, px_h = slope_hourly(curve, a), px_hourly(a) def cand(off, s=sl_h, p=px_h): s_d, r_d = _anchored(s, p, off) d = pd.DataFrame({"slope": s_d, "r": r_d}).dropna() if len(d) < 30: return pd.Series(dtype=float) vol = d["r"].rolling(14).std() * np.sqrt(365) dirn = (d["slope"] > 0).astype(float) tgt = np.clip(dirn * (0.20 / vol.replace(0, np.nan)), 0, 2.0).fillna(0.0) hold = tgt.shift(1).fillna(0.0) return (hold * d["r"] - A.FEE_SIDE * hold.diff().abs().fillna(0.0)).dropna() def base(off, p=px_h, s=sl_h): s_d, r_d = _anchored(s, p, off) d = pd.DataFrame({"r": r_d}).dropna() return d["r"] band = A.anchor_luck_band(cand, list(range(24)), canonical=0) print(f" {a}: Sharpe candidato — canonico(00:00) {band['canonical']:+.2f} " f"(pctl {band['canonical_pctl']:.2f}) | MEDIANA ONESTA {band['median']:+.2f} | " f"banda [{band['lo']:+.2f}, {band['hi']:+.2f}] | positivo in {band['frac_positive']:.0%} " f"delle 24 ancore | gate_pass={band['gate_pass']}") bh = A.anchor_luck_band(base, list(range(24)), canonical=0) print(f" buy&hold sulla stessa finestra/ancore: mediana {bh['median']:+.2f} " f"[{bh['lo']:+.2f}, {bh['hi']:+.2f}] <-- il vero termine di paragone in un toro") dlt = A.anchor_luck_delta(cand, base, list(range(24))) print(f" mediana delle DIFFERENZE APPAIATE candidato-B&H: {dlt['median_paired']:+.2f} " f"(positiva in {dlt['n_positive']}/{dlt['n_anchors']} ancore)") TRIALS.append((f"Q3 {a}: 24 ancore x 2 varianti", 48)) sub("(d) null del DE-LEVERING — obbligatorio su ogni claim di minor rischio") for a in ASSETS: sl_h, px_h = slope_hourly(curve, a), px_hourly(a) s_d, r_d = _anchored(sl_h, px_h, 0) d = pd.DataFrame({"slope": s_d, "r": r_d}).dropna() vol = d["r"].rolling(14).std() * np.sqrt(365) tgt = np.clip((d["slope"] > 0).astype(float) * (0.20 / vol.replace(0, np.nan)), 0, 2.0).fillna(0.0) hold = tgt.shift(1).fillna(0.0) net = (hold * d["r"] - A.FEE_SIDE * hold.diff().abs().fillna(0.0)).dropna() dd_c = A._dd_ret(net) best = None for k in np.arange(0.05, 1.01, 0.05): b = k * d["r"].reindex(net.index) if A._dd_ret(b) <= dd_c: best = (k, A._sh(b), A._dd_ret(b)) break print(f" {a}: candidato Sharpe {A._sh(net):+.2f} maxDD {dd_c:.2%} | " + (f"buy&hold de-leverato k={best[0]:.2f} -> stesso DD ({best[2]:.2%}) con Sharpe " f"{best[1]:+.2f} => {'IL CANDIDATO NON SOPRAVVIVE' if best[1] >= A._sh(net) else 'candidato sopra il null'}" if best else "nessun k<=1 raggiunge quel DD")) TRIALS.append((f"Q3 {a}: 20 valori di k nel null de-levering", 20)) # =========================================================================================== # SEZIONE 5 — IL TEST LUNGO. La pendenza ha un PROXY con 5 anni di storia? # =========================================================================================== def _proxy_slope(df: pd.DataFrame, asset: str, rvwin: int = 7) -> np.ndarray: """Proxy A PRIORI della pendenza su storia lunga: DVOL(30g) - RV(rvwin), tutto causale. Motivazione, non scelta a posteriori: iv_30 ~ DVOL (corr 0.99, SEZIONE 1) e iv_7 e' in larga parte una funzione della RV recente (SEZIONE 2c). Il proxy e' quindi la pendenza con la gamba corta sostituita dalla sua determinante. NON e' la stessa variabile: l'attenuazione e' misurata sotto e va portata dietro a ogni conclusione. """ c = df["close"].to_numpy(float) r = A.simple_returns(c) bpd = A.bars_per_day(df) rv = A.realized_vol(r, max(2, rvwin * bpd), bpd * 365.25) * 100.0 return A.dvol(df, asset) - rv def make_proxy_dir(long_only: bool = True, thr: float = 0.0): def fn(df, asset): p = _proxy_slope(df, asset) d = np.where(np.isfinite(p), np.where(p > thr, 1.0, 0.0 if long_only else -1.0), 0.0) return A.vol_target(d, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) return fn def sezione5(curve: pd.DataFrame) -> None: hr("SEZIONE 5 — IL TEST LUNGO: la pendenza ha un proxy con 5 anni di storia?") print(" Con 74 giorni nessun gate statistico ha potenza. L'unica via per DIRE qualcosa e' un") print(" proxy a priori su storia lunga — e dichiararne l'attenuazione.") sub("(a) quanto e' buono il proxy (misurato sulla finestra vera, non assunto)") for a in ASSETS: sl = slope_hourly(curve, a).resample("1D").last().dropna() df = A.get(a, "1d") p = pd.Series(_proxy_slope(df, a), index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))).dropna() j = pd.concat({"vero": sl, "proxy": p}, axis=1, join="inner").dropna() c = float(j.vero.corr(j.proxy)) print(f" {a}: corr(pendenza VERA, DVOL-RV7) = {c:+.3f} su n={len(j)} giorni " f"-> attenuazione: il proxy spiega il {c**2:.0%} della varianza della pendenza") print(" ⚠️ Un proxy a corr ~0.5-0.7 e' una lente SFOCATA: un risultato NEGATIVO sul proxy") print(" e' evidenza piu' debole di un negativo sulla variabile vera. Va detto, non nascosto.") sub("(b) il candidato direzionale sul proxy, 5 anni, coi gate veri") fn = make_proxy_dir(long_only=True) cau = A.causality_ok(fn, tf="1d") print(f" causality_ok: {cau}") cd = A.candidate_daily(fn, tf="1d") era = cd[cd.index >= pd.Timestamp("2021-10-01", tz="UTC")] print(f" era DVOL (2021-10 -> oggi, {len(era)} giorni): Sharpe {A._sh(era):+.3f} " f"maxDD {A._dd_ret(era):.2%}") rep = A.marginal_vs_tp01(cd) print(f" marginal_vs_tp01 -> {rep.get('marginal_verdict')} | corr a TP01 " f"{rep.get('corr_full')} | uplift w25 full {rep['blends']['w25']['uplift_full']:+.3f} " f"hold {rep['blends']['w25']['uplift_hold']}") for k in ("is_hedge", "has_insample_edge", "robust_oos", "beats_noise_null", "earns_slot"): if k in rep: print(f" {k:20s} = {rep[k]}") imp = A.implausible_sharpe(era) print(f" implausible_sharpe: flagged={imp.get('flagged')} ({imp.get('reasons')})") TRIALS.append(("Sez.5: 2 varianti proxy (long-only, L/S) x 1 soglia", 2)) sub("(c) il precedente che questo replica") print(" `dvol_directional.py` (2026-06-29) testo' VRP-Z = z-score causale di (IV-RV) come") print(" segnale DIREZIONALE su 5 anni, con la griglia completa: verdetto **HEDGE,") print(" earns_slot=False** — paga solo quando TP01 e' debole, quindi non e' alpha.") print(" La pendenza della term structure e', su questa lente, la stessa variabile con la") print(" gamba realizzata sostituita da una implicita. Il filone non e' nuovo: e' quello.") # =========================================================================================== # SEZIONE 6 — CONTI, GATE, VERDETTO # =========================================================================================== def sezione6() -> None: hr("SEZIONE 6 — GRIGLIA DICHIARATA E DEFLATED SHARPE") tot = sum(n for _, n in TRIALS) for name, n in TRIALS: print(f" {n:5d} {name}") print(f" {'-'*70}\n {tot:5d} TOTALE (contato al RIALZO: ogni variante valutata, anche scartata)") print("\n Non e' stata ridotta nessuna griglia per il budget: il vincolo di questo filone") print(" non e' il tempo di calcolo, e' che il campione dura 74 giorni.") print("\n deflated_sharpe: NON GIRATO su un candidato di questo studio, e il motivo e' una") print(" scelta, non una dimenticanza: il DSR e' una correzione per multiple-testing su UNA") print(" serie di ritorni; qui il candidato migliore ha 74 osservazioni e la sua incertezza") print(" di CAMPIONE (SE(Sharpe) ~ 2.2) domina di un ordine di grandezza quella da selezione.") print(f" Deflazionare {tot} trial su 74 giorni darebbe un numero preciso e privo di senso.") print(" Sul proxy a 5 anni il DSR e' invece pertinente ed e' riportato dal gate del 29/06") print(" su cui questo filone ricade (VRP-Z: earns_slot=False).") def main() -> None: hr("r0822_term_structure — TERM STRUCTURE DELLA VOL IMPLICITA BTC/ETH") print(__doc__.split("METODO.")[0].strip()[:0] or "", end="") chain = load_quoted_chain() sezione0(chain) atm_exp = build_atm_by_expiry(chain) curve = build_curve(atm_exp) sezione1(curve) sezione2(curve) sezione3(chain, atm_exp, curve) del chain sezione4(curve) sezione5(curve) sezione6() hr("FINE") if __name__ == "__main__": main()