堆、选择与流 · HTML
数学 / 逻辑形态
lower 为 max-heap，upper 为 min-heap；max(lower) <= min(upper)，且 lower 与 upper 等大或 lower 恰多一项。
识别信号
- 数据逐项到达，每次插入后查询中位数。
- 答案依赖有序集合中间边界，但不需完整排序。
- 集合要动态平分为较小与较大两半。
- 删除不是主操作，或可另加惰性删除。
- 希望插入 O(log n)、查询 O(1)。
容易误判的相似信号
| Signal | Why it is different |
|---|
| 静态数组只查一次 | Quickselect 期望线性；双堆为在线能力支付 log n。 |
| 滑动窗口中位数 | 仍可用双堆，但必须加 delayed deletion、有效大小计数和堆顶清理。 |
| 值域很小 | 频次桶与秩扫描可能更简单，并天然支持删除。 |
核心不变量
lower 中每项不大于 upper 中每项；lower.size 等于 upper.size 或恰好多 1。
状态、转移与处理顺序
| Item | Definition |
|---|
| State | 存负值的 lower 小根堆、upper 小根堆；删除版另有 delayed 计数与逻辑大小。 |
| Transition | 按 lower 顶把新值放入一侧，再把超出大小约定的一侧堆顶移动到另一侧。 |
| Frontier / order | lower 顶是较小半区最大值，upper 顶是较大半区最小值；它们构成中位边界。 |
Python 骨架
from heapq import heappop, heappush
class MedianStream:
def __init__(self):
self.lower, self.upper = [], []
def add(self, x):
if not self.lower or x <= -self.lower[0]:
heappush(self.lower, -x)
else:
heappush(self.upper, x)
if len(self.lower) > len(self.upper) + 1:
heappush(self.upper, -heappop(self.lower))
elif len(self.upper) > len(self.lower):
heappush(self.lower, -heappop(self.upper))
def median(self):
if not self.lower:
raise ValueError('empty stream')
if len(self.lower) > len(self.upper):
return -self.lower[0]
return (-self.lower[0] + self.upper[0]) / 2
正确性思路
- 按边界放入后，跨堆有序性保持。
- 若一侧过大，其堆顶是最靠近分界且唯一应跨界的元素；移动它同时恢复大小并保持顺序。
- 大小规则使中间一个或两个秩恰位于堆顶，因此查询正确。
复杂度
| Time | Space | Parameters |
|---|
| add O(log n)，median O(1) | O(n) | n 为当前有效元素数。 |
边界条件
- 空流查询
- 重复值
- 负数与 lower 取负
- 偶数长度的返回类型
- 删除版物理堆长不等于有效大小
错误模式
| Pattern | Why it fails / fix |
|---|
| 只平衡大小 | 两个随机半区即使等大，堆顶也不是中位边界；还必须维护跨堆顺序。 |
| lower 的负号遗漏 | 比较、移动、返回必须区分物理负值与逻辑值。 |
| 滑窗用 list.remove | 破坏堆复杂度；应 delayed counter，并维护有效大小。 |
| 未约定哪边多一项 | 插入、再平衡和奇数查询会彼此矛盾。 |
选择与排除规则
- 持续插入并持续查中位数：双堆。
- 静态一次查询：Quickselect。
- 滑动窗口：双堆加惰性删除或有序多重集合。
- 有限小值域：考虑频次桶。
相关模板
原创微型例子
| 环节 | 内容 |
|---|
| Result | 边界为 4 与 6，中位数为 5。 |
| Setup | 温度流依次为 4,10,6,2。 |
| Trace | 4 在 lower；10 在 upper；6 先入 upper 后移到 lower；2 入 lower 后把 6 移回 upper。 |
Recall prompts
我的补充