heapq --- 堆積佇列 (heap queue) 演算法

原始碼：Lib/heapq.py

這個模組實作了堆積佇列 (heap queue) 演算法，亦被稱為優先佇列 (priority queue) 演算法。

最小堆積 (min-heap) 是一顆二元樹，樹上每個父節點的值都小於或等於其子節點的值。我們將這種情況稱為堆積性質不變 (heap invariant)。

對於最小堆積，此實作使用串列，對於所有存在的被比較元素，k 都滿足 heap[k] <= heap[2*k+1] 和 heap[k] <= heap[2*k+2]。元素是從零開始計數。最小堆積有一個有趣的性質：最小的元素永遠在根節點 heap[0]。

Max-heaps satisfy the reverse invariant: every parent node has a value greater than any of its children. These are implemented as lists for which maxheap[2*k+1] <= maxheap[k] and maxheap[2*k+2] <= maxheap[k] for all k for which the compared elements exist. The root, maxheap[0], contains the largest element; heap.sort(reverse=True) maintains the max-heap invariant.

heapq API 與教科書上的堆積演算法在兩個方面不同：(a) 我們使用從零開始的索引。這使得節點索引和其子節點索引之間的關係變得不那麼明顯，但由於 Python 使用從零開始的索引，所以這樣更適合。(b) 教科書通常專注於最大堆積，因為它們適合原地排序。我們的實作偏向於最小堆積，因為它們更符合 Python 串列。

這兩個特性使得可以將堆積視為一個普通的 Python 串列而不會有意外：heap[0] 是最小的元素，而 heap.sort() 維持堆積性質不變！

Like list.sort(), this implementation uses only the < operator for comparisons, for both min-heaps and max-heaps.

In the API below, and in this documentation, the unqualified term heap generally refers to a min-heap. The API for max-heaps is named using a _max suffix.

要建立一個堆積，使用初始化為 [] 的串列，或者分別使用 heapify() 或 heapify_max() 函式將現有的串列轉換為最小堆積或最大堆積。

提供了以下針對最小堆積的函式：

heapq.heappush(heap, item)

將值 item 推入 heap，並維持最小堆積性質不變。

heapq.heappop(heap)

從 heap 取出並回傳最小的元素，維持最小堆積性質不變。如果堆積為空，會引發 IndexError。若要在不取出的情況下存取最小元素，請使用 heap[0]。

heapq.heappushpop(heap, item)

將 item 放入 heap ，接著從 heap 取出並回傳最小的元素。這個組合函式比呼叫 heappush() 之後呼叫 heappop() 更有效率。

heapq.heapify(x)

在線性時間內將串列 x 原地轉換為最小堆積。

heapq.heapreplace(heap, item)

從 heap 取出並回傳最小的元素，接著將新的 item 放進heap。heap 的大小不會改變。如果 heap 是空的會產生 IndexError 錯誤。

這個一次完成的操作會比呼叫 heappop() 之後呼叫 heappush() 更有效率，並在維護 heap 的大小不變時更為適當，取出/放入的組合函式一定會從 heap 回傳一個元素並用 item 取代他。

函式的回傳值可能會大於被加入的 item 。如果這不是你期望發生的，可以考慮使用 heappushpop() 替代，他會回傳 heap 的最小值和 item 兩個當中比較小的那個，並將大的留在 heap 內。

提供以下針對最大堆積的函式：

heapq.heapify_max(x)

在線性時間內將串列 x 原地轉換為最大堆積。

在 3.14 版被加入.

heapq.heappush_max(heap, item)

將值 item 推入最大堆積 heap，維持最大堆積性質不變。

在 3.14 版被加入.

heapq.heappop_max(heap)

從最大堆積 heap 取出並回傳最大的元素，維持最大堆積性質不變。如果最大堆積為空，會引發 IndexError。若要在不取出的情況下存取最大元素，請使用 maxheap[0]。

在 3.14 版被加入.

heapq.heappushpop_max(heap, item)

將 item 推入 max-heap heap，然後取出並回傳 heap 中最大的元素。這個組合動作比先呼叫 heappush_max() 再單獨呼叫 heappop_max() 更有效率。

在 3.14 版被加入.

heapq.heapreplace_max(heap, item)

從最大堆積 heap 取出並回傳最大的元素，同時推入新的 item。最大堆積的大小不會改變。如果最大堆積為空，會引發 IndexError。

The value returned may be smaller than the item added. Refer to the analogous function heapreplace() for detailed usage notes.

在 3.14 版被加入.

這個模組也提供三個利用 heap 實作的一般用途函式

heapq.merge(*iterables, key=None, reverse=False)

合併多個已排序的輸入並產生單一且已排序的輸出（舉例：合併來自多個 log 檔中有時間戳記的項目）。回傳一個 iterator 包含已經排序的值。

和 sorted(itertools.chain(*iterables)) 類似但回傳值是一個 iterable ，不會一次把所有資料都放進記憶體中，並且假設每一個輸入都已經（由小到大）排序過了。

有兩個選用參數，指定時必須被當作關鍵字參數指定。

key 參數指定了一個 key function 引數，用來從每一個輸入的元素中決定一個比較的依據。預設的值是 None （直接比較元素）。

reverse 是一個布林值，如果設定為 True ，則輸入的元素將以相反的比較順序進行合併。為了達成類似 sorted(itertools.chain(*iterables), reverse=True) 的行為，所有 iterables 必須由大到小排序。

在 3.5 版的變更: 加入選用參數 key 和 reverse 。

heapq.nlargest(n, iterable, key=None)

回傳一個包含資料 iterable 中前 n 大元素的 list 。如果有指定 key 引數，key 會是只有一個引數的函式，用來從每一個在 iterable 中的元素提取一個比較的依據（例如 key=str.lower ）。效果相當於 sorted(iterable, key=key, reverse=True)[:n] 。

heapq.nsmallest(n, iterable, key=None)

回傳一個包含資料 iterable 中前 n 小元素的 list 。如果有指定 key 引數，key 會是只有一個引數的函式，用來從每一個在 iterable 中的元素提取一個比較的依據（例如 key=str.lower ）。效果相當於 sorted(iterable, key=key)[:n] 。

後兩個函式在 n 值比較小時有最好的表現。對於較大的 n 值，只用 sorted() 函式會更有效率。同樣地，當 n==1 時，使用內建函式 min() 和 max() 會有更好的效率。如果需要重複使用這些函式，可以考慮將 iterable 轉成真正的 heap 。

基礎範例

堆積排序 (heapsort) 可以透過將所有的值推入一個 heap，並且從 heap 中一個接一個彈出最小元素來實作：

>>> def heapsort(iterable):
...     h = []
...     for value in iterable:
...         heappush(h, value)
...     return [heappop(h) for i in range(len(h))]
...
>>> heapsort([1, 3, 5, 7, 9, 2, 4, 6, 8, 0])
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

雖然類似 sorted(iterable) ，但跟 sorted() 不同的是，這個實作不是 stable 的排序。

Heap 中的元素可以是 tuple 。這有利於將要比較的值（例如一個 task 的優先度）和主要資料放在一起排序：

>>> h = []
>>> heappush(h, (5, 'write code'))
>>> heappush(h, (7, 'release product'))
>>> heappush(h, (1, 'write spec'))
>>> heappush(h, (3, 'create tests'))
>>> heappop(h)
(1, 'write spec')

Other Applications

Medians are a measure of central tendency for a set of numbers. In distributions skewed by outliers, the median provides a more stable estimate than an average (arithmetic mean). A running median is an online algorithm that updates continuously as new data arrives.

A running median can be efficiently implemented by balancing two heaps, a max-heap for values at or below the midpoint and a min-heap for values above the midpoint. When the two heaps have the same size, the new median is the average of the tops of the two heaps; otherwise, the median is at the top of the larger heap:

def running_median(iterable):
    "Yields the cumulative median of values seen so far."

    lo = []  # max-heap
    hi = []  # min-heap (same size as or one smaller than lo)

    for x in iterable:
        if len(lo) == len(hi):
            heappush_max(lo, heappushpop(hi, x))
            yield lo[0]
        else:
            heappush(hi, heappushpop_max(lo, x))
            yield (lo[0] + hi[0]) / 2

For example:

>>> list(running_median([5.0, 9.0, 4.0, 12.0, 8.0, 9.0]))
[5.0, 7.0, 5.0, 7.0, 8.0, 8.5]

優先佇列實作細節

優先佇列 (priority queue) 是 heap 的常見用途之一，實作優先佇列伴隨著下列挑戰：

  • 排序的穩定性：如何將兩個擁有相同優先次序 (priority) 的 task 按照他們被加入的順序回傳?

  • Tuple的排序在某些情況下會壞掉，例如當 Tuple (priority, task) 的 priorities 相等且 tasks 沒有一個預設的排序時。

  • 當一個 heap 中 task 的 priority 改變時，如何將它移到 heap 正確的位置上?

  • 或者一個還沒被解決的 task 需要被刪除時，要如何從佇列中找到並刪除指定的 task?

一個針對前兩個問題的解法是：儲存一個包含 priority、entry count 和 task 三個元素的 tuple 。兩個 task 有相同 priority 時，entry count 會讓兩個 task 能根據加入的順序排序。因為沒有任何兩個 task 擁有相同的 entry count，所以永遠不會直接使用 task 做比較。

task 無法比較的另一個解決方案是建立一個包裝器類別，該類別忽略 task 項目，只比較優先等級：

from dataclasses import dataclass, field
from typing import Any

@dataclass(order=True)
class PrioritizedItem:
    priority: int
    item: Any=field(compare=False)

剩下的問題可以藉由找到要刪除的 task 並更改它的 priority 或者直接將它移除。尋找一個 task 可以使用一個 dictionary 指向佇列當中的 entry 。

移除 entry 或更改它的 priority 更為困難，因為這會破壞 heap 的性質。所以一個可行的方案是將原本的 entry 做一個標記表示它已經被刪除，並新增一個擁有新的 priority 的 entry：

pq = []                         # 在 heap 中的 entry 串列
entry_finder = {}               # task 對應到 entry 的對映
REMOVED = '<removed-task>'      # 被刪除的 task 的佔位器
counter = itertools.count()     # 唯一的序列計數

def add_task(task, priority=0):
    '新增一個 task 或更新一個已存在 task 的 priority'
    if task in entry_finder:
        remove_task(task)
    count = next(counter)
    entry = [priority, count, task]
    entry_finder[task] = entry
    heappush(pq, entry)

def remove_task(task):
    '將一個已存在的 task 標記為 REMOVED。如果找不到會引發 KeyError。'
    entry = entry_finder.pop(task)
    entry[-1] = REMOVED

def pop_task():
    '移除並回傳最低 priority 的 task。如果 heap 是空的會引發 KeyError。'
    while pq:
        priority, count, task = heappop(pq)
        if task is not REMOVED:
            del entry_finder[task]
            return task
    raise KeyError('從空的優先佇列中 pop 出元素')

原理

Heap 是一個陣列對於所有從0開始的 index k 都存在性質 a[k] <= a[2*k+1] 和 a[k] <= a[2*k+2] 。為了方便比較，不存在的元素被視為無限大。Heap 的一個有趣的性質是：a[0] 永遠是最小的元素。

The strange invariant above is meant to be an efficient memory representation for a tournament. The numbers below are k, not a[k]:

Example (min-heap) binary tree.

在上面的樹當中，每個單元 k 都會位在 2*k+1 與 2*k+2 上方。如同體育賽事常見的錦標賽般，每個單元可視為其下方兩個單元當中的贏家，我們可以透過追溯整棵樹來找到該贏家曾經對戰過的所有對手。然而，在許多電腦應用中，我們不需要追溯贏家的完整對戰歷史。為了能更有效率地使用記憶體，當一個贏家晉級勝出時，我們用下方較低層級的另一個項目來取代它，至此規則變為一個單元以及它下方兩個單元，包含三個不同項目，但是最上方的單元「勝過」下方兩個單元。

如果能確保滿足這個 heap 的不變式，那麼索引 0 顯然是最終的贏家。移除並找到「下一個」贏家最簡單的演算法為：將一個輸家（例如上圖中的單元 30）移動到位置 0，然後從新的位置 0 不斷與下方的位置交換值來向下傳遞，直到滿足不變式為止。這個過程的複雜度顯然是樹的節點數目的對數級別。透過對所有項目疊代，可以得到一個複雜度為 O(n log n) 的排序。

這種排序有個好處，只要插入的項目沒有「贏過」你最後提取、索引為 0 的元素，你就可以在排序進行的同時有效率地插入新項目。這在模擬情境當中特別有用，其中樹能夠保存所有輸入事件，而「贏」意味著最小排程時間。當一個事件排程其它事件的執行時，因這些事件仍在等待進行，所以很容易將它們插入 heap 當中。因此， heap 是一個實現排程器的優秀資料結構（這就是我用以實作 MIDI 編曲器的方法 :-）。

多種用於實作排程器的結構現今已被廣泛研究，heap 對此非常有用，因為它們速度相當快，且速度幾乎不受其他因素影響，最壞情況與平均狀況差異無幾。也有其它整體說來更有效率的方法，然而它們的最壞情況可能會非常糟糕。

Heap 在為儲存於硬碟上的大量資料進行排序也非常有用。你可能已經知道，大量資料排序涉及 "runs" 的產生（也就是預先排序的序列，其大小通常與 CPU 記憶體的大小有關），之後再對這些 run 合併，而這些合併的過程通常相當巧妙 [1]。很重要的一點是，初始排序產生的 run 越長越好。錦標賽是達成這一點的好方法，若你用所有可用記憶體來舉行一場錦標賽，並透過替換與向下交換來處理所有適配目前 run 的值，那麼對於隨機產生的輸入，將可以產生長度兩倍於記憶體大小的 run。對於已模糊排序過的輸入，效果更好。

此外，若你將索引為 0 的項目輸出至磁碟，並取得一個無法適配目前錦標賽的輸入（因為該值「勝過」最後的輸出值），則該輸入值就無法插入至 heap 當中，因此 heap 的大小會減小。釋放出來的記憶體可以巧妙地立即再被運用，逐步建構出第二個 heap，其大小增加的速度會與第一個 heap 減少的速度一致。當第一個 heap 完全消失時，你可以切換至第二個 heap 開啟一個新 run 。這真是個聰明且相當有效率的做法！

總結來說，heap 是值得了解的有用記憶體結構。我在一些應用中使用它們，我認為能有一個 'heap' 模組是很棒的。:-)

註解