PythagorasGoal/Old/scripts/analysis/exit_policies/verify_16_overfit.py

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