Python heapq：優先度キューとヒープ操作をシンプルに

Q: Pythonのheapqとは？

heapqは通常のPythonリストを使用して最小ヒープを実装する標準ライブラリモジュールです。ヒープ特性を維持しながら、O(log n)時間でアイテムの挿入と取り出しを行う関数を提供します。Pythonで優先度キューを実装する標準的な方法です。

Q: Pythonで最大ヒープを作成するには？

Pythonのheapqは最小ヒープのみをサポートしています。最大ヒープをシミュレートするには、挿入時に値を否定し、取り出し時に再び否定します：挿入にheapq.heappush(heap, -value)、最大値の取得に-heapq.heappop(heap)を使用します。

Q: heapqとsorted()はいつ使い分けるべき？

heapq.nlargestやnsmallestは、Top N要素のみが必要で、Nがデータ全体のサイズに比べて小さい場合に使用します。すべての要素を順序付きで必要とする場合やNが全体のサイズに近い場合はsorted()を使用します。

Q: heapqはスレッドセーフですか？

いいえ。個々のheappushとheappopの呼び出しはアトミックではありません。スレッドセーフな優先度キューには、適切なロックでheapqをラップするqueue.PriorityQueueを使用してください。

Q: heapq操作の時間計算量は？

heappushとheappopはどちらもO(log n)です。heapifyはリストをO(n)でヒープに変換します。nlargest(k, data)とnsmallest(k, data)はO(n log k)で実行されます。

Name: Soren Atelier

Updated on 2026/2/10

優先度に従ってアイテムを処理する必要があります -- ジョブスケジューラのタスク、シミュレーションのイベント、グラフの最短経路など。ソートされたリストが明らかな選択肢に見えますが、ソート済みリストへの挿入はO(n)であり、数百万のアイテムを処理する際にボトルネックになります。要素を追加するたびにリスト全体をソートするのはO(n log n)でさらに悪化します。

コストは急速に蓄積されます。挿入のたびにキューを再ソートするジョブスケジューラは、時間の大部分を管理業務に費やします。各ステップで完全なリストをスキャンして最小距離を見つけるパスファインディングアルゴリズムは、線形時間の操作を二次時間に変えてしまいます。

Pythonのheapqモジュールは、O(log n)の挿入とO(log n)の最小要素抽出を提供するヒープベースの優先度キューを提供します。また、コレクション全体をソートせずに任意のイテラブルからN個の最大または最小要素を見つけるための最適化された関数も含まれています。このガイドでは、heapqのすべての関数を実践的な例とともに説明します。

📚

ヒープとは？

ヒープは、各親ノードがその子ノード以下である二分木です（最小ヒープ）。Pythonのheapqは通常のリストを使用して最小ヒープを実装し、最小の要素は常にインデックス0にあります。

主要な操作とその時間計算量：

操作	関数	時間計算量
要素の追加	`heapq.heappush(heap, item)`	O(log n)
最小を取り出す	`heapq.heappop(heap)`	O(log n)
追加してから取り出す	`heapq.heappushpop(heap, item)`	O(log n)
取り出してから追加	`heapq.heapreplace(heap, item)`	O(log n)
リストからヒープを構築	`heapq.heapify(list)`	O(n)
N個の最小要素	`heapq.nsmallest(n, iterable)`	O(n log k)
N個の最大要素	`heapq.nlargest(n, iterable)`	O(n log k)

基本的なヒープ操作

ヒープの作成と使用

import heapq
 
# Start with an empty heap (just a regular list)
heap = []
 
# Push items
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
heapq.heappush(heap, 2)
 
print(heap)        # [1, 2, 3, 5] (heap order, not sorted)
print(heap[0])     # 1 (smallest element, always at index 0)
 
# Pop smallest item
smallest = heapq.heappop(heap)
print(smallest)    # 1
print(heap)        # [2, 5, 3]

リストをヒープに変換

import heapq
 
data = [5, 3, 8, 1, 2, 7, 4, 6]
heapq.heapify(data)  # In-place, O(n)
print(data)  # [1, 2, 4, 3, 5, 7, 8, 6] (heap order)
 
# Pop elements in sorted order
sorted_data = []
while data:
    sorted_data.append(heapq.heappop(data))
print(sorted_data)  # [1, 2, 3, 4, 5, 6, 7, 8]

heappushpopとheapreplace

これらの複合操作は、個別のpush/pop呼び出しよりも効率的です：

import heapq
 
heap = [1, 3, 5, 7]
heapq.heapify(heap)
 
# heappushpop: push first, then pop smallest
result = heapq.heappushpop(heap, 2)
print(result)  # 1 (pushed 2, popped 1)
print(heap)    # [2, 3, 5, 7]
 
# heapreplace: pop smallest first, then push
result = heapq.heapreplace(heap, 10)
print(result)  # 2 (popped 2, pushed 10)
print(heap)    # [3, 7, 5, 10]

nlargestとnsmallest

これらの関数は、コレクション全体をソートせずにN個の最大または最小の要素を見つけます。

import heapq
 
scores = [85, 92, 78, 95, 88, 72, 91, 83, 97, 69]
 
# Top 3 scores
top_3 = heapq.nlargest(3, scores)
print(top_3)  # [97, 95, 92]
 
# Bottom 3 scores
bottom_3 = heapq.nsmallest(3, scores)
print(bottom_3)  # [69, 72, 78]

キー関数を使用

両方ともカスタム比較用のkeyパラメータを受け入れます：

import heapq
 
students = [
    {'name': 'Alice', 'gpa': 3.9},
    {'name': 'Bob', 'gpa': 3.5},
    {'name': 'Charlie', 'gpa': 3.8},
    {'name': 'Diana', 'gpa': 3.95},
    {'name': 'Eve', 'gpa': 3.2},
]
 
top_students = heapq.nlargest(3, students, key=lambda s: s['gpa'])
for s in top_students:
    print(f"{s['name']}: {s['gpa']}")
# Diana: 3.95
# Alice: 3.9
# Charlie: 3.8

nlargest/nsmallest vs sorted()の使い分け

シナリオ	最適なアプローチ	理由
Nが1	`min()` / `max()`	O(n)、最もシンプル
Nが全体に比べて小さい	`nlargest` / `nsmallest`	O(n log k)、メモリ使用が少ない
Nが全体の長さに近い	`sorted(iterable)[:n]`	O(n log n)だが実際には高速
すべての要素のソートが必要	`sorted()`	完全なソートが適切

最大ヒープの実装

Pythonのheapqは最小ヒープのみを提供します。最大ヒープを作成するには、値を否定します：

import heapq
 
# Max-heap using negation
max_heap = []
for val in [5, 1, 3, 8, 2]:
    heapq.heappush(max_heap, -val)
 
# Pop largest items
while max_heap:
    print(-heapq.heappop(max_heap), end=' ')
# 8 5 3 2 1

カスタムオブジェクトの優先度キュー

直接比較できないオブジェクトには、タプル(priority, counter, item)を使用します：

import heapq
from dataclasses import dataclass
 
@dataclass
class Task:
    name: str
    description: str
 
# Use (priority, tie_breaker, task) tuples
task_queue = []
counter = 0
 
def add_task(priority, task):
    global counter
    heapq.heappush(task_queue, (priority, counter, task))
    counter += 1
 
def get_next_task():
    if task_queue:
        priority, _, task = heapq.heappop(task_queue)
        return task
    return None
 
add_task(3, Task("Low", "Clean up logs"))
add_task(1, Task("Critical", "Fix production bug"))
add_task(2, Task("Medium", "Update docs"))
add_task(1, Task("Critical", "Patch security hole"))
 
while task_queue:
    task = get_next_task()
    print(f"{task.name}: {task.description}")
 
# Critical: Fix production bug
# Critical: Patch security hole
# Medium: Update docs
# Low: Clean up logs

カウンタはタイブレーカーとして機能し、同じ優先度のアイテムが挿入順序（FIFO）で返されるようにします。

実践的なユースケース

ソートされたイテラブルの結合

import heapq
 
list1 = [1, 4, 7, 10]
list2 = [2, 5, 8, 11]
list3 = [3, 6, 9, 12]
 
merged = list(heapq.merge(list1, list2, list3))
print(merged)  # [1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12]

heapq.merge()はメモリ効率が良く、すべてのリストを一度にメモリに読み込みません。

ランニングメディアン

import heapq
 
class RunningMedian:
    """Track the median of a stream of numbers."""
    def __init__(self):
        self.low = []   # max-heap (negated values)
        self.high = []  # min-heap
 
    def add(self, num):
        heapq.heappush(self.low, -num)
        heapq.heappush(self.high, -heapq.heappop(self.low))
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high))
 
    def median(self):
        if len(self.low) > len(self.high):
            return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2
 
rm = RunningMedian()
for num in [2, 1, 5, 7, 2, 0, 5]:
    rm.add(num)
    print(f"Added {num}, median = {rm.median()}")

ダイクストラの最短経路

import heapq
 
def dijkstra(graph, start):
    """Find shortest paths from start to all nodes."""
    distances = {node: float('inf') for node in graph}
    distances[start] = 0
    pq = [(0, start)]
 
    while pq:
        dist, node = heapq.heappop(pq)
        if dist > distances[node]:
            continue
        for neighbor, weight in graph[node]:
            new_dist = dist + weight
            if new_dist < distances[neighbor]:
                distances[neighbor] = new_dist
                heapq.heappush(pq, (new_dist, neighbor))
 
    return distances
 
graph = {
    'A': [('B', 1), ('C', 4)],
    'B': [('C', 2), ('D', 5)],
    'C': [('D', 1)],
    'D': [],
}
 
print(dijkstra(graph, 'A'))
# {'A': 0, 'B': 1, 'C': 3, 'D': 4}

Jupyterでヒープのパフォーマンスを分析する

ヒープベースのアルゴリズムを代替案と比較したり、優先度キューのパフォーマンスをプロファイリングしたりする際、インタラクティブな実験によりプロセスが加速します。RunCell (opens in a new tab)はJupyter用のAIエージェントで、ヒープ操作のベンチマーク、パフォーマンス特性の可視化、データ構造の選択をインタラクティブに最適化するのに役立ちます。

FAQ

Pythonのheapqとは？

heapqは通常のPythonリストを使用して最小ヒープを実装する標準ライブラリモジュールです。ヒープ特性（最小要素が常にインデックス0）を維持しながら、O(log n)時間でアイテムの挿入と取り出しを行う関数を提供します。Pythonで優先度キューを実装する標準的な方法です。

Pythonで最大ヒープを作成するには？

Pythonのheapqは最小ヒープのみをサポートしています。最大ヒープをシミュレートするには、挿入時に値を否定し、取り出し時に再び否定します：挿入にheapq.heappush(heap, -value)、最大値の取得に-heapq.heappop(heap)を使用します。複雑なオブジェクトの場合、タプルで優先度を否定します：(-priority, counter, item)。

heapqとsorted()はいつ使い分けるべき？

heapq.nlargest(n, data)やheapq.nsmallest(n, data)は、Top N要素のみが必要で、Nがデータ全体のサイズに比べて小さい場合に使用します。すべての要素を順序付きで必要とする場合やNが全体のサイズに近い場合はsorted()を使用します。Nが1の場合はmin()/max()を使用します。

heapqはスレッドセーフですか？

いいえ。個々のheappushとheappopの呼び出しはアトミックではありません。複数のスレッドが同じヒープにアクセスする場合、外部同期が必要です。スレッドセーフな優先度キューには、適切なロックでheapqをラップするqueue.PriorityQueueを使用してください。

heapq操作の時間計算量は？

heappushとheappopはどちらもO(log n)です。heapifyはリストをO(n)でヒープに変換します。nlargest(k, data)とnsmallest(k, data)はO(n log k)で実行されます（nはデータサイズ、kは要求された要素数）。

まとめ

Pythonのheapqモジュールは、優先度キューとTop-N選択の標準ツールです。ヒープベースのアプローチはO(log n)のプッシュとポップ操作を提供し、ソート済みリストの維持と比べて劇的に高速です。優先度キューパターンにはheappush/heappopを、Top-N選択にはnlargest/nsmallestを、ソートされたシーケンスの結合にはmergeを、既存のリストの変換にはheapifyを使用してください。

📚