Heaps & Task Schedulers

The classic, production-proven design for calculating running medians is the Dual-Heap Balancing Architecture (one Max-Heap and one Min-Heap).

1. The Mental Model

Imagine splitting all numbers you’ve seen so far into two equal halves:

• The Lower Half (all numbers $le$ median): We store these in a Max-Heap. Why? Because we only care about the largest number in this half!
• The Upper Half (all numbers $ge$ median): We store these in a Min-Heap. Why? Because we only care about the smallest number in this half!

   Lower Half (Max-Heap)          Upper Half (Min-Heap)
[1, 3, 5, 7, 9  -> (TOP: 9)]  <-->  [(TOP: 11) <- 11, 14, 18, 22]
                                     /
                        The Median is 
                     between 9 and 11!

The median is ALWAYS right at the fingertips: either the top of the Max-Heap, or the average of the two tops!

2. The Two Golden Invariants

To make this work 100% reliably, you must maintain two invariants after every single number is added:

1. Ordering Invariant: Every element in max_heap must be $le$ every element in min_heap. (If max_heap.top() > min_heap.top(), swap them).
2. Size Balance Invariant: The size difference between the two heaps must never exceed 1: 0 <= len(max_heap) - len(min_heap) <= 1.

Production Python 3.12 Implementation

import heapq

class MedianFinder:
    def __init__(self):
        # Python heapq is a min-heap by default.
        # To simulate a max-heap, multiply values by -1.
        self.small = []  # Max-heap (stores inverted values)
        self.large = []  # Min-heap (stores normal values)

    def add_num(self, num: int) -> None:
        # Step 1: Push to max-heap (small half)
        heapq.heappush(self.small, -num)
        
        # Step 2: Ensure ordering invariant (all small <= all large)
        if self.small and self.large and (-self.small[0] > self.large[0]):
            val = -heapq.heappop(self.small)
            heapq.heappush(self.large, val)
            
        # Step 3: Ensure size invariant (len(small) can be at most 1 larger than len(large))
        if len(self.small) > len(self.large) + 1:
            val = -heapq.heappop(self.small)
            heapq.heappush(self.large, val)
        elif len(self.large) > len(self.small):
            val = heapq.heappop(self.large)
            heapq.heappush(self.small, -val)

    def find_median(self) -> float:
        if len(self.small) > len(self.large):
            return float(-self.small[0])
        return (-self.small[0] + self.large[0]) / 2.0

Performance & Production Benchmarks

• add_num() Time: O(log N). Pushing and popping from heaps of size N/2 takes ~15-20 CPU instructions.
• find_median() Time: O(1). Simply peek at heap roots (index 0). Instantaneous!
• Space Complexity: O(N) total memory to store the incoming stream numbers.