fix: use historical correlation percentiles

src/quant_platform_kit/strategy_lifecycle/market_regime.py

@@ -142,6 +142,7 @@ class RegimeDetector:
     VOL_PERCENTILE_STRESS = 0.90      # vol above 90th %ile → stress
     CORR_PERCENTILE_STRESS = 0.85     # correlation above 85th %ile → stress
     MIN_HISTORY_DAYS = 252            # minimum history for percentile calculation
+    CORRELATION_WINDOW_DAYS = 60
 
     def __init__(self, *, benchmark_returns: pd.Series | None = None):
         self._benchmark = normalize_return_series(benchmark_returns) if benchmark_returns is not None else None
@@ -229,39 +230,54 @@ def _compute_correlation_metrics(
         benchmark: pd.Series,
         universe: pd.DataFrame | None,
     ) -> tuple[float, float]:
-        """Compute average pairwise correlation from a universe of returns."""
+        """Compute average pairwise correlation and its rolling-history percentile.
+
+        A correlation value is not itself a percentile: mapping ``[-1, 1]`` to
+        ``[0, 1]`` conflates high absolute correlation with correlation that is
+        unusually high for this universe.  We instead compare the current
+        60-day average pairwise correlation against the historical sequence of
+        equivalent 60-day observations.  The method deliberately returns the
+        neutral fallback when there is less than one full year of aligned
+        history, rather than manufacturing precision from a short sample.
+        """
         if universe is None or universe.empty:
             return float("nan"), 0.5
 
         frame = pd.DataFrame(universe).copy()
         # Align with benchmark
-        common = frame.index.intersection(benchmark.index)
-        if len(common) < 20:
+        common = frame.index.intersection(benchmark.index).sort_values()
+        if len(common) < self.MIN_HISTORY_DAYS:
             return float("nan"), 0.5
 
         # Use up to 10 columns to keep it cheap
         cols = [c for c in frame.columns if str(c).strip() and not str(c).startswith("buy_hold_")][:10]
         if len(cols) < 3:
             return float("nan"), 0.5
 
-        # 60-day rolling correlation
-        corr_matrix = frame[cols].tail(60).corr()
-        # Average of lower triangle (excluding diagonal)
-        n = len(cols)
-        if n < 2:
-            return float("nan"), 0.5
-
-        values = []
-        for i in range(n):
-            for j in range(i + 1, n):
-                values.append(corr_matrix.iloc[i, j])
-
-        avg = float(np.mean(values)) if values else float("nan")
+        aligned = frame.loc[common, cols]
+        rolling_averages = []
+        for end in range(self.CORRELATION_WINDOW_DAYS, len(aligned) + 1):
+            average = self._average_pairwise_correlation(
+                aligned.iloc[end - self.CORRELATION_WINDOW_DAYS:end]
+            )
+            if np.isfinite(average):
+                rolling_averages.append(average)
 
-        # Simple percentile: 0.5 is the default "normal"
-        corr_percentile = min(max((avg + 1.0) / 2.0, 0.0), 1.0)  # map [-1,1] → [0,1]
+        if not rolling_averages:
+            return float("nan"), 0.5
 
-        return avg, corr_percentile
+        average = rolling_averages[-1]
+        history = np.asarray(rolling_averages, dtype=float)
+        percentile = float((history < average).mean())
+        return average, percentile
+
+    @staticmethod
+    def _average_pairwise_correlation(frame: pd.DataFrame) -> float:
+        """Return the finite lower-triangle mean of a correlation matrix."""
+        corr_matrix = frame.corr()
+        values = corr_matrix.to_numpy()[np.tril_indices(len(corr_matrix), k=-1)]
+        finite_values = values[np.isfinite(values)]
+        return float(np.mean(finite_values)) if len(finite_values) else float("nan")
 
 
 # ── Convenience factory ──────────────────────────────────────────────

tests/test_market_regime.py

@@ -0,0 +1,51 @@
+import numpy as np
+import pandas as pd
+import pytest
+
+from quant_platform_kit.strategy_lifecycle.market_regime import RegimeDetector
+
+
+def test_correlation_percentile_compares_equivalent_rolling_windows():
+    """The percentile must be empirical, not a remapping of correlation [-1, 1]."""
+    rng = np.random.default_rng(17)
+    index = pd.bdate_range("2025-01-01", periods=320)
+    benchmark = pd.Series(rng.normal(0, 0.01, len(index)), index=index)
+    universe = pd.DataFrame(
+        {
+            "a": rng.normal(0, 0.01, len(index)),
+            "b": rng.normal(0, 0.01, len(index)),
+            "c": rng.normal(0, 0.01, len(index)),
+        },
+        index=index,
+    )
+    # Make only the latest correlation window materially more correlated.
+    anchor = universe["a"].iloc[-60:].to_numpy()
+    universe.loc[index[-60:], "b"] = 0.6 * anchor + 0.8 * rng.normal(0, 0.01, 60)
+    universe.loc[index[-60:], "c"] = 0.6 * anchor + 0.8 * rng.normal(0, 0.01, 60)
+
+    detector = RegimeDetector(benchmark_returns=benchmark)
+    average, percentile = detector._compute_correlation_metrics(benchmark, universe)
+
+    history = []
+    for end in range(detector.CORRELATION_WINDOW_DAYS, len(universe) + 1):
+        history.append(
+            detector._average_pairwise_correlation(
+                universe.iloc[end - detector.CORRELATION_WINDOW_DAYS:end]
+            )
+        )
+    expected_percentile = float((np.asarray(history) < history[-1]).mean())
+
+    assert average == pytest.approx(history[-1])
+    assert percentile == pytest.approx(expected_percentile)
+    assert percentile != pytest.approx((average + 1.0) / 2.0)
+
+
+def test_correlation_percentile_falls_back_without_full_year_of_aligned_data():
+    index = pd.bdate_range("2025-01-01", periods=251)
+    benchmark = pd.Series(0.001, index=index)
+    universe = pd.DataFrame({"a": 0.001, "b": 0.002, "c": 0.003}, index=index)
+
+    average, percentile = RegimeDetector()._compute_correlation_metrics(benchmark, universe)
+
+    assert np.isnan(average)
+    assert percentile == 0.5