Two Heaps (Median Finder)

The Intuition

The two-heap pattern maintains the median of a dynamic collection in O(log n) per insertion and O(1) per query. The dataset is split into two halves:

Lower half (max-heap, lo). The largest value in this heap is at its root and is always accessible in O(1).
Upper half (min-heap, hi). The smallest value in this heap is at its root.

Two invariants must hold after every operation:

Order: every value in lo is <= every value in hi.
Balance: len(lo) == len(hi) or len(lo) == len(hi) + 1. The lower half is allowed to have at most one extra element.

Given those invariants, the median is lo[0] when sizes differ, or (lo[0] + hi[0]) / 2 when they are equal.

The standard insertion algorithm in three lines:

Push the new value to lo.
Move lo's root to hi (hi.push(lo.pop())). This guarantees the order invariant.
If hi is now larger than lo, move hi's root back to lo. This guarantees the balance invariant.

The pattern generalizes beyond medians whenever a problem requires "smallest k values vs largest k values" or "below threshold vs above threshold" with dynamic balancing. Two-heap solutions show up for IPO (LC 502), where one heap holds projects ordered by capital required and the other holds available projects ordered by profit, and for sliding-window median (LC 480), where lazy deletion via a hash map handles values that have left the window.

When to use

Finding the median in a data stream, maintaining the median as elements are added or removed, sliding window median problems, or any scenario where you need quick access to the middle element of a growing collection.

When NOT to use

Need arbitrary element access (use balanced BST), data is static and you just need one median (sort and pick middle), or need more than just the median (use different statistics structure).

Pattern Recognition

Find the median of a dynamically changing collectionYou need to repeatedly insert values and query the middle elementThe problem involves balancing two groups where one tracks smaller values and the other tracks larger valuesSliding window problems that ask for the median of each window positionOptimization problems where you split elements into two balanced groups

Edge Cases

Only one element has been added, so the median is that element itself
Two elements added, median is their average
All elements are identical, both heaps contain the same value
Integer overflow when summing two heap tops for the average (use long or double)
Empty heaps when querying median before any insertions
Negative numbers mixed with positive numbers

Practice Problems

HardFind Median from Data Stream

HardSliding Window Median

HardIPO

Key Points

•Max-heap holds the smaller half, min-heap holds the larger half
•Balance the heaps so their sizes differ by at most 1
•The median is the top of the max-heap or the average of both tops
•Always add to max-heap first, then rebalance by moving tops between heaps
•Python uses negation to simulate a max-heap with heapq

Code Template

import heapq

class MedianFinder:
    """Two-heap median finder.
    max_heap: smaller half (negated for max-heap behavior).
    min_heap: larger half."""

    def __init__(self):
        self.max_heap = []  # stores negated values
        self.min_heap = []

    def add_num(self, num):
        # Push to max-heap (smaller half)
        heapq.heappush(self.max_heap, -num)

        # Ensure max-heap top <= min-heap top
        if (self.min_heap and
                -self.max_heap[0] > self.min_heap[0]):
            val = -heapq.heappop(self.max_heap)
            heapq.heappush(self.min_heap, val)

        # Balance sizes: max_heap can have at most 1 extra
        if len(self.max_heap) > len(self.min_heap) + 1:
            val = -heapq.heappop(self.max_heap)
            heapq.heappush(self.min_heap, val)
        elif len(self.min_heap) > len(self.max_heap):
            val = heapq.heappop(self.min_heap)
            heapq.heappush(self.max_heap, -val)

    def find_median(self):
        if len(self.max_heap) > len(self.min_heap):
            return -self.max_heap[0]
        return (-self.max_heap[0] + self.min_heap[0]) / 2.0

import "container/heap"

// IntMaxHeap implements a max-heap of ints.
type IntMaxHeap []int
func (h IntMaxHeap) Len() int            { return len(h) }
func (h IntMaxHeap) Less(i, j int) bool  { return h[i] > h[j] }
func (h IntMaxHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
func (h *IntMaxHeap) Push(x interface{}) { *h = append(*h, x.(int)) }
func (h *IntMaxHeap) Pop() interface{} {
    old := *h
    val := old[len(old)-1]
    *h = old[:len(old)-1]
    return val
}

// IntMinHeap implements a min-heap of ints.
type IntMinHeap []int
func (h IntMinHeap) Len() int            { return len(h) }
func (h IntMinHeap) Less(i, j int) bool  { return h[i] < h[j] }
func (h IntMinHeap) Swap(i, j int)       { h[i], h[j] = h[j], h[i] }
func (h *IntMinHeap) Push(x interface{}) { *h = append(*h, x.(int)) }
func (h *IntMinHeap) Pop() interface{} {
    old := *h
    val := old[len(old)-1]
    *h = old[:len(old)-1]
    return val
}

type MedianFinder struct {
    maxHeap *IntMaxHeap // smaller half
    minHeap *IntMinHeap // larger half
}

func NewMedianFinder() MedianFinder {
    lo := &IntMaxHeap{}
    hi := &IntMinHeap{}
    heap.Init(lo)
    heap.Init(hi)
    return MedianFinder{lo, hi}
}

func (mf *MedianFinder) AddNum(num int) {
    heap.Push(mf.maxHeap, num)
    // Ensure ordering: max-heap top <= min-heap top
    if mf.minHeap.Len() > 0 &&
        (*mf.maxHeap)[0] > (*mf.minHeap)[0] {
        val := heap.Pop(mf.maxHeap).(int)
        heap.Push(mf.minHeap, val)
    }
    // Balance sizes
    if mf.maxHeap.Len() > mf.minHeap.Len()+1 {
        val := heap.Pop(mf.maxHeap).(int)
        heap.Push(mf.minHeap, val)
    } else if mf.minHeap.Len() > mf.maxHeap.Len() {
        val := heap.Pop(mf.minHeap).(int)
        heap.Push(mf.maxHeap, val)
    }
}

func (mf *MedianFinder) FindMedian() float64 {
    if mf.maxHeap.Len() > mf.minHeap.Len() {
        return float64((*mf.maxHeap)[0])
    }
    return float64((*mf.maxHeap)[0]+(*mf.minHeap)[0]) / 2.0
}

class MedianFinder {
    // maxHeap: smaller half, minHeap: larger half
    private PriorityQueue<Integer> maxHeap;
    private PriorityQueue<Integer> minHeap;

    public MedianFinder() {
        maxHeap = new PriorityQueue<>(
            Collections.reverseOrder());
        minHeap = new PriorityQueue<>();
    }

    public void addNum(int num) {
        maxHeap.offer(num);
        // Ensure ordering: max-heap top <= min-heap top
        if (!minHeap.isEmpty()
                && maxHeap.peek() > minHeap.peek()) {
            minHeap.offer(maxHeap.poll());
        }
        // Balance sizes
        if (maxHeap.size() > minHeap.size() + 1) {
            minHeap.offer(maxHeap.poll());
        } else if (minHeap.size() > maxHeap.size()) {
            maxHeap.offer(minHeap.poll());
        }
    }

    public double findMedian() {
        if (maxHeap.size() > minHeap.size()) {
            return maxHeap.peek();
        }
        return (maxHeap.peek() + minHeap.peek()) / 2.0;
    }
}

Common Mistakes

Forgetting to negate values when using Python heapq as a max-heap
Not rebalancing heaps after every insertion
Getting the median wrong when both heaps have equal size (must average the two tops)
Using the wrong heap for initial insertion, breaking the ordering invariant

Related Patterns

The Intuition

The two-heap pattern maintains the median of a dynamic collection in O(log n) per insertion and O(1) per query. The dataset is split into two halves:

Lower half (max-heap, lo). The largest value in this heap is at its root and is always accessible in O(1).

Upper half (min-heap, hi). The smallest value in this heap is at its root.

Two invariants must hold after every operation:

Order: every value in lo is <= every value in hi.

Balance: len(lo) == len(hi) or len(lo) == len(hi) + 1. The lower half is allowed to have at most one extra element.

Given those invariants, the median is lo[0] when sizes differ, or (lo[0] + hi[0]) / 2 when they are equal.

The standard insertion algorithm in three lines:

Push the new value to lo.

Move lo's root to hi (hi.push(lo.pop())). This guarantees the order invariant.

If hi is now larger than lo, move hi's root back to lo. This guarantees the balance invariant.

import heapq class MedianFinder: """Two-heap median finder. max_heap: smaller half (negated for max-heap behavior). min_heap: larger half.""" def __init__(self): self.max_heap = [] # stores negated values self.min_heap = [] def add_num(self, num): # Push to max-heap (smaller half) heapq.heappush(self.max_heap, -num) # Ensure max-heap top <= min-heap top if (self.min_heap and -self.max_heap[0] > self.min_heap[0]): val = -heapq.heappop(self.max_heap) heapq.heappush(self.min_heap, val) # Balance sizes: max_heap can have at most 1 extra if len(self.max_heap) > len(self.min_heap) + 1: val = -heapq.heappop(self.max_heap) heapq.heappush(self.min_heap, val) elif len(self.min_heap) > len(self.max_heap): val = heapq.heappop(self.min_heap) heapq.heappush(self.max_heap, -val) def find_median(self): if len(self.max_heap) > len(self.min_heap): return -self.max_heap[0] return (-self.max_heap[0] + self.min_heap[0]) / 2.0