# /// script
# requires-python = ">=3.10"
# dependencies = ["numpy"]
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Measure your skew distribution, and price each nine.

One million arrival skews (arrival time minus event time) from a heavy-tailed
distribution. The p50, p99, p99.9 and p99.99 of the skew; for each, used as
the watermark delay δ, the share of records that arrive late and the extra
time every window waits.
"""
import argparse, json
import numpy as np

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--n", type=int, default=1_000_000, help="Records.")
p.add_argument("--shape", type=float, default=1.2, help="Pareto shape of the skew.")
p.add_argument("--scale", type=float, default=2.0, help="Skew scale (s).")
p.add_argument("--seed", type=int, default=0, help="Random seed.")
p.add_argument("--json", help="Also write the results to this JSON file.")
args = p.parse_args()

skew = np.random.default_rng(args.seed).pareto(args.shape, args.n) * args.scale    # heavy-tailed: mostly small, sometimes very large
rows = []
for q in (50, 99, 99.9, 99.99):
    d = float(np.percentile(skew, q))
    rows.append({"quantile": f"p{q:g}", "delta_s": d, "late_pct": float((skew > d).mean() * 100)})
for i, r in enumerate(rows): r["extra_wait_s"] = r["delta_s"] - (rows[i - 1]["delta_s"] if i else 0)
summary = {"p9999_over_p50": rows[-1]["delta_s"] / rows[0]["delta_s"]}
for r in rows: print(f"δ = {r['quantile']:<6} = {r['delta_s']:10.1f} s   late records {r['late_pct']:6.3f}%   extra wait over the previous nine {r['extra_wait_s']:10.1f} s")

if args.json:
    with open(args.json, "w") as f:
        json.dump({"summary": summary, "tables": {"quantiles": rows}}, f)