research(exit-lab): 34 agenti su exit dinamiche → EXIT-16 close-confirm SL PROMOSSO a livello PORT06
23 famiglie esplorate (harness condiviso exit_lab, train/OOS embargo nov-2023, tutto lo storico 1h 2018-2026) + 10 verifiche avversariali + test PORT06. 'Cavalcare il prezzo' non esiste (4a conferma: oltre il TP=media non c'e' coda). Scoperta: lo SL intrabar fisso e' il distruttore di valore n.1 delle fade (stop da wick = falsi negativi). Forma robusta: SL solo su CHIUSURA oltre sl0±0.5·ATR14 — PORT06 FULL Sharpe 6.47→7.84 DD 4.10→2.60, OOS 8.82→10.06. Collaterali: bias gap-through dell'engine sugli stop stretti; ramo -2% del worker morto con sl=0. Diario: docs/diary/2026-06-04-exit-lab.md Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,261 @@
|
||||
"""VERIFY EXIT-16 close_confirm_sl — lente STRESS (avversariale).
|
||||
|
||||
Ipotesi nulla: l'edge della close-confirm-SL e' fragile a frizioni reali.
|
||||
Quattro stress, tutti su segnali cache (params LIVE, hurst_max=0.55):
|
||||
|
||||
(1) FEE 2x: FEE_RT=0.002 (vs 0.001). Penalizza le policy che girano piu' capitale.
|
||||
(2) BEAR/CRASH 2021-01..2022-12 (LUNA/FTX/19-mag-21): worst-trade + 5 peggiori
|
||||
trade della policy vs base. Lo SL disattivato lascia correre le perdite?
|
||||
(3) SLIPPAGE AVVERSO 20bps sulle uscite della policy: ogni fill di USCITA paga
|
||||
+20bps contro la posizione (prezzo di uscita peggiorato). L'edge regge?
|
||||
NB: lo applico SOLO alle uscite della POLICY (la sua tesi e' "esco al close":
|
||||
il close-fill e' market, paga slippage; la base esce a livelli limite sl0/tp0).
|
||||
(4) OVERLAP/TURNOVER: la policy allunga la permanenza (no stop intrabar). Conto
|
||||
i segnali SALTATI per non-overlap (i <= last_exit) base vs policy, e quanto
|
||||
capitale-tempo (somma bars in posizione) gira in piu'.
|
||||
|
||||
Tutto via simulate() con monkeypatch di FEE_RT e una sottoclasse engine per lo
|
||||
slippage. Niente modifiche ad altri file.
|
||||
"""
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[3]))
|
||||
|
||||
import exit_lab # noqa: E402
|
||||
from exit_lab import ExitPolicy, simulate, OOS_START_MS, HARD_CAP, LEV, POS # noqa: E402
|
||||
|
||||
import importlib.util
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
"cc16", str(Path(__file__).resolve().parent / "16_close_confirm_sl.py"))
|
||||
cc16 = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(cc16)
|
||||
CloseConfirmSl = cc16.CloseConfirmSl
|
||||
|
||||
BUF = 0.5 # train-pick
|
||||
|
||||
|
||||
def fmt(r):
|
||||
if not r:
|
||||
return " n/a"
|
||||
return (f"ret{r['ret_pct']:>8.0f}% dd{r['dd_pct']:>5.1f} sh{r['sharpe_t']:>5.2f} "
|
||||
f"n{r['trades']:>4} win{r['win_pct']:>4.0f} bars{r['avg_bars']:>5.1f}")
|
||||
|
||||
|
||||
def sub(cls, sleeve, g, s, e):
|
||||
return simulate(cls, sleeve, g, start_ms=s, end_ms=e)
|
||||
|
||||
|
||||
def ms(d):
|
||||
return int(pd.Timestamp(d, tz="UTC").value // 1e6)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Engine "instrumented" che riproduce simulate() ma:
|
||||
# - applica uno slippage avverso (bps) su OGNI fill di USCITA (solo se policy)
|
||||
# - raccoglie la lista dei ret per-trade e i segnali SALTATI per non-overlap
|
||||
# - raccoglie capital-time (somma bars)
|
||||
# Lo tengo allineato 1:1 con exit_lab.simulate (stesso ordine SL-prima-di-TP).
|
||||
# ===========================================================================
|
||||
def simulate_instr(policy_cls, sleeve, params=None, start_ms=None, end_ms=None,
|
||||
exit_slip_bps=0.0):
|
||||
params = params or {}
|
||||
h, l, c, ts = sleeve["high"], sleeve["low"], sleeve["close"], sleeve["ts_ms"]
|
||||
n = len(c)
|
||||
ctx = dict(sleeve)
|
||||
policy_cls.prepare(ctx, **params)
|
||||
fee = exit_lab.FEE_RT * LEV
|
||||
slip = exit_slip_bps * 1e-4
|
||||
capital = peak = 1000.0
|
||||
max_dd = 0.0
|
||||
last_exit = -1
|
||||
trades = wins = 0
|
||||
bars_tot = 0
|
||||
skipped_overlap = 0
|
||||
rets = [] # (ret, ts_entry, bars)
|
||||
for (i, d, tp0, sl0, mb) in sleeve["signals"]:
|
||||
if start_ms is not None and ts[i] < start_ms:
|
||||
continue
|
||||
if end_ms is not None and ts[i] >= end_ms:
|
||||
continue
|
||||
if i + 1 >= n:
|
||||
continue
|
||||
if i <= last_exit:
|
||||
skipped_overlap += 1
|
||||
continue
|
||||
entry = c[i]
|
||||
pol = policy_cls(ctx, i, d, entry, tp0, sl0, mb, **params)
|
||||
horizon = min(int(pol.horizon), HARD_CAP)
|
||||
fills = []
|
||||
remaining = 1.0
|
||||
j = i
|
||||
for step in range(1, horizon + 1):
|
||||
j = i + step
|
||||
if j >= n:
|
||||
j = n - 1
|
||||
fills.append((remaining, c[j])); remaining = 0.0
|
||||
break
|
||||
tp, sl, tpfrac = pol.levels(j)
|
||||
hit_sl = sl is not None and ((d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl))
|
||||
hit_tp = tp is not None and ((d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp))
|
||||
if hit_sl:
|
||||
fills.append((remaining, sl)); remaining = 0.0
|
||||
break
|
||||
if hit_tp:
|
||||
f = min(max(tpfrac, 0.0), 1.0) * remaining
|
||||
if f > 0:
|
||||
fills.append((f, tp)); remaining -= f
|
||||
if remaining <= 1e-9:
|
||||
break
|
||||
pol.on_partial(j, tp, remaining)
|
||||
if pol.after_bar(j):
|
||||
fills.append((remaining, c[j])); remaining = 0.0
|
||||
break
|
||||
if step == horizon:
|
||||
fills.append((remaining, c[j])); remaining = 0.0
|
||||
if remaining > 1e-9:
|
||||
fills.append((remaining, c[j]))
|
||||
# slippage avverso sull'uscita: il prezzo di uscita peggiora di slip,
|
||||
# cioe' si vende piu' basso (long) / si ricompra piu' alto (short).
|
||||
def adj(p):
|
||||
return p * (1.0 - slip) if d == 1 else p * (1.0 + slip)
|
||||
ret = sum(f * (adj(p) - entry) for f, p in fills) / entry * d * LEV - fee
|
||||
capital = max(capital + capital * POS * ret, 10.0)
|
||||
peak = max(peak, capital)
|
||||
max_dd = max(max_dd, (peak - capital) / peak)
|
||||
last_exit = j
|
||||
trades += 1
|
||||
wins += ret > 0
|
||||
bars_tot += j - i
|
||||
rets.append((ret, int(ts[i]), j - i))
|
||||
if trades == 0:
|
||||
return {}
|
||||
r = np.array([x[0] for x in rets])
|
||||
return {
|
||||
"ret_pct": (capital / 1000.0 - 1) * 100,
|
||||
"dd_pct": max_dd * 100,
|
||||
"trades": trades,
|
||||
"win_pct": wins / trades * 100,
|
||||
"sharpe_t": float(r.mean() / r.std() * np.sqrt(len(r))) if r.std() else 0.0,
|
||||
"avg_bars": bars_tot / trades,
|
||||
"bars_tot": bars_tot,
|
||||
"skipped_overlap": skipped_overlap,
|
||||
"rets": rets,
|
||||
"worst5": sorted(r.tolist())[:5],
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
data = exit_lab.load_sleeves()
|
||||
|
||||
# ===================================================================
|
||||
print("=" * 104)
|
||||
print("TEST 1 — FEE 2x (FEE_RT 0.001 -> 0.002). base vs policy buffer=0.5 (OOS, dopo 2023-11)")
|
||||
print("=" * 104)
|
||||
orig_fee = exit_lab.FEE_RT
|
||||
survive_fee = True
|
||||
for fee in (0.001, 0.002):
|
||||
exit_lab.FEE_RT = fee
|
||||
print(f"\n--- FEE_RT={fee} ---")
|
||||
for (code, asset), sleeve in data.items():
|
||||
key = f"{code.split('_')[0]} {asset}"
|
||||
b = sub(ExitPolicy, sleeve, {}, OOS_START_MS, None)
|
||||
p = sub(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None)
|
||||
tag = ""
|
||||
if fee == 0.002 and b and p:
|
||||
# regge se sharpe policy >= base (la tesi e' che migliora)
|
||||
ok = p["sharpe_t"] >= b["sharpe_t"] - 0.10
|
||||
survive_fee &= ok
|
||||
tag = "OK" if ok else "WORSE"
|
||||
print(f" {key:<10} base {fmt(b)}")
|
||||
print(f" {'':<10} pol {fmt(p)} {tag}")
|
||||
exit_lab.FEE_RT = orig_fee
|
||||
print(f"\nFEE 2x: policy regge (>= base-0.10 sh su tutti gli sleeve OOS)? {survive_fee}")
|
||||
|
||||
# ===================================================================
|
||||
print("\n" + "=" * 104)
|
||||
print("TEST 2 — BEAR/CRASH 2021-01..2022-12 (LUNA/FTX/19-mag): worst-trade + 5 peggiori")
|
||||
print("=" * 104)
|
||||
s2, e2 = ms("2021-01-01"), ms("2023-01-01")
|
||||
tail_worse = 0
|
||||
tail_total = 0
|
||||
for (code, asset), sleeve in data.items():
|
||||
key = f"{code.split('_')[0]} {asset}"
|
||||
b = simulate_instr(ExitPolicy, sleeve, {}, s2, e2)
|
||||
p = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, s2, e2)
|
||||
print(f"\n{key}")
|
||||
print(f" base {fmt(b)}")
|
||||
print(f" pol {fmt(p)}")
|
||||
if b and p:
|
||||
bw = [f"{x*100:+.1f}%" for x in b["worst5"]]
|
||||
pw = [f"{x*100:+.1f}%" for x in p["worst5"]]
|
||||
print(f" base 5 peggiori (ret netto): {bw}")
|
||||
print(f" pol 5 peggiori (ret netto): {pw}")
|
||||
tail_total += 1
|
||||
# la policy peggiora la coda se il worst-trade e' piu' negativo
|
||||
if p["worst5"][0] < b["worst5"][0] - 0.005:
|
||||
tail_worse += 1
|
||||
print(f" -> CODA PEGGIORE: worst {p['worst5'][0]*100:+.1f}% < base {b['worst5'][0]*100:+.1f}%")
|
||||
else:
|
||||
print(f" -> coda OK: worst {p['worst5'][0]*100:+.1f}% vs base {b['worst5'][0]*100:+.1f}%")
|
||||
print(f" DD bear: base {b['dd_pct']:.1f}% pol {p['dd_pct']:.1f}%")
|
||||
print(f"\nBEAR: sleeve con coda PEGGIORE (worst-trade > 0.5pt sotto base): {tail_worse}/{tail_total}")
|
||||
|
||||
# ===================================================================
|
||||
print("\n" + "=" * 104)
|
||||
print("TEST 3 — SLIPPAGE AVVERSO 20bps sulle uscite della POLICY (OOS). base senza slippage")
|
||||
print("(la tesi della policy e' 'esco al close' = market fill -> paga slippage)")
|
||||
print("=" * 104)
|
||||
survive_slip = True
|
||||
for (code, asset), sleeve in data.items():
|
||||
key = f"{code.split('_')[0]} {asset}"
|
||||
b = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None, exit_slip_bps=0.0)
|
||||
p0 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=0.0)
|
||||
p20 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=20.0)
|
||||
ok = p20 and b and p20["sharpe_t"] >= b["sharpe_t"] - 0.10
|
||||
survive_slip &= bool(ok)
|
||||
print(f"\n{key}")
|
||||
print(f" base (no slip) {fmt(b)}")
|
||||
print(f" pol (no slip) {fmt(p0)}")
|
||||
print(f" pol (+20bps exit) {fmt(p20)} {'OK' if ok else 'WORSE vs base'}")
|
||||
print(f"\nSLIPPAGE 20bps: policy ancora >= base-0.10 sh su tutti? {survive_slip}")
|
||||
print("(test severo: lo slippage colpisce la policy ma NON la base — asimmetria pessimistica)")
|
||||
|
||||
# severita' extra: slippage anche sulla base (entrambe market) per fairness
|
||||
print("\n--- fairness: 20bps anche sulle uscite della BASE ---")
|
||||
fair = True
|
||||
for (code, asset), sleeve in data.items():
|
||||
key = f"{code.split('_')[0]} {asset}"
|
||||
b20 = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None, exit_slip_bps=20.0)
|
||||
p20 = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None, exit_slip_bps=20.0)
|
||||
ok = p20 and b20 and p20["sharpe_t"] >= b20["sharpe_t"] - 0.10
|
||||
fair &= bool(ok)
|
||||
print(f" {key:<10} base+20 {fmt(b20)}")
|
||||
print(f" {'':<10} pol +20 {fmt(p20)} {'OK' if ok else 'WORSE'}")
|
||||
print(f"fairness (entrambe +20bps): policy >= base-0.10 sh? {fair}")
|
||||
|
||||
# ===================================================================
|
||||
print("\n" + "=" * 104)
|
||||
print("TEST 4 — OVERLAP/TURNOVER: segnali saltati per non-overlap + capital-time (OOS)")
|
||||
print("=" * 104)
|
||||
for (code, asset), sleeve in data.items():
|
||||
key = f"{code.split('_')[0]} {asset}"
|
||||
b = simulate_instr(ExitPolicy, sleeve, {}, OOS_START_MS, None)
|
||||
p = simulate_instr(CloseConfirmSl, sleeve, {"buffer": BUF}, OOS_START_MS, None)
|
||||
if b and p:
|
||||
dskip = p["skipped_overlap"] - b["skipped_overlap"]
|
||||
dbars = p["bars_tot"] - b["bars_tot"]
|
||||
print(f" {key:<10} base: trades {b['trades']:>4} skip-overlap {b['skipped_overlap']:>4} "
|
||||
f"bars_tot {b['bars_tot']:>6} avg {b['avg_bars']:.1f}")
|
||||
print(f" {'':<10} pol : trades {p['trades']:>4} skip-overlap {p['skipped_overlap']:>4} "
|
||||
f"bars_tot {p['bars_tot']:>6} avg {p['avg_bars']:.1f}")
|
||||
print(f" {'':<10} -> +{dskip} segnali persi per overlap, "
|
||||
f"+{dbars} bars in posizione ({dbars/max(b['bars_tot'],1)*100:+.0f}% capital-time)")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user