16. Running Median of Latency

Problem

Fin logs the end-to-end response latency for every reply it sends to a customer. An operational dashboard wants the running median of these latencies as new measurements stream in continuously — it can't afford to re-sort the entire history on every single update, since the stream never stops.

Design a class that supports two operations: adding a new latency measurement, and querying the current median of everything added so far. The median must be queryable cheaply after every insert, even as the number of measurements grows without bound.

Source: src/16_median_latency_tracker.py

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

>>> t = MedianLatencyTracker()
>>> t.add(5)
>>> t.median()
5.0
>>> t.add(15)
>>> t.median()
10.0
>>> t.add(1)
>>> t.median()
5.0

Step-by-Step Approach

1. Keep two heaps: lo, a max-heap holding the smaller half of the values seen so far, and hi, a min-heap holding the larger half.
2. Python's heapq only gives you a min-heap, so simulate the max-heap by pushing negated values into lo.
3. On add(latency), always push into lo first, then move its new top over to hi. This routes every value through the "smaller half" heap once, which keeps the two heaps correctly partitioned (everything in lo ≤ everything in hi) without a separate comparison step.
4. Rebalance sizes: if hi ever grows larger than lo, pop its top back into lo. This keeps len(lo) equal to len(hi) or exactly one greater, never the other way around.
5. On median(): if lo has one extra element, the median is its top. Otherwise the two heaps are the same size and the median is the average of both tops.
6. Raise ValueError if median() is called before any value has been added.

The key insight is that a heap's top is O(1) to read, and the only work a stream of inserts needs to do is keep the "boundary" between the lower and upper half balanced — it never needs to look at, or re-sort, the rest of either half. That's what turns an O(n log n) full-resort-per-query into O(log n) per insert and O(1) per query.

Reference solution

import heapq


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

    def add(self, latency: float) -> None:
        # negate: _lo is a max-heap via heapq's min-heap
        heapq.heappush(self._lo, -latency)
        # route through _lo to keep lo <= hi partitioned
        heapq.heappush(self._hi, -heapq.heappop(self._lo))
        # rebalance: keep len(_lo) == len(_hi) or one more
        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

• import heapq — free functions over a plain list, no heap object
• heapq.heappush(heap, item) — insert and re-sift, O(log n)
• heapq.heappop(heap) — remove/return smallest, re-sift, O(log n)
• Negation trick — simulate a max-heap on top of heapq's min-heap by pushing -latency
• Route-through-lo balancing — push into _lo then move its new top to _hi, avoiding an explicit comparison
• Size-rebalancing step — if len(self._hi) > len(self._lo) keeps median readable at the top(s) in O(1)
• list[float] — PEP 585 generic-alias type hint, no runtime effect

How to Recognize This Pattern