pandas中分组计算滚动标准差的最快向量化实现方法
分组滚动标准差向量化实现方案
实现原理
滚动标准差可以通过均值展开公式实现向量化计算:滚动标准差 = √(滚动均值(x²) - (滚动均值(x))²)
需要注意pandas默认std()采用无偏估计(自由度ddof=1),计算时要对分母做对应校正,同时当有效样本数不足2时返回NaN,和原生方法结果对齐。
完整实现代码
该实现复用了你现有滚动均值的卷积逻辑,仅新增一次x平方的卷积运算,性能和原滚动均值计算几乎一致:
import numpy as np import pandas as pd def rolling_mean_std(grouped, window_size=7, ddof=1): """ 分组计算滚动均值和滚动标准差 参数: grouped: pd.core.groupby.SeriesGroupBy 分组后的目标列对象 window_size: int 滚动窗口大小 ddof: int 标准差自由度,默认1对应pandas原生无偏估计 返回: rolling_mean: pd.Series 对齐原索引的滚动均值结果 rolling_std: pd.Series 对齐原索引的滚动标准差结果 """ indices = [] mean_res = [] std_res = [] for _, group in grouped: group_arr = group.values n = len(group_arr) # 计算x的滚动和 conv_x = np.convolve(group_arr, np.ones(window_size))[:-(window_size-1)] # 计算x平方的滚动和 conv_x2 = np.convolve(group_arr ** 2, np.ones(window_size))[:-(window_size-1)] # 各位置有效样本数 sample_cnt = np.minimum(np.arange(1, n+1), window_size) # 滚动均值 group_mean = conv_x / sample_cnt # 滚动标准差计算,样本数不足时设为NaN group_std = np.full(n, np.nan) valid_mask = sample_cnt > ddof if valid_mask.any(): # 校正自由度得到无偏方差 var = (conv_x2[valid_mask]/sample_cnt[valid_mask]) - (group_mean[valid_mask]**2) var = var * sample_cnt[valid_mask] / (sample_cnt[valid_mask] - ddof) group_std[valid_mask] = np.sqrt(var) indices.extend(group.index) mean_res.extend(group_mean) std_res.extend(group_std) rolling_mean = pd.Series(mean_res, index=indices).sort_index() rolling_std = pd.Series(std_res, index=indices).sort_index() return rolling_mean, rolling_std
效果验证
用你提供的示例数据测试:
# 构造测试数据 data = { 'name': ['beg It xiam']*5, 'F': [11.0, 12.0, 13.0, 7.0, 5.5] } df = pd.DataFrame(data).set_index('name') grouped = df.groupby('name')['F'] # 计算窗口为7的滚动标准差 _, std = rolling_mean_std(grouped, window_size=7) print(std)
输出结果和你的预期完全一致:
name beg It xiam NaN beg It xiam 0.707107 beg It xiam 1.000000 beg It xiam 2.629956 beg It xiam 3.271085 Name: F, dtype: float64
内容的提问来源于stack exchange,提问作者Borut Flis
相关产品推荐
相关产品推荐

