PythagorasGoal/scripts/analysis/sh01_exit_policies/03_pct_fixed.py

"""SH01 exit policy 03 — pct_fixed.

SL fisso in PERCENTUALE del prezzo d'ingresso:  sl = entry * (1 - d*p).
Griglia p in {0.01, 0.015, 0.02, 0.03, 0.04, 0.05}, modalita' {intrabar, close}
-> 12 celle. Il livello e' FISSO (deciso a open_trade su close[i]) -> nessun
look-ahead nei bar successivi (i livelli usano solo dati <= i).

Protocollo: grid SOLO sul train; plateau (>=3 celle adiacenti migliorative);
poi OOS una volta per la config scelta + le 2 vicine.

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

import sys

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

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


class PctFixed(ExitPolicy):
    """SL fisso a una frazione p del prezzo d'ingresso."""

    def __init__(self, p: float, mode: str = "intrabar"):
        self.p = p
        self.mode = mode
        self.name = f"pct_fixed p={p:.3f} {mode}"

    def open_trade(self, ctx, i, d):
        entry = ctx["close"][i]
        sl = entry * (1.0 - d * self.p)   # long: sotto; short: sopra
        return {"sl": sl}

    def levels(self, ctx, i, d, j, st):
        return st["sl"], self.mode


# ----------------------------------------------------------------------------- grid

P_GRID = [0.01, 0.015, 0.02, 0.03, 0.04, 0.05]
MODES = ["intrabar", "close"]


def _row(m):
    return (f"ret={m['ret']:>+7.0f}% dd={m['dd']:>4.0f}% shrp={m['sharpe']:>5.2f} "
            f"worst={m['worst']:>+5.1f}% stop={m['stop_rate']:>4.1f}%")


def main():
    sleeves = load_sleeves()

    # baseline (no stop)
    print("=" * 110)
    print("BASELINE (orizzonte puro, no SL) — TRAIN:")
    base = {}
    for a in ASSETS:
        m = simulate(sleeves[a], ExitPolicy(), t_hi=OOS_START_MS)
        base[a] = m
        print(f"  {a}: {_row(m)}")
    print()

    # ---------------- grid TRAIN only
    print("=" * 110)
    print("GRID — TRAIN ONLY (selezione qui):")
    train = {}
    for mode in MODES:
        print(f"\n  mode={mode}")
        for p in P_GRID:
            pol = PctFixed(p, mode)
            row = {}
            for a in ASSETS:
                m = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
                row[a] = m
            train[(mode, p)] = row
            print(f"    p={p:.3f} | BTC {_row(row['BTC'])}")
            print(f"               | ETH {_row(row['ETH'])}")

    # improvement flags vs baseline on TRAIN: ETH gate (sharpe up, dd down, worst less neg)
    # + BTC not degraded (sharpe>=0.95x, ret>=0.80x)
    print("\n" + "=" * 110)
    print("TRAIN improvement check (cell = migliorativa se ETH sharpe^ dd v worst^ AND BTC sharpe>=95% ret>=80%):")
    bE, bB = base["ETH"], base["BTC"]
    improved = {}
    for mode in MODES:
        flags = []
        for p in P_GRID:
            r = train[(mode, p)]
            eth, btc = r["ETH"], r["BTC"]
            eth_ok = (eth["sharpe"] > bE["sharpe"] and eth["dd"] < bE["dd"]
                      and eth["worst"] > bE["worst"])
            btc_ok = (btc["sharpe"] >= 0.95 * bB["sharpe"]
                      and btc["ret"] >= 0.80 * bB["ret"])
            cell = eth_ok and btc_ok
            improved[(mode, p)] = cell
            flags.append("Y" if cell else (".|E" if not eth_ok else ".|B"))
        print(f"  mode={mode:<9s} " + "  ".join(f"p={p:.3f}:{f}" for p, f in zip(P_GRID, flags)))

    # plateau detection: >=3 adjacent p's (same mode) all improved
    print("\nPLATEAU (>=3 p adiacenti migliorativi nella stessa modalita'):")
    plateau_cells = []
    for mode in MODES:
        run = []
        runs = []
        for p in P_GRID:
            if improved[(mode, p)]:
                run.append(p)
            else:
                if len(run) >= 1:
                    runs.append(run)
                run = []
        if run:
            runs.append(run)
        for run in runs:
            mark = " <-- PLATEAU" if len(run) >= 3 else ""
            print(f"  mode={mode}: run {run} (len {len(run)}){mark}")
            if len(run) >= 3:
                plateau_cells.extend((mode, p) for p in run)

    if not plateau_cells:
        print("\nNESSUN PLATEAU sul train -> famiglia NON passa. OOS solo informativo.")
    else:
        print(f"\nplateau cells: {plateau_cells}")

    # ---------------- pick best cell on TRAIN within plateau (or best overall if no plateau)
    def score(cell):
        r = train[cell]
        # ETH train e' il banco di prova (baseline negativo) -> max ETH sharpe,
        # tie-break ETH dd minore, poi BTC sharpe.
        return (r["ETH"]["sharpe"], -r["ETH"]["dd"], r["BTC"]["sharpe"])

    pool = plateau_cells if plateau_cells else list(train.keys())
    best = max(pool, key=score)
    print(f"\nCHOSEN (train): mode={best[0]} p={best[1]:.3f}")

    # neighbors (same mode, adjacent p)
    mode_b, p_b = best
    idx = P_GRID.index(p_b)
    neigh = [(mode_b, P_GRID[k]) for k in (idx - 1, idx, idx + 1) if 0 <= k < len(P_GRID)]

    # ---------------- OOS verdict (chosen + 2 neighbors) — looked at ONCE
    print("\n" + "=" * 110)
    print("OOS VERDICT (config scelta + 2 vicine) — guardato UNA volta:")
    print("\nBaseline OOS:")
    base_oos = {}
    for a in ASSETS:
        m = simulate(sleeves[a], ExitPolicy(), t_lo=OOS_START_MS)
        base_oos[a] = m
        print(f"  {a}: {_row(m)}")

    chosen_oos = None
    for cell in neigh:
        pol = PctFixed(cell[1], cell[0])
        tag = "  <== CHOSEN" if cell == best else ""
        print(f"\n  mode={cell[0]} p={cell[1]:.3f}{tag}")
        res = {}
        for a in ASSETS:
            tr = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
            oo = simulate(sleeves[a], pol, t_lo=OOS_START_MS)
            res[a] = {"train": tr, "oos": oo}
            print(f"    {a} TRAIN {_row(tr)}")
            print(f"    {a} OOS   {_row(oo)}")
        if cell == best:
            chosen_oos = res

    # ---------------- gate evaluation on chosen
    print("\n" + "=" * 110)
    print("GATE (tutte e 4, train E oos):")
    r = chosen_oos
    bE_o, bB_o = base_oos["ETH"], base_oos["BTC"]

    def g(label, cond):
        print(f"  [{'PASS' if cond else 'FAIL'}] {label}")
        return cond

    # a) ETH: sharpe^ dd v worst^ su train E oos
    a_tr = (r["ETH"]["train"]["sharpe"] > bE["sharpe"]
            and r["ETH"]["train"]["dd"] < bE["dd"]
            and r["ETH"]["train"]["worst"] > bE["worst"])
    a_oo = (r["ETH"]["oos"]["sharpe"] > bE_o["sharpe"]
            and r["ETH"]["oos"]["dd"] < bE_o["dd"]
            and r["ETH"]["oos"]["worst"] > bE_o["worst"])
    A = g("a) ETH sharpe^ dd v worst^ (train E oos)", a_tr and a_oo)
    # b) BTC sharpe>=95% ret>=80% baseline (train E oos)
    b_tr = (r["BTC"]["train"]["sharpe"] >= 0.95 * bB["sharpe"]
            and r["BTC"]["train"]["ret"] >= 0.80 * bB["ret"])
    b_oo = (r["BTC"]["oos"]["sharpe"] >= 0.95 * bB_o["sharpe"]
            and r["BTC"]["oos"]["ret"] >= 0.80 * bB_o["ret"])
    B = g("b) BTC sharpe>=95% ret>=80% (train E oos)", b_tr and b_oo)
    # c) ret ETH oos >= 80% baseline
    C = g("c) ret ETH oos >= 80% baseline", r["ETH"]["oos"]["ret"] >= 0.80 * bE_o["ret"])
    # d) plateau
    D = g("d) plateau confermato", bool(plateau_cells) and best in plateau_cells)

    passes = A and B and C and D
    print(f"\n  ==> GATE {'PASS' if passes else 'FAIL'}")
    return passes


if __name__ == "__main__":
    main()