Python DataFrame中基于滚动最大值区间求Low列最低值的问题

核心问题解决代码 #

首先修正原代码中max5的计算逻辑（原写法存在窗口偏移问题），同时获取最大值对应的行索引，最终计算区间Low最小值：

import pandas as pd

# 初始化数据（假设已加载原始数据）
df.columns=['Timestamp', 'Open', 'High', 'Low', 'Close', 'Volume']
df['Timestamp'] = pd.to_datetime(df['Timestamp'], unit='ms')
df.set_index('Timestamp', inplace=True)
maxframes = 5

# 步骤1：获取往后maxframes行的High最大值及其对应索引
# 反转DataFrame实现"往后看"的滚动窗口
rev_df = df.iloc[::-1]
# 计算窗口内High最大值的索引
rev_df['max5_idx'] = rev_df['High'].rolling(window=maxframes, min_periods=1).apply(
    lambda window: window.idxmax(), raw=False
)
# 计算窗口内High的最大值
rev_df['max5'] = rev_df['High'].rolling(window=maxframes, min_periods=1).max()
# 反转回原顺序
df['max5'] = rev_df['max5'].iloc[::-1]
df['max5_idx'] = rev_df['max5_idx'].iloc[::-1]

# 步骤2：计算当前行到max5对应行的Low最小值
# 方法1：小数据集适用（逐行处理）
df['low_min_interval'] = df.apply(
    lambda row: df.loc[row.name:row['max5_idx'], 'Low'].min(),
    axis=1
)

# 方法2：大数据集适用（矢量化优化）
# 先映射索引到行位置，避免重复查找
df['pos'] = range(len(df))
df['max5_pos'] = df['max5_idx'].map(df['pos'])
# 用列表推导实现高效遍历
df['low_min_interval'] = [
    df['Low'].iloc[i:df['max5_pos'].iloc[i]+1].min() 
    for i in range(len(df))
]

关键问题拆解 #

1. 获取max5对应的行索引：
   利用rolling().apply()结合idxmax()，在每个滚动窗口中直接返回最大值对应的Timestamp索引。通过反转DataFrame，实现"往后看"的滚动窗口（pandas默认滚动窗口是"往前看"）。

2. 计算区间Low最小值：

   • 小数据集直接用apply逐行截取区间并取最小值；
   • 大数据集通过索引映射到行位置，用列表推导替代apply，大幅提升效率（列表推导比apply快2-5倍）。

代码优化建议 #

• 修正窗口逻辑：原代码中shift(-maxframes).rolling(maxframes).max()会导致窗口偏移到当前行之后的第maxframes到2*maxframes行，不符合"往后5行"的需求，改用反转+滚动的方式更准确。
• 优先矢量化操作：避免频繁使用apply逐行处理，大数据集下用列表推导、numpy广播或预计算区间统计值替代。
• 边界处理：rolling参数min_periods=1确保末尾不足maxframes行的窗口仍能计算结果，可根据业务需求调整为maxframes强制窗口满额。
• 代码复用：可将往后窗口的统计逻辑封装成函数，比如：

  def rolling_backward(df, col, window, func):
      rev_df = df.iloc[::-1]
      result = rev_df[col].rolling(window=window, min_periods=1).apply(func, raw=False)
      return result.iloc[::-1]
  

  调用时直接df['max5_idx'] = rolling_backward(df, 'High', maxframes, lambda x: x.idxmax())

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