"""FASE 2 — esplorazione larga per famiglie su BTC/ETH, harness onesto (research_lab). Famiglie (serie di posizione, causali, netto fee, vs buy&hold + null p-value): TSMOM (momentum) | REVERSAL | MA-cross | DONCHIAN breakout | VOL-TARGET overlay | LEAD-LAG BTC<->ETH | HURST-gated momentum. Multi-TF dove sensato (1h + 15m). La barra DA BATTERE è il buy&hold (Sharpe ~0.8 su BTC/ETH): una strategia di timing vale solo se fa MEGLIO net-fee. Per ogni famiglia: scan griglia (FULL Sharpe), poi report onesto sulla config migliore. Selezionare il best-di-griglia GONFIA -> i gate veri sono OOS-VAL + null p<0.05. uv run python scripts/analysis/phase2_families.py """ from __future__ import annotations import sys from pathlib import Path PROJECT_ROOT = Path(__file__).resolve().parents[2] sys.path.insert(0, str(PROJECT_ROOT)) import numpy as np import pandas as pd from src.data.downloader import load_data from scripts.analysis.research_lab import ( backtest, buy_hold, mc_pvalue, window_mask, ts, VAL_START, HOLDOUT_START, BARS_PER_YEAR, ) # --------------------------------- famiglie --------------------------------- def tsmom(df, L, mode="ls"): c = pd.Series(df["close"].values.astype(float)) pos = np.sign(np.nan_to_num((c / c.shift(L) - 1).values)) return np.maximum(pos, 0) if mode == "lo" else pos def reversal(df, L, mode="ls"): c = pd.Series(df["close"].values.astype(float)) pos = -np.sign(np.nan_to_num((c / c.shift(L) - 1).values)) return np.maximum(pos, 0) if mode == "lo" else pos def ma_cross(df, fast, slow, mode="ls"): c = pd.Series(df["close"].values.astype(float)) ef = c.ewm(span=fast, adjust=False).mean() es = c.ewm(span=slow, adjust=False).mean() pos = np.sign((ef - es).values) return np.maximum(pos, 0) if mode == "lo" else pos def donchian(df, L, mode="ls"): h = pd.Series(df["high"].values.astype(float)).rolling(L).max().shift(1).values l = pd.Series(df["low"].values.astype(float)).rolling(L).min().shift(1).values c = df["close"].values.astype(float) pos = np.zeros(len(c)); cur = 0 for i in range(len(c)): if not np.isnan(h[i]) and c[i] > h[i]: cur = 1 elif not np.isnan(l[i]) and c[i] < l[i]: cur = -1 if mode == "ls" else 0 pos[i] = cur return pos def vol_target(df, tf, target=0.6, L=72): """Overlay SEMPRE-LONG con esposizione scalata dalla vol realizzata (target vol annua).""" c = pd.Series(df["close"].values.astype(float)) rv_ann = c.pct_change().rolling(L).std().values * np.sqrt(BARS_PER_YEAR[tf]) pos = np.clip(np.nan_to_num(target / np.where(rv_ann > 0, rv_ann, np.nan), nan=0.0), 0, 1) return pos def rolling_hurst(c, W=120, step=6, lags=(2, 4, 8, 16, 32)): logc = np.log(c); n = len(c); H = np.full(n, np.nan) lg = np.log(lags) for i in range(W, n, step): seg = logc[i - W:i] tau = [np.std(seg[lag:] - seg[:-lag]) for lag in lags] if min(tau) > 0: H[i] = np.polyfit(lg, np.log(tau), 1)[0] return pd.Series(H).ffill().fillna(0.5).values def hurst_mom(df, L=48, W=120, mode="ls"): H = rolling_hurst(df["close"].values.astype(float), W) return np.where(H > 0.5, tsmom(df, L, mode), 0.0) def leadlag_df(target_df, other_df, L): """Costruisce un df col close del TARGET e la posizione = segno del rendimento a L barre dell'ALTRO asset (allineato per timestamp). Ritorna (df_merged, pos).""" a = target_df[["timestamp", "open", "high", "low", "close"]] b = other_df[["timestamp", "close"]].rename(columns={"close": "other"}) m = a.merge(b, on="timestamp", how="inner").reset_index(drop=True) o = pd.Series(m["other"].values.astype(float)) pos = np.sign(np.nan_to_num((o / o.shift(L) - 1).values)) return m, pos # --------------------------------- reporting --------------------------------- ROWS = [] def summarize(family, asset, tf, df, pos, mc_n=300): full = backtest(df, pos, tf) oos = backtest(df, pos, tf, lo=VAL_START, hi=HOLDOUT_START) bh = buy_hold(df, tf) gross = backtest(df, pos, tf, fee_rt=0.0).sharpe _, p, _, _ = mc_pvalue(df, pos, tf, n=mc_n) beats_bh = full.sharpe > bh.sharpe and oos.sharpe > 0 real = (full.sharpe > 0 and oos.sharpe > 0 and not np.isnan(p) and p < 0.05) verdict = "★EDGE?" if (real and beats_bh) else ("real?" if real else "rumore") ROWS.append(dict(fam=family, asset=asset, tf=tf, full=full.sharpe, oos=oos.sharpe, gross=gross, bh=bh.sharpe, p=p, trd=full.ntrades, verdict=verdict)) print(f" {family:<16s} {asset} {tf:<3s} | FULL {full.sharpe:>5.2f} OOS {oos.sharpe:>5.2f} " f"gross {gross:>5.2f} | B&H {bh.sharpe:>4.2f} | p {p:>.3f} | trd/y {full.ntrades:>6.0f} | {verdict}") def scan_best(family, asset, tf, df, fn, grid, label_fn): """Scansiona la griglia (FULL Sharpe), stampa la riga compatta, ritorna la pos migliore.""" best = None line = [] for params in grid: pos = fn(df, *params) s = backtest(df, pos, tf).sharpe line.append(f"{label_fn(params)}={s:>4.1f}") if best is None or s > best[0]: best = (s, params, pos) print(f" {asset} {tf} grid: " + " ".join(line)) return best[2], best[1] def main(): print("=" * 100) print(" FASE 2 — esplorazione famiglie BTC/ETH | netto fee 0.10% RT | barra = buy&hold | hold-out bloccato") print("=" * 100) D1 = {a: load_data(a, "1h") for a in ("BTC", "ETH")} D15 = {a: load_data(a, "15m") for a in ("BTC", "ETH")} def block(title): print("\n" + "#" * 100 + f"\n {title}\n" + "#" * 100) # ---- TSMOM (momentum) 1h + 15m, L/S e long-only ---- block("TSMOM (momentum)") Ls = [(12,), (24,), (48,), (96,), (192,)] for a in ("BTC", "ETH"): pos, p = scan_best("TSMOM-LS", a, "1h", D1[a], lambda d, L: tsmom(d, L, "ls"), Ls, lambda x: f"L{x[0]}") summarize("TSMOM-LS", a, "1h", D1[a], pos) pos, p = scan_best("TSMOM-LO", a, "1h", D1[a], lambda d, L: tsmom(d, L, "lo"), Ls, lambda x: f"L{x[0]}") summarize("TSMOM-LO", a, "1h", D1[a], pos) pos, p = scan_best("TSMOM-LS", a, "15m", D15[a], lambda d, L: tsmom(d, L, "ls"), [(48,),(96,),(192,),(384,)], lambda x: f"L{x[0]}") summarize("TSMOM-LS", a, "15m", D15[a], pos) # ---- REVERSAL 1h + 15m ---- block("REVERSAL (mean-reversion breve)") Lr = [(1,), (3,), (6,), (12,), (24,)] for a in ("BTC", "ETH"): pos, p = scan_best("REV-LS", a, "1h", D1[a], lambda d, L: reversal(d, L, "ls"), Lr, lambda x: f"L{x[0]}") summarize("REV-LS", a, "1h", D1[a], pos) pos, p = scan_best("REV-LS", a, "15m", D15[a], lambda d, L: reversal(d, L, "ls"), Lr, lambda x: f"L{x[0]}") summarize("REV-LS", a, "15m", D15[a], pos) # ---- MA cross ---- block("MA-CROSS (trend)") g = [(12, 48), (24, 96), (48, 192), (24, 200)] for a in ("BTC", "ETH"): pos, p = scan_best("MAX-LS", a, "1h", D1[a], lambda d, f, s: ma_cross(d, f, s, "ls"), g, lambda x: f"{x[0]}/{x[1]}") summarize("MAX-LS", a, "1h", D1[a], pos) pos, p = scan_best("MAX-LO", a, "1h", D1[a], lambda d, f, s: ma_cross(d, f, s, "lo"), g, lambda x: f"{x[0]}/{x[1]}") summarize("MAX-LO", a, "1h", D1[a], pos) # ---- Donchian breakout ---- block("DONCHIAN breakout") Ld = [(24,), (48,), (96,), (192,)] for a in ("BTC", "ETH"): pos, p = scan_best("DONCH-LS", a, "1h", D1[a], lambda d, L: donchian(d, L, "ls"), Ld, lambda x: f"L{x[0]}") summarize("DONCH-LS", a, "1h", D1[a], pos) pos, p = scan_best("DONCH-LO", a, "1h", D1[a], lambda d, L: donchian(d, L, "lo"), Ld, lambda x: f"L{x[0]}") summarize("DONCH-LO", a, "1h", D1[a], pos) # ---- Vol-target overlay (vs buy&hold) ---- block("VOL-TARGET overlay (sempre-long scalato) — riduce la vol/DD del buy&hold?") for a in ("BTC", "ETH"): pos, p = scan_best("VOLTGT", a, "1h", D1[a], lambda d, t: vol_target(d, "1h", t, 72), [(0.4,), (0.6,), (0.8,), (1.0,)], lambda x: f"t{x[0]}") summarize("VOLTGT", a, "1h", D1[a], pos) # ---- Hurst-gated momentum ---- block("HURST-gated momentum (momentum solo in regime trending H>0.5)") for a in ("BTC", "ETH"): pos, p = scan_best("HURST-MOM", a, "1h", D1[a], lambda d, L: hurst_mom(d, L, 120, "ls"), [(24,), (48,), (96,)], lambda x: f"L{x[0]}") summarize("HURST-MOM", a, "1h", D1[a], pos) # ---- Lead-lag BTC<->ETH ---- block("LEAD-LAG BTC<->ETH (posiziona un asset col rendimento passato dell'altro)") for tgt, oth in (("ETH", "BTC"), ("BTC", "ETH")): Ll = [1, 3, 6, 12, 24] best = None; line = [] for L in Ll: m, pos = leadlag_df(D1[tgt], D1[oth], L) s = backtest(m, pos, "1h").sharpe line.append(f"L{L}={s:>4.1f}") if best is None or s > best[0]: best = (s, L, m, pos) print(f" {oth}->{tgt} 1h grid: " + " ".join(line)) _, L, m, pos = best summarize(f"LL {oth}>{tgt}", tgt, "1h", m, pos) # ---- classifica finale ---- print("\n" + "=" * 100) print(" CLASSIFICA — net-fee FULL Sharpe (★EDGE? = batte B&H, OOS>0 e null p<0.05)") print("=" * 100) for r in sorted(ROWS, key=lambda r: -r["full"]): print(f" {r['fam']:<16s} {r['asset']} {r['tf']:<3s} | FULL {r['full']:>5.2f} | OOS {r['oos']:>5.2f} | " f"B&H {r['bh']:>4.2f} | p {r['p']:>.3f} | {r['verdict']}") edges = [r for r in ROWS if r["verdict"] == "★EDGE?"] print(f"\n Candidati che battono il buy&hold net-fee + OOS>0 + null p<0.05: {len(edges)}") for r in edges: print(f" -> {r['fam']} {r['asset']} {r['tf']}: FULL {r['full']:.2f} OOS {r['oos']:.2f} p {r['p']:.3f}") if __name__ == "__main__": main()