如何在Python中为pandas.rolling实现步幅（stride）功能？

问题场景 #

需要实现一种特殊的滚动窗口计算：窗口每次移动1步，但窗口内的元素不是连续的，而是按固定间隔从原数据中选取。例如窗口大小为3、间隔为3时，第6行的窗口包含第6、3、0行的元素，第7行包含第7、4、1行的元素，以此类推。

实现方法 #

方法1：手动移位求和（简单直观） #

对于固定的窗口大小和间隔，直接通过shift()方法将数据移位对应步数后求和，再对前窗口大小×间隔 -1行填充NaN即可：

import pandas as pd

# 构造示例数据
df = pd.DataFrame({'value': range(10)})

# 参数定义
window_size = 3
stride = 3

# 生成移位后的序列并求和
sum_series = sum(df['value'].shift(i * stride) for i in range(window_size))

# 前 window_size*stride -1 行设为缺失值
sum_series.iloc[:window_size * stride - 1] = pd.NA

print(sum_series)

输出：

0     <NA>
1     <NA>
2     <NA>
3     <NA>
4     <NA>
5     <NA>
6       9.0
7      12.0
8      15.0
9      18.0
dtype: float64

方法2：Numpy滑动窗口优化（高效处理大数据） #

当数据量较大或需要灵活调整参数时，用Numpy的索引操作可以提升效率：

import pandas as pd
import numpy as np

df = pd.DataFrame({'value': range(10)})
window_size = 3
stride = 3

values = df['value'].values
min_valid_idx = window_size * stride - 1

# 生成所有有效索引对应的窗口元素并求和
valid_indices = np.arange(min_valid_idx, len(df))
result_values = [values[i - np.arange(window_size)*stride].sum() for i in valid_indices]

# 构建完整结果序列
full_result = pd.Series(pd.NA, index=df.index)
full_result.loc[valid_indices] = result_values

print(full_result)

方法3：封装自定义函数（复用性强） #

如果需要多次使用该逻辑，可以封装成通用函数，支持不同的窗口大小和间隔：

import pandas as pd
import numpy as np

def strided_rolling_sum(series, window_size, stride):
    min_valid_idx = window_size * stride - 1
    result = pd.Series(pd.NA, index=series.index)
    
    if len(series) <= min_valid_idx:
        return result
    
    values = series.values
    for i in range(min_valid_idx, len(series)):
        window_elements = values[i - np.arange(window_size)*stride]
        result.iloc[i] = window_elements.sum()
    
    return result

# 调用示例
df = pd.DataFrame({'value': range(10)})
print(strided_rolling_sum(df['value'], window_size=3, stride=3))

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