docs(diary): exit ladder 80/20 con runner+stop dinamico SCARTATO (il runner non corre: TP=media, ~3% oltre TP, −38% ret su MR02 ETH)
Test onesto scripts/analysis/partial_tp_ladder.py: stessi segnali, engine intrabar fade_base, fee identiche. Il ladder compra win-rate/DD pagando i winner migliori — stesso profilo del vol-target gia' scartato. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,164 @@
|
||||
"""Exit 'ladder' per le fade: al tocco di una frazione f del TP esce la quota q,
|
||||
il runner (1-q) corre con STOP DINAMICO bloccato alla soglia (profitto lockato).
|
||||
|
||||
Proposta 2026-06-04 ("e se all'80% del TP usciamo con 80% e mettiamo un SL
|
||||
dinamico su quella soglia e lo lasciamo correre?"). Confronto ONESTO con l'exit
|
||||
canonico (TP pieno al livello) sugli STESSI segnali, stesso engine intrabar di
|
||||
fade_base (ingresso a close[i], SL prioritario nello stesso bar, fee 0.10% RT
|
||||
x leva su tutto il notional — il ladder NON paga fee extra: due uscite ma
|
||||
stesso notional totale).
|
||||
|
||||
Convenzioni intrabar del ladder (oneste/conservative):
|
||||
- SL pieno prioritario sulla soglia nello stesso bar;
|
||||
- se il bar che tocca la soglia CHIUDE oltre la soglia (rientro), il runner
|
||||
si considera stoppato subito alla soglia (non gli si regala il bar);
|
||||
- il runner non ha TP (corre), esce su ri-tocco della soglia o a max_bars.
|
||||
|
||||
Gate (metodologia repo): il ladder deve migliorare ret E DD (o chiaramente il
|
||||
rischio/rendimento) su ENTRAMBI gli asset, full E OOS, per tutte e 3 le fade.
|
||||
|
||||
uv run python scripts/analysis/partial_tp_ladder.py
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
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))
|
||||
|
||||
from src.data.downloader import load_data # noqa: E402
|
||||
from src.live.strategy_loader import load_strategy # noqa: E402
|
||||
|
||||
LIVE_PARAMS = dict(trend_max=3.0, ema_long=200, hurst_max=0.55, min_tp_frac=0.0015)
|
||||
OOS_START = pd.Timestamp("2023-11-01", tz="UTC")
|
||||
LEV, POS, FEE_RT = 3.0, 0.15, 0.001
|
||||
CODES = ["MR01_bollinger_fade", "MR02_donchian_fade", "MR07_return_reversal"]
|
||||
|
||||
|
||||
def simulate(signals, df, ts, policy: str, f: float = 0.8, q: float = 0.8,
|
||||
start=None) -> dict:
|
||||
"""Replay intrabar degli stessi segnali con exit 'base' o 'ladder'."""
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
n = len(c)
|
||||
fee = FEE_RT * LEV
|
||||
capital = peak = 1000.0
|
||||
max_dd = 0.0
|
||||
last_exit = -1
|
||||
trades = wins = 0
|
||||
runner_beyond_tp = runner_stopped = 0
|
||||
rets = []
|
||||
|
||||
for sig in signals:
|
||||
i, d = sig.idx, sig.direction
|
||||
if start is not None and ts.iloc[i] < start:
|
||||
continue
|
||||
if i <= last_exit or i + 1 >= n:
|
||||
continue
|
||||
entry = c[i]
|
||||
tp, sl, mb = sig.metadata["tp"], sig.metadata["sl"], sig.metadata["max_bars"]
|
||||
j = i
|
||||
|
||||
if policy == "base":
|
||||
exit_p = c[min(i + mb, n - 1)]
|
||||
for step in range(1, mb + 1):
|
||||
j = i + step
|
||||
if j >= n:
|
||||
j = n - 1; exit_p = c[j]; break
|
||||
hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
|
||||
hit_tp = (d == 1 and h[j] >= tp) or (d == -1 and l[j] <= tp)
|
||||
if hit_sl:
|
||||
exit_p = sl; break
|
||||
if hit_tp:
|
||||
exit_p = tp; break
|
||||
if step == mb:
|
||||
exit_p = c[j]
|
||||
ret = (exit_p - entry) / entry * d * LEV - fee
|
||||
else: # ladder
|
||||
th = entry + f * (tp - entry) # soglia f del percorso verso il TP
|
||||
p1 = p2 = None # fill quota q / runner (1-q)
|
||||
state = "full"
|
||||
for step in range(1, mb + 1):
|
||||
j = i + step
|
||||
if j >= n:
|
||||
j = n - 1
|
||||
if p1 is None:
|
||||
p1 = c[j]
|
||||
p2 = c[j]
|
||||
break
|
||||
if state == "full":
|
||||
hit_sl = (d == 1 and l[j] <= sl) or (d == -1 and h[j] >= sl)
|
||||
hit_th = (d == 1 and h[j] >= th) or (d == -1 and l[j] <= th)
|
||||
if hit_sl: # SL pieno prioritario (conservativo)
|
||||
p1 = p2 = sl; break
|
||||
if hit_th:
|
||||
p1 = th; state = "runner"
|
||||
# il bar chiude oltre la soglia -> runner stoppato subito
|
||||
if (d == 1 and c[j] < th) or (d == -1 and c[j] > th):
|
||||
p2 = th; runner_stopped += 1; break
|
||||
if step == mb:
|
||||
p2 = c[j]; break
|
||||
continue
|
||||
if step == mb:
|
||||
p1 = p2 = c[j]; break
|
||||
else: # runner: stop dinamico = soglia
|
||||
hit_stop = (d == 1 and l[j] <= th) or (d == -1 and h[j] >= th)
|
||||
if hit_stop:
|
||||
p2 = th; runner_stopped += 1; break
|
||||
if step == mb:
|
||||
p2 = c[j]; break
|
||||
if p2 is not None and (p2 - tp) * d > 0:
|
||||
runner_beyond_tp += 1
|
||||
ret = (q * (p1 - entry) + (1 - q) * (p2 - entry)) / 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
|
||||
rets.append(ret)
|
||||
|
||||
if trades == 0:
|
||||
return {}
|
||||
rets = np.array(rets)
|
||||
return {
|
||||
"ret_pct": (capital / 1000.0 - 1) * 100,
|
||||
"dd_pct": max_dd * 100,
|
||||
"trades": trades,
|
||||
"win_pct": wins / trades * 100,
|
||||
"avg_ret_bps": rets.mean() * 1e4,
|
||||
"sharpe_t": rets.mean() / rets.std() * np.sqrt(len(rets)) if rets.std() else 0,
|
||||
"runner_beyond_tp": runner_beyond_tp,
|
||||
"runner_stopped": runner_stopped,
|
||||
}
|
||||
|
||||
|
||||
def main() -> None:
|
||||
variants = [("base", None, None), ("ladder", 0.8, 0.8),
|
||||
("ladder", 0.8, 0.5), ("ladder", 0.5, 0.5)]
|
||||
for code in CODES:
|
||||
strat = load_strategy(code)
|
||||
for asset in ("BTC", "ETH"):
|
||||
df = load_data(asset, "1h")
|
||||
ts = pd.to_datetime(df["timestamp"], unit="ms", utc=True)
|
||||
signals = strat.generate_signals(df, ts, **LIVE_PARAMS)
|
||||
print(f"\n=== {code} {asset} 1h — {len(signals)} segnali (params live) ===")
|
||||
print(f"{'policy':<16}{'periodo':<6}{'ret%':>10}{'DD%':>8}{'win%':>7}"
|
||||
f"{'avg bps':>9}{'Sh(t)':>7}{'n':>6}{'run>TP':>8}{'run-stop':>9}")
|
||||
for policy, f, q in variants:
|
||||
tag = "base" if policy == "base" else f"ladder f{f} q{q}"
|
||||
for label, start in (("FULL", None), ("OOS", OOS_START)):
|
||||
r = simulate(signals, df, ts, policy, f or 0.8, q or 0.8, start)
|
||||
if not r:
|
||||
continue
|
||||
print(f"{tag:<16}{label:<6}{r['ret_pct']:>10.0f}{r['dd_pct']:>8.1f}"
|
||||
f"{r['win_pct']:>7.1f}{r['avg_ret_bps']:>9.1f}{r['sharpe_t']:>7.2f}"
|
||||
f"{r['trades']:>6}{r['runner_beyond_tp']:>8}{r['runner_stopped']:>9}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user