From ca88e62a11dcf71ecb7a17f68826c77dd092ca9c Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Thu, 28 May 2026 19:57:15 +0000 Subject: [PATCH] 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) --- scripts/analysis/intrabar_test.py | 188 +++++++++++++++++++++ scripts/analysis/oos_validation.py | 259 +++++++++++++++++++++++++++++ 2 files changed, 447 insertions(+) create mode 100644 scripts/analysis/intrabar_test.py create mode 100644 scripts/analysis/oos_validation.py diff --git a/scripts/analysis/intrabar_test.py b/scripts/analysis/intrabar_test.py new file mode 100644 index 0000000..154dd6f --- /dev/null +++ b/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() diff --git a/scripts/analysis/oos_validation.py b/scripts/analysis/oos_validation.py new file mode 100644 index 0000000..0846974 --- /dev/null +++ b/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()