feat(risk): stamp consecutive_losses before strategy evaluate (#216)

Reuse QPK stamp helper so entrypoint circuit breakers see trailing
loss streak from live equity history.

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

Author: 3 people

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "pytest",
     "pytz",
     "requests",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@53b2ca73a5a50257b5d1a3c769b75c40924e4ba6",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@69a0256934d081b5ef309a885384b9eb9f62cf90",
     "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@17ddb86c72d44b2c7b78ba7a10d8f71b21180166",
 ]
 license = "MIT"
@@ -82,5 +82,5 @@ show_missing = true
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@53b2ca73a5a50257b5d1a3c769b75c40924e4ba6",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@69a0256934d081b5ef309a885384b9eb9f62cf90",
 ]

qsl.toml

@@ -5,7 +5,7 @@ ring = 3
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "53b2ca73a5a50257b5d1a3c769b75c40924e4ba6"
+quant_platform_kit = "69a0256934d081b5ef309a885384b9eb9f62cf90"
 us_equity_strategies = "17ddb86c72d44b2c7b78ba7a10d8f71b21180166"
 
 [qsl.compat]

strategy_runtime.py

@@ -58,6 +58,20 @@ def managed_symbols(self) -> tuple[str, ...]:
         configured = self.merged_runtime_config.get("managed_symbols", ())
         return tuple(str(symbol) for symbol in configured)
 
+    def _stamp_portfolio_risk_metadata(self, available_inputs: Mapping[str, Any]) -> dict[str, Any]:
+        resolved = dict(available_inputs)
+        snapshot = resolved.get("portfolio_snapshot")
+        if snapshot is None:
+            return resolved
+        from quant_platform_kit.strategy_lifecycle.live_equity import stamp_consecutive_losses_on_snapshot
+
+        resolved["portfolio_snapshot"] = stamp_consecutive_losses_on_snapshot(
+            snapshot,
+            strategy_profile=self.profile,
+            logger=self.logger,
+        )
+        return resolved
+
     def evaluate(
         self,
         *,
@@ -74,11 +88,11 @@ def evaluate(
         if _FEATURE_SNAPSHOT_INPUT in frozenset(self.entrypoint.manifest.required_inputs):
             return self._evaluate_feature_snapshot_strategy(
                 runtime_config=runtime_config,
-                available_inputs=available_inputs,
+                available_inputs=self._stamp_portfolio_risk_metadata(available_inputs),
             )
 
         as_of = datetime.now(timezone.utc)
-        resolved_available_inputs = dict(available_inputs)
+        resolved_available_inputs = self._stamp_portfolio_risk_metadata(available_inputs)
         resolved_available_inputs.update(
             resolve_external_market_signal_inputs(
                 strategy_profile=self.profile,

tests/test_strategy_runtime.py

@@ -115,3 +115,64 @@ def test_dca_overrides_ignore_non_dca_profiles():
     settings = _runtime_settings(dca_mode="smart", dca_base_investment_usd=500.0)
 
     assert _build_runtime_overrides("global_etf_rotation", settings) == {}
+
+
+def test_stamp_portfolio_risk_metadata_calls_qpk_helper(monkeypatch):
+    from datetime import datetime, timezone
+    from unittest.mock import MagicMock
+
+    from quant_platform_kit.common.models import PortfolioSnapshot
+    from quant_platform_kit.strategy_contracts import (
+        StrategyDecision,
+        StrategyManifest,
+        StrategyRuntimeAdapter,
+    )
+    import strategy_runtime as strategy_runtime_module
+
+    class _Entrypoint:
+        manifest = StrategyManifest(
+            profile="global_etf_rotation",
+            domain="us_equity",
+            display_name="Global ETF",
+            description="test",
+            required_inputs=frozenset({"portfolio_snapshot"}),
+        )
+
+        def evaluate(self, ctx):
+            self.ctx = ctx
+            return StrategyDecision()
+
+    entrypoint = _Entrypoint()
+    runtime = strategy_runtime_module.LoadedStrategyRuntime(
+        entrypoint=entrypoint,
+        runtime_adapter=StrategyRuntimeAdapter(portfolio_input_name="portfolio_snapshot"),
+        runtime_settings=_runtime_settings(strategy_profile="global_etf_rotation"),
+        logger=lambda _msg: None,
+    )
+    snapshot = PortfolioSnapshot(
+        as_of=datetime.now(timezone.utc),
+        total_equity=1_000.0,
+        positions=(),
+        metadata={},
+    )
+    stamped = PortfolioSnapshot(
+        as_of=snapshot.as_of,
+        total_equity=snapshot.total_equity,
+        positions=(),
+        metadata={"consecutive_losses": 2},
+    )
+    stamp = MagicMock(return_value=stamped)
+    monkeypatch.setattr(
+        "quant_platform_kit.strategy_lifecycle.live_equity.stamp_consecutive_losses_on_snapshot",
+        stamp,
+    )
+    monkeypatch.setattr(
+        strategy_runtime_module,
+        "resolve_external_market_signal_inputs",
+        lambda **_kwargs: {},
+    )
+
+    runtime.evaluate(translator=lambda key, **_kwargs: key, portfolio_snapshot=snapshot)
+
+    stamp.assert_called_once()
+    assert entrypoint.ctx.portfolio.metadata["consecutive_losses"] == 2