How to Build a Consumer Lag Gauge in Python

import time
import random
from collections import deque


class ConsumerLagGauge:
    """Mock consumer lag gauge measuring how far behind a consumer is."""

    def __init__(self, producer_rate=10, consumer_rate=7, initial_lag=0):
        self.producer_rate = producer_rate
        self.consumer_rate = consumer_rate
        self.lag = initial_lag
        self.history = deque(maxlen=10)

    def tick(self):
        """Simulate one time unit of production and consumption."""
        produced = random.randint(0, self.producer_rate)
        consumed = random.randint(0, self.consumer_rate)
        self.lag += produced - consumed
        self.lag = max(0, self.lag)
        self.history.append(self.lag)
        return self.lag

    def current_lag(self):
        return self.lag

    def average_lag(self):
        if not self.history:
            return 0
        return sum(self.history) / len(self.history)

    def is_healthy(self, threshold=10):
        return self.lag <= threshold


if __name__ == "__main__":
    gauge = ConsumerLagGauge(producer_rate=10, consumer_rate=7, initial_lag=5)
    print("Timestamp | Lag | Healthy")
    for i in range(6):
        lag = gauge.tick()
        print(f"{i:9} | {lag:3} | {gauge.is_healthy()}")
    print(f"\nFinal lag: {gauge.current_lag()}")
    print(f"Average lag: {gauge.average_lag():.1f}")