import numpy as np import pytest import pandas as pd import pandas._testing as tm from pandas.core.algorithms import safe_sort @pytest.fixture( params=[ pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=[1, 0]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=[1, 1]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=["C", "C"]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=[1.0, 0]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=[0.0, 1]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=["C", 1]), pd.DataFrame( [[2.0, 4.0], [1.0, 2.0], [5.0, 2.0], [8.0, 1.0]], columns=[1, 0.0] ), pd.DataFrame([[2, 4.0], [1, 2.0], [5, 2.0], [8, 1.0]], columns=[0, 1.0]), pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1.0]], columns=[1.0, "X"]), ] ) def pairwise_frames(request): """Pairwise frames test_pairwise""" return request.param @pytest.fixture def pairwise_target_frame(): """Pairwise target frame for test_pairwise""" return pd.DataFrame([[2, 4], [1, 2], [5, 2], [8, 1]], columns=[0, 1]) @pytest.fixture def pairwise_other_frame(): """Pairwise other frame for test_pairwise""" return pd.DataFrame( [[None, 1, 1], [None, 1, 2], [None, 3, 2], [None, 8, 1]], columns=["Y", "Z", "X"], ) def test_rolling_cov(series): A = series B = A + np.random.default_rng(2).standard_normal(len(A)) result = A.rolling(window=50, min_periods=25).cov(B) tm.assert_almost_equal(result.iloc[-1], np.cov(A[-50:], B[-50:])[0, 1]) def test_rolling_corr(series): A = series B = A + np.random.default_rng(2).standard_normal(len(A)) result = A.rolling(window=50, min_periods=25).corr(B) tm.assert_almost_equal(result.iloc[-1], np.corrcoef(A[-50:], B[-50:])[0, 1]) @pytest.mark.parametrize( "method,expected_values", [ ("corr", [np.nan] * 5), ("cov", [np.nan, np.nan, np.nan, np.nan, 0.0]), ], ) def test_rolling_cov_corr_degenerate(method, expected_values): # GH#24019 a = pd.Series([1e5, 0, 0, 0, 0]) b = pd.Series([9.45] * 5) wind = a.rolling(5) result = getattr(wind, method)(b) expected = pd.Series(expected_values) tm.assert_series_equal(result, expected) @pytest.mark.parametrize( "method,expected_last", [("cov", 50.0), ("corr", 1.0)], ) def test_rolling_cov_corr_nan_gap_recovery(method, expected_last): # GH#65739 a window whose valid-pair count drops to 0 must not poison # subsequent overlapping windows with NaN a = pd.Series([1.0, 10.0, 20.0, 30.0]) b = pd.Series([1.0, np.nan, 20.0, 30.0]) result = getattr(a.rolling(2), method)(b) expected = pd.Series([np.nan, np.nan, np.nan, expected_last]) tm.assert_series_equal(result, expected) @pytest.mark.parametrize("method", ["cov", "corr"]) @pytest.mark.parametrize("swap_xy", [False, True]) @pytest.mark.parametrize("offset_y", [0.0, 1e13]) def test_rolling_cov_corr_outlier_exit(method, swap_xy, offset_y): # GH#65739 once a large value leaves the window, the running accumulators # keep round-off garbage on the scale of that value; every later window was # silently wrong. swap_xy covers the value landing in either operand, and # offset_y the case where the other operand is dominated by a shared offset, # which used to cost precision in every deviation taken against its mean. window = 9 values_x = [3, 3, 7, 9, 3, 3, 3.8e12, 3, 8, 2, 2, 8, 6, 7, 7, 3, 8, 4] values_y = [ offset_y + value for value in [6, 3, 3, 5, 9, 1, 2, 9, 2, 1, 4, 6, 6, 9, 7, 9, 3, 5] ] if swap_xy: values_x, values_y = values_y, values_x series_x = pd.Series(values_x, dtype="float64") series_y = pd.Series(values_y, dtype="float64") result = getattr(series_x.rolling(window), method)(series_y) # oracle: NumPy's two-pass computation over each window on its own, which # shares no code with the incremental accumulators under test. Both results # are unchanged by translation, so each window is recentred first -- without # that the oracle itself loses ~1e-8 on the offset_y case. pairwise = np.cov if method == "cov" else np.corrcoef array_x = series_x.to_numpy() array_y = series_y.to_numpy() expected = [np.nan] * (window - 1) for stop in range(window, len(values_x) + 1): chunk_x = array_x[stop - window : stop] chunk_y = array_y[stop - window : stop] expected.append(pairwise(chunk_x - chunk_x[0], chunk_y - chunk_y[0])[0, 1]) tm.assert_series_equal(result, pd.Series(expected), rtol=1e-12, atol=0) @pytest.mark.parametrize("method", ["cov", "corr"]) @pytest.mark.parametrize("offset", [1e10, 1e14]) def test_rolling_cov_corr_shared_offset(method, offset): # GH#65739 an offset shared by the whole series left almost no significant # digits in the deviations the accumulators are built from, so results were # wrong with no outlier anywhere in the data -- e.g. prices, or epoch # timestamps around 1.7e18 window = 5 series_x = pd.Series([1, 2, 4, 7, 3, 5, 9, 2, 6, 8], dtype="float64") + offset series_y = pd.Series([2, 1, 5, 3, 8, 4, 7, 9, 1, 6], dtype="float64") + offset result = getattr(series_x.rolling(window), method)(series_y) pairwise = np.cov if method == "cov" else np.corrcoef array_x = series_x.to_numpy() array_y = series_y.to_numpy() expected = [np.nan] * (window - 1) for stop in range(window, len(series_x) + 1): chunk_x = array_x[stop - window : stop] chunk_y = array_y[stop - window : stop] expected.append(pairwise(chunk_x - chunk_x[0], chunk_y - chunk_y[0])[0, 1]) tm.assert_series_equal(result, pd.Series(expected), rtol=1e-12, atol=0) @pytest.mark.parametrize("method", ["cov", "corr"]) def test_rolling_cov_corr_outlier_exit_no_nan(method): # GH#65739 cancellation could drive ssqdm_x negative, so calc_corr took the # square root of a negative number and returned NaN for every later window window = 3 series_x = pd.Series([1.0, 2.0, 1e12, 1e7, 6.0, 5.0, 8.0, 7.0]) series_y = pd.Series([2.0, 1.0, 3.0, -1e12, 4.0, 7.0, 6.0, 9.0]) result = getattr(series_x.rolling(window), method)(series_y) assert not result.iloc[window - 1 :].isna().any() # rtol here is the tolerance the incremental update is designed to hold to, # not float64 noise: some of these windows still contain 1e12 pairwise = np.cov if method == "cov" else np.corrcoef array_x = series_x.to_numpy() array_y = series_y.to_numpy() expected = [np.nan] * (window - 1) for stop in range(window, len(series_x) + 1): chunk_x = array_x[stop - window : stop] chunk_y = array_y[stop - window : stop] expected.append(pairwise(chunk_x - chunk_x[0], chunk_y - chunk_y[0])[0, 1]) tm.assert_series_equal(result, pd.Series(expected), rtol=1e-5, atol=0) @pytest.mark.parametrize("method", ["cov", "corr"]) def test_rolling_cov_corr_extreme_range_recovers(method): # GH#65739 a window spanning nearly the whole float64 range must not leave # the accumulators holding NaN, which would blank every later window window = 5 series_x = pd.Series([1e308, 1.0, 2.0, 3.0, -1e308, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]) series_y = pd.Series([0.2, 0.1, 0.5, 0.3, 0.8, 0.4, 0.9, 0.6, 0.7, 0.2, 0.5]) result = getattr(series_x.rolling(window), method)(series_y) assert not result.iloc[window - 1 :].isna().any() # once the extreme values leave, the windows are ordinary data and exact pairwise = np.cov if method == "cov" else np.corrcoef array_x = series_x.to_numpy() array_y = series_y.to_numpy() for stop in range(10, len(series_x) + 1): expected = pairwise( array_x[stop - window : stop], array_y[stop - window : stop] )[0, 1] assert result.iloc[stop - 1] == pytest.approx(expected, rel=1e-12) def test_rolling_corr_bias_correction(): # test for correct bias correction a = pd.Series( np.arange(20, dtype=np.float64), index=pd.date_range("2020-01-01", periods=20) ) b = a.copy() a[:5] = np.nan b[:10] = np.nan result = a.rolling(window=len(a), min_periods=1).corr(b) tm.assert_almost_equal(result.iloc[-1], a.corr(b)) @pytest.mark.parametrize("func", ["cov", "corr"]) def test_rolling_pairwise_cov_corr(func, frame): result = getattr(frame.rolling(window=10, min_periods=5), func)() result = result.loc[(slice(None), 1), 5] result.index = result.index.droplevel(1) expected = getattr(frame[1].rolling(window=10, min_periods=5), func)(frame[5]) tm.assert_series_equal(result, expected, check_names=False) @pytest.mark.parametrize("method", ["corr", "cov"]) def test_flex_binary_frame(method, frame): series = frame[1] res = getattr(series.rolling(window=10), method)(frame) res2 = getattr(frame.rolling(window=10), method)(series) exp = frame.apply(lambda x: getattr(series.rolling(window=10), method)(x)) tm.assert_frame_equal(res, exp) tm.assert_frame_equal(res2, exp) frame2 = pd.DataFrame( np.random.default_rng(2).standard_normal(frame.shape), index=frame.index, columns=frame.columns, ) res3 = getattr(frame.rolling(window=10), method)(frame2) res3.columns = pd.Index(list(res3.columns)) exp = pd.DataFrame( {k: getattr(frame[k].rolling(window=10), method)(frame2[k]) for k in frame} ) tm.assert_frame_equal(res3, exp) @pytest.mark.parametrize("window", range(7)) def test_rolling_corr_with_zero_variance(window): # GH 18430 s = pd.Series(np.zeros(20)) other = pd.Series(np.arange(20)) assert s.rolling(window=window).corr(other=other).isna().all() def test_corr_sanity(): # GH 3155 df = pd.DataFrame( np.array( [ [0.87024726, 0.18505595], [0.64355431, 0.3091617], [0.92372966, 0.50552513], [0.00203756, 0.04520709], [0.84780328, 0.33394331], [0.78369152, 0.63919667], ] ) ) res = df[0].rolling(5, center=True).corr(df[1]) assert all(np.abs(np.nan_to_num(x)) <= 1 for x in res) df = pd.DataFrame(np.random.default_rng(2).random((30, 2))) res = df[0].rolling(5, center=True).corr(df[1]) assert all(np.abs(np.nan_to_num(x)) <= 1 for x in res) def test_rolling_cov_diff_length(): # GH 7512 s1 = pd.Series([1, 2, 3], index=range(3)) s2 = pd.Series([1, 3], index=range(0, 4, 2)) result = s1.rolling(window=3, min_periods=2).cov(s2) expected = pd.Series([None, None, 2.0]) tm.assert_series_equal(result, expected) s2a = pd.Series([1, None, 3], index=range(3)) result = s1.rolling(window=3, min_periods=2).cov(s2a) tm.assert_series_equal(result, expected) def test_rolling_corr_diff_length(): # GH 7512 s1 = pd.Series([1, 2, 3], index=range(3)) s2 = pd.Series([1, 3], index=range(0, 4, 2)) result = s1.rolling(window=3, min_periods=2).corr(s2) expected = pd.Series([None, None, 1.0]) tm.assert_series_equal(result, expected) s2a = pd.Series([1, None, 3], index=range(3)) result = s1.rolling(window=3, min_periods=2).corr(s2a) tm.assert_series_equal(result, expected) @pytest.mark.parametrize("func", ["cov", "corr"]) def test_time_based_rolling_other_longer_raises(func): # GH#62937 idx_short = pd.date_range("2020-01-01", periods=3, freq="D") idx_long = pd.date_range("2020-01-01", periods=5, freq="D") s = pd.Series([1, 2, 3], index=idx_short) other = pd.Series([1, 2, 3, 4, 5], index=idx_long) msg = "Variable rolling window requires .* Got 3 < 5" with pytest.raises(ValueError, match=msg): getattr(s.rolling("2D"), func)(other) @pytest.mark.parametrize( "f", [ lambda x: x.rolling(window=10, min_periods=5).cov(x, pairwise=True), lambda x: x.rolling(window=10, min_periods=5).corr(x, pairwise=True), ], ) def test_rolling_functions_window_non_shrinkage_binary(f): # corr/cov return a MI DataFrame df = pd.DataFrame( [[1, 5], [3, 2], [3, 9], [-1, 0]], columns=pd.Index(["A", "B"], name="foo"), index=pd.Index(range(4), name="bar"), ) df_expected = pd.DataFrame( columns=pd.Index(["A", "B"], name="foo"), index=pd.MultiIndex.from_product([df.index, df.columns], names=["bar", "foo"]), dtype="float64", ) df_result = f(df) tm.assert_frame_equal(df_result, df_expected) @pytest.mark.parametrize( "f", [ lambda x: x.rolling(window=10, min_periods=5).cov(x, pairwise=True), lambda x: x.rolling(window=10, min_periods=5).corr(x, pairwise=True), ], ) def test_moment_functions_zero_length_pairwise(f): df1 = pd.DataFrame() df2 = pd.DataFrame( columns=pd.Index(["a"], name="foo"), index=pd.Index([], name="bar") ) df2["a"] = df2["a"].astype("float64") df1_expected = pd.DataFrame( index=pd.MultiIndex.from_product([df1.index, df1.columns]) ) df2_expected = pd.DataFrame( index=pd.MultiIndex.from_product( [df2.index, df2.columns], names=["bar", "foo"] ), columns=pd.Index(["a"], name="foo"), dtype="float64", ) df1_result = f(df1) tm.assert_frame_equal(df1_result, df1_expected) df2_result = f(df2) tm.assert_frame_equal(df2_result, df2_expected) class TestPairwise: # GH 7738 @pytest.mark.parametrize("f", [lambda x: x.cov(), lambda x: x.corr()]) def test_no_flex(self, pairwise_frames, pairwise_target_frame, f): # DataFrame methods (which do not call flex_binary_moment()) result = f(pairwise_frames) tm.assert_index_equal(result.index, pairwise_frames.columns) tm.assert_index_equal(result.columns, pairwise_frames.columns) expected = f(pairwise_target_frame) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values tm.assert_numpy_array_equal(result, expected, check_dtype=False) @pytest.mark.parametrize( "f", [ lambda x: x.expanding().cov(pairwise=True), lambda x: x.expanding().corr(pairwise=True), lambda x: x.rolling(window=3).cov(pairwise=True), lambda x: x.rolling(window=3).corr(pairwise=True), lambda x: x.ewm(com=3).cov(pairwise=True), lambda x: x.ewm(com=3).corr(pairwise=True), ], ) def test_pairwise_with_self(self, pairwise_frames, pairwise_target_frame, f): # DataFrame with itself, pairwise=True # note that we may construct the 1st level of the MI # in a non-monotonic way, so compare accordingly result = f(pairwise_frames) tm.assert_index_equal( result.index.levels[0], pairwise_frames.index, check_names=False ) tm.assert_index_equal( safe_sort(result.index.levels[1]), safe_sort(pairwise_frames.columns.unique()), ) tm.assert_index_equal(result.columns, pairwise_frames.columns) expected = f(pairwise_target_frame) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values tm.assert_numpy_array_equal(result, expected, check_dtype=False) @pytest.mark.parametrize( "f", [ lambda x: x.expanding().cov(pairwise=False), lambda x: x.expanding().corr(pairwise=False), lambda x: x.rolling(window=3).cov(pairwise=False), lambda x: x.rolling(window=3).corr(pairwise=False), lambda x: x.ewm(com=3).cov(pairwise=False), lambda x: x.ewm(com=3).corr(pairwise=False), ], ) def test_no_pairwise_with_self(self, pairwise_frames, pairwise_target_frame, f): # DataFrame with itself, pairwise=False result = f(pairwise_frames) tm.assert_index_equal(result.index, pairwise_frames.index) tm.assert_index_equal(result.columns, pairwise_frames.columns) expected = f(pairwise_target_frame) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values tm.assert_numpy_array_equal(result, expected, check_dtype=False) @pytest.mark.parametrize( "f", [ lambda x, y: x.expanding().cov(y, pairwise=True), lambda x, y: x.expanding().corr(y, pairwise=True), lambda x, y: x.rolling(window=3).cov(y, pairwise=True), lambda x, y: x.rolling(window=3).corr(y, pairwise=True), lambda x, y: x.ewm(com=3).cov(y, pairwise=True), lambda x, y: x.ewm(com=3).corr(y, pairwise=True), ], ) def test_pairwise_with_other( self, pairwise_frames, pairwise_target_frame, pairwise_other_frame, f ): # DataFrame with another DataFrame, pairwise=True result = f(pairwise_frames, pairwise_other_frame) tm.assert_index_equal( result.index.levels[0], pairwise_frames.index, check_names=False ) tm.assert_index_equal( safe_sort(result.index.levels[1]), safe_sort(pairwise_other_frame.columns.unique()), ) expected = f(pairwise_target_frame, pairwise_other_frame) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values # not exact: rolling corr rounds differently on 32-bit tm.assert_almost_equal(result, expected, check_dtype=False) @pytest.mark.parametrize( "f", [ lambda x, y: x.expanding().cov(y, pairwise=False), lambda x, y: x.expanding().corr(y, pairwise=False), lambda x, y: x.rolling(window=3).cov(y, pairwise=False), lambda x, y: x.rolling(window=3).corr(y, pairwise=False), lambda x, y: x.ewm(com=3).cov(y, pairwise=False), lambda x, y: x.ewm(com=3).corr(y, pairwise=False), ], ) def test_no_pairwise_with_other(self, pairwise_frames, pairwise_other_frame, f): # DataFrame with another DataFrame, pairwise=False result = ( f(pairwise_frames, pairwise_other_frame) if pairwise_frames.columns.is_unique else None ) if result is not None: # we can have int and str columns expected_index = pairwise_frames.index.union(pairwise_other_frame.index) expected_columns = pairwise_frames.columns.union( pairwise_other_frame.columns ) tm.assert_index_equal(result.index, expected_index) tm.assert_index_equal(result.columns, expected_columns) else: with pytest.raises(ValueError, match="'arg1' columns are not unique"): f(pairwise_frames, pairwise_other_frame) with pytest.raises(ValueError, match="'arg2' columns are not unique"): f(pairwise_other_frame, pairwise_frames) @pytest.mark.parametrize( "f", [ lambda x, y: x.expanding().cov(y), lambda x, y: x.expanding().corr(y), lambda x, y: x.rolling(window=3).cov(y), lambda x, y: x.rolling(window=3).corr(y), lambda x, y: x.ewm(com=3).cov(y), lambda x, y: x.ewm(com=3).corr(y), ], ) def test_pairwise_with_series(self, pairwise_frames, pairwise_target_frame, f): # DataFrame with a Series result = f(pairwise_frames, pd.Series([1, 1, 3, 8])) tm.assert_index_equal(result.index, pairwise_frames.index) tm.assert_index_equal(result.columns, pairwise_frames.columns) expected = f(pairwise_target_frame, pd.Series([1, 1, 3, 8])) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values tm.assert_numpy_array_equal(result, expected, check_dtype=False) result = f(pd.Series([1, 1, 3, 8]), pairwise_frames) tm.assert_index_equal(result.index, pairwise_frames.index) tm.assert_index_equal(result.columns, pairwise_frames.columns) expected = f(pd.Series([1, 1, 3, 8]), pairwise_target_frame) # since we have sorted the results # we can only compare non-nans result = result.dropna().values expected = expected.dropna().values tm.assert_numpy_array_equal(result, expected, check_dtype=False) def test_corr_freq_memory_error(self): # GH 31789 s = pd.Series(range(5), index=pd.date_range("2020", periods=5)) result = s.rolling("12h").corr(s) expected = pd.Series([np.nan] * 5, index=pd.date_range("2020", periods=5)) tm.assert_series_equal(result, expected) def test_cov_mulittindex(self): # GH 34440 columns = pd.MultiIndex.from_product([list("ab"), list("xy"), list("AB")]) index = range(3) df = pd.DataFrame(np.arange(24).reshape(3, 8), index=index, columns=columns) result = df.ewm(alpha=0.1).cov() index = pd.MultiIndex.from_product( [range(3), list("ab"), list("xy"), list("AB")] ) columns = pd.MultiIndex.from_product([list("ab"), list("xy"), list("AB")]) expected = pd.DataFrame( np.vstack( ( np.full((8, 8), np.nan), np.full((8, 8), 32.000000), np.full((8, 8), 63.881919), ) ), index=index, columns=columns, ) tm.assert_frame_equal(result, expected) def test_multindex_columns_pairwise_func(self): # GH 21157 columns = pd.MultiIndex.from_arrays([["M", "N"], ["P", "Q"]], names=["a", "b"]) df = pd.DataFrame(np.ones((5, 2)), columns=columns) result = df.rolling(3).corr() expected = pd.DataFrame( np.nan, index=pd.MultiIndex.from_arrays( [ np.repeat(np.arange(5, dtype=np.int64), 2), ["M", "N"] * 5, ["P", "Q"] * 5, ], names=[None, "a", "b"], ), columns=columns, ) tm.assert_frame_equal(result, expected)