config(PORT06): cap SHAPE 0.0588 — SH01 resta senza SL (ricerca multi-agente: 11 famiglie di stop, 0 sopravvissute)
Crash ETH 2026-06-05: SH01 ETH −15.6% su un trade (exit solo a orizzonte, nessuna protezione). Ricerca con harness dedicato sh01_exit_lab (cache walk-forward, engine fill gap-aware worse(livello,open), parity esatta con explore_lab, train<=2023-11-01): ATR intrabar/close-confirm, %, chandelier, breakeven, giveback, loser-timestop, disaster-cap close+intrabar, swing, vol-regime — NESSUNA passa il gate (ogni stop stretto rompe BTC, ogni stop largo non tocca la coda ETH; nei crash il fill e' al gap). Mitigazione: peso famiglia SHAPE 11.8%->5.9% in PORT06 (FULL 6.47->6.43 DD 4.10->3.96, OOS 8.82->8.58 DD 1.30->1.36) — la prossima coda impatta il conto per meta'. Regression-lock test aggiornato. Diario: docs/diary/2026-06-05-sh01-sl-research.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,252 @@
|
||||
"""SH01 EXIT LAB — harness onesto e CONDIVISO per la ricerca di STOP-LOSS su SH01.
|
||||
|
||||
SH01 (shape-ML, logit walk-forward W24 H12 th0.58) NON ha TP/SL: esce SOLO a
|
||||
orizzonte H=12 barre. Live (2026-06-05) si è preso il crash ETH intero: −15.6%
|
||||
in un trade (long 1727.8 → 1594.35, leva 2x). Domanda di ricerca: esiste uno SL
|
||||
che taglia le code SENZA distruggere l'edge (che vive nell'asimmetria dei
|
||||
winner, win-rate ~50%)?
|
||||
|
||||
CONTRATTO ANTI-LOOK-AHEAD (vincolante, verificato da agenti avversari):
|
||||
- i livelli attivi nel bar j (`levels(..., j)`) possono usare SOLO dati <= j-1
|
||||
(il worker live fissa i livelli al close del bar precedente; il bar j li tocca);
|
||||
- `after_bar(..., j)` decide sul CLOSE del bar j (eseguibile al poll del tick);
|
||||
- indicatori causali: usare l'indice j-1 (es. ctx["atr14"][j-1]).
|
||||
|
||||
FILL GAP-AWARE (lezione exit-lab 2026-06-04 + crash live 2026-06-05): lo stop
|
||||
intrabar NON filla "al livello" se il bar apre già oltre → fill = worse(level,
|
||||
open[j]). Senza questo il backtest ha un bias PRO stop-stretti (54% dei fill
|
||||
era ottimista). Il crash di oggi (feed flat 2h → gap 1655→1600) è il caso reale.
|
||||
|
||||
PROTOCOLLO ANTI-OVERFIT (vincolante, = exit_lab):
|
||||
- TRAIN = storico fino al 2023-11-01, OOS = dopo. SELEZIONE parametri SOLO
|
||||
sul train; OOS guardato una volta per il verdetto.
|
||||
- gate: miglioramento su ENTRAMBI gli asset (BTC e ETH), train E oos, con
|
||||
plateau sulla griglia (non una cella isolata). Metrica primaria: Sharpe e
|
||||
DD; il return non deve crollare (>= ~80% del baseline).
|
||||
- fee 0.10% RT × leva su tutto il notional.
|
||||
|
||||
Baseline = exit a orizzonte puro (max_bars=H, nessun TP/SL): parità ESATTA con
|
||||
`explore_lab.simulate` verificata da `parity_check()`.
|
||||
|
||||
uv run python scripts/analysis/sh01_exit_lab.py # build cache + parity check
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import pickle
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
||||
sys.path.insert(0, str(PROJECT_ROOT))
|
||||
|
||||
LEV, POS, FEE_RT = 3.0, 0.15, 0.001
|
||||
OOS_START_MS = int(pd.Timestamp("2023-11-01", tz="UTC").value // 1e6)
|
||||
ASSETS = ("BTC", "ETH")
|
||||
CACHE = PROJECT_ROOT / "data" / "cache" / "sh01_exit_lab.pkl"
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- cache
|
||||
|
||||
def build_cache() -> dict:
|
||||
"""Walk-forward SH01 (lento, ~minuti) → entries cache su disco."""
|
||||
from scripts.analysis.explore_lab import get_df # noqa: E402
|
||||
from scripts.analysis.shape_ml_research import ml_wf_entries, atr # noqa: E402
|
||||
from scripts.strategies.SH01_shape_ml import CONFIG # noqa: E402
|
||||
|
||||
out = {}
|
||||
for a in ASSETS:
|
||||
df = get_df(a, "1h")
|
||||
ents = ml_wf_entries(df, **CONFIG)
|
||||
out[a] = {
|
||||
"entries": [(int(e["i"]), int(e["d"]), int(e["max_bars"])) for e in ents],
|
||||
"open": df["open"].values.astype(float),
|
||||
"high": df["high"].values.astype(float),
|
||||
"low": df["low"].values.astype(float),
|
||||
"close": df["close"].values.astype(float),
|
||||
"ts_ms": df["timestamp"].values.astype("int64"),
|
||||
"atr14": atr(df, 14),
|
||||
}
|
||||
print(f" {a}: {len(ents)} entries, {len(df)} bars", flush=True)
|
||||
CACHE.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(CACHE, "wb") as f:
|
||||
pickle.dump(out, f)
|
||||
return out
|
||||
|
||||
|
||||
def load_sleeves(refresh: bool = False) -> dict:
|
||||
"""{asset: ctx}. ctx = {entries, open, high, low, close, ts_ms, atr14}."""
|
||||
if CACHE.exists() and not refresh:
|
||||
with open(CACHE, "rb") as f:
|
||||
return pickle.load(f)
|
||||
return build_cache()
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- policy
|
||||
|
||||
class ExitPolicy:
|
||||
"""Contratto per le policy di stop su SH01 (solo SL/uscite anticipate: il
|
||||
TP non esiste e l'exit a orizzonte max_bars resta SEMPRE il bound).
|
||||
|
||||
open_trade(ctx, i, d) -> state : livelli iniziali, SOLO dati <= i
|
||||
levels(ctx, i, d, j, st) -> (sl, mode) attivi nel bar j, SOLO dati <= j-1.
|
||||
sl=None → nessuno stop nel bar. mode: "intrabar" (tocco high/low, fill
|
||||
gap-aware worse(sl, open[j])) o "close" (stop solo se il CLOSE sfonda
|
||||
sl, uscita al close — stile EXIT-16).
|
||||
after_bar(ctx, i, d, j, st) -> bool : uscita discrezionale al CLOSE del bar
|
||||
j (dati <= j). Per giveback/time-stop/regime.
|
||||
Lo state è un dict mutabile per-trade (trailing ecc.)."""
|
||||
|
||||
name = "base"
|
||||
|
||||
def open_trade(self, ctx: dict, i: int, d: int) -> dict:
|
||||
return {}
|
||||
|
||||
def levels(self, ctx: dict, i: int, d: int, j: int, st: dict):
|
||||
return None, "intrabar"
|
||||
|
||||
def after_bar(self, ctx: dict, i: int, d: int, j: int, st: dict) -> bool:
|
||||
return False
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- engine
|
||||
|
||||
def simulate(ctx: dict, policy: ExitPolicy, fee_rt: float = FEE_RT,
|
||||
lev: float = LEV, pos: float = POS,
|
||||
t_lo: int | None = None, t_hi: int | None = None,
|
||||
gap_fill: bool = True, lag_close_exit: bool = False) -> dict:
|
||||
"""Engine intrabar con policy di stop. Entries non sovrapposte (come
|
||||
explore_lab.simulate). t_lo/t_hi: filtro ms-epoch sull'ENTRY (train/oos).
|
||||
gap_fill: fill stop intrabar a worse(sl, open[j]) — tenere True.
|
||||
lag_close_exit: stress — le uscite "al close" fillano al close del bar
|
||||
successivo (poll in ritardo)."""
|
||||
o, h, l, c = ctx["open"], ctx["high"], ctx["low"], ctx["close"]
|
||||
ts = ctx["ts_ms"]
|
||||
n = len(c)
|
||||
cap = peak = 1000.0
|
||||
max_dd = 0.0
|
||||
fee = fee_rt * lev
|
||||
trades = wins = stops = 0
|
||||
bars_in = 0
|
||||
last_exit = -1
|
||||
yearly: dict[int, float] = {}
|
||||
rets: list[float] = []
|
||||
trade_rows: list[dict] = []
|
||||
|
||||
for (i, d, mb) in ctx["entries"]:
|
||||
if i <= last_exit or i + 1 >= n:
|
||||
continue
|
||||
if t_lo is not None and ts[i] < t_lo:
|
||||
continue
|
||||
if t_hi is not None and ts[i] >= t_hi:
|
||||
continue
|
||||
entry = c[i]
|
||||
st = policy.open_trade(ctx, i, d)
|
||||
exit_p, j, reason = c[min(i + mb, n - 1)], min(i + mb, n - 1), "time"
|
||||
for k in range(1, mb + 1):
|
||||
j = i + k
|
||||
if j >= n:
|
||||
j, exit_p, reason = n - 1, c[n - 1], "eod"
|
||||
break
|
||||
sl, mode = policy.levels(ctx, i, d, j, st)
|
||||
if sl is not None and mode == "intrabar":
|
||||
hit = (l[j] <= sl) if d == 1 else (h[j] >= sl)
|
||||
if hit:
|
||||
if gap_fill:
|
||||
exit_p = min(sl, o[j]) if d == 1 else max(sl, o[j])
|
||||
else:
|
||||
exit_p = sl
|
||||
reason = "stop"
|
||||
break
|
||||
if sl is not None and mode == "close":
|
||||
brk = (c[j] < sl) if d == 1 else (c[j] > sl)
|
||||
if brk:
|
||||
jj = min(j + 1, n - 1) if lag_close_exit else j
|
||||
exit_p, j, reason = c[jj], jj, "stop"
|
||||
break
|
||||
if policy.after_bar(ctx, i, d, j, st):
|
||||
jj = min(j + 1, n - 1) if lag_close_exit else j
|
||||
exit_p, j, reason = c[jj], jj, "policy"
|
||||
break
|
||||
if k == mb:
|
||||
exit_p, reason = c[j], "time"
|
||||
ret = (exit_p - entry) / entry * d * lev - fee
|
||||
cb = cap
|
||||
cap = max(cb + cb * pos * ret, 10.0)
|
||||
peak = max(peak, cap)
|
||||
max_dd = max(max_dd, (peak - cap) / peak)
|
||||
trades += 1
|
||||
wins += ret > 0
|
||||
stops += reason == "stop"
|
||||
bars_in += (j - i)
|
||||
last_exit = j
|
||||
rets.append(ret * pos)
|
||||
yr = pd.Timestamp(ts[i], unit="ms", tz="UTC").year
|
||||
yearly[yr] = yearly.get(yr, 0.0) + ret * 100
|
||||
trade_rows.append({"i": i, "j": j, "d": d, "ret": ret, "reason": reason})
|
||||
|
||||
sharpe = (float(np.mean(rets) / np.std(rets) * np.sqrt(len(rets)))
|
||||
if len(rets) > 1 and np.std(rets) > 0 else 0.0)
|
||||
return {
|
||||
"trades": trades,
|
||||
"win": wins / trades * 100 if trades else 0.0,
|
||||
"stop_rate": stops / trades * 100 if trades else 0.0,
|
||||
"ret": (cap / 1000 - 1) * 100,
|
||||
"dd": max_dd * 100,
|
||||
"sharpe": sharpe,
|
||||
"worst": min(rets) * 100 if rets else 0.0, # peggior trade, % equity (ret*pos)
|
||||
"yearly": yearly,
|
||||
"_trades": trade_rows,
|
||||
}
|
||||
|
||||
|
||||
def evaluate(policy: ExitPolicy, sleeves: dict | None = None, **kw) -> dict:
|
||||
"""train (fino al 2023-11-01) e oos (dopo) per BTC e ETH. Stampa sintesi."""
|
||||
sleeves = sleeves or load_sleeves()
|
||||
out = {}
|
||||
for a in ASSETS:
|
||||
ctx = sleeves[a]
|
||||
tr = simulate(ctx, policy, t_hi=OOS_START_MS, **kw)
|
||||
oo = simulate(ctx, policy, t_lo=OOS_START_MS, **kw)
|
||||
out[a] = {"train": tr, "oos": oo}
|
||||
print(f" {policy.name:<28s} {a}: "
|
||||
f"TRAIN ret={tr['ret']:>+7.0f}% dd={tr['dd']:>4.0f}% shrp={tr['sharpe']:>5.2f} "
|
||||
f"worst={tr['worst']:>+5.1f}% stop={tr['stop_rate']:>4.1f}% | "
|
||||
f"OOS ret={oo['ret']:>+6.0f}% dd={oo['dd']:>4.0f}% shrp={oo['sharpe']:>5.2f} "
|
||||
f"worst={oo['worst']:>+5.1f}%", flush=True)
|
||||
return out
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- parity
|
||||
|
||||
def parity_check() -> bool:
|
||||
"""Baseline (nessuno stop) == explore_lab.simulate sugli stessi entries."""
|
||||
from scripts.analysis.explore_lab import get_df, simulate as ref_sim # noqa: E402
|
||||
|
||||
sleeves = load_sleeves()
|
||||
ok = True
|
||||
for a in ASSETS:
|
||||
ctx = sleeves[a]
|
||||
mine = simulate(ctx, ExitPolicy())
|
||||
df = get_df(a, "1h")
|
||||
ents = [{"i": i, "d": d, "max_bars": mb, "tp": None, "sl": None}
|
||||
for (i, d, mb) in ctx["entries"]]
|
||||
ref = ref_sim(ents, df)
|
||||
same = (abs(mine["ret"] - ref["ret"]) < 1e-6 and mine["trades"] == ref["trades"]
|
||||
and abs(mine["dd"] - ref["dd"]) < 1e-6)
|
||||
ok &= same
|
||||
print(f" parity {a}: mine ret={mine['ret']:+.2f}% trades={mine['trades']} "
|
||||
f"| ref ret={ref['ret']:+.2f}% trades={ref['trades']} -> {'OK' if same else 'MISMATCH'}")
|
||||
return ok
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("build cache (walk-forward SH01, puo' richiedere minuti)...")
|
||||
load_sleeves(refresh="--refresh" in sys.argv)
|
||||
print("parity check baseline vs explore_lab.simulate:")
|
||||
ok = parity_check()
|
||||
print("baseline train/oos:")
|
||||
evaluate(ExitPolicy())
|
||||
sys.exit(0 if ok else 1)
|
||||
@@ -0,0 +1,205 @@
|
||||
"""SH01 EXIT policy 01 — atr_fixed_intrabar.
|
||||
|
||||
SL fisso ad ATR, INTRABAR. In open_trade fissiamo il livello una volta sola:
|
||||
sl = entry - d * k * ATR14[i] (entry = close[i], ATR14[i] noto a close[i])
|
||||
levels() restituisce (sl, "intrabar") costante per tutta la vita del trade.
|
||||
Il fill è gap-aware (worse(sl, open[j])) nell'engine — realistico sui crash a
|
||||
gap (es. 2026-06-05: feed flat 2h -> gap ETH 1655->1600).
|
||||
|
||||
ANTI-LOOK-AHEAD: il livello usa SOLO dati <= i (ATR14[i], close[i]); levels usa
|
||||
quel valore congelato (nessun dato del bar j). OK.
|
||||
|
||||
PROTOCOLLO: grid su k SOLO sul train (t_hi=OOS_START_MS). Plateau >=3 celle
|
||||
adiacenti migliorative. Poi OOS una volta sulla config scelta + 2 vicine.
|
||||
|
||||
cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/01_atr_fixed_intrabar.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 AtrFixedIntrabar(ExitPolicy):
|
||||
def __init__(self, k: float):
|
||||
self.k = float(k)
|
||||
self.name = f"atr_fixed_intrabar k={k:.1f}"
|
||||
|
||||
def open_trade(self, ctx, i, d):
|
||||
atr = ctx["atr14"][i]
|
||||
entry = ctx["close"][i]
|
||||
# se atr nan/0 (early bars) -> nessuno stop attivo
|
||||
sl = entry - d * self.k * atr if atr == atr and atr > 0 else None
|
||||
return {"sl": sl}
|
||||
|
||||
def levels(self, ctx, i, d, j, st):
|
||||
return st["sl"], "intrabar"
|
||||
|
||||
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)},
|
||||
}
|
||||
|
||||
|
||||
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 main():
|
||||
sleeves = load_sleeves()
|
||||
KS = [1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0]
|
||||
|
||||
print("=" * 78)
|
||||
print("TRAIN GRID (selezione SOLO sul train, t_hi=OOS_START)")
|
||||
print("=" * 78)
|
||||
# baseline train
|
||||
print(" baseline (orizzonte puro):")
|
||||
base = evaluate(ExitPolicy(), sleeves=sleeves)
|
||||
print()
|
||||
|
||||
train = {} # k -> {asset: result}
|
||||
for k in KS:
|
||||
pol = AtrFixedIntrabar(k)
|
||||
row = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
row[a] = simulate(ctx, pol, t_hi=OOS_START_MS)
|
||||
train[k] = row
|
||||
print(f" k={k:.1f}")
|
||||
_row("TRAIN", "BTC", row["BTC"])
|
||||
_row("TRAIN", "ETH", row["ETH"])
|
||||
|
||||
print()
|
||||
print("=" * 78)
|
||||
print("PLATEAU CHECK (train): per ogni k, ETH sharpe up & dd down & worst up,")
|
||||
print(" BTC sharpe>=95% & ret>=80% baseline")
|
||||
print("=" * 78)
|
||||
b_btc, b_eth = BASELINE["BTC"]["train"], BASELINE["ETH"]["train"]
|
||||
improving = []
|
||||
for k in KS:
|
||||
bt, et = train[k]["BTC"], train[k]["ETH"]
|
||||
eth_ok = (et["sharpe"] > b_eth["sharpe"] and et["dd"] < b_eth["dd"]
|
||||
and et["worst"] > b_eth["worst"])
|
||||
btc_ok = (bt["sharpe"] >= 0.95 * b_btc["sharpe"]
|
||||
and bt["ret"] >= 0.80 * b_btc["ret"])
|
||||
ok = eth_ok and btc_ok
|
||||
if ok:
|
||||
improving.append(k)
|
||||
print(f" k={k:.1f} ETH_ok={eth_ok} BTC_ok={btc_ok} -> "
|
||||
f"{'IMPROVING' if ok else '-'}")
|
||||
print(f" improving cells (train): {improving}")
|
||||
|
||||
# plateau = >=3 k adiacenti improving
|
||||
plateau = []
|
||||
for idx in range(len(KS)):
|
||||
run = []
|
||||
for j in range(idx, len(KS)):
|
||||
if KS[j] in improving:
|
||||
run.append(KS[j])
|
||||
else:
|
||||
break
|
||||
if len(run) >= 3 and len(run) > len(plateau):
|
||||
plateau = run
|
||||
print(f" longest adjacent improving run: {plateau} "
|
||||
f"(plateau={'YES' if len(plateau) >= 3 else 'NO'})")
|
||||
|
||||
# scelgo centro del plateau (o miglior ETH sharpe fra gli improving)
|
||||
chosen = None
|
||||
if len(plateau) >= 3:
|
||||
chosen = plateau[len(plateau) // 2]
|
||||
elif improving:
|
||||
chosen = max(improving, key=lambda k: train[k]["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": plateau, "improving": improving,
|
||||
"train": train, "oos": None}
|
||||
|
||||
print(f"CHOSEN k={chosen:.1f} -> OOS (config + 2 vicine), guardato UNA volta")
|
||||
print("=" * 78)
|
||||
ci = KS.index(chosen)
|
||||
neigh = [KS[x] for x in (ci - 1, ci, ci + 1) if 0 <= x < len(KS)]
|
||||
oos = {}
|
||||
for k in neigh:
|
||||
pol = AtrFixedIntrabar(k)
|
||||
row = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
row[a] = {
|
||||
"train": train[k][a],
|
||||
"oos": simulate(ctx, pol, t_lo=OOS_START_MS),
|
||||
}
|
||||
oos[k] = row
|
||||
print(f" k={k:.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"])
|
||||
|
||||
# gate finale sulla config scelta
|
||||
print()
|
||||
print("=" * 78)
|
||||
print(f"GATE finale (k={chosen:.1f}):")
|
||||
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) ETH: sharpe up & dd down & worst up, train E oos
|
||||
a_train = (et_tr["sharpe"] > b_eth["sharpe"] and et_tr["dd"] < b_eth["dd"]
|
||||
and et_tr["worst"] > b_eth["worst"])
|
||||
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) BTC sharpe>=95% & ret>=80% baseline (train e oos)
|
||||
b_tr = (bt_tr["sharpe"] >= 0.95 * b_btc["sharpe"]
|
||||
and bt_tr["ret"] >= 0.80 * b_btc["ret"])
|
||||
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
|
||||
# c) ret ETH oos >= 80% baseline
|
||||
cond_c = et_oo["ret"] >= 0.80 * Be_o["ret"]
|
||||
# d) plateau
|
||||
cond_d = len(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} ({plateau})")
|
||||
passes = cond_a and cond_b and cond_c and cond_d
|
||||
print(f" PASSES GATE: {passes}")
|
||||
print("=" * 78)
|
||||
|
||||
return {"chosen": chosen, "plateau": plateau, "improving": improving,
|
||||
"passes": passes, "oos": oos, "conds": (cond_a, cond_b, cond_c, cond_d)}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,94 @@
|
||||
"""SH01 EXIT POLICY 02 — ATR-fixed stop, CLOSE-CONFIRM (stile EXIT-16 delle fade).
|
||||
|
||||
Stesso livello di stop fisso della policy 01 (intrabar):
|
||||
sl = entry - d * k * ATR14[i] (fissato all'ingresso, dati <= i)
|
||||
ma `levels` ritorna mode="close" → lo stop scatta SOLO se il CLOSE del bar j
|
||||
sfonda il livello, con uscita al close (immune ai wick). E' il trasferimento a
|
||||
SH01 della lezione EXIT-16 sulle fade: l'overshoot che buca lo stop e rientra e'
|
||||
un falso negativo; aspettare la conferma del CLOSE evita di farsi stoppare dai
|
||||
wick di un crash che poi rimbalza dentro l'orizzonte.
|
||||
|
||||
Griglia k in {1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0}.
|
||||
|
||||
ANTI-LOOK-AHEAD: sl usa SOLO atr14[i] e c[i] (dati <= i); mode="close" decide
|
||||
sul close del bar j (dati <= j, eseguibile al poll). Nessun indicatore al bar j.
|
||||
|
||||
cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/02_atr_fixed_close_confirm.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/opt/docker/PythagorasGoal")
|
||||
|
||||
from scripts.analysis.sh01_exit_lab import ( # noqa: E402
|
||||
ExitPolicy, evaluate, load_sleeves, simulate, OOS_START_MS,
|
||||
)
|
||||
|
||||
|
||||
class AtrFixedCloseConfirm(ExitPolicy):
|
||||
def __init__(self, k: float):
|
||||
self.k = float(k)
|
||||
self.name = f"atr_fixed_close k={k:g}"
|
||||
|
||||
def open_trade(self, ctx: dict, i: int, d: int) -> dict:
|
||||
atr = ctx["atr14"][i]
|
||||
entry = ctx["close"][i]
|
||||
sl = entry - d * self.k * atr
|
||||
return {"sl": float(sl)}
|
||||
|
||||
def levels(self, ctx: dict, i: int, d: int, j: int, st: dict):
|
||||
return st["sl"], "close"
|
||||
|
||||
|
||||
GRID = [1.0, 1.5, 2.0, 2.5, 3.0, 4.0, 5.0]
|
||||
|
||||
|
||||
def _fmt(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}% n={m['trades']}")
|
||||
|
||||
|
||||
def main():
|
||||
sleeves = load_sleeves()
|
||||
|
||||
# baseline (orizzonte puro) per riferimento
|
||||
base = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
base[a] = {
|
||||
"train": simulate(ctx, ExitPolicy(), t_hi=OOS_START_MS),
|
||||
"oos": simulate(ctx, ExitPolicy(), t_lo=OOS_START_MS),
|
||||
}
|
||||
|
||||
print("=" * 110)
|
||||
print("BASELINE (exit orizzonte puro):")
|
||||
for a in ("BTC", "ETH"):
|
||||
print(f" {a} TRAIN {_fmt(base[a]['train'])}")
|
||||
print(f" {a} OOS {_fmt(base[a]['oos'])}")
|
||||
|
||||
print("=" * 110)
|
||||
print("GRID — TRAIN ONLY (selezione parametri):")
|
||||
train_res = {}
|
||||
for k in GRID:
|
||||
pol = AtrFixedCloseConfirm(k)
|
||||
row = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
|
||||
train_res[k] = row
|
||||
print(f" k={k:>3g} | BTC {_fmt(row['BTC'])}")
|
||||
print(f" | ETH {_fmt(row['ETH'])}")
|
||||
|
||||
# verdetto OOS sulla config scelta + vicine (guardato una volta sola)
|
||||
print("=" * 110)
|
||||
print("OOS (verdetto, config scelta + vicine):")
|
||||
for k in GRID:
|
||||
pol = AtrFixedCloseConfirm(k)
|
||||
print(f" k={k:>3g} | BTC TRAIN {_fmt(train_res[k]['BTC'])}")
|
||||
for a in ("BTC", "ETH"):
|
||||
oo = simulate(sleeves[a], pol, t_lo=OOS_START_MS)
|
||||
print(f" | {a} OOS {_fmt(oo)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,203 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,238 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,267 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,260 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,107 @@
|
||||
"""SH01 EXIT POLICY 07 — LOSER time-stop condizionale (after_bar).
|
||||
|
||||
Idea: SH01 esce a orizzonte fisso H=12. Se a un check-point intermedio j=i+m il
|
||||
trade e' GIA' in perdita oltre una soglia, esci subito al close invece di tenere
|
||||
fino a H. L'ipotesi (lezione exit-lab fade: tagliare i loser presto rischia di
|
||||
tagliare anche i winner in drawdown temporaneo): su un orizzonte FISSO di 12 bar
|
||||
forse un loser conclamato a meta' corsa raramente recupera, mentre i winner del
|
||||
modello partono subito (asimmetria). Il time-stop e' UNA volta sola (al bar m),
|
||||
non un trailing: non insegue il prezzo, condiziona solo l'uscita a un istante.
|
||||
|
||||
Regola (after_bar):
|
||||
al bar j == i + m: se (close[j]-entry)/entry * d < -x * ATR14[i] / entry
|
||||
esci al close del bar j.
|
||||
Equivalente: directional_move[j] < -x*ATR14[i]. x=0.0 => esci se in QUALSIASI
|
||||
perdita direzionale al bar m.
|
||||
|
||||
Griglia m in {2, 3, 4, 6} x x in {0.0, 0.5, 1.0}.
|
||||
|
||||
ANTI-LOOK-AHEAD: ATR14[i] e entry=close[i] fissati all'ingresso (dati <= i);
|
||||
after_bar decide sul close del bar j (dati <= j, eseguibile al poll del tick).
|
||||
Nessun indicatore al bar j stesso.
|
||||
|
||||
cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/07_loser_timestop.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/opt/docker/PythagorasGoal")
|
||||
|
||||
from scripts.analysis.sh01_exit_lab import ( # noqa: E402
|
||||
ExitPolicy, evaluate, load_sleeves, simulate, OOS_START_MS,
|
||||
)
|
||||
|
||||
|
||||
class LoserTimestop(ExitPolicy):
|
||||
def __init__(self, m: int, x: float):
|
||||
self.m = int(m)
|
||||
self.x = float(x)
|
||||
self.name = f"loser_timestop m={m} x={x:g}"
|
||||
|
||||
def open_trade(self, ctx: dict, i: int, d: int) -> dict:
|
||||
return {
|
||||
"entry": float(ctx["close"][i]),
|
||||
"atr": float(ctx["atr14"][i]),
|
||||
}
|
||||
|
||||
def after_bar(self, ctx: dict, i: int, d: int, j: int, st: dict) -> bool:
|
||||
if j != i + self.m:
|
||||
return False
|
||||
move = (ctx["close"][j] - st["entry"]) * d # directional, in price
|
||||
thresh = -self.x * st["atr"]
|
||||
return move < thresh
|
||||
|
||||
|
||||
GRID_M = [2, 3, 4, 6]
|
||||
GRID_X = [0.0, 0.5, 1.0]
|
||||
|
||||
|
||||
def _fmt(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}% n={m['trades']}")
|
||||
|
||||
|
||||
def main():
|
||||
sleeves = load_sleeves()
|
||||
|
||||
base = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
base[a] = {
|
||||
"train": simulate(ctx, ExitPolicy(), t_hi=OOS_START_MS),
|
||||
"oos": simulate(ctx, ExitPolicy(), t_lo=OOS_START_MS),
|
||||
}
|
||||
|
||||
print("=" * 110)
|
||||
print("BASELINE (exit orizzonte puro):")
|
||||
for a in ("BTC", "ETH"):
|
||||
print(f" {a} TRAIN {_fmt(base[a]['train'])}")
|
||||
print(f" {a} OOS {_fmt(base[a]['oos'])}")
|
||||
|
||||
print("=" * 110)
|
||||
print("GRID — TRAIN ONLY (selezione parametri): m x rows")
|
||||
train_res = {}
|
||||
for mm in GRID_M:
|
||||
for xx in GRID_X:
|
||||
pol = LoserTimestop(mm, xx)
|
||||
row = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
|
||||
train_res[(mm, xx)] = row
|
||||
print(f" m={mm} x={xx:g} | BTC {_fmt(row['BTC'])}")
|
||||
print(f" | ETH {_fmt(row['ETH'])}")
|
||||
|
||||
print("=" * 110)
|
||||
print("OOS (verdetto, intera griglia per ispezione plateau):")
|
||||
for mm in GRID_M:
|
||||
for xx in GRID_X:
|
||||
pol = LoserTimestop(mm, xx)
|
||||
line_b = simulate(sleeves["BTC"], pol, t_lo=OOS_START_MS)
|
||||
line_e = simulate(sleeves["ETH"], pol, t_lo=OOS_START_MS)
|
||||
print(f" m={mm} x={xx:g} | BTC OOS {_fmt(line_b)}")
|
||||
print(f" | ETH OOS {_fmt(line_e)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,159 @@
|
||||
"""SH01 EXIT POLICY 08 — DISASTER-CAP LARGO, close-confirm (minimal intervention).
|
||||
|
||||
Ipotesi: SH01 (exit a orizzonte puro, niente TP/SL) si fa massacrare dalle code
|
||||
rare (crash ETH 2026-06-05 −15.6% in un trade, ETH 2020). Uno stop LARGO a
|
||||
k·ATR14[i] (k grande) dovrebbe toccare SOLO quei pochi trade-disastro, lasciando
|
||||
intatto il resto della distribuzione — e quindi l'edge asimmetrico dei winner.
|
||||
|
||||
sl = entry - d * k * ATR14[i] (fissato all'ingresso, dati <= i)
|
||||
mode = "close" (stop solo se il CLOSE sfonda, stile EXIT-16)
|
||||
|
||||
Griglia LARGA: k in {3.0, 4.0, 5.0, 6.0}. E' il complemento "wide-only" della
|
||||
policy 02 (che spazzava anche stop stretti): qui l'intento e' la NON-interferenza.
|
||||
|
||||
Strumentazione extra (richiesta dal mandato): per ogni k riporto
|
||||
- stop_rate (quanti trade vengono stoppati),
|
||||
- la DISTRIBUZIONE dei trade tagliati: erano tutti loser? quanti winner uccisi?
|
||||
Per ogni trade stoppato confronto il suo ret (post-stop, ⇒ negativo) con il
|
||||
ret che AVREBBE avuto a orizzonte puro (baseline, senza stop) → conto quanti
|
||||
sarebbero finiti winner (stop "dannoso") vs loser (stop "utile").
|
||||
|
||||
ANTI-LOOK-AHEAD: sl usa SOLO atr14[i] e c[i] (dati <= i); mode="close" decide sul
|
||||
close del bar j (dati <= j). Nessun indicatore valutato al bar j.
|
||||
|
||||
cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/08_disaster_wide_close.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/opt/docker/PythagorasGoal")
|
||||
|
||||
import numpy as np # noqa: E402
|
||||
|
||||
from scripts.analysis.sh01_exit_lab import ( # noqa: E402
|
||||
ExitPolicy, load_sleeves, simulate, OOS_START_MS, FEE_RT, LEV, POS,
|
||||
)
|
||||
|
||||
|
||||
class DisasterWideClose(ExitPolicy):
|
||||
def __init__(self, k: float):
|
||||
self.k = float(k)
|
||||
self.name = f"disaster_wide_close k={k:g}"
|
||||
|
||||
def open_trade(self, ctx: dict, i: int, d: int) -> dict:
|
||||
atr = ctx["atr14"][i]
|
||||
entry = ctx["close"][i]
|
||||
sl = entry - d * self.k * atr
|
||||
return {"sl": float(sl)}
|
||||
|
||||
def levels(self, ctx: dict, i: int, d: int, j: int, st: dict):
|
||||
return st["sl"], "close"
|
||||
|
||||
|
||||
GRID = [3.0, 4.0, 5.0, 6.0]
|
||||
|
||||
|
||||
def _fmt(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}% n={m['trades']}")
|
||||
|
||||
|
||||
def _baseline_ret_by_entry(ctx, t_lo=None, t_hi=None):
|
||||
"""Mappa entry-i -> ret a orizzonte puro (baseline, nessuno stop), stesso
|
||||
engine, stesso slice. Serve a classificare i trade stoppati."""
|
||||
base = simulate(ctx, ExitPolicy(), t_lo=t_lo, t_hi=t_hi)
|
||||
return {r["i"]: r["ret"] for r in base["_trades"]}
|
||||
|
||||
|
||||
def _stop_breakdown(ctx, policy, t_lo=None, t_hi=None):
|
||||
"""Esegue la policy e analizza SOLO i trade con reason=='stop'.
|
||||
Ritorna (n_stop, n_winner_killed, n_loser_cut, dettaglio_list)."""
|
||||
res = simulate(ctx, policy, t_lo=t_lo, t_hi=t_hi)
|
||||
base_ret = _baseline_ret_by_entry(ctx, t_lo=t_lo, t_hi=t_hi)
|
||||
killed = cut = 0
|
||||
detail = []
|
||||
for r in res["_trades"]:
|
||||
if r["reason"] != "stop":
|
||||
continue
|
||||
br = base_ret.get(r["i"])
|
||||
would_win = (br is not None and br > 0)
|
||||
killed += would_win
|
||||
cut += (not would_win)
|
||||
detail.append((r["i"], r["d"], r["ret"], br, would_win))
|
||||
return res, len(detail), killed, cut, detail
|
||||
|
||||
|
||||
def main():
|
||||
sleeves = load_sleeves()
|
||||
|
||||
# baseline orizzonte puro
|
||||
base = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
base[a] = {
|
||||
"train": simulate(ctx, ExitPolicy(), t_hi=OOS_START_MS),
|
||||
"oos": simulate(ctx, ExitPolicy(), t_lo=OOS_START_MS),
|
||||
}
|
||||
|
||||
print("=" * 118)
|
||||
print("BASELINE (exit orizzonte puro):")
|
||||
for a in ("BTC", "ETH"):
|
||||
print(f" {a} TRAIN {_fmt(base[a]['train'])}")
|
||||
print(f" {a} OOS {_fmt(base[a]['oos'])}")
|
||||
|
||||
print("=" * 118)
|
||||
print("GRID — TRAIN ONLY (selezione parametri):")
|
||||
train_res = {}
|
||||
for k in GRID:
|
||||
pol = DisasterWideClose(k)
|
||||
row = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
row[a] = simulate(sleeves[a], pol, t_hi=OOS_START_MS)
|
||||
train_res[k] = row
|
||||
print(f" k={k:>3g} | BTC {_fmt(row['BTC'])}")
|
||||
print(f" | ETH {_fmt(row['ETH'])}")
|
||||
|
||||
# ---- breakdown dei trade stoppati (TRAIN), per la domanda "minimal intervention"
|
||||
print("=" * 118)
|
||||
print("STOP BREAKDOWN — TRAIN (chi viene tagliato? winner uccisi vs loser tagliati):")
|
||||
for k in GRID:
|
||||
pol = DisasterWideClose(k)
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
res, ns, killed, cut, detail = _stop_breakdown(ctx, pol, t_hi=OOS_START_MS)
|
||||
print(f" k={k:>3g} {a} TRAIN: stop n={ns:>2d}/{res['trades']} "
|
||||
f"({res['stop_rate']:.1f}%) -> loser_tagliati={cut} winner_UCCISI={killed}")
|
||||
for (i, d, ret, br, ww) in detail:
|
||||
tag = "WINNER-KILLED" if ww else "loser-cut"
|
||||
brs = f"{br*100:>+6.1f}%" if br is not None else " n/a "
|
||||
print(f" i={i:>6d} d={d:>+d} stop_ret={ret*100:>+6.1f}% "
|
||||
f"baseline_ret={brs} [{tag}]")
|
||||
|
||||
# ---- verdetto OOS (config scelta + vicine, guardato una volta)
|
||||
print("=" * 118)
|
||||
print("OOS (verdetto):")
|
||||
for k in GRID:
|
||||
pol = DisasterWideClose(k)
|
||||
print(f" k={k:>3g} | BTC TRAIN {_fmt(train_res[k]['BTC'])}")
|
||||
for a in ("BTC", "ETH"):
|
||||
oo = simulate(sleeves[a], pol, t_lo=OOS_START_MS)
|
||||
print(f" | {a} OOS {_fmt(oo)}")
|
||||
print("=" * 118)
|
||||
print("STOP BREAKDOWN — OOS:")
|
||||
for k in GRID:
|
||||
pol = DisasterWideClose(k)
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
res, ns, killed, cut, detail = _stop_breakdown(ctx, pol, t_lo=OOS_START_MS)
|
||||
print(f" k={k:>3g} {a} OOS : stop n={ns:>2d}/{res['trades']} "
|
||||
f"({res['stop_rate']:.1f}%) -> loser_tagliati={cut} winner_UCCISI={killed}")
|
||||
for (i, d, ret, br, ww) in detail:
|
||||
tag = "WINNER-KILLED" if ww else "loser-cut"
|
||||
brs = f"{br*100:>+6.1f}%" if br is not None else " n/a "
|
||||
print(f" i={i:>6d} d={d:>+d} stop_ret={ret*100:>+6.1f}% "
|
||||
f"baseline_ret={brs} [{tag}]")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,256 @@
|
||||
"""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()
|
||||
@@ -0,0 +1,157 @@
|
||||
"""SH01 EXIT POLICY 10 — vol_regime_stop.
|
||||
|
||||
Stop CONDIZIONALE al regime di volatilita': lo SL esiste solo quando la vol
|
||||
sta esplodendo. Razionale: il danno (2020 ETH, crash live 2026-06-05) avviene
|
||||
in vol-expansion; quando la vol e' normale lo SL taglierebbe winner in
|
||||
drawdown temporaneo (l'edge SH01 e' nell'asimmetria, win ~50%).
|
||||
|
||||
Regime causale: vr[j] = ATR14[j] / SMA100(ATR14)[j]. Nel bar j si guarda
|
||||
vr[j-1] (dati <= j-1). Se vr[j-1] > r -> SL = entry - d*k1*ATR14[i]
|
||||
(ATR all'entry = dati <= i). Altrimenti nessuno stop.
|
||||
|
||||
r in {1.2, 1.5}
|
||||
k1 in {1.5, 2.0, 3.0}
|
||||
mode in {intrabar, close}
|
||||
|
||||
ANTI-LOOK-AHEAD:
|
||||
- vr e' un array precomputato module-level (SMA100 causale, no centering).
|
||||
- levels(j) usa vr[j-1] e atr14[i] (entry), entrambi <= j-1.
|
||||
- mode "close": stop solo se il CLOSE sfonda (stile EXIT-16).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, "/opt/docker/PythagorasGoal")
|
||||
|
||||
import numpy as np
|
||||
|
||||
from scripts.analysis.sh01_exit_lab import (
|
||||
ExitPolicy, evaluate, load_sleeves, simulate, OOS_START_MS,
|
||||
)
|
||||
|
||||
_VR_CACHE: dict[int, np.ndarray] = {}
|
||||
|
||||
|
||||
def _vol_ratio(atr14: np.ndarray, win: int = 100) -> np.ndarray:
|
||||
"""vr[j] = atr14[j] / SMA(atr14, win)[j], causale. NaN dove non definito."""
|
||||
key = id(atr14)
|
||||
if key in _VR_CACHE:
|
||||
return _VR_CACHE[key]
|
||||
a = np.asarray(atr14, dtype=float)
|
||||
n = len(a)
|
||||
sma = np.full(n, np.nan)
|
||||
# rolling mean causale (include il bar corrente j: e' OK perche' in levels
|
||||
# consumiamo vr[j-1], cioe' dati fino a j-1).
|
||||
csum = np.nancumsum(np.where(np.isnan(a), 0.0, a))
|
||||
cnt = np.cumsum(~np.isnan(a))
|
||||
for j in range(n):
|
||||
lo = j - win + 1
|
||||
if lo < 0:
|
||||
continue
|
||||
s = csum[j] - (csum[lo - 1] if lo > 0 else 0.0)
|
||||
k = cnt[j] - (cnt[lo - 1] if lo > 0 else 0)
|
||||
if k > 0:
|
||||
sma[j] = s / k
|
||||
vr = np.where((sma > 0) & ~np.isnan(a), a / sma, np.nan)
|
||||
_VR_CACHE[key] = vr
|
||||
return vr
|
||||
|
||||
|
||||
class VolRegimeStop(ExitPolicy):
|
||||
def __init__(self, r: float, k1: float, mode: str):
|
||||
self.r = float(r)
|
||||
self.k1 = float(k1)
|
||||
self.mode = mode
|
||||
self.name = f"volreg r{r} k{k1} {mode[:3]}"
|
||||
|
||||
def open_trade(self, ctx: dict, i: int, d: int) -> dict:
|
||||
atr_i = ctx["atr14"][i]
|
||||
if not np.isfinite(atr_i) or atr_i <= 0:
|
||||
atr_i = 0.0
|
||||
return {"vr": _vol_ratio(ctx["atr14"]), "atr_i": float(atr_i)}
|
||||
|
||||
def levels(self, ctx: dict, i: int, d: int, j: int, st: dict):
|
||||
if j - 1 < 0:
|
||||
return None, self.mode
|
||||
vr_prev = st["vr"][j - 1]
|
||||
if not np.isfinite(vr_prev) or vr_prev <= self.r:
|
||||
return None, self.mode # regime calmo -> nessuno stop
|
||||
atr_i = st["atr_i"]
|
||||
if atr_i <= 0:
|
||||
return None, self.mode
|
||||
entry = ctx["close"][i]
|
||||
sl = entry - d * self.k1 * atr_i
|
||||
return sl, self.mode
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------------- grid
|
||||
|
||||
def _fmt(m: dict) -> str:
|
||||
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
|
||||
print("=== BASELINE (orizzonte puro) ===")
|
||||
base = {}
|
||||
for a in ("BTC", "ETH"):
|
||||
ctx = sleeves[a]
|
||||
tr = simulate(ctx, ExitPolicy(), t_hi=OOS_START_MS)
|
||||
oo = simulate(ctx, ExitPolicy(), t_lo=OOS_START_MS)
|
||||
base[a] = {"train": tr, "oos": oo}
|
||||
print(f" {a} TRAIN {_fmt(tr)}")
|
||||
print(f" {a} OOS {_fmt(oo)}")
|
||||
|
||||
rs = [1.2, 1.5]
|
||||
ks = [1.5, 2.0, 3.0]
|
||||
modes = ["intrabar", "close"]
|
||||
|
||||
print("\n=== GRID (TRAIN only) ===")
|
||||
grid = {}
|
||||
for mode in modes:
|
||||
print(f"\n--- mode={mode} ---")
|
||||
for r in rs:
|
||||
for k1 in ks:
|
||||
pol = VolRegimeStop(r, k1, mode)
|
||||
btc = simulate(sleeves["BTC"], pol, t_hi=OOS_START_MS)
|
||||
eth = simulate(sleeves["ETH"], pol, t_hi=OOS_START_MS)
|
||||
grid[(mode, r, k1)] = (btc, eth)
|
||||
print(f" r={r} k1={k1}: BTC {_fmt(btc)} | ETH {_fmt(eth)}")
|
||||
|
||||
# plateau check sul train: cella migliorativa se
|
||||
# ETH sharpe up & dd down & worst less-neg, BTC sharpe>=95% & ret>=80%
|
||||
bB_tr = base["BTC"]["train"]; bE_tr = base["ETH"]["train"]
|
||||
print("\n=== TRAIN improvement map (cell = ETH sh^ dd_v worst^ AND BTC ok) ===")
|
||||
improved = {}
|
||||
for key, (btc, eth) in grid.items():
|
||||
eth_ok = (eth["sharpe"] > bE_tr["sharpe"] and eth["dd"] < bE_tr["dd"]
|
||||
and eth["worst"] > bE_tr["worst"])
|
||||
btc_ok = (btc["sharpe"] >= 0.95 * bB_tr["sharpe"]
|
||||
and btc["ret"] >= 0.80 * bB_tr["ret"])
|
||||
improved[key] = eth_ok and btc_ok
|
||||
flag = "YES" if improved[key] else " . "
|
||||
print(f" {key}: {flag} (ethSh {eth['sharpe']:+.2f} vs {bE_tr['sharpe']:+.2f}, "
|
||||
f"ethDD {eth['dd']:.0f} vs {bE_tr['dd']:.0f}, ethW {eth['worst']:+.1f} vs {bE_tr['worst']:+.1f}, "
|
||||
f"btcSh {btc['sharpe']:.2f} btcRet {btc['ret']:+.0f})")
|
||||
|
||||
n_imp = sum(improved.values())
|
||||
print(f"\nTRAIN improving cells: {n_imp}/{len(grid)}")
|
||||
|
||||
# OOS verdict on improving cells (guardato UNA volta)
|
||||
print("\n=== OOS verdict (improving train cells) ===")
|
||||
for key, ok in improved.items():
|
||||
if not ok:
|
||||
continue
|
||||
mode, r, k1 = key
|
||||
pol = VolRegimeStop(r, k1, mode)
|
||||
btc = simulate(sleeves["BTC"], pol, t_lo=OOS_START_MS)
|
||||
eth = simulate(sleeves["ETH"], pol, t_lo=OOS_START_MS)
|
||||
print(f" {key}: BTC {_fmt(btc)} | ETH {_fmt(eth)}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,105 @@
|
||||
"""SH01 EXIT policy 11 — disaster_wide_intrabar (COMPLETENESS PROBE).
|
||||
|
||||
Le 10 policy precedenti hanno tutte fallito. Diagnosi ricorrente:
|
||||
- close-confirm (02,08) ALLARGA la coda su momentum-continuation (caso live
|
||||
ETH 2026-06-05): il close corre oltre il livello.
|
||||
- intrabar fisso (01) cappa AL livello (worst limitato) ma degrada BTC anche a k=5.
|
||||
|
||||
QUESTA probe chiude il buco: intrabar cap MOLTO LARGO (k=6..12), gap-aware,
|
||||
il cui UNICO scopo e' tagliare la coda catastrofica (la -14.9% ETH / il crash
|
||||
live -15.6%) SENZA mai toccare i normali pullback. E' la domanda diretta:
|
||||
"esiste un k cosi' largo che NON tocca BTC ma cappa la coda ETH?".
|
||||
|
||||
Anti-look-ahead: sl = entry - d*k*ATR14[i], congelato in open_trade (dati<=i);
|
||||
levels restituisce il livello costante, fill gap-aware nell'engine. mode=intrabar
|
||||
cappa AL livello (a differenza del close-confirm che lascia correre il close).
|
||||
|
||||
cd /opt/docker/PythagorasGoal && uv run python scripts/analysis/sh01_exit_policies/11_disaster_wide_intrabar.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, load_sleeves, simulate,
|
||||
)
|
||||
|
||||
|
||||
class DisasterWideIntrabar(ExitPolicy):
|
||||
def __init__(self, k: float):
|
||||
self.k = float(k)
|
||||
self.name = f"disaster_wide_intrabar k={k:.1f}"
|
||||
|
||||
def open_trade(self, ctx, i, d):
|
||||
atr = ctx["atr14"][i]
|
||||
entry = ctx["close"][i]
|
||||
sl = entry - d * self.k * atr if atr == atr and atr > 0 else None
|
||||
return {"sl": sl}
|
||||
|
||||
def levels(self, ctx, i, d, j, st):
|
||||
return st["sl"], "intrabar"
|
||||
|
||||
|
||||
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)},
|
||||
}
|
||||
|
||||
|
||||
def _row(tag, a, r):
|
||||
print(f" {tag:<7s} {a}: ret={r['ret']:>+7.0f}% dd={r['dd']:>4.0f}% "
|
||||
f"shrp={r['sharpe']:>5.2f} worst={r['worst']:>+6.1f}% "
|
||||
f"stop={r['stop_rate']:>4.1f}% trades={r['trades']}")
|
||||
|
||||
|
||||
def main():
|
||||
sleeves = load_sleeves()
|
||||
KS = [5.0, 6.0, 7.0, 8.0, 10.0, 12.0]
|
||||
|
||||
print("=" * 78)
|
||||
print("TRAIN GRID (intrabar cap LARGO, fill gap-aware)")
|
||||
print("=" * 78)
|
||||
train = {}
|
||||
for k in KS:
|
||||
pol = DisasterWideIntrabar(k)
|
||||
row = {a: simulate(sleeves[a], pol, t_hi=OOS_START_MS) for a in ("BTC", "ETH")}
|
||||
train[k] = row
|
||||
print(f" k={k:.1f}")
|
||||
_row("TRAIN", "BTC", row["BTC"])
|
||||
_row("TRAIN", "ETH", row["ETH"])
|
||||
|
||||
b_btc, b_eth = BASELINE["BTC"]["train"], BASELINE["ETH"]["train"]
|
||||
improving = []
|
||||
for k in KS:
|
||||
bt, et = train[k]["BTC"], train[k]["ETH"]
|
||||
eth_ok = (et["sharpe"] > b_eth["sharpe"] and et["dd"] < b_eth["dd"]
|
||||
and et["worst"] > b_eth["worst"])
|
||||
btc_ok = (bt["sharpe"] >= 0.95 * b_btc["sharpe"]
|
||||
and bt["ret"] >= 0.80 * b_btc["ret"])
|
||||
if eth_ok and btc_ok:
|
||||
improving.append(k)
|
||||
print(f" k={k:.1f} ETH_ok={eth_ok} BTC_ok={btc_ok} "
|
||||
f"(BTC shrp={bt['sharpe']:.2f} ret={bt['ret']:.0f} | "
|
||||
f"ETH shrp={et['sharpe']:.2f} dd={et['dd']:.0f} worst={et['worst']:.1f})")
|
||||
print(f"\n improving cells (train): {improving}")
|
||||
|
||||
if not improving:
|
||||
print(" -> NESSUNA cella migliorativa: NO pulito, OOS non guardato.")
|
||||
return
|
||||
|
||||
# plateau >=3 adiacenti? poi OOS
|
||||
print("\n Plateau candidate -> OOS verdetto:")
|
||||
for k in improving:
|
||||
oos = {a: simulate(sleeves[a], DisasterWideIntrabar(k), t_lo=OOS_START_MS)
|
||||
for a in ("BTC", "ETH")}
|
||||
print(f" k={k:.1f}")
|
||||
_row("OOS", "BTC", oos["BTC"])
|
||||
_row("OOS", "ETH", oos["ETH"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -66,6 +66,11 @@ PORTFOLIOS = {
|
||||
weighting="cap", caps={"PAIRS": 0.33}),
|
||||
"PORT05": Portfolio("PORT05", "Master esteso", FADE + HONEST + PAIRS + TSM,
|
||||
weighting="cap", caps={"PAIRS": 0.33}),
|
||||
# SHAPE cappata a ~mezzo sleeve equal (2026-06-05): SH01 non ha SL per design e la
|
||||
# ricerca multi-agente (sh01_exit_lab, 11 famiglie di stop, 0 sopravvissute) dimostra
|
||||
# che NESSUNO stop taglia la coda ETH senza rompere l'edge -> si dimezza l'esposizione
|
||||
# (costo backtest ~0: FULL 6.47->6.43, OOS 8.82->8.58, FULL DD 4.10->3.96). Vedi
|
||||
# docs/diary/2026-06-05-sh01-sl-research.md.
|
||||
"PORT06": Portfolio("PORT06", "Master + shape", FADE + HONEST + PAIRS + TSM + SHAPE,
|
||||
weighting="cap", caps={"PAIRS": 0.33}, leverage=2.0),
|
||||
weighting="cap", caps={"PAIRS": 0.33, "SHAPE": 0.0588}, leverage=2.0),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user