fix(ops): Corr crashes in expanding mode (N=0)

qlib/data/ops.py

@@ -1491,10 +1491,13 @@ def _load_internal(self, instrument, start_index, end_index, *args):
         # NOTE: Load uses MemCache, so calling load again will not cause performance degradation
         series_left = self.feature_left.load(instrument, start_index, end_index, *args)
         series_right = self.feature_right.load(instrument, start_index, end_index, *args)
-        res.loc[
-            np.isclose(series_left.rolling(self.N, min_periods=1).std(), 0, atol=2e-05)
-            | np.isclose(series_right.rolling(self.N, min_periods=1).std(), 0, atol=2e-05)
-        ] = np.nan
+        if self.N == 0:
+            std_left = series_left.expanding(min_periods=1).std()
+            std_right = series_right.expanding(min_periods=1).std()
+        else:
+            std_left = series_left.rolling(self.N, min_periods=1).std()
+            std_right = series_right.rolling(self.N, min_periods=1).std()
+        res.loc[np.isclose(std_left, 0, atol=2e-05) | np.isclose(std_right, 0, atol=2e-05)] = np.nan
         return res
 
 

tests/ops/test_pair_rolling.py

@@ -0,0 +1,61 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+import unittest
+
+import numpy as np
+import pandas as pd
+
+from qlib.data.base import Expression
+from qlib.data.ops import Corr
+
+
+class StubFeature(Expression):
+    """Minimal expression returning a fixed series, so operators can be tested without a data provider."""
+
+    def __init__(self, values, name):
+        self._values = list(values)
+        self._name = name
+
+    def __str__(self):
+        # unique per stub: Expression.load caches series keyed on str(self)
+        return self._name
+
+    def _load_internal(self, instrument, start_index, end_index, *args):
+        return pd.Series(self._values, dtype=float)
+
+    def get_longest_back_rolling(self):
+        return 0
+
+    def get_extended_window_size(self):
+        return 0, 0
+
+
+class TestCorrExpanding(unittest.TestCase):
+    """Corr(x, y, 0) must work in expanding mode instead of crashing on rolling(0)."""
+
+    def test_corr_expanding_no_crash(self):
+        left = [1.0, 2.0, 3.0, 4.0, 5.0]
+        right = [2.0, 1.0, 4.0, 3.0, 6.0]
+        res = Corr(StubFeature(left, "exp_l"), StubFeature(right, "exp_r"), 0).load("inst", 0, 4, "day")
+        expected = pd.Series(left).expanding(min_periods=1).corr(pd.Series(right))
+        np.testing.assert_allclose(res.values, expected.values)
+
+    def test_corr_expanding_masks_constant_series(self):
+        # while the left series is constant its std is 0, so corr must be masked to NaN
+        left = [1.0, 1.0, 1.0, 2.0]
+        right = [2.0, 1.0, 4.0, 3.0]
+        res = Corr(StubFeature(left, "mask_l"), StubFeature(right, "mask_r"), 0).load("inst", 0, 3, "day")
+        self.assertTrue(np.isnan(res.values[:3]).all())
+        self.assertTrue(np.isfinite(res.values[3]))
+
+    def test_corr_rolling_unchanged(self):
+        left = [1.0, 2.0, 3.0, 4.0, 5.0]
+        right = [2.0, 1.0, 4.0, 3.0, 6.0]
+        res = Corr(StubFeature(left, "roll_l"), StubFeature(right, "roll_r"), 3).load("inst", 0, 4, "day")
+        expected = pd.Series(left).rolling(3, min_periods=1).corr(pd.Series(right))
+        np.testing.assert_allclose(res.values, expected.values)
+
+
+if __name__ == "__main__":
+    unittest.main()