Python如何计算移动窗口统计量？rolling函数详解

pandas的rolling()函数用于计算移动窗口统计量，常见聚合操作有1. .mean()计算移动平均值，2. .sum()计算移动总和，3. .std()计算移动标准差，4. .min()/.max()计算极值，5. .count()计算非nan数量，6. .median()计算移动中位数；窗口可定义为固定观测值或时间窗如'3d'、'2h'，且支持自定义函数通过.apply()应用，例如计算窗口范围或非零值计数；处理缺失值时，默认要求窗口内数据点等于window大小才计算，但可通过设置min_periods参数指定最小非nan数量，如设为1则允许窗口内只要有数据即计算，从而减少结果中的nan数量。

在Python中处理时间序列或任何顺序数据时，计算移动窗口统计量是家常便饭，它能帮助我们平滑数据、识别趋势或波动。Pandas库中的rolling()函数就是为此而生的核心工具。它能让你定义一个滑动的数据窗口，然后对这个窗口内的数据执行各种聚合操作，比如计算平均值、总和、标准差等等，非常灵活和高效。

解决方案

使用Pandas的rolling()函数来计算移动窗口统计量。这个函数通常会链式调用其他聚合方法，例如.mean()、.sum()、.std()等。它的核心在于window参数，它定义了窗口的大小。

import pandas as pd
import numpy as np

# 创建一个示例Series
data = pd.Series(np.random.rand(20) * 100)

# 计算5个元素的移动平均值
rolling_mean = data.rolling(window=5).mean()
print("5个元素的移动平均值:\n", rolling_mean)

# 计算3个元素的移动和
rolling_sum = data.rolling(window=3).sum()
print("\n3个元素的移动和:\n", rolling_sum)

# 计算7个元素的移动标准差
rolling_std = data.rolling(window=7).std()
print("\n7个元素的移动标准差:\n", rolling_std)

# 对于DataFrame，可以在特定列上应用
df = pd.DataFrame({'A': np.random.rand(20), 'B': np.random.randint(0, 100, 20)})
df['rolling_A_mean'] = df['A'].rolling(window=4).mean()
print("\nDataFrame上A列的4个元素移动平均值:\n", df)

rolling()函数返回一个Rolling对象，这个对象本身不会立即计算结果，而是等待你指定一个聚合函数（如.mean()、.sum()、.apply()等）后才进行实际计算。这种设计使得它非常高效，因为它允许你链式地应用不同的聚合逻辑。

立即学习“Python免费学习笔记（深入）”；

Pandas rolling()函数的基本用法和常见聚合操作有哪些？

当你拿到一串数据，想看看它在某个时间段或者某个连续序列中的表现时，rolling()函数就显得格外好用。它的基本用法其实挺直观的，就是先用rolling()定义窗口，再跟上你想要的聚合方法。

最基础的，你得告诉rolling()你的窗口有多大，这就是window参数。比如，window=5就意味着每次取最近的5个数据点来计算。这个窗口可以是固定数量的观测值，也可以是时间段（如果你的数据是时间序列的话）。

常见的聚合操作，我平时用得最多的无非就是那么几个：

• .mean(): 计算移动平均值，这是最常用的，能有效平滑数据，看出趋势。
• .sum(): 计算移动总和，比如你想知道过去7天的销售额总和。
• .std(): 计算移动标准差，这能帮你理解数据的波动性，比如股票价格的波动。
• .min() / .max(): 计算移动最小值/最大值，找出窗口内的极值。
• .count(): 计算窗口内的非NaN值的数量，这在处理缺失值时挺有用。
• .median(): 计算移动中位数，对异常值更鲁豫。

举个例子，如果你有一天的每小时温度数据，想看看过去3小时的平均温度，或者过去24小时的最高温度，rolling().mean()和rolling().max()就能轻松搞定。

# 延续上面的data Series
print("原始数据:\n", data.head(10))

# 3个元素的移动最小值
rolling_min = data.rolling(window=3).min()
print("\n3个元素的移动最小值:\n", rolling_min.head(10))

# 4个元素的移动中位数
rolling_median = data.rolling(window=4).median()
print("\n4个元素的移动中位数:\n", rolling_median.head(10))

注意，在窗口的初始阶段，也就是数据点不足以填满整个窗口时，rolling()默认会返回NaN。这很正常，毕竟它还没攒够足够的数据来计算完整的窗口统计量。

稿定AI绘图

稿定推出的AI绘画工具

下载

如何处理移动窗口中的缺失值？min_periods参数有什么用？

处理数据，尤其是真实世界的数据，缺失值总是避不开的话题。在计算移动窗口统计量时，如果窗口内有缺失值，或者窗口一开始数据不够，rolling()的默认行为是返回NaN。这在很多情况下是合理的，因为你可能确实需要一个完整的窗口才能进行有效计算。

但有时候，你可能希望即使窗口内的数据点不足，只要达到某个最小数量，就进行计算。这时，min_periods参数就派上用场了。

min_periods定义了窗口内进行计算所需的最小非NaN观测值数量。

• 默认行为： 如果不设置min_periods，它的默认值就是window的大小。这意味着，只有当窗口内有足够的数据点（等于window大小）时，才会返回一个非NaN的结果。比如，rolling(window=5).mean()，前4个结果都会是NaN。
• 自定义min_periods： 如果你设置min_periods=1，那么只要窗口内至少有一个非NaN值，就会进行计算。这在处理时间序列的开头部分特别有用，可以避免过多的NaN值。

举个例子，我经常在处理一些传感器数据时遇到这种情况，数据流刚开始的时候可能不稳定，或者中间偶尔有几秒钟的数据缺失。如果我用min_periods=1，那么即使窗口不完整，我依然能得到一个近似的统计量，这对于实时监控或者快速趋势判断很有帮助。当然，你需要权衡精度和可用性。

# 示例数据，包含一些NaN
data_with_nan = pd.Series([1, 2, np.nan, 4, 5, 6, np.nan, 8, 9, 10])

# 窗口大小为3，默认min_periods=3
rolling_default_min_periods = data_with_nan.rolling(window=3).mean()
print("默认min_periods (3):\n", rolling_default_min_periods)
# 解释：第一个非NaN值出现在索引2 (4.0)，因为它前面有两个NaN，所以第一个有效计算是 (NaN, NaN, 4) -> NaN。
# 接着 (NaN, 4, 5) -> NaN。
# (4, 5, 6) -> 5.0
# (5, 6, NaN) -> NaN
# (6, NaN, 8) -> NaN
# (NaN, 8, 9) -> NaN
# (8, 9, 10) -> 9.0

# 窗口大小为3，min_periods=1
rolling_custom_min_periods = data_with_nan.rolling(window=3, min_periods=1).mean()
print("\n自定义min_periods (1):\n", rolling_custom_min_periods)
# 解释：
# [1] -> 1.0
# [1, 2] -> 1.5
# [2, NaN] -> 2.0 (因为NaN被忽略了，只计算2)
# [NaN, 4] -> 4.0
# [4, 5] -> 4.5
# [4, 5, 6] -> 5.0
# [5, 6, NaN] -> 5.5
# [6, NaN, 8] -> 7.0
# [NaN, 8, 9] -> 8.5
# [8, 9, 10] -> 9.0

通过对比你会发现，min_periods极大地影响了结果中NaN的出现位置和数量，这取决于你对数据完整性的要求。

除了固定大小窗口，rolling()还能如何定义窗口？时间窗和自定义函数

rolling()的强大之处远不止于固定数量的观测值窗口。对于时间序列数据，它真正闪耀的地方在于支持“时间窗”定义。这意味着你可以指定一个时间长度，比如“过去7天”或“最近3小时”，而不是“过去5个数据点”。这在处理不规则采样或需要基于实际时间进行分析的数据时，简直是天赐之物。

要使用时间窗，你的Series或DataFrame的索引必须是DatetimeIndex。window参数这时接受一个字符串，比如'3D'表示3天，'7D'表示7天，'1H'表示1小时等等。Pandas会自动根据时间戳来确定哪些数据点落在当前窗口内。

# 创建一个时间序列数据
dates = pd.date_range(start='2023-01-01', periods=15, freq='H')
time_series_data = pd.Series(np.random.randint(1, 100, 15), index=dates)
print("时间序列数据:\n", time_series_data)

# 计算2小时的移动平均值
rolling_2h_mean = time_series_data.rolling(window='2H').mean()
print("\n2小时的移动平均值:\n", rolling_2h_mean)

# 计算5小时的移动总和，最小周期为1
rolling_5h_sum = time_series_data.rolling(window='5H', min_periods=1).sum()
print("\n5小时的移动总和 (min_periods=1):\n", rolling_5h_sum)

除了内置的聚合函数，rolling()还支持通过.apply()方法传入自定义函数。这意味着你可以对窗口内的数据执行任何你想要的复杂逻辑。你的自定义函数会接收一个Series或DataFrame的子集作为输入，然后返回一个标量值。这为数据分析提供了极大的灵活性，比如计算窗口内的众数、某个特定百分位数，或者执行更复杂的统计检验。

# 自定义函数：计算窗口内的范围 (最大值 - 最小值)
def calculate_range(x):
    return x.max() - x.min()

# 应用自定义函数
rolling_range = time_series_data.rolling(window='3H', min_periods=2).apply(calculate_range)
print("\n3小时窗口的移动范围 (最大值-最小值):\n", rolling_range)

# 另一个例子：计算窗口内非零值的计数
def count_non_zero(x):
    return (x != 0).sum()

data_with_zeros = pd.Series([1, 0, 3, 0, 5, 6, 0, 8, 9, 0])
rolling_non_zero_count = data_with_zeros.rolling(window=3, min_periods=1).apply(count_non_zero)
print("\n3个元素窗口的非零值计数:\n", rolling_non_zero_count)

通过时间窗和自定义函数，rolling()函数几乎可以满足你在移动窗口统计方面的所有需求，无论是简单的平均值，还是复杂的数据模式识别，它都能提供一个优雅且高效的解决方案。