Approach: use a fixed-size circular buffer of per-second buckets for a 60-second sliding window per key; keep bounded memory by capping number of tracked keys and evicting least-recently-used. For multi-instance, emit periodic partial aggregates (per-bucket counts) and merge by summing corresponding time buckets with clock alignment.python
import time
from collections import defaultdict, OrderedDict
class SlidingRate:
def __init__(self, window_secs=60, max_keys=10000):
self.window = window_secs
self.max_keys = max_keys
self.buckets = {} # key -> list of ints (length window)
self.timestamps = {} # key -> last_seen_epoch_sec
self.lru = OrderedDict() # maintain LRU of keys
def _ensure_key(self, key, now):
if key in self.buckets:
self.lru.move_to_end(key)
return
if len(self.buckets) >= self.max_keys:
old = next(iter(self.lru))
self.buckets.pop(old); self.timestamps.pop(old); self.lru.popitem(last=False)
self.buckets[key] = [0]*self.window
self.timestamps[key] = now
self.lru[key] = None
def add(self, key, n=1, ts=None):
now = int(ts or time.time())
self._ensure_key(key, now)
age = now - self.timestamps[key]
if age >= self.window:
# reset whole buffer
self.buckets[key] = [0]*self.window
elif age > 0:
# rotate buffer forward by age seconds
b = self.buckets[key]
self.buckets[key] = b[age:] + [0]*age
self.timestamps[key] = now
self.buckets[key][-1] += n # newest bucket
def get_rate_per_minute(self, key, ts=None):
if key not in self.buckets: return 0
now = int(ts or time.time())
age = now - self.timestamps[key]
if age >= self.window:
return 0
if age > 0:
b = self.buckets[key]
buf = b[age:] + [0]*age
else:
buf = self.buckets[key]
return sum(buf)
