fix: bind LongBridge SOXL runtime risk limits to account and UES

application/v7_paper_application.py

@@ -31,9 +31,9 @@
 V7_PAPER_SCOPE = "PAPER"
 V7_PAPER_PLATFORM = "longbridge"
 # The research contract's frozen source is 07b164..., while the package that
-# is actually approved for the disabled account process is the controlled
-# b83ef4b3... revision.  Keep those identities separate.
-V7_APPROVED_UES_REVISION = "b83ef4b3ae67c47d132ddd660ba3ccc60d474c85"
+# 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"
 
 _COMMIT_PATTERN = re.compile(r"^[0-9a-fA-F]{40}$")
 _TICKET_PATTERN = re.compile(r"^rpt_[0-9a-fA-F]{64}$")

pyproject.toml

@@ -19,7 +19,7 @@ dependencies = [
     "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@b83ef4b3ae67c47d132ddd660ba3ccc60d474c85",
+    "us-equity-strategies @ git+https://github.com/QuantStrategyLab/UsEquityStrategies.git@e2258223310913f6db9f40b810756db0ee2cfd68",
     "hk-equity-strategies @ git+https://github.com/QuantStrategyLab/HkEquityStrategies.git@709e5e1cde7841aed538d94eb26b552b46cb7806",
 ]
 

qsl.toml

@@ -6,7 +6,7 @@ allow_legacy = false
 
 [qsl.requires]
 quant_platform_kit = "68c51590da8a5097b7de6d75b4ccb6a175318b48"
-us_equity_strategies = "c31f28484b8489a1ecda8e595a6d2afd3bce8185"
+us_equity_strategies = "e2258223310913f6db9f40b810756db0ee2cfd68"
 hk_equity_strategies = "709e5e1cde7841aed538d94eb26b552b46cb7806"
 
 [qsl.compat]

runtime_config_support.py

@@ -187,6 +187,7 @@ class PlatformRuntimeSettings:
     strategy_plugin_alert_telegram_parse_mode: str | None = None
     strategy_plugin_alert_telegram_disable_web_page_preview: str | None = None
     strategy_plugin_alert_telegram_body_max_chars: str | None = None
+    trusted_runtime_risk_policy: Mapping[str, Any] | None = None
     runtime_target: RuntimeTarget | None = None
     strategy_metadata: Any = None
 
@@ -263,6 +264,26 @@ def _runtime_target_market_value(runtime_target: RuntimeTarget, field: str) -> s
     return str(value).strip() if value is not None and str(value).strip() else None
 
 
+
+def _load_trusted_runtime_risk_policy() -> Mapping[str, Any] | None:
+    """Read risk limits only from the deployment runtime target JSON."""
+    raw_target = os.getenv("RUNTIME_TARGET_JSON") or os.getenv("QSL_RUNTIME_TARGET_JSON")
+    if raw_target is None or not str(raw_target).strip():
+        return None
+    try:
+        payload = json.loads(raw_target)
+    except (TypeError, ValueError) as exc:
+        raise EnvironmentError("RUNTIME_TARGET_JSON must be valid JSON") from exc
+    if not isinstance(payload, dict):
+        raise EnvironmentError("RUNTIME_TARGET_JSON must decode to an object")
+    policy = payload.get("runtime_risk_limits")
+    if policy is None:
+        return None
+    if not isinstance(policy, dict):
+        raise EnvironmentError("RUNTIME_TARGET_JSON.runtime_risk_limits must be an object")
+    return dict(policy)
+
+
 def load_platform_runtime_settings(
     *,
     project_id_resolver: Callable[[], str | None],
@@ -276,6 +297,7 @@ def load_platform_runtime_settings(
         env=os.environ,
         expected_platform_id=LONGBRIDGE_PLATFORM,
     )
+    trusted_runtime_risk_policy = _load_trusted_runtime_risk_policy()
     strategy_definition = None
     strategy_metadata = None
     if runtime_target.strategy_profile == _bound_v7_profile():
@@ -516,6 +538,7 @@ def load_platform_runtime_settings(
             os.getenv("STRATEGY_PLUGIN_ALERT_TELEGRAM_BODY_MAX_CHARS")
         ),
         runtime_target=runtime_target,
+        trusted_runtime_risk_policy=trusted_runtime_risk_policy,
         strategy_metadata=strategy_metadata,
     )
 

strategy_runtime.py

@@ -26,6 +26,7 @@
     build_execution_timing_metadata,
     build_strategy_context_from_available_inputs,
 )
+from quant_platform_kit.risk.contracts import RuntimeRiskLimits
 from runtime_config_support import PlatformRuntimeSettings
 
 from strategy_loader import (
@@ -35,6 +36,24 @@
 
 
 _FEATURE_SNAPSHOT_INPUT = "feature_snapshot"
+_SOXL_PROFILE = "soxl_soxx_trend_income"
+
+def _installed_ues_revision() -> str | None:
+    """Read the VCS revision of the installed UES distribution."""
+    try:
+        distribution = importlib_metadata.distribution("us-equity-strategies")
+        raw_direct_url = distribution.read_text("direct_url.json")
+        if not raw_direct_url:
+            return None
+        payload = json.loads(raw_direct_url)
+        revision = payload.get("vcs_info", {}).get("commit_id")
+    except (ImportError, OSError, TypeError, ValueError, AttributeError):
+        return None
+    if not isinstance(revision, str) or not revision.strip():
+        return None
+    return revision.strip()
+
+
 DCA_PROFILES = frozenset({"nasdaq_sp500_smart_dca", "ibit_smart_dca"})
 IBIT_ZSCORE_EXIT_PROFILE = "ibit_smart_dca"
 
@@ -141,6 +160,103 @@ def _build_capital_base_capabilities(self, available_inputs: Mapping[str, Any])
         capabilities.update({"capital_base": capital, "capital_base_binding": binding})
         return capabilities
 
+    def _build_runtime_risk_capabilities(
+        self,
+        available_inputs: Mapping[str, Any],
+        capabilities: Mapping[str, Any],
+    ) -> tuple[dict[str, Any], str]:
+        """Bind explicit limits to the deployed account and installed UES."""
+        if self.profile != _SOXL_PROFILE:
+            return dict(capabilities), "unavailable:profile_not_supported"
+        policy = self.runtime_settings.trusted_runtime_risk_policy
+        runtime_target = self.runtime_settings.runtime_target
+        snapshot = available_inputs.get("portfolio_snapshot")
+        binding = capabilities.get("capital_base_binding")
+        if not isinstance(policy, Mapping):
+            return {**capabilities, "runtime_risk_limits": object()}, "unavailable:runtime_risk_policy"
+        if runtime_target is None or snapshot is None or binding is None:
+            return {**capabilities, "runtime_risk_limits": object()}, "unavailable:runtime_binding"
+        expected_policy_keys = {
+            "binding",
+            "allowed_symbols",
+            "product_leverage_factors",
+            "nominal_caps",
+            "total_nominal_exposure_cap",
+            "total_effective_exposure_cap",
+            "max_positions",
+            "exit_parameters",
+        }
+        if set(policy) != expected_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
+        policy_binding = policy["binding"]
+        expected_binding_keys = {
+            "account_scope",
+            "runtime_scope",
+            "account_hash",
+            "strategy_profile",
+            "ues_revision",
+            "execution_mode",
+            "cash_only_execution",
+            "reserved_cash_ratio",
+            "options_enabled",
+        }
+        if set(policy_binding) != expected_binding_keys:
+            return {**capabilities, "runtime_risk_limits": object()}, "unavailable:invalid_runtime_binding"
+        metadata = getattr(snapshot, "metadata", {})
+        account_scope = str(runtime_target.account_scope or "").strip()
+        runtime_scope = str(runtime_target.service_name or runtime_target.deployment_selector or "").strip()
+        actual_account_hash = str(metadata.get("account_hash") or "").strip() if isinstance(metadata, Mapping) else ""
+        actual_ues_revision = _installed_ues_revision()
+        actual_exit_buffer = self.merged_runtime_config.get("trend_exit_buffer")
+        if (
+            not account_scope
+            or not runtime_scope
+            or not actual_account_hash
+            or target_release is None
+            or str(policy_binding["account_scope"]).strip() != account_scope
+            or str(policy_binding["runtime_scope"]).strip() != runtime_scope
+            or str(policy_binding["account_hash"]).strip() != actual_account_hash
+            or str(policy_binding["strategy_profile"]).strip() != self.profile
+            or str(policy_binding["ues_revision"]).strip() != str(target_release.strategy_revision).strip()
+            or actual_ues_revision is None
+            or actual_ues_revision != str(policy_binding["ues_revision"]).strip()
+            or str(policy_binding["execution_mode"]).strip().lower() != runtime_target.execution_mode
+            or policy_binding["cash_only_execution"] is not True
+            or self.runtime_settings.cash_only_execution is not True
+            or policy_binding["reserved_cash_ratio"] != self.merged_runtime_config.get("cash_reserve_ratio")
+            or policy_binding["reserved_cash_ratio"] != self.runtime_settings.reserved_cash_ratio
+            or policy_binding["reserved_cash_ratio"] != 0.03
+            or policy_binding["options_enabled"] is not False
+            or any(
+                self.merged_runtime_config.get(key) is not False
+                for key in (
+                    "option_overlay_enabled",
+                    "option_growth_overlay_enabled",
+                    "option_income_overlay_enabled",
+                )
+            )
+            or not isinstance(policy.get("exit_parameters"), Mapping)
+            or actual_exit_buffer is None
+            or actual_exit_buffer != 0.02
+            or dict(policy["exit_parameters"]) != {"trend_exit_buffer": 0.02}
+            or dict(policy["exit_parameters"]) != {"trend_exit_buffer": actual_exit_buffer}
+        ):
+            return {**capabilities, "runtime_risk_limits": object()}, "unavailable:runtime_binding_mismatch"
+        try:
+            limits = RuntimeRiskLimits(
+                allowed_symbols=tuple(policy["allowed_symbols"]),
+                product_leverage_factors=policy["product_leverage_factors"],
+                nominal_caps=policy["nominal_caps"],
+                total_nominal_exposure_cap=policy["total_nominal_exposure_cap"],
+                total_effective_exposure_cap=policy["total_effective_exposure_cap"],
+                max_positions=policy["max_positions"],
+            )
+        except (TypeError, ValueError):
+            return {**capabilities, "runtime_risk_limits": object()}, "unavailable:invalid_runtime_risk_limits"
+        return {**capabilities, "runtime_risk_limits": limits}, "verified:runtime_risk_limits"
+
     def _build_feature_snapshot_context(self, request):
         return build_strategy_context_from_available_inputs(
             entrypoint=request.entrypoint,
@@ -184,6 +300,10 @@ def evaluate(
             )
         )
         capabilities = self._build_capital_base_capabilities(resolved_available_inputs)
+        capabilities, runtime_risk_status = self._build_runtime_risk_capabilities(
+            resolved_available_inputs,
+            capabilities,
+        )
         ctx = build_strategy_context_from_available_inputs(
             entrypoint=active_entrypoint,
             runtime_adapter=self.runtime_adapter,
@@ -198,6 +318,7 @@ def evaluate(
             metadata={
                 "strategy_profile": self.profile,
                 "strategy_display_name": self.display_name,
+                "runtime_risk_status": runtime_risk_status,
                 **build_execution_timing_metadata(
                     signal_date=as_of,
                     signal_effective_after_trading_days=(
@@ -286,6 +407,11 @@ def _build_runtime_overrides(profile: str, runtime_settings: PlatformRuntimeSett
         overrides["reserved_cash_floor_usd"] = float(reserved_cash_floor_usd)
     if reserved_cash_ratio is not None and float(reserved_cash_ratio or 0.0) > 0.0:
         overrides["reserved_cash_ratio"] = float(reserved_cash_ratio)
+        overrides["cash_reserve_ratio"] = float(reserved_cash_ratio)
+    if profile == _SOXL_PROFILE and bool(getattr(runtime_settings, "cash_only_execution", True)):
+        overrides["option_overlay_enabled"] = False
+        overrides["option_growth_overlay_enabled"] = False
+        overrides["option_income_overlay_enabled"] = False
     income_layer_enabled = getattr(runtime_settings, "income_layer_enabled", None)
     income_layer_start_usd = getattr(runtime_settings, "income_layer_start_usd", None)
     income_layer_max_ratio = getattr(runtime_settings, "income_layer_max_ratio", None)