基于Python deque实现滑动时间窗口的扩展移动求和问题

核心思路 #

因为你的deque是按时间从旧到新有序排列的，队首是最早的数据，所以不需要遍历整个队列检查过期元素：

• 每次处理新元素（或迭代时），以当前元素的时间为基准，计算窗口起始时间（当前时间 - 5分钟）
• 从队首开始，逐个移除所有时间早于窗口起始时间的元素（队列有序，一旦遇到第一个在窗口内的元素，后面的肯定都在窗口里）
• 最后对当前队列内所有元素的event字段求和即可

代码实现 #

假设你的time字段是字符串格式（如"07:20:40"），先转换为datetime对象方便计算。以下是完整可运行代码：

from collections import deque
from datetime import datetime, timedelta

# 初始化deque，模拟已有历史数据
event_deque = deque([
    {"time": "07:14:30", "event": 10},
    {"time": "07:16:20", "event": 20},
    {"time": "07:18:10", "event": 15},
])

# 窗口大小：5分钟
WINDOW_SIZE = timedelta(minutes=5)

def process_new_element(new_element):
    # 将新元素加入队列
    event_deque.append(new_element)
    
    # 解析当前元素的时间（假设是当天时间，有日期的话调整格式即可）
    current_time = datetime.strptime(new_element["time"], "%H:%M:%S")
    # 计算窗口的起始时间
    window_start = current_time - WINDOW_SIZE
    
    # 清理队首的过期元素
    while event_deque:
        oldest_time = datetime.strptime(event_deque[0]["time"], "%H:%M:%S")
        if oldest_time < window_start:
            event_deque.popleft()
        else:
            # 队列有序，后续元素时间都晚于队首，无需继续检查
            break
    
    # 计算当前窗口内的event求和
    window_sum = sum(item["event"] for item in event_deque)
    return window_sum

# 模拟添加新元素并执行处理
new_item = {"time": "07:20:40", "event": 25}
current_sum = process_new_element(new_item)
print(f"当前窗口求和结果：{current_sum}")
print(f"清理后的deque内容：{list(event_deque)}")

关键细节说明 #

• 时间格式适配：如果你的time字段包含日期，只需调整strptime的格式（比如"%Y-%m-%d %H:%M:%S"），确保能正确解析为datetime对象
• 高效清理：deque的popleft()是O(1)操作，结合队列有序的特性，每个元素只会被移除一次，整体清理过程平均时间复杂度为O(1)
• 内存控制：每次添加新元素后立即清理过期数据，避免deque无限膨胀占用内存

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