"""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()