deque: rolling median and the maxlen narrowing

Compute a rolling median over a stream in one pass, with a fixed-size buffer.

Implement three functions:

def rolling_median(values: Iterable[float], window: int) -> list[float]: ...
def headroom(buffer: deque[float]) -> int: ...
def solve(values: list[float], window: int) -> dict[str, object]: ...

• rolling_median yields one median per full window, so a stream of n values with window w produces max(0, n - w + 1) results. Raise ValueError when window < 1.
• The buffer must never hold more than window values. Take values: Iterable[float] — a real stream is not indexable, and the moment you write values[i - window : i] you have both copied the window on every step and locked yourself out of generators, file handles and DB cursors.
• headroom returns maxlen - len(buffer), or -1 when the deque is unbounded.
• solve returns {"error", "medians", "tail", "headroom"}:
  • "error" is "ValueError" when window < 1, otherwise "".
  • "medians" is the rolling medians ([] on error).
  • "tail" is the last window values as a list — build it with deque(values, maxlen=window), which consumes the whole stream in constant memory.
  • "headroom" is headroom(tail_deque), or -1 on error.

The production consequence. list.insert(0, x) and list.pop(0) are $O(n)$: every element behind the insertion point moves. A queue or a last-N-events buffer built that way is quadratic. Measured, 100,000 list.insert(0, i) calls took 1157 ms against 1.04 ms for deque.appendleft — about 1,100x. It passes every test at fixture size.

The typing detail. deque.maxlen is int | None, because an unbounded deque has no maximum. Arithmetic on it needs a narrowing branch, and writing that branch forces you to answer what an unbounded buffer’s headroom means — a question the untyped version let you skip.

The median of an even-sized window is the mean of the two middle values; statistics.median already does this.