Python deque 滑动窗口:旧记录自动淘汰,累计值也要同步减掉
窗口只留三条,总和为什么还一直增加
一个小型仪表脚本只显示最近三次读数,用带 maxlen 的 deque 保存历史后,第四次读数会自动挤掉最旧的一条。列表看起来正确,但旁边的总和却可能仍把所有历史读数加在一起。容器只负责限制自己保存的元素,不会替你更新另一个变量里的统计结果。
本文约定新读数从右侧进入,左侧是最旧记录,窗口最多三条。每次更新先检查窗口是否已满;如果已满,在追加前记住并减去左端旧值,再追加新值并加到总和。顺序很重要:先追加再取左端,拿到的已经是下一条仍应保留的读数。
把窗口和统计值放进同一个更新入口
下面代码可直接用 Python 三点九及以上运行,只使用标准库和内存数据。类只接受正整数容量与整数读数,排除布尔值,便于把本次演示限定为精确整数求和。返回历史时生成元组快照,调用者正常使用时通过 add 更新,避免只修改容器却忘记维护总和。
AI生成概念示意图,非真实界面
from collections import deque
class RecentTotal:
def __init__(self, capacity):
if type(capacity) is not int or capacity <= 0:
raise ValueError('capacity must be a positive integer')
self._items = deque(maxlen=capacity)
self.total = 0
def add(self, value):
if type(value) is not int:
raise TypeError('reading must be an integer')
if len(self._items) == self._items.maxlen:
self.total -= self._items[0]
self._items.append(value)
self.total += value
return self.total
def snapshot(self):
return tuple(self._items)
window = RecentTotal(3)
for value, expected in [(4, 4), (7, 11), (2, 13), (9, 18), (1, 12)]:
actual = window.add(value)
assert actual == expected == sum(window.snapshot())
assert len(window.snapshot()) <= 3
print('window:', window.snapshot(), 'total:', actual)
single = RecentTotal(1)
assert single.add(-5) == -5
assert single.add(2) == 2
assert RecentTotal(3).snapshot() == ()
try:
RecentTotal(0)
except ValueError:
print('zero capacity: rejected')
else:
raise AssertionError('zero capacity was accepted')
for invalid_capacity in (-1, True, 2.5):
try:
RecentTotal(invalid_capacity)
except ValueError:
pass
else:
raise AssertionError('invalid capacity was accepted')
for invalid_reading in (True, 2.5, '3', None):
before = (window.snapshot(), window.total)
try:
window.add(invalid_reading)
except TypeError:
pass
else:
raise AssertionError('invalid reading was accepted')
assert (window.snapshot(), window.total) == before
print('invalid inputs: rejected; window unchanged')
directions = deque([1, 2, 3], maxlen=3)
directions.appendleft(0)
assert list(directions) == [0, 1, 2]
print('appendleft:', list(directions))
directions.extend([8, 9, 10, 11])
assert list(directions) == [9, 10, 11]
print('extend:', list(directions))
directions.extendleft([7, 6])
assert list(directions) == [6, 7, 9]
print('extendleft:', list(directions))
copy = directions.copy()
assert copy.maxlen == 3
assert deque(directions).maxlen is None
copy.append(99)
assert list(copy) == [7, 9, 99]
assert list(directions) == [6, 7, 9]
print('copy capacity:', copy.maxlen, 'contents:', list(copy))五次更新的总和依次为四、十一、十三、十八、十二。第四次窗口从四、七、二变成七、二、九,因此先减四再加九;第五次再减七加一。最后留下二、九、一,总和十二。每一步都与当前窗口重新求和对照,可以把漏减旧值的错误定位到首次满窗后的更新。
容量是规则,当前长度只是状态
窗口尚未填满时,这里也输出部分窗口总和,不等待凑够三条。如果业务只接受完整窗口,应在展示层先检查长度是否等于容量。计算平均值时也要明确分母:预热阶段除以当前条数与固定容量含义不同,空窗口更不能直接做除法。
maxlen 保存的是容量上限,len 返回当前元素数量,两者不能混用。deque 的零容量在标准库里合法,但会立即丢弃加入的元素;本例为避免没有历史却保留总和的混乱,在入口拒绝零容量。负数、非整数和布尔容量也被明确拒绝。
复制历史时要继续保留容量。示例使用 copy 得到仍为三条上限的浅拷贝,再追加一个值确认会发生淘汰。直接写 deque(已有队列) 会得到未设置容量的新队列;若需要显式重建,可同时传入原队列的 maxlen。复制后的数值相同,并不自动证明保留策略相同。
换一个进入方向,淘汰端也随之改变
后半段用独立队列演示方向,避免干扰前面的总和。满窗时 appendleft 把新项放到左边,右边的旧项被移走。所以“左端总是最旧”只在我们坚持从右端追加的约定下成立,不能看见双端队列就把这个业务含义固定在它的某一端。
extend 会按输入顺序逐项从右侧追加。示例一次给四个值,容量只有三,最后只保留九、十、十一,原来的内容以及本批最先加入的八都被淘汰。批量加入不是把列表当作一个元素;如果要把整组数据保存成一项,应使用追加一个对象的接口。
extendleft 则逐项往左边追加,因此输入七、六,最终六在七的左边,满窗时从右端淘汰。它既改变顺序又可能淘汰多项。需要为批量读数维护总和时,最容易核验的做法是逐项调用同一个 add 入口,不要绕过去扩展内部队列。
验算可以全量做,日常更新只处理进出项
示例里的 sum 断言是测试对照,会遍历窗口;日常更新只查看端点、追加和调整整数总和,避免每次重新遍历全部历史。容量为一、包含负数以及空输入也值得测试,负数离开窗口时仍应执行减法,不能简单把总和限制为非负。
这个窗口按条数保留,不代表最近三秒;按时间淘汰还需要时间戳和过期规则。多个线程共同更新时,检查容量、读取旧值、追加与改总和是一组相关动作,不能只因单个队列操作可用就假定整组更新不可分割。本例面向单线程脚本,便于观察每一步状态。


