Sliding Window Average with Deque in Python
Computes the running average of a sliding window over streaming numbers using a collections.deque for O(1) pop-left operations.
Python code
26 linesfrom collections import deque
class SlidingAverage:
def __init__(self, window_size):
self.window_size = window_size
self.window = deque()
self.total = 0
def add(self, value):
self.window.append(value)
self.total += value
if len(self.window) > self.window_size:
self.total -= self.window.popleft()
return self.average()
def average(self):
if not self.window:
return 0.0
return self.total / len(self.window)
if __name__ == "__main__":
sa = SlidingAverage(3)
for v in [5, 10, 15, 20, 25]:
avg = sa.add(v)
print(f"after adding {v:>3}: window={list(sa.window):>15} avg={avg:.2f}")
Output
after adding 5: window=[5] avg=5.00
after adding 10: window=[5, 10] avg=7.50
after adding 15: window=[5, 10, 15] avg=10.00
after adding 20: window=[10, 15, 20] avg=15.00
after adding 25: window=[15, 20, 25] avg=20.00
How it works
The deque maintains the elements in the current window; when the window exceeds the specified size, the oldest value is popped from the left. The running total is updated incrementally, so the average is computed in O(1) time per add operation. The average method handles an empty window by returning 0.0. This pattern is suitable for streaming data where new values arrive continuously and you need a moving average.
Common mistakes
- Using a list and popping the first element, which is O(n) and slow for large windows.
- Forgetting to update the total when removing the oldest element, leading to a wrong average.
- Assuming the window is always full; the average is only over the current elements in the window.
Variations
- Use a fixed-length list with an index pointer to emulate a ring buffer.
- Use numpy's rolling window or pandas' rolling mean for batch data.
Real-world use cases
- Monitoring server response times to compute a rolling latency average for health checks.
- Processing real-time sensor readings to smooth noisy data with a moving average.
- Calculating rolling connection rate in a message queue consumer to detect spikes.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.