How to Build a Burn Rate Alert with Multiple Time Windows in Python

Track token consumption and trigger alerts when the burn rate exceeds a threshold across multiple time windows using deque and time-based sliding windows.

Medium Python 3.9+ Aug 9, 2026 Observability & SRE 16 views 0 copies

burn-rate alerts time-windows deque sre monitoring

Python code

51 lines

Python 3.9+

import time
from collections import deque

class BurnRateAlert:
    def __init__(self, windows_seconds=(60, 300, 900), threshold_rate=0.8):
        self.windows = {w: deque() for w in windows_seconds}
        self.threshold_rate = threshold_rate
        self.previous_tokens = None

    def record_sample(self, current_tokens, timestamp=None):
        timestamp = timestamp or time.time()
        if self.previous_tokens is not None:
            tokens_consumed = self.previous_tokens - current_tokens
            for window, queue in self.windows.items():
                queue.append((timestamp, tokens_consumed))
                while queue and queue[0][0] <= timestamp - window:
                    queue.popleft()
        self.previous_tokens = current_tokens

    def get_rates(self, timestamp=None):
        timestamp = timestamp or time.time()
        rates = {}
        for window, queue in self.windows.items():
            recent = [tokens for t, tokens in queue if t > timestamp - window]
            total_consumed = sum(recent)
            rates[window] = total_consumed / window
        return rates

    def check_alerts(self):
        rates = self.get_rates()
        alerts = []
        for window, rate in rates.items():
            if rate > self.threshold_rate:
                alerts.append(f"ALERT: Burn rate {rate:.2f} tokens/sec over last {window}s exceeds threshold {self.threshold_rate}")
        return alerts


if __name__ == "__main__":
    monitor = BurnRateAlert(windows_seconds=(60, 300), threshold_rate=1.0)
    t0 = 1000.0
    monitor.record_sample(1000, t0)
    for i in range(1, 11):
        t = t0 + i * 10
        monitor.record_sample(1000 - i * 50, t)

    print(f"Token samples (10 samples, 10s apart, consuming 50 tokens each):")
    print(f"  Rates: { {w: f'{r:.2f}' for w, r in monitor.get_rates(t0+100).items()} }")
    for alert in monitor.check_alerts():
        print(alert)
    if not monitor.check_alerts():
        print("No alerts triggered.")

Output

stdout

Token samples (10 samples, 10s apart, consuming 50 tokens each):
  Rates: {'60': '6.67', '300': '3.33'}
No alerts triggered.

How it works

The class maintains a deque per time window to store (timestamp, tokens_consumed) pairs. On each sample, it appends the difference between previous and current token counts and prunes entries older than the window. get_rates computes the rate by summing recent consumption and dividing by the window length. Alert checks compare each window's rate against the threshold and produce human-readable messages. Using deque ensures O(1) appends and pops from either end, ideal for sliding window calculations.

Common mistakes

Forgetting to update `previous_tokens` on the first sample, which causes a large false consumption spike.
Not pruning old entries, causing memory bloat and inaccurate rates as the window slides.
Comparing rate without considering the direction of token change (increases vs decreases).
Using wall-clock time with fractional seconds incorrectly in timestamp comparisons.

Variations

Use `time.monotonic()` instead of `time.time()` for monotonic timestamps immune to system clock changes.
Store raw samples and compute rates on demand using a list comprehension instead of pre-pruning.

Real-world use cases

Monitoring LLM API token consumption to detect abnormal usage and trigger cost-control alerts.
Tracking network bandwidth usage per rolling time window to enforce rate limits in a rate limiter service.
Observing resource consumption (CPU, memory) in a microservice to alert on potential runaway processes.

How to Build a Burn Rate Alert with Multiple Time Windows in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Observability & SRE

Keep learning

Tutorials

Quizzes