feat(risk): stamp consecutive_losses before strategy evaluate

pyproject.toml

@@ -18,7 +18,7 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "longport==3.0.23",
-    "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",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@b6a8ac2ad3c8110b5ea74fb059c8206388d63bcd",
 ]
@@ -61,5 +61,5 @@ include = [
 
 [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"
 hk_equity_strategies = "b6a8ac2ad3c8110b5ea74fb059c8206388d63bcd"
 

strategy_runtime.py

@@ -59,6 +59,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,
         *,
@@ -75,11 +89,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)
         from us_equity_strategies.signals import resolve_external_market_signal_inputs
         resolved_available_inputs.update(
             resolve_external_market_signal_inputs(

tests/test_strategy_runtime.py

@@ -570,6 +570,62 @@ def test_feature_snapshot_runtime_loads_mega_cap_top50_snapshot_into_context(sel
         self.assertEqual(result.metadata["managed_symbols"], ("NVDA", "META", "BOXX"))
         self.assertEqual(result.metadata["status_icon"], "👑")
 
+    def test_evaluate_stamps_consecutive_losses_on_portfolio_snapshot(self):
+        from quant_platform_kit.common.models import PortfolioSnapshot
+
+        class _GlobalEntrypoint:
+            def __init__(self):
+                self.manifest = StrategyManifest(
+                    profile="global_etf_rotation",
+                    domain="us_equity",
+                    display_name="Global ETF Rotation",
+                    description="test",
+                    required_inputs=frozenset({"market_history", "portfolio_snapshot"}),
+                )
+                self.ctx = None
+
+            def evaluate(self, ctx):
+                self.ctx = ctx
+                return StrategyDecision()
+
+        entrypoint = _GlobalEntrypoint()
+        runtime = strategy_runtime_module.LoadedStrategyRuntime(
+            entrypoint=entrypoint,
+            runtime_adapter=StrategyRuntimeAdapter(
+                portfolio_input_name="portfolio_snapshot",
+                runtime_policy=StrategyRuntimePolicy(signal_effective_after_trading_days=0),
+            ),
+            runtime_settings=_build_runtime_settings("global_etf_rotation"),
+            logger=lambda _message: None,
+        )
+        snapshot = PortfolioSnapshot(
+            as_of=datetime.now(timezone.utc),
+            total_equity=10_000.0,
+            positions=(),
+            metadata={},
+        )
+        stamped = PortfolioSnapshot(
+            as_of=snapshot.as_of,
+            total_equity=snapshot.total_equity,
+            positions=(),
+            metadata={"consecutive_losses": 4},
+        )
+
+        with patch(
+            "quant_platform_kit.strategy_lifecycle.live_equity.stamp_consecutive_losses_on_snapshot",
+            return_value=stamped,
+        ) as stamp:
+            result = runtime.evaluate(
+                market_history=lambda *_args, **_kwargs: [1.0, 2.0],
+                portfolio_snapshot=snapshot,
+                translator=lambda key, **_kwargs: key,
+            )
+
+        stamp.assert_called_once()
+        self.assertIs(entrypoint.ctx.portfolio, stamped)
+        self.assertEqual(entrypoint.ctx.portfolio.metadata["consecutive_losses"], 4)
+        self.assertEqual(result.metadata["strategy_profile"], "global_etf_rotation")
+
 
 if __name__ == "__main__":
     unittest.main()