Moving Average from Data Stream

Difficulty: Easy | Acceptance: N/A | Paid: No Topics: Array, Design, Queue, Data Stream

Given a stream of integers and a window size, calculate the moving average of all integers in the sliding window.

Implement the MovingAverage class:

MovingAverage(int size) Initializes the object with the size of the window size. double next(int val) Returns the moving average of the last size values of the stream.

• Examples
• Constraints
• Brute Force
• Queue with Running Sum
• Circular Buffer

Examples

Example 1: Input: [“MovingAverage”, “next”, “next”, “next”, “next”] [[3], [1], [10], [3], [5]]

Output: [null, 1.0, 5.5, 4.66667, 6.0]

Explanation: MovingAverage m = new MovingAverage(3); m.next(1); // return 1.0 = 1 / 1 m.next(10); // return 5.5 = (1 + 10) / 2 m.next(3); // return 4.66667 = (1 + 10 + 3) / 3 m.next(5); // return 6.0 = (10 + 3 + 5) / 3

Constraints

1 <= size <= 1000
-10⁵ <= val <= 10⁵
At most 10⁵ calls will be made to next.

Brute Force

Intuition Store all incoming values in a list and calculate the sum of the last size elements for each call to next().

Steps

• Store all values in a list/array
• For each next() call, sum the last min(size, total_count) elements
• Return the average

class MovingAverage:

    def __init__(self, size: int):
        self.size = size
        self.values = []

    def next(self, val: int) -> float:
        self.values.append(val)
        window = self.values[-self.size:] if len(self.values) >= self.size else self.values
        return sum(window) / len(window)

Complexity

• Time: O(n) for each next() call where n is the window size
• Space: O(n) to store all values
• Notes: Simple but inefficient for large windows

Queue with Running Sum

Intuition Use a queue to maintain the sliding window and track the running sum to avoid recalculating it each time.

Steps

• Maintain a queue and a running sum
• When adding a new value, add it to the queue and sum
• If queue exceeds size, remove the oldest element and subtract from sum
• Return sum / queue size

from collections import deque

class MovingAverage:

    def __init__(self, size: int):
        self.size = size
        self.queue = deque()
        self.window_sum = 0

    def next(self, val: int) -> float:
        self.queue.append(val)
        self.window_sum += val
        
        if len(self.queue) > self.size:
            self.window_sum -= self.queue.popleft()
        
        return self.window_sum / len(self.queue)

Complexity

• Time: O(1) for each next() call
• Space: O(n) where n is the window size
• Notes: Optimal solution with constant time operations

Circular Buffer

Intuition Use a fixed-size array with circular indexing to avoid dynamic memory allocation and maintain O(1) operations.

Steps

• Initialize a fixed-size array and track the current index and count
• Store values at the current index and update the running sum
• If buffer is full, subtract the value being overwritten
• Update index circularly and return average

class MovingAverage:

    def __init__(self, size: int):
        self.size = size
        self.buffer = [0] * size
        self.idx = 0
        self.count = 0
        self.window_sum = 0

    def next(self, val: int) -> float:
        if self.count < self.size:
            self.count += 1
            self.window_sum += val
        else:
            self.window_sum += val - self.buffer[self.idx]
        
        self.buffer[self.idx] = val
        self.idx = (self.idx + 1) % self.size
        
        return self.window_sum / self.count

Complexity

• Time: O(1) for each next() call
• Space: O(n) where n is the window size
• Notes: Memory efficient with pre-allocated buffer, no dynamic allocation overhead