替代pandas iterrows的高效方案：批量计算价格异动触发时间

需求概述 #

现有含Time（时间）和Price（价格）的Pandas数据表，需新增两列：

• Up_Time：记录当前行之后首次出现价格上涨超100美元的时间
• Down_Time：记录当前行之后首次出现价格下跌超100美元的时间

示例：

• 09:19行的Up_Time为14:02，Down_Time为11:39
• 09:56行的Up_Time为14:02，Down_Time为12:18

原始数据表：

Time        Price    Up_Time   Down_Time
09:19:00    3252.25     
09:24:00    3259.9      
09:56:00    3199.4      
10:17:00    3222.5      
10:43:00    3191.25     
11:39:00    3143        
12:18:00    2991.7      
13:20:00    3196.35     
13:26:00    3176.1      
13:34:00    3198.85     
13:37:00    3260.75     
14:00:00    3160.85     
14:02:00    3450        
14:19:00    3060.5      
14:30:00    2968.7      
14:31:00    2895.8      
14:52:00    2880.7      
14:53:00    2901.55     
14:55:00    2885.55     
14:57:00    2839.05     
14:58:00    2871.5      
15:00:00    2718.95     

原代码问题 #

当前使用的双重循环代码时间复杂度为O(n²)，当数据量较大时（如几万行），会导致处理时间长达15-20分钟：

for i, row in df.iterrows():
    time_up = np.nan
    time_down = np.nan

    for j in range(i+1, len(df)):
        diff = df.iloc[j]['Price'] - row['Price']
        if diff > 100:
            time_up = df.iloc[j]['Time']
        elif diff < -100:
            time_down = df.iloc[j]['Time']

        if not pd.isna(time_up) or not pd.isna(time_down):
            break

    df.at[i, 'Up_Time'] = time_up
    df.at[i, 'Down_Time'] = time_down

高效优化方案：单调栈实现 #

使用单调栈可以将时间复杂度降至O(n)，每个元素仅入栈和出栈一次，处理速度大幅提升。

代码实现 #

import pandas as pd
import numpy as np

# 重置索引确保连续（若原索引已连续可跳过）
df = df.reset_index(drop=True)

# 计算Up_Time：找当前行之后第一个价格超当前价+100的时间
up_stack = []
up_times = [np.nan] * len(df)

for i in range(len(df)-1, -1, -1):
    target_price = df.loc[i, 'Price'] + 100
    # 弹出栈中所有不满足条件的元素
    while up_stack and df.loc[up_stack[-1], 'Price'] <= target_price:
        up_stack.pop()
    # 栈不为空则取栈顶对应时间
    if up_stack:
        up_times[i] = df.loc[up_stack[-1], 'Time']
    # 将当前索引压入栈
    up_stack.append(i)

df['Up_Time'] = up_times

# 计算Down_Time：找当前行之后第一个价格低于当前价-100的时间
down_stack = []
down_times = [np.nan] * len(df)

for i in range(len(df)-1, -1, -1):
    target_price = df.loc[i, 'Price'] - 100
    # 弹出栈中所有不满足条件的元素
    while down_stack and df.loc[down_stack[-1], 'Price'] >= target_price:
        down_stack.pop()
    # 栈不为空则取栈顶对应时间
    if down_stack:
        down_times[i] = df.loc[down_stack[-1], 'Time']
    # 将当前索引压入栈
    down_stack.append(i)

df['Down_Time'] = down_times

原理说明 #

• 从后往前遍历数据表，用栈维护后续行的索引，确保栈中元素对应的价格始终满足目标条件（涨超100/跌超100）
• 每次遍历当前行时，先弹出栈中不满足条件的元素，剩余栈顶即为当前行之后第一个符合要求的行索引
• 最终将栈顶对应的时间赋值给当前行的目标列

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