Lesson 39 — Heap III: Two Heaps, One Median

~25 minutes · Day 48 · two heaps · Python

Tonight's interview problem is Find Median from Data Stream. The trap is that a stream keeps changing. Sorting once is easy; answering the median again and again after new numbers arrive is the real problem. The reusable interview move is to keep the smaller half and larger half separated so the middle never gets lost.

The invariant

A max-heap of the lower half and a min-heap of the upper half, kept balanced, put the median at the tops.

The slow way first

The brute force idea is honest: store every number, sort the full list whenever findMedian() is called, then read the middle. That makes adding cheap, but median queries expensive: each query pays O(n log n) for sorting.

def find_median_slow(nums):
    nums = sorted(nums)              # sort the whole stream snapshot
    mid = len(nums) // 2
    if len(nums) % 2 == 1:
        return nums[mid]
    return (nums[mid - 1] + nums[mid]) / 2

A data stream asks for updates, so repeated sorting wastes work. A heap is built for repeated “give me the best item” updates: Python's heapq module gives a min-heap, where the smallest item is at index 0. To simulate a max-heap for the lower half, store negative numbers. The problem statement and required API live on LeetCode.

The two-heap way

Keep two piles:

• low: the lower half, as a max-heap made from negative values.
• high: the upper half, as Python's normal min-heap.

After every insert, enforce two facts: every value in low is less than or equal to every value in high, and low is either the same size as high or one item bigger. Then the median is either the top of low, or the average of both tops.

The two heaps do not fully sort the stream; they keep the median boundary exposed at the roots.

import heapq
class MedianFinder:
    def __init__(self): self.low, self.high = [], []       # negatives, positives
    def addNum(self, num):
        heapq.heappush(self.low, -num)                     # try lower half first
        heapq.heappush(self.high, -heapq.heappop(self.low))# restore low <= high
        if len(self.high) > len(self.low):
            heapq.heappush(self.low, -heapq.heappop(self.high)) # rebalance
    def findMedian(self):
        if len(self.low) > len(self.high): return -self.low[0]
        return (-self.low[0] + self.high[0]) / 2

Read that as a ritual, not magic: push into low, move the biggest lower-half value into high, then move one back only if high became too large. The tops now describe the middle of the whole stream.

Transfer: same invariant, new pressure

Sliding Window Median asks for the median of every length-k window. Do not solve it tonight. Just predict the shape: if one number enters the window and one number leaves, what structure should still hold the lower half and upper half?

What the trade buys you

The trade is exactly what the stream needs: spend a little work on every insert so each median query is instant.

Practice — answer before you scroll past

heapq.heappush(low, -num)
if len(low) > len(high) + 1:
    heapq.heappush(high, -heapq.heappop(low))

Your move (tonight)

Solve Find Median from Data Stream in your Notion tracker using the cycle — predict → attempt → compare → encode. In the predict step, say the brute force and its complexity out loud. In the attempt step, write the two heaps and rebalance rule without watching anything first.

For the encode step, write the invariant at the top of this page in your own words. The revision gate for heap problems is: “min-heap or max-heap, and what lives in it?”

Stuck or curious? Ask your teacher (the agent) — that's what it's for.