feat(analysis): validazione out-of-sample fee-aware delle strategie

oos_validation.py: backtest OOS fedele al worker live (non-overlap, hold,
stop, fee, leva) su finestra held-out. Mostra che l'edge storico 76-79%
e' un artefatto di look-ahead (ingresso a close[i-1]) e che nessuna regola
di direzione onesta supera il lancio di moneta; le fee sono secondarie
(4/6 config perdono anche a fee zero).

intrabar_test.py: ingresso intra-barra su 5m vs close 15m a parita' di exit.
Lo "scatto" del breakout e' avverso (rientro immediato alla media), quindi
la granularita' piu' fine non recupera edge.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

scripts/analysis/intrabar_test.py

@@ -0,0 +1,188 @@
+"""Test ingresso intra-barra: rottura banda squeeze rilevata sul 5m vs close 15m.
+
+Domanda: entrando sul 5m appena il prezzo rompe la banda di Bollinger dello
+squeeze (bande dall'ultima barra 15m CHIUSA -> nessun look-ahead), si recupera
+parte del movimento che l'ingresso al close della barra 15m si perde?
+
+Confronto a parita' di EXIT (stesso wall-clock): l'unica differenza e' il prezzo
+d'ingresso (5m anticipato vs close 15m ritardato). La differenza di rendimento e'
+esattamente lo "scatto" del breakout catturato in piu'.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from src.data.downloader import load_data
+from src.live.signal_engine import keltner_ratio
+
+OOS_START = "2023-11-20"
+BB_W = 14
+SQ_THR = 0.8
+MIN_DUR = 5
+LEV = 3.0
+POS = 0.15
+M15 = 15 * 60 * 1000
+M5 = 5 * 60 * 1000
+
+
+def build_15m_levels(df15: pd.DataFrame) -> pd.DataFrame:
+    c = df15["close"].values
+    h = df15["high"].values
+    l = df15["low"].values
+    n = len(c)
+    kcr = keltner_ratio(c, h, l, BB_W)
+    ma = np.full(n, np.nan)
+    sd = np.full(n, np.nan)
+    for t in range(BB_W, n):
+        w = c[t - BB_W + 1 : t + 1]
+        ma[t] = w.mean()
+        sd[t] = w.std()
+    upper = ma + 2 * sd
+    lower = ma - 2 * sd
+
+    # durata squeeze consecutiva e maturita'
+    dur = np.zeros(n, dtype=int)
+    run = 0
+    for t in range(n):
+        if not np.isnan(kcr[t]) and kcr[t] < SQ_THR:
+            run += 1
+        else:
+            run = 0
+        dur[t] = run
+    mature = dur >= MIN_DUR
+
+    return pd.DataFrame({
+        "ts15": df15["timestamp"].values,
+        "close_time15": df15["timestamp"].values + M15,
+        "close15": c,
+        "upper": upper,
+        "lower": lower,
+        "mature": mature,
+    })
+
+
+def run_asset(asset: str, hold_min: int, fee_rt: float) -> dict:
+    df5 = load_data(asset, "5m").reset_index(drop=True)
+    df15 = load_data(asset, "15m").reset_index(drop=True)
+    lvl = build_15m_levels(df15)
+
+    d5 = pd.DataFrame({
+        "ts5": df5["timestamp"].values,
+        "close_time5": df5["timestamp"].values + M5,
+        "close5": df5["close"].values,
+    })
+
+    # banda armata: ultima barra 15m CHIUSA prima della chiusura del bar 5m
+    armed = pd.merge_asof(
+        d5.sort_values("close_time5"),
+        lvl[["close_time15", "upper", "lower", "mature"]].sort_values("close_time15"),
+        left_on="close_time5", right_on="close_time15", direction="backward",
+    )
+    # barra 15m CONTENENTE il bar 5m (per l'ingresso ritardato a close 15m)
+    cont = pd.merge_asof(
+        d5.sort_values("ts5"),
+        lvl[["ts15", "close15", "close_time15"]].rename(
+            columns={"close_time15": "cont_close_time"}).sort_values("ts15"),
+        left_on="ts5", right_on="ts15", direction="backward",
+    )
+
+    m = armed.copy()
+    m["cont_close"] = cont["close15"].values
+    m["cont_close_time"] = cont["cont_close_time"].values
+
+    oos_ms = int(pd.Timestamp(OOS_START, tz="UTC").timestamp() * 1000)
+    close5 = m["close5"].values
+    ct5 = m["close_time5"].values
+    upper = m["upper"].values
+    lower = m["lower"].values
+    mature = m["mature"].values
+    cont_close = m["cont_close"].values
+    cont_ct = m["cont_close_time"].values
+    n = len(m)
+
+    cap_e = cap_l = 1000.0   # equity ingresso early(5m) e late(15m)
+    peak_e = peak_l = 1000.0
+    dd_e = dd_l = 0.0
+    trades = win_e = win_l = 0
+    thrust_sum = 0.0
+    fee = fee_rt * LEV
+    busy_until = -1
+
+    for i in range(n):
+        if ct5[i] < oos_ms or ct5[i] <= busy_until:
+            continue
+        if not mature[i] or np.isnan(upper[i]):
+            continue
+        if close5[i] > upper[i]:
+            d = 1
+        elif close5[i] < lower[i]:
+            d = -1
+        else:
+            continue
+
+        entry_e = close5[i]
+        entry_l = cont_close[i]
+        exit_time = cont_ct[i] + hold_min * 60 * 1000
+        # primo close 5m al/oltre exit_time
+        j = np.searchsorted(ct5, exit_time, side="left")
+        if j >= n:
+            break
+        exit_p = close5[j]
+
+        ret_e = ((exit_p - entry_e) / entry_e) * d * LEV - fee
+        ret_l = ((exit_p - entry_l) / entry_l) * d * LEV - fee
+        thrust_sum += (entry_l - entry_e) / entry_e * d * 100  # scatto % (no leva)
+
+        cb_e, cb_l = cap_e, cap_l
+        cap_e = max(cb_e + cb_e * POS * ret_e, 10.0)
+        cap_l = max(cb_l + cb_l * POS * ret_l, 10.0)
+        peak_e = max(peak_e, cap_e); dd_e = max(dd_e, (peak_e - cap_e) / peak_e)
+        peak_l = max(peak_l, cap_l); dd_l = max(dd_l, (peak_l - cap_l) / peak_l)
+        trades += 1
+        win_e += ret_e > 0
+        win_l += ret_l > 0
+        busy_until = exit_time
+
+    return {
+        "trades": trades,
+        "avg_thrust": thrust_sum / trades if trades else 0.0,
+        "early_win": win_e / trades * 100 if trades else 0.0,
+        "late_win": win_l / trades * 100 if trades else 0.0,
+        "early_ret": (cap_e / 1000 - 1) * 100,
+        "late_ret": (cap_l / 1000 - 1) * 100,
+        "early_dd": dd_e * 100,
+        "late_dd": dd_l * 100,
+    }
+
+
+def main():
+    for fee_rt in (0.002, 0.001):
+        print("=" * 104)
+        print(f"  INGRESSO INTRA-BARRA 5m vs CLOSE 15m — OOS da {OOS_START} | leva={LEV:.0f}x "
+              f"| fee={fee_rt*100:.2f}% RT")
+        print("  EARLY = entra al close 5m che rompe la banda | LATE = entra al close della barra 15m | stesso exit")
+        print("=" * 104)
+        print(f"  {'Asset':>5s}{'Hold':>6s}{'Trd':>6s}{'Scatto%':>9s}"
+              f"{'EARLY win%':>12s}{'EARLY ret%':>12s}{'LATE win%':>11s}{'LATE ret%':>11s}{'Δret%':>9s}")
+        print("  " + "-" * 100)
+        for asset in ["BTC", "ETH"]:
+            for hold_min in (15, 30, 45):
+                r = run_asset(asset, hold_min, fee_rt)
+                print(f"  {asset:>5s}{hold_min:>5d}m{r['trades']:>6d}{r['avg_thrust']:>+9.3f}"
+                      f"{r['early_win']:>12.1f}{r['early_ret']:>+12.1f}"
+                      f"{r['late_win']:>11.1f}{r['late_ret']:>+11.1f}"
+                      f"{r['early_ret']-r['late_ret']:>+9.1f}")
+            print("  " + "-" * 100)
+        print("  Scatto% = movimento medio (no leva) catturato tra rottura 5m e close 15m, nella direzione.")
+        print("  Δret% = vantaggio dell'ingresso anticipato. Se ~0 o negativo, il 5m non aiuta.\n")
+
+
+if __name__ == "__main__":
+    main()

scripts/analysis/oos_validation.py

@@ -0,0 +1,259 @@
+"""Validazione out-of-sample fee-aware di tutte le strategie live.
+
+Per ognuna delle 6 config in strategies.yml:
+  - split temporale held-out (train = primi (1-test_frac), test = ultimo test_frac)
+  - ML01 (SignalEngine): allena sul train, predice sul test (come il worker live)
+  - rule-based: i segnali sono causali, si valutano quelli nella finestra test
+  - simulazione fedele al worker live: una posizione per volta (non-overlap),
+    uscita a `hold` barre o stop a -2%, fee round-trip e leva inclusi
+
+Stampa, per ogni config: numero trade nel test, win% lordo e netto, return netto,
+costo commissioni, e confronto lordo-vs-netto per isolare l'impatto delle fee.
+Usa i parquet locali (data/raw), nessuna chiamata di rete.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import numpy as np
+import pandas as pd
+import yaml
+
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from src.data.downloader import load_data
+from src.live.strategy_loader import load_strategy
+from src.live.signal_engine import SignalEngine, keltner_ratio, build_features
+
+TEST_FRAC = 0.30
+STOP_PCT = -0.02
+
+
+def simulate(entries: list[tuple[int, int]], close: np.ndarray, hold: int,
+             fee_rt: float, lev: float, pos: float,
+             initial: float = 1000.0, entry_offset: int = 0) -> dict:
+    """FSM fedele al worker live: non-overlap, hold N barre o stop -2%.
+
+    entry_offset: 0 = ingresso a close[i] (worker live); 1 = close[i-1]
+    (convenzione del backtest storico, che conosce la direzione di barra i).
+    """
+    n = len(close)
+    capital = peak = initial
+    max_dd = 0.0
+    fees_eur = gross_eur = 0.0
+    wins_gross = wins_net = n_trades = 0
+    last_exit = -1
+
+    for i, d in entries:
+        e = i - entry_offset
+        if e <= last_exit or e < 0 or e + 1 >= n:
+            continue
+        entry = close[e]
+        exit_price = close[min(e + hold, n - 1)]
+        for k in range(1, hold + 1):
+            j = e + k
+            if j >= n:
+                exit_price = close[n - 1]
+                break
+            if k < hold and (close[j] - entry) / entry * d <= STOP_PCT:
+                exit_price = close[j]
+                break
+            if k == hold:
+                exit_price = close[j]
+
+        actual = (exit_price - entry) / entry * d   # movimento prezzo * direzione (no leva)
+        gross = actual * lev
+        fee = fee_rt * lev
+        net = gross - fee
+
+        cap_before = capital
+        capital = max(cap_before + cap_before * pos * net, 10.0)
+        gross_eur += cap_before * pos * gross
+        fees_eur += cap_before * pos * fee
+        peak = max(peak, capital)
+        max_dd = max(max_dd, (peak - capital) / peak)
+
+        n_trades += 1
+        wins_gross += actual > 0
+        wins_net += net > 0
+        last_exit = e + hold
+
+    return {
+        "trades": n_trades,
+        "win_gross": wins_gross / n_trades * 100 if n_trades else 0.0,
+        "win_net": wins_net / n_trades * 100 if n_trades else 0.0,
+        "net_return_pct": (capital / initial - 1) * 100,
+        "net_eur": capital - initial,
+        "gross_eur": gross_eur,
+        "fees_eur": fees_eur,
+        "final_capital": capital,
+        "max_dd": max_dd * 100,
+    }
+
+
+def rule_entries(name: str, df: pd.DataFrame, params: dict, split: int) -> list[tuple[int, int]]:
+    strat = load_strategy(name)
+    ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
+    sigs = strat.generate_signals(df, ts, **params)
+    return [(s.idx, s.direction) for s in sigs if s.idx >= split]
+
+
+def ml_entries(df: pd.DataFrame, params: dict, split: int, hold: int) -> tuple[list[tuple[int, int]], dict]:
+    bb_w = params.get("bb_window", 14)
+    sq_thr = params.get("sq_threshold", 0.8)
+    ml_thr = params.get("ml_threshold", 0.70)
+
+    eng = SignalEngine(bb_w=bb_w, sq_thr=sq_thr, ml_thr=ml_thr)
+    train_res = eng.train(df.iloc[:split].reset_index(drop=True), lookahead=hold)
+    if not eng.trained:
+        return [], train_res
+
+    close = df["close"].values
+    high = df["high"].values
+    low = df["low"].values
+    volume = df["volume"].values
+    n = len(df)
+    kcr = keltner_ratio(close, high, low, bb_w)
+    up_idx = list(eng.model.classes_).index(1)
+
+    entries: list[tuple[int, int]] = []
+    in_sq = False
+    sq_start = 0
+    for i in range(bb_w + 1, n):
+        if np.isnan(kcr[i]):
+            continue
+        is_sq = kcr[i] < sq_thr
+        if is_sq and not in_sq:
+            in_sq, sq_start = True, i
+        elif not is_sq and in_sq:
+            in_sq = False
+            dur = i - sq_start
+            if dur < eng.min_squeeze_bars or i < split or i + hold >= n:
+                continue
+            avg_vol = float(np.mean(volume[sq_start:i]))
+            feats = build_features(df, i, dur, avg_vol, kcr[i])
+            if feats is None:
+                continue
+            p_up = eng.model.predict_proba(eng.scaler.transform(feats.reshape(1, -1)))[0][up_idx]
+            if p_up >= ml_thr:
+                entries.append((i, 1))
+            elif p_up <= (1 - ml_thr):
+                entries.append((i, -1))
+    return entries, train_res
+
+
+def squeeze_releases(df: pd.DataFrame, bb_w: int, sq_thr: float, min_dur: int,
+                     split: int) -> list[int]:
+    """Indici delle barre di rilascio squeeze nella finestra test (idx >= split)."""
+    close = df["close"].values
+    high = df["high"].values
+    low = df["low"].values
+    kcr = keltner_ratio(close, high, low, bb_w)
+    rels: list[int] = []
+    in_sq = False
+    sq_start = 0
+    for i in range(bb_w + 1, len(df)):
+        if np.isnan(kcr[i]):
+            continue
+        is_sq = kcr[i] < sq_thr
+        if is_sq and not in_sq:
+            in_sq, sq_start = True, i
+        elif not is_sq and in_sq:
+            in_sq = False
+            if i - sq_start >= min_dur and i >= split:
+                rels.append(i)
+    return rels
+
+
+def honest_entries(df: pd.DataFrame, rels: list[int], rule: str, mom: int = 4) -> list[tuple[int, int]]:
+    """Direzione da regole honest (solo dati <= i-1) o baseline breakout.
+
+    breakout: sign(close[i]-close[i-1])  -> conoscibile solo a close[i] (= live attuale)
+    premom:   sign(close[i-1]-close[i-1-mom]) -> trend pre-release, 100% honest
+    fade:     -sign(close[i]-close[i-1]) -> mean-reversion del breakout
+    """
+    close = df["close"].values
+    out: list[tuple[int, int]] = []
+    for i in rels:
+        if i - 1 - mom < 0:
+            continue
+        if rule == "premom":
+            d = np.sign(close[i - 1] - close[i - 1 - mom])
+        elif rule == "fade":
+            d = -np.sign(close[i] - close[i - 1])
+        else:  # breakout
+            d = np.sign(close[i] - close[i - 1])
+        if d != 0:
+            out.append((i, int(d)))
+    return out
+
+
+def main():
+    cfg = yaml.safe_load((PROJECT_ROOT / "strategies.yml").read_text())
+    defaults = cfg.get("defaults", {})
+    hold = defaults.get("hold_bars", 3)
+    lev = defaults.get("leverage", 3)
+    fee_rt = 0.002
+
+    fee_grid = [0.0, 0.0005, 0.001, 0.0015, 0.002]
+
+    # ---- (b) SENSIBILITA' ALLE FEE (config live, ingresso close[i]) ----
+    print("=" * 104)
+    print(f"  (b) SENSIBILITA' ALLE FEE — config live, ingresso close[i] | OOS {int(TEST_FRAC*100)}% | hold={hold} leva={lev}x")
+    print("=" * 104)
+    print(f"  {'Strategia':<26s}{'Asset':>5s}{'Trd':>5s}{'Lordo€':>9s}"
+          + "".join(f"{f'{f*100:.2f}%':>10s}" for f in fee_grid))
+    print("  " + "-" * 100)
+
+    for entry in cfg.get("strategies", []):
+        if not entry.get("enabled", True):
+            continue
+        name, asset, tf = entry["name"], entry["asset"], entry["tf"]
+        pos = entry.get("position_size", defaults.get("position_size", 0.15))
+        params = dict(entry.get("params", {}))
+        params["asset"], params["tf"] = asset, tf
+
+        df = load_data(asset, tf).reset_index(drop=True)
+        split = int(len(df) * (1 - TEST_FRAC))
+        close = df["close"].values
+        entries = (ml_entries(df, params, split, hold)[0] if name.startswith("ML01")
+                   else rule_entries(name, df, params, split))
+
+        gross = simulate(entries, close, hold, 0.0, lev, pos)["net_eur"]
+        rets = [simulate(entries, close, hold, f, lev, pos)["net_return_pct"] for f in fee_grid]
+        print(f"  {name:<26s}{asset:>5s}{len(entries):>5d}{gross:>+9.0f}"
+              + "".join(f"{r:>+10.1f}" for r in rets))
+
+    print("  " + "-" * 100)
+    print("  Colonne = Ret% netto al variare della fee RT. 0.00% isola l'edge puro (senza costi).")
+    print("  Deribit perp reale: taker ~0.10% RT, maker ~0%. Il modello live usa 0.20% RT.")
+
+    # ---- (a) HONEST-ENTRY squeeze: direzione decisa <= i-1, ingresso close[i] ----
+    print("\n" + "=" * 104)
+    print(f"  (a) HONEST-ENTRY squeeze (bb14 sq0.8 dur>=5) — ingresso close[i], fee={fee_rt*100:.1f}% RT")
+    print("=" * 104)
+    print(f"  {'Asset':>5s}{'Regola direzione':>20s}{'Trd':>6s}{'Win%g':>8s}{'Win%n':>8s}{'Netto€':>9s}{'Ret%':>9s}{'DD%':>7s}")
+    print("  " + "-" * 100)
+
+    rules = [("breakout (=live)", "breakout"), ("pre-trend mom4", "premom"),
+             ("pre-trend mom8", "premom8"), ("fade breakout", "fade")]
+    for asset in ["BTC", "ETH"]:
+        df = load_data(asset, "15m").reset_index(drop=True)
+        split = int(len(df) * (1 - TEST_FRAC))
+        close = df["close"].values
+        rels = squeeze_releases(df, 14, 0.8, 5, split)
+        for label, rule in rules:
+            mom = 8 if rule == "premom8" else 4
+            ents = honest_entries(df, rels, "premom" if rule == "premom8" else rule, mom=mom)
+            r = simulate(ents, close, hold, fee_rt, lev, 0.15)
+            print(f"  {asset:>5s}{label:>20s}{r['trades']:>6d}{r['win_gross']:>8.1f}"
+                  f"{r['win_net']:>8.1f}{r['net_eur']:>+9.0f}{r['net_return_pct']:>+9.1f}{r['max_dd']:>7.1f}")
+        print("  " + "-" * 100)
+    print("  pre-trend = direzione dal trend PRIMA del rilascio (solo dati <= i-1): 100% honest.")
+    print("  Se nessuna regola honest batte ~breakeven, non esiste edge direzionale tradeable.")
+
+
+if __name__ == "__main__":
+    main()