PYTHON
Implement a Fixed-Size Sliding Window with collections.deque
Learn to efficiently manage a fixed-size window of data using Python's collections.deque, perfect for moving averages or recent history tracking without costly list shifts.
from collections import deque
def calculate_sliding_average(data_stream, window_size):
"""
Calculates the sliding average of a data stream using a deque.
Maintains a fixed-size window and updates the sum efficiently.
"""
if not isinstance(data_stream, list) or not data_stream:
return []
if not isinstance(window_size, int) or window_size <= 0:
raise ValueError("Window size must be a positive integer.")
window = deque(maxlen=window_size)
current_sum = 0
averages = []
for item in data_stream:
window.append(item)
current_sum += item
if len(window) > window_size: # Should only happen if maxlen wasn't set, but good for clarity
removed_item = window.popleft()
current_sum -= removed_item
if len(window) == window_size:
averages.append(current_sum / window_size)
else:
# If window is not yet full, just for initial elements
# Or, if you want average only when window is full, remove this else block
averages.append(current_sum / len(window))
return averages
# Example usage:
data = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
window_s = 3
result = calculate_sliding_average(data, window_s)
# print(f"Data: {data}")
# print(f"Window Size: {window_s}")
# print(f"Sliding Averages: {result}")
How it works: This snippet demonstrates using `collections.deque` to implement a fixed-size sliding window. `deque` (double-ended queue) allows efficient O(1) appending and popping from both ends. By setting `maxlen`, the deque automatically removes elements from the opposite end when the maximum size is reached. This is crucial for problems like calculating moving averages, where you need to maintain a constant-size window of recent data without costly re-slicing or shifting of a standard list. The `current_sum` is updated incrementally to avoid re-summing the window for each step, further boosting efficiency.