From ab9ce18809bf860a8365adffb21474859d38c762 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Tue, 30 Jun 2026 17:33:17 +0200 Subject: [PATCH] feat: trailing-stop exit sweep in the backtest MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third exit model alongside target-vs-stop and the fixed take-profit. The TP sweep showed the edge lives in the fat tail (avg R keeps rising as you let winners run), but a fixed wide target is win-rate-brutal and gives everything back on a reversal. A trailing stop harvests the tail while protecting gains. Per setup the replay computes the realized R for several trail widths (3/5/7/10/ 15/20%) in a single conservative pass — stop ratchets up via max(initial_stop, peak*(1-trail)), exit on the pullback or at the horizon close, R vs the initial risk. Aggregated into a trailing sweep (win rate = share closed in profit, avg R, total R) over the qualified set and shown as a new table in the Backtest panel. Co-Authored-By: Claude Opus 4.8 --- app/services/backtest_service.py | 82 +++++++++++++++++++ .../src/components/signals/BacktestPanel.tsx | 50 +++++++++++ frontend/src/lib/types.ts | 10 +++ tests/unit/test_backtest_service.py | 37 +++++++++ 4 files changed, 179 insertions(+) diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 147096d..97aba27 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -251,6 +251,59 @@ def _tp_primitives( return risk_pct, stopped, mfe, close_pct +def _trailing_exits( + direction: str, entry: float, init_stop: float, trail_fracs, forward: list, horizon: int +) -> dict[int, float]: + """Realized R per trailing-stop width, in one pass over the post-entry bars. + + The stop ratchets up (never below the initial stop): ``max(init_stop, + peak*(1-trail))`` for a long. Exit when a bar pierces the current stop (filled + at the stop level), else at the horizon-end close. Each width is keyed by its + integer percent (5 for 0.05). Conservative: the stop for a bar uses the peak + through the *previous* bar (this bar's high is folded in only afterwards). + R is relative to the initial risk (entry → init_stop). + """ + long = direction == "long" + risk = abs(entry - init_stop) / entry if entry else 0.0 + if risk <= 0: + return {round(f * 100): 0.0 for f in trail_fracs} + bars = forward[:horizon] + if not bars: + return {round(f * 100): 0.0 for f in trail_fracs} + + result: dict[int, float] = {} + peak = entry + active = list(trail_fracs) + for r in bars: + remaining = [] + for f in active: + if long: + stop_level = max(init_stop, peak * (1 - f)) + if r.low <= stop_level: + result[round(f * 100)] = ((stop_level - entry) / entry) / risk + continue + else: + stop_level = min(init_stop, peak * (1 + f)) + if r.high >= stop_level: + result[round(f * 100)] = ((entry - stop_level) / entry) / risk + continue + remaining.append(f) + active = remaining + if not active: + break + if long: + if r.high > peak: + peak = r.high + elif r.low < peak: + peak = r.low + + last_close = bars[-1].close + timeout_r = (((last_close - entry) / entry) if long else ((entry - last_close) / entry)) / risk + for f in active: + result[round(f * 100)] = timeout_r + return result + + def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) -> list[dict]: """Walk one ticker's history weekly, building setups and their realized outcomes.""" candidates: list[dict] = [] @@ -281,6 +334,9 @@ def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) - risk_pct, tp_stopped, mfe_pct, tp_close_pct = _tp_primitives( s["direction"], s["entry"], s["stop"], forward, HORIZON ) + trail_r = _trailing_exits( + s["direction"], s["entry"], s["stop"], TRAIL_LEVELS, forward, HORIZON + ) iso = records[i].date.isocalendar() candidates.append({ "symbol": symbol, @@ -300,6 +356,7 @@ def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) - "tp_stopped": tp_stopped, "mfe_pct": mfe_pct, "tp_close_pct": tp_close_pct, + "trail_r": trail_r, }) return candidates @@ -327,6 +384,9 @@ def _bucket_stats(cands: list[dict]) -> dict: # it's a standalone fixed-% exit; exiting at the target is the target model. TP_LEVELS = (0.04, 0.06, 0.08, 0.10, 0.12, 0.15, 0.20, 0.25, 0.30) +# Trailing-stop widths (give-back from the peak) swept for the trailing exit model. +TRAIL_LEVELS = (0.03, 0.05, 0.07, 0.10, 0.15, 0.20) + def _take_profit_bucket(cands: list[dict], tp: float) -> dict: """Stats for a fixed take-profit exit at +``tp`` (fraction): bank +tp if it's @@ -357,6 +417,27 @@ def _take_profit_bucket(cands: list[dict], tp: float) -> dict: } +def _trailing_bucket(cands: list[dict], trail_pct: int) -> dict: + """Stats for a trailing-stop exit of width ``trail_pct`` (integer percent). + Each candidate carries its realized R for this width in ``trail_r``; a "win" + is simply an exit in profit (R > 0).""" + rs = [ + c["trail_r"][trail_pct] + for c in cands + if c.get("trail_r", {}).get(trail_pct) is not None + ] + total = len(rs) + wins = sum(1 for r in rs if r > 0) + return { + "trail_pct": trail_pct, + "total": total, + "wins": wins, + "win_rate": round(wins / total * 100, 1) if total else None, + "avg_r": round(sum(rs) / total, 3) if total else None, + "total_r": round(sum(rs), 2) if total else None, + } + + def _calibration(cands: list[dict]) -> list[dict]: """Predicted target probability vs realized hit rate, per probability bucket.""" rows: list[dict] = [] @@ -792,6 +873,7 @@ async def run_backtest( "min_momentum_percentile": current_min_pct, "sweep": sweep, "take_profit_sweep": [_take_profit_bucket(qualified, tp) for tp in TP_LEVELS], + "trailing_sweep": [_trailing_bucket(qualified, round(f * 100)) for f in TRAIL_LEVELS], "calibration": _calibration(candidates), "signal_eval": _signal_evaluation(collected), "signal_eval_note": ( diff --git a/frontend/src/components/signals/BacktestPanel.tsx b/frontend/src/components/signals/BacktestPanel.tsx index f3dd116..6a14e3a 100644 --- a/frontend/src/components/signals/BacktestPanel.tsx +++ b/frontend/src/components/signals/BacktestPanel.tsx @@ -89,6 +89,10 @@ export function BacktestPanel() { report?.take_profit_sweep && report.take_profit_sweep.length > 0 ? Math.max(...report.take_profit_sweep.map((r) => r.avg_r ?? -Infinity)) : null; + const bestTrailAvgR = + report?.trailing_sweep && report.trailing_sweep.length > 0 + ? Math.max(...report.trailing_sweep.map((r) => r.avg_r ?? -Infinity)) + : null; const run = useMutation({ mutationFn: () => triggerJob('backtest'), @@ -286,6 +290,52 @@ export function BacktestPanel() { )} + {report.trailing_sweep && report.trailing_sweep.length > 0 && ( +
+

+ Trailing-stop exit +

+

+ Let it run, but exit when price gives back X% from its + peak (the stop only ratchets up, never below the initial stop). Captures the tail + without the fixed take-profit's all-or-nothing miss, and protects gains. In R vs the initial + risk. Win Rate = share closed in profit. ★ = best avg R. +

+
+ + + + + + + + + + + + + {report.trailing_sweep.map((row) => { + const best = row.avg_r != null && row.avg_r === bestTrailAvgR; + return ( + + + + + + + + + ); + })} + +
TrailSetupsProfitableWin RateAvg RTotal R
+ {best && ★} + {row.trail_pct}% + {row.total}{row.wins}{fmtPct(row.win_rate)}{fmtR(row.avg_r)}{fmtR(row.total_r)}
+
+
+ )} +

Probability calibration diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index 66b735d..aaf11ed 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -239,6 +239,15 @@ export interface BacktestTakeProfitRow { total_r: number | null; } +export interface BacktestTrailingRow { + trail_pct: number; + total: number; + wins: number; + win_rate: number | null; + avg_r: number | null; + total_r: number | null; +} + export interface BacktestSignalEvalRow { signal: string; weeks: number; @@ -262,6 +271,7 @@ export interface BacktestReport { min_momentum_percentile: number; sweep: BacktestSweepRow[]; take_profit_sweep?: BacktestTakeProfitRow[]; + trailing_sweep?: BacktestTrailingRow[]; calibration: BacktestCalibrationRow[]; signal_eval?: BacktestSignalEvalRow[]; signal_eval_note?: string; diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index df6aa56..35f6fdf 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -95,6 +95,43 @@ class TestTakeProfitBucket: assert b["avg_r"] is None +class TestTrailingExits: + def test_locks_gain_on_pullback(self): + # Runs to 120, then a 10% trail (from peak 120 → 108) is pierced on the drop. + res = bt._trailing_exits("long", 100.0, 90.0, (0.10,), [_bar(120, 110, 118), _bar(130, 100, 105)], 30) + assert res[10] == pytest.approx(0.8) # (108-100)/100 / 0.10 risk + + def test_initial_stop_caps_loss(self): + # Trail (20%) is looser than the initial stop → initial stop governs = -1R. + res = bt._trailing_exits("long", 100.0, 90.0, (0.20,), [_bar(101, 89, 90)], 30) + assert res[20] == pytest.approx(-1.0) + + def test_timeout_exits_at_close(self): + res = bt._trailing_exits("long", 100.0, 90.0, (0.20,), [_bar(105, 98, 104), _bar(106, 100, 105)], 30) + assert res[20] == pytest.approx(0.5) # close 105 → +5% / 10% risk + + def test_multiple_widths_one_pass(self): + # Tighter trail locks in more here (exit at 114 vs 108). + res = bt._trailing_exits("long", 100.0, 90.0, (0.10, 0.05), [_bar(120, 110, 118), _bar(130, 100, 105)], 30) + assert res[10] == pytest.approx(0.8) + assert res[5] == pytest.approx(1.4) + + +class TestTrailingBucket: + def test_bucket(self): + cands = [ + {"trail_r": {5: 1.4, 10: 0.8}}, + {"trail_r": {5: -1.0, 10: -1.0}}, + {"trail_r": {5: 0.5, 10: 0.5}}, + ] + b = bt._trailing_bucket(cands, 5) + assert b["total"] == 3 + assert b["wins"] == 2 + assert b["win_rate"] == pytest.approx(66.7, abs=0.1) + assert b["total_r"] == pytest.approx(0.9, abs=0.01) + assert b["avg_r"] == pytest.approx(0.3, abs=0.01) + + def test_bucket_stats_counts_and_expectancy(): cands = [ _cand(70, OUTCOME_TARGET_HIT, 3.0), # +3R win