PythagorasGoal/scripts/analysis/sh01_exit_policies/06_giveback.py

"""SH01 EXIT policy 06 — giveback (profit-protection).

Protezione del profitto via after_bar (mode "close" implicito: uscita sempre al
close del bar j). Lo state traccia il PEAK FAVOREVOLE dei close da i (per long il
max close; per short il min close, specchiato). Si esce al close del bar j se:

    giveback = (peak_fav - close[j]) * d   >=  g * ATR14[j-1]      (retrace)
    profit_at_peak = (peak_fav - entry) * d >=  m * ATR_ref         (era in gain)

cioe' il trade aveva raggiunto un profitto di almeno m*ATR e poi ha ritracciato
di g*ATR dal massimo favorevole. Idea: lascia correre il momentum SH01 finche'
sale, ma protegge il guadagno quando rifiata — senza toccare i trade che non sono
mai andati in profitto (quelli muoiono a orizzonte come nel baseline, cosi' non
si crea un trailing-stop mascherato che taglia i winner-in-drawdown).

Griglia g in {1.0, 1.5, 2.0, 3.0} x m in {0.5, 1.0}.

ANTI-LOOK-AHEAD: after_bar(j) decide sul CLOSE del bar j (dato <= j, eseguibile
al poll). Il peak favorevole include close[j] (gia' chiuso quando si decide).
ATR di riferimento: usiamo ATR14[j-1] per la soglia di giveback (causale, come
i livelli) e ATR14[i] per la soglia di profit-at-peak (noto a close[i], cioe'
all'apertura del trade). open_trade usa solo close[i]/ATR14[i]. Nessun dato di
un bar futuro. OK.

Profilo SH01: hold a orizzonte (momentum), win ~50%, edge nell'asimmetria dei
winner. La famiglia "ride/trailing" sulle fade e' stata SCARTATA; il giveback e'
una variante condizionata-al-profitto, pensata per NON toccare i loser-che-
recuperano. Pronti a un NO se taglia comunque l'edge.

PROTOCOLLO: grid (g x m) SOLO sul train (t_hi=OOS_START_MS). Plateau >=3 celle
adiacenti migliorative (adiacenza su g, m fisso). Poi OOS una volta sulla config
scelta + 2 vicine.

    cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/06_giveback.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,
)


class Giveback(ExitPolicy):
    def __init__(self, g: float, m: float):
        self.g = float(g)
        self.m = float(m)
        self.name = f"giveback g={g:.1f} m={m:.1f}"

    def open_trade(self, ctx, i, d):
        entry = ctx["close"][i]
        a0 = ctx["atr14"][i]
        a0 = float(a0) if (a0 == a0 and a0 > 0) else 0.0
        # peak favorevole inizializzato all'entry; atr_ref per il profit-at-peak.
        return {"entry": entry, "peak": entry, "atr0": a0}

    def levels(self, ctx, i, d, j, st):
        # nessuno stop a livello: il giveback e' tutto in after_bar.
        return None, "close"

    def after_bar(self, ctx, i, d, j, st):
        close = ctx["close"]
        atr = ctx["atr14"]
        cj = close[j]
        # aggiorna il peak FAVOREVOLE con close[j] (gia' chiuso quando decidiamo).
        # per long: max close; per short: min close (= peak favorevole specchiato).
        if d == 1:
            if cj > st["peak"]:
                st["peak"] = cj
        else:
            if cj < st["peak"]:
                st["peak"] = cj

        a_gb = atr[j - 1]
        if not (a_gb == a_gb and a_gb > 0):
            return False
        a_pk = st["atr0"]
        if a_pk <= 0:
            return False

        # profitto raggiunto al peak favorevole (in direzione del trade).
        profit_at_peak = (st["peak"] - st["entry"]) * d
        if profit_at_peak < self.m * a_pk:
            return False
        # ritracciamento dal peak favorevole fino al close corrente.
        giveback = (st["peak"] - cj) * d
        return giveback >= self.g * a_gb


# 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)},
}

GS = [1.0, 1.5, 2.0, 3.0]
MS = [0.5, 1.0]


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[(m,g)] -> {asset: result}
    train = {}
    for m in MS:
        print(f"  --- m={m:.1f} (profit-at-peak threshold) ---")
        for g in GS:
            pol = Giveback(g, m)
            row = {}
            for a in ("BTC", "ETH"):
                row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
            train[(m, g)] = row
            print(f"  g={g:.1f} m={m:.1f}")
            _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 = {}  # m -> [g...]
    for m in MS:
        imp = []
        for g in GS:
            bt, et = train[(m, g)]["BTC"], train[(m, g)]["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(g)
            print(f"  m={m:.1f} g={g:.1f}  ETH_ok={eth_ok}  BTC_ok={btc_ok}  -> "
                  f"{'IMPROVING' if ok else '-'}")
        improving[m] = imp
        print(f"  improving cells (m={m:.1f}): {imp}")

    # plateau = >=3 g adiacenti improving in QUALCHE m
    best_plateau, best_m = [], None
    for m in MS:
        imp = improving[m]
        for idx in range(len(GS)):
            run = []
            for j in range(idx, len(GS)):
                if GS[j] in imp:
                    run.append(GS[j])
                else:
                    break
            if len(run) >= 3 and len(run) > len(best_plateau):
                best_plateau, best_m = run, m
    print(f"  longest adjacent improving run: {best_plateau} (m={best_m}) "
          f"plateau={'YES' if len(best_plateau) >= 3 else 'NO'}")

    chosen = None
    if len(best_plateau) >= 3:
        chosen_g = best_plateau[len(best_plateau) // 2]
        chosen = (best_m, chosen_g)
    else:
        cands = [(m, g) for m in MS for g in improving[m]]
        if cands:
            chosen = max(cands, key=lambda mg: train[mg]["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_m, c_g = chosen
    print(f"CHOSEN g={c_g:.1f} m={c_m:.1f} -> OOS (config + 2 vicine g), 1 volta")
    print("=" * 78)
    gi = GS.index(c_g)
    neigh = [GS[x] for x in (gi - 1, gi, gi + 1) if 0 <= x < len(GS)]
    oos = {}
    for g in neigh:
        pol = Giveback(g, c_m)
        row = {}
        for a in ("BTC", "ETH"):
            row[a] = {"train": train[(c_m, g)][a],
                      "oos": simulate(sleeves[a], pol, t_lo=OOS_START_MS)}
        oos[g] = row
        print(f"  g={g:.1f} m={c_m:.1f}")
        _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 (g={c_g:.1f} m={c_m:.1f}):")
    bt_tr, et_tr = oos[c_g]["BTC"]["train"], oos[c_g]["ETH"]["train"]
    bt_oo, et_oo = oos[c_g]["BTC"]["oos"], oos[c_g]["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} m={best_m})")
    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()