"""VERIFICA AVVERSARIALE di TP01 (branch strategy-research-2026-06) col MIO gauntlet onesto. TP01 = TSMOM multi-orizzonte (30/90/180g) long-flat, vol-target 20%, leva cap 2x, portafoglio 50/50 BTC+ETH. Codice riprodotto VERBATIM dal branch (src/strategies/trend_portfolio.py). La loro tesi: 'positiva ogni anno 2019-2026, Sharpe ~1.32'. Il mio test decisivo: il HOLD-OUT 2025-26 (che ha bocciato il mio trend 1h in Fase 3) + cross-asset + multi-TF (cherry-picking 4h?). uv run python scripts/analysis/verify_tp01.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 HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") CANONICAL = dict(target_vol=0.20, leverage=2.0, long_only=True, horizons_days=(30, 90, 180), vol_win_days=30, fee_side=0.0005) # ---- TP01 riprodotto VERBATIM dal branch ---- def simple_returns(c): r = np.zeros(len(c)); r[1:] = c[1:] / c[:-1] - 1.0; return r def realized_vol(r, win, bpy): return pd.Series(r).rolling(win, min_periods=win // 2).std().values * np.sqrt(bpy) def tsmom_blend(c, horizons): n = len(c); acc = np.zeros(n); cnt = np.zeros(n) for h in horizons: s = np.full(n, np.nan); s[h:] = np.sign(c[h:] / c[:-h] - 1.0) v = np.isfinite(s); acc[v] += s[v]; cnt[v] += 1 out = np.zeros(n); nz = cnt > 0; out[nz] = acc[nz] / cnt[nz]; return out def target_series(df, p, bpd): c = df["close"].values.astype(float); bpy = bpd * 365.25 r = simple_returns(c) vol = realized_vol(r, p["vol_win_days"] * bpd, bpy) direction = tsmom_blend(c, tuple(d * bpd for d in p["horizons_days"])) if p["long_only"]: direction = np.clip(direction, 0, None) scal = np.where((vol > 0) & np.isfinite(vol), p["target_vol"] / vol, 0.0) tgt = np.clip(direction * scal, -p["leverage"], p["leverage"]); tgt[~np.isfinite(tgt)] = 0.0 return tgt def net_returns(df, p, bpd): c = df["close"].values.astype(float); r = simple_returns(c) tgt = target_series(df, p, bpd) pos_held = np.zeros(len(tgt)); pos_held[1:] = tgt[:-1] # decisa a close[t-1], tenuta in t -> causale gross = pos_held * r turn = np.abs(np.diff(pos_held, prepend=0.0)) net = gross - p["fee_side"] * turn; net[0] = 0.0 return np.clip(net, -0.99, None), pos_held def resample(df_1h, rule): g = df_1h.copy(); idx = pd.to_datetime(g["timestamp"], unit="ms", utc=True); g.index = idx out = g.resample(rule, label="left", closed="left").agg( {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}).dropna(subset=["open"]) out["timestamp"] = out.index return out.reset_index(drop=True) def metrics(combo, idx): rr = combo[np.isfinite(combo)] if len(rr) < 2 or np.std(rr) == 0: return dict(sharpe=0, cagr=0, dd=0, ret=0, n=len(rr)) dt = pd.Series(idx).diff().dt.total_seconds().median() bpy = 86400 * 365.25 / dt eq = np.cumprod(1 + rr); peak = np.maximum.accumulate(eq) years = (idx[-1] - idx[0]).total_seconds() / 86400 / 365.25 return dict(sharpe=float(np.mean(rr) / np.std(rr) * np.sqrt(bpy)), cagr=float(eq[-1] ** (1 / years) - 1) if years > 0 else 0, dd=float(np.max((peak - eq) / peak)), ret=float(eq[-1] - 1), n=len(rr)) def portfolio_combo(tf_rule, bpd): series = {} for a in ("BTC", "ETH"): df = load_data(a, "1h") if tf_rule: df = resample(df, tf_rule) net, _ = net_returns(df, CANONICAL, bpd) series[a] = pd.Series(net, index=pd.to_datetime(df["timestamp"], unit="ms", utc=True) if not tf_rule else pd.DatetimeIndex(df["timestamp"])) J = pd.concat(series, axis=1, join="inner").fillna(0.0) combo = 0.5 * J["BTC"].values + 0.5 * J["ETH"].values return combo, J.index, J def line(label, combo, idx): m = metrics(combo, idx) return f" {label:<22s} Sharpe {m['sharpe']:>5.2f} | ret {m['ret']*100:>+8.1f}% CAGR {m['cagr']*100:>+6.1f}% | DD {m['dd']*100:>5.1f}% | n {m['n']}" def main(): print("=" * 92) print(" VERIFICA TP01 (TSMOM 30/90/180 vol-target 20% lev2x long-flat, 50/50 BTC+ETH)") print(" col gauntlet onesto: FULL vs buy&hold | HOLD-OUT 2025-26 bloccato | per-anno | multi-TF") print("=" * 92) TFS = [("15m", "15min", 96), ("1h", None, 24), ("4h", "4h", 6), ("1d", "1D", 1)] print("\n (A) MULTI-TF: il 4h e' cherry-picked? FULL + HOLD-OUT per ogni timeframe") for tf, rule, bpd in TFS: combo, idx, J = portfolio_combo(rule, bpd) ho = idx >= HOLDOUT full = metrics(combo, idx) hold = metrics(combo[ho], idx[ho]) tag = " <- canonica" if tf == "4h" else "" print(f" {tf:<3s} FULL Sh {full['sharpe']:>5.2f} CAGR {full['cagr']*100:>+6.1f}% DD {full['dd']*100:>4.1f}% " f"| HOLD-OUT Sh {hold['sharpe']:>5.2f} ret {hold['ret']*100:>+6.1f}% DD {hold['dd']*100:>4.1f}%{tag}") # focus 4h canonica combo, idx, J = portfolio_combo("4h", 6) print("\n (B) 4h CANONICA — per anno (la tesi: positiva OGNI anno 2019-2026)") s = pd.Series(combo, index=idx) for y, g in s.groupby(s.index.year): eq = np.cumprod(1 + g.values); pk = np.maximum.accumulate(eq) ho_flag = " <- HOLD-OUT (mai usato per scegliere config?)" if y >= 2025 else "" print(f" {y}: ret {(eq[-1]-1)*100:>+7.1f}% DD {np.max((pk-eq)/pk)*100:>5.1f}%{ho_flag}") print("\n (C) HOLD-OUT 2025-26 — TP01 vs buy&hold 50/50 (4h)") ho = idx >= HOLDOUT print(line("TP01 portfolio HO", combo[ho], idx[ho])) # buy&hold 50/50 sullo stesso indice/finestra bh = {} for a in ("BTC", "ETH"): df = resample(load_data(a, "1h"), "4h") r = simple_returns(df["close"].values.astype(float)) bh[a] = pd.Series(r, index=pd.DatetimeIndex(df["timestamp"])) Jb = pd.concat(bh, axis=1, join="inner").reindex(idx).fillna(0.0) bh_combo = 0.5 * Jb["BTC"].values + 0.5 * Jb["ETH"].values print(line("buy&hold 50/50 HO", bh_combo[ho], idx[ho])) print(line("TP01 portfolio FULL", combo, idx)) print(line("buy&hold 50/50 FULL", bh_combo, idx)) print("\n (D) CROSS-ASSET nel HOLD-OUT (lo stesso edge regge su ENTRAMBI?)") for a in ("BTC", "ETH"): df = resample(load_data(a, "1h"), "4h") net, _ = net_returns(df, CANONICAL, 6) ix = pd.DatetimeIndex(df["timestamp"]); m = ix >= HOLDOUT print(line(f"TP01 {a} sleeve HO", net[m], ix[m])) if __name__ == "__main__": main()