Pandas Rolling时间窗口报错及10秒移动平均实现求助

原代码的核心问题 #

1. 使用rolling('10sec')时，未将ltt列设为DataFrame的datetime类型索引，pandas无法识别时间窗口的基准列，导致报错。
2. 按hour+minute分组不符合“每10秒区间”的需求，这种分组会把同一分钟内的所有数据放在一起，无法实现10秒级的区间划分。
3. 流式数据场景下，每次重新创建DataFrame、重复转换datetime类型会导致效率极低，且未做数据窗口的裁剪（旧数据无需一直保留）。

解决步骤与修正代码 #

1. 解决rolling('10sec')报错的核心要求 #

要使用时间偏移量作为滚动窗口参数，必须满足：

• 数据的索引是datetime64类型
• 调用rolling时基于索引计算

2. 实现两种10秒均值计算方式 #

根据需求，分两种场景处理：

场景A：固定10秒区间的均值（如12:12:00-12:12:09、12:12:10-12:12:19等区间的均值）

这种是按固定时间桶分组计算，代码如下：

import pandas as pd

records1 = [
{'ltt': 'Mon Nov  7 12:12:05 2022', 'last': 258},
{'ltt': 'Mon Nov  7 12:12:05 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:07 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:08 2022', 'last': 260},
{'ltt': 'Mon Nov  7 12:12:09 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:10 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:11 2022', 'last': 261},
{'ltt': 'Mon Nov  7 12:12:12 2022', 'last': 262},
{'ltt': 'Mon Nov  7 12:12:12 2022', 'last': 260},
{'ltt': 'Mon Nov  7 12:12:14 2022', 'last': 258},
{'ltt': 'Mon Nov  7 12:12:15 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:16 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:17 2022', 'last': 260},
{'ltt': 'Mon Nov  7 12:12:18 2022', 'last': 258},
{'ltt': 'Mon Nov  7 12:12:19 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:20 2022', 'last': 260},
{'ltt': 'Mon Nov  7 12:12:21 2022', 'last': 260},
{'ltt': 'Mon Nov  7 12:12:22 2022', 'last': 258},
{'ltt': 'Mon Nov  7 12:12:23 2022', 'last': 259},
{'ltt': 'Mon Nov  7 12:12:24 2022', 'last': 260}
]

# 初始化DataFrame并处理时间列
df = pd.DataFrame(records1)
df['ltt'] = pd.to_datetime(df['ltt'], format="%a %b %d %H:%M:%S %Y")

# 按10秒区间分桶，生成区间标签
df['10sec_bin'] = df['ltt'].dt.floor('10S')

# 计算每个10秒区间的均值，并合并回原数据
df['10sec_avg'] = df.groupby('10sec_bin')['last'].transform('mean')

print(df[['ltt', 'last', '10sec_bin', '10sec_avg']])

场景B：滚动10秒窗口的均值（每个数据点往前10秒内所有数据的均值）

这种是滑动时间窗口，代码如下：

import pandas as pd

records1 = [
# 数据同上，省略
]

df = pd.DataFrame(records1)
df['ltt'] = pd.to_datetime(df['ltt'], format="%a %b %d %H:%M:%S %Y")

# 将ltt设为索引，排序后计算滚动均值
df = df.set_index('ltt').sort_index()
df['rolling_10sec_avg'] = df['last'].rolling('10s', min_periods=1).mean()

# 重置索引恢复原结构
df = df.reset_index()

print(df[['ltt', 'last', 'rolling_10sec_avg']])

3. 流式数据场景优化 #

对于流式数据，不需要每次处理全部历史数据，只需维护最近10秒的数据集：

import pandas as pd
from datetime import timedelta

# 全局维护一个带datetime索引的DataFrame
stream_df = pd.DataFrame(columns=['ltt', 'last']).set_index('ltt')

def process_stream_record(record):
    global stream_df
    # 转换时间并添加到数据集
    record['ltt'] = pd.to_datetime(record['ltt'], format="%a %b %d %H:%M:%S %Y")
    new_row = pd.DataFrame([record]).set_index('ltt')
    stream_df = pd.concat([stream_df, new_row])
    
    # 裁剪窗口：只保留最近10秒的数据
    cutoff_time = stream_df.index.max() - timedelta(seconds=10)
    stream_df = stream_df[stream_df.index >= cutoff_time]
    
    # 计算当前滚动10秒均值（或固定区间均值）
    stream_df['rolling_10sec_avg'] = stream_df['last'].rolling('10s', min_periods=1).mean()
    
    # 打印最新结果
    print(stream_df.tail(1))

# 模拟流式数据输入
for rec in records1:
    process_stream_record(rec)

关键说明 #

• 使用rolling('10s')时，必须确保索引是datetime类型，这是解决原报错的核心。
• 固定区间均值用dt.floor('10S')生成时间桶，适合统计整10秒段的聚合值；滚动窗口均值适合实时计算每个数据点的历史10秒平均。
• 流式场景下，定期裁剪旧数据可以避免内存占用过高，同时保证计算效率。

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