PythagorasGoal/scripts/analysis/sh01_exit_policies/09_swing_stop.py

"""SH01 EXIT policy 09 — swing_stop.

Stop STRUTTURALE sullo swing recente, fissato all'ingresso:
    long : sl = min(low[i-N+1 .. i]) - pad * ATR14[i]
    short: sl = max(high[i-N+1 .. i]) + pad * ATR14[i]
Specchiato per d=-1. Il livello e' congelato in open_trade (SOLO dati <= i:
low/high della finestra fino a i incluso, ATR14[i] noto a close[i]). levels()
restituisce quel livello costante per tutta la vita del trade -> nessun dato del
bar j -> anti-look-ahead OK.

Idea: invece di uno stop a distanza fissa (ATR/%), ancora lo stop alla STRUTTURA
del prezzo (minimo/massimo dello swing recente). Un long viene stoppato solo se
rompe il supporto strutturale che lo ha generato; il pad in ATR da' un cuscinetto
sotto il livello per evitare i wick (mode intrabar) o per richiedere conferma sul
close (mode close, stile EXIT-16).

Griglia: N in {6, 12, 24} x pad in {0.0, 0.25, 0.5} x mode {intrabar, close}.

    cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/09_swing_stop.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 SwingStop(ExitPolicy):
    def __init__(self, n: int, pad: float, mode: str):
        self.n = int(n)
        self.pad = float(pad)
        self.mode = str(mode)
        self.name = f"swing n={n} pad={pad:.2f} {mode}"

    def open_trade(self, ctx, i, d):
        lo, hi = ctx["low"], ctx["high"]
        atr = ctx["atr14"][i]
        lo0 = max(0, i - self.n + 1)
        if atr != atr or atr <= 0:           # nan/0 (early bars) -> nessuno stop
            return {"sl": None}
        if d == 1:
            swing = float(lo[lo0:i + 1].min())
            sl = swing - self.pad * atr
        else:
            swing = float(hi[lo0:i + 1].max())
            sl = swing + self.pad * atr
        return {"sl": sl}

    def levels(self, ctx, i, d, j, st):
        return st["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)},
}

NS = [6, 12, 24]
PADS = [0.0, 0.25, 0.5]
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, n, pad) -> {asset: result}
    train = {}
    for mode in MODES:
        print(f"  --- mode={mode} ---")
        for n in NS:
            for pad in PADS:
                pol = SwingStop(n, pad, mode)
                row = {}
                for a in ("BTC", "ETH"):
                    row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
                train[(mode, n, pad)] = row
                print(f"  n={n:<2d} pad={pad:.2f}")
                _row("TRAIN", "BTC", row["BTC"])
                _row("TRAIN", "ETH", row["ETH"])
        print()

    print("=" * 78)
    print("PLATEAU CHECK (train): per ogni cella, ETH(shrp up & dd down & worst up)")
    print("                       & BTC(shrp>=95% & ret>=80% baseline)")
    print("=" * 78)
    improving = []
    grid_imp = {}  # (mode,n,pad) -> bool
    for mode in MODES:
        for n in NS:
            for pad in PADS:
                bt, et = train[(mode, n, pad)]["BTC"], train[(mode, n, pad)]["ETH"]
                ok = _eth_ok(et, b_eth) and _btc_ok(bt, b_btc)
                grid_imp[(mode, n, pad)] = ok
                if ok:
                    improving.append((mode, n, pad))
                print(f"  {mode:<8s} n={n:<2d} pad={pad:.2f}  "
                      f"ETH_ok={_eth_ok(et, b_eth)!s:<5} BTC_ok={_btc_ok(bt, b_btc)!s:<5} "
                      f"-> {'IMPROVING' if ok else '-'}")
    print(f"  improving cells (train): {len(improving)}/{len(train)} -> {improving}")

    # PLATEAU = adiacenza nella griglia N x pad (stesso mode). Adiacenti = vicini
    # nelle liste NS/PADS. Cerco il blocco contiguo piu' grande di celle improving.
    def adjacent_block_size(mode):
        cells = [(NS.index(n), PADS.index(p))
                 for (m, n, p) in improving if m == mode]
        cells_set = set(cells)
        best = []
        for start in cells:
            # BFS sul reticolo 4-connesso
            seen, stack = set(), [start]
            while stack:
                cur = stack.pop()
                if cur in seen:
                    continue
                seen.add(cur)
                ci, cj = cur
                for di, dj in ((1, 0), (-1, 0), (0, 1), (0, -1)):
                    nb = (ci + di, cj + dj)
                    if nb in cells_set and nb not in seen:
                        stack.append(nb)
            if len(seen) > len(best):
                best = list(seen)
        return best

    plateau_cells = []
    plateau_mode = None
    for mode in MODES:
        blk = adjacent_block_size(mode)
        if len(blk) > len(plateau_cells):
            plateau_cells = blk
            plateau_mode = mode
    plateau_ok = len(plateau_cells) >= 3
    if plateau_mode is not None:
        readable = [(plateau_mode, NS[i], PADS[j]) for (i, j) in plateau_cells]
    else:
        readable = []
    print(f"  largest adjacent improving block: {len(plateau_cells)} cells "
          f"mode={plateau_mode} -> {readable} (plateau={'YES' if plateau_ok else 'NO'})")

    # scelta: centro del plateau (miglior ETH sharpe fra le celle del blocco),
    # altrimenti miglior ETH sharpe fra gli improving.
    chosen = None
    if plateau_ok:
        chosen = max(readable, key=lambda c: train[c]["ETH"]["sharpe"])
    elif improving:
        chosen = max(improving, key=lambda c: train[c]["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": readable, "improving": improving,
                "passes": False}

    print(f"CHOSEN {chosen} -> OOS (config + vicine), guardato UNA volta")
    print("=" * 78)
    mode, n, pad = chosen
    # vicine: stesso mode, pad +-1 step e n +-1 step (se esistono e improving o no)
    ni, pi = NS.index(n), PADS.index(pad)
    neigh = set([chosen])
    for di, dj in ((0, 0), (1, 0), (-1, 0), (0, 1), (0, -1)):
        a, b = ni + di, pi + dj
        if 0 <= a < len(NS) and 0 <= b < len(PADS):
            neigh.add((mode, NS[a], PADS[b]))
    oos = {}
    for c in sorted(neigh, key=lambda c: (c[1], c[2])):
        m, nn, pp = c
        pol = SwingStop(nn, pp, m)
        row = {}
        for a in ("BTC", "ETH"):
            row[a] = {"train": train[c][a],
                      "oos": simulate(sleeves[a], pol, t_lo=OOS_START_MS)}
        oos[c] = row
        print(f"  {m} n={nn} pad={pp:.2f}")
        _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 ({chosen}):")
    bt_tr, et_tr = oos[chosen]["BTC"]["train"], oos[chosen]["ETH"]["train"]
    bt_oo, et_oo = oos[chosen]["BTC"]["oos"], oos[chosen]["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
    cond_b = _btc_ok(bt_tr, b_btc) and (bt_oo["sharpe"] >= 0.95 * Bb_o["sharpe"]
                                        and bt_oo["ret"] >= 0.80 * Bb_o["ret"])
    cond_c = et_oo["ret"] >= 0.80 * Be_o["ret"]
    cond_d = plateau_ok

    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} ({len(plateau_cells)} cells)")
    passes = cond_a and cond_b and cond_c and cond_d
    print(f"  PASSES GATE: {passes}")
    print("=" * 78)

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


if __name__ == "__main__":
    main()