PythagorasGoal/scripts/analysis/research_lab.py

"""HARNESS DI RICERCA ONESTO — BTC/ETH, v2.0.0 (Fase 0).

Dopo che l'intera libreria precedente si è rivelata artefatto di feed/harness disonesti,
la prima cosa di cui fidarsi NON è una strategia ma il banco di prova. Questo modulo è
quel banco: causale per costruzione, netto fee, con baseline e null model.

MODELLO CANONICO = SERIE DI POSIZIONE.
  Una strategia è una funzione  signal(df, **params) -> pd.Series/np.array  che dà la
  posizione target per barra in [-1, +1]. REGOLA: position[i] è decisa con dati FINO a
  close[i] (mai oltre) e GUADAGNA il rendimento close[i] -> close[i+1]. L'engine moltiplica
  position[i] * fwd[i] (fwd strettamente futuro rispetto alla decisione) -> niente look-ahead
  per costruzione, e niente fill sull'estremo di candela (si entra al close). La fee è
  addebitata sul TURNOVER |Δposition| (un round-trip 0->1->0 = 2 unità = fee_rt intera).

GATE (vedi CLAUDE.md): ingresso eseguibile (qui per costruzione), netto fee 0.10% RT,
OOS held-out, robustezza su griglia, onestà statistica (null model + buy&hold), walk-forward
per i modelli fittati, liquidità (BTC/ETH ok).

    uv run python scripts/analysis/research_lab.py          # self-test del banco
"""
from __future__ import annotations
import sys
from dataclasses import dataclass
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

FEE_RT = 0.001                      # 0.10% round-trip taker Deribit (0.05%/lato)
BARS_PER_YEAR = {"5m": 105192.0, "15m": 35064.0, "1h": 8766.0,
                 "4h": 2191.5, "12h": 730.5, "1d": 365.25}


def load_tf(asset: str, tf: str):
    """Carica un TF certificato. 5m/15m/1h diretti; 4h/12h/1d DERIVATI per resample dal 1h
    (confini 00:00 UTC). >=12h e' il regime raccomandato (sotto, costi+overfit dominano)."""
    if tf in ("5m", "15m", "1h"):
        return load_data(asset, tf)
    rule = {"4h": "4h", "12h": "12h", "1d": "1D"}[tf]
    df = load_data(asset, "1h").copy()
    df.index = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
    out = df.resample(rule, label="left", closed="left").agg(
        {"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"}).dropna(subset=["open"])
    epoch = pd.Timestamp("1970-01-01", tz="UTC")
    out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
    return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume"]]
# Hold-out FINALE bloccato: NIENTE ricerca/tuning lo tocca finché non è il verdetto (Fase 3).
HOLDOUT_START = "2025-01-01"
# Finestra di validazione OOS usata in ricerca (out-of-sample ma PRE hold-out).
VAL_START = "2023-01-01"


def ts(df) -> pd.Series:
    return pd.to_datetime(df["timestamp"], unit="ms", utc=True)


def window_mask(df, lo: str | None = None, hi: str | None = None) -> np.ndarray:
    t = ts(df)
    m = np.ones(len(df), bool)
    if lo is not None:
        m &= (t >= pd.Timestamp(lo, tz="UTC")).values
    if hi is not None:
        m &= (t < pd.Timestamp(hi, tz="UTC")).values
    return m


@dataclass
class BT:
    n: int
    ret: float        # rendimento composto sulla finestra (pos 1x, leva 1x)
    cagr: float
    sharpe: float     # annualizzato
    maxdd: float      # % (positivo)
    exposure: float   # |pos| medio
    turnover: float   # Σ|Δpos| / anno
    ntrades: float    # round-trip equivalenti / anno

    def line(self, label="") -> str:
        return (f"  {label:<22s} Sh {self.sharpe:>6.2f} | ret {self.ret*100:>+8.1f}% "
                f"CAGR {self.cagr*100:>+6.1f}% | DD {self.maxdd*100:>5.1f}% | "
                f"expo {self.exposure:>4.2f} trd/y {self.ntrades:>6.1f} | n {self.n}")


def _net_series(df, position, fee_rt=FEE_RT):
    """Ritorna (net, gross, fwd, pos) per barra. net[i] = pos[i]*fwd[i] - fee sul cambio a i."""
    c = df["close"].values.astype(float)
    pos = np.nan_to_num(np.asarray(position, float), nan=0.0)
    pos = np.clip(pos, -1.0, 1.0)
    n = len(c)
    fwd = np.zeros(n)
    fwd[:-1] = c[1:] / c[:-1] - 1.0          # rendimento close[i]->close[i+1] (futuro vs decisione a i)
    gross = pos * fwd
    dpos = np.abs(np.diff(np.concatenate([[0.0], pos])))   # cambio di posizione a i (si tradea al close[i])
    fee = dpos * (fee_rt / 2.0)              # fee_rt = round-trip (2 unità di turnover); /2 per unità
    net = gross - fee
    return net, gross, fwd, pos


def backtest(df, position, tf="1h", fee_rt=FEE_RT, lo=None, hi=None) -> BT:
    net, gross, fwd, pos = _net_series(df, position, fee_rt)
    m = window_mask(df, lo, hi)
    net_w, pos_w = net[m], pos[m]
    dpos_w = np.abs(np.diff(np.concatenate([[0.0], pos_w])))
    bpy = BARS_PER_YEAR[tf]
    n = int(m.sum())
    if n < 2:
        return BT(n, 0, float("nan"), 0, 0, 0, 0, 0)
    eq = np.cumprod(1.0 + net_w)
    total = float(eq[-1] - 1.0)
    years = n / bpy
    cagr = float((1 + total) ** (1 / years) - 1) if years > 0 and total > -1 else float("nan")
    mu, sd = float(net_w.mean()), float(net_w.std())
    sharpe = mu / sd * np.sqrt(bpy) if sd > 0 else 0.0
    peak = np.maximum.accumulate(eq)
    maxdd = float(np.max((peak - eq) / peak)) if n else 0.0
    expo = float(np.mean(np.abs(pos_w)))
    turn_y = float(dpos_w.sum() / years) if years > 0 else 0.0
    return BT(n, total, cagr, sharpe, maxdd, expo, turn_y, turn_y / 2.0)


def buy_hold(df, tf="1h", fee_rt=FEE_RT, lo=None, hi=None) -> BT:
    return backtest(df, np.ones(len(df)), tf, fee_rt, lo, hi)


def mc_pvalue(df, position, tf="1h", fee_rt=FEE_RT, n=500, lo=None, hi=None, seed=0):
    """Null model a ROTAZIONE CIRCOLARE: ruota la serie di posizione di un offset casuale.
    Preserva ESATTAMENTE exposure, turnover e distribuzione degli holding; distrugge solo
    l'allineamento col mercato. p = P(Sharpe_ruotato >= Sharpe_reale). p alto = il timing
    non batte il caso (nessuna skill)."""
    pos = np.nan_to_num(np.asarray(position, float))
    base = backtest(df, pos, tf, fee_rt, lo, hi).sharpe
    N = len(pos)
    if np.abs(np.diff(pos)).sum() == 0:       # posizione costante -> rotazione degenere
        return base, float("nan"), float("nan"), float("nan")
    rng = np.random.default_rng(seed)
    sims = np.empty(n)
    for k in range(n):
        off = int(rng.integers(1, N))
        sims[k] = backtest(df, np.roll(pos, off), tf, fee_rt, lo, hi).sharpe
    p = float((np.sum(sims >= base) + 1) / (n + 1))
    return base, p, float(sims.mean()), float(sims.std())


def report(name, df, position, tf="1h", fee_rt=FEE_RT, mc_n=400):
    """Stampa il verdetto onesto: FULL / OOS-VAL / vs buy&hold / null p-value / sweep fee."""
    print(f"\n  === {name} ({tf}) ===")
    print(backtest(df, position, tf, fee_rt).line("FULL"))
    print(backtest(df, position, tf, fee_rt, lo=VAL_START, hi=HOLDOUT_START).line(f"OOS-VAL {VAL_START[:4]}-24"))
    print(buy_hold(df, tf, fee_rt).line("buy&hold FULL"))
    base, p, msh, ssd = mc_pvalue(df, position, tf, fee_rt, n=mc_n)
    verdict = "RUMORE" if (np.isnan(p) or p > 0.05) else "batte il null"
    print(f"  null (rotazione, n={mc_n}): Sharpe reale {base:.2f} vs random {msh:.2f}±{ssd:.2f} "
          f"-> p={p if not np.isnan(p) else float('nan'):.3f}  [{verdict}]")
    print("  sweep fee RT:", "  ".join(
        f"{f*100:.2f}%→Sh{backtest(df, position, tf, f).sharpe:.2f}" for f in (0.0, 0.0005, 0.001, 0.002)))


# ============================ SELF-TEST DEL BANCO ============================
def self_test():
    """Valida l'HARNESS, non una strategia. Tre prove:
      (1) buy&hold: Sharpe positivo, DD grande (sanity dei numeri).
      (2) CHEAT look-ahead (pos = segno del rendimento FUTURO): Sharpe enorme, p≈0
          -> l'engine SA vedere un edge quando esiste davvero.
      (3) NOISE causale (pos da rumore del passato): Sharpe≈0, p≈0.5
          -> l'engine NON inventa edge dal nulla (niente leak)."""
    print("=" * 78)
    print("  SELF-TEST HARNESS — deve: vedere il cheat, NON vedere il rumore")
    print("=" * 78)
    df = load_data("BTC", "1h")
    t = ts(df)
    c = df["close"].values.astype(float)
    bh = buy_hold(df, "1h")
    print(bh.line("(1) buy&hold BTC"))
    assert bh.sharpe > 0, "buy&hold dovrebbe avere Sharpe>0 sullo storico BTC"

    # (2) CHEAT: posizione = segno del rendimento del prossimo bar (USA IL FUTURO)
    fwd = np.zeros(len(c)); fwd[:-1] = c[1:] / c[:-1] - 1.0
    cheat = np.sign(fwd)
    bt_cheat = backtest(df, cheat, "1h")
    _, p_cheat, _, _ = mc_pvalue(df, cheat, "1h", n=200, seed=1)
    print(bt_cheat.line("(2) CHEAT look-ahead"))
    print(f"      -> null p={p_cheat:.4f} (atteso ≈0: l'edge finto È enorme e battibile dal caso ~mai)")
    assert bt_cheat.sharpe > 20, "il cheat dovrebbe dare Sharpe enorme se l'engine è corretto"
    assert p_cheat < 0.02, "il cheat dovrebbe battere il null in modo schiacciante"

    # (3) NOISE causale a BASSO turnover (blocchi ~50 barre): isola la SKILL dalla fee-death.
    # Posizione casuale (non usa il futuro) tenuta a blocchi -> turnover basso -> se l'engine non
    # inventa edge dal nulla, Sharpe≈0 e il null p≈0.5 (random rotazioni indistinguibili).
    rng = np.random.default_rng(42)
    blk = 50
    raw = np.sign(rng.standard_normal(len(c) // blk + 1))
    noise_pos = np.repeat(raw, blk)[:len(c)]
    noise_pos = pd.Series(noise_pos).shift(1).fillna(0).values        # solo passato
    bt_noise = backtest(df, noise_pos, "1h")
    base_n, p_noise, msh, ssd = mc_pvalue(df, noise_pos, "1h", n=400, seed=2)
    print(bt_noise.line("(3) NOISE causale"))
    print(f"      -> null p={p_noise:.3f} (atteso alto/≈0.5: nessuna skill, indistinguibile dal caso)")
    assert bt_noise.sharpe < 2.0, "il rumore causale non deve sembrare SKILLATO (Sharpe positivo grande = leak)"
    assert p_noise > 0.10, "il rumore causale non deve battere il null (p basso = edge spurio/leak)"

    print("\n  ✓ HARNESS VALIDATO: vede il cheat (Sharpe enorme, p≈0), non inventa edge dal rumore (p alto).")
    print(f"  Hold-out finale BLOCCATO da {HOLDOUT_START} (non usato in ricerca). OOS-VAL: {VAL_START}→hold-out.")


if __name__ == "__main__":
    self_test()