From e0a36565b4d11deaa7034280ab443bd8ff14cd1f Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 17 Sep 2026 05:17:37 +0800 Subject: [PATCH 1/2] feat(risk): add RuntimeRiskLimits and completed-session daily bars Enable trusted multi-ETF exposure caps for SOXL recovery without weakening the default single-position gate, and map LongPort naive timestamps via process-local time to completed XNYS sessions. Co-authored-by: Cursor --- .../longbridge/market_data.py | 59 ++++- src/quant_platform_kit/risk/__init__.py | 2 + src/quant_platform_kit/risk/contracts.py | 72 ++++++ src/quant_platform_kit/risk/gate.py | 73 +++++- tests/test_longbridge_market_data.py | 141 ++++++++++- tests/test_risk_gate.py | 232 ++++++++++++++++++ 6 files changed, 573 insertions(+), 6 deletions(-) diff --git a/src/quant_platform_kit/longbridge/market_data.py b/src/quant_platform_kit/longbridge/market_data.py index 23f66fdc..f3f309a1 100644 --- a/src/quant_platform_kit/longbridge/market_data.py +++ b/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,12 +164,22 @@ 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, @@ -139,3 +187,6 @@ def calculate_rotation_indicators( 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 diff --git a/src/quant_platform_kit/risk/__init__.py b/src/quant_platform_kit/risk/__init__.py index ecffdf9a..5eebf70e 100644 --- a/src/quant_platform_kit/risk/__init__.py +++ b/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", diff --git a/src/quant_platform_kit/risk/contracts.py b/src/quant_platform_kit/risk/contracts.py index f05481ca..e156ff05 100644 --- a/src/quant_platform_kit/risk/contracts.py +++ b/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.""" diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index e08c74b1..004a2ab8 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/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, diff --git a/tests/test_longbridge_market_data.py b/tests/test_longbridge_market_data.py index 361db300..7419377a 100644 --- a/tests/test_longbridge_market_data.py +++ b/tests/test_longbridge_market_data.py @@ -1,11 +1,17 @@ from __future__ import annotations +import contextlib +import ctypes +import os import sys import types import unittest from unittest.mock import patch +import pandas as pd + from quant_platform_kit.longbridge.market_data import ( + _completed_daily_closes, calculate_rotation_indicators, fetch_last_price, fetch_last_prices, @@ -19,8 +25,9 @@ def __init__(self, symbol, last_done): class FakeBar: - def __init__(self, close): + def __init__(self, close, timestamp=None): self.close = close + self.timestamp = timestamp class FakeQuoteContext: @@ -34,6 +41,26 @@ def candlesticks(self, symbol, period, count, adjust_type): return [FakeBar(200.0 + i) for i in range(count)] +@contextlib.contextmanager +def _process_timezone(tz_name: str): + """Set process-local TZ for naive datetime.astimezone semantics. + + This Python build may omit ``time.tzset``; libc ``tzset`` still applies + the ``TZ`` environment variable on macOS/Linux. + """ + previous = os.environ.get("TZ") + os.environ["TZ"] = tz_name + ctypes.CDLL(None).tzset() + try: + yield + finally: + if previous is None: + os.environ.pop("TZ", None) + else: + os.environ["TZ"] = previous + ctypes.CDLL(None).tzset() + + class LongBridgeMarketDataTests(unittest.TestCase): def test_fetch_last_price(self) -> None: self.assertEqual(fetch_last_price(FakeQuoteContext(), "SOXL.US"), 123.45) @@ -108,6 +135,118 @@ def test_calculate_rotation_indicators(self) -> None: indicators["soxx"]["realized_volatility_20"], ) + def test_completed_daily_closes_rejects_missing_stale_duplicate_and_timezone_boundary(self) -> None: + expected = pd.Timestamp("2026-09-16") + complete = [FakeBar(100.0, pd.Timestamp("2026-09-16 20:00:00+00:00"))] + self.assertEqual(_completed_daily_closes(complete, expected).iloc[-1]["close"], 100.0) + self.assertIsNone(_completed_daily_closes([FakeBar(100.0)], expected)) + self.assertIsNone(_completed_daily_closes([FakeBar(float("nan"), pd.Timestamp("2026-09-16 20:00:00+00:00"))], expected)) + self.assertIsNone(_completed_daily_closes([FakeBar(float("inf"), pd.Timestamp("2026-09-16 20:00:00+00:00"))], expected)) + self.assertIsNone(_completed_daily_closes([FakeBar(0.0, pd.Timestamp("2026-09-16 20:00:00+00:00"))], expected)) + self.assertIsNone(_completed_daily_closes([FakeBar(100.0, pd.Timestamp("2026-09-15 20:00:00+00:00"))], expected)) + self.assertIsNone(_completed_daily_closes(complete + [FakeBar(101.0, pd.Timestamp("2026-09-16 21:00:00+00:00"))], expected)) + self.assertEqual(_completed_daily_closes([FakeBar(100.0, pd.Timestamp("2026-09-17 00:30:00+00:00"))], expected).iloc[-1]["session"], expected) + + def test_completed_daily_closes_naive_process_local_dst_and_timezone_matrix(self) -> None: + # LongPort 3.x emits naive process-local wall times for broker instants. + # Recover the instant via local semantics, then map to the NY session date. + # Do not reject naive timestamps, and do not blindly stamp UTC/NY. + cases = ( + # Summer (EDT): exchange local midnight 2026-06-16 == 04:00 UTC + ("America/New_York", "2026-06-16 00:00:00", "2026-06-16"), + ("UTC", "2026-06-16 04:00:00", "2026-06-16"), + ("Asia/Shanghai", "2026-06-16 12:00:00", "2026-06-16"), + # Winter (EST): exchange local midnight 2026-01-15 == 05:00 UTC + ("America/New_York", "2026-01-15 00:00:00", "2026-01-15"), + ("UTC", "2026-01-15 05:00:00", "2026-01-15"), + ("Asia/Shanghai", "2026-01-15 13:00:00", "2026-01-15"), + ) + for process_tz, naive_wall, session_date in cases: + with self.subTest(process_tz=process_tz, naive_wall=naive_wall): + expected = pd.Timestamp(session_date) + with _process_timezone(process_tz): + # Sanity: naive wall must equal the process-local view of the + # same UTC instant that yields this NY session. + local_view = ( + pd.Timestamp(f"{session_date} 00:00:00", tz="America/New_York") + .tz_convert(process_tz) + .tz_localize(None) + ) + self.assertEqual(local_view, pd.Timestamp(naive_wall)) + frame = _completed_daily_closes( + [FakeBar(100.0, pd.Timestamp(naive_wall))], + expected, + ) + self.assertIsNotNone(frame) + self.assertEqual(frame.iloc[-1]["session"], expected) + self.assertEqual(frame.iloc[-1]["close"], 100.0) + + # Aware timestamps remain accepted and agree with the NY session calendar. + aware = FakeBar( + 100.0, + pd.Timestamp("2026-06-16 00:00:00", tz="America/New_York").tz_convert("UTC"), + ) + aware_frame = _completed_daily_closes([aware], pd.Timestamp("2026-06-16")) + self.assertIsNotNone(aware_frame) + self.assertEqual(aware_frame.iloc[-1]["session"], pd.Timestamp("2026-06-16")) + + def test_completed_session_opt_in_excludes_open_session_bar(self) -> None: + class TimedContext(FakeQuoteContext): + def candlesticks(self, symbol, period, count, adjust_type): + sessions = list(pd.bdate_range(end="2026-09-16", periods=count - 1)) + [pd.Timestamp("2026-09-17")] + base = 100.0 if symbol == "SOXL.US" else 200.0 + return [ + FakeBar( + base + index, + session.tz_localize("America/New_York") + .replace(hour=9, minute=30) + .tz_convert("UTC"), + ) + for index, session in enumerate(sessions) + ] + + longport_module = types.ModuleType("longport") + openapi_module = types.ModuleType("longport.openapi") + openapi_module.Period = types.SimpleNamespace(Day="Day") + openapi_module.AdjustType = types.SimpleNamespace(ForwardAdjust="ForwardAdjust") + with patch.dict(sys.modules, {"longport": longport_module, "longport.openapi": openapi_module}): + indicators = calculate_rotation_indicators( + TimedContext(), trend_window=150, + completed_session_date="2026-09-16", + ) + self.assertIsNotNone(indicators) + self.assertEqual(indicators["completed_session"]["date"], "2026-09-16") + self.assertEqual(indicators["soxl"]["price"], 100.0 + 418) + self.assertEqual(indicators["soxx"]["price"], 200.0 + 418) + + def test_completed_session_opt_in_rejects_missing_asset_session(self) -> None: + class MissingSessionContext(FakeQuoteContext): + def candlesticks(self, symbol, period, count, adjust_type): + sessions = list(pd.bdate_range(end="2026-09-15", periods=count - 1)) + if symbol == "SOXL.US": + sessions.append(pd.Timestamp("2026-09-17")) + base = 100.0 if symbol == "SOXL.US" else 200.0 + return [ + FakeBar( + base + index, + session.tz_localize("America/New_York") + .replace(hour=9, minute=30) + .tz_convert("UTC"), + ) + for index, session in enumerate(sessions) + ] + + longport_module = types.ModuleType("longport") + openapi_module = types.ModuleType("longport.openapi") + openapi_module.Period = types.SimpleNamespace(Day="Day") + openapi_module.AdjustType = types.SimpleNamespace(ForwardAdjust="ForwardAdjust") + with patch.dict(sys.modules, {"longport": longport_module, "longport.openapi": openapi_module}): + indicators = calculate_rotation_indicators( + MissingSessionContext(), trend_window=150, + completed_session_date="2026-09-16", + ) + self.assertIsNone(indicators) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 82025ca5..d6a5d22e 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -22,6 +22,7 @@ CandidateRiskIdentity, RiskAction, RiskSignal, + RuntimeRiskLimits, ) from quant_platform_kit.risk.engine import RiskEngine from quant_platform_kit.risk.gate import ( @@ -143,6 +144,237 @@ def test_finite_portfolio_snapshot_remains_approved(self) -> None: class ApplyRiskGateTests(unittest.TestCase): + _EIGHT_ETF_SYMBOLS = ( + "SOXL", + "SOXX", + "BOXX", + "SCHD", + "DGRO", + "SGOV", + "SPYI", + "QQQI", + ) + + @staticmethod + def _runtime_limits() -> RuntimeRiskLimits: + return RuntimeRiskLimits( + allowed_symbols=("SOXL", "SOXX", "BOXX"), + product_leverage_factors={"SOXL": 3, "SOXX": 1, "BOXX": 1}, + nominal_caps={"SOXL": 0.679, "SOXX": 0.873, "BOXX": 0.97}, + total_nominal_exposure_cap=0.97, + total_effective_exposure_cap=2.328, + max_positions=8, + ) + + @classmethod + def _approved_eight_etf_limits(cls) -> RuntimeRiskLimits: + """User-accepted SOXL runtime mapping used by the 8-ETF recovery case.""" + symbols = cls._EIGHT_ETF_SYMBOLS + return RuntimeRiskLimits( + allowed_symbols=symbols, + product_leverage_factors={ + "SOXL": 3, + **{symbol: 1 for symbol in symbols if symbol != "SOXL"}, + }, + nominal_caps={ + "SOXL": 0.679, + "SOXX": 0.873, + **{ + symbol: 0.97 + for symbol in symbols + if symbol not in {"SOXL", "SOXX"} + }, + }, + total_nominal_exposure_cap=0.97, + total_effective_exposure_cap=2.328, + max_positions=8, + ) + + def test_explicit_runtime_limits_approve_three_etf_targets(self) -> None: + result = apply_risk_gate( + _decision( + positions=( + PositionTarget(symbol="SOXL", target_weight=0.20), + PositionTarget(symbol="SOXX", target_weight=0.30), + PositionTarget(symbol="BOXX", target_weight=0.40), + ) + ), + max_single_weight=1.0, + max_total_exposure=1.0, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=self._runtime_limits(), + ) + + self.assertIn("risk_gate:passed", result.risk_flags) + self.assertEqual(len(result.positions), 3) + + def test_explicit_runtime_limits_reject_unknown_asset_and_cap_overrun(self) -> None: + for decision in ( + _decision(positions=(PositionTarget(symbol="SPY", target_weight=0.10),)), + _decision(positions=(PositionTarget(symbol="SOXL", target_weight=0.70),)), + _decision( + positions=( + PositionTarget(symbol="SOXL", target_weight=0.10), + PositionTarget(symbol="SOXL", target_weight=0.10), + ) + ), + _decision(positions=(PositionTarget(symbol="SOXL", target_weight=-0.10),)), + ): + with self.subTest(decision=decision): + result = apply_risk_gate( + decision, + max_single_weight=1.0, + max_total_exposure=1.0, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=self._runtime_limits(), + ) + self.assertEqual(result.positions, ()) + self.assertEqual(result.risk_flags, ("rejected:runtime_risk_limits",)) + + def test_explicit_runtime_limits_accept_approved_machine_precision_boundary(self) -> None: + result = apply_risk_gate( + _decision( + positions=( + PositionTarget(symbol="SOXL", target_weight=0.679), + PositionTarget(symbol="SOXX", target_weight=0.194), + PositionTarget(symbol="BOXX", target_weight=0.097), + ) + ), + max_single_weight=1.0, + max_total_exposure=1.0, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=self._runtime_limits(), + ) + + self.assertIn("risk_gate:passed", result.risk_flags) + + def test_invalid_explicit_runtime_limits_fail_closed_without_fallback(self) -> None: + result = apply_risk_gate( + _decision(positions=(PositionTarget(symbol="SOXL", target_weight=0.10),)), + max_single_weight=1.0, + max_total_exposure=1.0, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=object(), # type: ignore[arg-type] + ) + + self.assertEqual(result.positions, ()) + self.assertEqual(result.risk_flags, ("rejected:runtime_risk_limits",)) + + def test_approved_eight_etf_full_nominal_budget_approve_and_reject_overruns(self) -> None: + # Full 8-ETF allocation at the accepted total nominal boundary (0.97). + # Overrun must REJECT rather than silently scale. + full_weights = { + "SOXL": 0.20, + "SOXX": 0.11, + "BOXX": 0.11, + "SCHD": 0.11, + "DGRO": 0.11, + "SGOV": 0.11, + "SPYI": 0.11, + "QQQI": 0.11, + } + self.assertAlmostEqual(sum(full_weights.values()), 0.97) + limits = self._approved_eight_etf_limits() + + approved = apply_risk_gate( + _decision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=weight) + for symbol, weight in full_weights.items() + ) + ), + max_single_weight=1.0, + max_total_exposure=1.0, + max_positions=8, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=limits, + ) + self.assertIn("risk_gate:passed", approved.risk_flags) + self.assertEqual(len(approved.positions), 8) + + over_total = dict(full_weights) + over_total["QQQI"] = 0.111 + rejected_total = apply_risk_gate( + _decision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=weight) + for symbol, weight in over_total.items() + ) + ), + max_single_weight=1.0, + max_total_exposure=1.0, + max_positions=8, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=limits, + ) + self.assertEqual(rejected_total.positions, ()) + self.assertEqual(rejected_total.risk_flags, ("rejected:runtime_risk_limits",)) + + over_soxl = { + "SOXL": 0.680, + "SOXX": 0.05, + "BOXX": 0.04, + "SCHD": 0.04, + "DGRO": 0.04, + "SGOV": 0.04, + "SPYI": 0.04, + "QQQI": 0.03, + } + self.assertLess(sum(over_soxl.values()), 0.97) + rejected_soxl = apply_risk_gate( + _decision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=weight) + for symbol, weight in over_soxl.items() + ) + ), + max_single_weight=1.0, + max_total_exposure=1.0, + max_positions=8, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=limits, + ) + self.assertEqual(rejected_soxl.positions, ()) + self.assertEqual(rejected_soxl.risk_flags, ("rejected:runtime_risk_limits",)) + + def test_missing_runtime_risk_limits_falls_back_to_legacy_single_position_rule(self) -> None: + # When capability is absent (None), gate keeps the pre-authorization + # path: unauthorized configs may hold only one nonzero position. + # Callers must not treat missing limits as permission for the 8-ETF book. + eight_etf = _decision( + positions=tuple( + PositionTarget(symbol=symbol, target_weight=0.10) + for symbol in self._EIGHT_ETF_SYMBOLS + ) + ) + + result = apply_risk_gate( + eight_etf, + max_single_weight=1.0, + max_total_exposure=1.0, + max_positions=8, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=None, + ) + + self.assertEqual(result.positions, ()) + self.assertEqual(result.risk_flags, ("rejected:too_many_positions",)) + + def test_explicit_runtime_limits_reject_uncovered_budget_symbol(self) -> None: + result = apply_risk_gate( + StrategyDecision( + positions=(PositionTarget(symbol="SOXL", target_weight=0.10),), + budgets=(BudgetIntent(name="reserve", symbol="SPY", amount=100.0),), + ), + max_single_weight=1.0, + max_total_exposure=1.0, + portfolio_snapshot=_portfolio_snapshot(), + runtime_risk_limits=self._runtime_limits(), + ) + + self.assertEqual(result.positions, ()) + self.assertEqual(result.risk_flags, ("rejected:runtime_risk_limits",)) + def test_no_mandate_does_not_allow_caller_to_expand_default_cap(self) -> None: decision = _decision( positions=(PositionTarget(symbol="SPY", target_weight=0.11),), From 7fcf22adba7914b23df05613034bdbc72beb211a Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 17 Sep 2026 05:23:01 +0800 Subject: [PATCH 2/2] test: freeze research resume clocks past the 7-day observation TTL Fixture as_of dates aged past max_age_days on wall-clock CI after 2026-09-14, parking cycles as observation_stale unrelated to RuntimeRiskLimits. Co-authored-by: Cursor --- tests/test_research_promotion_resume.py | 32 ++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 3 deletions(-) diff --git a/tests/test_research_promotion_resume.py b/tests/test_research_promotion_resume.py index 3ba9a7fc..f9b739f0 100644 --- a/tests/test_research_promotion_resume.py +++ b/tests/test_research_promotion_resume.py @@ -1,7 +1,7 @@ """Durable research stages, using synthetic evidence and no external services.""" from dataclasses import replace -from datetime import date, datetime, timezone +from datetime import date, datetime, timedelta, timezone from pathlib import Path from unittest.mock import Mock, patch @@ -57,6 +57,7 @@ def test_new_request_uses_shared_cycle_without_fabricating_drift(tmp_path): new_request=_new_request(), research_identity=IDENTITY, ticket_dir=tmp_path, optimize=optimize, enforce_backtest_gates=Mock(return_value=_promotion_backtest_evidence()), record_shadow=Mock(return_value={"evidence_kind": "paired_shadow", "passed": True}), + evaluation_date=date(2026, 9, 8), ) assert result["status"] == "awaiting_human" assert result["ticket"]["drift_score"] is None @@ -70,6 +71,7 @@ def test_same_new_request_reuses_completed_stages(tmp_path): new_request=_new_request(), research_identity=IDENTITY, ticket_dir=tmp_path, optimize=optimize, enforce_backtest_gates=Mock(return_value=_promotion_backtest_evidence()), record_shadow=Mock(return_value={"evidence_kind": "paired_shadow", "passed": True}), + evaluation_date=date(2026, 9, 8), ) first = cycle.run_saved_research_promotion_cycle(**kwargs) second = cycle.run_saved_research_promotion_cycle(**kwargs) @@ -441,13 +443,15 @@ def test_normal_cycle_reuses_declined_diagnosis_across_runs(tmp_path): from quant_platform_kit.strategy_lifecycle.codex_integration import run_auto_pilot_cycle from quant_platform_kit.strategy_lifecycle.contracts import StrategyPerformanceSnapshot + fresh_as_of = date.today() - timedelta(days=1) + fresh_drift = replace(_drift(), as_of=fresh_as_of, source_revision="observation-v1") store = Mock(local_root=tmp_path) store.load_latest_snapshot.return_value = StrategyPerformanceSnapshot( - strategy_profile="demo_strategy", domain="us_equity", platform="test", as_of=date(2026, 9, 7), + strategy_profile="demo_strategy", domain="us_equity", platform="test", as_of=fresh_as_of, source_revision="observation-v1") prefix = "quant_platform_kit.strategy_lifecycle.codex_integration." with patch(prefix + "_run_monitor_phase", return_value=[]), \ - patch(prefix + "_run_drift_phase", return_value=([_drift()], [_drift()])), \ + patch(prefix + "_run_drift_phase", return_value=([fresh_drift], [fresh_drift])), \ patch(prefix + "_drift_freshness_reason", return_value=None), \ patch(prefix + "call_ai_optimization_decision", return_value={"optimization_needed": False}) as ai: for _ in range(2): @@ -734,8 +738,19 @@ def stop_before_delivery(ticket, path): sync, optimize, gates, shadow = Mock(return_value=True), Mock(), Mock(), Mock() prefix = "quant_platform_kit.strategy_lifecycle.codex_integration." identity = {**IDENTITY, "input_revision": "new-input"} if identity_changes else IDENTITY + + class _FrozenDateTime(datetime): + @classmethod + def now(cls, tz=None): + return datetime(2026, 9, 8, tzinfo=timezone.utc) + with patch(prefix + "_run_monitor_phase", return_value=[]), \ patch(prefix + "_run_drift_phase", return_value=([_drift()], [_drift()])), \ + patch(prefix + "_drift_freshness_reason", return_value=None), \ + patch( + "quant_platform_kit.strategy_lifecycle.production_drift_health_probe.datetime", + _FrozenDateTime, + ), \ patch(prefix + "call_ai_optimization_decision") as ai: run_auto_pilot_cycle("us_equity", store=store, create_issues=False, research_identity=identity, optimize=optimize, enforce_backtest_gates=gates, record_shadow=shadow, @@ -803,8 +818,19 @@ def test_actual_auto_pilot_passes_new_admission_before_ai(tmp_path): source_revision="observation-v1")) admission = Mock(return_value=False) prefix = "quant_platform_kit.strategy_lifecycle.codex_integration." + + class _FrozenDateTime(datetime): + @classmethod + def now(cls, tz=None): + return datetime(2026, 9, 8, tzinfo=timezone.utc) + with patch(prefix + "_run_monitor_phase", return_value=[]), \ patch(prefix + "_run_drift_phase", return_value=([_drift()], [_drift()])), \ + patch(prefix + "_drift_freshness_reason", return_value=None), \ + patch( + "quant_platform_kit.strategy_lifecycle.production_drift_health_probe.datetime", + _FrozenDateTime, + ), \ patch(prefix + "call_ai_optimization_decision") as ai: result = run_auto_pilot_cycle("us_equity", store=store, create_issues=False, research_identity=IDENTITY, optimize=Mock(), enforce_backtest_gates=Mock(), record_shadow=Mock(), admit_new_research=admission)