Кучи и heapq
Куча
Двоичная куча (min-heap) — почти полное двоичное дерево, в котором каждый узел не больше своих детей. Значит, минимум всегда в корне.
Куча хранится в обычном массиве без указателей: у элемента с индексом i дети — 2i + 1 и 2i + 2, родитель — (i - 1) // 2.
Операции:
- push — положить в конец и просеять вверх (менять с родителем, пока он больше) — O(log n);
- pop — взять корень, переставить последний элемент в корень и просеять вниз (менять с меньшим ребёнком) — O(log n);
- heapify — превратить массив в кучу снизу вверх — O(n).
def sift_up(heap, i):
while i > 0 and heap[(i - 1) // 2] > heap[i]:
parent = (i - 1) // 2
heap[i], heap[parent] = heap[parent], heap[i]
i = parent
def sift_down(heap, i):
n = len(heap)
while True:
smallest, left, right = i, 2 * i + 1, 2 * i + 2
if left < n and heap[left] < heap[smallest]:
smallest = left
if right < n and heap[right] < heap[smallest]:
smallest = right
if smallest == i:
return
heap[i], heap[smallest] = heap[smallest], heap[i]
i = smallest
def push(heap, x):
heap.append(x)
sift_up(heap, len(heap) - 1)
def pop(heap):
heap[0], heap[-1] = heap[-1], heap[0]
smallest = heap.pop()
if heap:
sift_down(heap, 0)
return smallest
h = []
for x in [5, 3, 8, 1, 9, 2]:
push(h, x)
print(f"push {x}: {h}")
print([pop(h) for _ in range(6)])
Пройдите пример 👣 Пошагово и проследите, как элемент «всплывает» к корню.
Модуль heapq
import heapq
nums = [5, 3, 8, 1, 9, 2]
heapq.heapify(nums) # на месте, O(n)
print(nums, nums[0])
heapq.heappush(nums, 0)
print(heapq.heappop(nums), heapq.heappop(nums))
print(heapq.heappushpop(nums, 4)) # push, затем pop — быстрее двух вызовов
print(heapq.nlargest(3, [5, 1, 8, 3, 9]), heapq.nsmallest(2, [5, 1, 8, 3, 9]))
# max-куча: отрицательные значения
max_heap = [-x for x in [5, 1, 8]]
heapq.heapify(max_heap)
print(-heapq.heappop(max_heap))
# приоритетная очередь: кортежи (приоритет, счётчик, данные)
tasks = []
for order, (priority, name) in enumerate([(2, "отчёт"), (1, "пожар"), (3, "почта"), (1, "звонок")]):
heapq.heappush(tasks, (priority, order, name)) # order — чтобы не сравнивать данные и сохранить порядок
while tasks:
print(heapq.heappop(tasks))
Кортежи с несравнимыми данными
Если приоритеты равны, Python сравнит следующий элемент кортежа. Словари и объекты без __lt__ сравнить нельзя → TypeError. Добавляйте уникальный счётчик вторым элементом.
Top-k за O(n log k)
Чтобы найти k наибольших, держите min-кучу размера k: в корне — наименьший из «лучших». Новый элемент больше корня — заменяет его.
import heapq
from collections import Counter
def top_k(nums, k):
heap = []
for x in nums:
if len(heap) < k:
heapq.heappush(heap, x)
elif x > heap[0]:
heapq.heapreplace(heap, x) # pop + push
return sorted(heap, reverse=True)
print(top_k([3, 1, 5, 12, 2, 11, 7], 3))
def top_k_frequent(words, k):
counts = Counter(words)
return heapq.nlargest(k, counts, key=counts.get)
print(top_k_frequent(["a", "b", "a", "c", "b", "a"], 2))
def k_closest(points, k):
return heapq.nsmallest(k, points, key=lambda p: p[0] ** 2 + p[1] ** 2)
print(k_closest([(1, 3), (-2, 2), (5, 8), (0, 1)], 2))
Сравнение подходов для k наибольших из n: сортировка — O(n log n); куча размера k — O(n log k) и O(k) памяти (подходит для потока данных!); quickselect — O(n) в среднем.
Слияние k отсортированных списков
import heapq
def merge_k(lists):
heap = [(lst[0], i, 0) for i, lst in enumerate(lists) if lst]
heapq.heapify(heap)
result = []
while heap:
value, i, j = heapq.heappop(heap) # минимальный среди «голов» списков
result.append(value)
if j + 1 < len(lists[i]):
heapq.heappush(heap, (lists[i][j + 1], i, j + 1))
return result
print(merge_k([[1, 4, 7], [2, 5, 8], [0, 3, 6, 9]]))
print(list(heapq.merge([1, 4, 7], [2, 5, 8]))) # готовая ленивая версия
Применяется во внешней сортировке (слияние отсортированных кусков огромного файла), в объединении логов из нескольких серверов по времени.
Медиана потока: две кучи
import heapq
class MedianFinder:
def __init__(self):
self.low = [] # max-куча меньшей половины (храним -x)
self.high = [] # min-куча большей половины
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
m = MedianFinder()
for x in [5, 15, 1, 3, 8]:
m.add(x)
print(f"добавили {x}: медиана {m.median()}")
Где ещё кучи
- алгоритм Дейкстры (кратчайшие пути) — следующий урок;
- планировщики задач и таймеров (asyncio хранит отложенные вызовы в куче);
- «встречи в переговорных»: минимальное число комнат;
- пирамидальная сортировка (heapsort) — O(n log n), на месте.
Сигналы задачи на кучу
«k наибольших/наименьших/частых/ближайших», «медиана потока», «слить k отсортированных», «следующее по приоритету событие», «минимальное число ресурсов одновременно».
Итог
- Куча — дерево в массиве, минимум в
heap[0]; push/pop — O(log n), heapify — O(n). heapq— только min-куча; max-куча через-x; приоритеты — кортежи с уникальным счётчиком.- Top-k — min-куча размера k, O(n log k); слияние k списков — куча из голов.
- Медиана потока — две кучи.
Практика решено 0 из 5
k-й наибольший элемент. Верните k-й по величине элемент списка (не k-й различный) за O(n log k): держите min-кучу размера k.
Какие тесты проверяют код (3)
assert kth_largest([3, 2, 1, 5, 6, 4], 2) == 5 and kth_largest([3, 2, 3, 1, 2, 4, 5, 5, 6], 4) == 4 assert kth_largest([7], 1) == 7 and kth_largest([1, 1, 1], 2) == 1 import inspect; assert "sort" not in inspect.getsource(kth_largest), "Без сортировки — кучей размера k"
import heapq
def kth_largest(nums, k):
heap = []
for x in nums:
heapq.heappush(heap, x)
if len(heap) > k:
heapq.heappop(heap)
return heap[0]
Слияние k отсортированных списков. Слейте k отсортированных списков в один отсортированный
за O(N log k), где N — общее число элементов. Используйте кучу из кортежей
(значение, номер списка, индекс в списке). Без sorted и без heapq.merge.
Какие тесты проверяют код (3)
assert merge_k([[1, 4, 5], [1, 3, 4], [2, 6]]) == [1, 1, 2, 3, 4, 4, 5, 6] assert merge_k([]) == [] and merge_k([[], []]) == [] and merge_k([[], [1]]) == [1] import inspect, re; assert not re.search(r"sort|heapq\.merge", inspect.getsource(merge_k)), "Без sorted и heapq.merge"
import heapq
def merge_k(lists):
heap = [(lst[0], i, 0) for i, lst in enumerate(lists) if lst]
heapq.heapify(heap)
result = []
while heap:
value, i, j = heapq.heappop(heap)
result.append(value)
if j + 1 < len(lists[i]):
heapq.heappush(heap, (lists[i][j + 1], i, j + 1))
return result
Медиана потока. Реализуйте MedianFinder с методами add(num) — O(log n) и median() — O(1).
Используйте две кучи: max-кучу для меньшей половины (храните отрицательные числа) и min-кучу для большей.
Какие тесты проверяют код (3)
m = MedianFinder()
m.add(1); m.add(2)
assert m.median() == 1.5
m.add(3)
assert m.median() == 2
import random
r = random.Random(5); m = MedianFinder(); seen = []
for _ in range(301):
x = r.randint(-100, 100); m.add(x); seen.append(x)
s = sorted(seen); n = len(s)
expected = s[n // 2] if n % 2 else (s[n // 2 - 1] + s[n // 2]) / 2
assert m.median() == expected
import inspect; assert all("sort" not in inspect.getsource(f) for f in (MedianFinder.add, MedianFinder.median)), "Без сортировки — двумя кучами"
import heapq
class MedianFinder:
def __init__(self):
self.low = [] # max-куча (отрицательные числа)
self.high = [] # min-куча
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
Как получить max-кучу с помощью heapq, который реализует только min-кучу?
Сложность операций двоичной кучи из n элементов
| O(1) | O(log n) | O(n) | |
|---|---|---|---|
посмотреть минимум heap[0] |
|||
heappush |
|||
heappop |
|||
heapify списка |
|||
| найти произвольный элемент |
Войдите, чтобы отмечать пройденные уроки.