"""STATARB-MULTI — il meccanismo di STATARB-RESID generalizza fuori dalla coppia ETH/BTC? PERCHE' QUESTA E' LA DOMANDA PIU' IMPORTANTE APERTA. STATARB-RESID e' il miglior lead del progetto (+4,4% in ~4 settimane di forward, haircut fill ~0, ortogonale al book, ESEGUIBILE a $600 — non STAT-MODE) e ha un gate di deploy pre-registrato al 2026-09-27. Ma e' stato scoperto su UNA coppia, ETH/BTC, dentro uno sweep a 5 thread: Sharpe 0.84, deflated-Sharpe 0.929 (sotto 0.95). Se il meccanismo e' reale deve funzionare anche su coppie MAI usate per scoprirlo; se ETH/BTC e' un outlier, il gate del 27/09 va letto con un prior molto peggiore — e va saputo PRIMA, non dopo. Questo e' un test out-of-PAIR-sample: i 50 alt di Hyperliquid non hanno partecipato alla scoperta. MECCANISMO CONGELATO (nessun rifit, nessuna griglia): W=45, sgn=+1, vol-target 20%, cap 2x, z-score su finestra W, beta OLS rolling causale, dollar-neutral, fee 0.05%/gamba per |delta pos|. Identico a scripts/research/orthogonal_signals.f_statarb_resid, riscritto qui in forma generica (base, target) invece che (BTC, ETH) — il test di identita' e' nel passo (0). IL MODO IN CUI QUESTO SEGNALE PUO' MENTIRE (pre-registrato, non scoperto dopo): con sgn=+1 la posizione e' proporzionale al LIVELLO del residuo z-scorato, cioe' "long l'asset che ha sovraperformato il suo fit su BTC". Se un alt sottoperforma BTC in modo persistente (2024-2026 e' stato esattamente questo), il segnale resta quasi sempre SHORT e la strategia degenera in una SCOMMESSA STATICA "short alt vs BTC" che ha pagato per motivi di beta, non di timing. E' la stessa degenerazione ("predictor affine nel ritorno proprio") gia' codificata il 2026-07-24. Quindi: NULL 1 — STATICA, in DUE versioni (la prima stesura di questo script ne aveva una sola, col segno preso da `sign(mean(segnale))` su TUTTO il campione: era look-ahead, cioe' un null che conosce gia' la direzione giusta. Corretta qui): (1a) CAUSALE: segno = sign(media espandente del segnale fino a i-1) — usa solo il passato; (1b) A PRIORI: sempre SHORT alt vs BTC, la scommessa beta nota del 2024-2026 (gli alt hanno sottoperformato BTC). Nessun parametro stimato, nessun look-ahead. Se il segnale non batte entrambe, non c'e' timing: c'e' solo direzione media. NULL 2 — PERMUTAZIONE: 200 permutazioni a blocchi (20 barre) del segnale contro lo stesso spread, per un p-value che non assume normalita'. CHECK 3 — DEGENERAZIONE: frazione di barre a segno costante; una coppia >85% mono-segno e' una scommessa statica travestita, e va marcata. CHECK 4 — RIDONDANZA vs XS01: correlazione del paniere di coppie ai rendimenti dello sleeve XS01 (long-i-forti/short-i-deboli): se corr e' alta, non e' uno sleeve nuovo, e' XS01. CHECK 5 — ETH/BTC e' un outlier? rango e percentile della coppia scopritrice fra tutte. uv run python scripts/research/r0725_statarb_multi.py """ from __future__ import annotations 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)) RAW = ROOT / "data" / "raw" ANN = np.sqrt(365.0) # --- CONFIG CONGELATA (dallo sweep 2026-06-29; NON toccare) -------------------------------------- W = 45 SGN = +1 TARGET_VOL = 0.20 VOL_WIN = 30 CAP = 2.0 FEE_LEG = 0.0005 # 0.05% per gamba per unita' di |delta posizione| (2 gambe) BASE = "BTC" MIN_BARS = 400 # ~13 mesi: sotto questa soglia la coppia non e' valutabile N_PERM = 200 BLOCK = 20 SEED = 20260725 def load_hl(sym: str) -> pd.Series: p = RAW / f"hl_{sym.lower()}_1d.parquet" if not p.exists(): raise FileNotFoundError(p) d = pd.read_parquet(p) s = pd.Series(d["close"].values.astype(float), index=pd.to_datetime(d["timestamp"], unit="ms", utc=True)).sort_index() return s[~s.index.duplicated(keep="last")] def universe() -> list[str]: out = [] for p in sorted(RAW.glob("hl_*_1d.parquet")): sym = p.stem.replace("hl_", "").replace("_1d", "").upper() if sym != BASE: out.append(sym) return out def _zscore(v: np.ndarray, w: int) -> np.ndarray: s = pd.Series(v) m = s.rolling(w, min_periods=w).mean() sd = s.rolling(w, min_periods=w).std() return ((s - m) / sd.replace(0, np.nan)).values def signal(base_px: pd.Series, tgt_px: pd.Series) -> tuple[np.ndarray, np.ndarray]: """Ritorna (posizione dollar-neutral causale, ritorno spread) — meccanismo congelato.""" x = np.log(base_px.values) y = np.log(tgt_px.values) sx, sy = pd.Series(x), pd.Series(y) mx = sx.rolling(W, min_periods=W).mean() my = sy.rolling(W, min_periods=W).mean() cov = (sx * sy).rolling(W, min_periods=W).mean() - mx * my var = (sx * sx).rolling(W, min_periods=W).mean() - mx * mx beta = cov / var.replace(0, np.nan) resid = (sy - (my - beta * mx) - beta * sx).values z = _zscore(np.nan_to_num(resid), W) d = SGN * np.tanh(np.nan_to_num(z)) r_b = np.concatenate([[0.0], np.diff(base_px.values) / base_px.values[:-1]]) r_t = np.concatenate([[0.0], np.diff(tgt_px.values) / tgt_px.values[:-1]]) spread = r_t - r_b vol = pd.Series(spread).rolling(VOL_WIN, min_periods=VOL_WIN).std().values * ANN scal = np.where((vol > 0) & np.isfinite(vol), TARGET_VOL / vol, 0.0) pos = np.clip(np.nan_to_num(d) * scal, -CAP, CAP) pos[~np.isfinite(pos)] = 0.0 return pos, spread def pnl(pos: np.ndarray, spread: np.ndarray) -> np.ndarray: """Ritorno netto: posizione decisa a i-1 tenuta in i, fee su 2 gambe per |delta pos|.""" held = np.concatenate([[0.0], pos[:-1]]) turn = np.abs(np.diff(held, prepend=0.0)) return held * spread - 2.0 * FEE_LEG * turn def _sh(r: np.ndarray) -> float: r = r[np.isfinite(r)] return float(r.mean() / r.std() * ANN) if len(r) > 30 and r.std() > 0 else 0.0 def _dd(r: np.ndarray) -> float: eq = np.cumprod(1.0 + np.nan_to_num(r)) return float(np.min(eq / np.maximum.accumulate(eq) - 1.0)) def main() -> None: print("=" * 104) print(" STATARB-MULTI — il meccanismo congelato (W=45, sgn=+1) fuori dalla coppia scopritrice") print("=" * 104) # ---------------- (0) identita' col segnale dello sweep, sulla coppia originale print("\n" + "-" * 104) print(" (0) CONTROLLO DI IDENTITA' — ETH/BTC ricalcolato qui vs implementazione dello sweep") print("-" * 104) try: sys.path.insert(0, str(ROOT / "scripts" / "research")) from orthogonal_signals import build_joint, f_statarb_resid, spread_ret j = build_joint("1d") pos_ref = f_statarb_resid(tf="1d", W=W, sgn=SGN, tv=TARGET_VOL, vw=VOL_WIN, cap=CAP)(j) r_ref = np.concatenate([[0.0], pos_ref[:-1]]) * spread_ret(j) print(f" sweep (feed Deribit certificato) n={len(j):>5} Sharpe LORDA {_sh(r_ref):>5.2f}") except Exception as e: print(f" [riferimento sweep non disponibile: {e.__class__.__name__}: {e}]") eth_hl, btc_hl = load_hl("ETH"), load_hl(BASE) idx = eth_hl.index.intersection(btc_hl.index) p, s = signal(btc_hl[idx], eth_hl[idx]) print(f" qui (feed Hyperliquid) n={len(idx):>5} Sharpe NETTA {_sh(pnl(p, s)):>5.2f}") print(" (venue e finestra diversi -> i due numeri non devono coincidere; serve a escludere") print(" un errore di implementazione, non a replicare il backtest)") # ---------------- (1) tutte le coppie alt/BTC print("\n" + "-" * 104) print(f" (1) MECCANISMO CONGELATO SU TUTTE LE COPPIE alt/{BASE} — out-of-pair-sample") print("-" * 104) base_px = load_hl(BASE) rng = np.random.default_rng(SEED) rows = [] for sym in universe(): try: tgt = load_hl(sym) except FileNotFoundError: continue ix = base_px.index.intersection(tgt.index) if len(ix) < MIN_BARS: continue b, t = base_px[ix], tgt[ix] pos, spread = signal(b, t) r = pnl(pos, spread) live = pos[np.isfinite(pos)] live = live[live != 0.0] if len(live) < MIN_BARS // 2: continue mono = float(max((live > 0).mean(), (live < 0).mean())) # degenerazione vol = pd.Series(spread).rolling(VOL_WIN, min_periods=VOL_WIN).std().values * ANN scal = np.where((vol > 0) & np.isfinite(vol), TARGET_VOL / vol, 0.0) scal = np.nan_to_num(scal) def _static(sign_vec: np.ndarray) -> np.ndarray: ps = np.clip(sign_vec * scal, -CAP, CAP) ps[~np.isfinite(ps)] = 0.0 return pnl(ps, spread) # (1a) CAUSALE: segno = media espandente del segnale fino a i-1 (nessun look-ahead) cum = np.cumsum(np.nan_to_num(pos)) cnt = np.arange(1, len(pos) + 1) run_mean = np.concatenate([[0.0], (cum / cnt)[:-1]]) sign_causal = np.where(run_mean >= 0, 1.0, -1.0) sign_causal[:VOL_WIN + W] = 0.0 # prima che il segnale esista r_stat = _static(sign_causal) # (1b) A PRIORI: sempre short alt vs BTC (beta noto del regime 2024-2026) r_short = _static(np.full(len(pos), -1.0)) # NULL 2: permutazione a blocchi del segnale nb = int(np.ceil(len(pos) / BLOCK)) perm = [] for _ in range(N_PERM): order = rng.permutation(nb) pp = np.concatenate([pos[k * BLOCK:(k + 1) * BLOCK] for k in order])[:len(pos)] perm.append(_sh(pnl(pp, spread))) perm = np.array(perm) rows.append(dict(sym=sym, n=len(ix), sh=_sh(r), dd=_dd(r), mono=mono, sh_stat=_sh(r_stat), uplift=_sh(r) - _sh(r_stat), sh_short=_sh(r_short), uplift_s=_sh(r) - _sh(r_short), pperm=float((perm >= _sh(r)).mean()), ret=r, pos=pos, idx=ix)) if not rows: print(" nessuna coppia valutabile — controlla data/raw/hl_*_1d.parquet") return rows.sort(key=lambda d: -d["sh"]) print(f" {'coppia':<10}{'n':>6}{'Sh netta':>10}{'maxDD':>9}{'mono%':>7}" f"{'stat.caus':>11}{'uplift':>8}{'sempre-sh':>11}{'uplift':>8}{'p perm':>9}") for d in rows: flag = " <-- DEGENERE" if d["mono"] > 0.85 else "" print(f" {d['sym']+'/'+BASE:<10}{d['n']:>6}{d['sh']:>10.2f}{d['dd']*100:>8.1f}%" f"{d['mono']*100:>6.0f}%{d['sh_stat']:>11.2f}{d['uplift']:>8.2f}" f"{d['sh_short']:>11.2f}{d['uplift_s']:>8.2f}{d['pperm']:>9.3f}{flag}") sh = np.array([d["sh"] for d in rows]) up = np.array([d["uplift"] for d in rows]) ups = np.array([d["uplift_s"] for d in rows]) pp = np.array([d["pperm"] for d in rows]) mono = np.array([d["mono"] for d in rows]) n = len(rows) print("\n" + "-" * 104) print(" (2) VERDETTO STATISTICO SULL'INSIEME") print("-" * 104) tstat = float(sh.mean() / (sh.std(ddof=1) / np.sqrt(n))) if n > 2 else 0.0 print(f" coppie valutate : {n}") print(f" Sharpe netta media {sh.mean():>6.2f} mediana {np.median(sh):>6.2f} " f"p10 {np.percentile(sh,10):>5.2f} p90 {np.percentile(sh,90):>5.2f}") print(f" frazione con Sharpe > 0 : {(sh>0).mean()*100:>5.0f}% " f"(t della media cross-coppie = {tstat:>5.2f})") print(f" UPLIFT vs statica CAUSALE media {up.mean():>6.2f} mediana {np.median(up):>6.2f} " f"frazione > 0: {(up>0).mean()*100:>3.0f}%") print(f" UPLIFT vs SEMPRE-SHORT media {ups.mean():>6.2f} mediana {np.median(ups):>6.2f} " f"frazione > 0: {(ups>0).mean()*100:>3.0f}%") print(f" p-value permutazione < 0.05 : {(pp<0.05).mean()*100:>5.0f}% delle coppie " f"(atteso per caso: 5%)") print(f" coppie DEGENERI (mono > 85%) : {(mono>0.85).sum()}/{n} " f"mono medio {mono.mean()*100:.0f}%") print("\n ⚠ il t cross-coppie SOVRASTIMA: le 50 coppie condividono la gamba BTC e gli alt") print(" co-muovono -> non sono 50 scommesse indipendenti. Il test onesto e' sul PANIERE (3).") # ---------------- (3) paniere EW print("\n" + "-" * 104) print(" (3) PANIERE EW DELLE COPPIE (sleeve ipotetica) + ridondanza vs XS01") print("-" * 104) ser = {d["sym"]: pd.Series(d["ret"], index=d["idx"]) for d in rows} M = pd.concat(ser, axis=1, sort=True).sort_index() bask = M.mean(axis=1, skipna=True).dropna() sh_b = _sh(bask.values) yrs = len(bask) / 365.0 print(f" paniere EW {len(ser)} coppie: n={len(bask)} ({yrs:.1f} anni) Sharpe {sh_b:>5.2f} " f"maxDD {_dd(bask.values)*100:>5.1f}% ret tot {(np.prod(1+bask.values)-1)*100:>6.1f}%") # ampiezza EFFETTIVA: quante scommesse indipendenti ci sono davvero C = M.corr().values off = C[~np.eye(len(C), dtype=bool)] rbar = float(np.nanmean(off)) n_eff = len(C) / (1.0 + (len(C) - 1) * rbar) if rbar > -1 / (len(C) - 1) else float(len(C)) print(f" correlazione media fra coppie {rbar:>5.3f} -> ampiezza EFFETTIVA ~{n_eff:.1f} " f"scommesse indipendenti (non {len(C)})") # intervallo di confidenza onesto sul paniere: block bootstrap (blocchi 20g) rng2 = np.random.default_rng(SEED + 7) v = bask.values nb2 = int(np.ceil(len(v) / BLOCK)) boot = [] for _ in range(2000): k = rng2.integers(0, nb2, size=nb2) x = np.concatenate([v[i * BLOCK:(i + 1) * BLOCK] for i in k])[:len(v)] boot.append(_sh(x)) boot = np.array(boot) lo, hi = np.percentile(boot, [2.5, 97.5]) print(f" block-bootstrap Sharpe paniere: IC95% [{lo:>5.2f}, {hi:>5.2f}] " f"P(Sharpe>0) = {(boot>0).mean()*100:.0f}% P(Sharpe>0.5) = {(boot>0.5).mean()*100:.0f}%") print(f" errore standard atteso a {yrs:.1f} anni ~ {1/np.sqrt(yrs):.2f} Sharpe -> " f"t del paniere ~ {sh_b*np.sqrt(yrs):.2f}") try: from src.portfolio.sleeves import _xsec_returns xs = _xsec_returns().dropna() xs.index = pd.to_datetime(xs.index, utc=True) if xs.index.tz is None else xs.index jj = pd.concat({"b": bask, "x": xs}, axis=1, sort=True).dropna() print(f" corr(paniere, XS01) = {jj['b'].corr(jj['x']):>6.3f} su {len(jj)} giorni comuni" f" -> {'RIDONDANTE con XS01' if abs(jj['b'].corr(jj['x'])) > 0.5 else 'non ridondante'}") except Exception as e: print(f" [corr XS01 non calcolabile: {e.__class__.__name__}: {e}]") # ---------------- (4) ETH/BTC e' un outlier? print("\n" + "-" * 104) print(" (4) LA COPPIA SCOPRITRICE E' UN OUTLIER?") print("-" * 104) eth = next((d for d in rows if d["sym"] == "ETH"), None) if eth is None: print(" ETH/BTC non fra le coppie valutate (storia insufficiente?)") else: rank = 1 + sum(1 for d in rows if d["sh"] > eth["sh"]) print(f" ETH/{BASE}: Sharpe {eth['sh']:.2f} -> rango {rank}/{n} " f"(percentile {100*(1-(rank-1)/n):.0f}°) mono {eth['mono']*100:.0f}% " f"uplift vs statica {eth['uplift']:+.2f} p perm {eth['pperm']:.3f}") print("\n" + "=" * 104) if __name__ == "__main__": main()