基于DataFrame列值的Pandas滚动窗口计算问题

pandas的rolling()方法不支持直接传入Series作为动态窗口大小，因为它是为统一窗口的向量化计算设计的。以下是两种可行的实现方式：

方案一：逐行自定义计算 #

通过apply()遍历每行，根据当前行的period值提取对应窗口的历史数据并计算最小值：

import pandas as pd

# 构造示例数据
data = {
    'price': [124.22, 126.51, 121.3, 125.25],
    'period': [5, 6, 6, 4]
}
df = pd.DataFrame(data)

def dynamic_rolling_min(row):
    row_idx = row.name
    window_size = row['period']
    # 确定窗口起始索引，避免越界
    start_idx = max(0, row_idx - window_size)
    # 提取当前行之前的窗口数据（对应shift(1)的逻辑，不含当前行）
    window_data = df.loc[start_idx:row_idx-1, 'price']
    return window_data.min() if not window_data.empty else pd.NA

# 生成结果列
df['lowest'] = df.apply(dynamic_rolling_min, axis=1)

运行后结果：

方案二：基于expanding窗口优化 #

利用expanding()窗口获取到当前行的所有历史数据，再根据period截取对应长度的子窗口计算最小值：

def get_dynamic_min(window):
    # 获取当前行的窗口大小
    current_period = window['period'].iloc[-1]
    # 截取历史数据中最后current_period个price（排除当前行）
    price_window = window['price'].iloc[:-1][-current_period:]
    return price_window.min() if not price_window.empty else pd.NA

df['lowest'] = df.expanding().apply(get_dynamic_min, raw=False)

注意事项 #

• 若窗口大小大于当前行之前的有效数据量，上述代码会取所有可用历史数据计算最小值；若需这种场景返回NaN，可修改判断条件为return pd.NA if row_idx < window_size else window_data.min()。
• 逐行处理的效率略低于固定窗口滚动计算，若数据量极大，可考虑用numpy向量化逻辑进一步优化。

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