From 70d17c1491cd95799d7eee5dad06035ff61e4a4f Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Sat, 10 Oct 2026 12:17:27 +0200 Subject: [PATCH] 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 --- app/services/backtest_service.py | 332 ++++++++++++++----------------- 1 file changed, 151 insertions(+), 181 deletions(-) diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 2d0b4ab..dcfca57 100644 --- a/app/services/backtest_service.py +++ b/app/services/backtest_service.py @@ -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,