feat(risk): RuntimeRiskLimits and completed-session daily bars

src/quant_platform_kit/longbridge/market_data.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import time
+import math
+from datetime import timezone
 from typing import Any
 
 import pandas as pd
@@ -96,6 +98,41 @@ def fetch_lot_sizes(q_ctx: Any, symbols: list[str]) -> dict[str, int]:
     return lot_sizes
 
 
+def _completed_daily_closes(bars: list[Any], expected_session: pd.Timestamp) -> pd.DataFrame | None:
+    rows = []
+    for bar in bars:
+        timestamp = getattr(bar, "timestamp", None)
+        if timestamp is None:
+            return None
+        try:
+            instant = pd.Timestamp(timestamp)
+            if instant is pd.NaT or pd.isna(instant):
+                return None
+            if instant.tzinfo is None:
+                # LongPort 3.x returns ``fromtimestamp(epoch, None)`` from
+                # its native extension.  A naive value therefore carries the
+                # process-local representation of the broker instant; using
+                # astimezone preserves that local-time meaning before the
+                # session calendar conversion.
+                instant = pd.Timestamp(instant.to_pydatetime().astimezone(timezone.utc))
+            else:
+                instant = instant.tz_convert("UTC")
+            session = instant.tz_convert("America/New_York").normalize().tz_localize(None)
+            close = float(bar.close)
+        except (TypeError, ValueError, OverflowError):
+            return None
+        if not math.isfinite(close) or close <= 0:
+            return None
+        if session <= expected_session:
+            rows.append({"session": session, "close": close})
+    if not rows:
+        return None
+    frame = pd.DataFrame(rows)
+    if frame["session"].duplicated().any():
+        return None
+    frame = frame.sort_values("session")
+    return frame if frame.iloc[-1]["session"] == expected_session else None
+
 def calculate_rotation_indicators(
     q_ctx: Any,
     *,
@@ -104,7 +141,8 @@ def calculate_rotation_indicators(
     dynamic_rsi_quantile_window: int = 252,
     dynamic_volatility_delever_window: int = 10,
     dynamic_volatility_delever_quantile_window: int = 252,
-) -> dict[str, dict[str, float]] | None:
+    completed_session_date: str | None = None,
+) -> dict[str, dict[str, Any]] | None:
     from longport.openapi import AdjustType, Period
 
     effective_lookback = (
@@ -126,16 +164,29 @@ def calculate_rotation_indicators(
     if not soxl_bars or not soxx_bars:
         return None
 
-    df_soxl = pd.DataFrame([{"close": float(k.close)} for k in soxl_bars])
-    df_soxx = pd.DataFrame([float(k.close) for k in soxx_bars], columns=["close"])
+    completed_session = None
+    if completed_session_date is None:
+        df_soxl = pd.DataFrame([{"close": float(k.close)} for k in soxl_bars])
+        df_soxx = pd.DataFrame([float(k.close) for k in soxx_bars], columns=["close"])
+    else:
+        completed_session = pd.Timestamp(completed_session_date).normalize()
+        df_soxl = _completed_daily_closes(soxl_bars, completed_session)
+        df_soxx = _completed_daily_closes(soxx_bars, completed_session)
+        if df_soxl is None or df_soxx is None:
+            return None
+        df_soxl = df_soxl[["close"]]
+        df_soxx = df_soxx[["close"]]
     if len(df_soxl) < trend_window or len(df_soxx) < trend_window:
         return None
 
-    return build_semiconductor_rotation_indicators_from_history(
+    indicators = build_semiconductor_rotation_indicators_from_history(
         soxl_history=df_soxl["close"],
         soxx_history=df_soxx["close"],
         trend_ma_window=trend_window,
         dynamic_rsi_quantile_window=dynamic_rsi_quantile_window,
         dynamic_volatility_delever_window=dynamic_volatility_delever_window,
         dynamic_volatility_delever_quantile_window=dynamic_volatility_delever_quantile_window,
     )
+    if completed_session is not None:
+        indicators["completed_session"] = {"date": completed_session.date().isoformat()}
+    return indicators

src/quant_platform_kit/risk/__init__.py

@@ -14,6 +14,7 @@
     RiskGateAssessment,
     RiskGateResult,
     RiskSignal,
+    RuntimeRiskLimits,
 )
 from quant_platform_kit.risk.engine import (
     RiskEngine,
@@ -101,6 +102,7 @@
     "RiskGateResult",
     "RiskEngine",
     "RiskSignal",
+    "RuntimeRiskLimits",
     "aggregate_risk_signals",
     "assess_with_evidence",
     "apply_risk_gate",

src/quant_platform_kit/risk/contracts.py

@@ -7,10 +7,13 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping as ABCMapping
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 import hashlib
 import json
+import math
+from types import MappingProxyType
 from typing import Any, Mapping
 
 
@@ -153,6 +156,75 @@ class RiskAction:
     notify: bool = True
 
 
+@dataclass(frozen=True)
+class RuntimeRiskLimits:
+    """Explicit, immutable limits supplied by a verified runtime binding."""
+
+    allowed_symbols: tuple[str, ...]
+    product_leverage_factors: Mapping[str, int]
+    nominal_caps: Mapping[str, float]
+    total_nominal_exposure_cap: float
+    total_effective_exposure_cap: float
+    max_positions: int
+
+    def __post_init__(self) -> None:
+        if isinstance(self.allowed_symbols, (str, bytes)):
+            raise ValueError("allowed_symbols must be a sequence of symbols")
+        symbols = tuple(self.allowed_symbols)
+        if not symbols or any(
+            type(symbol) is not str
+            or not symbol
+            or symbol != symbol.strip()
+            or symbol != symbol.upper()
+            for symbol in symbols
+        ):
+            raise ValueError("allowed_symbols must contain canonical symbols")
+        if len(set(symbols)) != len(symbols):
+            raise ValueError("allowed_symbols must not contain duplicates")
+
+        def _mapping(value: Mapping[str, Any], field_name: str) -> dict[str, Any]:
+            if not isinstance(value, ABCMapping):
+                raise ValueError(f"{field_name} must be a mapping")
+            result = dict(value)
+            if set(result) != set(symbols):
+                raise ValueError(f"{field_name} must cover allowed_symbols exactly")
+            return result
+
+        factors = _mapping(self.product_leverage_factors, "product_leverage_factors")
+        if any(
+            type(symbol) is not str
+            or type(factor) is not int
+            or isinstance(factor, bool)
+            or factor < 1
+            for symbol, factor in factors.items()
+        ):
+            raise ValueError("product_leverage_factors must contain positive integers")
+
+        caps = _mapping(self.nominal_caps, "nominal_caps")
+        if any(
+            type(symbol) is not str
+            or type(cap) not in (int, float)
+            or not math.isfinite(float(cap))
+            or float(cap) < 0.0
+            for symbol, cap in caps.items()
+        ):
+            raise ValueError("nominal_caps must contain finite nonnegative values")
+
+        for field_name in (
+            "total_nominal_exposure_cap",
+            "total_effective_exposure_cap",
+        ):
+            value = getattr(self, field_name)
+            if type(value) not in (int, float) or not math.isfinite(float(value)) or float(value) < 0.0:
+                raise ValueError(f"{field_name} must be finite and nonnegative")
+        if type(self.max_positions) is not int or isinstance(self.max_positions, bool) or self.max_positions < 0:
+            raise ValueError("max_positions must be a nonnegative integer")
+
+        object.__setattr__(self, "allowed_symbols", symbols)
+        object.__setattr__(self, "product_leverage_factors", MappingProxyType(factors))
+        object.__setattr__(self, "nominal_caps", MappingProxyType(caps))
+
+
 @dataclass(frozen=True)
 class CandidateRiskIdentity:
     """Immutable identity of one mandate-bound promotion candidate."""

src/quant_platform_kit/risk/gate.py

@@ -26,6 +26,7 @@
     CandidateRiskIdentity,
     RiskGateAssessment,
     RiskGateResult,
+    RuntimeRiskLimits,
 )
 from quant_platform_kit.risk.engine import build_risk_engine
 from quant_platform_kit.common.strategy_contracts import (
@@ -180,6 +181,62 @@ def _canonical_numeric_mapping(
     return result
 
 
+def _runtime_risk_limits_rejection(
+    limits: RuntimeRiskLimits,
+    *,
+    positions: tuple[PositionTarget, ...],
+    budgets: tuple[BudgetIntent, ...],
+    weights: list[tuple[PositionTarget, float]],
+    value_target_exposure_enforced: bool,
+) -> tuple[str, str] | None:
+    """Return a fail-closed finding for an explicit runtime limit set."""
+    if type(limits) is not RuntimeRiskLimits:
+        return ("rejected:runtime_risk_limits", "invalid_runtime_risk_limits")
+
+    allowed = set(limits.allowed_symbols)
+    seen_symbols: set[str] = set()
+    for position in positions:
+        if type(position) is not PositionTarget:
+            return ("rejected:runtime_risk_limits", "invalid_runtime_risk_limits")
+        if position.symbol not in allowed:
+            return ("rejected:runtime_risk_limits", "symbol_not_allowed")
+        if position.symbol in seen_symbols:
+            return ("rejected:runtime_risk_limits", "duplicate_symbol")
+        seen_symbols.add(position.symbol)
+        if position.target_weight is not None and position.target_value is not None:
+            return ("rejected:runtime_risk_limits", "conflicting_target_modes")
+        if position.target_weight is not None and (
+            _finite_number(position.target_weight) is None
+            or float(position.target_weight) < 0.0
+        ):
+            return ("rejected:runtime_risk_limits", "negative_weight_not_allowed")
+        if position.target_value is not None and not value_target_exposure_enforced:
+            return ("rejected:runtime_risk_limits", "value_target_enforcement_required")
+
+    if budgets:
+        return ("rejected:runtime_risk_limits", "budgets_not_supported")
+
+    if len(weights) > limits.max_positions:
+        return ("rejected:runtime_risk_limits", "max_positions_exceeded")
+
+    nominal_total = 0.0
+    effective_total = 0.0
+    for position, weight in weights:
+        factor = limits.product_leverage_factors.get(position.symbol)
+        nominal_cap = limits.nominal_caps.get(position.symbol)
+        if factor is None or nominal_cap is None:
+            return ("rejected:runtime_risk_limits", "asset_limit_missing")
+        if weight > nominal_cap:
+            return ("rejected:runtime_risk_limits", "nominal_cap_exceeded")
+        nominal_total += weight
+        effective_total += weight * factor
+    if nominal_total > limits.total_nominal_exposure_cap + 1e-9:
+        return ("rejected:runtime_risk_limits", "total_nominal_cap_exceeded")
+    if effective_total > limits.total_effective_exposure_cap + 1e-9:
+        return ("rejected:runtime_risk_limits", "total_effective_cap_exceeded")
+    return None
+
+
 def _canonical_cap_material(
     value: Any,
     *,
@@ -2138,6 +2195,7 @@ def _apply_risk_gate_static(
     enforce_value_target_exposure: Any,
     capital_base: CapitalBaseSnapshot | Mapping[str, Any] | None,
     capital_base_binding: CapitalBaseBinding | Mapping[str, Any] | None,
+    runtime_risk_limits: RuntimeRiskLimits | None,
     now: datetime,
     engine_action: Any,
     engine_failed: bool,
@@ -2390,6 +2448,17 @@ def _apply_risk_gate_static(
             if weight > 0.0:
                 weights.append((position, weight))
 
+    if static_rejection is None and runtime_risk_limits is not None:
+        runtime_rejection = _runtime_risk_limits_rejection(
+            runtime_risk_limits,
+            positions=positions,
+            budgets=raw_budgets,
+            weights=weights,
+            value_target_exposure_enforced=value_target_exposure_enforced,
+        )
+        if runtime_rejection is not None:
+            static_rejection = runtime_rejection
+
     if (
         static_rejection is None
         and positions
@@ -2440,7 +2509,7 @@ def _apply_risk_gate_static(
                             "rejected:overexposed",
                             f"名义仓位 {weight:.1%} > 可用账户容量",
                         )
-    elif static_rejection is None and positions:
+    elif static_rejection is None and positions and runtime_risk_limits is None:
         effective_single_weight = min(
             requested_single_weight,
             _DEFAULT_MAX_SINGLE_WEIGHT,
@@ -2543,6 +2612,7 @@ def apply_risk_gate(
     enforce_value_target_exposure: bool = False,
     capital_base: CapitalBaseSnapshot | Mapping[str, Any] | None = None,
     capital_base_binding: CapitalBaseBinding | Mapping[str, Any] | None = None,
+    runtime_risk_limits: RuntimeRiskLimits | None = None,
 ) -> StrategyDecision:
     """Apply hard checks and call RiskEngine.assess exactly once.
 
@@ -2578,6 +2648,7 @@ def apply_risk_gate(
             enforce_value_target_exposure=enforce_value_target_exposure,
             capital_base=capital_base,
             capital_base_binding=capital_base_binding,
+            runtime_risk_limits=runtime_risk_limits,
             now=now,
             engine_action=engine_action,
             engine_failed=engine_failed,