使用Pandas实现基于产品价格变动的后续时间匹配需求

DevOps 小满

2026-4-17

Pandas实现基于产品价格变动的后续时间匹配需求

我来帮你搞定这个重复的数据分析任务！针对你需要为每一行记录匹配后续首次价格变动超$100的上行时间和后续首次价格变动超$100的下行时间的需求，我整理了一套可复用的Pandas方案，完美适配你批量处理多数据集的场景。

先明确需求核心：对每一行的当前价格，我们要往后找第一个比它高$100以上的价格对应的时间（上行），以及第一个比它低$100以上的价格对应的时间（下行）。

步骤详解与代码实现

1. 数据预处理：确保时间列格式正确

首先得把Time列转换成Pandas的datetime类型，这样才能保证数据按时间顺序处理（如果你的数据已经是datetime格式，这一步可以跳过）：

import pandas as pd

# 假设你的数据来自CSV文件，或者已有现成的DataFrame
df = pd.read_csv("your_data.csv")
# 转换时间列格式
df['Time'] = pd.to_datetime(df['Time'], format='%H:%M:%S')
# 确保数据按时间排序（原始数据已有序的话可省略）
df = df.sort_values('Time').reset_index(drop=True)

2. 定义匹配后续时间的函数

我们写一个小函数，接收当前行的索引和价格，在后续行中筛选符合条件的第一个时间：

def find_next_time(row, df, direction):
    current_price = row['Price_of_product']
    # 只关注当前行之后的数据
    future_data = df.iloc[row.name+1:]
    
    if direction == 'up':
        # 筛选价格比当前高100以上的行
        mask = future_data['Price_of_product'] >= current_price + 100
    elif direction == 'down':
        # 筛选价格比当前低100以上的行
        mask = future_data['Price_of_product'] <= current_price - 100
    else:
        return pd.NaT
    
    # 返回第一个符合条件的时间，无匹配则返回空值
    matching_rows = future_data[mask]
    return matching_rows['Time'].iloc[0] if not matching_rows.empty else pd.NaT

3. 生成目标列

用apply方法给每一行计算两个目标列：

# 生成"后续首次涨价超100的时间"列
df['Time when next price moving by 100 up'] = df.apply(find_next_time, args=(df, 'up'), axis=1)
# 生成"后续首次降价超100的时间"列
df['Time when next price moving by 100 Down'] = df.apply(find_next_time, args=(df, 'down'), axis=1)

4. 验证结果（完全匹配你的示例）

咱们检查你提到的两个示例行：

对于09:19:00的行，当前价格3252.25：
- 第一个涨价超100的是14:02:00的3450（差值197.75>100）
- 第一个降价超100的是11:39:00的3143（差值109.25>100）
对于09:56:00的行，当前价格3199.4：
- 第一个涨价超100的同样是14:02:00的3450（差值250.6>100）
- 第一个降价超100的是12:18:00的2991.7（差值207.7>100）
  运行代码后，这两行的结果会完全符合你的预期！

5. 批量处理多数据集的优化

如果要处理大量数据集，把逻辑封装成函数就能批量复用：

def process_price_data(file_path):
    df = pd.read_csv(file_path)
    df['Time'] = pd.to_datetime(df['Time'], format='%H:%M:%S')
    df = df.sort_values('Time').reset_index(drop=True)
    
    def find_next_time(row, df, direction):
        current_price = row['Price_of_product']
        future_data = df.iloc[row.name+1:]
        
        if direction == 'up':
            mask = future_data['Price_of_product'] >= current_price + 100
        elif direction == 'down':
            mask = future_data['Price_of_product'] <= current_price - 100
        else:
            return pd.NaT
        
        matching_rows = future_data[mask]
        return matching_rows['Time'].iloc[0] if not matching_rows.empty else pd.NaT
    
    df['Time when next price moving by 100 up'] = df.apply(find_next_time, args=(df, 'up'), axis=1)
    df['Time when next price moving by 100 Down'] = df.apply(find_next_time, args=(df, 'down'), axis=1)
    
    # 保存处理后的文件
    df.to_csv(f"processed_{file_path}", index=False)
    return df

# 批量处理示例：遍历多个数据文件
file_list = ["data1.csv", "data2.csv", "data3.csv"]
for file in file_list:
    process_price_data(file)

额外提示

如果你的数据集特别大，apply方法效率可能不够高，这时候可以考虑用向量化方法（比如结合shift或布尔索引）优化，但对于常规规模的数据集，上面的方法已经足够清晰高效了。

备注：内容来源于stack exchange，提问作者rakesh