PythagorasGoal/scripts/analysis/sh01_exit_policies/04_chandelier_trail.py

"""SH01 EXIT policy 04 — chandelier_trail.

Trailing chandelier CAUSALE. Lo state tiene il running peak dei CLOSE da i a
j-1; lo stop per il bar j e':
    long : sl = peak - k * ATR14[j-1]
    short: sl = trough + k * ATR14[j-1]   (specchiato)
Il peak/trough viene aggiornato dentro levels() usando SOLO close[j-1] (dato
gia' chiuso quando il worker fissa il livello per il bar j). ATR14[j-1] e'
causale. Griglia k x mode {intrabar, close}.

ANTI-LOOK-AHEAD: levels(j) usa peak su close[<=j-1] e ATR14[j-1] -> nessun dato
del bar j. open_trade usa solo close[i]/ATR14[i]. OK.

Profilo SH01: hold a orizzonte (momentum), win ~50%, edge nell'asimmetria dei
winner. Sulle fade la famiglia trailing e' stata SCARTATA (taglia i winner che
vanno in drawdown e poi recuperano) -> qui si testa se su SH01 va diversamente,
pronti a un NO.

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

    cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/04_chandelier_trail.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 ChandelierTrail(ExitPolicy):
    def __init__(self, k: float, mode: str = "intrabar"):
        self.k = float(k)
        self.mode = mode
        self.name = f"chandelier_trail k={k:.1f} {mode}"

    def open_trade(self, ctx, i, d):
        # peak/trough inizializzato all'entry (close[i]); atr14[i] noto a close[i].
        entry = ctx["close"][i]
        return {"peak": entry, "trough": entry}

    def levels(self, ctx, i, d, j, st):
        close = ctx["close"]
        atr = ctx["atr14"]
        # aggiorna il running peak/trough con close[j-1] (gia' chiuso). j>=i+1
        # sempre nell'engine, quindi j-1>=i e' definito.
        cprev = close[j - 1]
        if cprev > st["peak"]:
            st["peak"] = cprev
        if cprev < st["trough"]:
            st["trough"] = cprev
        a = atr[j - 1]
        if not (a == a and a > 0):   # nan/0 -> nessuno stop attivo
            return None, self.mode
        if d == 1:
            sl = st["peak"] - self.k * a
        else:
            sl = st["trough"] + self.k * a
        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)},
}

KS = [2.0, 2.5, 3.0, 4.0, 5.0]
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,k)] -> {asset: result}
    train = {}
    for mode in MODES:
        print(f"  --- mode={mode} ---")
        for k in KS:
            pol = ChandelierTrail(k, mode)
            row = {}
            for a in ("BTC", "ETH"):
                row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
            train[(mode, k)] = row
            print(f"  k={k:.1f} ({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 -> [k...]
    for mode in MODES:
        imp = []
        for k in KS:
            bt, et = train[(mode, k)]["BTC"], train[(mode, k)]["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(k)
            print(f"  {mode:<9s} k={k:.1f}  ETH_ok={eth_ok}  BTC_ok={btc_ok}  -> "
                  f"{'IMPROVING' if ok else '-'}")
        improving[mode] = imp
        print(f"  improving cells ({mode}): {imp}")

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

    chosen = None
    if len(best_plateau) >= 3:
        chosen_k = best_plateau[len(best_plateau) // 2]
        chosen = (best_mode, chosen_k)
    else:
        # fallback: miglior ETH sharpe fra tutti gli improving (per diagnosi OOS)
        cands = [(m, k) for m in MODES for k in improving[m]]
        if cands:
            chosen = max(cands, key=lambda mk: train[mk]["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_k = chosen
    print(f"CHOSEN k={c_k:.1f} mode={c_mode} -> OOS (config + 2 vicine k), 1 volta")
    print("=" * 78)
    ci = KS.index(c_k)
    neigh = [KS[x] for x in (ci - 1, ci, ci + 1) if 0 <= x < len(KS)]
    oos = {}
    for k in neigh:
        pol = ChandelierTrail(k, c_mode)
        row = {}
        for a in ("BTC", "ETH"):
            row[a] = {"train": train[(c_mode, k)][a],
                      "oos": simulate(sleeves[a], pol, t_lo=OOS_START_MS)}
        oos[k] = row
        print(f"  k={k:.1f} ({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 (k={c_k:.1f} mode={c_mode}):")
    bt_tr, et_tr = oos[c_k]["BTC"]["train"], oos[c_k]["ETH"]["train"]
    bt_oo, et_oo = oos[c_k]["BTC"]["oos"], oos[c_k]["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} mode={best_mode})")
    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()