research(backtest): breakeven-stop exit test, rejected

Pre-registered test of raising the stop to entry once a trade is +1R or
+2R, on top of the 3x ATR trail. be_1r fails the frozen rule (train worse
on all three graded metrics, test drawdown worse); be_2r changes 2 of 480
trades because the trail already reaches entry near +2R. The production
exit stays.

Breakeven exits get their own breakeven_stop label. They follow live
re-entry semantics (only initial-stop exits lock re-entry), and an
ungraded be_1r_gated diagnostic arm sends them through the post-stop gate
reset instead. The trail and breakeven ratchet share one helper, so
breakeven also applies on a next_open fill bar. breakeven_r is validated,
verdicts flag passes that are only ties, and the graded capacity comes
from the shadow book config. Every graded number and the parity row are
unchanged from the first run.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/services/backtest_service.py

@@ -30,6 +30,7 @@ Environment variables (see also run_backtest_snapshot.py):
     BACKTEST_FALLBACK_CLEAR_AIR_ONLY=1
     BACKTEST_RESEARCH_EXITS=1
     BACKTEST_MIN_RR_SWEEP=1
+    BACKTEST_BREAKEVEN_STUDY=1          # docs/research/breakeven-stop.md
 
   Broad-universe signal research (local snapshots only; inert when unset):
     BACKTEST_LIQUID_BREADTH=1500   # PIT top-N by 63d median $vol, price floor
@@ -2298,6 +2299,8 @@ def _simulate_portfolio(
     corr_lookback: int = 120,
     corr_action: str = "skip",
     corr_min_overlap: int = 60,
+    breakeven_r: float | None = None,
+    breakeven_gate_reset: bool = False,
 ) -> dict | None:
     """Replay the qualified setups as ONE capital-constrained book and report
     portfolio economics from the daily equity curve (return, CAGR, drawdown,
@@ -2330,6 +2333,14 @@ def _simulate_portfolio(
     caps skip or half-size candidates whose max pairwise 120d return correlation
     with open holdings exceeds ``corr_max``.
 
+    ``breakeven_r`` (research only, ATR-trail policies): once the highest close
+    reaches entry + ``breakeven_r`` x initial risk, raise the stop to the entry
+    price. Like the trail, it is active from the next bar and only moves up.
+    A stop it set exits as "breakeven_stop". Live gate locks key on initial-stop
+    exits only, so by default that exit, like a trailing stop, neither starts
+    the cooldown nor the post-stop episode; ``breakeven_gate_reset`` makes it
+    do both, as an initial stop would.
+
     Returns None when there is nothing to trade. ``cost_per_side`` is charged on
     entry and exit and therefore changes both cash availability and subsequent
     position sizing.
@@ -2347,6 +2358,13 @@ def _simulate_portfolio(
         raise ValueError("corr_action must be 'skip' or 'half_size'")
     if vol_target is not None and vol_target <= 0:
         raise ValueError("vol_target must be positive when set")
+    if breakeven_r is not None:
+        if not math.isfinite(breakeven_r) or breakeven_r <= 0:
+            raise ValueError("breakeven_r must be positive and finite when set")
+        if exit_policy not in ("atr_trail3", "atr_trail3_target"):
+            raise ValueError("breakeven_r only applies to the ATR-trail exit policies")
+    if breakeven_gate_reset and breakeven_r is None:
+        raise ValueError("breakeven_gate_reset needs breakeven_r")
     clamp_lo, clamp_hi = float(vol_clamp[0]), float(vol_clamp[1])
     if clamp_lo <= 0 or clamp_hi < clamp_lo:
         raise ValueError("vol_clamp must satisfy 0 < lo <= hi")
@@ -2478,6 +2496,23 @@ def _simulate_portfolio(
             atr_cache[key] = None
         return atr_cache[key]
 
+    def _ratchet_stop(pos: dict, sym: str, idx: int, close: float) -> None:
+        """ATR-trail policies: raise the stop from this bar's close — the trail
+        off the highest close, then the breakeven floor. Effective from the next
+        bar; the stop never moves down."""
+        pos["highest_close"] = max(pos["highest_close"], close)
+        atr = _atr(sym, idx)
+        if atr is not None:
+            next_stop = pos["highest_close"] - atr_trail_multiplier * atr
+            if next_stop < close and next_stop > pos["stop"]:
+                pos["stop"] = next_stop
+                pos["stop_source"] = "trail"
+        if breakeven_r is not None and pos["entry"] > pos["stop"]:
+            risk = pos["entry"] - pos["initial_stop"]
+            if risk > 0 and pos["highest_close"] >= pos["entry"] + breakeven_r * risk:
+                pos["stop"] = pos["entry"]
+                pos["stop_source"] = "breakeven"
+
     def _close_trade(sym: str, fill: float, reason: str) -> dict:
         nonlocal cash
         pos = positions.pop(sym)
@@ -2520,11 +2555,12 @@ def _simulate_portfolio(
             if bar.low <= pos["stop"]:
                 # Same-bar stop+target resolves as the loss (conservative, like
                 # the evaluator); gap through the stop fills at the open.
-                reason = (
-                    "trailing_stop"
-                    if pos["stop"] > pos["initial_stop"] + 1e-9
-                    else "stop"
-                )
+                if pos["stop_source"] == "breakeven":
+                    reason = "breakeven_stop"
+                elif pos["stop"] > pos["initial_stop"] + 1e-9:
+                    reason = "trailing_stop"
+                else:
+                    reason = "stop"
                 survived_refresh = False
                 if reason == "stop" and initial_stop_refresh_fn is not None:
                     stop_refresh_attempts += 1
@@ -2545,9 +2581,12 @@ def _simulate_portfolio(
                 if not survived_refresh:
                     fill = min(pos["stop"], bar.open)
                     closed_pos = _close_trade(sym, fill, reason)
-                    if reason == "stop" and cooldown_sessions:
+                    resets_gate = reason == "stop" or (
+                        reason == "breakeven_stop" and breakeven_gate_reset
+                    )
+                    if resets_gate and cooldown_sessions:
                         cooldown_until_index[sym] = calendar_index + cooldown_sessions
-                    if reason == "stop" and post_stop_reentry_fn is not None:
+                    if resets_gate and post_stop_reentry_fn is not None:
                         post_stop_events += 1
                         post_stop_states[sym] = {
                             "stop_ord": o,
@@ -2584,12 +2623,7 @@ def _simulate_portfolio(
                 _close_trade(sym, bar.close, "time")
                 continue
             if exit_policy in ("atr_trail3", "atr_trail3_target"):
-                pos["highest_close"] = max(pos["highest_close"], bar.close)
-                atr = _atr(sym, bar.idx)
-                if atr is not None:
-                    next_stop = pos["highest_close"] - atr_trail_multiplier * atr
-                    if next_stop < bar.close:
-                        pos["stop"] = max(pos["stop"], next_stop)
+                _ratchet_stop(pos, sym, bar.idx, bar.close)
 
         # 2) entries — close-fill at signal close, or next-open fills of prior signals
         equity = _marked_equity()
@@ -2725,6 +2759,7 @@ def _simulate_portfolio(
                 "entry_ord": entry_ord,
                 "initial_stop": stop,
                 "stop": stop,
+                "stop_source": "initial",
                 "target": float(c["target"]) if c.get("target") else None,
                 "entry_cost": entry_cost,
                 "bars_held": 0,
@@ -2744,7 +2779,6 @@ def _simulate_portfolio(
             # bars_held stays 0 on the fill day (matches close-fill cadence).
             if fill_mode == FILL_MODE_NEXT_OPEN and fill_bar is not None:
                 positions[sym]["last_close"] = fill_bar.close
-                positions[sym]["highest_close"] = max(entry, fill_bar.close)
                 if fill_bar.low <= stop:
                     fill = min(stop, fill_bar.open)
                     closed = _close_trade(sym, fill, "stop")
@@ -2765,16 +2799,7 @@ def _simulate_portfolio(
                             "gate_went_unqualified": False,
                         }
                 elif exit_policy in ("atr_trail3", "atr_trail3_target"):
-                    atr = _atr(sym, fill_bar.idx)
-                    if atr is not None:
-                        next_stop = (
-                            positions[sym]["highest_close"]
-                            - atr_trail_multiplier * atr
-                        )
-                        if next_stop < fill_bar.close:
-                            positions[sym]["stop"] = max(
-                                positions[sym]["stop"], next_stop
-                            )
+                    _ratchet_stop(positions[sym], sym, fill_bar.idx, fill_bar.close)
             equity = _marked_equity()
 
         for c in fill_candidates:
@@ -3059,6 +3084,17 @@ def _simulate_portfolio(
     if max_entry_gap_pct is not None:
         result["max_entry_gap_pct"] = max_entry_gap_pct
         result["skipped_gap_cap"] = skipped_gap_cap
+    if breakeven_r is not None:
+        result["breakeven_r"] = breakeven_r
+        result["breakeven_gate_reset"] = breakeven_gate_reset
+        breakeven_exit_keys = {
+            (t["symbol"], t["exit_ord"]) for t in trades if t["reason"] == "breakeven_stop"
+        }
+        result["breakeven_exits"] = len(breakeven_exit_keys)
+        # Bought back in the session it was stopped out of breakeven.
+        result["breakeven_same_session_reentries"] = sum(
+            1 for t in trades if (t["symbol"], t["entry_ord"]) in breakeven_exit_keys
+        )
     if curve_payload is not None:
         result["equity_curve"] = curve_payload
     if benchmark_payload is not None:
@@ -3806,6 +3842,231 @@ def _holdout_evaluation(
     }
 
 
+# Pre-registered 2026-10-09 — docs/research/breakeven-stop.md. Two trigger levels,
+# no sweep; graded on both halves of a fixed split at the live book capacity.
+BREAKEVEN_STUDY_ARMS: tuple[tuple[str, float | None], ...] = (
+    ("control", None),
+    ("be_1r", 1.0),
+    ("be_2r", 2.0),
+)
+# Added 2026-10-10 after the first run, so reported but never graded: be_1r with its
+# breakeven exits starting the post-stop gate reset, as an initial stop does. The
+# graded arms follow live, where only initial-stop exits lock re-entry; the gap
+# between the two shows how much of be_1r is same-session re-buys.
+BREAKEVEN_STUDY_DIAGNOSTIC_ARMS: tuple[tuple[str, float], ...] = (("be_1r_gated", 1.0),)
+BREAKEVEN_STUDY_SPLIT = date(2024, 7, 1)
+_BREAKEVEN_PARITY_KEYS = ("sharpe", "cagr_pct", "max_drawdown_pct", "trades")
+_BREAKEVEN_GRADED_KEYS = ("sharpe", "cagr_pct", "max_drawdown_pct")
+
+
+def _breakeven_study_enabled() -> bool:
+    return os.getenv("BACKTEST_BREAKEVEN_STUDY", "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _breakeven_checks(arm: dict, control: dict) -> dict[str, bool]:
+    """The frozen pass rule for one window: no worse on Sharpe, CAGR or drawdown.
+    A missing metric fails — an arm cannot pass on numbers it did not produce."""
+
+    def ok(key: str, better) -> bool:
+        a, c = arm.get(key), control.get(key)
+        return a is not None and c is not None and better(a, c)
+
+    return {
+        "sharpe": ok("sharpe", lambda a, c: a >= c),
+        "cagr": ok("cagr_pct", lambda a, c: a >= c),
+        "max_drawdown": ok("max_drawdown_pct", lambda a, c: a <= c),
+    }
+
+
+def _breakeven_exit_study(
+    candidates: list[dict],
+    prices: dict[str, tuple],
+    _spy_closes: dict[date, float] | None,
+    hold_days: int,
+    live_exit_policy: dict | None = None,
+    cadence: str = DEFAULT_BACKTEST_CADENCE,
+    monitor_report: dict | None = None,
+    live_capacity: int | None = None,
+) -> dict:
+    """Production book with a breakeven stop added to the ATR trail, vs control.
+
+    Mirrors ``_holdout_evaluation``'s production configuration and changes only
+    ``breakeven_r`` and, for every graded row, capacity: ``live_capacity`` (the
+    shadow book's setting) when the caller could read it, else
+    ``SIM_MAX_POSITIONS``, rather than the monitor's 10. A capacity-10 control on
+    full history must reproduce the monitor's production row, or the study is not
+    grading the production book.
+    """
+    strategy = next(
+        (s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")), None
+    )
+    if strategy is None:
+        return {}
+    entry_cfg = _entry_variant_config(str(strategy["entry_variant"]))
+    if entry_cfg is None:
+        return {}
+
+    exit_policy = str(strategy["exit_policy"])
+    row_hold_days = hold_days
+    trail_multiplier = ATR_TRAIL_MULTIPLIER
+    reentry_policy = str(strategy.get("reentry_policy", "immediate"))
+    if strategy.get("use_live_config") and live_exit_policy is not None:
+        exit_policy = LIVE_EXIT_MODE_TO_SIM.get(
+            str(live_exit_policy.get("mode", "atr_trailing")), "atr_trail3"
+        )
+        row_hold_days = int(live_exit_policy.get("hold_days", hold_days))
+        trail_multiplier = float(
+            live_exit_policy.get("atr_multiplier", ATR_TRAIL_MULTIPLIER)
+        )
+    if exit_policy != "atr_trail3":
+        return {"error": f"live exit maps to {exit_policy!r}; the study needs atr_trail3"}
+    qualified_fn = (
+        None if strategy.get("use_live_config")
+        else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config)
+    )
+    ranking_key = str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"])
+    post_stop_reentry_fn = (
+        _make_gate_reset_reentry_fn(
+            candidates,
+            prices,
+            cadence=cadence,
+            qualified_fn=qualified_fn,
+            ranking_key=ranking_key,
+        )
+        if reentry_policy == "gate_reset"
+        else None
+    )
+
+    def run(
+        breakeven_r: float | None,
+        max_positions: int,
+        start: date | None = None,
+        end: date | None = None,
+        gate_reset: bool = False,
+    ) -> dict | None:
+        sim = _simulate_portfolio(
+            candidates,
+            prices,
+            _spy_closes,
+            exit_policy,
+            row_hold_days,
+            qualified_fn=qualified_fn,
+            ranking_key=ranking_key,
+            max_positions=max_positions,
+            risk_per_trade=float(entry_cfg["risk_per_trade"]),
+            atr_trail_multiplier=trail_multiplier,
+            post_stop_reentry_fn=post_stop_reentry_fn,
+            start_date=start,
+            end_date=end,
+            include_trades=True,
+            breakeven_r=breakeven_r,
+            breakeven_gate_reset=gate_reset,
+        )
+        if sim is None:
+            return None
+        rs = [t["r"] for t in sim.pop("trade_details", [])]
+        sim.pop("reentry_events", None)
+        sim["avg_r"] = round(sum(rs) / len(rs), 3) if rs else None
+        sim["trades_5r_plus"] = sum(1 for r in rs if r >= 5.0)
+        return sim
+
+    capacity = int(live_capacity) if live_capacity is not None else SIM_MAX_POSITIONS
+    split = BREAKEVEN_STUDY_SPLIT
+    windows = (("full", None, None), ("train", None, split), ("test", split, None))
+    rows: list[dict] = []
+    for arm, breakeven_r in BREAKEVEN_STUDY_ARMS:
+        for window, start, end in windows:
+            sim = run(breakeven_r, capacity, start, end)
+            if sim is not None:
+                rows.append({"arm": arm, "window": window, **sim})
+    diagnostic_rows: list[dict] = []
+    for arm, breakeven_r in BREAKEVEN_STUDY_DIAGNOSTIC_ARMS:
+        for window, start, end in windows:
+            sim = run(breakeven_r, capacity, start, end, gate_reset=True)
+            if sim is not None:
+                diagnostic_rows.append({"arm": arm, "window": window, **sim})
+
+    parity_row = run(None, int(entry_cfg["max_positions"]))
+    monitor_row = next(
+        (
+            r for r in (monitor_report or {}).get("runs") or []
+            if r.get("is_production") and r.get("lookback") == "all"
+        ),
+        None,
+    )
+    parity_ok = (
+        parity_row is not None
+        and monitor_row is not None
+        and all(parity_row.get(k) == monitor_row.get(k) for k in _BREAKEVEN_PARITY_KEYS)
+    )
+
+    def row(arm: str, window: str) -> dict | None:
+        return next((r for r in rows if r["arm"] == arm and r["window"] == window), None)
+
+    def same_as_control(arm: str, window: str) -> bool:
+        treated, control = row(arm, window), row("control", window)
+        return (
+            treated is not None
+            and control is not None
+            and all(treated.get(k) == control.get(k) for k in _BREAKEVEN_GRADED_KEYS)
+        )
+
+    verdicts: dict[str, dict] = {}
+    for arm, _ in BREAKEVEN_STUDY_ARMS[1:]:
+        verdict: dict[str, Any] = {}
+        for window in ("train", "test"):
+            treated, control = row(arm, window), row("control", window)
+            verdict[window] = (
+                _breakeven_checks(treated, control)
+                if treated is not None and control is not None
+                else {"sharpe": False, "cagr": False, "max_drawdown": False}
+            )
+        verdict["passes"] = parity_ok and all(
+            all(verdict[w].values()) for w in ("train", "test")
+        )
+        # The frozen rule passes ties. An arm that matches control on every
+        # graded metric in both windows is the control exit, not a finding.
+        verdict["identical_to_control"] = all(
+            same_as_control(arm, w) for w in ("train", "test")
+        )
+        verdicts[arm] = verdict
+
+    return {
+        "protocol": "docs/research/breakeven-stop.md",
+        "split_date": split.isoformat(),
+        "capacity": capacity,
+        "capacity_source": (
+            "shadow_book_capacity" if live_capacity is not None else "SIM_MAX_POSITIONS"
+        ),
+        "cadence": cadence,
+        "exit_policy": exit_policy,
+        "atr_trail_multiplier": trail_multiplier,
+        "hold_days": row_hold_days,
+        "reentry_policy": reentry_policy,
+        "parity": {
+            "ok": parity_ok,
+            "capacity": int(entry_cfg["max_positions"]),
+            "study_control": (
+                {k: parity_row.get(k) for k in _BREAKEVEN_PARITY_KEYS} if parity_row else None
+            ),
+            "monitor_production_all": (
+                {k: monitor_row.get(k) for k in _BREAKEVEN_PARITY_KEYS} if monitor_row else None
+            ),
+        },
+        "rows": rows,
+        "verdicts": verdicts,
+        "diagnostic_rows": diagnostic_rows,
+        "note": (
+            "Graded on train and test only: an arm passes when Sharpe and CAGR are "
+            "no lower and max drawdown no higher than control in both windows; "
+            "identical_to_control marks a pass that is only ties. Win rate and avg R "
+            "are reported, not graded. diagnostic_rows were added after the first run "
+            "and are never graded. A pass earns forward paper validation, not a "
+            "production change."
+        ),
+    }
+
+
 def _portfolio_monitor(
     candidates: list[dict],
     prices: dict[str, tuple],
@@ -4508,6 +4769,7 @@ async def run_backtest(
     portfolio_monitor_report: dict | None = None
     holdout_report: dict | None = None
     min_rr_sweep_report: dict | None = None
+    breakeven_study_report: dict | None = None
     if not _signal_eval_only():
         try:
             qual_symbols = sorted({
@@ -4572,6 +4834,25 @@ async def run_backtest(
                     candidates, price_columns, spy_closes, activation, current_min_pct,
                     hold_horizon, live_exit_policy=live_exit_policy, cadence=cadence,
                 )
+            if _breakeven_study_enabled():
+                live_capacity: int | None = None
+                try:
+                    from app.services.shadow_book_service import get_config
+
+                    live_capacity = int((await get_config(db))["capacity"])
+                except Exception:
+                    logger.exception(
+                        "Shadow book capacity load failed; breakeven study uses "
+                        "SIM_MAX_POSITIONS"
+                    )
+                    await _rollback_quietly(db, "shadow book capacity load")
+                breakeven_study_report = _breakeven_exit_study(
+                    candidates, price_columns, spy_closes, hold_horizon,
+                    live_exit_policy=live_exit_policy,
+                    cadence=cadence,
+                    monitor_report=portfolio_monitor_report,
+                    live_capacity=live_capacity,
+                )
         except Exception:
             logger.exception("Portfolio simulation failed")
             # Catches the price_columns fetch loop, which has no handler of its
@@ -4703,6 +4984,9 @@ async def run_backtest(
             "~6 months ≈ one market regime — treat as directional, not gospel."
         ),
     }
+    if breakeven_study_report is not None:
+        # Added only when opted in, so the default report keeps its exact shape.
+        report["breakeven_stop_study"] = breakeven_study_report
     report["recommendation"] = _build_recommendation(report)
     report["research_recommendation"] = _build_research_recommendation(report)
     return report