第 06 章 堆与优先队列

本章覆盖原书第二部分「排序与顺序统计量」的导言和第 6 章「堆排序」。原书第 2 章的插入排序、归并排序与第 3 章的渐近记号见本册第 01、03 章；本章起默认读者熟悉 \(O,\Theta,\Omega\) 与循环不变式（loop invariant）。

学习目标

1. 说清排序问题的输入输出、"关键字 + 卫星数据"的记录模型，以及原书第二部分七种排序算法的复杂度对比。
2. 掌握二叉堆的数组表示、最大堆/最小堆性质，能手工演示 MAX-HEAPIFY、BUILD-MAX-HEAP、HEAPSORT。
3. 能推导 MAX-HEAPIFY 的 \(O(\lg n)\) 与 BUILD-MAX-HEAP 的 \(O(n)\)（而不是 \(O(n\lg n)\)）。
4. 会用堆实现最大/最小优先队列的四个操作，理解"句柄"（handle）为什么在实际系统里必不可少。
5. 能用堆写出量化系统中的常用组件：价格-时间优先订单簿、流式 Top-K 选股、双堆滚动中位数、多路行情归并和事件驱动回测的事件队列。

读前导读

这一章在解决什么问题

这一章第一次正式讲数据结构。数据结构就是"数据在内存里怎么摆放"的约定，摆法决定了哪些操作快、哪些慢。这和公司档案的归档方式一样：按日期归档，查"上周的凭证"很快，查"某客户的全部凭证"就很慢；按客户归档则反过来。没有一种摆法所有操作都快，要看你最常做什么。

本章的主角是堆，它专门为一类需求服务：数据不断进进出出，而你随时都要拿到"当前最大（或最小）的那个"。量化里这种需求遍地都是：

• 订单簿：随时要知道最高买价和最低卖价，同时不断有挂单、撤单、成交。
• 事件驱动回测：成千上万个未来事件（行情、成交回报、定时任务），每次都要取出"时间最早"的那个。
• 实时排行榜：全市场 5000 只股票的因子得分不断流入，只想保留前 10 名。

两种朴素办法各有短板：数据始终保持完全排好序，取最大值很快，但每插入一个新数据可能要挪动一大片（\(n\) 量级）；数据不排序，插入很快，但每次找最大值要全部扫一遍（也是 \(n\) 量级）。堆走中间路线：只保证"每个上级不小于它的直接下级"，像一个只要求"领导比下属强"、不要求同级之间排名的组织架构。这么一个弱条件就足以让"最强者"永远在顶端，而插入和删除只需要沿着一条从顶到底的汇报线调整，代价是树的层数 \(\lg n\)。用"翻倍"来说：数据翻倍，堆只多一层，每次操作只多一步。100 万个数据的堆只有 20 层。

本章还有一个反直觉的结论：把一个乱序数组整理成堆只需要 \(O(n)\) 时间，而不是看上去的 \(O(n\lg n)\)。原因是绝大多数结点都在底层，几乎不用动。

需要先想起来的数学

• 二叉树的基本词汇。 树由"结点"组成，最上面的叫根，每个结点下面最多挂两个"孩子"，没有孩子的叫"叶子"。从根往下数，第 0 层 1 个结点、第 1 层 2 个、第 2 层 4 个……第 \(h\) 层 \(2^h\) 个。所以前 \(h+1\) 层装满共 \(2^{h+1}-1\) 个结点，例如 3 层装满是 7 个。结点的"高度"是它往下到叶子的最长步数。
• 对数与层数。 \(n\) 个结点的堆高度是 \(\lfloor\lg n\rfloor\)。例：\(n=10\) 时 \(\lg10\approx3.32\)，高度 3（共 4 层）。详见 第 00 册第 04 章 级数与收敛。
• 几何级数及其"加权"版本。 \(\sum_{k\ge0}x^k=\frac1{1-x}\)；两边对 \(x\) 求导再乘 \(x\)，得 \(\sum_{k\ge0}kx^k=\frac{x}{(1-x)^2}\)。这个"加权"级数你其实见过：它就是计算永续年金麦考利久期时的分子。建堆是 \(O(n)\) 的证明靠它。详见 第 00 册第 04 章 级数与收敛 与 第 00 册第 02 章 导数与泰勒展开。
• 取整。 \(\lfloor i/2\rfloor\) 是"\(i\) 除以 2 后向下取整"。\(\lfloor 7/2\rfloor=3\)，\(\lfloor 6/2\rfloor=3\)：结点 6 和 7 的父亲都是结点 3。

怎么读这一章

6.0 节是第二部分的导言，那张七种排序的对比表值得收藏，其余可以快速浏览。6.1 节（数组怎么表示树）、6.2 节（下沉操作 MAX-HEAPIFY）和 6.5 节（优先队列）是核心，建议拿纸画出原书例子的树形，对着数组下标手动走一遍。6.3 节的"建堆为什么是线性时间"是本章唯一的数学推导，第一次读可以只记结论和"大多数结点很矮"这个直觉。6.4 节堆排序很短，顺着读即可。6.5 节末尾的"几个有用的扩展"和"更快的优先队列"第一次可以跳过。量化实战里订单簿和事件队列两段最贴近实务，代码较长，先读每段之后的文字说明，再回头看代码。

6.0 第二部分导言：为什么花这么多篇幅讲排序

排序问题与记录

排序问题（sorting problem）的输入是 \(n\) 个数 \(\langle a_1,a_2,\dots,a_n\rangle\)，输出是它的一个排列 \(\langle a'_1,\dots,a'_n\rangle\)，满足 \(a'_1\le a'_2\le\cdots\le a'_n\)。

实践中被排序的很少是孤立的数，而是记录（record）。每条记录有一个关键字（key），排序就按它来排；其余部分叫卫星数据（satellite data），跟着关键字一起移动。卫星数据很大时，通常只重排指向记录的指针数组，而不搬动记录本身。量化里最常见的例子是：一条"股票-日期"记录的关键字是因子值，卫星数据是代码、行业、市值、下期收益；排序时我们其实是在重排行索引（argsort），而不是复制整张表。

算法讨论时一律假设"输入就是数"，因为排序算法只关心如何确定顺序，与排的是单个数还是大记录无关。把它落成程序时的工程细节（内存布局、指针还是复制）则可能影响很大。

为什么研究排序

原书给出五个理由：有些应用天然需要排序；很多算法把排序当关键子程序；排序算法汇集了分治、随机化、数据结构等几乎所有核心设计技术；排序有可证明的非平凡下界（第 8 章），而最好的上界与之匹配，所以我们知道现有算法是渐近最优的；实现排序会碰到大量工程问题（cache、虚拟内存、对关键字分布的先验知识），很多问题最好在算法层面而不是靠"调代码"解决。

本部分的算法总览

前四种是比较排序（comparison sort），只通过比较元素获得顺序信息，第 08 章会证明它们在最坏情况下至少要 \(\Omega(n\lg n)\) 次比较；后三种利用了输入的额外信息（整数范围、位数、分布），因此能突破这一下界。

第 09 章讨论顺序统计量（order statistics）：第 \(i\) 小的元素。排序后取第 \(i\) 个需要 \(\Omega(n\lg n)\)，但选择问题其实可以在 \(O(n)\) 内解决，这对计算截面分位数和中位数非常重要。

阅读这一部分需要的数学不多，但快速排序、桶排序和随机选择的分析要用到指示器随机变量（第 5 章；概率基础见第 02 册）。

6.1 堆：用数组装下一棵树

直觉

我们想要一个结构，能随时拿到"当前最大的元素"，插入和删除都比较快。排好序的数组拿最大值是 \(O(1)\)，但插入要 \(O(n)\)；无序数组插入 \(O(1)\)，找最大要 \(O(n)\)。堆处在中间：它只维护一个"偏序"——父亲不小于孩子——这个弱条件足以让最大值待在根上，同时把插入、删除的代价压到树高 \(O(\lg n)\)。

定义

二叉堆（binary heap）是一个数组，可看成一棵近似完全二叉树（nearly complete binary tree）：除最底层外每层都是满的，最底层从左向右填到某处为止。数组 \(A\) 有两个属性：A.length 是数组容量，A.heap-size 是其中属于堆的元素个数，只有 \(A[1..\text{heap-size}]\) 是有效元素。根是 \(A[1]\)。用 1 起始下标时：

三者都可以用一条移位指令完成。Python 等 0 起始的语言里改为 \(\text{parent}(i)=\lfloor (i-1)/2\rfloor\)、\(\text{left}(i)=2i+1\)、\(\text{right}(i)=2i+2\)，本章代码都用这一套。

堆性质（heap property）有两种：

• 最大堆（max-heap）：除根以外的每个结点 \(i\) 都有 \(A[\text{PARENT}(i)]\ge A[i]\)。于是最大元素在根上，任何子树的根都是该子树的最大值。
• 最小堆（min-heap）：\(A[\text{PARENT}(i)]\le A[i]\)，最小元素在根上。

堆排序用最大堆；优先队列两者都用，量化里最小堆出现得更多（事件按时间最早、卖单按价格最低）。

例（原书图 6.1）：\(A=\langle16,14,10,8,7,9,3,2,4,1\rangle\) 是一个最大堆。结点 2（值 14）的孩子是结点 4、5（值 8、7），结点 4 的孩子是结点 8、9（值 2、4）。父结点在数组中总在孩子左边。

白话解释：把树"按层从左到右"编号，再按编号依次放进数组，就不需要任何额外的指针来记录谁是谁的孩子，编号本身就说明了亲属关系。把上面的例子画出来： 第 0 层：结点 1（16） 第 1 层：结点 2（14）、结点 3（10） 第 2 层：结点 4（8）、5（7）、6（9）、7（3） 第 3 层：结点 8（2）、9（4）、10（1） 每层第一个结点的编号是 1、2、4、8，正好翻倍，所以"孩子编号 = 父亲编号 × 2（或 × 2 + 1）"，"父亲编号 = 孩子编号 ÷ 2 取整"。例如结点 5 的孩子是 10 和 11（11 不存在），结点 9 的父亲是 \(\lfloor9/2\rfloor=4\)。 最大堆性质只约束"上下级"：每条从根到叶的路径上，数值单调不增（如 16→14→8→4）。同一层之间没有顺序要求，比如第 2 层的 8、7、9、3 是乱的，这就是"偏序"的意思。

高度

结点的高度（height）是它到叶子的最长向下简单路径上的边数，堆的高度就是根的高度。含 \(n\) 个元素的堆高度为 \(\lfloor\lg n\rfloor\)：高度为 \(h\) 的堆至少有 \(2^h\) 个元素（前 \(h\) 层满、最底层 1 个），至多有 \(2^{h+1}-1\) 个元素（全满），所以 \(2^h\le n\le 2^{h+1}-1\)，取对数即得（练习 6.1-1、6.1-2）。

另一个后面要反复用的事实：在 \(n\) 元堆里，叶子恰好是下标 \(\lfloor n/2\rfloor+1,\dots,n\)（练习 6.1-7）。原因是结点 \(i\) 有左孩子当且仅当 \(2i\le n\)。

所有堆操作的代价都至多和树高成正比，因此是 \(O(\lg n)\)。本章的过程一览：

6.2 维护堆的性质：MAX-HEAPIFY

MAX-HEAPIFY\((A,i)\) 的前提是：以 LEFT\((i)\) 和 RIGHT\((i)\) 为根的子树都已经是最大堆，只有 \(A[i]\) 可能比孩子小。它让 \(A[i]\) 一层层"下沉"（float down），直到以 \(i\) 为根的子树满足最大堆性质。

MAX-HEAPIFY(A, i)
1  l = LEFT(i);  r = RIGHT(i)
2  if l <= A.heap-size and A[l] > A[i]  then largest = l  else largest = i
3  if r <= A.heap-size and A[r] > A[largest]  then largest = r
4  if largest != i
5      exchange A[i] with A[largest]
6      MAX-HEAPIFY(A, largest)

每一步在 \(A[i]\) 和两个孩子中找出最大者。若 \(A[i]\) 已最大，子树已是最大堆，结束；否则把较大的孩子换上来，此时结点 \(i\) 合格了，但被换下去的旧 \(A[i]\) 可能又比它的新孩子小，于是对那个位置递归。

例（原书图 6.2）：heap-size \(=10\)，\(A=\langle16,4,10,14,7,9,3,2,8,1\rangle\)，调用 MAX-HEAPIFY\((A,2)\)。\(A[2]=4\) 比孩子 14 小，与 \(A[4]\) 交换得 \(\langle16,14,10,4,7,9,3,2,8,1\rangle\)；在 \(i=4\) 处，4 比孩子 8（\(A[9]\)）小，交换得 \(\langle16,14,10,8,7,9,3,2,4,1\rangle\)；在 \(i=9\) 处已是叶子，结束。

运行时间。在规模为 \(n\) 的子树上，比较和交换花 \(\Theta(1)\)，然后在一个孩子的子树上递归。孩子子树最多含 \(2n/3\) 个结点——最坏情况是最底层恰好填满一半，此时左子树比右子树多一整层。于是

由主定理情况 2（第 04b 章）得 \(T(n)=O(\lg n)\)。等价地说，在高度为 \(h\) 的结点上调用的代价是 \(O(h)\)。

推导拆解：\(2n/3\) 从哪来？设左子树是高度为 \(h\) 的满树、右子树是高度为 \(h-1\) 的满树（最底层恰好只填满了左半边）。高度 \(h\) 的满二叉树有 \(2^{h+1}-1\) 个结点，所以左子树约 \(2^{h+1}\) 个、右子树约 \(2^h\) 个，整棵树约 \(2^{h+1}+2^h=3\cdot2^h\) 个。左子树占比约 \(\frac{2^{h+1}}{3\cdot2^h}=\frac23\)。 递归式 \(T(n)\le T(2n/3)+\Theta(1)\) 的意思是：每下沉一层花常数时间，问题规模至少缩到原来的 \(\frac23\)。规模要乘多少次 \(\frac23\) 才降到 1？\(\log_{3/2}n\) 次，这是 \(\lg n\) 的常数倍。更直接的说法：下沉最多走到叶子，路径长度不超过树高 \(\lfloor\lg n\rfloor\)。递归版需要 \(O(\lg n)\) 栈空间；把第 6 行改成循环（练习 6.2-5）就只需 \(O(1)\) 额外空间，这也是工程实现的标准写法。

练习 6.2-6 说明这个界是紧的：把最小值放在根上，它会一路沉到叶子，每层都要交换，所以最坏代价是 \(\Omega(\lg n)\)。

6.3 建堆：为什么是线性时间

叶子 \(A[\lfloor n/2\rfloor+1..n]\) 各自已经是 1 个元素的堆。BUILD-MAX-HEAP 从最后一个非叶结点开始，自底向上对每个结点调用 MAX-HEAPIFY：

BUILD-MAX-HEAP(A)
1  A.heap-size = A.length
2  for i = floor(A.length / 2) downto 1
3      MAX-HEAPIFY(A, i)

例（原书图 6.3）：\(A=\langle4,1,3,2,16,9,10,14,8,7\rangle\)。依次处理 \(i=5,4,3,2,1\)：\(i=5\)（16）不动；\(i=4\) 时 2 与 14 交换；\(i=3\) 时 3 与 10 交换；\(i=2\) 时 1 先与 16、再与 7 交换；\(i=1\) 时 4 依次与 16、14、8 交换。结果 \(\langle16,14,10,8,7,9,3,2,4,1\rangle\)。

正确性。循环不变式：每次第 2 行迭代开始时，结点 \(i+1,i+2,\dots,n\) 都是某个最大堆的根。

• 初始化：\(i=\lfloor n/2\rfloor\) 时，\(\lfloor n/2\rfloor+1..n\) 全是叶子，平凡成立。
• 保持：结点 \(i\) 的孩子编号都大于 \(i\)，按不变式它们都是最大堆的根，这正是 MAX-HEAPIFY\((A,i)\) 的前提；调用后 \(i\) 也成为最大堆的根，且 \(i+1..n\) 的性质不被破坏。
• 终止：\(i=0\)，于是结点 1 是最大堆的根。

为什么必须从 \(\lfloor n/2\rfloor\) 递减到 1，而不能递增（练习 6.3-2）？因为 MAX-HEAPIFY 要求孩子子树已经是堆，只有自底向上的顺序能保证这一点。

运行时间的精细分析。粗略看，\(O(n)\) 次调用、每次 \(O(\lg n)\)，总共 \(O(n\lg n)\)——正确但不紧。关键观察是：大多数结点都很矮。\(n\) 元堆中高度为 \(h\) 的结点至多 \(\lceil n/2^{h+1}\rceil\) 个（练习 6.3-3），在高度 \(h\) 上调用 MAX-HEAPIFY 花 \(O(h)\)，所以总代价

利用级数 \(\sum_{k\ge0}kx^k=x/(1-x)^2\)（\(|x|<1\)），代入 \(x=1/2\) 得

故 BUILD-MAX-HEAP 是 \(O(n)\)。直觉是：一半结点是叶子不用动，四分之一结点最多下沉 1 层，八分之一最多下沉 2 层……总下沉次数不超过 \(n\cdot(0/2+1/4+2/8+3/16+\cdots)=n\) 层左右。下面的实验中，随机排列建堆的交换次数稳定在约 \(0.74n\)。

推导拆解：分三步看这个求和。 第一步，"按高度分组记账"。不按结点逐个算，而是问：高度为 \(h\) 的结点有几个、每个最多花多少。高度 \(h\) 的结点至多 \(\lceil n/2^{h+1}\rceil\) 个（高度 0 即叶子约 \(n/2\) 个，高度 1 约 \(n/4\) 个……），每个下沉最多 \(h\) 层。所以总量 \(\approx\sum_h\frac{n}{2^{h+1}}\cdot h=\frac n2\sum_h\frac{h}{2^h}\)。 第二步，级数公式 \(\sum_{k\ge0}kx^k=\frac{x}{(1-x)^2}\) 的来历：从几何级数 \(\sum_kx^k=\frac1{1-x}\) 出发，两边对 \(x\) 求导得 \(\sum_kkx^{k-1}=\frac1{(1-x)^2}\)，再两边乘 \(x\) 即得。代入 \(x=\frac12\) 得 2。把有限项换成无穷项只会让和变大，所以仍是上界。 第三步，结论：总量 \(\le\frac n2\cdot2=n\) 量级，即 \(O(n)\)。 金融直觉：\(\frac{\sum_kkx^k}{\sum_kx^k}\) 正是现金流权重按 \(x^k\) 衰减时的"加权平均期限"，也就是永续年金的麦考利久期 \(\frac{1+r}{r}\)（这里 \(x=\frac1{1+r}\)）。\(x=\frac12\) 相当于 \(r=100\%\)，久期只有 2 期：贴现得越狠，权重越集中在近处，加权平均期限越短。建堆是同一回事：结点数按高度每层减半，权重集中在矮结点上，按结点数加权的平均下沉深度至多约 1 层，是个与 \(n\) 无关的常数。 为什么粗估是 \(O(n\lg n)\)？因为它假设每个结点都可能下沉 \(\lg n\) 层，但只有根附近极少数结点才有这么深的路可走。

BUILD-MIN-HEAP 只需把 MAX-HEAPIFY 换成 MIN-HEAPIFY，同样是线性时间。Python 的 heapq.heapify 就是这个算法。

思考题 6-1 对比了另一种建堆方法：逐个 MAX-HEAP-INSERT。它在最坏情况（升序输入，每个新元素都要上浮到根）下是 \(\Theta(n\lg n)\)，而且与 BUILD-MAX-HEAP 得到的堆不一定相同（输入 \(\langle1,2,3\rangle\) 时一个得 \(\langle3,1,2\rangle\)，一个得 \(\langle3,2,1\rangle\)）。所以一次性拿到全部数据时，应该用 heapify 而不是循环 heappush。

6.4 堆排序

有了最大堆，排序就很自然：根是最大值，把它和最后一个元素交换，它就落到了最终位置；把堆规模减 1，新根可能违反性质，调用一次 MAX-HEAPIFY\((A,1)\) 修复；重复到堆只剩 1 个元素。

HEAPSORT(A)
1  BUILD-MAX-HEAP(A)
2  for i = A.length downto 2
3      exchange A[1] with A[i]
4      A.heap-size = A.heap-size - 1
5      MAX-HEAPIFY(A, 1)

循环不变式（练习 6.4-2）：第 2 行每次迭代开始时，\(A[1..i]\) 是一个最大堆，包含 \(A[1..n]\) 中最小的 \(i\) 个元素；\(A[i+1..n]\) 已排好序，包含最大的 \(n-i\) 个元素。

例（原书图 6.4）：从最大堆 \(\langle16,14,10,8,7,9,3,2,4,1\rangle\) 出发，堆顶依次是 \(16,14,10,9,8,7,4,3,2\)，每次被换到末尾，最终 \(A=\langle1,2,3,4,7,8,9,10,14,16\rangle\)。

运行时间：建堆 \(O(n)\)，再做 \(n-1\) 次 \(O(\lg n)\) 的 MAX-HEAPIFY，总共 \(O(n\lg n)\)；且这个界是紧的——即使输入已排好序（升序或降序），堆排序也是 \(\Theta(n\lg n)\)（练习 6.4-3），元素互异时最好情况也是 \(\Omega(n\lg n)\)（练习 6.4-5★）。

堆排序的优点是原地、最坏 \(O(n\lg n)\)、不需要随机数；缺点是不稳定，而且访问模式在数组里跳来跳去，对 cache 不友好，实践中通常比精心实现的快速排序慢。它的真正价值在于两处：作为 introsort（第 07 章）的保底方案，以及堆这个数据结构本身。

6.5 优先队列

定义与应用

优先队列（priority queue）维护一个元素集合 \(S\)，每个元素有关键字。最大优先队列支持：

• INSERT\((S,x)\)：\(S\leftarrow S\cup\{x\}\)；
• MAXIMUM\((S)\)：返回关键字最大的元素；
• EXTRACT-MAX\((S)\)：删除并返回关键字最大的元素；
• INCREASE-KEY\((S,x,k)\)：把 \(x\) 的关键字增大到 \(k\)（要求 \(k\) 不小于原值）。

最小优先队列对称地支持 INSERT、MINIMUM、EXTRACT-MIN、DECREASE-KEY。原书举了两个应用：分时系统的作业调度（最大优先队列，随时取优先级最高的作业），以及事件驱动模拟（最小优先队列，以事件发生时间为关键字，每次取最早的事件；模拟一个事件可能产生新的未来事件，用 INSERT 加入）。后者就是事件驱动回测引擎的骨架。

用堆实现

HEAP-MAXIMUM(A)                         // Θ(1)
1  return A[1]

HEAP-EXTRACT-MAX(A)                     // O(lg n)
1  if A.heap-size < 1  error "heap underflow"
2  max = A[1]
3  A[1] = A[A.heap-size]
4  A.heap-size = A.heap-size - 1
5  MAX-HEAPIFY(A, 1)
6  return max

HEAP-INCREASE-KEY(A, i, key)            // O(lg n)
1  if key < A[i]  error "new key is smaller than current key"
2  A[i] = key
3  while i > 1 and A[PARENT(i)] < A[i]
4      exchange A[i] with A[PARENT(i)]
5      i = PARENT(i)

MAX-HEAP-INSERT(A, key)                 // O(lg n)
1  A.heap-size = A.heap-size + 1
2  A[A.heap-size] = -∞
3  HEAP-INCREASE-KEY(A, A.heap-size, key)

HEAP-EXTRACT-MAX 与堆排序的循环体几乎一样：把最后一个元素搬到根，再下沉。HEAP-INCREASE-KEY 则相反：关键字变大后它只可能比父亲大，于是像插入排序的内循环一样沿着到根的路径"上浮"，路径长 \(O(\lg n)\)。

例（原书图 6.5）：在最大堆 \(\langle16,14,10,8,7,9,3,2,4,1\rangle\) 中把下标 9 的元素（值 4）增大到 15：先与父亲 8 交换，再与父亲 14 交换，此时父亲 16 \(\ge\) 15，停止。

MAX-HEAP-INSERT 先追加一个关键字为 \(-\infty\) 的叶子，再把它"增大"到目标值。设成 \(-\infty\) 是为了让 HEAP-INCREASE-KEY 的前提"新值不小于旧值"一定成立（练习 6.5-4）。

结论：在 \(n\) 元集合上，堆能在 \(O(\lg n)\) 时间内支持任何优先队列操作。

金融直觉：把最大堆想成一个买单簿，堆顶永远是最优买价。三种操作对应三种场景： 新挂一笔买单（INSERT）：先排到队尾（数组末尾），如果它比"上级"出价高，就一级级往上挤，直到上级比它高为止，即"上浮"。 最优买单成交离场（EXTRACT-MAX）：堆顶空出来，临时把队尾那笔拉上来顶位，它多半不够格，于是和两个下级中出价较高者换位，一级级往下沉，即"下沉"。 某笔买单改价调高（INCREASE-KEY）：它只可能比上级更强，不可能比下级更弱，所以只需上浮。 每种情况都只沿一条上下级链条调整，链条长度就是树高 \(\lg n\)。10 万笔挂单时最多调整约 17 步。

句柄

真实应用里，队列元素对应外部对象（作业、事件、订单）。经常需要双向定位：从堆元素找到对象（在堆元素里存对象的指针或 ID），以及从对象找到它在堆中的位置（在对象里存数组下标）。后者麻烦在于元素在堆操作中不断移动，每次交换都必须更新下标。原书说"本书不展开，但实践中必须正确维护"。下面的订单簿例子里，pos 字典就是句柄表，它让撤单（任意位置删除）成为 \(O(\lg n)\)。

白话解释：句柄就是"每笔订单当前坐在第几号座位"的登记表。堆只擅长处理堆顶，但撤单要删的是任意一笔。如果不登记座位号，就得把整个数组翻一遍才能找到它（\(O(n)\)）；有了登记表，一查就知道在哪，然后用"末尾元素补位 + 上浮或下沉"在 \(O(\lg n)\) 内删掉。麻烦在于座位一直在换：每次上浮、下沉都会交换两笔订单的位置，登记表必须同步更新，漏一次就会删错单。这就是代码里 _swap 每次都改写 pos 的原因。

几个有用的扩展

• 删除任意元素（练习 6.5-8）：HEAP-DELETE\((A,i)\) 用最后一个元素填到位置 \(i\)，然后视情况上浮或下沉，\(O(\lg n)\)。
• 一次赋值代替交换（练习 6.5-6）：上浮时不必每层交换（三次赋值），可以像插入排序那样把父亲往下挪，最后一次性写入 key。
• 用优先队列实现队列和栈（练习 6.5-7）：给元素一个递增的时间戳当关键字，用最小优先队列得到先进先出，用最大优先队列得到后进先出。这也提示我们：在同价位按到达顺序排队（时间优先）只需把"(价格, 序号)"当复合关键字。
• \(k\) 路归并（练习 6.5-9）：\(k\) 个有序表共 \(n\) 个元素，用一个装着各表当前首元素的最小堆，每次取最小者并补入它在原表中的后继，\(O(n\lg k)\) 时间、\(O(k)\) 额外空间。
• \(d\) 叉堆（思考题 6-2）：每个结点 \(d\) 个孩子，数组下标为第 \(j\) 个孩子 \(d(i-1)+j+1\)、父亲 \(\lfloor(i-2)/d\rfloor+1\)（1 起始），高度 \(\Theta(\log_d n)\)。INSERT 和 INCREASE-KEY 只需上浮，\(O(\log_d n)\)；EXTRACT-MAX 下沉时每层要比较 \(d\) 个孩子，\(O(d\log_d n)\)。插入/改键远多于提取时取大 \(d\) 更划算。
• Young 氏矩阵（思考题 6-3）：\(m\times n\) 矩阵每行、每列都有序，空位记为 \(\infty\)。EXTRACT-MIN 取走左上角后像 MAX-HEAPIFY 一样与右边、下边较小者交换，\(O(m+n)\)；插入从右下角向上、向左交换，\(O(m+n)\)；查找某数从右上角出发，大了左移、小了下移，\(O(m+n)\)。这是"二维有序矩阵查找"的标准解法。

更快的优先队列（只需知道存在）

原书章末注记提到：斐波那契堆（原书第 19 章，本册第 18 章）把 DECREASE-KEY 降到摊还 \(O(1)\)；关键字是有界整数时，van Emde Boas 树（原书第 20 章，本册第 18 章）、基数堆（radix heap）等结构更快。一个重要特例是单调优先队列：EXTRACT-MIN 返回的值随时间单调不减。Dijkstra 最短路和离散事件模拟都属于这一类——回测引擎中的事件时间也是单调的，所以在超大规模回测里，按时间分桶的"日历队列"有时比二叉堆更快。日常规模下，二叉堆（Python heapq）已经足够。

6.6 量化实战

6.6.1 从零实现堆，并验证建堆是线性的

下面用 0 起始下标实现迭代版 MAX-HEAPIFY、BUILD-MAX-HEAP 和 HEAPSORT，用原书例子检查，再数一数建堆时的交换次数。

import numpy as np

# ---- 0 基下标版本：parent(i)=(i-1)//2, left=2i+1, right=2i+2 ----
def max_heapify(A, i, heap_size, counter=None):
    """迭代版 MAX-HEAPIFY（练习 6.2-5），让 A[i] 逐级下沉。"""
    while True:
        l, r = 2 * i + 1, 2 * i + 2
        largest = i
        if l < heap_size and A[l] > A[largest]:
            largest = l
        if r < heap_size and A[r] > A[largest]:
            largest = r
        if largest == i:
            return
        A[i], A[largest] = A[largest], A[i]
        if counter is not None:
            counter[0] += 1          # 统计交换次数
        i = largest

def build_max_heap(A, counter=None):
    n = len(A)
    for i in range(n // 2 - 1, -1, -1):   # 自底向上：从最后一个非叶结点到根
        max_heapify(A, i, n, counter)

def heapsort(A):
    build_max_heap(A)
    for end in range(len(A) - 1, 0, -1):
        A[0], A[end] = A[end], A[0]        # 当前最大值放到最终位置
        max_heapify(A, 0, end)             # 堆规模减 1 后修复根
    return A

# 书中例子（图 6.3）与练习 6.3-1
A = [4, 1, 3, 2, 16, 9, 10, 14, 8, 7]
build_max_heap(A); print("图6.3 建堆结果:", A)
B = [5, 3, 17, 10, 84, 19, 6, 22, 9]
build_max_heap(B); print("练习6.3-1 建堆:", B)

# 随机测试正确性
rng = np.random.default_rng(0)
for _ in range(200):
    x = list(rng.integers(0, 50, rng.integers(0, 60)))
    assert heapsort(x[:]) == sorted(x)
print("heapsort 200 组随机测试通过")

# BUILD-MAX-HEAP 是线性时间：交换次数 / n 有上界（理论上 <= n·Σh/2^h = 2n）
for n in [10**3, 10**4, 10**5, 10**6]:
    A = list(rng.permutation(n)); c = [0]
    build_max_heap(A, c)
    print(f"n={n:>8d}  建堆交换次数={c[0]:>8d}  交换/n={c[0]/n:.3f}")

输出：

图6.3 建堆结果: [16, 14, 10, 8, 7, 9, 3, 2, 4, 1]
练习6.3-1 建堆: [84, 22, 19, 10, 3, 17, 6, 5, 9]
heapsort 200 组随机测试通过
n=    1000  建堆交换次数=     723  交换/n=0.723
n=   10000  建堆交换次数=    7415  交换/n=0.742
n=  100000  建堆交换次数=   74478  交换/n=0.745
n= 1000000  建堆交换次数=  743825  交换/n=0.744

图 6.3 和练习 6.3-1 的结果与原书一致。交换次数与 \(n\) 之比稳定在 0.74 左右，不随 \(n\) 增长，直观印证了 \(O(n)\) 的建堆分析。

6.6.2 价格-时间优先的限价订单簿

限价订单簿（limit order book，详见第 07 册）的买方要"最高价优先、同价先到优先"，卖方要"最低价优先、同价先到优先"。把买单关键字设为 \((-\text{价格},\text{序号})\)、卖单设为 \((\text{价格},\text{序号})\)，两边都变成最小堆问题。撤单需要删除堆中任意位置的元素，这就是练习 6.5-8 加上句柄表。

import itertools

class IndexedMinHeap:
    """带句柄（handle）的二叉最小堆：pos[h] 记录句柄 h 当前在数组中的下标。
    支持 push / peek / pop / delete(h)（练习 6.5-8）/ update(h, key)，均为 O(lg n)。"""
    def __init__(self):
        self.a = []          # 元素为 (key, handle)
        self.pos = {}        # handle -> 下标
    def __len__(self):
        return len(self.a)
    def _swap(self, i, j):
        a = self.a
        a[i], a[j] = a[j], a[i]
        self.pos[a[i][1]] = i          # 每次移动都要更新句柄
        self.pos[a[j][1]] = j
    def _up(self, i):                  # 类似 HEAP-DECREASE-KEY 的上浮
        while i > 0 and self.a[(i - 1) // 2][0] > self.a[i][0]:
            self._swap(i, (i - 1) // 2); i = (i - 1) // 2
    def _down(self, i):                # MIN-HEAPIFY 下沉
        n = len(self.a)
        while True:
            l, r, s = 2 * i + 1, 2 * i + 2, i
            if l < n and self.a[l][0] < self.a[s][0]: s = l
            if r < n and self.a[r][0] < self.a[s][0]: s = r
            if s == i: return
            self._swap(i, s); i = s
    def push(self, key, h):
        self.a.append((key, h)); self.pos[h] = len(self.a) - 1
        self._up(len(self.a) - 1)
    def peek(self):
        return self.a[0]
    def delete(self, h):               # 用最后一个元素填洞，再上浮或下沉
        i = self.pos.pop(h)
        last = self.a.pop()
        if i < len(self.a):
            self.a[i] = last; self.pos[last[1]] = i
            self._up(i); self._down(i)
    def pop(self):
        key, h = self.a[0]; self.delete(h); return key, h
    def update(self, h, key):
        i = self.pos[h]; self.a[i] = (key, h); self._up(i); self._down(i)

class OrderBook:
    """价格-时间优先的极简限价订单簿：买方按 (-价格, 序号) 取最小 = 最高价最早；卖方按 (价格, 序号)。"""
    def __init__(self):
        self.bids, self.asks = IndexedMinHeap(), IndexedMinHeap()
        self.orders = {}               # oid -> [side, price, qty]
        self.seq = itertools.count()
        self.trades = []
    def best_bid(self):
        return -self.bids.peek()[0][0] if len(self.bids) else None
    def best_ask(self):
        return self.asks.peek()[0][0] if len(self.asks) else None
    def cancel(self, oid):
        side, price, qty = self.orders.pop(oid)
        (self.bids if side == 'B' else self.asks).delete(oid)
    def limit(self, oid, side, price, qty):
        book = self.asks if side == 'B' else self.bids        # 对手方
        while qty > 0 and len(book):
            (k, _), rid = book.peek()            # k 为价格（买簿中存的是 -价格）
            rprice = k if side == "B" else -k
            if (side == 'B' and rprice > price) or (side == 'S' and rprice < price):
                break                                          # 价格不再可成交
            fill = min(qty, self.orders[rid][2])
            self.trades.append((oid, rid, rprice, fill))
            qty -= fill; self.orders[rid][2] -= fill
            if self.orders[rid][2] == 0:
                book.pop(); del self.orders[rid]
        if qty > 0:                                            # 剩余部分挂单
            self.orders[oid] = [side, price, qty]
            key = (-price, next(self.seq)) if side == 'B' else (price, next(self.seq))
            (self.bids if side == 'B' else self.asks).push(key, oid)

ob = OrderBook()
ob.limit(1, 'B', 100, 5); ob.limit(2, 'B', 101, 3); ob.limit(3, 'B', 101, 4)
ob.limit(4, 'S', 103, 6); ob.limit(5, 'S', 102, 2)
print("挂单后  best bid/ask =", ob.best_bid(), ob.best_ask())
ob.cancel(2)                                   # 撤掉 101 价位上更早的那笔
print("撤单2后 best bid/ask =", ob.best_bid(), ob.best_ask())
ob.limit(6, 'S', 100, 6)                       # 卖单吃掉 101 的 4 手和 100 的 2 手
print("成交记录 (taker, maker, 价, 量):", ob.trades)
print("成交后  best bid/ask =", ob.best_bid(), ob.best_ask(), " 订单1剩余量 =", ob.orders[1][2])

# 随机压力测试：与暴力扫描的最优价比对
import random
random.seed(1)
ob = OrderBook(); live = []
for oid in range(20000):
    if live and random.random() < 0.3:
        o = live.pop(random.randrange(len(live)))
        if o in ob.orders: ob.cancel(o)
    else:
        side = random.choice('BS'); px = 1000 + random.randint(-20, 20)
        ob.limit(oid, side, px, random.randint(1, 10)); live.append(oid)
    if oid % 1000 == 0:
        bb = max([p for s, p, q in ob.orders.values() if s == 'B'], default=None)
        ba = min([p for s, p, q in ob.orders.values() if s == 'S'], default=None)
        assert (bb, ba) == (ob.best_bid(), ob.best_ask())
        assert bb is None or ba is None or bb < ba         # 订单簿不交叉
print("20000 笔随机订单测试通过，成交笔数 =", len(ob.trades), " 剩余挂单 =", len(ob.orders))

输出：

挂单后  best bid/ask = 101 102
撤单2后 best bid/ask = 101 102
成交记录 (taker, maker, 价, 量): [(6, 3, 101, 4), (6, 1, 100, 2)]
成交后  best bid/ask = 100 102  订单1剩余量 = 3
20000 笔随机订单测试通过，成交笔数 = 9956  剩余挂单 = 1747

白话解释：读这段代码需要知道几样 Python 语法。class IndexedMinHeap: 定义一种自制的数据类型，self 指"这个对象自己"，self.a 就是它内部的数组；以下划线开头的 _up、_down、_swap 按惯例是内部辅助函数。元组 (key, handle) 比大小时按字典序：先比第一项，相等再比第二项，所以关键字 (-价格, 序号) 天然实现"价格优先、同价时间优先"。买方取负价是因为 heapq 和这里的堆都是最小堆：\(-101<-100\)，于是出价 101 的买单排在 100 前面。itertools.count() 是一个每次调用 next() 就加 1 的计数器，用来给订单编到达序号。dict.pop(h) 取出并删除字典里的一项。

几点说明。第一，撤单 2 后最优买价仍是 101，因为订单 3 也在 101；但它现在排在 101 价位的第一位，所以随后的卖单先和订单 3 成交——这正是"时间优先"。第二，成交价取挂单方（maker）价格。第三，随机测试每 1000 笔用暴力扫描核对一次最优价，并检查买卖价不交叉。

工程上要知道：生产级撮合引擎通常按价位组织（价格档数组或平衡树，每档挂一个 FIFO 链表），因为同价位的大量订单只需一次价位查找，撤单可以 \(O(1)\) 从链表摘除；价格离散且范围有限时直接按价格下标寻址（第 08 章的"值作下标"）。堆版本的优点是代码短、正确性容易验证，适合做回测撮合模拟器和理解机制。另一种常见写法是"延迟删除"：撤单时只打标记，等它浮到堆顶再丢弃，代码更短但堆里会积累垃圾。

6.6.3 双堆滚动中位数

滚动中位数是稳健统计的基础工具：对错价 tick、闪崩数据不敏感。双堆法的思路：用最大堆 lo 装较小的一半、最小堆 hi 装较大的一半，保持 lo 比 hi 多 0 或 1 个元素，中位数就在堆顶。窗口滑动时要删除旧元素，这里用延迟删除：先在 delayed 里记账，等它到达堆顶再真正弹出。

白话解释：把窗口里的数从小到大排成一队，中位数就是站在中间的人。双堆法把队伍从中间劈开：左半队用最大堆，队首（堆顶）是左半最大的那个；右半队用最小堆，队首是右半最小的那个。两个队首正好站在分界线两侧，中位数就在这里，不用排整个队。以窗口 \(\{3,1,4,1,5\}\) 为例：左半 \(\{1,1,3\}\)（堆顶 3），右半 \(\{4,5\}\)（堆顶 4），左边多一个，中位数是 3。新数进来时先按"是否不大于左堆顶"决定进哪边，再检查两边人数，差太多就把一边的堆顶挪到另一边（rebalance）。每次挪动只是一次堆操作，所以每步 \(O(\lg w)\)。 代码语法：heapq 只有最小堆，最大堆靠"存负数"模拟，所以 -lo[0] 才是左半的最大值；defaultdict(int) 是查不到时自动返回 0 的字典，方便计数；nonlocal 让内部函数可以修改外层函数的变量 lo_size、hi_size。

import heapq, time
from collections import defaultdict
import numpy as np, pandas as pd

def rolling_median(x, w):
    """双堆滚动中位数：lo 为最大堆（存负数）放较小的一半，hi 为最小堆放较大的一半。
    窗口滑出的元素用“延迟删除”：先记账，等它浮到堆顶时再真正弹出。每步均摊 O(lg w)。"""
    lo, hi = [], []
    delayed = defaultdict(int)
    lo_size = hi_size = 0                     # 有效元素个数（不含待删元素）

    def prune(h, sign):                       # 弹掉堆顶所有待删元素
        while h and delayed[sign * h[0]]:
            delayed[sign * h[0]] -= 1
            heapq.heappop(h)

    def rebalance():                          # 保持 lo_size == hi_size 或 lo_size == hi_size + 1
        nonlocal lo_size, hi_size
        if lo_size > hi_size + 1:
            heapq.heappush(hi, -heapq.heappop(lo)); lo_size -= 1; hi_size += 1; prune(lo, -1)
        elif lo_size < hi_size:
            heapq.heappush(lo, -heapq.heappop(hi)); hi_size -= 1; lo_size += 1; prune(hi, 1)

    out = np.full(len(x), np.nan)
    for t, v in enumerate(x):
        if not lo or v <= -lo[0]:
            heapq.heappush(lo, -v); lo_size += 1
        else:
            heapq.heappush(hi, v); hi_size += 1
        if t >= w:                            # 移出 x[t-w]
            old = x[t - w]; delayed[old] += 1
            if old <= -lo[0]:
                lo_size -= 1
                if old == -lo[0]: prune(lo, -1)
            else:
                hi_size -= 1
                if hi and old == hi[0]: prune(hi, 1)
        rebalance()
        if t >= w - 1:
            out[t] = -lo[0] if w % 2 else (-lo[0] + hi[0]) / 2
    return out

rng = np.random.default_rng(42)
n = 200_000
ret = rng.standard_t(df=3, size=n) * 0.01            # 厚尾日收益
price = np.round(100 * np.exp(np.cumsum(ret)), 2)    # 价格按 0.01 取整，制造大量重复值
for w in (20, 21, 250):
    t0 = time.perf_counter(); m1 = rolling_median(price.tolist(), w); t1 = time.perf_counter()
    m2 = pd.Series(price).rolling(w).median().to_numpy()
    print(f"w={w:>3d}: 与 pandas 最大差异 = {np.nanmax(np.abs(m1 - m2)):.2e}，双堆用时 {t1 - t0:.2f}s")

# 稳健性：一个错价 tick 对滚动均值和滚动中位数的影响
p = price[:300].copy(); p[150] = p[150] * 10        # 人为注入一笔 10 倍的错价
ma = pd.Series(p).rolling(21).mean(); md = pd.Series(rolling_median(p.tolist(), 21))
clean_ma = pd.Series(price[:300]).rolling(21).mean()
clean_md = pd.Series(price[:300]).rolling(21).median()
print("错价对滚动均值的最大扭曲: %.2f，对滚动中位数的最大扭曲: %.2f"
      % ((ma - clean_ma).abs().max(), (md - clean_md).abs().max()))

输出：

w= 20: 与 pandas 最大差异 = 0.00e+00，双堆用时 0.13s
w= 21: 与 pandas 最大差异 = 0.00e+00，双堆用时 0.13s
w=250: 与 pandas 最大差异 = 0.00e+00，双堆用时 0.14s
错价对滚动均值的最大扭曲: 43.18，对滚动中位数的最大扭曲: 0.39

价格按 0.01 取整，制造了大量重复值，双堆结果与 pandas 完全一致（奇偶窗口都测了）。最后一行显示：一笔 10 倍的错价把 21 日滚动均值拉偏了 43 元，而滚动中位数只偏了 0.39 元。这就是为什么做数据清洗、计算"参考价"时更常用中位数。每步代价是摊还 \(O(\lg w)\)；pandas 的 rolling().median() 内部用的是跳表，复杂度同级。需要滚动任意分位数时，可把两个堆的大小比例从 1:1 改成 \(q:(1-q)\)。

6.6.4 流式 Top-K、多路行情归并与事件队列

import heapq
import numpy as np

# ---------- (1) 流式 Top-K：大小为 k 的最小堆，堆顶是“入围门槛” ----------
def top_k_stream(stream, k):
    h = []                                   # (score, code)
    for code, score in stream:
        if len(h) < k:
            heapq.heappush(h, (score, code))
        elif score > h[0][0]:                # 比门槛高才入围，O(lg k)
            heapq.heapreplace(h, (score, code))
    return sorted(h, reverse=True)

rng = np.random.default_rng(7)
n, k = 5000, 10
codes = [f"{600000 + i:06d}" for i in range(n)]
score = rng.standard_normal(n)               # 某因子的截面得分
top = top_k_stream(zip(codes, score), k)
ref = np.argsort(-score)[:k]
assert [c for _, c in top] == [codes[i] for i in ref]
print("Top-5:", [(c, round(float(s), 3)) for s, c in top[:5]])

# ---------- (2) k 路归并多个按时间排序的 tick 流（练习 6.5-9），O(n lg k) ----------
def k_way_merge(streams):
    h = [(s[0][0], j, 0) for j, s in enumerate(streams) if s]   # (时间戳, 流编号, 位置)
    heapq.heapify(h)                                           # BUILD-MIN-HEAP，O(k)
    out = []
    while h:
        ts, j, t = h[0]
        out.append(streams[j][t])
        if t + 1 < len(streams[j]):
            heapq.heapreplace(h, (streams[j][t + 1][0], j, t + 1))  # 取出最小 + 插入后继
        else:
            heapq.heappop(h)
    return out

venues = ["SH", "SZ", "FUT"]
streams = []
for v in venues:
    ts = np.sort(rng.integers(0, 10_000_000, 4)).tolist()     # 微秒时间戳
    streams.append([(t, v, round(100 + rng.normal(), 2)) for t in ts])
merged = k_way_merge(streams)
assert [m[0] for m in merged] == sorted(m[0] for s in streams for m in s)
for m in merged[:6]:
    print(m)

# ---------- (3) 事件驱动回测的核心：以时间为关键字的最小优先队列 ----------
events, seq = [], 0
def schedule(t, kind, payload):
    global seq
    heapq.heappush(events, (t, seq, kind, payload)); seq += 1     # seq 打破同一时刻的平局
for t, v, px in merged[:4]:
    schedule(t, "TICK", (v, px))
log = []
while events:
    t, _, kind, payload = heapq.heappop(events)                    # EXTRACT-MIN
    log.append((t, kind, payload))
    if kind == "TICK" and payload[0] == "SH":
        schedule(t + 350, "FILL", payload)          # 下单后 350 微秒才收到成交回报：产生“未来事件”
assert [e[0] for e in log] == sorted(e[0] for e in log)
for e in log:
    print(e)

输出：

Top-5: [('604657', 4.062), ('603086', 3.193), ('603593', 3.15), ('602675', 3.148), ('601533', 2.988)]
(340454, 'SH', 100.43)
(1686743, 'FUT', 99.21)
(1741444, 'SZ', 100.43)
(2954567, 'FUT', 99.81)
(3309102, 'FUT', 99.74)
(3793777, 'FUT', 99.75)
(340454, 'TICK', ('SH', 100.43))
(340804, 'FILL', ('SH', 100.43))
(1686743, 'TICK', ('FUT', 99.21))
(1741444, 'TICK', ('SZ', 100.43))
(2954567, 'TICK', ('FUT', 99.81))

这段代码包含三个模式：

白话解释：Top-K 为什么用最小堆？想象一个只有 10 个席位的榜单，最该关心的是"最弱的入围者"，因为新来者只要比他强就能把他挤掉。最小堆的堆顶正是这位守门员，比较一次就知道新来者能否入围；能的话 heapreplace 一步完成"踢掉守门员 + 新人入座 + 重新找出新守门员"，代价 \(O(\lg k)\)。5000 只股票选前 10，每只最多花约 \(\lg10\approx3\) 步，远比全部排序省。

1. 流式 Top-K：维护大小为 \(k\) 的最小堆，堆顶是当前入围门槛，新股票得分高于门槛才替换。总代价 \(O(n\lg k)\)，内存 \(O(k)\)，适合实时涨幅榜、成交额榜，或数据太大不能一次装进内存时的选股。若数据已在内存中，第 09 章的 argpartition 方法（\(O(n+k\lg k)\)）更快。
2. \(k\) 路归并：把多个交易所、多个品种各自按时间排好的 tick 流合成一条统一时间线，\(O(n\lg k)\)。标准库 heapq.merge 实现的就是这个算法，而且是惰性的，可以直接处理文件迭代器。
3. 事件队列：以 (时间, 序号) 为关键字的最小优先队列。处理 SH 的 tick 时"下单"，350 微秒后的成交回报作为未来事件插回队列。只要所有事件都从队列按时间顺序取出，策略就不可能看到未来数据——这是从结构上杜绝前视偏差（look-ahead bias）的办法。序号用来打破同一时刻的平局，保证结果可复现。

本章小结

堆是一棵存放在数组里的近似完全二叉树，只维护"父亲不小于（或不大于）孩子"这一弱条件，换来 \(O(1)\) 读最值、\(O(\lg n)\) 插入和删除。MAX-HEAPIFY 沿一条根到叶的路径下沉，代价与高度成正比；BUILD-MAX-HEAP 自底向上调用它，因为大多数结点很矮，总代价只有 \(O(n)\)。堆排序是原地、最坏 \(O(n\lg n)\) 的排序，但不稳定、cache 不友好。堆更重要的身份是优先队列：订单簿的最优价、Top-K 选股、滚动中位数、多路归并、事件驱动回测都建立在它之上。实际系统中要记得维护句柄，才能高效撤单或改键。

练习

基础

1. 数组 \(\langle23,17,14,6,13,10,1,5,7,12\rangle\) 是最大堆吗？（练习 6.1-6） 提示：检查每个结点与父亲。\(A[9]=7\) 的父亲是 \(A[4]=6\)，违反性质，不是。
2. 证明元素互异时，最大堆的最小元素一定在叶子上。一个升序排列的数组是最小堆吗？（练习 6.1-4、6.1-5） 提示：非叶结点至少有一个更小的孩子；升序数组满足 \(A[\lfloor i/2\rfloor]\le A[i]\)，是最小堆。
3. 手工演示 MAX-HEAPIFY\((A,3)\) 作用于 \(\langle27,17,3,16,13,10,1,5,7,12,4,8,9,0\rangle\)。（练习 6.2-1） 答案要点：3 与孩子中较大的 10 交换，再在位置 6 与孩子中较大的 9 交换，停止。
4. 演示 HEAP-EXTRACT-MAX 和 MAX-HEAP-INSERT\((A,10)\) 分别作用于 \(\langle15,13,9,5,12,8,7,4,0,6,2,1\rangle\)。（练习 6.5-1、6.5-2） 提示：前者把 1 搬到根后下沉（1 与 13、12、6 依次交换），结果 \(\langle13,12,9,5,6,8,7,4,0,1,2\rangle\)；后者新叶在下标 13（父亲 8），10 与 8、再与 9 交换后停在下标 3。
5. 用 Python heapq（只有最小堆）实现"同价位按到达顺序成交"的卖方订单队列，说明关键字应该怎样设计。 提示：\((\text{价格},\text{递增序号})\)；买方用 \((-\text{价格},\text{序号})\)。

进阶

6. 证明 \(n\) 元堆中高度为 \(h\) 的结点至多 \(\lceil n/2^{h+1}\rceil\) 个，并据此重做 BUILD-MAX-HEAP 的线性时间分析。（练习 6.3-3） 提示：对 \(h\) 归纳；\(h=0\) 时叶子数为 \(\lceil n/2\rceil\)；去掉所有叶子后剩下的树中原高度为 \(h\) 的结点变成高度 \(h-1\)。
7. 写出 HEAP-DELETE\((A,i)\)，并解释为什么填入最后一个元素后可能需要上浮而不只是下沉。（练习 6.5-8） 提示：最后一个元素可能来自另一棵子树，比位置 \(i\) 的新父亲还大。
8. 用 \(\Theta(n\lg k)\) 时间合并 \(k\) 个有序表；若这些表是 \(k\) 个交易所的逐笔成交文件，内存只够放 \(k\) 条记录，算法还可用吗？（练习 6.5-9） 提示：可用，堆中只放各流当前首元素，读一条补一条。
9. 修改 6.6.3 节的双堆，使它输出滚动 \(q\) 分位数（如 \(q=0.9\)），并与 pandas.Series.rolling(w).quantile(q, interpolation='lower') 对照。 提示：令 lo 的有效大小保持为 \(\lceil qw\rceil\)。
10. （思考题 6-2）分析 \(d\) 叉堆的 INSERT 与 EXTRACT-MAX 代价。在一个每秒大量挂单、撤单，而成交（提取）相对少的模拟器里，\(d\) 取 2 还是 4 更合适？ 提示：插入 \(O(\log_d n)\)、提取 \(O(d\log_d n)\)；插入多时 \(d=4\) 更好，并且 4 个孩子常在同一 cache 行内。

原书推荐习题：6.1-7、6.2-5、6.3-3、6.4-2、6.5-6、6.5-8、6.5-9；思考题 6-1、6-2、6-3。

原书对照