"""VERIFY EXIT-16 close_confirm_sl — lente OVERFIT/ROBUSTEZZA (avversariale). Ipotesi nulla: il risultato e' un artefatto (overfit di cella / di regime / di dipendenza dal loss-guard Hurst gia' applicato in cache). Tre test: (1) JITTER PARAMETRI: buffer fuori griglia {0.4, 0.6, 0.75, 1.0} + ponte verso la base con SL fisso a 3x/4x ATR (no_sl come limite). Il plateau tiene? (2) STABILITA' TEMPORALE: train 2018-20 vs 21-22; OOS 2023-11/2025-01 vs 2025-01/2026-05. Il miglioramento e' in OGNI finestra o concentrato? (3) DIPENDENZA HURST (decisivo): rigenero i segnali con hurst_max=None (NESSUN loss-guard, NON tocco la cache) e ripeto base-vs-policy. Se senza il guard la policy crolla, funziona SOLO grazie al guard -> condizione di validita'. """ import sys from pathlib import Path import numpy as np import pandas as pd sys.path.insert(0, str(Path(__file__).resolve().parents[1])) sys.path.insert(0, str(Path(__file__).resolve().parents[3])) import exit_lab # noqa: E402 from exit_lab import ExitPolicy, simulate, OOS_START_MS, _atr14 # noqa: E402 import importlib.util spec = importlib.util.spec_from_file_location( "cc16", str(Path(__file__).resolve().parent / "16_close_confirm_sl.py")) cc16 = importlib.util.module_from_spec(spec) spec.loader.exec_module(cc16) CloseConfirmSl = cc16.CloseConfirmSl from src.data.downloader import load_data # noqa: E402 from src.live.strategy_loader import load_strategy # noqa: E402 CODES = ["MR01_bollinger_fade", "MR02_donchian_fade", "MR07_return_reversal"] ASSETS = ("BTC", "ETH") # ---- policy ponte: SL fisso a multiplo di ATR (no_sl come limite) ---- class WideSlPolicy(ExitPolicy): """SL intrabar spostato a k*ATR oltre sl0 (ponte tra base e no-sl).""" name = "wide_sl" def __init__(self, ctx, i, d, entry, tp0, sl0, mb, **params): super().__init__(ctx, i, d, entry, tp0, sl0, mb, **params) self.k = float(params.get("k_atr", 2.0)) self.atr = ctx["atr14"] def levels(self, j: int): a = self.atr[j - 1] if np.isfinite(self.atr[j - 1]) else 0.0 # sl0 e' sotto (long) / sopra (short) l'entry; allarga di k*atr sl = self.sl0 - self.k * a if self.d == 1 else self.sl0 + self.k * a return self.tp0, sl, 1.0 def sub(cls, sleeve, g, s, e): return simulate(cls, sleeve, g, start_ms=s, end_ms=e) def fmt(r): if not r: return " n/a" return (f"ret{r['ret_pct']:>7.0f}% dd{r['dd_pct']:>5.1f} sh{r['sharpe_t']:>5.2f} " f"n{r['trades']:>4} bars{r['avg_bars']:>5.1f}") def build_signals(hurst_max): """Rigenera sleeve in memoria con hurst_max dato (None = no guard). NON tocca cache.""" out = {} params = dict(trend_max=3.0, ema_long=200, hurst_max=hurst_max, min_tp_frac=0.0015) for code in CODES: strat = load_strategy(code) for asset in ASSETS: df = load_data(asset, "1h") ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True) sigs = strat.generate_signals(df, ts, **params) h = df["high"].values.astype(float) l = df["low"].values.astype(float) c = df["close"].values.astype(float) out[(code, asset)] = { "signals": [(int(s.idx), int(s.direction), float(s.metadata["tp"]), float(s.metadata["sl"]), int(s.metadata["max_bars"])) for s in sigs], "open": df["open"].values.astype(float), "high": h, "low": l, "close": c, "ts_ms": df["timestamp"].values.astype(np.int64), "atr14": _atr14(h, l, c), } return out def main(): data = exit_lab.load_sleeves() keys = list(data.keys()) # ===== TEST 1: JITTER PARAMETRI ===== print("=" * 100) print("TEST 1 — JITTER buffer fuori griglia + ponte WIDE-SL (OOS, dopo 2023-11)") print("=" * 100) jit_buffers = [0.4, 0.6, 0.75, 1.0] all_pos = True for (code, asset), sleeve in data.items(): key = f"{code.split('_')[0]} {asset}" base = sub(ExitPolicy, sleeve, {}, OOS_START_MS, None) line = f"{key:<10} BASE {fmt(base)}" print(line) for b in jit_buffers: r = sub(CloseConfirmSl, sleeve, {"buffer": b}, OOS_START_MS, None) better = r and base and r["sharpe_t"] >= base["sharpe_t"] - 0.10 all_pos &= bool(better) print(f" buf={b:<4} {fmt(r)} {'OK' if better else 'WORSE'}") print() print(f"JITTER buffer: tutte >= base-0.10 sharpe? {all_pos}\n") print("-" * 100) print("PONTE WIDE-SL: SL intrabar fisso allargato a k*ATR (k grande -> verso no-sl)") print("-" * 100) for (code, asset), sleeve in data.items(): key = f"{code.split('_')[0]} {asset}" base = sub(ExitPolicy, sleeve, {}, OOS_START_MS, None) print(f"{key:<10} BASE(k=0) {fmt(base)}") for k in [1.5, 3.0, 4.0]: r = sub(WideSlPolicy, sleeve, {"k_atr": k}, OOS_START_MS, None) print(f" k={k:<4} {fmt(r)}") print() # ===== TEST 2: STABILITA' TEMPORALE ===== print("=" * 100) print("TEST 2 — STABILITA' TEMPORALE (base vs policy buffer=0.5)") print("=" * 100) ms = lambda d: int(pd.Timestamp(d, tz="UTC").value // 1e6) windows = [ ("TRAIN 2018-20", None, ms("2021-01-01")), ("TRAIN 2021-22", ms("2021-01-01"), OOS_START_MS), ("OOS 23/11-25/01", OOS_START_MS, ms("2025-01-01")), ("OOS 25/01-26/05", ms("2025-01-01"), None), ] win_verdict = {w[0]: 0 for w in windows} win_total = {w[0]: 0 for w in windows} for (code, asset), sleeve in data.items(): key = f"{code.split('_')[0]} {asset}" print(f"\n{key}") for wname, s, e in windows: b = sub(ExitPolicy, sleeve, {}, s, e) p = sub(CloseConfirmSl, sleeve, {"buffer": 0.5}, s, e) if b and p: win_total[wname] += 1 # criterio: policy non peggio della base su sharpe (tol 0.15) imp = p["sharpe_t"] >= b["sharpe_t"] - 0.15 win_verdict[wname] += int(imp) tag = "OK " if imp else "BAD" else: tag = "n/a" print(f" {wname:<18} base {fmt(b)}") print(f" {'':<18} pol {fmt(p)} -> {tag}") print("\nPer-finestra (policy >= base-0.15 sharpe):") for w in windows: wn = w[0] print(f" {wn:<18} {win_verdict[wn]}/{win_total[wn]} sleeve OK") # ===== TEST 3: DIPENDENZA HURST (DECISIVO) ===== print("\n" + "=" * 100) print("TEST 3 — DIPENDENZA dal loss-guard HURST (DECISIVO)") print("Rigenero i segnali con hurst_max=None (NO guard, regime trending incluso).") print("Se la policy crolla -> funziona SOLO grazie al guard.") print("=" * 100) print("Generazione segnali SENZA hurst (puo' richiedere ~1-2 min)...") data_nohurst = build_signals(hurst_max=None) n_guard = sum(len(s["signals"]) for s in data.values()) n_nohurst = sum(len(s["signals"]) for s in data_nohurst.values()) print(f"Segnali totali: con guard {n_guard}, senza guard {n_nohurst} " f"(+{n_nohurst - n_guard} segnali in regime trending)\n") holds = True for region_name, s, e in [("TRAIN", None, OOS_START_MS), ("OOS", OOS_START_MS, None)]: print(f"--- {region_name} (segnali SENZA hurst guard) ---") for (code, asset), sleeve in data_nohurst.items(): key = f"{code.split('_')[0]} {asset}" b = sub(ExitPolicy, sleeve, {}, s, e) p = sub(CloseConfirmSl, sleeve, {"buffer": 0.5}, s, e) if b and p: imp = p["sharpe_t"] >= b["sharpe_t"] - 0.15 ddimp = p["dd_pct"] <= b["dd_pct"] + 1.0 holds &= bool(imp) tag = "OK " if imp else "POLICY WORSE" else: tag = "n/a" print(f" {key:<10} base {fmt(b)}") print(f" {'':<10} pol {fmt(p)} -> {tag}") print() print(f"TEST 3 verdict: policy regge SENZA il guard (>= base-0.15 sharpe ovunque)? {holds}") print("Se False -> la tesi 'SL dannoso' dipende dal guard (condizione di validita').") if __name__ == "__main__": main()