Pandas中基于30秒时间窗口的列聚合及Flag更新优化问题

前置准备 #

先确保时间列格式正确并完成排序，这是后续窗口计算的基础：

import pandas as pd

# 转换时间列为datetime类型
df['local_time'] = pd.to_datetime(df['local_time'])
# 按ID和时间排序，保证窗口计算的连续性
df = df.sort_values(['ID', 'local_time']).reset_index(drop=True)

方案一：利用区间索引标记覆盖范围 #

通过反转时间实现"未来30秒"窗口求和，再用区间索引快速判断每行是否需要标记：

def process_group(g):
    # 反转时间，用rolling向前窗口模拟"未来30秒"的求和逻辑
    g_rev = g.iloc[::-1].copy()
    g_rev['window_sum'] = g_rev.rolling('30s', on='local_time', closed='both')['data'].sum()
    g['window_sum'] = g_rev['window_sum'].iloc[::-1]
    
    # 筛选出总和≥4的行，提取对应的时间区间
    valid_intervals = g[g['window_sum'] >= 4].assign(
        end_time=lambda x: x['local_time'] + pd.Timedelta(seconds=30)
    )[['local_time', 'end_time']]
    
    # 无符合条件的区间时直接返回0标记
    if valid_intervals.empty:
        g['Flag'] = 0
        return g.drop('window_sum', axis=1)
    
    # 创建区间索引，判断每行时间是否落在任一有效区间内
    interval_index = pd.IntervalIndex.from_arrays(
        valid_intervals['local_time'], valid_intervals['end_time'], closed='both'
    )
    g['Flag'] = g['local_time'].apply(lambda t: 1 if any(t in iv for iv in interval_index) else 0)
    
    return g.drop('window_sum', axis=1)

# 分组处理得到最终结果
df_result = df.groupby('ID', group_keys=False).apply(process_group)

方案二：事件累加标记（适合大数据量） #

通过事件计数的方式快速定位需要标记的时间段，避免逐个区间判断，性能更优：

def process_group(g):
    # 计算每行未来30秒窗口的data总和
    g_rev = g.iloc[::-1].copy()
    g_rev['window_sum'] = g_rev.rolling('30s', on='local_time', closed='both')['data'].sum()
    g['window_sum'] = g_rev['window_sum'].iloc[::-1]
    
    valid_rows = g[g['window_sum'] >= 4]
    if valid_rows.empty:
        g['Flag'] = 0
        return g.drop('window_sum', axis=1)
    
    # 创建事件：窗口开始标记+1，窗口结束标记-1
    events = []
    for t in valid_rows['local_time']:
        events.append((t, 1))
        events.append((t + pd.Timedelta(seconds=30), -1))
    
    # 合并所有时间点并排序，计算累加活跃值（活跃值>0表示处于需要标记的区间）
    all_times = sorted(set(g['local_time'].tolist() + [t for t, _ in events]))
    event_df = pd.DataFrame(events, columns=['time', 'delta']).sort_values('time')
    full_event = pd.DataFrame({'time': all_times}).merge(event_df, on='time', how='left').fillna(0)
    full_event['active'] = full_event['delta'].cumsum()
    
    # 匹配每行时间对应的活跃状态，完成标记
    merged = pd.merge_asof(g, full_event, left_on='local_time', right_on='time', direction='backward')
    g['Flag'] = merged['active'].apply(lambda x: 1 if x > 0 else 0)
    
    return g.drop('window_sum', axis=1)

df_result = df.groupby('ID', group_keys=False).apply(process_group)

效率说明 #

两种方案均避免了Python层面的循环，全部使用Pandas内置的矢量化操作（rolling、merge_asof、IntervalIndex），底层基于C实现，速度远快于原生循环。其中方案二的事件累加方式时间复杂度为O(n log n)，更适合百万级以上的大数据量场景。

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