PythagorasGoal/scripts/analysis/sh01_exit_policies/05_breakeven_ratchet.py

"""SH01 EXIT policy 05 — breakeven_ratchet.

Disaster-stop ampio + ratchet a breakeven. Idea: NON tagliare i loser presto
(quello distrugge l'edge, lezione exit-lab sulle fade), ma proteggere SOLO i
trade gia' andati in profitto, alzando lo stop a entry (o entry+b*ATR) una volta
che il move e' partito. Il disaster-stop iniziale (4*ATR14[i]) taglia la coda
estrema (il crash ETH -15.6%) senza interferire coi trade normali.

Logica:
  long :
    sl_init = entry - 4 * ATR14[i]
    quando close[<=j-1] >= entry + a * ATR14[i]  ->  sl = entry + b * ATR14[i]
  short: specchiato.
  RATCHET: una volta alzato lo stop a breakeven, NON riscende (st["armed"]).

Lo stop iniziale (4*ATR14[i]) e' FISSO sul valore noto a close[i] (open_trade);
il ratchet si arma leggendo close[j-1] (gia' chiuso quando il worker fissa il
livello per il bar j) -> nessun dato del bar j. ATR14[i] e' causale.

ANTI-LOOK-AHEAD: open_trade usa solo close[i]/ATR14[i]; levels(j) legge solo
close[j-1] per decidere l'arming e ATR14[i] (gia' fissato). OK.

Griglia: a in {0.5, 1.0, 1.5, 2.0} (soglia di arming in ATR) x b in {0, 0.25}
(dove va lo stop una volta armato: entry o entry+0.25 ATR) x mode {intrabar,
close}. Il disaster-stop 4*ATR e' fisso (la coda da tagliare e' a -15%, ~3 ATR).

Profilo SH01: hold a orizzonte, win ~50%, edge nell'asimmetria. Il rischio del
breakeven e' di chiudere a 0 i winner che vanno prima in drawdown leggero e poi
recuperano -> pronti a un NO se BTC degrada.

PROTOCOLLO: grid SOLO sul train (t_hi=OOS_START_MS). Plateau >=3 celle adiacenti
migliorative (adiacenza su a, con b/mode fissi). Poi OOS una volta su config +
2 vicine.

    cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/05_breakeven_ratchet.py
"""
from __future__ import annotations

import sys

sys.path.insert(0, "/opt/docker/PythagorasGoal")

from scripts.analysis.sh01_exit_lab import (  # noqa: E402
    ExitPolicy, OOS_START_MS, evaluate, load_sleeves, simulate,
)

DISASTER_ATR = 4.0


class BreakevenRatchet(ExitPolicy):
    def __init__(self, a: float, b: float = 0.0, mode: str = "intrabar"):
        self.a = float(a)
        self.b = float(b)
        self.mode = mode
        self.name = f"be_ratchet a={a:.1f} b={b:.2f} {mode}"

    def open_trade(self, ctx, i, d):
        entry = ctx["close"][i]
        a14 = ctx["atr14"][i]
        if not (a14 == a14 and a14 > 0):
            # nessun ATR valido -> nessuno stop (degenera a baseline su quel trade)
            return {"entry": entry, "atr": None, "sl_disaster": None, "armed": False}
        if d == 1:
            sl_dis = entry - DISASTER_ATR * a14
        else:
            sl_dis = entry + DISASTER_ATR * a14
        return {"entry": entry, "atr": a14, "sl_disaster": sl_dis, "armed": False}

    def levels(self, ctx, i, d, j, st):
        a14 = st["atr"]
        if a14 is None:
            return None, self.mode
        entry = st["entry"]
        cprev = ctx["close"][j - 1]   # gia' chiuso quando si fissa il livello per j
        # arming del ratchet (una volta armato resta armato)
        if not st["armed"]:
            if d == 1:
                if cprev >= entry + self.a * a14:
                    st["armed"] = True
            else:
                if cprev <= entry - self.a * a14:
                    st["armed"] = True
        if st["armed"]:
            if d == 1:
                sl = entry + self.b * a14
            else:
                sl = entry - self.b * a14
        else:
            sl = st["sl_disaster"]
        return sl, self.mode

    def after_bar(self, ctx, i, d, j, st):
        return False


# baseline numbers (exit a orizzonte puro) — dal prompt/harness
BASELINE = {
    "BTC": {"train": dict(ret=127, dd=23, sharpe=2.09, worst=-5.5),
            "oos":   dict(ret=41, dd=8, sharpe=2.18, worst=-3.1)},
    "ETH": {"train": dict(ret=-26, dd=61, sharpe=-0.16, worst=-14.9),
            "oos":   dict(ret=143, dd=7, sharpe=3.60, worst=-4.6)},
}

A_VALS = [0.5, 1.0, 1.5, 2.0]
B_VALS = [0.0, 0.25]
MODES = ["intrabar", "close"]


def _row(tag, a, r):
    print(f"    {tag:<10s} {a}: ret={r['ret']:>+7.0f}% dd={r['dd']:>4.0f}% "
          f"shrp={r['sharpe']:>5.2f} worst={r['worst']:>+5.1f}% "
          f"stop={r['stop_rate']:>4.1f}% trades={r['trades']}")


def _eth_ok(et, b_eth):
    return (et["sharpe"] > b_eth["sharpe"] and et["dd"] < b_eth["dd"]
            and et["worst"] > b_eth["worst"])


def _btc_ok(bt, b_btc):
    return (bt["sharpe"] >= 0.95 * b_btc["sharpe"]
            and bt["ret"] >= 0.80 * b_btc["ret"])


def main():
    sleeves = load_sleeves()
    b_btc, b_eth = BASELINE["BTC"]["train"], BASELINE["ETH"]["train"]

    print("=" * 78)
    print("TRAIN GRID (selezione SOLO sul train, t_hi=OOS_START)")
    print("=" * 78)
    print("  baseline (orizzonte puro):")
    evaluate(ExitPolicy(), sleeves=sleeves)
    print()

    # train[(mode,b,a)] -> {asset: result}
    train = {}
    for mode in MODES:
        for b in B_VALS:
            print(f"  --- mode={mode} b={b:.2f} ---")
            for a in A_VALS:
                pol = BreakevenRatchet(a, b, mode)
                row = {}
                for asset in ("BTC", "ETH"):
                    row[asset] = simulate(sleeves[asset], pol, t_hi=OOS_START_MS)
                train[(mode, b, a)] = row
                print(f"  a={a:.1f} b={b:.2f} ({mode})")
                _row("TRAIN", "BTC", row["BTC"])
                _row("TRAIN", "ETH", row["ETH"])
            print()

    print("=" * 78)
    print("PLATEAU CHECK (train): ETH sharpe up & dd down & worst up,")
    print("                       BTC sharpe>=95% & ret>=80% baseline")
    print("=" * 78)
    # improving[(mode,b)] -> [a...]
    improving = {}
    for mode in MODES:
        for b in B_VALS:
            imp = []
            for a in A_VALS:
                bt, et = train[(mode, b, a)]["BTC"], train[(mode, b, a)]["ETH"]
                eth_ok = _eth_ok(et, b_eth)
                btc_ok = _btc_ok(bt, b_btc)
                ok = eth_ok and btc_ok
                if ok:
                    imp.append(a)
                print(f"  {mode:<9s} b={b:.2f} a={a:.1f}  ETH_ok={eth_ok}  "
                      f"BTC_ok={btc_ok}  -> {'IMPROVING' if ok else '-'}")
            improving[(mode, b)] = imp
            print(f"  improving cells ({mode}, b={b:.2f}): {imp}")

    # plateau = >=3 a adiacenti improving in QUALCHE (mode,b)
    best_plateau, best_key = [], None
    for key in improving:
        imp = improving[key]
        for idx in range(len(A_VALS)):
            run = []
            for jj in range(idx, len(A_VALS)):
                if A_VALS[jj] in imp:
                    run.append(A_VALS[jj])
                else:
                    break
            if len(run) >= 3 and len(run) > len(best_plateau):
                best_plateau, best_key = run, key
    print(f"  longest adjacent improving run: {best_plateau} (key={best_key}) "
          f"plateau={'YES' if len(best_plateau) >= 3 else 'NO'}")

    chosen = None
    if len(best_plateau) >= 3:
        chosen_a = best_plateau[len(best_plateau) // 2]
        chosen = (best_key[0], best_key[1], chosen_a)
    else:
        cands = [(m, b, a) for (m, b) in improving for a in improving[(m, b)]]
        if cands:
            chosen = max(cands, key=lambda mba: train[mba]["ETH"]["sharpe"])

    print()
    print("=" * 78)
    if chosen is None:
        print("NESSUNA cella migliorativa sul train -> verdetto NO (niente OOS).")
        print("=" * 78)
        return {"chosen": None, "plateau": best_plateau, "improving": improving,
                "passes": False, "train": train}

    c_mode, c_b, c_a = chosen
    print(f"CHOSEN a={c_a:.1f} b={c_b:.2f} mode={c_mode} -> OOS (config + 2 vicine a)")
    print("=" * 78)
    ai = A_VALS.index(c_a)
    neigh = [A_VALS[x] for x in (ai - 1, ai, ai + 1) if 0 <= x < len(A_VALS)]
    oos = {}
    for a in neigh:
        pol = BreakevenRatchet(a, c_b, c_mode)
        row = {}
        for asset in ("BTC", "ETH"):
            row[asset] = {"train": train[(c_mode, c_b, a)][asset],
                          "oos": simulate(sleeves[asset], pol, t_lo=OOS_START_MS)}
        oos[a] = row
        print(f"  a={a:.1f} b={c_b:.2f} ({c_mode})")
        _row("TRAIN", "BTC", row["BTC"]["train"])
        _row("OOS", "BTC", row["BTC"]["oos"])
        _row("TRAIN", "ETH", row["ETH"]["train"])
        _row("OOS", "ETH", row["ETH"]["oos"])

    print()
    print("=" * 78)
    print(f"GATE finale (a={c_a:.1f} b={c_b:.2f} mode={c_mode}):")
    bt_tr, et_tr = oos[c_a]["BTC"]["train"], oos[c_a]["ETH"]["train"]
    bt_oo, et_oo = oos[c_a]["BTC"]["oos"], oos[c_a]["ETH"]["oos"]
    Bb_o, Be_o = BASELINE["BTC"]["oos"], BASELINE["ETH"]["oos"]

    a_train = _eth_ok(et_tr, b_eth)
    a_oos = (et_oo["sharpe"] > Be_o["sharpe"] and et_oo["dd"] < Be_o["dd"]
             and et_oo["worst"] > Be_o["worst"])
    cond_a = a_train and a_oos
    b_tr = _btc_ok(bt_tr, b_btc)
    b_oo = (bt_oo["sharpe"] >= 0.95 * Bb_o["sharpe"]
            and bt_oo["ret"] >= 0.80 * Bb_o["ret"])
    cond_b = b_tr and b_oo
    cond_c = et_oo["ret"] >= 0.80 * Be_o["ret"]
    cond_d = len(best_plateau) >= 3

    print(f"  a) ETH sharpe up & dd down & worst up (train&oos): {cond_a}")
    print(f"     train: shrp {et_tr['sharpe']:.2f} vs {b_eth['sharpe']:.2f} | "
          f"dd {et_tr['dd']:.0f} vs {b_eth['dd']:.0f} | "
          f"worst {et_tr['worst']:.1f} vs {b_eth['worst']:.1f}")
    print(f"     oos:   shrp {et_oo['sharpe']:.2f} vs {Be_o['sharpe']:.2f} | "
          f"dd {et_oo['dd']:.0f} vs {Be_o['dd']:.0f} | "
          f"worst {et_oo['worst']:.1f} vs {Be_o['worst']:.1f}")
    print(f"  b) BTC sharpe>=95% & ret>=80% (train&oos): {cond_b}")
    print(f"     train: shrp {bt_tr['sharpe']:.2f} (>={0.95*b_btc['sharpe']:.2f}) | "
          f"ret {bt_tr['ret']:.0f} (>={0.80*b_btc['ret']:.0f})")
    print(f"     oos:   shrp {bt_oo['sharpe']:.2f} (>={0.95*Bb_o['sharpe']:.2f}) | "
          f"ret {bt_oo['ret']:.0f} (>={0.80*Bb_o['ret']:.0f})")
    print(f"  c) ETH oos ret>=80% baseline ({0.80*Be_o['ret']:.0f}): {cond_c} "
          f"(ret={et_oo['ret']:.0f})")
    print(f"  d) plateau: {cond_d} ({best_plateau} key={best_key})")
    passes = cond_a and cond_b and cond_c and cond_d
    print(f"  PASSES GATE: {passes}")
    print("=" * 78)

    return {"chosen": chosen, "plateau": best_plateau, "improving": improving,
            "passes": passes, "oos": oos, "conds": (cond_a, cond_b, cond_c, cond_d)}


if __name__ == "__main__":
    main()