PythagorasGoal/Old/scripts/analysis/exit_lab.py

"""EXIT LAB — harness onesto e CONDIVISO per la ricerca di policy di uscita
(TP dinamico, SL dinamico/trailing, partial, ride) sulle fade attive.

Ricerca 2026-06-04 (>=20 agenti): ogni agente implementa una ExitPolicy in
scripts/analysis/exit_policies/<id>_<nome>.py e la valuta QUI, sugli STESSI
segnali (cache su disco) e con lo stesso engine intrabar di fade_base.

CONTRATTO ANTI-LOOK-AHEAD (vincolante, verra' verificato da agenti avversari):
  - i livelli attivi nel bar j (`levels(j)`) possono usare SOLO dati <= j-1
    (il worker live li fissa al close del bar precedente, poi il bar j li tocca);
  - `after_bar(j)` decide sul CLOSE del bar j (eseguibile al poll del tick);
  - indicatori: usare l'indice j-1 degli array causali (es. ctx["atr14"][j-1]).

PROTOCOLLO ANTI-OVERFIT (vincolante):
  - TRAIN = storico fino al 2023-11-01, OOS = dopo. La SELEZIONE dei parametri
    si fa SOLO sul train; l'OOS si guarda una volta, per il verdetto.
  - gate: il miglioramento deve tenere su ENTRAMBI gli asset e su TUTTE e 3 le
    strategie (train E oos), con plateau sulla griglia (non una cella isolata).
  - fee 0.10% RT x leva su tutto il notional; nessuna fee scontata sui limit.

Baseline = exit attuale (TP/SL fissi dall'entrata + max_bars): la parita' con
`partial_tp_ladder.py --base` e' verificata da `parity_check()`.

    uv run python scripts/analysis/exit_lab.py          # build cache + parity check
"""
from __future__ import annotations

import pickle
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            # noqa: E402
from src.live.strategy_loader import load_strategy   # noqa: E402

LIVE_PARAMS = dict(trend_max=3.0, ema_long=200, hurst_max=0.55, min_tp_frac=0.0015)
OOS_START_MS = int(pd.Timestamp("2023-11-01", tz="UTC").value // 1e6)
LEV, POS, FEE_RT = 3.0, 0.15, 0.001
CODES = ["MR01_bollinger_fade", "MR02_donchian_fade", "MR07_return_reversal"]
ASSETS = ("BTC", "ETH")
CACHE = PROJECT_ROOT / "data" / "cache" / "exit_lab_signals.pkl"
HARD_CAP = 240        # bound assoluto ai bar in posizione (policy "ride" comprese)


# ----------------------------------------------------------------------------- dati

def _atr14(h: np.ndarray, l: np.ndarray, c: np.ndarray) -> np.ndarray:
    pc = np.roll(c, 1); pc[0] = c[0]
    tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
    return pd.Series(tr).rolling(14).mean().values


def load_sleeves(refresh: bool = False) -> dict:
    """{(code, asset): sleeve} con cache. sleeve = {signals, open, high, low,
    close, ts_ms, atr14}. signals = [(i, d, tp0, sl0, mb), ...] dai params LIVE."""
    if CACHE.exists() and not refresh:
        with open(CACHE, "rb") as f:
            return pickle.load(f)
    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, **LIVE_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),
            }
            print(f"  cache {code} {asset}: {len(sigs)} segnali, {len(c)} barre "
                  f"({ts.iloc[0].date()} -> {ts.iloc[-1].date()})")
    CACHE.parent.mkdir(parents=True, exist_ok=True)
    with open(CACHE, "wb") as f:
        pickle.dump(out, f)
    return out


# ----------------------------------------------------------------------------- policy

class ExitPolicy:
    """Baseline = exit live attuale. Le sottoclassi ridefinisco levels/after_bar.

    Una ISTANZA per trade. `ctx` e' il dict sleeve (array completi + indicatori
    aggiunti da prepare()): per contratto si legge SOLO fino a j-1 in levels(j)
    e fino a j in after_bar(j)/on_partial(j).
    """
    name = "base"

    @classmethod
    def prepare(cls, ctx: dict, **params) -> None:
        """Pre-calcola array causali per-sleeve (una volta), es. SMA/EMA."""

    def __init__(self, ctx: dict, i: int, d: int, entry: float,
                 tp0: float, sl0: float, mb: int, **params):
        self.ctx, self.i, self.d, self.entry = ctx, i, d, entry
        self.tp0, self.sl0, self.mb = tp0, sl0, mb
        self.horizon = mb                     # le sottoclassi possono estendere (cap HARD_CAP)

    def levels(self, j: int):
        """Livelli ATTIVI nel bar j -> (tp, sl, tp_frac). None = livello assente.
        tp_frac = quota del RESIDUO che esce al tocco del TP (1.0 = tutta)."""
        return self.tp0, self.sl0, 1.0

    def on_partial(self, j: int, price: float, remaining: float) -> None:
        """Notifica del fill parziale al TP nel bar j (aggiorna lo stato qui)."""

    def after_bar(self, j: int) -> bool:
        """True = chiudi il residuo al close[j] (decisione sul close, eseguibile)."""
        return False


# ----------------------------------------------------------------------------- engine

def simulate(policy_cls, sleeve: dict, params: dict | None = None,
             start_ms: int | None = None, end_ms: int | None = None) -> dict:
    """Replay intrabar dei segnali dello sleeve con la policy. SL prioritario
    sul TP nello stesso bar (conservativo); fill parziali pesati; max_bars/
    horizon esce al close; non-overlap (una posizione per volta)."""
    params = params or {}
    h, l, c, ts = sleeve["high"], sleeve["low"], sleeve["close"], sleeve["ts_ms"]
    n = len(c)
    ctx = dict(sleeve)
    policy_cls.prepare(ctx, **params)
    fee = FEE_RT * LEV
    capital = peak = 1000.0
    max_dd = 0.0
    last_exit = -1
    trades = wins = 0
    bars_tot = 0
    rets = []

    for (i, d, tp0, sl0, mb) in sleeve["signals"]:
        if start_ms is not None and ts[i] < start_ms:
            continue
        if end_ms is not None and ts[i] >= end_ms:
            continue
        if i <= last_exit or i + 1 >= n:
            continue
        entry = c[i]
        pol = policy_cls(ctx, i, d, entry, tp0, sl0, mb, **params)
        horizon = min(int(pol.horizon), HARD_CAP)
        fills: list[tuple[float, float]] = []
        remaining = 1.0
        j = i
        for step in range(1, horizon + 1):
            j = i + step
            if j >= n:
                j = n - 1
                fills.append((remaining, c[j])); remaining = 0.0
                break
            tp, sl, tpfrac = pol.levels(j)
            hit_sl = sl is not None and ((d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl))
            hit_tp = tp is not None and ((d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp))
            if hit_sl:                                  # conservativo: SL prima del TP
                fills.append((remaining, sl)); remaining = 0.0
                break
            if hit_tp:
                f = min(max(tpfrac, 0.0), 1.0) * remaining
                if f > 0:
                    fills.append((f, tp)); remaining -= f
                if remaining <= 1e-9:
                    break
                pol.on_partial(j, tp, remaining)
            if pol.after_bar(j):
                fills.append((remaining, c[j])); remaining = 0.0
                break
            if step == horizon:
                fills.append((remaining, c[j])); remaining = 0.0
        if remaining > 1e-9:                            # safety (non dovrebbe accadere)
            fills.append((remaining, c[j]))

        ret = sum(f * (p - entry) for f, p in fills) / entry * d * LEV - fee
        capital = max(capital + capital * POS * ret, 10.0)
        peak = max(peak, capital)
        max_dd = max(max_dd, (peak - capital) / peak)
        last_exit = j
        trades += 1
        wins += ret > 0
        bars_tot += j - i
        rets.append(ret)

    if trades == 0:
        return {}
    r = np.array(rets)
    return {
        "ret_pct": (capital / 1000.0 - 1) * 100,
        "dd_pct": max_dd * 100,
        "trades": trades,
        "win_pct": wins / trades * 100,
        "avg_ret_bps": r.mean() * 1e4,
        "sharpe_t": float(r.mean() / r.std() * np.sqrt(len(r))) if r.std() else 0.0,
        "avg_bars": bars_tot / trades,
    }


# ----------------------------------------------------------------------------- report

def evaluate(policy_cls, grid: list[dict], data: dict | None = None,
             quiet: bool = False) -> dict:
    """Protocollo train/OOS su tutta la griglia. La selezione dei parametri va
    fatta SUL TRAIN (l'OOS si riporta, non si ottimizza). Ritorna dict
    {params_str: {sleeve: {train: {...}, oos: {...}}}} + baseline."""
    data = data or load_sleeves()
    out: dict = {}
    rows = [("base", ExitPolicy, {})] + [
        (", ".join(f"{k}={v}" for k, v in g.items()) or "default", policy_cls, g)
        for g in grid]
    for tag, cls, g in rows:
        out[tag] = {}
        for (code, asset), sleeve in data.items():
            key = f"{code.split('_')[0]} {asset}"
            tr = simulate(cls, sleeve, g, end_ms=OOS_START_MS)
            oo = simulate(cls, sleeve, g, start_ms=OOS_START_MS)
            out[tag][key] = {"train": tr, "oos": oo}
            if not quiet:
                print(f"{tag:<28}{key:<10}"
                      f"TRAIN ret{tr.get('ret_pct', 0):>7.0f}% dd{tr.get('dd_pct', 0):>5.1f} "
                      f"sh{tr.get('sharpe_t', 0):>5.2f} n{tr.get('trades', 0):>4} | "
                      f"OOS ret{oo.get('ret_pct', 0):>6.0f}% dd{oo.get('dd_pct', 0):>5.1f} "
                      f"sh{oo.get('sharpe_t', 0):>5.2f} n{oo.get('trades', 0):>4} "
                      f"bars{oo.get('avg_bars', 0):>5.1f}")
    return out


def parity_check() -> None:
    """La baseline qui deve riprodurre i numeri FULL di partial_tp_ladder (base):
    MR01 BTC ~92%/13.8dd, MR01 ETH ~194%/16.5dd, MR02 ETH ~2135%/16.2dd..."""
    data = load_sleeves()
    print("\nParity check baseline (FULL, atteso = partial_tp_ladder base):")
    expected = {("MR01_bollinger_fade", "BTC"): 92, ("MR01_bollinger_fade", "ETH"): 194,
                ("MR02_donchian_fade", "BTC"): 129, ("MR02_donchian_fade", "ETH"): 2135,
                ("MR07_return_reversal", "BTC"): 78, ("MR07_return_reversal", "ETH"): 115}
    ok = True
    for key, sleeve in data.items():
        r = simulate(ExitPolicy, sleeve)
        exp = expected[key]
        match = abs(r["ret_pct"] - exp) < 1.0
        ok &= match
        print(f"  {key[0].split('_')[0]} {key[1]}: ret {r['ret_pct']:.0f}% "
              f"(atteso ~{exp}) {'OK' if match else 'MISMATCH'}")
    print("PARITY", "OK" if ok else "FAILED")


if __name__ == "__main__":
    load_sleeves(refresh="--refresh" in sys.argv)
    parity_check()