如何用Pandas高效计算滚动窗口内high与后续low的最大差值

问题描述 #

现有一个包含多列的Pandas DataFrame，需为每一行计算n个观测窗口内，'high'列与后续所有'low'列值的最大差值。以n=3为例，df.max_loss[0]为对应窗口内所有满足ih≤il的high[ih]-low[il]的最大值（示例结果为0.88）。目前已有针对单窗口子集的计算函数，但希望找到更优雅的实现方式（如rolling.apply），避免嵌套循环，求解可行方案。

可行实现方案 #

方案1：rolling.apply结合自定义函数 #

这是最贴合需求的优雅实现，直接利用Pandas滚动窗口API，将单窗口计算逻辑封装为函数传入：

import pandas as pd
import numpy as np

def calc_max_loss(window):
    # 提取窗口内的high和low数组
    high = window['high'].values
    low = window['low'].values
    # 生成所有ih ≤ il的差值矩阵（上三角含对角线）
    diffs = high[:, np.newaxis] - low
    upper_diffs = np.triu(diffs)
    # 取最大值，若所有差值为负则返回0（可按需调整为NaN）
    max_val = upper_diffs.max()
    return max_val if max_val > 0 else 0

# 假设df是目标DataFrame，设置窗口大小n=3
n = 3
df['max_loss'] = df.rolling(window=n, min_periods=1).apply(calc_max_loss, raw=False)

方案2：预计算差值矩阵优化性能 #

如果数据集规模较大，rolling.apply的函数调用开销会比较明显，可通过预计算全局差值矩阵，再滑动提取窗口内的最大值：

# 预计算所有i ≤ j的high[i]-low[j]差值
total_rows = len(df)
diff_matrix = np.zeros((total_rows, total_rows))
for i in range(total_rows):
    diff_matrix[i, i:] = df['high'].iloc[i] - df['low'].iloc[i:]

# 遍历每个位置，提取对应n窗口内的最大值
max_loss_list = []
for i in range(total_rows):
    end_idx = min(i + n, total_rows)
    # 取窗口内的上三角区域（i<=j <= end_idx-1）
    window_diffs = diff_matrix[i:end_idx, i:end_idx]
    max_loss_list.append(window_diffs.max())

df['max_loss'] = pd.Series(max_loss_list)

注意事项 #

• 使用rolling.apply时，raw=False确保传递的是窗口子DataFrame，方便直接按列名取值；若设为raw=True，需按数组索引对应列位置。
• 若业务允许窗口内无有效正差值时返回NaN，可将函数中的返回逻辑改为return max_val if max_val > 0 else np.nan。
• 方案2的内存开销会随数据集增大而增加，若数据量极大，可考虑分块处理或用Numba加速循环。

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