feat(risk): stamp consecutive_losses before strategy evaluate

pyproject.toml

@@ -13,8 +13,8 @@ authors = [{ name = "QuantStrategyLab" }]
 dependencies = [
   "flask>=3.0",
   "pandas>=2.0",
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@37c81901160c5b31127a27dba1c63944933fb6bf",
-  "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@73844e92a8570a61e5a9dc6c245809d0b27b89bc",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@69a0256934d081b5ef309a885384b9eb9f62cf90",
+  "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@12c0cd4801060fcb2f9452ffd9a7f48df446ddd0",
 ]
 
 [project.optional-dependencies]

qsl.toml

@@ -6,8 +6,8 @@ allow_legacy = false
 enforce_bundle = true
 
 [qsl.requires]
-quant_platform_kit = "37c81901160c5b31127a27dba1c63944933fb6bf"
-cn_equity_strategies = "73844e92a8570a61e5a9dc6c245809d0b27b89bc"
+quant_platform_kit = "69a0256934d081b5ef309a885384b9eb9f62cf90"
+cn_equity_strategies = "12c0cd4801060fcb2f9452ffd9a7f48df446ddd0"
 
 [qsl.compat]
 bundle = "2026.07.2"

strategy_runtime.py

@@ -47,18 +47,28 @@ def evaluate(self, *, available_inputs: Mapping[str, Any]) -> StrategyEvaluation
         runtime_config = dict(self.runtime_config)
         apply_runtime_policy_to_runtime_config(runtime_config, self.runtime_adapter)
 
+        stamped_inputs = dict(available_inputs)
+        snapshot = stamped_inputs.get("portfolio_snapshot")
+        if snapshot is not None:
+            from quant_platform_kit.strategy_lifecycle.live_equity import stamp_consecutive_losses_on_snapshot
+
+            stamped_inputs["portfolio_snapshot"] = stamp_consecutive_losses_on_snapshot(
+                snapshot,
+                strategy_profile=self.profile,
+            )
+
         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=stamped_inputs,
             )
 
         as_of = datetime.now(timezone.utc)
         ctx = build_strategy_context_from_available_inputs(
             entrypoint=self.entrypoint,
             runtime_adapter=self.runtime_adapter,
             as_of=as_of,
-            available_inputs=dict(available_inputs),
+            available_inputs=stamped_inputs,
             runtime_config=runtime_config,
         )
         decision = self.entrypoint.evaluate(ctx)

tests/test_qsl_metadata.py

@@ -15,5 +15,5 @@ def test_qsl_metadata_has_runtime_platform_fields() -> None:
     assert qsl["enforce_bundle"] is True
     assert qsl["compat"]["bundle"] == "2026.07.2"
     requires = qsl["requires"]
-    assert requires["quant_platform_kit"] == "37c81901160c5b31127a27dba1c63944933fb6bf"
-    assert requires["cn_equity_strategies"] == "73844e92a8570a61e5a9dc6c245809d0b27b89bc"
+    assert requires["quant_platform_kit"] == "69a0256934d081b5ef309a885384b9eb9f62cf90"
+    assert requires["cn_equity_strategies"] == "12c0cd4801060fcb2f9452ffd9a7f48df446ddd0"

tests/test_strategy_runtime.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from unittest.mock import patch
+
+from quant_platform_kit.common.models import PortfolioSnapshot
+from quant_platform_kit.strategy_contracts import (
+    StrategyDecision,
+    StrategyManifest,
+    StrategyRuntimeAdapter,
+)
+from runtime_config_support import PlatformRuntimeSettings
+import strategy_runtime as strategy_runtime_module
+
+
+def _runtime_settings() -> PlatformRuntimeSettings:
+    return PlatformRuntimeSettings(
+        strategy_profile="cn_industry_etf_rotation",
+        strategy_display_name="CN Industry ETF Rotation",
+        strategy_domain="cn_equity",
+        dry_run_only=True,
+        market_history_path=None,
+        feature_snapshot_path=None,
+        feature_snapshot_manifest_path=None,
+    )
+
+
+def test_evaluate_stamps_consecutive_losses_on_portfolio_snapshot():
+    class _Entrypoint:
+        manifest = StrategyManifest(
+            profile="cn_industry_etf_rotation",
+            domain="cn_equity",
+            display_name="CN Industry ETF Rotation",
+            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(),
+    )
+    snapshot = PortfolioSnapshot(
+        as_of=datetime.now(timezone.utc),
+        total_equity=100_000.0,
+        positions=(),
+        metadata={},
+    )
+    stamped = PortfolioSnapshot(
+        as_of=snapshot.as_of,
+        total_equity=snapshot.total_equity,
+        positions=(),
+        metadata={"consecutive_losses": 3},
+    )
+
+    with patch(
+        "quant_platform_kit.strategy_lifecycle.live_equity.stamp_consecutive_losses_on_snapshot",
+        return_value=stamped,
+    ) as stamp:
+        result = runtime.evaluate(available_inputs={"portfolio_snapshot": snapshot})
+
+    stamp.assert_called_once()
+    assert entrypoint.ctx.portfolio is stamped
+    assert entrypoint.ctx.portfolio.metadata["consecutive_losses"] == 3
+    assert result.metadata["strategy_profile"] == "cn_industry_etf_rotation"