Replay determinism check

# /// script
# requires-python = ">=3.10"
# ///
# Copyright (c) 2026 Aayush Sharma · SPDX-License-Identifier: MIT
# https://thedataengineering.com/lab-code/LICENSE.txt
"""Replay determinism check.

A daily-totals job over a fixed event file that calls datetime.now() and
reads a mutable lookup table; run twice and compare output hashes. Then the
fixed version: the event time comes from the event, and the lookup is
versioned by date.
"""
import argparse, datetime, hashlib, json, random

p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
p.add_argument("--events", type=int, default=50_000, help="Events.")
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()
random.seed(args.seed)
events = [{"ts": f"2025-10-{random.randint(1, 30):02d}T{random.randint(0, 23):02d}:00", "sku": random.choice("ABCD"), "qty": random.randint(1, 5)}
          for _ in range(args.events)]
prices = {"A": 10, "B": 20, "C": 30, "D": 40}                     # a mutable lookup someone edits
price_versions = {"2025-10-01": {"A": 10, "B": 20, "C": 30, "D": 40}}

def job_v1(run_time):
    totals = {}
    for e in events:
        day = run_time.date().isoformat()                         # bug: processing time instead of event time
        totals[day] = totals.get(day, 0) + e["qty"] * prices[e["sku"]]
    return hashlib.sha256(json.dumps(totals, sort_keys=True).encode()).hexdigest()[:16]

def job_v2():
    totals = {}
    for e in events:
        day = e["ts"][:10]
        table = price_versions[max(k for k in price_versions if k <= day)]       # the version valid on the event's date
        totals[day] = totals.get(day, 0) + e["qty"] * table[e["sku"]]
    return hashlib.sha256(json.dumps(totals, sort_keys=True).encode()).hexdigest()[:16]

a = job_v1(datetime.datetime(2025, 11, 1, 9)); prices["C"] = 35
b = job_v1(datetime.datetime(2025, 11, 2, 9))
c = job_v2(); price_versions["2025-11-02"] = {"A": 10, "B": 20, "C": 35, "D": 40}
d = job_v2()
rows = [{"version": "now() + mutable lookup", "run1": a, "run2": b, "identical": "yes" if a == b else "no"},
        {"version": "event time + versioned lookup", "run1": c, "run2": d, "identical": "yes" if c == d else "no"}]
for r in rows: print(f"{r['version']:<32} run 1 {r['run1']}  run 2 {r['run2']}  identical: {r['identical']}")

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