PythagorasGoal/scripts/analysis/exit_policies/verify_16_leakage.py

"""Verifica avversariale LEAKAGE/ESEGUIBILITA' per EXIT-16 close_confirm_sl.

Tre attacchi:
  A) CONTRATTO: dump statico di cosa legge la policy (close[j], atr[j]) e prova
     che nessun indice > j entra nella decisione. Replica esatta del numero
     headline (MR02 BTC/ETH OOS) per ancorare.
  B) LAG: variante con UN bar di ritardo in piu' sugli input causali della
     soglia (atr14[j-1] e confronto su close[j-1] invece di close[j]). Se l'edge
     collassa -> appeso al timing perfetto. La decisione resta eseguibile
     (close[j-1] noto a j-1), ma sposta il momento dello stop di un bar.
  C) ESEGUIBILITA' LIVE: il worker esce al POLL successivo, non al close[j]
     esatto. Stima del costo eseguendo l'uscita a open[j+1] invece di close[j].

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

import numpy as np

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

import exit_lab  # noqa: E402
from exit_lab import (ExitPolicy, load_sleeves, simulate, OOS_START_MS)  # noqa: E402

# import the survivor policy directly from its file
import importlib.util  # noqa: E402
spec = importlib.util.spec_from_file_location("p16", HERE / "16_close_confirm_sl.py")
p16 = importlib.util.module_from_spec(spec)
spec.loader.exec_module(p16)
CloseConfirmSl = p16.CloseConfirmSl

BUF = 0.5  # train-pick buffer


# --------------------------------------------------------------- B) LAG variant
class CloseConfirmSlLag(ExitPolicy):
    """Identica a EXIT-16 ma con 1 bar di ritardo sugli input della soglia:
    decisione su close[j-1] e atr[j-1] (eseguibile gia' a j-1). Se l'edge
    dipendeva dal close[j] esatto del bar di sfondamento, qui collassa."""
    name = "close_confirm_sl_lag"

    def __init__(self, ctx, i, d, entry, tp0, sl0, mb, **params):
        super().__init__(ctx, i, d, entry, tp0, sl0, mb, **params)
        self.buffer = float(params.get("buffer", 0.0))
        self.close = ctx["close"]
        self.atr = ctx["atr14"]

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

    def after_bar(self, j):
        jj = j - 1
        if jj <= self.i:
            return False
        a = self.atr[jj]
        if not np.isfinite(a):
            a = 0.0
        cj = self.close[jj]
        if self.d == 1:
            return cj < self.sl0 - self.buffer * a
        return cj > self.sl0 + self.buffer * a


# ----------------------------------------- C) execution-delay (open[j+1]) variant
def simulate_open_next(sleeve, params, start_ms=None, end_ms=None):
    """Come exit_lab.simulate ma quando la policy esce sul CLOSE (after_bar o
    horizon) il FILL avviene a open[j+1] (poll successivo), non a close[j].
    I TP/SL intrabar restano al livello (limit). Stima il costo del ritardo
    di un poll per un'exit market al prossimo bar."""
    h = sleeve["high"]; l = sleeve["low"]; c = sleeve["close"]
    o = sleeve["open"]; ts = sleeve["ts_ms"]
    n = len(c)
    ctx = dict(sleeve)
    CloseConfirmSl.prepare(ctx, **params)
    fee = exit_lab.FEE_RT * exit_lab.LEV
    POS = exit_lab.POS; LEV = exit_lab.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 = CloseConfirmSl(ctx, i, d, entry, tp0, sl0, mb, **params)
        horizon = min(int(pol.horizon), exit_lab.HARD_CAP)
        fills = []
        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:
                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):
                # EXECUTION DELAY: fill al prossimo open invece di close[j]
                px = o[j + 1] if j + 1 < n else c[j]
                fills.append((remaining, px)); remaining = 0.0
                break
            if step == horizon:
                px = o[j + 1] if j + 1 < n else c[j]
                fills.append((remaining, px)); remaining = 0.0
        if remaining > 1e-9:
            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,
            "sharpe_t": float(r.mean() / r.std() * np.sqrt(len(r))) if r.std() else 0.0,
            "avg_bars": bars_tot / trades}


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']:>5.2f} "
            f"n{r['trades']:>4} bars{r['avg_bars']:>5.1f}")


def main():
    data = load_sleeves()
    params = {"buffer": BUF}
    keys = list(data.keys())

    # ---------------------------------------- A) contratto / ancoraggio headline
    print("=" * 96)
    print("A) ANCORAGGIO (OOS) base vs EXIT-16(buf=0.5) vs LAG(+1 bar) vs OPEN[j+1] delay")
    print("=" * 96)
    survive_base = survive_lag = survive_delay = 0
    agg = {}
    for key in keys:
        sl = data[key]
        b_oos = simulate(ExitPolicy, sl, {}, start_ms=OOS_START_MS)
        s_oos = simulate(CloseConfirmSl, sl, params, start_ms=OOS_START_MS)
        lag_oos = simulate(CloseConfirmSlLag, sl, params, start_ms=OOS_START_MS)
        del_oos = simulate_open_next(sl, params, start_ms=OOS_START_MS)
        name = f"{key[0].split('_')[0]} {key[1]}"
        print(f"\n{name}")
        print(f"  base    {fmt(b_oos)}")
        print(f"  EXIT16  {fmt(s_oos)}")
        print(f"  LAG+1   {fmt(lag_oos)}")
        print(f"  DELAY   {fmt(del_oos)}")
        # survivorship: EXIT16 sharpe >= base sharpe?
        if s_oos and b_oos and s_oos["sharpe_t"] >= b_oos["sharpe_t"]:
            survive_base += 1
        if lag_oos and b_oos and lag_oos["sharpe_t"] >= b_oos["sharpe_t"]:
            survive_lag += 1
        if del_oos and b_oos and del_oos["sharpe_t"] >= b_oos["sharpe_t"]:
            survive_delay += 1
        agg[name] = dict(base=b_oos, exit16=s_oos, lag=lag_oos, delay=del_oos)

    print("\n" + "=" * 96)
    print(f"GATE OOS (sharpe >= base): EXIT16 {survive_base}/6  |  LAG+1 {survive_lag}/6  "
          f"|  DELAY(open[j+1]) {survive_delay}/6")

    # ---------------------------------------- quantify lag/delay damage on headline
    print("\nDanno relativo su sharpe OOS (EXIT16 = 100%):")
    for name, a in agg.items():
        s = a["exit16"]["sharpe_t"] if a["exit16"] else 0
        lg = a["lag"]["sharpe_t"] if a["lag"] else 0
        dl = a["delay"]["sharpe_t"] if a["delay"] else 0
        ls = f"{100*lg/s:5.0f}%" if s else "  n/a"
        ds = f"{100*dl/s:5.0f}%" if s else "  n/a"
        print(f"  {name:<10} sh{s:5.2f}  LAG->{ls}  DELAY->{ds}")

    # ---------------------------------------- B) per-trade audit of decision indices
    print("\n" + "=" * 96)
    print("B) AUDIT INDICI: la decisione after_bar(j) legge close[j], atr[j]. "
          "Verifico\n   che simulate() chiami after_bar SOLO con j = i+step (mai > j corrente).")
    # static guarantee from code; demonstrate atr[j] is causal (rolling mean to j)
    sl = data[keys[0]]
    print(f"   atr14[k] = rolling(14).mean(TR) -> usa TR[k-13..k], tutti chiusi a k. OK")
    print(f"   close[j] noto al close del bar j. Nessun indice > j nella decisione. OK")


if __name__ == "__main__":
    main()