Two Heaps Pattern | DSA Roadmap

Why this matters more than it seems

A single heap only ever gives you the extreme (min or max) of a set. But a family of interview problems asks for something in the middle — a running median, or "the boundary between what's affordable and what isn't." The two-heaps pattern solves this by gluing a max-heap and a min-heap together at a seam, so the seam itself tracks whatever order statistic you care about. Once you recognize the shape, a cluster of problems that look unrelated on the surface (running median, sliding-window median, order-book matching, capital-unlocking optimization) collapse into the same six lines of logic.

The core idea: split the data at the median

To track a running median efficiently, maintain two heaps that partition the data in half:

lower — a max-heap holding the smaller half of the numbers seen so far. Its root is the largest of the small half.
upper — a min-heap holding the larger half. Its root is the smallest of the large half.

Maintain the invariant:

Every element in lower is ≤ every element in upper.
The sizes differ by at most 1: len(lower) == len(upper) or len(lower) == len(upper) + 1.

Given that invariant, the median falls right out:

If len(lower) > len(upper): median = lower's root.
If sizes are equal: median = average of lower's root and upper's root.

import heapq

class RunningMedian:
    def __init__(self):
        self.lower = []   # max-heap (store negated values)
        self.upper = []   # min-heap

    def add(self, num: int) -> None:
        # Always push into lower first, then let the smallest of lower
        # migrate into upper — this keeps the "every lower <= every upper"
        # invariant correct even when num belongs in upper.
        heapq.heappush(self.lower, -num)
        heapq.heappush(self.upper, -heapq.heappop(self.lower))

        # Rebalance: lower is allowed at most one more element than upper.
        if len(self.upper) > len(self.lower):
            heapq.heappush(self.lower, -heapq.heappop(self.upper))

    def median(self) -> float:
        if len(self.lower) > len(self.upper):
            return -self.lower[0]
        return (-self.lower[0] + self.upper[0]) / 2.0

Notice the "push into lower, then immediately shuffle its top into upper" trick — it's a clean way to guarantee the cross-heap ordering invariant without a branch that separately decides which heap a new number belongs in. It costs one extra O(log n) operation per insert but eliminates a common source of bugs.

Complexity

Insert (add): O(log n) — up to three heap pushes/pops, each O(log n).
Query (median): O(1) — just peek both roots.

Compare this to the alternatives:

Approach	Insert	Median query
Unsorted array, sort on demand	O(1)	O(n log n)
Sorted array (insert in place)	O(n) shift	O(1)
Balanced BST / order-statistics tree	O(log n)	O(log n)
Two heaps	O(log n)	O(1)

Two heaps gives you the best insert cost and the best query cost simultaneously — the trade-off other structures make (fast insert or fast query, not both) is exactly what this pattern avoids.

Generalizing beyond the median: two heaps as "two competing priorities"

The same shape — one heap ordered by criterion A, another ordered by criterion B, with elements flowing between them as some threshold changes — solves more than medians. A common variant: you have a pool of candidates gated by a capacity condition (e.g., "can only be considered once you can afford it"), and among the unlocked candidates you always want the best by some other metric (e.g., highest profit). The pattern is:

A min-heap ordered by the gating criterion (e.g., cost), holding everything not yet unlocked.
A max-heap ordered by the selection criterion (e.g., profit), holding everything currently unlocked.
Each round: drain everything from heap 1 whose gate is now satisfied into heap 2, then pop the best from heap 2.

import heapq

def unlock_and_select(locked: list[tuple[int, int]], rounds: int, budget: int) -> int:
    """locked: list of (gate_value, reward_value). Returns the max total
    reward obtainable by unlocking items whose gate <= current budget,
    one per round, always taking the best available reward."""
    locked.sort()  # sort ascending by gate value
    available = []  # max-heap of unlocked rewards (negated)
    i = 0
    for _ in range(rounds):
        while i < len(locked) and locked[i][0] <= budget:
            heapq.heappush(available, -locked[i][1])
            i += 1
        if not available:
            break
        budget += -heapq.heappop(available)
    return budget

This "min-heap of gates, max-heap of unlocked options" shape also underlies order-matching simulations (a max-heap of buy prices, a min-heap of sell prices) and appears again, in a different guise, when you reach Meeting Rooms–style interval scheduling in the Intervals topic.

Pitfalls and interview gotchas

Getting the heap types backwards. It's lower = max-heap, upper = min-heap — not the reverse. A quick sanity check: the root of lower and the root of upper need to sit right next to each other at the median seam, so lower needs fast access to its largest element and upper needs fast access to its smallest.
Off-by-one on the balance invariant. Decide up front whether lower is allowed to have one more element than upper (as above) or vice versa, and use that same convention consistently in both add and the median formula. Mixing conventions mid-solution is the single most common bug here.
Integer vs. float division. When the sizes are equal, the median is the average of two integers — in a statically typed language this often means an accidental integer division (e.g., (a + b) / 2 truncating in Java/C++). Use floating-point division explicitly.
Overflow when averaging. (a + b) / 2 can overflow in a fixed-width integer type if both values are near the type's max; prefer a + (b - a) / 2.0 or a wider type if the input range is large.
Python's heapq negation with tuples. If you store (-value, tiebreaker) in a max-heap, remember you're comparing the whole tuple — make sure the tiebreaker's natural ordering still does what you want once the primary field is negated.
Confusing this with a sliding window. A pure "running median over everything seen so far" never removes elements. If a variant asks for a median over only the last k elements, you additionally need a way to remove an arbitrary (not just extreme) element from a heap — since heaps don't support efficient arbitrary deletion, the standard trick is lazy deletion: track which values are "stale" in a hash map, and only actually pop them once they bubble up to a heap's root.

How to state this in an interview

"I'll maintain two heaps split at the median: a max-heap for the lower half, a min-heap for the upper half, kept within one element of each other in size. Each insert is O(log n) since it's a constant number of heap operations, and reading the median back out is O(1) since it's just the root(s) — that beats re-sorting on every query, which would be O(n log n) per read."

Further Resources (Optional)

Tech Interview Handbook — Heap cheatsheetArticle10m
GeeksforGeeks — Median of a Stream of Running IntegersArticle15m
GeeksforGeeks — Median of Sliding Window in an ArrayArticle20m
Wikipedia — Selection algorithmReference15m
NeetCode — Find Median from Data Stream (video walkthrough)Video25m
Abstract Algorithms — Two Heaps Pattern deep dive and interview pitfallsArticle15m
Wikipedia — Min-max heap (a single-heap alternative to the two-heap trick)Reference15m

Practice Problems

Interview relevancy:!!!Critical — must-have!!Important — highly recommended!Good to have~Niche — rarely asked

Find Median from Data StreamHard!!!3/535m
IPOHard!4/540m

Optional Practice (Extra Reps)

For once you've cleared the main set above and want more reps on this pattern. These don't count toward the roadmap's progress stats — solve them purely for your own benefit.

Number of Orders in the BacklogMedium~3/530m
Sliding Window MedianHard~5/550m
Sliding Window CostCSES~5/545m
Minimum Number of Refueling StopsHard!4/535m
Minimum Cost to Hire K WorkersHard!4/535m