fix: aggregate product caps and reject duplicate return dates

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")

src/quant_platform_kit/strategy_lifecycle/performance_metrics.py

@@ -21,28 +21,33 @@
 
 
 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(
     frame: pd.DataFrame,
     *,
     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()
         df = df.dropna(subset=[date_column]).set_index(date_column)
     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()

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")

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()

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),),