双堆法用 heapq 维护最大堆(存负值)和最小堆动态求中位数,时间 O(n log k)、空间 O(k);流式数据首选此法;大数据可选 numpy.percentile(平均 O(n))或计数排序(O(n+range),需已知值域)。

用 heapq 实现双堆法找中位数,不排序也能 O(n log k)
直接排序整个列表(sorted() 或 list.sort())在数据量大时太重——时间复杂度 O(n log n),内存也扛不住。真正高效的做法是维护两个堆:一个最大堆存较小一半,一个最小堆存较大一半。Python 没有内置最大堆,但可以用 heapq 存负值模拟。
关键点在于动态平衡两堆大小:
- 小根堆(
large)存较大半,直接用heapq.heappush(large, x) - 大根堆(
small)存较小半,存-x,取时用-small[0] - 每次插入后检查长度差是否 ≤1;否则把多出的堆顶移过去
中位数就看两堆顶:len(small) == len(large) 时取平均,否则取长堆的堆顶(注意 small 要加负号)。
处理流式数据或内存受限场景,优先选 numpy.percentile + 分块
如果数据来自文件、数据库或生成器,无法一次性加载进内存,heapq 双堆仍是首选——它只存最多 O(n) 个元素中的约 2×中位数邻域数据,空间可控。
立即学习“Python免费学习笔记(深入)”;
但如果数据能转成 numpy.ndarray(比如从 CSV 读入后切片),且你接受轻微近似,numpy.percentile(arr, 50, method='linear') 内部用 introselect 类算法,平均 O(n),比全排序快不少,还支持 axis 和 out 参数节省内存。
注意坑点:
Python 编程规范与最佳实践。强制 PEP 8 代码风格、语法校验、单元测试执行,适用于代码编写、审查和重构。触发词:Python、代码规范、py_compile、单元测试。
-
method默认是'linear',对偶数长度返回插值结果;若要严格按定义(下中位数),改用method='lower' - 传入 float32 数组时,
percentile可能降精度,建议显式指定dtype=np.float64 - 超大数组(> RAM)别硬转
np.array,先用np.memmap或分块统计频次再估算
当数据范围有限(如整数、固定精度浮点),试试计数排序思路
如果知道数据分布范围窄(比如传感器读数在 0–1000 之间,精度到 0.1),可把值映射为桶索引,用数组计数。中位数位置 = 总数 // 2(向下取整),然后顺序累加频次直到跨过该位置。
优势是 O(n + range),远快于堆或 select;劣势是内存随范围线性增长,且需预知 min/max。
实操示例:
counts = [0] * 10001 # 覆盖 0.0 ~ 1000.0,步长 0.1 → 10001 个桶
for x in data_stream:
idx = int(round(x * 10) # 0.1→0, 0.15→2...
if 0 <= idx < len(counts):
counts[idx] += 1
<p>target = len(data_stream) // 2
cum = 0
for i, c in enumerate(counts):
cum += c
if cum > target:
median = i / 10.0
break</p>别踩 statistics.median 的坑:它会强制转 list 并排序
看起来最省事的 statistics.median(iterable) 实际上内部做了 sorted(list(iterable)) ——对生成器或大数据集,这等于把全部数据拉进内存再排序,完全违背“无需完全排序”的初衷。
除非你确认 iterable 是小列表(pandas.Series.median() 虽然底层优化过,但 DataFrame 整体加载仍可能爆内存,不如先用 chunksize 分批算频次或用 dask.dataframe。
真正要小心的是“看似懒实则急”的接口:它们不报错,但悄悄吃光内存,最后卡死在 GC 阶段,连错误都来不及抛。

















