如何在Pandas滚动窗口中对应匹配第二个DataFrame元素做比较？

问题背景 #

给定两个数据集：

import pandas as pd 
import numpy as np
df = pd.DataFrame([1, 2, 3, 2, 5, 4, 3, 6, 7])

check_df = pd.DataFrame([3, 2, 5, 4, 3, 6, 4, 2, 1])

需要完成以下判断：

• 若df[0:3]中任意数值大于check_df[0]，返回1，否则返回0；
• 若df[1:4]中任意数值大于check_df[1]，返回1，否则返回0；
• 以此类推完成所有对应滚动窗口的判断。

原方案使用rolling.apply结合自定义函数，但自定义函数始终调用check_df[0]，无法实现第i个窗口与check_df[i]的对应匹配。

解决方案 #

方法一：手动遍历滑动窗口（直观易懂） #

先提取所有符合要求的滑动窗口，再逐行与check_df的对应值比较：

import pandas as pd
import numpy as np

df = pd.DataFrame([1, 2, 3, 2, 5, 4, 3, 6, 7])
check_df = pd.DataFrame([3, 2, 5, 4, 3, 6, 4, 2, 1])

# 获取所有长度为3的滑动窗口（过滤掉不足3个元素的窗口）
windows = df.rolling(3, min_periods=3).agg(list).dropna().reset_index(drop=True)
# 匹配对应数量的校验值
check_vals = check_df.iloc[:len(windows)].values.flatten()

# 逐窗口判断，生成结果
result = [1 if any(val > check_val for val in window) else 0 for window, check_val in zip(windows[0], check_vals)]
# 补全原DataFrame长度的结果（末尾补NaN）
result_df = pd.DataFrame(result + [np.nan]*2, columns=['result'])

print(result_df)

输出结果：

result
0     0.0
1     1.0
2     0.0
3     1.0
4     1.0
5     0.0
6     1.0
7     NaN
8     NaN

方法二：矩阵化滑动窗口（高效适合大数据） #

利用numpy的滑动窗口工具生成窗口矩阵，通过广播实现批量比较：

import pandas as pd
import numpy as np

df = pd.DataFrame([1, 2, 3, 2, 5, 4, 3, 6, 7])
check_df = pd.DataFrame([3, 2, 5, 4, 3, 6, 4, 2, 1])

window_size = 3
# 生成滑动窗口矩阵
sliding_windows = np.lib.stride_tricks.sliding_window_view(df.values.flatten(), window_size)
# 获取对应数量的校验值
check_vals = check_df.values.flatten()[:len(sliding_windows)]

# 批量判断并转为1/0格式
result = (sliding_windows > check_vals[:, np.newaxis]).any(axis=1).astype(int)
# 补全结果长度
result_df = pd.DataFrame(
    np.pad(result, (0, len(df)-len(result)), mode='constant', constant_values=np.nan),
    columns=['result']
)

print(result_df)

原方案问题说明 #

rolling.apply的自定义函数仅能接收当前窗口内的数值，无法直接获取窗口对应的索引位置，因此无法动态匹配check_df的对应行值。上述两种方法通过先提取所有窗口，再与check_df的对应值一一绑定，解决了索引匹配的核心问题。

内容的提问来源于stack exchange，提问作者John