演算法圖鑑
Heap / Priority Queue · 03 / 03

Two Heaps雙堆積

左最大堆、右最小堆，維持平衡。

用在：資料流中位數、滑動視窗中位數

時間複雜度O(log n)
空間複雜度O(n)
難度困難
前置知識Binary Heap

01為什麼需要它

監控儀表板的 p50 延遲

每秒幾千筆請求延遲進來，儀表板要即時顯示中位數。中位數不像平均可以累加，每次重算要先排序，O(n log n)。

為什麼用它把資料分成「較小的一半」和「較大的一半」，各用一個堆積管理。中位數永遠是兩個堆頂之一或它們的平均，新資料進來只要 O(log n) 調整。

有門檻的排程：IPO 問題

手上有一筆資本，每個專案需要一定資本才能啟動，做完能賺利潤。最多做 K 個，怎麼讓最後資本最多？

為什麼用它一個堆積按「需要的資本」排，負責解鎖目前做得起的專案；另一個按「利潤」排，負責從解鎖的裡面挑最賺的。兩個堆積各管一個維度，這是雙堆積的另一種形狀。

滑動視窗中位數

股價的過去 30 天中位數，每天往前滑一格。除了加入新資料，還要移除舊資料。

為什麼用它同樣兩個堆積，加上「延遲刪除」：被移除的元素先記在雜湊表裡，等它浮到堆頂時再真正丟掉。這是資料流中位數的進階版。

看到這些關鍵字就想到它：中位數、資料流、一半一半、兩個維度各自排序、既要最大又要最小。

02核心概念

一個堆積只能回答「最大是誰」或「最小是誰」，回答不了「中間是誰」。但把資料切成兩半就可以：較小的一半放進最大堆積，它的堆頂是左半的最大值；較大的一半放進最小堆積，它的堆頂是右半的最小值。這兩個堆頂正好是整體最中間的兩個數。

要維持兩條不變量。第一，左半的每個數都不大於右半的每個數，也就是 max(low) ≤ min(high)：新元素和左半堆頂比一下，決定進哪一邊。第二，兩邊大小相差不超過 1：放完後若某邊多太多，就把它的堆頂搬去另一邊。約定左邊可以多一個，那麼奇數筆時中位數就是左邊堆頂，偶數筆時是兩個堆頂的平均。

每筆資料最多一次 push 加一次搬移，O(log n)；查中位數只看堆頂，O(1)。相比每次排序的 O(n log n)，這就是「串流」問題典型的做法：不重算，只維護。

雙堆積不只用在中位數。更一般的形狀是兩個堆積各管一個維度：IPO 問題用一個堆積按資本門檻解鎖、另一個按利潤挑選；會議室問題用一個堆積管開始時間、另一個管結束時間。認出「有兩個排序準則要同時處理」，就想到它。

03演算法步驟

  1. 1準備 low（最大堆積）與 high（最小堆積）。Python 沒有最大堆積，low 存負值。
  2. 2新元素 x：若 low 是空的或 x ≤ max(low)，推入 low；否則推入 high。這一步維持「左半 ≤ 右半」。
  3. 3重新平衡：若 len(low) > len(high) + 1，把 low 的堆頂搬到 high；若 len(high) > len(low)，把 high 的堆頂搬到 low。
  4. 4查中位數：low 較多時回傳 max(low)；一樣多時回傳 (max(low) + min(high)) / 2。
  5. 5需要移除舊元素（滑動視窗）時，用雜湊表記下「待刪除」，等該元素浮到堆頂再真的 pop，並在計算大小時扣掉待刪除的數量。

04互動示範

八筆延遲資料依序進來。留意每一筆做了幾件事：先決定進哪邊，必要時搬一個過去，然後中位數直接從堆頂讀出。

資料流中位數
資料流（延遲 ms）
12045300809521060150
左半（較小）
空
最大堆積 · 左邊第一個是堆頂（左半最大）
中位數
—
右半（較大）
空
最小堆積 · 左邊第一個是堆頂（右半最小）
步驟 0/21兩個堆積：左邊的最大堆積放較小的一半（堆頂是左半最大值），右邊的最小堆積放較大的一半（堆頂是右半最小值）。中位數永遠在兩個堆頂上。

05程式碼

資料流中位數的完整實作，加上 IPO 問題示範「兩個堆積各管一個維度」的另一種用法。

import heapq


class MedianFinder:
    """資料流中位數（LeetCode 295）。
    low：最大堆積，放較小的一半（用取負模擬）
    high：最小堆積，放較大的一半
    不變量：len(low) == len(high) 或 len(low) == len(high) + 1，且 max(low) <= min(high)
    """

    def __init__(self):
        self.low = []    # 存 -x
        self.high = []   # 存 x

    def add_num(self, x):
        if not self.low or x <= -self.low[0]:
            heapq.heappush(self.low, -x)
        else:
            heapq.heappush(self.high, x)
        # 重新平衡：low 最多比 high 多一個
        if len(self.low) > len(self.high) + 1:
            heapq.heappush(self.high, -heapq.heappop(self.low))
        elif len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))

    def find_median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2


# 同一個骨架的另一個應用：IPO（LeetCode 502）
# 兩個堆積分工：一個按「門檻」排（最小堆積），一個按「利潤」排（最大堆積）
def maximize_capital(k, w, profits, capital):
    by_capital = sorted(zip(capital, profits))     # (需要的資本, 利潤)
    available = []                                 # 最大堆積：存 -利潤
    i = 0
    for _ in range(k):
        while i < len(by_capital) and by_capital[i][0] <= w:
            heapq.heappush(available, -by_capital[i][1])   # 資本夠了，解鎖
            i += 1
        if not available:
            break
        w += -heapq.heappop(available)             # 做利潤最高的那個
    return w

06練習題

  • LeetCode 295Find Median from Data StreamHard
  • LeetCode 502IPOHard
  • LeetCode 480Sliding Window Median（延遲刪除）Hard
  • LeetCode 253Meeting Rooms II（一個堆積管結束時間）Medium
  • LeetCode 1825Finding MK AverageHard
上一篇Top-K Problems下一篇—