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

"""ADVERSARIAL VERIFY — EXIT-02 trail_atr_keep_tp, LENTE OVERFIT/ROBUSTEZZA.

Tesi del sopravvissuto: lo SL intrabar fisso distrugge valore nelle fade; il
Chandelier trail (k=1.5) + TP fisso migliora Sharpe/DD ovunque (6/6 train, 5/6 OOS).

Ipotesi nulla del verificatore: e' un artefatto. Tre attacchi:
 (1) JITTER parametri: k vicini non provati (1.25/1.75) + ponte SL fisso a 3x/4x
     ATR (no_sl). Il plateau tiene o e' una cresta?
 (2) STABILITA' TEMPORALE: train 2018-20 vs 21-22, OOS 23-11/25-01 vs 25-01/26-05.
     Il miglioramento c'e' in OGNI finestra o concentrato in un regime?
 (3) DIPENDENZA HURST (decisivo): i segnali in cache hanno hurst_max=0.55 (toglie
     il regime trending). Rigenero i segnali SENZA hurst (hurst_max=None) IN MEMORIA
     (non tocco la cache) e ripeto base-vs-policy: la tesi "SL dannoso" regge anche
     dove gli stop servivano (regime persistente)?

    cd /opt/docker/PythagorasGoal && PYTHONPATH=. uv run python \
        scripts/analysis/exit_policies/verify_02_overfit.py
"""
import sys
from pathlib import Path

import numpy as np
import pandas as pd

HERE = Path(__file__).resolve()
sys.path.insert(0, str(HERE.parents[1]))            # scripts/analysis
sys.path.insert(0, str(HERE.parents[3]))            # project root

import exit_lab  # noqa: E402
from exit_lab import (ExitPolicy, simulate, load_sleeves, OOS_START_MS,  # noqa: E402
                      CODES, ASSETS, LIVE_PARAMS, _atr14)
from importlib import import_module                  # noqa: E402

mod = import_module("exit_policies.02_trail_atr_keep_tp")
TrailATRKeepTP = mod.TrailATRKeepTP

from src.data.downloader import load_data            # noqa: E402
from src.live.strategy_loader import load_strategy   # noqa: E402

SLEEVE_KEYS = [(c, a) for c in CODES for a in ASSETS]


# --------------------------------------------------------------- fixed-SL bridge
class FixedSLmultATR(ExitPolicy):
    """Ponte fra base (SL=sl0) e no_sl: SL fisso a m*ATR(entry) dall'entrata,
    TP fisso. Se il trail (k piccolo) batte uno SL fisso GIA' largo (3x/4x),
    allora il guadagno e' nel trailing, non solo nell'allontanare lo SL."""
    name = "fixed_sl_mult_atr"

    def __init__(self, ctx, i, d, entry, tp0, sl0, mb, **params):
        super().__init__(ctx, i, d, entry, tp0, sl0, mb, **params)
        m = float(params.get("m", 3.0))
        a = ctx["atr14"][i]
        if a is None or a != a:
            self.sl = sl0
        else:
            self.sl = entry - m * a if d == 1 else entry + m * a

    def levels(self, j: int):
        return self.tp0, self.sl, 1.0


# --------------------------------------------------------------- no-SL bridge
class NoSL(ExitPolicy):
    """Solo TP fisso + horizon, NESSUNO stop. Isola: il valore e' nel TOGLIERE
    lo stop (qualsiasi) o nel TRAIL dinamico? Se NoSL ~ trail, il driver e'
    'niente SL'; se il trail batte NoSL, il trail aggiunge."""
    name = "no_sl"

    def levels(self, j: int):
        return self.tp0, None, 1.0


def _fmt(r):
    if not r:
        return "  (no trades)"
    return (f"ret{r['ret_pct']:>7.0f}% dd{r['dd_pct']:>5.1f} sh{r['sharpe_t']:>6.2f} "
            f"n{r['trades']:>4} bars{r['avg_bars']:>5.1f}")


def _summary(rows):
    """rows: list of (sleeve_key, base_dict, pol_dict). Ritorna conteggi miglioramento."""
    sh_up = dd_dn = ret_up = n = 0
    for _, b, p in rows:
        if not b or not p:
            continue
        n += 1
        sh_up += p["sharpe_t"] > b["sharpe_t"]
        dd_dn += p["dd_pct"] < b["dd_pct"]
        ret_up += p["ret_pct"] > b["ret_pct"]
    return sh_up, dd_dn, ret_up, n


# ============================================================= TEST 1: JITTER
def test_jitter(data):
    print("\n" + "=" * 78)
    print("TEST 1 — JITTER: k vicini (1.25/1.75) + ponte SL fisso 3x/4x ATR + NoSL")
    print("=" * 78)
    print("\n[1a] Trail k in {1.25, 1.5, 1.75} — plateau o cresta? (OOS)")
    for k in (1.25, 1.5, 1.75):
        rows = []
        for key in SLEEVE_KEYS:
            sl = data[key]
            b = simulate(ExitPolicy, sl, start_ms=OOS_START_MS)
            p = simulate(TrailATRKeepTP, sl, {"k": k}, start_ms=OOS_START_MS)
            rows.append((key, b, p))
        sh, dd, ret, n = _summary(rows)
        print(f"  k={k:<5} OOS: Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}")

    print("\n[1b] SL fisso a m*ATR dall'entrata (m=3,4) — uno stop largo basta? (OOS)")
    for m in (3.0, 4.0):
        rows = []
        for key in SLEEVE_KEYS:
            sl = data[key]
            b = simulate(ExitPolicy, sl, start_ms=OOS_START_MS)
            p = simulate(FixedSLmultATR, sl, {"m": m}, start_ms=OOS_START_MS)
            rows.append((key, b, p))
        sh, dd, ret, n = _summary(rows)
        print(f"  m={m:<5} OOS: Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}")

    print("\n[1c] NoSL (solo TP+horizon) — il driver e' 'togliere lo SL'? (OOS)")
    rows = []
    for key in SLEEVE_KEYS:
        sl = data[key]
        b = simulate(ExitPolicy, sl, start_ms=OOS_START_MS)
        p = simulate(NoSL, sl, start_ms=OOS_START_MS)
        rows.append((key, b, p))
    sh, dd, ret, n = _summary(rows)
    print(f"  NoSL  OOS: Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}")

    print("\n[1d] dettaglio per sleeve: base vs k=1.5 vs NoSL vs SLx3 (OOS)")
    for key in SLEEVE_KEYS:
        sl = data[key]
        b = simulate(ExitPolicy, sl, start_ms=OOS_START_MS)
        t = simulate(TrailATRKeepTP, sl, {"k": 1.5}, start_ms=OOS_START_MS)
        ns = simulate(NoSL, sl, start_ms=OOS_START_MS)
        f3 = simulate(FixedSLmultATR, sl, {"m": 3.0}, start_ms=OOS_START_MS)
        tag = f"{key[0].split('_')[0]} {key[1]}"
        print(f"  {tag:<10} base sh{b.get('sharpe_t',0):>6.2f} | trail1.5 sh{t.get('sharpe_t',0):>6.2f} "
              f"| NoSL sh{ns.get('sharpe_t',0):>6.2f} | SLx3 sh{f3.get('sharpe_t',0):>6.2f}")


# ============================================================= TEST 2: TEMPORAL
def test_temporal(data):
    print("\n" + "=" * 78)
    print("TEST 2 — STABILITA' TEMPORALE (Sharpe base -> trail k=1.5)")
    print("=" * 78)
    W = [
        ("train 2018-20", None, int(pd.Timestamp("2021-01-01", tz="UTC").value // 1e6)),
        ("train 2021-22", int(pd.Timestamp("2021-01-01", tz="UTC").value // 1e6), OOS_START_MS),
        ("OOS  23-11/25-01", OOS_START_MS, int(pd.Timestamp("2025-01-01", tz="UTC").value // 1e6)),
        ("OOS  25-01/26-05", int(pd.Timestamp("2025-01-01", tz="UTC").value // 1e6), None),
    ]
    for label, s, e in W:
        rows = []
        for key in SLEEVE_KEYS:
            sl = data[key]
            b = simulate(ExitPolicy, sl, start_ms=s, end_ms=e)
            p = simulate(TrailATRKeepTP, sl, {"k": 1.5}, start_ms=s, end_ms=e)
            rows.append((key, b, p))
        sh, dd, ret, n = _summary(rows)
        # mediana del delta-Sharpe
        deltas = [p["sharpe_t"] - b["sharpe_t"] for _, b, p in rows if b and p]
        med = float(np.median(deltas)) if deltas else 0.0
        print(f"  {label:<20} Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}  "
              f"median dSharpe {med:+.2f}")


# ============================================================= TEST 3: HURST
def _build_sleeves_no_hurst():
    """Rigenera i segnali SENZA il loss-guard Hurst (hurst_max=None), IN MEMORIA.
    Replica esattamente load_sleeves() ma con LIVE_PARAMS modificati."""
    params = dict(LIVE_PARAMS)
    params["hurst_max"] = None
    out = {}
    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 test_hurst(data):
    print("\n" + "=" * 78)
    print("TEST 3 — DIPENDENZA DAL FILTRO HURST (decisivo)")
    print("=" * 78)
    print("Rigenero segnali con hurst_max=None (loss-guard OFF -> include il regime")
    print("trending/persistente dove gli stop dovrebbero servire). Confronto base->trail.")
    nh = _build_sleeves_no_hurst()

    # quanti segnali in piu' (il guard ne toglieva)
    print("\n  segnali: con-guard -> senza-guard")
    for key in SLEEVE_KEYS:
        ng = len(data[key]["signals"])
        nn = len(nh[key]["signals"])
        tag = f"{key[0].split('_')[0]} {key[1]}"
        print(f"    {tag:<10} {ng:>4} -> {nn:>4}  (+{nn-ng})")

    for scope, s, e in [("TRAIN", None, OOS_START_MS), ("OOS", OOS_START_MS, None)]:
        print(f"\n  [{scope}] base vs trail k=1.5 — SENZA hurst guard")
        rows = []
        for key in SLEEVE_KEYS:
            sl = nh[key]
            b = simulate(ExitPolicy, sl, start_ms=s, end_ms=e)
            p = simulate(TrailATRKeepTP, sl, {"k": 1.5}, start_ms=s, end_ms=e)
            rows.append((key, b, p))
            tag = f"{key[0].split('_')[0]} {key[1]}"
            print(f"    {tag:<10} base {_fmt(b)}")
            print(f"    {'':<10} trail{_fmt(p)}")
        sh, dd, ret, n = _summary(rows)
        print(f"    --> Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}")

    # contro-prova: con-guard sugli STESSI scope, per isolare l'effetto guard
    print("\n  [CONTROLLO] stesso confronto CON hurst guard (cache):")
    for scope, s, e in [("TRAIN", None, OOS_START_MS), ("OOS", OOS_START_MS, None)]:
        rows = []
        for key in SLEEVE_KEYS:
            sl = data[key]
            b = simulate(ExitPolicy, sl, start_ms=s, end_ms=e)
            p = simulate(TrailATRKeepTP, sl, {"k": 1.5}, start_ms=s, end_ms=e)
            rows.append((key, b, p))
        sh, dd, ret, n = _summary(rows)
        print(f"    [{scope}] con-guard --> Sharpe-up {sh}/{n}  DD-down {dd}/{n}  ret-up {ret}/{n}")


if __name__ == "__main__":
    data = load_sleeves()
    test_jitter(data)
    test_temporal(data)
    test_hurst(data)
    print("\nDONE")