The Two Heaps Pattern
Two heaps maintain the lower and upper halves of a data stream, allowing O(1) access to the median and O(log n) insertion. Use when: “find the median of a data stream,” “sliding window median,” or any problem requiring the middle value of a dynamically changing dataset.
Structure: a max-heap (left) holds the smaller half; a min-heap (right) holds the larger half. Invariants: (1) every element in left ≤ every element in right. (2) |len(left) – len(right)| ≤ 1. Median: if sizes are equal → (max(left) + min(right)) / 2; if left is larger → max(left).
Find Median from Data Stream (LC 295)
import heapq
class MedianFinder:
def __init__(self):
self.small = [] # max-heap (negate values)
self.large = [] # min-heap
def addNum(self, num):
# Always push to small first, then balance
heapq.heappush(self.small, -num)
# Ensure all of small self.large[0]):
heapq.heappush(self.large, -heapq.heappop(self.small))
# Balance sizes: small can be at most 1 larger than large
if len(self.small) > len(self.large) + 1:
heapq.heappush(self.large, -heapq.heappop(self.small))
if len(self.large) > len(self.small):
heapq.heappush(self.small, -heapq.heappop(self.large))
def findMedian(self):
if len(self.small) > len(self.large):
return -self.small[0]
return (-self.small[0] + self.large[0]) / 2
Sliding Window Median (LC 480)
Same two-heap approach, but remove elements as they leave the window. Python’s heapq doesn’t support O(log n) deletion, so use a lazy deletion map: mark deleted elements, skip them when they reach the heap top.
from collections import defaultdict
import heapq
def medianSlidingWindow(nums, k):
small, large = [], [] # max-heap (negated), min-heap
to_remove = defaultdict(int)
def add(num):
if not small or num <= -small[0]:
heapq.heappush(small, -num)
else:
heapq.heappush(large, num)
balance()
def remove(num):
to_remove[num] += 1
if num len(large) + 1:
heapq.heappush(large, -heapq.heappop(small))
clean(large, False)
elif len(large) > len(small):
heapq.heappush(small, -heapq.heappop(large))
clean(small, True)
def get_median():
if len(small) > len(large):
return float(-small[0])
return (-small[0] + large[0]) / 2.0
for num in nums[:k]:
add(num)
result = [get_median()]
for i in range(k, len(nums)):
add(nums[i])
remove(nums[i - k])
clean(small, True); clean(large, False)
result.append(get_median())
return result
IPO (LC 502) — Two Heaps for Greedy Selection
Given projects with capital requirements and profits, select at most k projects to maximize total capital. Greedy: always pick the most profitable project you can currently afford. Two heaps: (1) min-heap of (capital, profit) for unavailable projects. (2) max-heap of profits for available projects. At each step: move all projects with capital ≤ current_capital from the min-heap to the max-heap. Pick the max profit from the max-heap.
def findMaximizedCapital(k, w, profits, capital):
available = [] # max-heap of profits (negate)
unavailable = sorted(zip(capital, profits)) # sort by capital
idx = 0
for _ in range(k):
while idx < len(unavailable) and unavailable[idx][0] <= w:
heapq.heappush(available, -unavailable[idx][1])
idx += 1
if not available: break
w += -heapq.heappop(available)
return w
When to Use Two Heaps
- Median of a data stream — O(log n) insert, O(1) median
- Sliding window median — two heaps + lazy deletion
- Problems requiring the “middle” of a sorted set — the k-th largest/smallest of a dynamic stream
- Greedy problems where you need both the minimum of one group and the maximum of another simultaneously (e.g., meeting rooms, task scheduling with priority)
Google coding interviews test heap and priority queue patterns. See common questions for Google interview: heap and priority queue problems.
Meta coding interviews test heap and median problems. Review patterns for Meta interview: median and heap data structure problems.
Amazon coding interviews cover heap and data stream problems. See patterns for Amazon interview: heap and median stream problems.