在实时监控、行情推送和日志分析等场景中，数据以流的形式持续到达，我们还需要在任意时刻快速知道最近一段时间内的最大值或最小值。如果每次都遍历全部数据，时间开销会随数据量线性增长，难以满足实时性要求。下面介绍几种在Python中常用的动态最值查找策略。
基于堆的滑动窗口最值
当只需要维护固定大小滑动窗口的最值时，可以使用两个堆并延迟删除过期元素。以下示例用最大堆和最小堆分别维护窗口最大值与最小值：
import heapq
class SlidingWindowExtreme:
def __init__(self, k):
self.k = k
self.buffer = []
self.max_heap = [] # 存负值模拟最大堆
self.min_heap = []
self.counter = 0
self.lazy_max = {}
self.lazy_min = {}
def add(self, val):
self.buffer.append((self.counter, val))
heapq.heappush(self.max_heap, (-val, self.counter))
heapq.heappush(self.min_heap, (val, self.counter))
self.counter += 1
if len(self.buffer) > self.k:
old_idx, old_val = self.buffer.pop(0)
self.lazy_max[(-old_val, old_idx)] = True
self.lazy_min[(old_val, old_idx)] = True
self._clean()
def _clean(self):
while self.max_heap and self.lazy_max.get(self.max_heap[0], False):
heapq.heappop(self.max_heap)
while self.min_heap and self.lazy_min.get(self.min_heap[0], False):
heapq.heappop(self.min_heap)
def get_max(self):
self._clean()
return -self.max_heap[0][0]
def get_min(self):
self._clean()
return self.min_heap[0][0]
w = SlidingWindowExtreme(3)
for v in [5, 1, 9, 3]:
w.add(v)
print(w.get_max(), w.get_min())
单调队列法
若仅求滑动窗口最大值或最小值，单调队列可在均摊O(1)时间内完成。核心是保证队列内元素下标对应的值单调，并在队首移除窗口外元素：
from collections import deque
def sliding_max(nums, k):
q = deque()
res = []
for i, x in enumerate(nums):
while q and nums[q[-1]] <= x:
q.pop()
q.append(i)
if q[0] == i - k:
q.popleft()
if i >= k - 1:
res.append(nums[q[0]])
return res
print(sliding_max([5, 1, 9, 3, 7], 3))
使用平衡树结构
当窗口大小不固定或需频繁按值删除时，可借助sortedcontainers库中的SortedList，它支持O(log n)插入与删除，并可直接取首尾元素：
from sortedcontainers import SortedList
sl = SortedList()
for v in [4, 2, 8, 1]:
sl.add(v)
print('min', sl[0], 'max', sl[-1])
sl.remove(2)
print('min', sl[0], 'max', sl[-1])
策略对比
| 方法 | 适用场景 | 时间复杂度 |
|---|---|---|
| 双堆延迟删除 | 同时需最大最小，窗口固定 | 均摊O(log n) |
| 单调队列 | 仅单最值，窗口固定 | 均摊O(1) |
| SortedList | 窗口变动，按值删改多 | O(log n) |
小结
在Python实时数据流中，动态最值查找应根据窗口特征和操作类型选型。固定窗口单最值优先用单调队列，双最值可用堆，复杂变动场景用平衡树。合理运用这些策略能显著降低延迟。