基于双队列实现的RunningMax类（已验证正确性）优化咨询

你已经实现的RunningMax类能够正确计算滑动窗口最大值，输出与Pandas的滚动最大值完全一致，接下来我们看看这个实现的优化空间，以及更高效的写法：

当前实现的冗余点 #

1. 多余的values队列：维护self.values只是为了判断队首最大值是否滑出窗口，但这个信息可以通过给最大值队列绑定索引来替代，无需额外存储所有元素。
2. append方法中无效的循环条件：i < min(len(self.values),self.window)这个限制没有实际意义——只要新元素比maxima队尾元素小，就应该一直弹出，直到队列为空或遇到更大的元素，这个i的限制反而属于冗余判断。
3. 拆分的append与pop调用：在compute_max中每次先调用append再调用pop，两次函数调用可以合并为一次，减少调用开销。

优化后的实现 #

from collections import deque

class RunningMax:
    def __init__(self, window):
        self.window = window
        self.maxima = deque()  # 存储(元素值, 元素索引)的元组
        self.current_idx = 0

    def add_and_get_max(self, x):
        # 移除所有比当前元素小的历史元素，它们不可能成为后续窗口的最大值
        while self.maxima and self.maxima[-1][0] <= x:
            self.maxima.pop()
        # 将当前元素和其索引加入队列
        self.maxima.append((x, self.current_idx))
        
        # 移除已经滑出当前窗口的最大值
        while self.maxima[0][1] <= self.current_idx - self.window:
            self.maxima.popleft()
        
        # 当前窗口的最大值就是队列首元素的值
        max_val = self.maxima[0][0]
        self.current_idx += 1
        return max_val

def compute_max(a, window):
    rm = RunningMax(window)
    return [rm.add_and_get_max(ai) for ai in a]

优化效果说明 #

• 内存开销降低：去掉了values队列，仅用maxima队列存储可能成为最大值的元素及其索引，内存占用更少，尤其当窗口较大时更明显。
• 逻辑更简洁高效：移除冗余判断，合并操作后，减少了不必要的循环条件检查和函数调用次数，整体执行速度更快。
• 鲁棒性提升：通过索引判断元素是否在窗口内，逻辑更直观，避免了维护双队列可能带来的同步错误。

可以用以下代码验证优化后的实现正确性：

import pandas as pd
import numpy as np

# 生成测试数据
test_data = np.random.randint(0, 100, size=1000).tolist()
window_size = 15

# 对比结果
optimized_result = compute_max(test_data, window_size)
pandas_result = pd.Series(test_data).rolling(window_size, min_periods=1).max().tolist()

print(optimized_result == pandas_result)  # 输出True

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