How to Implement a Sliding Window Counter in Python

from collections import deque
from time import time


class SlidingWindowCounter:
    def __init__(self, window_size, bucket_size=1):
        self.window_size = window_size
        self.bucket_size = bucket_size
        self.buckets = deque()

    def _evict_expired(self, now):
        while self.buckets and self.buckets[0][0] <= now - self.window_size:
            self.buckets.popleft()

    def add_event(self):
        now = time.time()
        self._evict_expired(now)
        if self.buckets and now - self.buckets[-1][0] < self.bucket_size:
            self.buckets[-1][1] += 1
        else:
            self.buckets.append([int(now), 1])
        self._evict_expired(now)

    def count(self):
        now = time.time()
        self._evict_expired(now)
        return sum(count for _, count in self.buckets)


if __name__ == "__main__":
    window = SlidingWindowCounter(window_size=10, bucket_size=2)
    for _ in range(5):
        window.add_event()
    print(f"Approximate event count in last 10s: {window.count()}")