Algorithm Search⌘K

Heap-based selection in Python

Keep a heap of at most k candidates to find the k smallest elements in O(n log k) time.

File
heap_select.py, 59 lines
Level
I0 not checked yet
Toolchain
Python 3.14, pytest 9
Test framework
pytest
Run the tests
python3 -m pytest
heap_select.pyRawGitHub
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/heap-select/. The authors are listed in prov/0YXEH5.jsonl.

from typing import Iterable, Protocol, TypeVar


class Ordered(Protocol):
    def __lt__(self, other: "Ordered", /) -> bool: ...


T = TypeVar("T", bound=Ordered)


def sift_down(h: list[T], i: int, m: int) -> None:
    """Restore the max-heap order of h[0 .. m - 1] below index i.

    Assumes that the subtrees of the children of i are max-heaps. Moves h[i]
    down, swapping it with its larger child, until no child is larger.
    """
    while True:  # @line L11
        c = 2 * i + 1  # @line L12
        if c >= m:  # @line L13
            return
        if c + 1 < m and h[c] < h[c + 1]:  # @line L14
            c += 1
        if not h[i] < h[c]:  # @line L15
            return
        h[i], h[c] = h[c], h[i]  # @line L16
        i = c


def heap_select(a: Iterable[T], k: int) -> list[T]:
    """Return the k smallest items of a, in increasing order.

    Reads a once and keeps a max-heap of the k smallest items seen so far,
    so it works on a stream. It takes O(n log k) time and O(k) extra space.
    If a has fewer than k items, all of them are returned, sorted.
    """
    if k < 0:
        raise ValueError("k must be at least 0")
    it = iter(a)
    h: list[T] = []
    if k == 0:
        return h
    for x in it:  # @line L1
        h.append(x)
        if len(h) == k:
            break
    m = len(h)
    for i in range(m // 2 - 1, -1, -1):  # @line L2
        sift_down(h, i, m)
    for x in it:  # @line L3
        if x < h[0]:  # @line L4
            h[0] = x  # @line L5
            sift_down(h, 0, m)  # @line L6
    for j in range(m - 1, 0, -1):  # @line L7
        h[0], h[j] = h[j], h[0]  # @line L8
        sift_down(h, 0, j)  # @line L9
    return h  # @line L10
test_heap_select.py 40 lines
# SPDX-License-Identifier: MIT
# Part of https://algorithm.tamnd.com/heap-select/. The authors are listed in prov/0YXEH5.jsonl.

import random

import pytest

from heap_select import heap_select, sift_down


def test_example() -> None:
    a = [31, 41, 59, 26, 53, 58, 97, 93, 23, 84]
    assert heap_select(a, 3) == [23, 26, 31]
    assert heap_select(a, 1) == [23]
    assert heap_select(a, 10) == sorted(a)


def test_sift_down() -> None:
    h = [3, 9, 8, 5, 7]
    sift_down(h, 0, 5)
    assert h == [9, 7, 8, 5, 3]


def test_edges() -> None:
    assert heap_select([], 3) == []
    assert heap_select([5, 1], 0) == []
    assert heap_select([5, 1], 5) == [1, 5]
    assert heap_select([4, 4, 4, 4], 2) == [4, 4]
    assert heap_select(iter(range(100, 0, -1)), 4) == [1, 2, 3, 4]
    with pytest.raises(ValueError):
        heap_select([1], -1)


def test_random_against_sorted() -> None:
    rng = random.Random(4471)
    for _ in range(800):
        n = rng.randint(0, 60)
        a = [rng.randint(-15, 15) for _ in range(n)]
        k = rng.randint(0, n + 2)
        assert heap_select(a, k) == sorted(a)[:k]

What the next level needs

  • the tests did not run in this build

Read the full page about Heap-based selection