对顶堆实战：5分钟掌握动态中位数计算的Python实现（附LeetCode例题）

对顶堆实战：5分钟掌握动态中位数计算的Python实现（附LeetCode例题）

在算法面试和竞赛中，动态中位数问题是一个经典的高频考点。想象这样一个场景：数据流不断涌入，你需要实时维护并快速返回当前所有数字的中位数。传统做法每次查询都重新排序显然效率太低，而对顶堆（Dual Heap）结构能以O(logN)时间复杂度优雅解决这个问题。本文将用Python带你从零实现这一数据结构，并解决LeetCode 295. Find Median from Data Stream等实际问题。

1. 什么是对顶堆？

对顶堆由两个堆组成：一个大根堆保存较小的一半数字，一个小根堆保存较大的一半数字。这种结构始终保持两个特性：

1. 大根堆的所有元素 ≤ 小根堆的所有元素
2. 两个堆的大小差不超过1

当满足这两个条件时，中位数计算就变得非常简单：

• 如果总数是奇数：元素较多的堆顶即为中位数
• 如果总数是偶数：两个堆顶的平均值即为中位数

import heapq class MedianFinder: def __init__(self): self.max_heap = [] # 保存较小的一半（Python默认小根堆，通过存储负数模拟大根堆） self.min_heap = [] # 保存较大的一半

2. Python实现核心操作

2.1 添加元素

每次新元素插入时需要决定放入哪个堆，并动态平衡两个堆的大小：

def addNum(self, num: int) -> None: if not self.max_heap or num <= -self.max_heap[0]: heapq.heappush(self.max_heap, -num) # 大根堆插入 else: heapq.heappush(self.min_heap, num) # 小根堆插入 # 平衡两个堆的大小 if len(self.max_heap) > len(self.min_heap) + 1: heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap)) elif len(self.min_heap) > len(self.max_heap): heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap))

注意：Python的heapq模块只实现小根堆，我们通过存储负数来模拟大根堆行为

2.2 查询中位数

根据堆的大小关系返回相应结果：

def findMedian(self) -> float: if len(self.max_heap) == len(self.min_heap): return (-self.max_heap[0] + self.min_heap[0]) / 2 else: return -self.max_heap[0]

3. LeetCode 295完整解法

将上述代码组合起来就是完整的解决方案：

import heapq class MedianFinder: def __init__(self): self.max_heap = [] # 较小的一半 self.min_heap = [] # 较大的一半 def addNum(self, num: int) -> None: if not self.max_heap or num <= -self.max_heap[0]: heapq.heappush(self.max_heap, -num) else: heapq.heappush(self.min_heap, num) # Balance heaps if len(self.max_heap) > len(self.min_heap) + 1: heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap)) elif len(self.min_heap) > len(self.max_heap): heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap)) def findMedian(self) -> float: if len(self.max_heap) == len(self.min_heap): return (-self.max_heap[0] + self.min_heap[0]) / 2 else: return -self.max_heap[0]

时间复杂度分析：

• 添加操作：O(logN)
• 查询操作：O(1)

4. 常见错误与调试技巧

4.1 堆平衡不及时

初学者常犯的错误是在插入元素后忘记检查堆的大小平衡。这会导致中位数计算错误：

# 错误示例：忘记平衡堆 def addNum(self, num: int) -> None: if num <= -self.max_heap[0]: heapq.heappush(self.max_heap, -num) else: heapq.heappush(self.min_heap, num) # 缺少平衡代码！

4.2 边界条件处理

当堆为空时的特殊处理：

# 正确处理空堆情况 def addNum(self, num: int) -> None: if not self.max_heap or num <= -self.max_heap[0]: # 检查max_heap是否为空 heapq.heappush(self.max_heap, -num)

4.3 测试用例验证

建议用以下测试案例验证你的实现：

5. 性能优化与变种问题

5.1 延迟删除技巧

在某些场景下需要支持删除操作，可以采用"延迟删除"策略：

• 维护一个哈希表记录待删除元素
• 当元素出现在堆顶时才真正删除
• 空间换时间，保持O(logN)时间复杂度

5.2 滑动窗口中位数

LeetCode 480. Sliding Window Median是对顶堆的进阶应用。解决方案：

1. 维护一个大小固定的对顶堆
2. 窗口滑动时删除离开的元素，添加新元素
3. 使用延迟删除处理非堆顶元素

def medianSlidingWindow(nums: List[int], k: int) -> List[float]: pass # 实现留给读者练习

5.3 多语言实现对比

不同语言的对顶堆实现差异：

掌握对顶堆不仅能够解决动态中位数问题，它的思想还可以扩展到：

• 实时统计系统的百分位数
• 股票市场中间价计算
• 游戏中的动态难度调整