PythagorasGoal/scripts/analysis/verify_tp01.py

"""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()