PythagorasGoal/scripts/analysis/exit16_port06_impact.py

"""TEST DECISIVO: impatto di EXIT-16 (close_confirm_sl, buffer=0.5 ATR) sul PORT06,
nel PATH CANONICO del backtest (NON exit_lab).

EXIT-16: lo SL intrabar e' DISATTIVATO; si esce al close del bar j solo se il close
ha sfondato il livello di buffer*ATR14:
  long  (d=1):  esci a close[j] se close[j] < sl0 - 0.5*atr14[j]
  short (d=-1): esci a close[j] se close[j] > sl0 + 0.5*atr14[j]
TP intrabar al livello e max_bars al close restano INVARIATI.

Metodo (come fu fatto per il loss-guard Hurst):
  1. build_everything() canonico -> equity giornaliere di TUTTI gli sleeve (cache intatta).
  2. ricostruisco le 6 equity fade in variante EXIT-16 replicando ESATTAMENTE
     fade_daily_equity/build_trades (stessi segnali fn(df,**params), trend_max=3.0,
     fee 0.10%RT*lev3, pos 0.15, compounding, non-overlap), cambiando SOLO il ramo SL.
  3. PARITA': con la SL-rule originale il replay deve riprodurre le equity canoniche.
  4. PORT06 base vs EXIT-16 con la STESSA matematica dei pesi (Portfolio.backtest):
     weighting cap, caps PAIRS 0.33, ribilancio 1D, metriche FULL e OOS.

NB: la leva 2x del portfolios.yml NON entra nel backtest (Portfolio.backtest la ignora;
e' un knob live). Le equity fade gia' includono lev=3 dentro build_trades.
"""
from __future__ import annotations

import sys
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(PROJECT_ROOT))

from src.data.downloader import load_data
from scripts.analysis.risk_management import strats_for
from scripts.analysis.combine_portfolio import OOS_DATE
from scripts.analysis._port06_gate_common import (
    build_trades_variant, equity_from_trades, port_metrics, dd as _dd,
)
from scripts.portfolios._defs import PORTFOLIOS


def fade_equity_variant(asset, fn, params, mode):
    """Stesso flusso di combine_portfolio.fade_daily_equity ma con build_trades_variant."""
    df = load_data(asset, "1h")
    trades = build_trades_variant(fn(df, **params), df, mode=mode, trend_max=3.0)
    return equity_from_trades(df, trades)


def main():
    p = PORTFOLIOS["PORT06"]
    fade_ids = [s.sid for s in p.sleeves if s.sid.startswith("MR")]
    print("=" * 96)
    print("  TEST DECISIVO EXIT-16 (close_confirm_sl buffer=0.5 ATR) su PORT06 — path canonico")
    print(f"  fade sleeve: {fade_ids}")
    print("=" * 96)

    # --- 1. equity canoniche di TUTTI gli sleeve (cache intatta) ---
    print("\n[1] build_everything() canonico (pesante, ~2-3 min)...")
    from src.portfolio.sleeves import all_sleeve_equities
    eq_base = dict(all_sleeve_equities())  # {sid: equity giornaliera}
    print(f"    sleeve totali: {len(eq_base)}")

    # --- 2. PARITA': replay 'orig' deve riprodurre le equity canoniche ---
    print("\n[2] PARITA' replay (mode=orig) vs canonico (fade_daily_equity):")
    print(f"    {'sleeve':<10s}{'corr':>10s}{'ret_canon%':>14s}{'ret_replay%':>14s}{'diff%':>9s}")
    parity_ok = True
    eq_orig, eq_e16 = {}, {}
    for asset in ("BTC", "ETH"):
        for nm, (fn, params) in strats_for(asset).items():
            sid = f"{nm}_{asset}"
            if sid not in fade_ids:
                continue
            eq_orig[sid] = fade_equity_variant(asset, fn, params, mode="orig")
            eq_e16[sid] = fade_equity_variant(asset, fn, params, mode="exit16")
            base = eq_base[sid]
            rep = eq_orig[sid]
            corr = base.pct_change().fillna(0).corr(rep.pct_change().fillna(0))
            rb = (base.iloc[-1] / base.iloc[0] - 1) * 100
            rr = (rep.iloc[-1] / rep.iloc[0] - 1) * 100
            diff = rr - rb
            flag = "" if (corr > 0.999 and abs(diff) <= max(1.0, abs(rb) * 0.01)) else "  <-- MISMATCH"
            if flag:
                parity_ok = False
            print(f"    {sid:<10s}{corr:>10.5f}{rb:>14.1f}{rr:>14.1f}{diff:>+9.2f}{flag}")
    print(f"\n    PARITA' {'OK' if parity_ok else 'FALLITA'} "
          f"(corr>0.999 e ret finale entro 1%).")
    if not parity_ok:
        print("\n    >>> Parita' non raggiunta: NON forzo. Diagnostico sopra. STOP.")
        return

    # --- 3. PORT06 base vs EXIT-16: stessi pesi cap, stessa matematica ---
    members_base = dict(eq_base)
    members_e16 = dict(eq_base)
    for sid in fade_ids:
        members_e16[sid] = eq_e16[sid]  # sostituisco SOLO le 6 colonne fade

    # pesi cap canonici (gli stessi che usa Portfolio.backtest) dentro port_metrics
    f_b, o_b = port_metrics(members_base, p)
    f_e, o_e = port_metrics(members_e16, p)

    print("\n" + "=" * 96)
    print(f"  [3] PORT06 — pesi={p.weighting} caps={p.caps} | OOS da {OOS_DATE} | leva3x interna fade, pos0.15")
    print("=" * 96)
    print(f"  {'variante':<14s}{'FULL Sh':>9s}{'FULL DD%':>10s}{'FULL CAGR':>11s}"
          f"   | {'OOS Sh':>8s}{'OOS DD%':>9s}{'OOS CAGR':>10s}")
    print("  " + "-" * 90)
    print(f"  {'BASE':<14s}{f_b['sharpe']:>9.2f}{f_b['dd']:>10.2f}{f_b['cagr']:>10.0f}%"
          f"   | {o_b['sharpe']:>8.2f}{o_b['dd']:>9.2f}{o_b['cagr']:>9.0f}%")
    print(f"  {'EXIT-16':<14s}{f_e['sharpe']:>9.2f}{f_e['dd']:>10.2f}{f_e['cagr']:>10.0f}%"
          f"   | {o_e['sharpe']:>8.2f}{o_e['dd']:>9.2f}{o_e['cagr']:>9.0f}%")
    print("  " + "-" * 90)
    print(f"  {'DELTA':<14s}{f_e['sharpe']-f_b['sharpe']:>+9.2f}{f_e['dd']-f_b['dd']:>+10.2f}"
          f"{f_e['cagr']-f_b['cagr']:>+10.0f}%   | {o_e['sharpe']-o_b['sharpe']:>+8.2f}"
          f"{o_e['dd']-o_b['dd']:>+9.2f}{o_e['cagr']-o_b['cagr']:>+9.0f}%")

    # --- per-sleeve fade: differenze principali ---
    print("\n  Per-sleeve fade (equity FULL ret%, EXIT-16 vs orig-replay):")
    print(f"    {'sleeve':<10s}{'orig ret%':>12s}{'exit16 ret%':>14s}{'delta%':>10s}"
          f"{'orig DD%':>10s}{'e16 DD%':>10s}")
    for sid in fade_ids:
        ro = eq_orig[sid]; re = eq_e16[sid]
        rro = (ro.iloc[-1] / ro.iloc[0] - 1) * 100
        rre = (re.iloc[-1] / re.iloc[0] - 1) * 100
        print(f"    {sid:<10s}{rro:>12.1f}{rre:>14.1f}{rre-rro:>+10.1f}"
              f"{_dd(ro):>10.1f}{_dd(re):>10.1f}")

    # --- GATE ---
    print("\n" + "=" * 96)
    print("  GATE (stesso del loss-guard): PROMOSSO se OOS Sharpe migliora/pari E DD non peggiora")
    print("              materialmente, E in FULL non degrada.")
    print("=" * 96)
    oos_sh_ok = o_e['sharpe'] >= o_b['sharpe'] - 0.02
    oos_dd_ok = o_e['dd'] <= o_b['dd'] + 0.20            # no peggioramento materiale DD
    full_ok = f_e['sharpe'] >= f_b['sharpe'] - 0.02 and f_e['dd'] <= f_b['dd'] + 0.20
    promoted = oos_sh_ok and oos_dd_ok and full_ok
    print(f"  OOS Sharpe {o_b['sharpe']:.2f} -> {o_e['sharpe']:.2f} "
          f"({'OK' if oos_sh_ok else 'KO'})")
    print(f"  OOS DD%    {o_b['dd']:.2f} -> {o_e['dd']:.2f} "
          f"({'OK' if oos_dd_ok else 'KO'})")
    print(f"  FULL Sharpe {f_b['sharpe']:.2f} -> {f_e['sharpe']:.2f} | "
          f"FULL DD {f_b['dd']:.2f} -> {f_e['dd']:.2f} ({'OK' if full_ok else 'KO'})")
    print("\n  VERDETTO: " + (">>> PROMOSSO <<<" if promoted else ">>> BOCCIATO <<<"))


if __name__ == "__main__":
    main()