4. 滚动窗口统计
任务说明
本案例针对一段包含时间戳与数值的时序数据,实现滚动窗口统计计算。需要完成:滚动均值、滚动标准差、滚动最大值,并明确窗口边界(数据起始位置不足一个窗口长度时)的处理策略。最终输出每个时间点对应的统计结果,便于后续特征工程直接使用。
实现方案
采用Python标准库与NumPy实现,不依赖pandas,以展示滚动窗口的核心计算逻辑。窗口大小固定为3,边界策略为:窗口不足时输出NaN(不填充),即前2个位置无有效统计值。代码中通过循环遍历每个位置,截取当前点及之前共window_size个元素构成窗口,分别计算均值、标准差(总体标准差,ddof=0)与最大值。
import numpy as np
def rolling_stats(data, window_size=3):
"""
计算滚动均值、标准差、最大值。
窗口不足时输出 NaN。
参数:
data: list或np.ndarray,一维时序数值
window_size: int,窗口大小
返回:
dict: 包含 'mean', 'std', 'max' 三个numpy数组
"""
n = len(data)
means = np.full(n, np.nan)
stds = np.full(n, np.nan)
maxs = np.full(n, np.nan)
for i in range(n):
if i + 1 < window_size:
continue # 窗口不足,保持NaN
window = data[i - window_size + 1 : i + 1]
means[i] = np.mean(window)
stds[i] = np.std(window) # ddof=0,总体标准差
maxs[i] = np.max(window)
return {'mean': means, 'std': stds, 'max': maxs}
# 样例数据(10个点)
data = [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]
result = rolling_stats(data, window_size=3)
# 打印结果
print("数据:", data)
print("滚动均值:", result['mean'])
print("滚动标准差:", result['std'])
print("滚动最大值:", result['max'])
预期输出
根据代码逻辑,窗口大小为3,前2个位置为NaN,从索引2开始有有效值。以索引2为例,窗口为[1,2,3],均值为2.0,标准差为0.81649658,最大值为3.0。完整预期输出如下:
数据: [1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0]
滚动均值: [nan nan 2.0 3.0 4.0 5.0 6.0 7.0 8.0 9.0]
滚动标准差: [nan nan 0.81649658 0.81649658 0.81649658 0.81649658 0.81649658 0.81649658 0.81649658 0.81649658]
滚动最大值: [nan nan 3.0 4.0 5.0 6.0 7.0 8.0 9.0 10.0]
落地与验证
运行方式:将上述代码保存为rolling_stats.py,在已安装NumPy的Python环境中执行python rolling_stats.py。
验证方法:观察输出数组长度与输入数据一致(均为10);前2个位置为NaN;从第3个位置起,均值等于窗口内三个数的算术平均,最大值等于窗口内最大元素,标准差可通过手工计算核对(例如索引2处窗口[1,2,3]的总体标准差为√(2/3)≈0.8165)。
判定标准:若输出与上述预期一致,且窗口边界处正确输出NaN,则案例实现通过。