Streaming moving average

Latency dashboards, sensor feeds and training-loss curves all show a moving average. The values arrive one at a time and the stream may never end, so you can’t collect them first.

Write a generator moving_average(stream, window):

• For each value from stream, yield the mean of the last window values seen so far, including this one.
• Before window values have arrived, average whatever you have.
• Be lazy. It must work on an infinite stream.
• Each step should be O(1), not O(window).
• window is at least 1.

>>> list(moving_average([2, 4, 6, 8], 2))
[2.0, 3.0, 5.0, 7.0]

Approach

Keep the last window values in a deque and their sum in a variable. Each new value adds to the sum. Once the window is over-full, the oldest value leaves and is subtracted.

from collections import deque


def moving_average(stream, window):
    buf = deque()
    total = 0
    for x in stream:
        buf.append(x)
        total += x
        if len(buf) > window:
            total -= buf.popleft()
        yield total / len(buf)

deque(maxlen=window) drops the oldest item for you, but it doesn’t tell you what it dropped. You still need that value to subtract, so read buf[0] before appending if you go that way.

Complexity

O(1) time per value and O(window) memory, however long the stream runs.

Why this matters

This is the shape of every streaming aggregate: keep a small summary, update it with each event, and emit a result without looking back over history. Window functions in SQL (AVG(x) OVER (ROWS 99 PRECEDING)) and stream processors like Flink use the same trick. One caveat with floats: adding and subtracting for millions of steps lets rounding error creep in. Long-running systems recompute the sum from the buffer now and then, or use integers or math.fsum.