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

"""VERIFY EXIT-16 close_confirm_sl — lente STRESS (avversariale).

Ipotesi nulla: l'edge della close-confirm-SL e' fragile a frizioni reali.
Quattro stress, tutti su segnali cache (params LIVE, hurst_max=0.55):

(1) FEE 2x: FEE_RT=0.002 (vs 0.001). Penalizza le policy che girano piu' capitale.
(2) BEAR/CRASH 2021-01..2022-12 (LUNA/FTX/19-mag-21): worst-trade + 5 peggiori
    trade della policy vs base. Lo SL disattivato lascia correre le perdite?
(3) SLIPPAGE AVVERSO 20bps sulle uscite della policy: ogni fill di USCITA paga
    +20bps contro la posizione (prezzo di uscita peggiorato). L'edge regge?
    NB: lo applico SOLO alle uscite della POLICY (la sua tesi e' "esco al close":
    il close-fill e' market, paga slippage; la base esce a livelli limite sl0/tp0).
(4) OVERLAP/TURNOVER: la policy allunga la permanenza (no stop intrabar). Conto
    i segnali SALTATI per non-overlap (i <= last_exit) base vs policy, e quanto
    capitale-tempo (somma bars in posizione) gira in piu'.

Tutto via simulate() con monkeypatch di FEE_RT e una sottoclasse engine per lo
slippage. Niente modifiche ad altri file.
"""
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, HARD_CAP, LEV, POS  # 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

BUF = 0.5  # train-pick


def fmt(r):
    if not r:
        return "   n/a"
    return (f"ret{r['ret_pct']:>8.0f}% dd{r['dd_pct']:>5.1f} sh{r['sharpe_t']:>5.2f} "
            f"n{r['trades']:>4} win{r['win_pct']:>4.0f} bars{r['avg_bars']:>5.1f}")


def sub(cls, sleeve, g, s, e):
    return simulate(cls, sleeve, g, start_ms=s, end_ms=e)


def ms(d):
    return int(pd.Timestamp(d, tz="UTC").value // 1e6)


# ===========================================================================
# Engine "instrumented" che riproduce simulate() ma:
#  - applica uno slippage avverso (bps) su OGNI fill di USCITA (solo se policy)
#  - raccoglie la lista dei ret per-trade e i segnali SALTATI per non-overlap
#  - raccoglie capital-time (somma bars)
# Lo tengo allineato 1:1 con exit_lab.simulate (stesso ordine SL-prima-di-TP).
# ===========================================================================
def simulate_instr(policy_cls, sleeve, params=None, start_ms=None, end_ms=None,
                   exit_slip_bps=0.0):
    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 = exit_lab.FEE_RT * LEV
    slip = exit_slip_bps * 1e-4
    capital = peak = 1000.0
    max_dd = 0.0
    last_exit = -1
    trades = wins = 0
    bars_tot = 0
    skipped_overlap = 0
    rets = []           # (ret, ts_entry, bars)
    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 + 1 >= n:
            continue
        if i <= last_exit:
            skipped_overlap += 1
            continue
        entry = c[i]
        pol = policy_cls(ctx, i, d, entry, tp0, sl0, mb, **params)
        horizon = min(int(pol.horizon), 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):
                fills.append((remaining, c[j])); remaining = 0.0
                break
            if step == horizon:
                fills.append((remaining, c[j])); remaining = 0.0
        if remaining > 1e-9:
            fills.append((remaining, c[j]))
        # slippage avverso sull'uscita: il prezzo di uscita peggiora di slip,
        # cioe' si vende piu' basso (long) / si ricompra piu' alto (short).
        def adj(p):
            return p * (1.0 - slip) if d == 1 else p * (1.0 + slip)
        ret = sum(f * (adj(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, int(ts[i]), j - i))
    if trades == 0:
        return {}
    r = np.array([x[0] for x in 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,
        "bars_tot": bars_tot,
        "skipped_overlap": skipped_overlap,
        "rets": rets,
        "worst5": sorted(r.tolist())[:5],
    }


def main():
    data = exit_lab.load_sleeves()

    # ===================================================================
    print("=" * 104)
    print("TEST 1 — FEE 2x (FEE_RT 0.001 -> 0.002).  base vs policy buffer=0.5  (OOS, dopo 2023-11)")
    print("=" * 104)
    orig_fee = exit_lab.FEE_RT
    survive_fee = True
    for fee in (0.001, 0.002):
        exit_lab.FEE_RT = fee
        print(f"\n--- FEE_RT={fee} ---")
        for (code, asset), sleeve in data.items():
            key = f"{code.split('_')[0]} {asset}"
            b = sub(ExitPolicy, sleeve, {}, OOS_START_MS, None)
            p = sub(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None)
            tag = ""
            if fee == 0.002 and b and p:
                # regge se sharpe policy >= base (la tesi e' che migliora)
                ok = p["sharpe_t"] >= b["sharpe_t"] - 0.10
                survive_fee &= ok
                tag = "OK" if ok else "WORSE"
            print(f"  {key:<10} base {fmt(b)}")
            print(f"  {'':<10} pol  {fmt(p)} {tag}")
    exit_lab.FEE_RT = orig_fee
    print(f"\nFEE 2x: policy regge (>= base-0.10 sh su tutti gli sleeve OOS)? {survive_fee}")

    # ===================================================================
    print("\n" + "=" * 104)
    print("TEST 2 — BEAR/CRASH 2021-01..2022-12 (LUNA/FTX/19-mag): worst-trade + 5 peggiori")
    print("=" * 104)
    s2, e2 = ms("2021-01-01"), ms("2023-01-01")
    tail_worse = 0
    tail_total = 0
    for (code, asset), sleeve in data.items():
        key = f"{code.split('_')[0]} {asset}"
        b = simulate_instr(ExitPolicy, sleeve, {}, s2, e2)
        p = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, s2, e2)
        print(f"\n{key}")
        print(f"  base {fmt(b)}")
        print(f"  pol  {fmt(p)}")
        if b and p:
            bw = [f"{x*100:+.1f}%" for x in b["worst5"]]
            pw = [f"{x*100:+.1f}%" for x in p["worst5"]]
            print(f"  base 5 peggiori (ret netto): {bw}")
            print(f"  pol  5 peggiori (ret netto): {pw}")
            tail_total += 1
            # la policy peggiora la coda se il worst-trade e' piu' negativo
            if p["worst5"][0] < b["worst5"][0] - 0.005:
                tail_worse += 1
                print(f"  -> CODA PEGGIORE: worst {p['worst5'][0]*100:+.1f}% < base {b['worst5'][0]*100:+.1f}%")
            else:
                print(f"  -> coda OK: worst {p['worst5'][0]*100:+.1f}% vs base {b['worst5'][0]*100:+.1f}%")
            print(f"  DD bear: base {b['dd_pct']:.1f}%  pol {p['dd_pct']:.1f}%")
    print(f"\nBEAR: sleeve con coda PEGGIORE (worst-trade > 0.5pt sotto base): {tail_worse}/{tail_total}")

    # ===================================================================
    print("\n" + "=" * 104)
    print("TEST 3 — SLIPPAGE AVVERSO 20bps sulle uscite della POLICY (OOS).  base senza slippage")
    print("(la tesi della policy e' 'esco al close' = market fill -> paga slippage)")
    print("=" * 104)
    survive_slip = True
    for (code, asset), sleeve in data.items():
        key = f"{code.split('_')[0]} {asset}"
        b = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None, exit_slip_bps=0.0)
        p0 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=0.0)
        p20 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=20.0)
        ok = p20 and b and p20["sharpe_t"] >= b["sharpe_t"] - 0.10
        survive_slip &= bool(ok)
        print(f"\n{key}")
        print(f"  base  (no slip)        {fmt(b)}")
        print(f"  pol   (no slip)        {fmt(p0)}")
        print(f"  pol   (+20bps exit)    {fmt(p20)}  {'OK' if ok else 'WORSE vs base'}")
    print(f"\nSLIPPAGE 20bps: policy ancora >= base-0.10 sh su tutti? {survive_slip}")
    print("(test severo: lo slippage colpisce la policy ma NON la base — asimmetria pessimistica)")

    # severita' extra: slippage anche sulla base (entrambe market) per fairness
    print("\n--- fairness: 20bps anche sulle uscite della BASE ---")
    fair = True
    for (code, asset), sleeve in data.items():
        key = f"{code.split('_')[0]} {asset}"
        b20 = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None, exit_slip_bps=20.0)
        p20 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=20.0)
        ok = p20 and b20 and p20["sharpe_t"] >= b20["sharpe_t"] - 0.10
        fair &= bool(ok)
        print(f"  {key:<10} base+20 {fmt(b20)}")
        print(f"  {'':<10} pol +20 {fmt(p20)}  {'OK' if ok else 'WORSE'}")
    print(f"fairness (entrambe +20bps): policy >= base-0.10 sh? {fair}")

    # ===================================================================
    print("\n" + "=" * 104)
    print("TEST 4 — OVERLAP/TURNOVER: segnali saltati per non-overlap + capital-time (OOS)")
    print("=" * 104)
    for (code, asset), sleeve in data.items():
        key = f"{code.split('_')[0]} {asset}"
        b = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None)
        p = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None)
        if b and p:
            dskip = p["skipped_overlap"] - b["skipped_overlap"]
            dbars = p["bars_tot"] - b["bars_tot"]
            print(f"  {key:<10} base: trades {b['trades']:>4} skip-overlap {b['skipped_overlap']:>4} "
                  f"bars_tot {b['bars_tot']:>6} avg {b['avg_bars']:.1f}")
            print(f"  {'':<10} pol : trades {p['trades']:>4} skip-overlap {p['skipped_overlap']:>4} "
                  f"bars_tot {p['bars_tot']:>6} avg {p['avg_bars']:.1f}")
            print(f"  {'':<10} -> +{dskip} segnali persi per overlap, "
                  f"+{dbars} bars in posizione ({dbars/max(b['bars_tot'],1)*100:+.0f}% capital-time)")


if __name__ == "__main__":
    main()