如何在Python中构建Tumbling Window并计算分段最大值？

当然可以轻松实现你需要的这种**非重叠滚动窗口（Tumbling Window）**计算！针对你的示例数据，我整理了几种实用的方案，从纯原生Python到适合大数据的库方法都有：

方法1：纯Python原生实现（无依赖） #

这是最直观的方法，不需要安装任何额外库，适合小数据量或者快速验证需求：

listA = [3,2,5,9,4,6,3,8,7,9]
window_size = 3
result = []

# 按步长window_size遍历每个窗口的起始索引
for i in range(0, len(listA), window_size):
    # 截取当前窗口的元素
    current_window = listA[i:i+window_size]
    # 计算窗口最大值并加入结果列表
    result.append(max(current_window))

print(result)  # 输出: [5, 9, 8, 9]

如果你的需求是只保留完整的3元素窗口（忽略最后不足3个的元素），只需要调整循环的终止条件：

result = []
for i in range(0, len(listA) - window_size + 1, window_size):
    current_window = listA[i:i+window_size]
    result.append(max(current_window))

print(result)  # 输出: [5, 9, 8]

方法2：使用NumPy库（适合大型数组） #

如果你处理的是ndarray或者大型数据集，NumPy的方法效率更高，代码也简洁：

import numpy as np

arr = np.array([3,2,5,9,4,6,3,8,7,9])
window_size = 3

# 将数组分割为非重叠窗口（最后一个窗口可能长度不足）
windows = np.array_split(arr, np.ceil(len(arr)/window_size).astype(int))
# 计算每个窗口的最大值
result = [window.max() for window in windows]

print(result)  # 输出: [5, 9, 8, 9]

方法3：使用itertools（适合迭代器/大数据流） #

如果你的数据是迭代器类型（比如从文件读取的数据流），不想一次性加载全部数据到内存，可以用itertools.islice来分块处理：

from itertools import islice

listA = [3,2,5,9,4,6,3,8,7,9]
window_size = 3
result = []
data_iter = iter(listA)

while True:
    # 每次从迭代器中取window_size个元素
    current_window = list(islice(data_iter, window_size))
    # 没有元素时退出循环
    if not current_window:
        break
    result.append(max(current_window))

print(result)  # 输出: [5, 9, 8, 9]

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