To maintain time-based sliding-window mean/variance with O(1) update, O(1) query, and O(1) memory per active key, use a fixed number of time buckets per key (B = ceil(T / bucket_size)). Each bucket holds aggregated summary (count, sum, sumsq, bucket_ts). On each update you map timestamp→bucket index, if bucket is stale subtract its summary from the key-level aggregate, reset it, then add the new value to that bucket and to the key-level aggregate. Query uses the key-level aggregates to compute mean/variance in O(1).python
import math
from collections import defaultdict
class SlidingWindowAggregator:
def __init__(self, window_seconds, bucket_size_seconds):
# number of buckets per key (bounded memory)
self.window = window_seconds
self.bucket_size = bucket_size_seconds
self.buckets_per_key = (window_seconds + bucket_size_seconds - 1) // bucket_size_seconds
self.store = {} # key -> {'buckets': [...], 'agg_count':int, 'agg_sum':float, 'agg_sumsq':float}
def _make_empty_bucket(self):
return {'ts': -1, 'count': 0, 'sum': 0.0, 'sumsq': 0.0}
def update(self, value, key, timestamp):
if key not in self.store:
buckets = [self._make_empty_bucket() for _ in range(self.buckets_per_key)]
self.store[key] = {'buckets': buckets, 'agg_count': 0, 'agg_sum': 0.0, 'agg_sumsq': 0.0}
entry = self.store[key]
idx = (timestamp // self.bucket_size) % self.buckets_per_key
b = entry['buckets'][idx]
bucket_start = (timestamp // self.bucket_size) * self.bucket_size
# if bucket is stale (outside window) reset by removing old contribution
if b['ts'] != bucket_start:
# remove old bucket's contribution from aggregate
entry['agg_count'] -= b['count']
entry['agg_sum'] -= b['sum']
entry['agg_sumsq'] -= b['sumsq']
# reset bucket
b['ts'] = bucket_start
b['count'] = 0
b['sum'] = 0.0
b['sumsq'] = 0.0
# add value to bucket and to aggregate
b['count'] += 1
b['sum'] += value
b['sumsq'] += value * value
entry['agg_count'] += 1
entry['agg_sum'] += value
entry['agg_sumsq'] += value * value
def query_mean_variance(self, key):
if key not in self.store or self.store[key]['agg_count'] == 0:
return None # or (0.0, 0.0) based on desired semantics
e = self.store[key]
n = e['agg_count']
s = e['agg_sum']
ss = e['agg_sumsq']
mean = s / n
# population variance; for sample variance use (ss - s*s/n)/(n-1) when n>1
var = (ss - (s * s) / n) / n if n > 0 else 0.0
# numerical safety
var = max(var, 0.0)
return mean, var
Key points:- Fixed bucket count ensures O(1) memory per active key.- Updates are O(1): compute index, subtract stale bucket, add value, update aggregates.- Queries are O(1): use aggregated count/sum/sumsq to compute mean & variance.Time complexity: update O(1), query O(1). Space: O(B) per active key where B = ceil(T / bucket_size) (constant given T and bucket_size).Edge cases: choose bucket_size to trade accuracy vs memory; if bucket too large, time-window granularity is coarse. Decide population vs sample variance for your use case.