"""GATE PORT06 del candidato index_comp_disp W=168 (ricerca dispersion 2026-06-08). Edge confermato avversarialmente: fade della componente idiosincratica di BTC verso l'indice EW, gated da alta dispersione. Config: rel_len=12, z_win=336, z_thr=1.5, disp_168 >= quantile rolling 0.7 (win 720), TP=1.0*ATR14, SL=1.5*ATR14, max_bars=24. Domanda del gate (lezione FR01: robusto != migliora-il-portafoglio): 1) correlazione daily col MASTER e con le fade BTC esistenti (e' un diversificatore?) 2) PORT06 BASE (17 sleeve) vs +DISP (18 sleeve) con pesi cap: DeltaSharpe/DeltaDD FULL e OOS. PROMOSSO solo se decorrela E migliora (o non degrada) l'OOS. uv run python scripts/analysis/dispersion_edges/gate_index_comp_disp.py """ from __future__ import annotations import sys from pathlib import Path import numpy as np import pandas as pd PROJECT_ROOT = Path(__file__).resolve().parents[3] sys.path.insert(0, str(PROJECT_ROOT)) from scripts.analysis.dispersion_lab import features, align_to from scripts.analysis.explore_lab import get_df, atr from scripts.analysis.combine_portfolio import _norm, IDX, port_returns, metrics, SPLIT, OOS_DATE from scripts.analysis.honest_improve2 import _daily_equity from scripts.portfolios._defs import PORTFOLIOS from src.portfolio import weighting as W FEE_RT, LEV, POS, INIT = 0.001, 3.0, 0.15, 1000.0 CFG = dict(rel_len=12, z_win=336, z_thr=1.5, disp_q=0.7, disp_q_win=720, tp_atr=1.0, sl_atr=1.5, max_bars=24) def _last_rank(x): if x.shape[0] < 2: return np.nan return float((x[:-1] < x[-1]).mean()) def build_trades(asset="BTC"): """Entries CAUSALI + exit intrabar (TP/SL/max_bars) -> [(i, j, ret_netto)].""" df = get_df(asset, "1h") F = features() fa = align_to(F, df) c, h, l = df["close"].values, df["high"].values, df["low"].values n = len(c) a14 = atr(df, 14) rel = fa[f"rel_{asset}"].values.astype(float) disp = fa["disp_168"].values.astype(float) # somma rolling rel su rel_len, z-score causale (mean/std rolling z_win shift 1) rs = pd.Series(rel).rolling(CFG["rel_len"]).sum() rmean = rs.rolling(CFG["z_win"]).mean().shift(1) rstd = rs.rolling(CFG["z_win"]).std().shift(1) z = ((rs - rmean) / rstd.replace(0, np.nan)).values dpct = pd.Series(disp).rolling(CFG["disp_q_win"]).apply(_last_rank, raw=True).values fee = FEE_RT * LEV out = [] last = -1 for i in range(n - 1): if i <= last or not np.isfinite(z[i]) or not np.isfinite(dpct[i]): continue if dpct[i] < CFG["disp_q"] or abs(z[i]) < CFG["z_thr"]: continue ai = a14[i] if not np.isfinite(ai) or ai <= 0: continue d = -1 if z[i] > 0 else 1 tp = c[i] + d * CFG["tp_atr"] * ai sl = c[i] - d * CFG["sl_atr"] * ai mb = CFG["max_bars"] j = min(i + mb, n - 1) exit_p = c[j] for k in range(1, mb + 1): j = i + k if j >= n: j = n - 1; exit_p = c[j]; break if d == 1: if l[j] <= sl: exit_p = sl; break if h[j] >= tp: exit_p = tp; break else: if h[j] >= sl: exit_p = sl; break if l[j] <= tp: exit_p = tp; break if k == mb: exit_p = c[j] out.append((i, j, (exit_p - c[i]) / c[i] * d * LEV - fee)) last = j return df, out def daily_equity(df, trades): ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) cap = INIT; eq_ts, eq_v = [], [] for i, j, ret in sorted(trades, key=lambda t: t[1]): cap = max(cap + cap * POS * ret, 10.0) eq_ts.append(ts.iloc[j]); eq_v.append(cap) return _norm(_daily_equity(eq_ts, eq_v, IDX)) def pmetrics(members, p, extra=None): ids = list(p.sleeve_ids) + ([extra] if extra else []) dr = pd.DataFrame({i: members[i].pct_change().fillna(0.0) for i in ids}) if extra: caps = dict(p.caps); caps["DISP"] = caps.get("DISP", None) w = W.weight_vector(p.weighting, ids, dr, weights=p.weights, caps=p.caps, clusters={**{i:(p.clusters or {}).get(i,i) for i in p.sleeve_ids}, **({extra:"disp"} if extra else {})}, lookback=p.vol_lookback) drp = port_returns({i: members[i] for i in ids}, w) return metrics(drp), metrics(drp, lo=SPLIT) def main(): p = PORTFOLIOS["PORT06"] print("=" * 100) print(" GATE PORT06 — candidato index_comp_disp W=168 (BTC) | famiglia DISP nuova") print(f" config {CFG} | OOS da {OOS_DATE}") print("=" * 100) from src.portfolio.sleeves import all_sleeve_equities eq_base = dict(all_sleeve_equities()) df, trades = build_trades("BTC") disp_eq = daily_equity(df, trades) fr = (disp_eq.iloc[-1] / disp_eq.iloc[0] - 1) * 100 o = disp_eq.iloc[SPLIT:]; ofr = (o.iloc[-1] / o.iloc[0] - 1) * 100 print(f"\n[1] candidato standalone: {len(trades)} trade | FULL {fr:+.0f}% | OOS {ofr:+.0f}%") # correlazione daily col MASTER e con le fade BTC dr_cand = disp_eq.pct_change().fillna(0.0) print("\n[2] correlazione daily col candidato (decorrela?):") for sid in ["MR01_BTC", "MR02_BTC", "MR07_BTC", "DIP01_BTC"]: corr = dr_cand.corr(eq_base[sid].pct_change().fillna(0.0)) print(f" {sid:<12} corr {corr:+.3f}") master_dr = pd.DataFrame({i: eq_base[i].pct_change().fillna(0.0) for i in p.sleeve_ids}).mean(axis=1) print(f" {'MASTER(EW)':<12} corr {dr_cand.corr(master_dr):+.3f}") # PORT06 base vs +DISP f_b, o_b = pmetrics(eq_base, p) members = dict(eq_base); members["DISP_BTC"] = disp_eq f_e, o_e = pmetrics(members, p, extra="DISP_BTC") print("\n[3] PORT06 BASE (17) vs +DISP (18):") print(f" {'':<10}{'FULL Sh':>9}{'FULL DD%':>10}{'OOS Sh':>9}{'OOS DD%':>9}") print(f" {'BASE':<10}{f_b['sharpe']:>9.2f}{f_b['dd']:>10.2f}{o_b['sharpe']:>9.2f}{o_b['dd']:>9.2f}") print(f" {'+DISP':<10}{f_e['sharpe']:>9.2f}{f_e['dd']:>10.2f}{o_e['sharpe']:>9.2f}{o_e['dd']:>9.2f}") print(f" {'DELTA':<10}{f_e['sharpe']-f_b['sharpe']:>+9.2f}{f_e['dd']-f_b['dd']:>+10.2f}" f"{o_e['sharpe']-o_b['sharpe']:>+9.2f}{o_e['dd']-o_b['dd']:>+9.2f}") promoted = (o_e['sharpe'] >= o_b['sharpe'] - 0.02 and o_e['dd'] <= o_b['dd'] + 0.20 and f_e['sharpe'] >= f_b['sharpe'] - 0.02) print("\n VERDETTO: " + (">>> PROMOSSO <<<" if promoted else ">>> BOCCIATO (diluisce, come FR01) <<<")) if __name__ == "__main__": main()