TL;DR: A deque plus a running total. On each push, append and add to the total; evict from the left (and subtract) once the window exceeds size, then return total over current length. O(1) per call.
How to approach it This is a stateful API design warm-up, not an algorithms question. It's checking whether you can shape a small class cleanly, because the rate-limiter and event-stream questions that follow build on the same skeleton. Clarify: window by count (last N values) or by time (last N seconds)? Start with count, but structure the code so the time-window variant is a small change, and say you're doing that.
A strong answer Keep a running sum; never re-sum the window:
from collections import deque
class MovingAverage:
def __init__(self, size: int):
if size <= 0:
raise ValueError("size must be positive")
self.size = size
self.window: deque[float] = deque()
self.total = 0.0
def next(self, value: float) -> float:
self.window.append(value)
self.total += value
if len(self.window) > self.size:
self.total -= self.window.popleft()
return self.total / len(self.window)
O(1) per call, O(size) memory. Narrate the choice: deque gives O(1) pops from the left (a list's pop(0) is O(n)); the running total avoids O(size) re-summing, trivial here, but the same discipline matters when the window is a million events.
ma = MovingAverage(3)
assert ma.next(1) == 1.0
assert ma.next(10) == 5.5 # partial window: average of what exists
assert ma.next(3) == 14 / 3
assert ma.next(5) == 6.0 # 1 evicted: (10+3+5)/3
Note the deliberate decision: before the window fills, return the average of what exists (and you asked whether that's the desired semantics rather than assuming).
What interviewers probe next (1) Time-based window, the real follow-up: store (timestamp, value) pairs and evict while now - window[0][0] >= window_seconds. The class shape survives; only the eviction predicate changes, and that refactorability is what's being graded:
def next_t(self, value: float, now: float) -> float:
self.window.append((now, value)); self.total += value
while self.window and now - self.window[0][0] >= self.window_seconds:
_, old = self.window.popleft(); self.total -= old
return self.total / len(self.window)
Passing now as a parameter instead of calling time.time() inside is a testability point; mention it. (2) Floating-point drift? The running total accumulates error over millions of updates, and the effect is small but real and measurable: we ran 10 million pushes of 0.1 through a 1,000-element window and the running total ended at 99.9999999999986 against a re-summed 100.0, a drift of about 1.4e-12. Harmless for a latency dashboard, disqualifying for money, where a ledger that reconciles to within-epsilon is a ledger that does not reconcile. So: integers in cents or Decimal for anything financial, or a periodic re-sum (every N pushes, recompute total = sum(window)) to pin the error, and knowing the drift is bounded-but-nonzero is exactly the level of numeric literacy the follow-up is probing for. (3) Max/min over the window? Different structure entirely: monotonic deque; recognizing that the running-sum trick doesn't transfer is the signal. (4) Thread safety? One lock around next; note GIL-protected single ops aren't enough because append/evict/total must be atomic together.
Common mistakes list.pop(0) without flagging the O(n) cost. Re-summing the window each call. Hard-coding time.time() inside the method, making tests flaky. Not asking the count-vs-time question, then rewriting everything when the interviewer "extends" the problem they always intended to extend. And skipping input validation (size <= 0): one line, disproportionate signal on practical screens.