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,则案例实现通过。

滚动窗口统计计算流程 (窗口大小=3) 原始数据: 1.0 2.0 3.0 4.0 5.0 6.0 7.0 8.0 9.0 10.0 窗口示例 (索引2): 窗口 [1.0, 2.0, 3.0] 输出统计量: 均值 = 2.0 标准差 = 0.8165 最大值 = 3.0 索引0,1: 窗口不足 → 输出 NaN