From 08239ce1f554dfc26c75bc93743e1a1de18565b6 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 10 Sep 2026 02:37:22 +0800 Subject: [PATCH] fix: enforce aggregate product caps and unique daily returns Co-Authored-By: Codex --- src/quant_platform_kit/risk/gate.py | 7 +- .../strategy_lifecycle/performance_metrics.py | 11 ++- tests/test_lifecycle_metrics.py | 25 ++++++ tests/test_lifecycle_performance_monitor.py | 65 ++++++++++++++ tests/test_risk_gate.py | 90 +++++++++++++++++++ 5 files changed, 190 insertions(+), 8 deletions(-) diff --git a/src/quant_platform_kit/risk/gate.py b/src/quant_platform_kit/risk/gate.py index a2b377ef..e08c74b1 100644 --- a/src/quant_platform_kit/risk/gate.py +++ b/src/quant_platform_kit/risk/gate.py @@ -1756,11 +1756,8 @@ def _assess_with_evidence_static( reason_codes.add("invalid_risk_metadata") continue target_weights[symbol] = combined_weight - policy_positions = ( - list(target_weights.items()) - if is_tqqq_evidence_mandate - else active_positions - ) + # Product caps apply to the combined holding, regardless of row/role. + policy_positions = list(target_weights.items()) weighted_exposure = 0.0 if mandate_provenance is None and len(active_positions) > 1: reason_codes.add("fallback_position_count") diff --git a/src/quant_platform_kit/strategy_lifecycle/performance_metrics.py b/src/quant_platform_kit/strategy_lifecycle/performance_metrics.py index ab7868b4..148784b8 100644 --- a/src/quant_platform_kit/strategy_lifecycle/performance_metrics.py +++ b/src/quant_platform_kit/strategy_lifecycle/performance_metrics.py @@ -21,13 +21,16 @@ def normalize_return_series(series: pd.Series) -> pd.Series: - """Clean and normalize a daily return series.""" + """Clean daily returns; reject repeated dates rather than compound twice.""" s = pd.Series(series).copy() if not pd.api.types.is_datetime64_any_dtype(s.index): s.index = pd.to_datetime(s.index, errors="coerce") s.index = s.index.tz_localize(None).normalize() + s = s.loc[s.index.notna()] + if s.index.has_duplicates: + raise ValueError("daily return dates must be unique") s = pd.to_numeric(s, errors="coerce") - return s.loc[s.index.notna()].dropna().sort_index() + return s.dropna().sort_index() def normalize_return_matrix( @@ -35,7 +38,7 @@ def normalize_return_matrix( *, date_column: str = "as_of", ) -> pd.DataFrame: - """Normalize a return matrix: datetime index, numeric values.""" + """Normalize a daily return matrix, rejecting repeated normalized dates.""" df = pd.DataFrame(frame).copy() if date_column in df.columns: df[date_column] = pd.to_datetime(df[date_column], errors="coerce").dt.tz_localize(None).dt.normalize() @@ -43,6 +46,8 @@ def normalize_return_matrix( else: df.index = pd.to_datetime(df.index, errors="coerce").tz_localize(None).normalize() df = df.loc[df.index.notna()] + if df.index.has_duplicates: + raise ValueError("daily return dates must be unique") for col in df.columns: df[col] = pd.to_numeric(df[col], errors="coerce") return df.sort_index() diff --git a/tests/test_lifecycle_metrics.py b/tests/test_lifecycle_metrics.py index f41dd261..d7984b6b 100644 --- a/tests/test_lifecycle_metrics.py +++ b/tests/test_lifecycle_metrics.py @@ -10,6 +10,7 @@ from quant_platform_kit.strategy_lifecycle.performance_metrics import ( compute_window_metrics, compute_windows, + normalize_return_matrix, normalize_return_series, DEFAULT_WINDOWS, ) @@ -27,6 +28,30 @@ def test_normalize_return_series(self) -> None: # Index should be datetime self.assertTrue(pd.api.types.is_datetime64_any_dtype(s.index)) + def test_daily_return_normalization_rejects_duplicate_dates(self) -> None: + for dates in ( + ["2026-09-08", "2026-09-08"], + ["2026-09-08T09:00:00", "2026-09-08T16:00:00"], + ["2026-09-08T09:00:00+08:00", "2026-09-08T16:00:00+08:00"], + ): + for values in ([0.01, 0.01], [0.01, -0.02]): + series = pd.Series(values, index=pd.to_datetime(dates)) + with self.subTest(dates=dates, values=values): + for operation in ( + lambda: normalize_return_series(series), + lambda: normalize_return_matrix(series.to_frame("strategy")), + lambda: normalize_return_matrix(pd.DataFrame({"as_of": dates, "strategy": values})), + lambda: compute_window_metrics(series, window_days=1), + ): + with self.assertRaisesRegex(ValueError, "^daily return dates must be unique$"): + operation() + + def test_duplicate_benchmark_dates_are_rejected_before_comparison(self) -> None: + returns = pd.Series([0.01, -0.02], index=self.dates[:2]) + benchmark = pd.Series([0.01, 0.01], index=[self.dates[0], self.dates[0]]) + with self.assertRaisesRegex(ValueError, "^daily return dates must be unique$"): + compute_window_metrics(returns, benchmark_returns=benchmark) + def test_compute_window_metrics_basic(self) -> None: r = self.returns wp = compute_window_metrics(r, window_days=126, window_label="test_6m") diff --git a/tests/test_lifecycle_performance_monitor.py b/tests/test_lifecycle_performance_monitor.py index b354fdd3..3a73157c 100644 --- a/tests/test_lifecycle_performance_monitor.py +++ b/tests/test_lifecycle_performance_monitor.py @@ -3,6 +3,7 @@ import tempfile import unittest from pathlib import Path +from unittest.mock import patch import pandas as pd @@ -14,6 +15,7 @@ try_record_platform_execution, ) from quant_platform_kit.strategy_lifecycle.performance_store import PerformanceStore +from quant_platform_kit.strategy_lifecycle.return_collector import ReturnCollector class PerformanceMonitorTests(unittest.TestCase): @@ -71,6 +73,69 @@ def collect(self, _domain: str) -> dict[str, pd.Series]: with self.assertRaisesRegex(RuntimeError, "No strategy return series found"): run_monitor("us_equity", collector=EmptyCollector()) + def test_csv_collector_to_monitor_rejects_duplicate_daily_returns(self) -> None: + for dates, values in ( + (["2026-09-08", "2026-09-08"], [0.01, 0.01]), + (["2026-09-08T09:00:00", "2026-09-08T16:00:00"], [0.01, -0.02]), + ): + with self.subTest(dates=dates), tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + matrix = root / "portfolio_and_tracker_returns.csv" + pd.DataFrame({"as_of": dates, "synthetic_soxl": values}).to_csv(matrix, index=False) + store = PerformanceStore(local_root=root / "store") + collector = ReturnCollector(artifact_roots={"us_equity": root}, projects_root=root, store=store) + with patch.object(PerformanceStore, "save_snapshot", autospec=True) as save: + with self.assertRaisesRegex(RuntimeError, "No strategy return series found"): + run_monitor("us_equity", strategy_profile="synthetic_soxl", collector=collector, + store=store, windows=(2,), min_observations=2) + self.assertEqual(run_monitor( + "us_equity", collector=collector, store=store, min_observations=2, fail_on_empty=False, + ), []) + save.assert_not_called() + + def test_csv_collector_to_monitor_preserves_unique_daily_returns(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + pd.DataFrame({ + "as_of": ["2026-09-08T16:00:00", "2026-09-09T16:00:00"], + "synthetic_soxl": [0.01, -0.02], + "SPY": [0.01, -0.02], + }).to_csv(root / "portfolio_and_tracker_returns.csv", index=False) + store = PerformanceStore(local_root=root / "store") + collector = ReturnCollector(artifact_roots={"us_equity": root}, projects_root=root, store=store) + snapshots = run_monitor( + "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store, + windows=(2,), min_observations=2, require_explicit_benchmark=True, + strategy_benchmarks={"synthetic_soxl": "SPY"}, + ) + self.assertEqual(len(snapshots), 1) + metrics = snapshots[0].windows[2] + self.assertEqual(metrics.observation_count, 2) + self.assertAlmostEqual(metrics.total_return, 1.01 * 0.98 - 1.0) + self.assertAlmostEqual(metrics.excess_cagr, 0.0) + + def test_csv_collector_to_monitor_rejects_duplicate_required_benchmark(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + pd.DataFrame({"as_of": ["2026-09-08", "2026-09-09"], "synthetic_soxl": [0.01, -0.02]}).to_csv( + root / "portfolio_and_tracker_returns.csv", index=False, + ) + benchmark_dir = root / "benchmark" + benchmark_dir.mkdir() + pd.DataFrame({"as_of": ["2026-09-08", "2026-09-08"], "SPY": [0.01, -0.02]}).to_csv( + benchmark_dir / "portfolio_and_tracker_returns.csv", index=False, + ) + store = PerformanceStore(local_root=root / "store") + collector = ReturnCollector(artifact_roots={"us_equity": root}, projects_root=root, store=store) + with patch.object(PerformanceStore, "save_snapshot", autospec=True) as save: + with self.assertRaisesRegex(RuntimeError, "explicit benchmark data is unavailable or insufficient"): + run_monitor( + "us_equity", strategy_profile="synthetic_soxl", collector=collector, store=store, + windows=(2,), min_observations=2, require_explicit_benchmark=True, + strategy_benchmarks={"synthetic_soxl": "SPY"}, + ) + save.assert_not_called() + if __name__ == "__main__": unittest.main() diff --git a/tests/test_risk_gate.py b/tests/test_risk_gate.py index 209dd823..82025ca5 100644 --- a/tests/test_risk_gate.py +++ b/tests/test_risk_gate.py @@ -738,6 +738,96 @@ def test_approved_receipt_is_immutable_and_redacts_digest_inputs(self) -> None: self.assertEqual(first.assessment.assessment_sha256, redacted_equivalent.assessment.assessment_sha256) self.assertEqual(len(first.decision.positions), 1) + def test_generic_product_caps_apply_to_combined_symbol_targets(self) -> None: + candidate = self._candidate(strategy_profile="soxl_soxx_trend_income") + for cap_field, cap in ( + ("product_caps", 0.10), + ("nominal_caps", 0.10), + ("product_effective_caps", 0.30), + ): + mandate = self._mandate( + candidate, + product_leverage_factors={"SOXL": 3}, + allowed_nonzero_assets=["SOXL"], + **{cap_field: {"SOXL": cap}}, + ) + for mode in ("weight", "value", "mixed"): + positions = ( + PositionTarget(symbol="SOXL", target_weight=0.08, role="core") + if mode != "value" else + PositionTarget(symbol="SOXL", target_value=8_000.0, role="core"), + PositionTarget(symbol="SOXL", target_weight=0.08, role="income") + if mode == "weight" else + PositionTarget(symbol="SOXL", target_value=8_000.0, role="income"), + ) + with self.subTest(cap_field=cap_field, mode=mode), patch( + "quant_platform_kit.risk.gate._utc_now", return_value=self._NOW, + ): + result = assess_with_evidence( + _decision(positions=positions), self._snapshot(), scope="ACCOUNT", + mandate_provenance=mandate, candidate_identity=candidate, market_data={}, + capital_base=_capital_base(as_of=self._NOW), capital_base_binding=_capital_base_binding(), + ) + self.assertEqual(result.assessment.outcome, "REJECT") + reason = "product_effective_exposure_cap" if cap_field == "product_effective_caps" else "product_exposure_cap" + self.assertIn(reason, result.assessment.reason_codes) + self.assertAlmostEqual(result.assessment.proposed_effective_exposure, 0.48) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(result.decision.positions, ()) + + def test_generic_split_targets_at_cap_preserve_decision_identity(self) -> None: + candidate = self._candidate(strategy_profile="soxl_soxx_trend_income") + mandate = self._mandate( + candidate, product_leverage_factors={"SOXL": 3}, allowed_nonzero_assets=["SOXL"], + product_caps={"SOXL": 0.10}, nominal_caps={"SOXL": 0.10}, + product_effective_caps={"SOXL": 0.30}, + ) + positions = ( + PositionTarget(symbol="SOXL", target_weight=0.04, role="core"), + PositionTarget(symbol="SOXL", target_value=6_000.0, role="income"), + PositionTarget(symbol="SOXL", target_weight=0.0, role="inactive"), + ) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + split = assess_with_evidence( + _decision(positions=positions), self._snapshot(), scope="ACCOUNT", + mandate_provenance=mandate, candidate_identity=candidate, market_data={}, + capital_base=_capital_base(as_of=self._NOW), capital_base_binding=_capital_base_binding(), + ) + single = assess_with_evidence( + _decision(positions=(PositionTarget(symbol="SOXL", target_weight=0.10),)), + self._snapshot(), scope="ACCOUNT", mandate_provenance=mandate, + candidate_identity=candidate, market_data={}, + capital_base=_capital_base(as_of=self._NOW), capital_base_binding=_capital_base_binding(), + ) + for result in (split, single): + self.assertEqual(result.assessment.outcome, "APPROVE") + self.assertAlmostEqual(result.assessment.proposed_effective_exposure, 0.30) + self.assertFalse(result.assessment.execution_authorized) + self.assertEqual(split.decision.positions, positions) + self.assertNotEqual(split.assessment.decision_digest_sha256, single.assessment.decision_digest_sha256) + + def test_generic_split_targets_still_obey_account_exposure_cap(self) -> None: + candidate = self._candidate(strategy_profile="soxl_soxx_trend_income") + mandate = self._mandate( + candidate, product_leverage_factors={"SOXL": 3, "SOXX": 1}, + allowed_nonzero_assets=["SOXL", "SOXX"], + ) + decision = _decision(positions=( + PositionTarget(symbol="SOXL", target_weight=0.08, role="core"), + PositionTarget(symbol="SOXL", target_value=8_000.0, role="income"), + PositionTarget(symbol="SOXX", target_weight=0.03), + )) + with patch("quant_platform_kit.risk.gate._utc_now", return_value=self._NOW): + result = assess_with_evidence( + decision, self._snapshot(), scope="ACCOUNT", mandate_provenance=mandate, + candidate_identity=candidate, market_data={}, + capital_base=_capital_base(as_of=self._NOW), capital_base_binding=_capital_base_binding(), + ) + self.assertEqual(result.assessment.outcome, "REJECT") + self.assertIn("effective_exposure_cap", result.assessment.reason_codes) + self.assertAlmostEqual(result.assessment.proposed_effective_exposure, 0.51) + self.assertFalse(result.assessment.execution_authorized) + def test_mandate_requires_typed_candidate_and_still_assesses_once(self) -> None: decision = _decision( positions=(PositionTarget(symbol="BTCUSDT", target_weight=0.10),),