PythagorasGoal/scripts/analysis/phase2_families.py

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