36. Running Median of Eval Latency

Problem

An inference-serving dashboard ingests per-request eval latencies one at a time as they stream in from a live evaluation job, and needs to show the current median latency after every new sample without re-sorting the entire history on each update.

Given a stream of latencies processed left to right, return a list where element i is the median of all latencies seen through index i. This must use two heaps — a max-heap for the lower half of values, a min-heap for the upper half — so each new sample costs O(log n), not an O(n log n) full re-sort after every sample.

Source: src/36_running_median_eval_latency.py

def running_medians(latencies: list[float]) -> list[float]:
    ...

Examples:
>>> running_medians([5, 2, 8])
[5.0, 3.5, 5.0]

>>> running_medians([1])
[1.0]

Step-by-Step Approach

1. Split the running dataset into two halves: a "low" half holding the smaller values and a "high" half holding the larger values, kept balanced so their sizes never differ by more than one.
2. Store the low half as a max-heap (Python's heapq is min-heap only, so negate values on push and pop) and the high half as an ordinary min-heap, so each half's boundary element sits at its heap's root.
3. On each new value: push it into the low heap, then immediately pop the low heap's max and push it into the high heap. This routes the value to the correct side in one pass without a separate comparison branch.
4. Rebalance: if the high heap ends up larger than the low heap, pop its min back into the low heap. This keeps the low heap always the same size as, or one larger than, the high heap.
5. Read the median off the two roots: if the low heap is larger, its root alone is the median (odd count); otherwise average the two roots (even count).
6. Append that median to the output list after every insert, so the final result tracks the median at every prefix of the stream, not just at the end.

The key insight is that "push into low, then bubble its max into high" always routes each new value to the correct side without an explicit if/else, and rebalancing by size (not value) is enough to guarantee the roots straddle the true median.

Reference solution

import heapq


def running_medians(latencies: list[float]) -> list[float]:
    # lo: max-heap (negated) for the smaller half; hi: min-heap for the larger half.
    # Rebalanced after every insert so sizes differ by at most 1; O(log n) per sample.
    lo: list[float] = []
    hi: list[float] = []
    medians: list[float] = []
    for value in latencies:
        heapq.heappush(lo, -value)
        # always push into lo first, then bubble its max over to hi -- keeps hi's
        # min >= lo's max without a separate comparison branch
        heapq.heappush(hi, -heapq.heappop(lo))
        if len(hi) > len(lo):
            heapq.heappush(lo, -heapq.heappop(hi))
        if len(lo) > len(hi):
            medians.append(float(-lo[0]))
        else:
            medians.append((-lo[0] + hi[0]) / 2.0)
    return medians

Key Functions & Tricks

• heapq.heappush(lo, -value) — negation trick to simulate a max-heap with Python's min-heap-only heapq
• Two-heap median maintenance — splits the dataset at its median, each half a heap, O(log n) insert and O(1) median read
• "Push-then-bubble" routing — push into lo unconditionally, then move its max to hi, avoiding a manual comparison to decide which heap a value belongs in
• Size-based rebalancing — keeping len(lo) - len(hi) in {0, 1} is enough to guarantee correctness, without ever comparing cross-heap values directly
• -lo[0] / hi[0] — O(1) peek at each heap's root to read the median

How to Recognize This Pattern