feat(risk): small-account hold binding and QPK/UES pin

application/v7_paper_application.py

@@ -33,7 +33,7 @@
 # The research contract's frozen source is 07b164..., while the package that
 # is actually approved for the disabled account process tracks the platform UES pin.
 # Keep research source_commit separate from approved_ues_revision.
-V7_APPROVED_UES_REVISION = "e2258223310913f6db9f40b810756db0ee2cfd68"
+V7_APPROVED_UES_REVISION = "4a3943883cd6b5bbfe32a559e56a91b40a81b7ce"
 
 _COMMIT_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
 _TICKET_PATTERN = re.compile(r"^rpt_[0-9a-fA-F]{64}$")

pyproject.toml

@@ -18,8 +18,8 @@ dependencies = [
     "google-cloud-storage",
     "google-auth",
     "longport==3.0.23",
-    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48",
-    "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@e2258223310913f6db9f40b810756db0ee2cfd68",
+    "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@a90937621a7a05c9f72d0ca8a29be3fcb18a327c",
+    "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@4a3943883cd6b5bbfe32a559e56a91b40a81b7ce",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@709e5e1cde7841aed538d94eb26b552b46cb7806",
 ]
 
@@ -61,5 +61,5 @@ include = [
 
 [tool.uv]
 override-dependencies = [
-  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@68c51590da8a5097b7de6d75b4ccb6a175318b48",
+  "quant-platform-kit @ git+https://github.com/QuantStrategyLab/QuantPlatformKit.git@a90937621a7a05c9f72d0ca8a29be3fcb18a327c",
 ]

qsl.toml

@@ -5,8 +5,8 @@ upgrade_ring = "ring_d"
 allow_legacy = false
 
 [qsl.requires]
-quant_platform_kit = "68c51590da8a5097b7de6d75b4ccb6a175318b48"
-us_equity_strategies = "e2258223310913f6db9f40b810756db0ee2cfd68"
+quant_platform_kit = "a90937621a7a05c9f72d0ca8a29be3fcb18a327c"
+us_equity_strategies = "4a3943883cd6b5bbfe32a559e56a91b40a81b7ce"
 hk_equity_strategies = "709e5e1cde7841aed538d94eb26b552b46cb7806"
 
 [qsl.compat]

strategy_runtime.py

@@ -26,7 +26,7 @@
     build_execution_timing_metadata,
     build_strategy_context_from_available_inputs,
 )
-from quant_platform_kit.risk.contracts import RuntimeRiskLimits
+from quant_platform_kit.risk.contracts import RuntimeRiskLimits, SmallAccountRiskHoldPolicy
 from runtime_config_support import PlatformRuntimeSettings
 
 from strategy_loader import (
@@ -38,6 +38,22 @@
 _FEATURE_SNAPSHOT_INPUT = "feature_snapshot"
 _SOXL_PROFILE = "soxl_soxx_trend_income"
 
+
+def _parse_small_account_hold_policy(raw: Any) -> SmallAccountRiskHoldPolicy | None:
+    """Parse optional deployment hold policy; invalid shapes return None."""
+    if raw is None:
+        return None
+    if not isinstance(raw, Mapping):
+        return None
+    try:
+        return SmallAccountRiskHoldPolicy(
+            enabled=raw["enabled"],
+            hold_below_nav=raw["hold_below_nav"],
+            require_cash_only=raw.get("require_cash_only", True),
+        )
+    except (KeyError, TypeError, ValueError):
+        return None
+
 def _installed_ues_revision() -> str | None:
     """Read the VCS revision of the installed UES distribution."""
     try:
@@ -186,7 +202,13 @@ def _build_runtime_risk_capabilities(
             "max_positions",
             "exit_parameters",
         }
-        if set(policy) != expected_policy_keys or not isinstance(policy.get("binding"), Mapping):
+        optional_policy_keys = {"small_account_hold"}
+        policy_keys = set(policy)
+        if (
+            not expected_policy_keys.issubset(policy_keys)
+            or (policy_keys - expected_policy_keys - optional_policy_keys)
+            or not isinstance(policy.get("binding"), Mapping)
+        ):
             return {**capabilities, "runtime_risk_limits": object()}, "unavailable:invalid_runtime_risk_policy"
 
         target_release = runtime_target.strategy_release
@@ -255,7 +277,25 @@ def _build_runtime_risk_capabilities(
             )
         except (TypeError, ValueError):
             return {**capabilities, "runtime_risk_limits": object()}, "unavailable:invalid_runtime_risk_limits"
-        return {**capabilities, "runtime_risk_limits": limits}, "verified:runtime_risk_limits"
+        capability_payload: dict[str, Any] = {
+            **capabilities,
+            "runtime_risk_limits": limits,
+            "cash_only_execution": bool(self.runtime_settings.cash_only_execution),
+        }
+        hold_policy = _parse_small_account_hold_policy(policy.get("small_account_hold"))
+        if isinstance(policy.get("small_account_hold"), Mapping) and hold_policy is None:
+            return {
+                **capabilities,
+                "runtime_risk_limits": object(),
+            }, "unavailable:invalid_small_account_hold"
+        if hold_policy is not None:
+            if hold_policy.require_cash_only and self.runtime_settings.cash_only_execution is not True:
+                return {
+                    **capabilities,
+                    "runtime_risk_limits": object(),
+                }, "unavailable:small_account_hold_cash_only"
+            capability_payload["small_account_hold_policy"] = hold_policy
+        return capability_payload, "verified:runtime_risk_limits"
 
     def _build_feature_snapshot_context(self, request):
         return build_strategy_context_from_available_inputs(

tests/test_runtime_risk_binding.py

@@ -138,6 +138,55 @@ def test_binds_verified_limits(self):
         self.assertEqual(entrypoint.ctx.capabilities["runtime_risk_limits"].max_positions, 8)
         self.assertIsInstance(entrypoint.ctx.capabilities["runtime_risk_limits"], RuntimeRiskLimits)
 
+    def test_attaches_small_account_hold_policy(self):
+        from quant_platform_kit.risk.contracts import SmallAccountRiskHoldPolicy
+
+        policy = _policy()
+        policy["small_account_hold"] = {
+            "enabled": True,
+            "hold_below_nav": 1000.0,
+            "require_cash_only": True,
+        }
+        entrypoint = _SoxlEntrypoint()
+        runtime = strategy_runtime_module.LoadedStrategyRuntime(
+            entrypoint=entrypoint,
+            runtime_adapter=StrategyRuntimeAdapter(portfolio_input_name="portfolio_snapshot"),
+            runtime_settings=_settings(policy),
+            merged_runtime_config=dict(entrypoint.manifest.default_config),
+        )
+        with patch.object(strategy_runtime_module, "_installed_ues_revision", return_value="ues-revision"):
+            with patch("us_equity_strategies.signals.resolve_external_market_signal_inputs", return_value={}):
+                result = runtime.evaluate(
+                    translator=lambda key, **_k: key,
+                    benchmark_history=[{"close": 1.0}],
+                    portfolio_snapshot=_snapshot(),
+                )
+        self.assertEqual(result.metadata["runtime_risk_status"], "verified:runtime_risk_limits")
+        hold = entrypoint.ctx.capabilities["small_account_hold_policy"]
+        self.assertIsInstance(hold, SmallAccountRiskHoldPolicy)
+        self.assertEqual(hold.hold_below_nav, 1000.0)
+        self.assertTrue(entrypoint.ctx.capabilities["cash_only_execution"])
+
+    def test_rejects_invalid_small_account_hold(self):
+        policy = _policy()
+        policy["small_account_hold"] = {"enabled": True, "hold_below_nav": -1}
+        entrypoint = _SoxlEntrypoint()
+        runtime = strategy_runtime_module.LoadedStrategyRuntime(
+            entrypoint=entrypoint,
+            runtime_adapter=StrategyRuntimeAdapter(portfolio_input_name="portfolio_snapshot"),
+            runtime_settings=_settings(policy),
+            merged_runtime_config=dict(entrypoint.manifest.default_config),
+        )
+        with patch.object(strategy_runtime_module, "_installed_ues_revision", return_value="ues-revision"):
+            with patch("us_equity_strategies.signals.resolve_external_market_signal_inputs", return_value={}):
+                result = runtime.evaluate(
+                    translator=lambda key, **_k: key,
+                    benchmark_history=[{"close": 1.0}],
+                    portfolio_snapshot=_snapshot(),
+                )
+        self.assertEqual(result.metadata["runtime_risk_status"], "unavailable:invalid_small_account_hold")
+        self.assertNotIn("small_account_hold_policy", entrypoint.ctx.capabilities)
+
     def test_rejects_wrong_account_hash(self):
         entrypoint = _SoxlEntrypoint()
         runtime = strategy_runtime_module.LoadedStrategyRuntime(