Anomaly count equals rate times window

# /// script
# requires-python = ">=3.10"
# dependencies = ["numpy"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Anomaly count equals rate times window.

Conflicting operations arrive at random (a Poisson process); for each one,
count how many others fall inside a window W after it, and compare the
average with lambda x W for windows from one SQL statement to lag under load.
"""
import argparse, json
import numpy as np

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--rate", type=float, default=667, help="lambda: conflicting operations per second (667/s = 40,000 per minute).")
p.add_argument("--seconds", type=float, default=60, help="Simulated time.")
p.add_argument("--seed", type=int, default=7, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()
rng = np.random.default_rng(args.seed)

lam, T = args.rate, args.seconds
n = rng.poisson(lam * T)
t = np.sort(rng.uniform(0, T, n))
named = [("single statement 50us", 50e-6), ("replica lag 20ms", 0.020), ("app read-modify-write 50ms", 0.050),
         ("lag under load 2s", 2.0)]
grid = sorted(set([w for _, w in named] + list(np.logspace(-5, 0.5, 12))))
rows = []
for W in grid:
    inside = np.searchsorted(t, t + W) - np.arange(n) - 1      # other operations in the next W seconds
    rows.append({"window_s": float(W), "measured": float(inside.mean()), "lambda_W": lam * W})
for name, W in named:
    r = next(r for r in rows if r["window_s"] == W)
    print(f"{name:30s} measured {r['measured']:10.4f}   lambda*W {r['lambda_W']:10.4f}")
a, b = (next(r for r in rows if r["window_s"] == W)["measured"] for W in (0.050, 50e-6))
print(f"\napp read-modify-write window / single statement window: {a / b:,.0f}x more conflicts")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"tables": {"windows": rows, "named": [{"window": nm, **next(r for r in rows if r["window_s"] == W)} for nm, W in named],
                              "summary": [{"ratio": a / b, "n": int(n)}]}}, f)