feat: trailing-stop exit sweep in the backtest
Deploy / lint (push) Successful in 6s
Deploy / test (push) Successful in 55s
Deploy / deploy (push) Successful in 32s

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 <noreply@anthropic.com>
This commit is contained in:
2026-06-30 17:33:17 +02:00
parent c5f6b07a3e
commit ab9ce18809
4 changed files with 179 additions and 0 deletions
+82
View File
@@ -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": (