Skip to content
AITroveRead. Build. Understand.
Reading list
Make this comfortable

Two heaps: maintain an exact running median

Last updated: 5 Oct 20269 min read
tutorial
IntermediateBy AITrove Editorial

An append-only median index divides recorded integers into a lower half and an upper half. The lower half is a max-heap, represented here by negated integers in Python's min-heap; the upper half is an ordinary min-heap. Every lower value must be no greater than every upper value, and the lower heap has either the same number of entries or one extra. The top of the lower heap is the median for odd counts. For even counts, the mean of both heap tops is returned as an exact rational number, avoiding a rounded floating-point answer. A call before the first record raises. This API does not remove or revise old observations.

Operational case

Record scan counts 47, 19, 61, 26, and 83. The sorted order is 19, 26, 47, 61, 83, so the current median is 47. Record 52; now the middle two are 47 and 52, and the exact median is 99/2. Both heap sizes differ by at most one after each append. A burst of high values can move an upper-heap root into the lower heap, but it cannot make the lower top exceed the upper top when insertion and rebalancing preserve their partition. A sliding-window median would need an explicit deletion strategy; simply dropping an old count from a side list does not remove its heap entry.

Working Python program

python
"""Append-only exact running median using two heaps."""

import heapq
from fractions import Fraction


class ScanMedian:
    def __init__(self):
        self.lower: list[int] = []
        self.upper: list[int] = []

    def record(self, scan_count: int) -> None:
        if self.lower and scan_count <= -self.lower[0]:
            heapq.heappush(self.lower, -scan_count)
        else:
            heapq.heappush(self.upper, scan_count)
        if len(self.upper) > len(self.lower):
            heapq.heappush(self.lower, -heapq.heappop(self.upper))
        elif len(self.lower) > len(self.upper) + 1:
            heapq.heappush(self.upper, -heapq.heappop(self.lower))

    def median(self) -> Fraction:
        if not self.lower:
            raise ValueError("no scans recorded")
        if len(self.lower) > len(self.upper):
            return Fraction(-self.lower[0])
        return Fraction(-self.lower[0] + self.upper[0], 2)


scan_median = ScanMedian()
for count in (47, 19, 61, 26, 83):
    scan_median.record(count)
print(scan_median.median())
scan_median.record(52)
print(scan_median.median())

Output

Output
47
99/2

Time, space, and tradeoff

Each append makes one heap insertion and at most one transfer, taking O(log N) time for N recorded values. Reading the two roots takes O(1) heap operations; constructing an exact rational has arithmetic cost tied to integer bit length. The heaps retain all N values, using O(N) space. Sorting all values after each append would cost O(N log N) per query; a sorted array would make inserts O(N) due shifting. These heaps are a good fit when the stream only grows. They are not a bounded-memory window and they do not support arbitrary removal or timestamp expiration in this example.

Common Mistakes

  • Do not average two roots when the count is odd.
  • Do not let one heap grow two entries larger than the other.
  • Do not use floating-point division when an exact even-count median is required.
  • Do not describe this append-only index as a sliding-window median.

Connected lessons

Apply the invariant in the scan batch audit project, then check the operations quiz.

Double-ended queues: reconcile min and max heaps adds another queue operation contract.

Sliding medians: expire heap entries by event identity examines a related structure with a different operation boundary.

data structures
range-query-structures
Storage details