DSA Moving Average from a Stream

The problem

A moving average is the average of the most recent values in a stream of readings. A window of fixed size holds the values that count. Older values stop counting once they leave the window.

The design contract is a class with a chosen window size. Each new reading goes in, and the current moving average comes out. Division is true division, so every average is a decimal.

size = 3
readings = [10, 20, 30, 40, 50]
expected = [10.0, 15.0, 20.0, 30.0, 40.0]
for reading, average in zip(readings, expected):
    print("reading", reading, "-> average", average)

Read one row at a time. The first average covers one reading, because the window is not full yet. The last covers 30, 40 and 50: the 10 and 20 have left the window.

Intuition: keep everything and slice the tail

The first idea stores every reading in a list. Each average takes the last few readings and means them. The idea is correct, and simple to check.

Two costs grow quietly. The list grows forever, even though only the last window of readings can ever matter. And each average re-reads the whole window, even though only one reading changed.

readings = []
size = 3
for value in (10, 20, 30, 40, 50):
    readings.append(value)
    window = readings[-size:]
    print("stored:", readings, "window:", window, "average:", sum(window) / len(window))

The stored list reaches length five for a window of three. On a stream that never ends, the stored list never stops growing. The window needs a structure that forgets.

Brute force: re-sum the window every time

A leaner straightforward version keeps only the window, in a deque. It still recomputes the sum from scratch on every reading. The additions counter shows what that habit costs.

from collections import deque


class SlowMovingAverage:
    def __init__(self, size):
        self.size = size
        self.window = deque()
        self.additions = 0

    def add(self, value):
        if len(self.window) == self.size:
            self.window.popleft()
        self.window.append(value)
        total = 0
        for item in self.window:
            total += item
            self.additions += 1
        return total / len(self.window)

average = SlowMovingAverage(3)
for reading in (10, 20, 30, 40, 50, 60):
    print(reading, "->", average.add(reading))
print("additions:", average.additions)

Six readings cost fifteen additions with a window of three. The cost per reading grows with the window size, not with the change. Only two values actually changed per step: one left, one joined.

The optimal solution: a window and a running sum

Keep the window in a deque beside a running sum: subtract the leaving value, add the joining value, and divide the sum by the window length. The sum is never rebuilt, so each reading costs at most two updates.

from collections import deque


class MovingAverage:
    def __init__(self, size):
        self.size = size
        self.window = deque()
        self.total = 0
        self.updates = 0

    def add(self, value):
        if len(self.window) == self.size:
            self.total -= self.window.popleft()
            self.updates += 1
        self.window.append(value)
        self.total += value
        self.updates += 1
        return self.total / len(self.window)

average = MovingAverage(3)
for reading in (10, 20, 30, 40, 50):
    print(reading, "->", average.add(reading))
print("sum updates:", average.updates)

Every average matches the script. Five readings cost seven sum updates: one per reading while the window fills, then two per reading after that.

Walkthrough: the window filling, then sliding

Watch the window and the sum together. The first three readings only add. From the fourth reading on, each step removes before it adds.

from collections import deque


class MovingAverage:
    def __init__(self, size):
        self.size = size
        self.window = deque()
        self.total = 0
        self.updates = 0

    def add(self, value):
        if len(self.window) == self.size:
            self.total -= self.window.popleft()
            self.updates += 1
        self.window.append(value)
        self.total += value
        self.updates += 1
        return self.total / len(self.window)

average = MovingAverage(3)
for reading in (10, 20, 30, 40, 50):
    result = average.add(reading)
    print("reading", reading, "window:", list(average.window), "total:", average.total, "average:", result)

The 40 arrives to a full window. The 10 leaves the sum, the 40 joins it, and the total moves from 60 to 90. The division then reads the average straight off the sum.

Complexity

Time is O(1) per reading: one reading causes at most one subtraction and one addition, however wide the window is.

Space is O(size) for the window size: the deque holds at most that many readings, and nothing else grows.

from collections import deque


def slow_updates(size, count):
    window = deque()
    additions = 0
    for value in range(count):
        if len(window) == size:
            window.popleft()
        window.append(value)
        for _ in window:
            additions += 1
    return additions

def fast_updates(size, count):
    window = deque()
    updates = 0
    for value in range(count):
        if len(window) == size:
            window.popleft()
            updates += 1
        window.append(value)
        updates += 1
    return updates

for size in (3, 10, 50):
    print("window", size, "readings 100 -> re-sum additions:", slow_updates(size, 100), "running-sum updates:", fast_updates(size, 100))

At window 50 the re-sum pays 4,775 additions against 197 updates. The running-sum count barely moves when the window widens. That flatness, printed rather than timed, is the O(1) claim made visible.

Edge cases and pitfalls

A window that is not yet full averages what it has. The divisor is the current window length, not the target size. Dividing by the full size early would shrink every early average.

A window of size one echoes each reading. Every new reading evicts the previous one, and the average is the reading itself.

from collections import deque


class MovingAverage:
    def __init__(self, size):
        self.size = size
        self.window = deque()
        self.total = 0
        self.updates = 0

    def add(self, value):
        if len(self.window) == self.size:
            self.total -= self.window.popleft()
            self.updates += 1
        self.window.append(value)
        self.total += value
        self.updates += 1
        return self.total / len(self.window)

average = MovingAverage(3)
print("first reading:", average.add(8))
print("second reading:", average.add(4))
single = MovingAverage(1)
print("size one:", single.add(7), single.add(9))
negatives = MovingAverage(2)
print("with negatives:", negatives.add(-10), negatives.add(10))

Negative readings need no special case. The sum simply carries them, and the pair -10 and 10 averages to 0.0. The running sum works for any readings that can be added at all.