feat: take-profit exit sweep in the backtest (alongside target-vs-stop)
The target-vs-stop model counts a near-miss of a far S/R target as a full loss and ignores the partial gains you actually bank — so it measures a different strategy than "scalp the early pop, take +8%". Add a realistic take-profit exit model next to it (original untouched). Per setup the replay now also records risk%, whether the stop was hit, the favourable excursion reachable before the stop (MFE), and the horizon-close move. From those a fixed-take-profit sweep (4/6/8/10/12/15%) is scored in R: bank +X% if reached before the stop, else -1R, else the horizon close. Hit rate = how often +X% was banked (the MFE CDF), so you can pick the EV-optimal TP without top-ticking fantasy. Shown as a new table in the Backtest panel; the IC, calibration and momentum sweep are unchanged. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -215,6 +215,42 @@ def _window_setups(
|
||||
return out
|
||||
|
||||
|
||||
def _tp_primitives(
|
||||
direction: str, entry: float, stop: float, forward: list, horizon: int
|
||||
) -> tuple[float, bool, float, float]:
|
||||
"""Primitives for the take-profit exit model, from the bars after detection.
|
||||
|
||||
Returns ``(risk_pct, stopped, mfe_pct, close_pct)``:
|
||||
- ``risk_pct`` fraction from entry to stop (the 1R distance)
|
||||
- ``stopped`` whether the stop was hit within the horizon
|
||||
- ``mfe_pct`` best favourable excursion (fraction) reachable *before* the
|
||||
stop — strictly before the stop bar, so a same-bar tp+stop
|
||||
counts as a loss (matching the conservative target model);
|
||||
over the whole horizon if the stop is never hit
|
||||
- ``close_pct`` directional return at the horizon-end close (the timeout exit)
|
||||
|
||||
From these any fixed take-profit level can be scored without re-walking bars:
|
||||
tp reached before stop (``mfe_pct >= tp``) → +tp; else stop → −1R; else the
|
||||
horizon-close move.
|
||||
"""
|
||||
long = direction == "long"
|
||||
risk_pct = abs(entry - stop) / entry if entry else 0.0
|
||||
bars = forward[:horizon]
|
||||
if not bars:
|
||||
return risk_pct, False, 0.0, 0.0
|
||||
mfe = 0.0
|
||||
stopped = False
|
||||
for r in bars:
|
||||
if (r.low <= stop) if long else (r.high >= stop):
|
||||
stopped = True
|
||||
break
|
||||
fav = (r.high - entry) / entry if long else (entry - r.low) / entry
|
||||
if fav > mfe:
|
||||
mfe = fav
|
||||
close_pct = ((bars[-1].close - entry) / entry) * (1.0 if long else -1.0)
|
||||
return risk_pct, stopped, mfe, close_pct
|
||||
|
||||
|
||||
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] = []
|
||||
@@ -240,6 +276,11 @@ def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) -
|
||||
realized_r = -1.0
|
||||
else: # expired
|
||||
realized_r = 0.0
|
||||
# Take-profit exit primitives (parallel to the target-vs-stop outcome
|
||||
# above; aggregated separately into the take-profit sweep).
|
||||
risk_pct, tp_stopped, mfe_pct, tp_close_pct = _tp_primitives(
|
||||
s["direction"], s["entry"], s["stop"], forward, HORIZON
|
||||
)
|
||||
iso = records[i].date.isocalendar()
|
||||
candidates.append({
|
||||
"symbol": symbol,
|
||||
@@ -255,6 +296,10 @@ def _replay_ticker(symbol: str, records: list, config: dict, activation: dict) -
|
||||
"outcome": outcome,
|
||||
"target_hit": target_hit,
|
||||
"realized_r": realized_r,
|
||||
"risk_pct": risk_pct,
|
||||
"tp_stopped": tp_stopped,
|
||||
"mfe_pct": mfe_pct,
|
||||
"tp_close_pct": tp_close_pct,
|
||||
})
|
||||
return candidates
|
||||
|
||||
@@ -276,6 +321,39 @@ def _bucket_stats(cands: list[dict]) -> dict:
|
||||
}
|
||||
|
||||
|
||||
# Fixed take-profit levels (fractions) swept for the take-profit exit model.
|
||||
TP_LEVELS = (0.04, 0.06, 0.08, 0.10, 0.12, 0.15)
|
||||
|
||||
|
||||
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
|
||||
reached before the stop, else −1R on a stop, else exit at the horizon close.
|
||||
Results are in R (gain% / risk%) so they're comparable to the target model.
|
||||
``hit_rate`` here = share that reached +tp before the stop (the MFE CDF)."""
|
||||
rs: list[float] = []
|
||||
wins = 0
|
||||
for c in cands:
|
||||
risk = c.get("risk_pct") or 0.0
|
||||
if risk <= 0:
|
||||
continue
|
||||
if c.get("mfe_pct", 0.0) >= tp:
|
||||
rs.append(tp / risk)
|
||||
wins += 1
|
||||
elif c.get("tp_stopped"):
|
||||
rs.append(-1.0)
|
||||
else:
|
||||
rs.append((c.get("tp_close_pct", 0.0)) / risk)
|
||||
total = len(rs)
|
||||
return {
|
||||
"tp_pct": round(tp * 100, 1),
|
||||
"total": total,
|
||||
"wins": wins,
|
||||
"hit_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] = []
|
||||
@@ -710,6 +788,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],
|
||||
"calibration": _calibration(candidates),
|
||||
"signal_eval": _signal_evaluation(collected),
|
||||
"signal_eval_note": (
|
||||
|
||||
Reference in New Issue
Block a user