#!/usr/bin/env python """r0822_oi_pin.py — OI-PIN: il prezzo e' attratto verso gli strike a massimo open interest nelle ore che precedono la scadenza Deribit? (ondata 2026-08-22, filone OI-PIN) CONTESTO / PERCHE' QUESTO FILONE NON E' "CALENDARIO GIA' MORTO" -------------------------------------------------------------- Il 02/07 il progetto ha bocciato il **calendario** delle scadenze Deribit (drift post-expiry weekly/monthly/quarterly): 0/24 celle a Bonferroni, e il pattern si INVERTE proprio sul quarterly, dove l'open interest massimo dovrebbe amplificarlo. Quel test guardava le DATE. Qui si guarda l'OPEN INTEREST VERO, per strike, ora per ora (`data/raw/cb_chain/`): il meccanismo proposto non e' "il venerdi' e' speciale" ma "il market maker corto gamma ricompra/rivende verso lo strike dove sta la sua esposizione". E' un meccanismo NON di calendario, quindi ammissibile secondo la sezione 5 del brief. IPOTESI (dichiarata PRIMA di misurare) -------------------------------------- H1 Il prezzo deriva VERSO il livello a massima concentrazione di OI / gamma / max-pain nelle ultime ore prima della scadenza (statistica `toward = sign(K - S_t) * r_{t->T}` > 0). H2 L'effetto e' piu' forte a poche ore dalla scadenza e sulle scadenze grandi (venerdi'). H0-ATTESA DELL'AUTORE: **negativa**. Motivi dichiarati in anticipo: (a) il crypto e' 24/7, senza il "chiusura di borsa" che genera il pinning azionario classico; (b) l'OI Deribit e' grande ma il perp/spot che lo dovrebbe muovere e' molto piu' grande; (c) il progetto ha gia' misurato che ogni effetto di finestra su BTC/ETH e' rumore; (d) e soprattutto: `K_oi` sta quasi sempre VICINO allo spot, quindi qualunque statistica "il prezzo va verso K" e' contaminata dalla POSIZIONE del livello, non dal suo OI. Per questo il null location-matched (regola codificata il 02/07 sui livelli di Fibonacci) non e' un contorno: e' IL test. CONVENZIONE CAUSALE (verificata da un assert nel codice) -------------------------------------------------------- Barre 1h open-labeled: la barra etichettata `tau` copre [tau, tau+1h) e il suo close e' il prezzo a `tau+1h`. * snapshot catena all'ora `hh` -> righe con ts in [hh, hh+1h) * decisione a fine barra `hh` -> prezzo eseguibile = close(hh) = prezzo a hh+1h * quindi entrata a `hte` ore dalla scadenza <=> snapshot a dte = hte+1 ore * uscita al prezzo di regolamento = close della barra (E-1h) = prezzo a E (08:00 UTC) `A.causality_ok` NON e' applicabile (il segnale non e' funzione del solo feed OHLC): la causalita' e' garantita per costruzione e verificata da `_assert_causalita()`. Uso: nice -n 19 timeout 900 uv run python scripts/research/r0822_oi_pin.py """ from __future__ import annotations import glob import sys 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 STORE = ROOT / "data" / "raw" / "cb_chain" ASSETS = ("BTC", "ETH") FEE_SIDE_REAL = 0.00035 # 3,5 bps/lato = fee Deribit REALE del libro (7 bps RT) FEE_SIDE_CONS = A.FEE_SIDE # 5 bps/lato = convenzione conservativa dei backtest del progetto MAX_H = 50 # ore prima della scadenza da tenere in memoria HTE_GRID = (1, 2, 3, 6, 12, 24, 36, 48) # ore-alla-scadenza all'INGRESSO RULES = ("oi", "gex", "mp") # max OI · max gamma*OI · max pain PLACEBOS_GLOB = ("rand", "mid", "mr7") # non dipendono dalla regola PLACEBOS_RULE = ("adj", "jit") # costruiti sulla distanza DELLA regola SEED = 20260822 RNG = np.random.default_rng(SEED) # =========================================================================== dati def _files() -> list[str]: fs = sorted(glob.glob(str(STORE / "*.parquet"))) if not fs: raise FileNotFoundError(f"{STORE} vuoto o assente") return fs def load_near(max_h: int = MAX_H) -> pd.DataFrame: """Catena filtrata alle sole ore vicine a scadenza, letta file per file (RAM: 7 GB in due).""" cols = ["asset", "strike", "option_type", "ts", "exp", "open_interest", "gamma", "quote_status"] out = [] for p in _files(): d = pd.read_parquet(p, columns=cols) d["ts"] = pd.to_datetime(d["ts"], utc=True) d["exp"] = pd.to_datetime(d["exp"], utc=True) d["hh"] = d["ts"].dt.floor("h") d["dte_h"] = (d["exp"] - d["hh"]).dt.total_seconds() / 3600.0 d = d[(d["dte_h"] > 0) & (d["dte_h"] <= max_h)] if len(d): out.append(d) del d d = pd.concat(out, ignore_index=True) # una riga presente non e' un dato presente: OI mancante o quota rotta non conta d = d[d["open_interest"].notna() & (d["open_interest"] > 0)] d = (d.sort_values("ts") .drop_duplicates(subset=["asset", "exp", "hh", "strike", "option_type"], keep="last") .reset_index(drop=True)) return d def spot_1h() -> dict[str, pd.Series]: s = {} for a in ASSETS: df = A.get(a, "1h") s[a] = pd.Series(df["close"].to_numpy(float), index=pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True))) return s # =========================================================================== livelli def _max_pain(K: np.ndarray, oi_c: np.ndarray, oi_p: np.ndarray) -> float: """Strike che minimizza il payout complessivo verso i detentori (max pain classico).""" d = K[None, :] - K[:, None] # d[i,j] = K_j - K_i (K_i = candidato) pay = (np.maximum(-d, 0.0) * oi_c[None, :]).sum(1) + (np.maximum(d, 0.0) * oi_p[None, :]).sum(1) return float(K[int(np.argmin(pay))]) def levels_table(chain: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFrame: """Un record per (asset, scadenza, ora): i 3 livelli candidati + i 4 placebo di strike + le misure di concentrazione. I placebo sono LOCATION-MATCHED (regola 02/07).""" rows = [] piv = (chain.groupby(["asset", "exp", "hh", "strike", "option_type"], sort=False) .agg(oi=("open_interest", "sum"), gm=("gamma", "mean")).reset_index()) for (asset, exp, hh), g in piv.groupby(["asset", "exp", "hh"], sort=False): S = spot[asset].asof(hh) # prezzo a hh+1h = close(barra hh) if not np.isfinite(S) or S <= 0: continue K = np.sort(g["strike"].unique().astype(float)) if len(K) < 3: continue oi_c = np.zeros(len(K)); oi_p = np.zeros(len(K)); gx = np.zeros(len(K)) idx = {k: i for i, k in enumerate(K)} for k, ot, o, gm in zip(g["strike"].to_numpy(float), g["option_type"].to_numpy(), g["oi"].to_numpy(float), g["gm"].to_numpy(float)): i = idx[float(k)] (oi_c if ot == "C" else oi_p)[i] += o gx[i] += o * (gm if np.isfinite(gm) else 0.0) oi = oi_c + oi_p tot = oi.sum() if tot <= 0: continue i_oi = int(np.argmax(oi)) lv = { "oi": float(K[i_oi]), "gex": float(K[int(np.argmax(gx))]) if gx.max() > 0 else np.nan, "mp": _max_pain(K, oi_c, oi_p), } # ---- placebo LOCATION-MATCHED, uno per OGNI regola -------------------------------- # ⚠ difetto trovato e corretto in sessione: la prima stesura costruiva `adj`/`jit` # attorno al SOLO max-OI e poi li usava come null anche per max-pain e gamma. Ma # |K_oi - S| mediano e' ~2,4% e |K_mp - S| ~1,1%: confrontare mp con un placebo # tarato sulla distanza di oi NON e' location-matched, e' proprio l'errore che il # null doveva prevenire. Ogni regola ha ora i placebo alla PROPRIA distanza. dist = np.abs(K - S) for nm, kv in list(lv.items()): if not np.isfinite(kv): out_adj = out_jit = np.nan else: i_r = int(np.argmin(np.abs(K - kv))) j = i_r + (1 if (i_r == 0 or (i_r < len(K) - 1 and RNG.random() < 0.5)) else -1) out_adj = float(K[int(np.clip(j, 0, len(K) - 1))]) d0 = abs(kv - S) cand = np.where((dist >= 0.5 * d0) & (dist <= 1.5 * d0) & (K != kv))[0] out_jit = float(K[cand[RNG.integers(len(cand))]]) if len(cand) else np.nan lv[f"adj_{nm}"] = out_adj lv[f"jit_{nm}"] = out_jit # placebo GLOBALI (non dipendono dalla regola) # rand: permutare l'OI fra gli strike e riprenderne l'argmax equivale ESATTAMENTE a # estrarre uno strike uniforme dalla griglia (l'argmax di un vettore permutato e' # uniforme sulle posizioni) -> lo si implementa cosi', ed e' esatto, non simulato. lv["rand"] = float(K[RNG.integers(len(K))]) # mid: centro della griglia quotata = puro artefatto di copertura, zero informazione OI lv["mid"] = float(0.5 * (K[0] + K[-1])) rows.append(dict( asset=asset, exp=exp, hh=hh, dte_h=float((exp - hh).total_seconds() / 3600.0), S=float(S), n_k=len(K), tot_oi=float(tot), top_share=float(oi[i_oi] / tot), hhi=float(((oi / tot) ** 2).sum()), k_step=float(np.median(np.diff(K))) if len(K) > 1 else np.nan, **{f"L_{k}": v for k, v in lv.items()}, )) return pd.DataFrame(rows) # =========================================================================== il FATTO def build_events(lev: pd.DataFrame, spot: dict[str, pd.Series]) -> pd.DataFrame: """Un evento per (asset, scadenza, hte): livelli, prezzo d'ingresso, prezzo di regolamento.""" ev = [] mr7 = {a: spot[a].rolling(24 * 7, min_periods=24).mean() for a in ASSETS} for a in ASSETS: s = spot[a] sub = lev[lev["asset"] == a] for exp, g in sub.groupby("exp"): t_set = pd.Timestamp(exp) - pd.Timedelta(hours=1) # barra il cui close e' a E if t_set not in s.index: continue S_T = float(s.loc[t_set]) gi = g.set_index("hh") for hte in HTE_GRID: hh = pd.Timestamp(exp) - pd.Timedelta(hours=hte + 1) if hh not in gi.index or hh not in s.index: continue r = gi.loc[hh] S = float(s.loc[hh]) # close(hh) = prezzo a E-hte if not np.isfinite(S) or S <= 0: continue rec = dict(asset=a, exp=exp, hte=hte, hh=hh, S=S, S_T=S_T, ret=S_T / S - 1.0, n_k=int(r["n_k"]), tot_oi=float(r["tot_oi"]), top_share=float(r["top_share"]), hhi=float(r["hhi"]), k_step=float(r["k_step"]), dow=int(pd.Timestamp(exp).dayofweek)) lvls = {c[2:]: r[c] for c in r.index if str(c).startswith("L_")} lvls["mr7"] = float(mr7[a].asof(hh)) for nm, K in lvls.items(): K = float(K) if K is not None else np.nan if not np.isfinite(K) or K <= 0: rec[f"d_{nm}"] = np.nan; rec[f"tw_{nm}"] = np.nan rec[f"sh_{nm}"] = np.nan continue d = (K - S) / S rec[f"K_{nm}"] = K rec[f"d_{nm}"] = d rec[f"tw_{nm}"] = np.sign(d) * rec["ret"] # verso il livello rec[f"sh_{nm}"] = (abs(S - K) - abs(S_T - K)) / S # avvicinamento ev.append(rec) E = pd.DataFrame(ev) # NULL STATICO CAUSALE (lezione 25/07: il primo null statico di STATARB usava # sign(mean(segnale)) su tutto il campione = look-ahead; qui la maggioranza e' ESPANDENTE # e shiftata). Risponde a: "il candidato e' solo 'sempre nello stesso verso' prima della # scadenza?" — che su 3,7 mesi di mercato direzionale e' l'ipotesi piu' probabile. E = E.sort_values(["asset", "hte", "hh"]).reset_index(drop=True) sgn = np.sign(E["d_mp"].to_numpy(float)) maj = (pd.Series(sgn).groupby([E["asset"], E["hte"]]) .transform(lambda x: x.expanding().mean().shift(1))) E["d_maj"] = np.where(np.isfinite(maj), np.sign(maj) * 0.01, np.nan) E["tw_maj"] = np.sign(E["d_maj"]) * E["ret"] E["sh_maj"] = np.nan E["d_long"] = 0.01; E["tw_long"] = E["ret"]; E["sh_long"] = np.nan E["d_short"] = -0.01; E["tw_short"] = -E["ret"]; E["sh_short"] = np.nan return E def _assert_causalita(chain: pd.DataFrame, ev: pd.DataFrame) -> str: """Il livello usato a `hh` deve venire da quote osservate PRIMA del prezzo d'ingresso.""" bad = int((chain["ts"] < chain["hh"]).sum() + (chain["ts"] >= chain["hh"] + pd.Timedelta("1h")).sum()) assert bad == 0, "snapshot fuori dalla propria ora" dte = (pd.to_datetime(ev["exp"]) - pd.to_datetime(ev["hh"])).dt.total_seconds() / 3600.0 assert bool((dte == ev["hte"] + 1).all()), "hte non coerente con lo snapshot" assert bool(((pd.to_datetime(ev["hh"]) + pd.Timedelta(hours=1)) <= pd.to_datetime(ev["exp"])).all()), "ingresso oltre la scadenza" return (f"OK — 0/{len(chain)} righe di catena fuori dalla propria ora; " f"{len(ev)} eventi con dte == hte+1 (ingresso = close(hh), quota vista in [hh,hh+1h))") # =========================================================================== inferenza def cluster_boot(x: np.ndarray, clusters: np.ndarray, n: int = 2000, seed: int = SEED): """IC95 della media con ricampionamento dei CLUSTER (una scadenza = 1 cluster: BTC ed ETH della stessa scadenza non sono osservazioni indipendenti).""" x = np.asarray(x, float) m = np.isfinite(x) x, clusters = x[m], np.asarray(clusters)[m] if len(x) < 5: return np.nan, np.nan, np.nan, 0 uc = np.unique(clusters) groups = [x[clusters == c] for c in uc] rng = np.random.default_rng(seed) draws = np.empty(n) for i in range(n): pick = rng.integers(0, len(groups), len(groups)) draws[i] = np.mean(np.concatenate([groups[j] for j in pick])) return float(np.mean(x)), float(np.percentile(draws, 2.5)), float(np.percentile(draws, 97.5)), len(uc) def mde(x: np.ndarray, clusters: np.ndarray) -> float: """Effetto minimo rilevabile (alpha .05 bilaterale, potenza 80%) con questo campione.""" x = np.asarray(x, float) m = np.isfinite(x) x, cl = x[m], np.asarray(clusters)[m] if len(x) < 3: return np.nan nc = max(len(np.unique(cl)), 1) return float(2.802 * np.std(x) / np.sqrt(nc)) # =========================================================================== strategia _W: dict = {} def win_1h(asset: str, lo: pd.Timestamp, hi: pd.Timestamp) -> pd.DataFrame: """Feed 1h tagliato alla finestra della catena: fuori da li' la posizione e' zero per costruzione, e valutare 70k barre 288 volte costerebbe minuti per nulla.""" key = (asset, lo, hi) if key not in _W: df = A.get(asset, "1h") dt = pd.DatetimeIndex(pd.to_datetime(df["datetime"], utc=True)) m = (dt >= lo - pd.Timedelta("2D")) & (dt <= hi + pd.Timedelta("2D")) _W[key] = df.loc[m].reset_index(drop=True) return _W[key] def hourly_target(ev: pd.DataFrame, asset: str, rule: str, hte: int, band: tuple[float, float] | None, sizing: str, subsample: str, df1h: pd.DataFrame) -> np.ndarray: """Posizione oraria: entra `hte` ore prima della scadenza nel verso del livello, esce a scadenza. Se piu' finestre si sovrappongono vince la scadenza PIU' VICINA (una posizione alla volta = cio' che il conto Deribit puo' davvero fare: strumento unico, netting).""" idx = pd.DatetimeIndex(pd.to_datetime(df1h["datetime"], utc=True)) pos = np.zeros(len(idx)) loc = pd.Series(np.arange(len(idx)), index=idx) e = ev[(ev["asset"] == asset) & (ev["hte"] == hte)].copy() if subsample == "fri": e = e[e["dow"] == 4] elif subsample == "big": e = e[e["tot_oi"] >= e["tot_oi"].median()] d = e[f"d_{rule}"].to_numpy(float) keep = np.isfinite(d) if band is not None: keep &= (np.abs(d) >= band[0]) & (np.abs(d) <= band[1]) e = e[keep] d = d[keep] if len(e) == 0: return pos w = np.sign(d) if sizing == "sign" else np.clip(d / 0.01, -1.0, 1.0) order = np.argsort(e["exp"].to_numpy())[::-1] # lontane prima: la scadenza VICINA sovrascrive for j in order: hh = pd.Timestamp(e["hh"].iloc[j]); exp = pd.Timestamp(e["exp"].iloc[j]) last = exp - pd.Timedelta(hours=2) if hh not in loc.index or last not in loc.index: continue i0, i1 = int(loc[hh]), int(loc[last]) if i1 >= i0: pos[i0:i1 + 1] = w[j] return pos def strat_daily(ev: pd.DataFrame, rule: str, hte: int, band, sizing: str, subsample: str, fee_side: float) -> tuple[pd.Series, dict]: """Serie giornaliera netta 50/50 BTC+ETH + metriche, sulla sola finestra della catena.""" nets, info = {}, {} lo_w = pd.Timestamp(ev["hh"].min()); hi_w = pd.Timestamp(ev["exp"].max()) for a in ASSETS: df = win_1h(a, lo_w, hi_w) tgt = hourly_target(ev, a, rule, hte, band, sizing, subsample, df) r = A.eval_weights(df, tgt, fee_side=fee_side) nets[a] = pd.Series(r["net"], index=r["idx"]) info[a] = int(np.sum(np.abs(np.diff(np.r_[0.0, tgt])) > 1e-12)) J = pd.concat(nets, axis=1, join="inner").fillna(0.0) daily = A._to_daily(0.5 * J[ASSETS[0]] + 0.5 * J[ASSETS[1]]) lo = pd.Timestamp(ev["hh"].min()).normalize() daily = daily[daily.index >= lo] return daily, info def metr(s: pd.Series) -> dict: r = s.dropna() if len(r) < 5: return dict(sharpe=np.nan, dd=np.nan, cagr=np.nan, n=len(r)) eq = np.cumprod(1 + r.to_numpy(float)); pk = np.maximum.accumulate(eq) yrs = max((r.index[-1] - r.index[0]).total_seconds() / 86400 / 365.25, 1e-6) return dict(sharpe=A._sh(r), dd=float(np.max((pk - eq) / pk)), cagr=float(eq[-1] ** (1 / yrs) - 1), n=int(len(r))) # =========================================================================== main def main() -> None: P = print P("=" * 100) P("OI-PIN — pinning verso gli strike a massimo open interest (catena Deribit oraria)") P("=" * 100) # ---------------------------------------------------------------- 1. DATO chain = load_near() spot = spot_1h() lev = levels_table(chain, spot) ev = build_events(lev, spot) P("\n[1] DATO") P(f" righe catena a <= {MAX_H}h dalla scadenza : {len(chain):,}") P(f" scadenze distinte : {ev.groupby('asset').exp.nunique().to_dict()}") P(f" finestra : {ev.hh.min()} -> {ev.exp.max()}") P(f" eventi (asset x scadenza x hte) : {len(ev)}") P(f" strike per snapshot mediana/min : {lev.n_k.median():.0f} / {lev.n_k.min():.0f}") P(f" causalita' : {_assert_causalita(chain, ev)}") P(" ⚠ il collettore scarta gli strumenti con OI<100 -> la griglia e' TRONCATA sulle code:") P(" il max-OI ne soffre poco (e' un argmax), il MAX PAIN si', perche' e' una somma su tutta") P(" la catena. Il verdetto su 'mp' va letto come 'max pain sulla catena OI>=100'.") # ---------------------------------------------------------------- 2. POTENZA (PRIMA del risultato) P("\n[2] POTENZA DICHIARATA — calcolata PRIMA di leggere qualunque effetto") P(" Unita' statistica = una SCADENZA (BTC ed ETH della stessa data non sono indipendenti).") P(f" {'hte':>4} {'n oss':>6} {'n clus':>7} {'sd(ret)':>9} {'MDE medio/trade':>16} {'MDE ann.':>10}") for hte in HTE_GRID: e = ev[ev["hte"] == hte] m = mde(e["ret"].to_numpy(float), e["exp"].to_numpy()) ann = m * (365.25 * 24 / hte) if np.isfinite(m) else np.nan P(f" {hte:>4} {len(e):>6} {e.exp.nunique():>7} {np.nanstd(e['ret']):>9.4f}" f" {m:>16.4f} {ann:>9.1%}") P(" LETTURA OBBLIGATORIA: il campione rileva solo effetti ENORMI. A 24h l'MDE e' ~0,5-0,7%") P(" per trade contro 0,07% di fee: un pinning realistico (pochi bps) e' INVISIBILE qui.") P(" Quindi un 'non significativo' NON prova assenza; un 'significativo' sarebbe cosi' grande") P(" da essere sospetto. Il verdetto massimo raggiungibile e' LEAD.") # ---------------------------------------------------------------- 3. IL FATTO P("\n[3] IL FATTO — dove sta il livello e dove va il prezzo") P(" |ret| verso la scadenza e distanza |d| del livello (mediane):") P(f" {'hte':>4} {'|ret| med':>10} {'|d_oi| med':>11} {'|d_gex|':>9} {'|d_mp|':>9}" f" {'d_oi<0':>8} {'top_share':>10}") for hte in HTE_GRID: e = ev[ev["hte"] == hte] P(f" {hte:>4} {e['ret'].abs().median():>10.4f} {e['d_oi'].abs().median():>11.4f}" f" {e['d_gex'].abs().median():>9.4f} {e['d_mp'].abs().median():>9.4f}" f" {(e['d_oi'] < 0).mean():>8.2f} {e['top_share'].median():>10.3f}") P(" ⚠ se 'd_oi<0' e' lontano da 0.50 il livello ha un BIAS DI SEGNO (griglia troncata o") P(" mercato in trend): una statistica 'il prezzo va verso K' lo erediterebbe.") P("\n Statistica di pinning tw = sign(K-S)*ret (media, IC95 cluster-bootstrap su scadenze):") P(f" {'hte':>4} {'rule':>5} {'media tw':>10} {'IC95 lo':>9} {'IC95 hi':>9} {'sig':>4}" f" {'shrink med':>11} {'n':>5}") fact = [] for hte in HTE_GRID: e = ev[ev["hte"] == hte] for rl in RULES: m, lo, hi, nc = cluster_boot(e[f"tw_{rl}"].to_numpy(float), e["exp"].to_numpy()) sg = "***" if (np.isfinite(lo) and lo > 0) else ("neg" if (np.isfinite(hi) and hi < 0) else "-") P(f" {hte:>4} {rl:>5} {m:>10.5f} {lo:>9.5f} {hi:>9.5f} {sg:>4}" f" {np.nanmedian(e[f'sh_{rl}']):>11.5f} {int(np.isfinite(e[f'tw_{rl}']).sum()):>5}") fact.append(dict(hte=hte, rule=rl, mean=m, lo=lo, hi=hi)) F = pd.DataFrame(fact) P(f" celle con IC95 tutto sopra zero: {int((F.lo > 0).sum())}/{len(F)}" f" | tutto sotto zero: {int((F.hi < 0).sum())}/{len(F)}") P("\n LA STESSA STATISTICA SUI LIVELLI PLACEBO — se sale anche li', NON e' pinning:") P(f" {'hte':>4} " + " ".join(f"{('tw_' + n):>10}" for n in ("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short"))) for hte in HTE_GRID: e = ev[ev["hte"] == hte] P(f" {hte:>4} " + " ".join( f"{np.nanmean(e[f'tw_{n}']):>10.5f}" for n in ("oi", "mp", "rand", "jit_mp", "mid", "mr7", "maj", "short"))) P("\n QUANTO SONO LO STESSO SEGNALE? accordo del SEGNO di (K-S) fra max-pain e i placebo:") for n in ("mid", "mr7", "oi", "maj"): agr = [float(np.nanmean(np.sign(ev.loc[ev.hte == h, "d_mp"]) == np.sign(ev.loc[ev.hte == h, f"d_{n}"]))) for h in HTE_GRID] P(f" mp vs {n:<5}: " + " ".join(f"h{h}={a:.2f}" for h, a in zip(HTE_GRID, agr))) P(f" base rate sign<0 : " + " ".join( f"{n}={np.nanmean(np.sign(ev[f'd_{n}']) < 0):.2f}" for n in ("oi", "gex", "mp", "mid", "mr7"))) for a in ASSETS: sp_ = spot[a] w = sp_[(sp_.index >= ev.hh.min()) & (sp_.index <= ev.exp.max())] P(f" deriva {a} nella finestra: {w.iloc[-1] / w.iloc[0] - 1:+.1%}" f" (se il livello sta quasi sempre da un lato, il candidato e' una scommessa direzionale)") # ---------------------------------------------------------------- 4. NULL LOCATION-MATCHED P("\n[4] NULL LOCATION-MATCHED (regola 02/07: e' la POSIZIONE o e' l'OI?)") P(" Confronto APPAIATO per evento: tw(livello speciale) - tw(placebo alla stessa distanza).") P(" placebo PER REGOLA: adj_r = strike adiacente a K_r · jit_r = strike arbitrario a") P(" distanza 0.5-1.5x |K_r - S| (analogo del Fib±jitter del 02/07).") P(" placebo GLOBALI: rand = strike uniforme (== permutazione dell'OI, equivalenza esatta) ·") P(" mid = centro della griglia quotata (zero info OI) · mr7 = media 7g dello spot.") P(f" {'hte':>4} {'rule':>5} {'Δ vs adj_r':>12} {'Δ vs jit_r':>12}" f" {'Δ vs rand':>12} {'Δ vs mid':>12} {'Δ vs mr7':>12}") npos = ntot = 0 for hte in HTE_GRID: e = ev[ev["hte"] == hte] for rl in RULES: cells = [] for p in (f"adj_{rl}", f"jit_{rl}") + PLACEBOS_GLOB: dd = e[f"tw_{rl}"].to_numpy(float) - e[f"tw_{p}"].to_numpy(float) m, lo, hi, _ = cluster_boot(dd, e["exp"].to_numpy()) cells.append(f"{m:>11.5f}{'*' if (np.isfinite(lo) and lo > 0) else ' '}") ntot += 1 npos += int(np.isfinite(lo) and lo > 0) P(f" {hte:>4} {rl:>5} " + " ".join(cells)) P(f" celle (rule x hte x placebo) in cui lo SPECIALE batte il placebo con IC95>0:" f" {npos}/{ntot} (atteso per caso ~{0.025 * ntot:.0f})") P(" LETTURA: `adj_r`/`jit_r` chiedono 'e' proprio QUELLO strike?'; `mid`/`mr7` chiedono") P(" 'serve l'open interest, o basta il centro recente del prezzo?'. La seconda e' la") P(" domanda che uccide o salva il filone.") # ---------------------------------------------------------------- 5. CONDIZIONAMENTO P("\n[5] CONDIZIONAMENTO — la concentrazione di OI e le scadenze grandi cambiano qualcosa?") P(f" {'hte':>4} {'gruppo':>16} {'media tw_oi':>12} {'IC95 lo':>9} {'IC95 hi':>9} {'n':>4}") for hte in (3, 12, 24): e = ev[ev["hte"] == hte] hi_c = e["top_share"] >= e["top_share"].median() big = e["tot_oi"] >= e["tot_oi"].quantile(0.75) for nm, sel in (("conc ALTA", hi_c), ("conc BASSA", ~hi_c), ("OI top-25%", big), ("venerdi'", e["dow"] == 4)): x = e.loc[sel, "tw_oi"].to_numpy(float) m, lo, hh_, _ = cluster_boot(x, e.loc[sel, "exp"].to_numpy()) P(f" {hte:>4} {nm:>16} {m:>12.5f} {lo:>9.5f} {hh_:>9.5f} {int(np.isfinite(x).sum()):>4}") # ---------------------------------------------------------------- 6. STRATEGIA P("\n[6] STRATEGIA — expectancy netta fee (3,5 bps/lato reali) e serie giornaliera") bands = {"nessuna": None, "0.2-3%": (0.002, 0.03)} subs = ("all", "big", "fri") sizings = ("sign", "prop") cells, n_trials = [], 0 for rl in RULES: for hte in HTE_GRID: for bn, bd in bands.items(): for sz in sizings: for sb in subs: n_trials += 1 d, info = strat_daily(ev, rl, hte, bd, sz, sb, FEE_SIDE_REAL) mm = metr(d) cells.append(dict(rule=rl, hte=hte, band=bn, sizing=sz, sub=sb, **mm, ntrade=sum(info.values()))) C = pd.DataFrame(cells) P(f" griglia strategia valutata: {n_trials} celle " f"({len(RULES)} rule x {len(HTE_GRID)} hte x {len(bands)} bande x {len(sizings)} sizing x {len(subs)} sottocampioni)") P(f" celle con Sharpe > 0: {(C.sharpe > 0).sum()}/{len(C)} (una famiglia di RUMORE sta a ~50%)") P(" migliori 8 celle per Sharpe (NON e' una selezione: e' il MASSIMO della griglia, che serve al DSR):") P(C.sort_values("sharpe", ascending=False).head(8).to_string(index=False, float_format=lambda v: f"{v:.3f}")) P(" peggiori 3:") P(C.sort_values("sharpe").head(3).to_string(index=False, float_format=lambda v: f"{v:.3f}")) P("\n [6-bis] LA STESSA MACCHINA SUI LIVELLI PLACEBO (stesse finestre, stessa fee, stesso") P(" sizing): se un placebo regge quanto il livello 'speciale', l'OI non aggiunge nulla.") nulls = [] for rl in ("mp", "oi", "mid", "mr7", "rand", "jit_mp", "maj", "short", "long"): for hte in (3, 6, 12, 24): for bn, bd in bands.items(): d, _i = strat_daily(ev, rl, hte, bd, "sign", "all", FEE_SIDE_REAL) mm = metr(d) nulls.append(dict(rule=rl, hte=hte, band=bn, sharpe=mm["sharpe"], cagr=mm["cagr"], dd=mm["dd"])) N = pd.DataFrame(nulls) piv = N.pivot_table(index=["hte", "band"], columns="rule", values="sharpe") P(piv.reindex(columns=["mp", "oi", "mid", "mr7", "rand", "jit_mp", "maj", "short", "long"]) .to_string(float_format=lambda v: f"{v:6.2f}")) P(f" celle placebo valutate: {len(N)} (contano come trial: la griglia dichiarata sale)") best = C.sort_values("sharpe", ascending=False).iloc[0] P(f"\n CELLA MIGLIORE = {best['rule']}/hte{int(best['hte'])}/{best['band']}/{best['sizing']}/{best['sub']}") # expectancy per trade, in % e in R, sulla cella migliore e = ev[ev["hte"] == int(best["hte"])].copy() if best["sub"] == "big": e = e[e["tot_oi"] >= e["tot_oi"].median()] elif best["sub"] == "fri": e = e[e["dow"] == 4] dd = e[f"d_{best['rule']}"].to_numpy(float) keep = np.isfinite(dd) if best["band"] != "nessuna": keep &= (np.abs(dd) >= 0.002) & (np.abs(dd) <= 0.03) tw = e.loc[keep, f"tw_{best['rule']}"].to_numpy(float) dk = np.abs(dd[keep]) net = tw - 2 * FEE_SIDE_REAL wins, losses = net[net > 0], net[net < 0] rr = (wins.mean() / abs(losses.mean())) if len(wins) and len(losses) else np.nan P(f" trade: {len(net)} · WR {np.mean(net > 0):.1%} · RR medio {rr:.2f}" f" · WR-knob atteso 1/(1+RR) = {1 / (1 + rr):.1%}" if np.isfinite(rr) else " RR n.d.") P(f" expectancy NETTA per trade: {np.mean(net):+.5f} ({np.mean(net) * 100:+.3f}%)" f" | in R (R = distanza dal livello): {np.mean(net / np.maximum(dk, 1e-6)):+.3f} R") P(f" a fee ZERO: {np.mean(tw):+.5f} -> {'l edge lordo non esiste' if np.mean(tw) <= 0 else 'edge lordo positivo, la fee lo mangia?'}") dbest, info = strat_daily(ev, best["rule"], int(best["hte"]), None if best["band"] == "nessuna" else (0.002, 0.03), best["sizing"], best["sub"], FEE_SIDE_REAL) dcons, _ = strat_daily(ev, best["rule"], int(best["hte"]), None if best["band"] == "nessuna" else (0.002, 0.03), best["sizing"], best["sub"], FEE_SIDE_CONS) mb, mc = metr(dbest), metr(dcons) P(f" serie giornaliera cella migliore fee 3,5bps: Sharpe {mb['sharpe']:+.2f}" f" · maxDD {mb['dd']:.2%} · CAGR {mb['cagr']:+.2%} · n={mb['n']}g") P(f" fee 5,0bps: Sharpe {mc['sharpe']:+.2f}" f" · maxDD {mc['dd']:.2%} · CAGR {mc['cagr']:+.2%}") # scomposizione: per mese e per ERA DI COLLETTORE (bite:research fino al 30/07, poi pyg) P("\n SCOMPOSIZIONE (regola 22/08: un contributo positivo si scompone prima di crederci)") mo = dbest.groupby([dbest.index.year, dbest.index.month]) P(" " + " · ".join(f"{y}-{m:02d}: Sh {A._sh(g):+.2f} ret {float(np.prod(1 + g) - 1):+.2%} ({len(g)}g)" for (y, m), g in mo)) cut = pd.Timestamp("2026-07-30", tz="UTC") e1, e2 = dbest[dbest.index < cut], dbest[dbest.index >= cut] P(f" era collettore bite:research (<30/07): Sh {A._sh(e1):+.2f} ret {float(np.prod(1 + e1) - 1):+.2%} ({len(e1)}g)" f" | pyg (>=30/07): Sh {A._sh(e2):+.2f} ret {float(np.prod(1 + e2) - 1):+.2%} ({len(e2)}g)") top5 = dbest.sort_values(ascending=False).head(5) P(f" concentrazione: i 5 giorni migliori valgono {float(np.prod(1 + top5) - 1):+.2%}" f" su {float(np.prod(1 + dbest) - 1):+.2%} totali" f" = {100 * np.log1p(top5).sum() / max(np.log1p(dbest).sum(), 1e-9):.0f}% del log-equity") P(" ⚠ HOLD-OUT: NON ESISTE. La catena parte dal 2026-05-01 e l'hold-out del progetto e'") P(" 2025-01-01: TUTTA la serie sta dentro l'hold-out, quindi non c'e' nessuna finestra") P(" fuori campione da mostrare. Questo NON e' un dettaglio: e' il motivo per cui") P(" 3,7 mesi di catena non possono produrre uno sleeve.") # ---------------------------------------------------------------- 7. GATE P("\n[7] GATE") P(f" causalita' : {_assert_causalita(chain, ev)}") P(" A.causality_ok : NON GIRATO — non applicabile (il segnale non e' funzione del solo") P(" feed OHLC; la causalita' e' garantita dalla convenzione hte=dte-1") P(" e verificata dall'assert qui sopra).") # DSR: conta TUTTI i trial al rialzo all_sr = [float(v) for v in C["sharpe"].to_numpy(float) if np.isfinite(v)] all_sr_wide = all_sr + [float(v) for v in N["sharpe"].to_numpy(float) if np.isfinite(v)] n_fact = len(HTE_GRID) * len(RULES) * (1 + len(PLACEBOS_GLOB) + len(PLACEBOS_RULE)) + 3 * 4 # celle del fatto + condiz. trials_tot = len(all_sr_wide) + n_fact dsr, sr0 = A.deflated_sharpe(mb["sharpe"], all_sr, dbest, dpy=365.25) dsr_w, _ = A.deflated_sharpe(mb["sharpe"], all_sr_wide, dbest, dpy=365.25) P(f" deflated Sharpe : DSR {dsr:.3f} (soglia 0.95) · Sharpe-null atteso {sr0:.2f}" f" · trial contati {len(all_sr)} (griglia strategia)") P(f" DSR contando ANCHE le celle placebo ({len(all_sr_wide)} trial): {dsr_w:.3f}") P(f" trial TOTALI dichiarati al rialzo (fatto+null+condiz.+strategia): {trials_tot}") # marginal vs TP01 mv = A.marginal_vs_tp01(dbest) P(f" marginal vs TP01 : {mv.get('marginal_verdict')} · corr {mv.get('corr_full')}" f" · uplift w25 full {mv.get('blends', {}).get('w25', {}).get('uplift_full')}" f" · giorni in comune {mv.get('n_days')}") P(" ⚠ il gate NON PUO' dare ADDS con questo campione: `multicut_persistent`") P(" richiede >=2 tagli annuali da >=120 giorni e la serie ne ha ~113 in UN") P(" anno solo; e `has_insample_edge` e' True per DEFAULT perche' non c'e'") P(" alcun pre-2025. Il verdetto marginale qui e' strutturale, non informativo.") # null del de-levering (obbligatorio su ogni claim di DD) B = A.tp01_baseline_daily() J = pd.concat({"B": B, "C": dbest}, axis=1, join="inner").dropna() if len(J) > 20: bl = 0.75 * J["B"] + 0.25 * J["C"] dd_bl = A._dd_ret(bl) ks = np.linspace(0.05, 1.0, 96) ok = [(k, A._sh(k * J["B"]), A._dd_ret(k * J["B"])) for k in ks] cand = [(k, s) for k, s, d_ in ok if d_ <= dd_bl] kbest = max(cand, key=lambda t: t[1]) if cand else None P(f" null de-levering : blend 0.75TP01+0.25cand -> Sharpe {A._sh(bl):+.2f} / DD {dd_bl:.2%}" f" | k*TP01 a pari DD -> k={kbest[0]:.2f} Sharpe {kbest[1]:+.2f}" if kbest else " null de-levering : non calcolabile") if kbest: P(f" esito: {'REFUTED (il de-levering fa meglio)' if kbest[1] >= A._sh(bl) else 'superato'}") else: P(" null de-levering : NON GIRATO (sovrapposizione con TP01 troppo corta)") # implausible imp = A.implausible_sharpe(dbest, n_trades=int(best["ntrade"])) P(f" implausible_sharpe : implausible={imp['implausible']} · {imp.get('reasons')}") # anchor: l'ANCORA di questa strategia e' hte (quale ora prima della scadenza si entra) ab = A.anchor_luck_band( lambda h: strat_daily(ev, best["rule"], int(h), None if best["band"] == "nessuna" else (0.002, 0.03), best["sizing"], best["sub"], FEE_SIDE_REAL)[0], offsets=list(HTE_GRID), canonical=int(best["hte"])) P(f" anchor (hte) : canonico {ab.get('canonical'):.2f} al {100 * ab.get('canonical_pctl', np.nan):.0f}° pctl" f" · MEDIANA ONESTA {ab.get('median'):.2f} · banda [{ab.get('lo'):.2f},{ab.get('hi'):.2f}]" f" · positive {ab.get('frac_positive'):.0%} · gate_pass={ab.get('gate_pass')}") # LA DOMANDA DECISIVA, sulla banda d'ancora: max-pain aggiunge qualcosa a un livello # che NON usa opzioni? Mediana delle DIFFERENZE APPAIATE sui 8 hte (mai differenza # delle mediane: lezione 26/07). def _fn(rule): return lambda h: strat_daily(ev, rule, int(h), (0.002, 0.03), "sign", "all", FEE_SIDE_REAL)[0] for other in ("mr7", "mid", "oi", "rand"): dl = A.anchor_luck_delta(_fn("mp"), _fn(other), offsets=list(HTE_GRID)) P(f" Δ appaiato mp-{other:<4}: mediana {dl.get('median_paired'):+.2f}" f" · positiva in {dl.get('n_positive')}/{dl.get('n_anchors')} hte" f" · banda [{dl.get('lo'):+.2f},{dl.get('hi'):+.2f}] · gate_pass={dl.get('gate_pass')}") # eseguibilita' a $600 P(" eseguibilita' $600 :", end=" ") hc = [] for a in ASSETS: df = win_1h(a, pd.Timestamp(ev["hh"].min()), pd.Timestamp(ev["exp"].max())) tgt = hourly_target(ev, a, best["rule"], int(best["hte"]), None if best["band"] == "nessuna" else (0.002, 0.03), best["sizing"], best["sub"], df) sc = A.eval_weights_smallcap(df, tgt, capital=300.0, fee_side=FEE_SIDE_REAL) hc.append(f"{a} haircut {sc['sharpe_haircut']:+.3f} ({sc['n_executed_trades']} trade eseguiti)") P(" · ".join(hc)) P(" (300$/asset = il cap per-asset del libro live a questo capitale;") P(" una gamba ±1 muove 300$ >> min-order 5$ -> nessun ordine saltato)") # ---------------------------------------------------------------- 8. SINTESI P("\n[8] SINTESI") n_fact_pos = int((F.lo > 0).sum()) P(f" fatto : {n_fact_pos}/{len(F)} celle (rule x hte) con pinning significativo a IC95") P(f" null loc. : {npos}/{ntot} confronti appaiati in cui il livello 'speciale' batte il placebo") P(f" strategia : miglior Sharpe della griglia {mb['sharpe']:+.2f} su {mb['n']} giorni, DSR {dsr:.3f}") P(f" marginale : {mv.get('marginal_verdict')} (strutturalmente non ADDS-abile a 3,7 mesi)") P("=" * 100) if __name__ == "__main__": main()