Кучи и heapq

⏱ 40 мин · 🧩 заданий: 5

Куча

Двоичная куча (min-heap) — почти полное двоичное дерево, в котором каждый узел не больше своих детей. Значит, минимум всегда в корне.

Куча хранится в обычном массиве без указателей: у элемента с индексом i дети — 2i + 1 и 2i + 2, родитель — (i - 1) // 2.

массив: [1, 3, 2, 7, 4, 5, 8] 1 (i=0) / \ 3 (1) 2 (2) / \ / \ 7(3) 4(4) 5(5) 8(6) каждый родитель ≤ детей; минимум — heap[0]

Операции:

  • 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

Вы не вошли — ответы проверяются, но прогресс не сохранится. Войти или зарегистрироваться.
Задание 1 Программа на Python

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"
Задание 2 Программа на Python

Слияние 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"
Задание 3 Программа на Python

Медиана потока. Реализуйте 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)), "Без сортировки — двумя кучами"
Задание 4 Один вариант ответа

Как получить max-кучу с помощью heapq, который реализует только min-кучу?

Задание 5 Заполнить таблицу

Сложность операций двоичной кучи из n элементов

O(1)O(log n)O(n)
посмотреть минимум heap[0]
heappush
heappop
heapify списка
найти произвольный элемент
Отметьте ячейку — «+», пустая — «−».

Войдите, чтобы отмечать пройденные уроки.