Python中一维时间序列后向移动平均的计算与验证方法

验证后向移动平均实现正确性的方法 #

1. 先对齐后向移动平均的定义 #

因果型10分钟后向移动平均的标准计算逻辑为：采样间隔5秒，120个点对应10分钟，对任意按时间升序排列的第i行数据，ma[i]等于第i-119到i行（共120个连续历史采样点，包含当前点）的MAP_MEDIAN均值，前119行因无法凑够120个历史点，结果应为空值（或按业务规则处理）。

2. 逐点手动计算校验 #

写最基础的逐行遍历逻辑，手动计算部分点的结果和你现有结果对比，逻辑无任何封装，结果可以作为基准：

window_size = 120
# 仅校验前200行即可覆盖前序窗口和部分正常窗口的校验
for i in range(window_size-1, 200):
    # 取当前点往前120个采样点计算均值
    manual_calc_mean = df_1.MAP_MEDIAN.iloc[i-window_size+1 : i+1].mean()
    # 和你最终得到的正序df中对应位置的ma值对比，浮点误差小于1e-6即可认为一致
    is_correct = abs(manual_calc_mean - df_1.ma.iloc[i]) < 1e-6
    print(f"行索引{i}: 手动计算值{manual_calc_mean:.4f}, 你的结果{df_1.ma.iloc[i]:.4f}, 是否一致{is_correct}")

如果所有校验点都输出True，说明逻辑正确，否则存在计算偏差。

3. 用pandas原生参数实现的基准结果对比 #

pandas rolling本身原生支持因果窗口计算，无需倒序、shift操作，结果可作为标准答案：

window_size = 120
# closed='right'表示窗口包含当前右端点，是标准的后向因果窗口配置
df_1['ma_benchmark'] = df_1.MAP_MEDIAN.rolling(window_size, closed='right').mean()
# 计算全量差值的最大值
max_diff = (df_1['ma'] - df_1['ma_benchmark']).abs().max()
print(f"全量数据最大误差：{max_diff}")
# 最大误差小于1e-6即可认为你的实现完全正确

注：你现有代码中的shift(-2)属于多余操作，大概率会导致结果整体偏移，建议去掉该操作后再做对比。

4. 边缘场景和可视化校验 #

• 前序空值校验：正确实现的后向移动平均前119行应为空值，若前几行就出现非空结果，说明窗口范围不符合要求。
• 尾行校验：手动计算df最后120个MAP_MEDIAN的均值，和你结果中最后一行的ma值对比，应完全一致。
• 曲线滞后性校验：放大时序图中任意一个明显的尖峰/谷值，后向移动平均的曲线拐点出现时间应晚于原始MAP曲线，不会出现提前预判拐点的情况（非因果中心窗口才会出现曲线和原始值几乎对齐的情况）。

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