Sliding Window Streaming Mock in Python

A simple Python class that maintains a sliding window of recent streaming values and computes the running average.

Easy Python 3.9+ Aug 9, 2026 Big data & Spark 12 views 0 copies

streaming sliding-window averages simulation data-structures

Python code

33 lines

Python 3.9+

import time
import random

class StreamingMock:
    """Produces a stream of numbers using a sliding window."""
    
    def __init__(self, window_size=5):
        self.window = []
        self.window_size = window_size
        
    def push(self, value):
        """Add a value, sliding the window forward."""
        self.window.append(value)
        if len(self.window) > self.window_size:
            self.window.pop(0)
        return self.window
    
    def average(self):
        """Return average of current window, or None if empty."""
        if not self.window:
            return None
        return sum(self.window) / len(self.window)


if __name__ == "__main__":
    stream = StreamingMock(window_size=3)
    
    # Simulate streaming data: feed values one at a time
    for i in range(1, 8):
        value = random.randint(1, 100)
        window = stream.push(value)
        print(f"Pushed {value:3d} | Window: {window} | Avg: {stream.average():.2f}")
        time.sleep(0.1)  # simulate real-time streaming

Output

stdout

Pushed  42 | Window: [42] | Avg: 42.00
Pushed  87 | Window: [42, 87] | Avg: 64.50
Pushed  15 | Window: [42, 87, 15] | Avg: 48.00
Pushed  63 | Window: [87, 15, 63] | Avg: 55.00
Pushed  91 | Window: [15, 63, 91] | Avg: 56.33
Pushed  28 | Window: [63, 91, 28] | Avg: 60.67
Pushed  54 | Window: [91, 28, 54] | Avg: 57.67

How it works

The push method appends each new value to the internal list and, if the list exceeds the configured window_size, removes the oldest element via pop(0). This ensures the window always holds the most recent N items, simulating a fixed-size sliding window over a streaming source. The average method computes the mean of the current window using sum and len, returning None when the window is empty. The main loop feeds random integers one at a time with a small delay, printing the current window and its average after each push to mimic real-time data ingestion.

Common mistakes

Using `pop(0)` on a large window is O(n) — prefer `collections.deque` for better performance.
Forgetting to handle the empty window case in `average`, leading to a ZeroDivisionError.
Assuming the window is always full — the average is computed over whatever values are present until the window size is reached.
Not resetting the window between streams, causing stale data to persist.

Variations

Use `collections.deque(maxlen=window_size)` to automatically drop the oldest item without manual pop.
Use NumPy's `np.convolve` or `pandas.Series.rolling` for vectorized sliding window averages on batches.

Real-world use cases

Computing a rolling average of sensor readings in an IoT monitoring system.
Tracking moving averages of financial market prices in real-time trading algorithms.
Aggregating recent user activity metrics for live dashboard updates in web analytics.

Sliding Window Streaming Mock in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Big data & Spark

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Big data & Spark

Keep learning

Tutorials

Quizzes