Moving Average from Data Stream

How it works

A circular queue (ring buffer) avoids the overhead of shifting elements when removing from the front. We use a fixed-size array where the head pointer wraps around when it reaches the end. The element being overwritten is the one leaving the window, so we subtract it from the running sum before adding the new value. This achieves constant time per operation with minimal memory overhead.

Steps

1. Initialize a fixed-size array of length size, a head pointer, a running sum, and a count of elements seen.
2. When next(val) is called:
   • Increment the count.
   • Calculate the tail position as (head + 1) % size, which points to the oldest element that will be replaced.
   • Subtract the value at the tail position from the running sum and add the new value.
   • Move the head pointer forward: head = (head + 1) % size.
   • Store the new value at the head position.
3. Return the running sum divided by the current window size (minimum of count and size).

Complexity

• Time complexity: O(M)
  • Time complexity per next() call is O(1). Therefore, the total time complexity for M calls is O(M)
• Space complexity: O(N)

Where N is the size of the moving window and M is the number of calls made to next.

Things to watch for

Dividing by Window Size Before It Is Full

When fewer than size elements have been added, the average should be calculated using the actual count of elements, not the window size. Dividing by size when only a few elements exist produces incorrect results. For example, with size = 3 and only one element 5, the average should be 5.0, not 5/3 = 1.67.

Not Removing the Oldest Element When Window Is Full

When using a running sum approach, forgetting to subtract the element leaving the window causes the sum to grow unbounded. After receiving more than size elements, each new element should trigger removal of the oldest one from both the queue and the running sum. Failing to do this results in the average including more elements than the window size allows.

class MovingAverage:
    def __init__(self, size: int):
        self.size = size
        self.queue = [0] * self.size
        self.head = self.window_sum = 0

        # number of elements seen so far
        self.count = 0

    def next(self, val: int) -> float:
        self.count += 1

        # calculate the new sum by shifting the window
        tail = (self.head + 1) % self.size
        self.window_sum = self.window_sum - self.queue[tail] + val

        # move on to the next head
        self.head = (self.head + 1) % self.size
        self.queue[self.head] = val
        return self.window_sum / min(self.size, self.count)

class MovingAverage {
private:
    int size;
    int head = 0;
    int windowSum = 0;
    int count = 0;
    vector<int> queue;

public:
    MovingAverage(int size) {
        this->size = size;
        queue = vector<int>(size);
    }

    double next(int val) {
        ++count;

        // calculate the new sum by shifting the window
        int tail = (head + 1) % size;
        windowSum = windowSum - queue[tail] + val;

        // move on to the next head
        head = (head + 1) % size;
        queue[head] = val;
        return windowSum * 1.0 / min(size, count);
    }
};