Pandas：计算每行自定义Start和End时间区间内的Value最大值

现有包含Timestamp、Value、Start、End四列的DataFrame，需要为每一行计算所有满足Timestamp >= 当前行Start且Timestamp <= 当前行End的行对应的Value最大值，因每行区间长度不固定，无法直接使用rolling.max()实现。
示例数据参考：

方法1：逐行apply（适合1万行以内小数据量） #

逻辑简单易理解，时间复杂度O(n²)，数据量过大时运行较慢

import pandas as pd

def calc_interval_max(row, full_df):
    match_mask = (full_df['Timestamp'] >= row['Start']) & (full_df['Timestamp'] <= row['End'])
    return full_df.loc[match_mask, 'Value'].max()

df['interval_max_value'] = df.apply(calc_interval_max, axis=1, full_df=df)

方法2：numpy广播向量化（适合10万行以内中等数据量） #

通过矩阵运算替代逐行遍历，性能比逐行apply提升5~10倍

import numpy as np

timestamps_arr = df['Timestamp'].values
starts_arr = df['Start'].values[:, None]
ends_arr = df['End'].values[:, None]
value_arr = df['Value'].values

# 生成匹配掩码矩阵，不匹配的位置填充负无穷不影响最大值计算
match_mask = (timestamps_arr >= starts_arr) & (timestamps_arr <= ends_arr)
df['interval_max_value'] = np.max(np.where(match_mask, value_arr, -np.inf), axis=1)

方法3：排序+双指针优化（适合10万行以上大数据量） #

时间复杂度O(nlogn)，是性能最优的实现方案，适合超大数据集

import numpy as np

# 先按Timestamp排序原数据集
df_sorted = df.sort_values('Timestamp', ignore_index=True)
sorted_ts = df_sorted['Timestamp'].values
sorted_val = df_sorted['Value'].values

# 打包所有区间的起始、结束、原索引
intervals = [(s, e, idx) for idx, (s, e) in enumerate(zip(df['Start'], df['End']))]
# 按区间起始值排序
intervals.sort(key=lambda x: x[0])

current_max = -np.inf
right_pointer = 0
result = [0] * len(df)

for s, e, original_idx in intervals:
    # 右指针右移，把所有小于等于当前区间结束时间的Value纳入统计
    while right_pointer < len(sorted_ts) and sorted_ts[right_pointer] <= e:
        if sorted_val[right_pointer] > current_max:
            current_max = sorted_val[right_pointer]
        right_pointer += 1
    # 区间已按起始值排序，所有已统计的时间戳都满足>=区间起始值，直接取最大值即可
    result[original_idx] = current_max

df['interval_max_value'] = result

注意事项 #

• 运行前请确保Timestamp、Start、End三列均为pandas datetime类型，可通过pd.to_datetime()转换
• 如果Value列存在空值，可在计算最大值前先过滤空值，或添加skipna=True参数

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