refactor(backtest): resolve the production book config in one place

The portfolio monitor, holdout evaluation, min-R:R sweep and breakeven study each rebuilt the strategy's exit, hold, trail, gate, ranking and re-entry settings by hand, and the copies had drifted: only the monitor honored a strategy-level ranking_key. They now share _strategy_book_config / _production_book_config, which follow the monitor's rule. No output changes. On the prod snapshot (daily cadence, holdout split, min-R:R sweep and breakeven study all on), all four functions return byte-identical JSON to the previous commit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-10 12:17:27 +02:00
1 parent 3545b483d8
commit 70d17c1491
@@ -3535,6 +3535,96 @@ def _entry_variant_config(variant: str) -> dict | None:
    return next((cfg for cfg in STRATEGY_VARIANTS if cfg["variant"] == variant), None)


+@dataclass(frozen=True)
+class _BookConfig:
+    """How one ``PORTFOLIO_MONITOR_STRATEGIES`` entry is replayed. The monitor and
+    every production-book study resolve it here, so they cannot drift apart."""
+
+    strategy: dict
+    entry_cfg: dict
+    exit_policy: str
+    live_exit_mode: str | None
+    hold_days: int
+    trail_multiplier: float
+    reentry_policy: str
+    qualified_fn: Callable[[dict], bool] | None
+    ranking_key: str
+    max_positions: int
+    risk_per_trade: float
+
+    def post_stop_reentry_fn(
+        self,
+        candidates: list[dict],
+        prices: dict[str, tuple],
+        cadence: str,
+        qualified_fn: Callable[[dict], bool] | None,
+    ) -> Callable[[str, int, dict, Any], dict | None] | None:
+        """The gate-reset callback for this book, built on ``qualified_fn`` (a
+        sweep passes its own gate), or None for immediate re-entry."""
+        if self.reentry_policy != "gate_reset":
+            return None
+        return _make_gate_reset_reentry_fn(
+            candidates,
+            prices,
+            cadence=cadence,
+            qualified_fn=qualified_fn,
+            ranking_key=self.ranking_key,
+        )
+
+
+def _strategy_book_config(
+    strategy: dict, hold_days: int, live_exit_policy: dict | None
+) -> _BookConfig | None:
+    entry_cfg = _entry_variant_config(str(strategy["entry_variant"]))
+    if entry_cfg is None:
+        return None
+    # Live-config rows replay the runtime qualification flag and Admin exit
+    # policy; research rows replay their frozen variant gate and exit.
+    use_live = bool(strategy.get("use_live_config"))
+    exit_policy = str(strategy["exit_policy"])
+    row_hold_days = hold_days
+    trail_multiplier = ATR_TRAIL_MULTIPLIER
+    live_exit_mode: str | None = None
+    if use_live and live_exit_policy is not None:
+        live_exit_mode = str(live_exit_policy.get("mode", "atr_trailing"))
+        exit_policy = LIVE_EXIT_MODE_TO_SIM.get(live_exit_mode, "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)
+        )
+    return _BookConfig(
+        strategy=strategy,
+        entry_cfg=entry_cfg,
+        exit_policy=exit_policy,
+        live_exit_mode=live_exit_mode,
+        hold_days=row_hold_days,
+        trail_multiplier=trail_multiplier,
+        reentry_policy=str(strategy.get("reentry_policy", "immediate")),
+        qualified_fn=(
+            None if use_live
+            else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config)
+        ),
+        ranking_key=str(
+            strategy.get("ranking_key")
+            or entry_cfg.get("ranking_key")
+            or entry_cfg["percentile_key"]
+        ),
+        max_positions=int(entry_cfg["max_positions"]),
+        risk_per_trade=float(entry_cfg["risk_per_trade"]),
+    )
+
+
+def _production_book_config(
+    hold_days: int, live_exit_policy: dict | None
+) -> _BookConfig | None:
+    strategy = next(
+        (s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")), None
+    )
+    if strategy is None:
+        return None
+    return _strategy_book_config(strategy, hold_days, live_exit_policy)
+
+
 def _exit_policy_sims(
    candidates: list[dict],
    prices: dict[str, tuple],
@@ -3632,23 +3722,9 @@ def _min_rr_sweep(
    Graded on Sharpe/CAGR/DD under the real exit — not per-setup expectancy —
    because that is the metric every other promotion decision used.
    """
-    strategy = next((s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")), None)
-    if strategy is None:
+    book = _production_book_config(hold_days, live_exit_policy)
+    if book 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))

    live_min_rr = float(activation.get("min_rr", 0.0))
    live_qualified = sum(1 for c in candidates if c.get("qualified"))
@@ -3675,25 +3751,15 @@ def _min_rr_sweep(
            candidates,
            prices,
            _spy_closes,
-            exit_policy,
-            row_hold_days,
+            book.exit_policy,
+            book.hold_days,
            qualified_fn=qualified_fn,
-            ranking_key=str(entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]),
-            max_positions=int(entry_cfg["max_positions"]),
-            risk_per_trade=float(entry_cfg["risk_per_trade"]),
-            atr_trail_multiplier=trail_multiplier,
-            post_stop_reentry_fn=(
-                _make_gate_reset_reentry_fn(
-                    candidates,
-                    prices,
-                    cadence=cadence,
-                    qualified_fn=qualified_fn,
-                    ranking_key=str(
-                        entry_cfg.get("ranking_key") or entry_cfg["percentile_key"]
-                    ),
-                )
-                if reentry_policy == "gate_reset"
-                else None
+            ranking_key=book.ranking_key,
+            max_positions=book.max_positions,
+            risk_per_trade=book.risk_per_trade,
+            atr_trail_multiplier=book.trail_multiplier,
+            post_stop_reentry_fn=book.post_stop_reentry_fn(
+                candidates, prices, cadence, qualified_fn
            ),
            start_date=sweep_start,
        )
@@ -3718,8 +3784,8 @@ def _min_rr_sweep(
        "live_min_rr": live_min_rr,
        "live_qualified_setups": live_qualified,
        "reproduces_production_gate": reproduces,
-        "exit_policy": exit_policy,
-        "reentry_policy": reentry_policy,
+        "exit_policy": book.exit_policy,
+        "reentry_policy": book.reentry_policy,
        "entries_from": sweep_start.isoformat() if sweep_start else None,
        "window": "out-of-sample (test)" if sweep_start else "full history (in-sample)",
        "rows": rows,
@@ -3763,45 +3829,11 @@ def _holdout_evaluation(
    chosen on. This does the real thing: the test window is disjoint from the
    train window, so a rule settled on train has never seen it.
    """
-    strategy = next(
-        (s for s in PORTFOLIO_MONITOR_STRATEGIES if s.get("is_production")),
-        None,
-    )
-    if strategy is None:
+    book = _production_book_config(hold_days, live_exit_policy)
+    if book 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)
-        )
-    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
+    post_stop_reentry_fn = book.post_stop_reentry_fn(
+        candidates, prices, cadence, book.qualified_fn
    )

    rows: list[dict] = []
@@ -3813,13 +3845,13 @@ def _holdout_evaluation(
            candidates,
            prices,
            _spy_closes,
-            exit_policy,
-            row_hold_days,
-            qualified_fn=qualified_fn,
-            ranking_key=ranking_key,
-            max_positions=int(entry_cfg["max_positions"]),
-            risk_per_trade=float(entry_cfg["risk_per_trade"]),
-            atr_trail_multiplier=trail_multiplier,
+            book.exit_policy,
+            book.hold_days,
+            qualified_fn=book.qualified_fn,
+            ranking_key=book.ranking_key,
+            max_positions=book.max_positions,
+            risk_per_trade=book.risk_per_trade,
+            atr_trail_multiplier=book.trail_multiplier,
            post_stop_reentry_fn=post_stop_reentry_fn,
            start_date=start,
            end_date=end,
@@ -3827,12 +3859,12 @@ def _holdout_evaluation(
        )
        if sim is None:
            continue
-        rows.append({"window": window, "exit_policy": exit_policy, **sim})
+        rows.append({"window": window, "exit_policy": book.exit_policy, **sim})

    return {
        "split_date": split.isoformat(),
-        "strategy": strategy["strategy"],
-        "reentry_policy": reentry_policy,
+        "strategy": book.strategy["strategy"],
+        "reentry_policy": book.reentry_policy,
        "rows": rows,
        "note": (
            "Train = entries before the split; test = entries on/after it. The two "
@@ -3897,44 +3929,15 @@ def _breakeven_exit_study(
    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:
+    book = _production_book_config(hold_days, live_exit_policy)
+    if book 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
+    if book.exit_policy != "atr_trail3":
+        return {
+            "error": f"live exit maps to {book.exit_policy!r}; the study needs atr_trail3"
+        }
+    post_stop_reentry_fn = book.post_stop_reentry_fn(
+        candidates, prices, cadence, book.qualified_fn
    )

    def run(
@@ -3948,13 +3951,13 @@ def _breakeven_exit_study(
            candidates,
            prices,
            _spy_closes,
-            exit_policy,
-            row_hold_days,
-            qualified_fn=qualified_fn,
-            ranking_key=ranking_key,
+            book.exit_policy,
+            book.hold_days,
+            qualified_fn=book.qualified_fn,
+            ranking_key=book.ranking_key,
            max_positions=max_positions,
-            risk_per_trade=float(entry_cfg["risk_per_trade"]),
-            atr_trail_multiplier=trail_multiplier,
+            risk_per_trade=book.risk_per_trade,
+            atr_trail_multiplier=book.trail_multiplier,
            post_stop_reentry_fn=post_stop_reentry_fn,
            start_date=start,
            end_date=end,
@@ -3986,7 +3989,7 @@ def _breakeven_exit_study(
            if sim is not None:
                diagnostic_rows.append({"arm": arm, "window": window, **sim})

-    parity_row = run(None, int(entry_cfg["max_positions"]))
+    parity_row = run(None, book.max_positions)
    monitor_row = next(
        (
            r for r in (monitor_report or {}).get("runs") or []
@@ -4039,13 +4042,13 @@ def _breakeven_exit_study(
            "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,
+        "exit_policy": book.exit_policy,
+        "atr_trail_multiplier": book.trail_multiplier,
+        "hold_days": book.hold_days,
+        "reentry_policy": book.reentry_policy,
        "parity": {
            "ok": parity_ok,
-            "capacity": int(entry_cfg["max_positions"]),
+            "capacity": book.max_positions,
            "study_control": (
                {k: parity_row.get(k) for k in _BREAKEVEN_PARITY_KEYS} if parity_row else None
            ),
@@ -4079,44 +4082,11 @@ def _portfolio_monitor(
    rows: list[dict] = []
    strategies = _portfolio_monitor_strategies()
    for strategy in strategies:
-        entry_cfg = _entry_variant_config(str(strategy["entry_variant"]))
-        if entry_cfg is None:
+        book = _strategy_book_config(strategy, hold_days, live_exit_policy)
+        if book is None:
            continue
-        ranking_key = str(
-            strategy.get("ranking_key")
-            or entry_cfg.get("ranking_key")
-            or entry_cfg["percentile_key"]
-        )
-        # Live-config rows replay the runtime qualification flag and Admin exit
-        # policy. The overlay opts into this deliberately so only ordering
-        # changes relative to the production row.
-        use_live = bool(strategy.get("use_live_config"))
-        reentry_policy = str(strategy.get("reentry_policy", "immediate"))
-        exit_policy = str(strategy["exit_policy"])
-        row_hold_days = hold_days
-        trail_multiplier = ATR_TRAIL_MULTIPLIER
-        live_exit_mode: str | None = None
-        if use_live and live_exit_policy is not None:
-            live_exit_mode = str(live_exit_policy.get("mode", "atr_trailing"))
-            exit_policy = LIVE_EXIT_MODE_TO_SIM.get(live_exit_mode, "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)
-            )
-        qualified_fn = (
-            None if use_live
-            else lambda c, config=entry_cfg: _qualifies_strategy_variant(c, config)
-        )
-        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
+        post_stop_reentry_fn = book.post_stop_reentry_fn(
+            candidates, prices, cadence, book.qualified_fn
        )
        for lookback in PORTFOLIO_MONITOR_LOOKBACKS:
            start = _lookback_start(latest_ord, lookback["days"])
@@ -4124,13 +4094,13 @@ def _portfolio_monitor(
                candidates,
                prices,
                _spy_closes,
-                exit_policy,
-                row_hold_days,
-                qualified_fn=qualified_fn,
-                ranking_key=ranking_key,
-                max_positions=int(entry_cfg["max_positions"]),
-                risk_per_trade=float(entry_cfg["risk_per_trade"]),
-                atr_trail_multiplier=trail_multiplier,
+                book.exit_policy,
+                book.hold_days,
+                qualified_fn=book.qualified_fn,
+                ranking_key=book.ranking_key,
+                max_positions=book.max_positions,
+                risk_per_trade=book.risk_per_trade,
+                atr_trail_multiplier=book.trail_multiplier,
                post_stop_reentry_fn=post_stop_reentry_fn,
                start_date=start,
                include_curve=True,
@@ -4144,10 +4114,10 @@ def _portfolio_monitor(
                "is_production": bool(strategy.get("is_production")),
                "comparison_arm": strategy.get("comparison_arm"),
                "entry_variant": strategy["entry_variant"],
-                "ranking_key": ranking_key,
-                "exit_policy": exit_policy,
-                "live_exit_mode": live_exit_mode,
-                "reentry_policy": reentry_policy,
+                "ranking_key": book.ranking_key,
+                "exit_policy": book.exit_policy,
+                "live_exit_mode": book.live_exit_mode,
+                "reentry_policy": book.reentry_policy,
                "lookback": lookback["lookback"],
                "lookback_label": lookback["label"],
                **sim,