feat(risk): stamp consecutive_losses before strategy evaluate

Resolve trailing loss streak from live equity history into portfolio
metadata so entrypoint risk gates can trip the circuit breaker.

Co-Authored-By: Claude <noreply@anthropic.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 3 people

application/rebalance_service.py

@@ -58,7 +58,13 @@ def _record_platform_execution_telemetry(
             "orders_filled": list(summary.get("orders_filled") or ()),
             "orders_skipped": list(summary.get("orders_skipped") or ()),
             "trade_date": summary.get("trade_date") or metadata.get("trade_date"),
-            "total_equity": summary.get("equity") or metadata.get("equity"),
+            "total_equity": (
+                summary.get("equity")
+                or summary.get("total_equity")
+                or metadata.get("equity")
+                or metadata.get("portfolio_total_equity")
+                or metadata.get("total_equity")
+            ),
         },
         domain=str(metadata.get("domain") or ""),
     )

strategy_runtime.py

@@ -185,6 +185,46 @@ def _attach_strategy_plugin_metadata(
             return portfolio_snapshot
         return attach_strategy_plugin_metadata(portfolio_snapshot, tuple(strategy_plugin_signals or ()))
 
+    def _with_consecutive_loss_metadata(self, portfolio_snapshot: Any | None) -> Any | None:
+        """Stamp trailing consecutive_losses onto portfolio metadata before evaluate."""
+        if portfolio_snapshot is None:
+            return None
+        metadata = dict(getattr(portfolio_snapshot, "metadata", None) or {})
+        if metadata.get("consecutive_losses") is not None:
+            return portfolio_snapshot
+        try:
+            from quant_platform_kit.strategy_lifecycle.live_equity import resolve_consecutive_losses
+            from quant_platform_kit.strategy_lifecycle.performance_monitor import infer_strategy_domain
+
+            streak = resolve_consecutive_losses(
+                domain=infer_strategy_domain(self.profile),
+                strategy_profile=self.profile,
+            )
+        except Exception as exc:
+            self.logger(
+                "strategy_consecutive_losses_resolve_failed | "
+                f"profile={self.profile} error_type={type(exc).__name__} error={exc}"
+            )
+            return portfolio_snapshot
+        if streak is None:
+            return portfolio_snapshot
+        metadata["consecutive_losses"] = int(streak)
+        return replace(portfolio_snapshot, metadata=metadata)
+
+    def _prepare_portfolio_snapshot(
+        self,
+        portfolio_snapshot: Any | None,
+        strategy_symbols=(),
+        strategy_plugin_signals=(),
+        *,
+        project: bool = True,
+    ) -> Any | None:
+        snapshot = portfolio_snapshot
+        if project:
+            snapshot = self._project_portfolio_snapshot(snapshot, strategy_symbols)
+        snapshot = self._attach_strategy_plugin_metadata(snapshot, strategy_plugin_signals)
+        return self._with_consecutive_loss_metadata(snapshot)
+
     @staticmethod
     def _normalize_symbols(symbols) -> tuple[str, ...]:
         normalized = []
@@ -680,11 +720,11 @@ def _evaluate_direct_market_data_strategy(
             ib,
             required=False,
         )
-        portfolio_snapshot = self._project_portfolio_snapshot(
+        portfolio_snapshot = self._prepare_portfolio_snapshot(
             portfolio_snapshot,
             self._configured_strategy_symbols(include_ranking_pool=True),
+            strategy_plugin_signals,
         )
-        portfolio_snapshot = self._attach_strategy_plugin_metadata(portfolio_snapshot, strategy_plugin_signals)
         option_chains = self._fetch_option_chains_for_runtime(ib, runtime_config, portfolio_snapshot)
         if option_chains:
             runtime_config["option_chains"] = option_chains
@@ -760,11 +800,11 @@ def _evaluate_market_data_strategy(
             ib,
             required=requires_portfolio,
         )
-        portfolio_snapshot = self._project_portfolio_snapshot(
+        portfolio_snapshot = self._prepare_portfolio_snapshot(
             portfolio_snapshot,
             self._configured_strategy_symbols(include_ranking_pool=True),
+            strategy_plugin_signals,
         )
-        portfolio_snapshot = self._attach_strategy_plugin_metadata(portfolio_snapshot, strategy_plugin_signals)
         option_chains = self._fetch_option_chains_for_runtime(ib, runtime_config, portfolio_snapshot)
         if option_chains:
             runtime_config["option_chains"] = option_chains
@@ -835,8 +875,11 @@ def _evaluate_value_target_strategy(
         apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)
         managed_symbols = self._configured_strategy_symbols()
         portfolio_snapshot = self._fetch_portfolio_snapshot_for_context(ib, required=True)
-        portfolio_snapshot = self._project_portfolio_snapshot(portfolio_snapshot, managed_symbols)
-        portfolio_snapshot = self._attach_strategy_plugin_metadata(portfolio_snapshot, strategy_plugin_signals)
+        portfolio_snapshot = self._prepare_portfolio_snapshot(
+            portfolio_snapshot,
+            managed_symbols,
+            strategy_plugin_signals,
+        )
         option_chains = self._fetch_option_chains_for_runtime(ib, runtime_config, portfolio_snapshot)
         if option_chains:
             runtime_config["option_chains"] = option_chains
@@ -965,7 +1008,11 @@ def build_available_inputs(feature_snapshot) -> Mapping[str, Any]:
                 ib,
                 required=requires_portfolio,
             )
-            portfolio_snapshot = self._attach_strategy_plugin_metadata(portfolio_snapshot, strategy_plugin_signals)
+            portfolio_snapshot = self._prepare_portfolio_snapshot(
+                portfolio_snapshot,
+                strategy_plugin_signals=strategy_plugin_signals,
+                project=False,
+            )
             if portfolio_snapshot is not None:
                 portfolio_snapshot_holder["portfolio_snapshot"] = portfolio_snapshot
             option_chains = self._fetch_option_chains_for_runtime(ib, runtime_config, portfolio_snapshot)

tests/test_strategy_runtime.py

@@ -1221,3 +1221,85 @@ def evaluate(self, ctx):
     assert enriched["unrealized_pnl_pct"] == -0.05
     assert enriched["consecutive_losses"] == 2
     assert enriched["portfolio_total_equity"] == 10_000.0
+
+
+def test_with_consecutive_loss_metadata_stamps_from_resolver(monkeypatch):
+    from datetime import datetime, timezone
+
+    class FakeEntrypoint:
+        manifest = StrategyManifest(
+            profile="global_etf_rotation",
+            domain="us_equity",
+            display_name="Global ETF",
+            description="test",
+            required_inputs=frozenset(),
+        )
+
+        def evaluate(self, ctx):
+            return StrategyDecision(positions=())
+
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=FakeEntrypoint(),
+        runtime_adapter=StrategyRuntimeAdapter(),
+        runtime_settings=_build_runtime_settings("global_etf_rotation"),
+        logger=lambda _msg: None,
+    )
+    monkeypatch.setattr(
+        "quant_platform_kit.strategy_lifecycle.live_equity.resolve_consecutive_losses",
+        lambda **_kwargs: 4,
+    )
+    snapshot = PortfolioSnapshot(
+        as_of=datetime.now(timezone.utc),
+        total_equity=10_000.0,
+        positions=(),
+        metadata={},
+    )
+
+    stamped = runtime._with_consecutive_loss_metadata(snapshot)
+
+    assert stamped is not snapshot
+    assert stamped.metadata["consecutive_losses"] == 4
+
+
+def test_with_consecutive_loss_metadata_preserves_explicit_value(monkeypatch):
+    from datetime import datetime, timezone
+
+    class FakeEntrypoint:
+        manifest = StrategyManifest(
+            profile="global_etf_rotation",
+            domain="us_equity",
+            display_name="Global ETF",
+            description="test",
+            required_inputs=frozenset(),
+        )
+
+        def evaluate(self, ctx):
+            return StrategyDecision(positions=())
+
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=FakeEntrypoint(),
+        runtime_adapter=StrategyRuntimeAdapter(),
+        runtime_settings=_build_runtime_settings("global_etf_rotation"),
+        logger=lambda _msg: None,
+    )
+    called = {"count": 0}
+
+    def _boom(**_kwargs):
+        called["count"] += 1
+        raise AssertionError("should not resolve when already set")
+
+    monkeypatch.setattr(
+        "quant_platform_kit.strategy_lifecycle.live_equity.resolve_consecutive_losses",
+        _boom,
+    )
+    snapshot = PortfolioSnapshot(
+        as_of=datetime.now(timezone.utc),
+        total_equity=10_000.0,
+        positions=(),
+        metadata={"consecutive_losses": 1},
+    )
+
+    stamped = runtime._with_consecutive_loss_metadata(snapshot)
+
+    assert stamped is snapshot
+    assert called["count"] == 0