Streaming median¶
A sorted array exposes the median but makes arbitrary insertion linear. Two heaps preserve only the order information that the median requires.
Ownership model¶
- A max-heap owns the lower half and exposes its largest value.
- A min-heap owns the upper half and exposes its smallest value.
Invariants:
- every value in the lower heap is at most every value in the upper heap;
- the lower heap has the same size as the upper heap or one extra element.
Add operation¶
The tested template routes every new value through the lower heap, moves the largest lower value into the upper heap, then restores the size rule. This single sequence enforces both invariants.
flowchart LR
accTitle: Two heaps preserve the streaming median invariants
accDescr: Insert a new value into the lower max-heap, move its largest value to the upper min-heap to restore order, then move the upper minimum back only when the lower heap is too small.
subgraph O["1 · Restore order"]
direction TB
A["New value"] --> B["Push into lower max-heap"]
B --> C["Move lower maximum<br/>to upper min-heap"]
end
subgraph S["2 · Restore size"]
direction TB
D{"Is lower smaller<br/>than upper?"} -->|"yes"| E["Move upper minimum<br/>back to lower"]
D -->|"no"| F["Keep the heaps"]
E --> G["Balanced"]
F --> G
end
C --> D
G --> H["Invariant holds<br/>median is exposed at heap roots"]Order is repaired before size. Once all lower values are no greater than all upper values, at most one root move is needed to restore the size rule.
Tested templates¶
class MedianFinder:
"""Maintain the median of an online integer stream."""
def __init__(self) -> None:
self.lower: list[int] = [] # negated max-heap
self.upper: list[int] = [] # min-heap
def add(self, value: int) -> None:
# Route through lower so every lower value stays <= every upper value.
heappush(self.lower, -value)
heappush(self.upper, -heappop(self.lower))
if len(self.upper) > len(self.lower):
# Keep lower the same size as upper or one element larger.
heappush(self.lower, -heappop(self.upper))
def median(self) -> float:
if not self.lower:
raise ValueError("median is undefined for an empty stream")
if len(self.lower) > len(self.upper):
return float(-self.lower[0])
return (-self.lower[0] + self.upper[0]) / 2.0
class MedianFinder {
public:
void add(int value) {
// Route through lower so every lower value stays <= every upper value.
lower_.push(value);
upper_.push(lower_.top());
lower_.pop();
if (upper_.size() > lower_.size()) {
// Keep lower the same size as upper or one element larger.
lower_.push(upper_.top());
upper_.pop();
}
}
double median() const {
if (lower_.empty()) {
throw std::logic_error("median is undefined for an empty stream");
}
if (lower_.size() > upper_.size()) {
return static_cast<double>(lower_.top());
}
return (static_cast<double>(lower_.top()) +
static_cast<double>(upper_.top())) /
2.0;
}
private:
std::priority_queue<int> lower_;
std::priority_queue<int, std::vector<int>, std::greater<int>> upper_;
};
class MedianFinder {
public:
void add(int value) {
// Route through lower so every lower value stays <= every upper value.
lower_.push(value);
upper_.push(lower_.top());
lower_.pop();
if (upper_.size() > lower_.size()) {
// Keep lower the same size as upper or one element larger.
lower_.push(upper_.top());
upper_.pop();
}
}
double median() const {
if (lower_.empty()) {
throw std::logic_error("median is undefined for an empty stream");
}
if (lower_.size() > upper_.size()) {
return static_cast<double>(lower_.top());
}
return (static_cast<double>(lower_.top()) +
static_cast<double>(upper_.top())) /
2.0;
}
private:
std::priority_queue<int> lower_;
std::priority_queue<int, std::vector<int>, std::greater<int>> upper_;
};
Complexity and traps¶
- Add:
O(log n). - Median:
O(1). - Empty-stream median must have explicit behavior.
- Duplicate values do not need special handling.
- In C++, convert heap tops before adding them so the average cannot overflow in integer arithmetic.