30. Running Median of Response Latency

Problem

An operational dashboard for a real-time voice pipeline streams in the end-to-end response latency of every request as it completes, and needs the running median at any moment without ever re-sorting the full history collected so far.

Maintain the running median using two heaps: a max-heap for the lower half of latencies seen so far, and a min-heap for the upper half, so the median is O(1) to read after each O(log n) insert.

Source: src/30_latency_median_tracker.py

class LatencyMedianTracker:
    def add(self, latency: float) -> None: ...
    def median(self) -> float: ...  # raises ValueError if empty

>>> t = LatencyMedianTracker()
>>> t.add(120.0); t.median()
120.0
>>> t.add(80.0); t.median()
100.0
>>> t.add(200.0); t.median()
120.0

Step-by-Step Approach

1. Maintain two heaps: _lo, a max-heap holding the smaller half of all values seen so far (implemented as a min-heap of negated values, since Python's heapq is min-heap only), and _hi, a min-heap holding the larger half.
2. Keep the invariant that every value in _lo is ≤ every value in _hi, and the two heaps' sizes never differ by more than 1.
3. On add(latency), always push into _lo first, regardless of the new value's actual magnitude.
4. Immediately rebalance: pop _lo's max and push it into _hi. This single push-then-promote sequence is what guarantees the cross-heap ordering invariant holds after every insert, no matter which heap the new value "should" end up in.
5. If that left _hi strictly larger than _lo, pop _hi's min back into _lo — this restores the size-balance invariant (sizes differ by at most 1).
6. On median(): if _lo is strictly larger, the median is its top (the extra element from an odd-sized stream). Otherwise the streams are balanced and the median is the average of both heaps' tops.

The key insight is the "push into _lo, then always promote its max into _hi, then rebalance sizes" sequence — it looks like it does unnecessary work, but it's exactly what keeps both the value-ordering invariant and the size-balance invariant correct after every single insert, without ever needing to compare the new value against the current median explicitly.

Reference solution

import heapq


class LatencyMedianTracker:
    # _lo is a max-heap (values stored negated) holding the smaller half;
    # _hi is a min-heap holding the larger half. Kept within size 1 of each
    # other so the median is always at the top of one or both heaps.
    def __init__(self):
        self._lo: list[float] = []
        self._hi: list[float] = []

    def add(self, latency: float) -> None:
        # always push into _lo first, then rebalance by promoting its max
        # into _hi -- guarantees every value in _lo <= every value in _hi
        heapq.heappush(self._lo, -latency)
        heapq.heappush(self._hi, -heapq.heappop(self._lo))
        if len(self._hi) > len(self._lo):
            heapq.heappush(self._lo, -heapq.heappop(self._hi))

    def median(self) -> float:
        if not self._lo and not self._hi:
            raise ValueError("no latencies added")
        if len(self._lo) > len(self._hi):
            return float(-self._lo[0])
        return (-self._lo[0] + self._hi[0]) / 2.0

Key Functions & Tricks

• Negated min-heap as a max-heap — Python's heapq only implements a min-heap, so negating values on push/pop simulates a max-heap.
• heapq.heappush / heapq.heappop — O(log n) per call, dominating add()'s cost.
• Push-then-promote-then-rebalance sequence — the three-step dance in add() that maintains both invariants unconditionally, regardless of where the new value belongs.
• Size-balance invariant (differ by at most 1) — what makes median() an O(1) lookup instead of requiring a scan.
• len(self._lo) > len(self._hi) branch in median() — handles the odd-total-count case where one heap has the extra element.

How to Recognize This Pattern