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
# 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