#!/usr/bin/env python3 """ The Data Sentinel — six daily checks on the four tables under every report, with printed severities, a quarantine that stops a wrong number reaching a report, and a fever chart. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --partial 0.15 --outage 0.2 # a worse feed python3 engine.py --storm 1 # the bad fortnight: days 20-33, every incident x4 python3 engine.py --vol-warn 5 # a tight volume threshold: watch the false alarms Everything here is deterministic. The 60 days of partitions are generated from the printed rules below with a tiny 32-bit PRNG (mulberry32) that the in-browser engine on the case page implements bit-for-bit, so the page's reference run reproduces this file's results.json exactly. No real data is involved. Stdlib only. """ import argparse, json, math, os # ----------------------------------------------------------------------------- deterministic randomness class RNG: """mulberry32 — identical to the JavaScript port on the case page.""" M = 0xFFFFFFFF def __init__(self, seed): self.a = seed & self.M def random(self): self.a = (self.a + 0x6D2B79F5) & self.M t = self.a t = ((t ^ (t >> 15)) * (1 | t)) & self.M t = ((t + (((t ^ (t >> 7)) * (61 | t)) & self.M)) & self.M) ^ t return ((t ^ (t >> 14)) & self.M) / 4294967296 def normal(self, mu=0.0, sd=1.0): u = self.random() or 1e-12; v = self.random() return mu + sd * math.sqrt(-2 * math.log(u)) * math.cos(6.283185307179586 * v) def rnd(x, d=0): """Half-up rounding computed the same way on both sides (floor(x*10^d + 0.5)); no tie surprises.""" m = 10 ** d return math.floor(x * m + 0.5) / m def rint(x): return int(math.floor(x + 0.5)) def f1(x): return "%.1f" % rnd(x, 1) def f2(x): return "%.2f" % rnd(x, 2) DAYS = 60 SAMPLE = 400 # amount values sampled per table per day for the distribution check BASELINE_DAYS = 30 # the PSI baseline: 30 days x 400 values, frozen before day 0 Z_DECILES = [-1.2816, -0.8416, -0.5244, -0.2533, 0.0, 0.2533, 0.5244, 0.8416, 1.2816] # 10 printed buckets # ----------------------------------------------------------------------------- the tables (printed) # name, daily rows, key columns with their baseline null rate, amount column ~ lognormal(mu, sd) or None TABLES = [ {"name": "customers", "rows": 2000, "cols": [("email", 0.04), ("gstin", 0.12)], "amount": None}, {"name": "invoices", "rows": 45000, "cols": [("po_ref", 0.06), ("due_date", 0.005)], "amount": (10.3, 0.8)}, {"name": "payments", "rows": 30000, "cols": [("invoice_id", 0.01), ("utr", 0.03)], "amount": (10.3, 0.8)}, {"name": "arrears_snapshot", "rows": 850000, "cols": [("segment", 0.02), ("balance", 0.0)], "amount": (8.5, 0.9)}, ] VOL_JITTER = 0.03 # daily volume ~ rows x (1 + normal(0, 3%)) NULL_JITTER = 0.006 # daily null rate ~ baseline + normal(0, 0.6 pp) DUP_BASE, DUP_SD = 0.0003, 0.0002 # duplicate key share on a normal day ORPH_BASE, ORPH_SD = 0.0005, 0.0004 # orphan payments (no matching invoice) on a normal day ARRIVE_MU, ARRIVE_SD = 345, 30 # partition lands 05:45 +- 30 min (minutes after midnight) # ----------------------------------------------------------------------------- the checks (printed) CHECKS = ["C1", "C2", "C3", "C4", "C5", "C6"] CHECK_NAME = {"C1": "VOLUME DRIFT", "C2": "NULL RATE", "C3": "DUPLICATE KEYS", "C4": "REFERENTIAL INTEGRITY", "C5": "FRESHNESS", "C6": "DISTRIBUTION SHIFT"} VOL_CRIT = 0.35 # |delta vs 7-day median| > vol_warn -> WARN, > 35% -> CRITICAL NULL_WARN, NULL_CRIT = 0.05, 0.15 # null rate above the column's baseline: +5 pp WARN, +15 pp CRITICAL DUP_WARN, DUP_CRIT = 0.001, 0.01 # duplicate keys > 0.1% WARN, > 1% CRITICAL ORPH_WARN, ORPH_CRIT = 0.002, 0.02 # payments -> invoices orphans > 0.2% WARN, > 2% CRITICAL FRESH_WARN, FRESH_CRIT = 420, 540 # partition later than 07:00 WARN, missing by 09:00 CRITICAL PSI_WARN, PSI_CRIT = 0.10, 0.25 # PSI vs the 30-day baseline > 0.10 WARN, > 0.25 CRITICAL ESCALATE_WINDOW = 2 # a WARN on the same check within the previous 2 days (3 days incl. today) -> CRITICAL P_CLEAR = 0.75 # a quarantined table is cleared by its owner next morning with p 0.75 STORM = (20, 33, 4.0) # "the bad fortnight": days 20-33, every incident probability x4 SEV_WEIGHT = {0: 0, 1: 1, 2: 3} # the fever score: WARN 1, CRITICAL 3, summed over every check on every table # ----------------------------------------------------------------------------- incidents (printed) # type, tables it can hit (indexes into TABLES), the check(s) that should catch it INCIDENTS = [ ("PARTIAL_LOAD", [0, 1, 2, 3], ["C1"]), # volume -40% ("SCHEMA_NULL", [0, 1, 2, 3], ["C2"]), # a renamed column arrives empty for a drawn share (10-100%) of rows ("RERUN", [0, 1, 2, 3], ["C1", "C3"]), # the day loaded twice: rows x2, half the keys duplicated ("OUTAGE", [0, 1, 2, 3], ["C5"]), # the partition lands late, 07:00-11:00 (drawn) ("FX_SHIFT", [1, 2, 3], ["C6"]), # amounts shift: log-mean +0.15..0.60 (drawn) ("ORPHANS", [2], ["C4"]), # a late invoice feed: 0.3-5.3% of payments have no invoice (drawn) ] INCIDENT_INDEX = {k[0]: i for i, k in enumerate(INCIDENTS)} # the six downstream reports and the tables each one reads REPORTS = [ ("DAILY ARREARS POSITION", [3, 0]), ("COLLECTIONS DASHBOARD", [1, 2, 0]), ("AGEING BY SEGMENT", [3, 0]), ("CASH FORECAST", [1, 2]), ("BOARD PACK", [0, 1, 2, 3]), ("REGULATORY RETURN", [3, 0]), ] def bucket(v, edges): b = 0 for e in edges: if v > e: b += 1 return b def psi(p, q): """Population stability index over the 10 buckets, shares floored at 1e-4.""" s = 0.0 for i in range(10): a = max(p[i], 1e-4); b = max(q[i], 1e-4) s += (a - b) * math.log(a / b) return s def hhmm(m): m = rint(m) return "%02d:%02d" % (m // 60, m % 60) # ----------------------------------------------------------------------------- the engine def run(seed=42, partial=0.05, schema=0.04, rerun=0.03, outage=0.06, shift=0.04, orphans=0.05, vol_warn=15, storm=0): params = {"seed": seed, "p_partial_load": partial, "p_schema_null": schema, "p_rerun": rerun, "p_outage": outage, "p_fx_shift": shift, "p_orphans": orphans, "vol_warn_pct": vol_warn, "storm": storm} rng = RNG(seed) probs = [partial, schema, rerun, outage, shift, orphans] vol_warn_f = vol_warn / 100.0 # 0. the PSI baseline: 30 days x 400 values per amount table, in table order, before day 0 edges, base_shares = {}, {} for t in TABLES: if t["amount"] is None: continue mu, sd = t["amount"] e = [math.exp(mu + sd * z) for z in Z_DECILES] counts = [0] * 10 for _ in range(BASELINE_DAYS * SAMPLE): counts[bucket(math.exp(rng.normal(mu, sd)), e)] += 1 edges[t["name"]] = e base_shares[t["name"]] = [c / float(BASELINE_DAYS * SAMPLE) for c in counts] hist = [[t["rows"]] * 7 for t in TABLES] # the 7-day volume history, seeded with the printed daily rows quarantined = [False] * len(TABLES) last_warn = [[-99] * 6 for _ in TABLES] grid = [[] for _ in TABLES] # status per table per day grid_sev = [[] for _ in TABLES] # the six check severities per table per day (0/1/2) alerts, incidents, fever, quarantine_days_marks = [], [], [], [] psi_series = {t["name"]: [] for t in TABLES if t["amount"] is not None} pages = digest = escalations = clears = 0 reports_prevented = 0 reports_by_name = {r[0]: 0 for r in REPORTS} log = [] for day in range(DAYS): # 1. overnight: the owner of each quarantined table clears it with p 0.75 (one draw per quarantined table, in table order) for ti in range(len(TABLES)): if quarantined[ti]: if rng.random() < P_CLEAR: quarantined[ti] = False; clears += 1 log.append("day %02d %-16s cleared by owner, back in service" % (day, TABLES[ti]["name"])) # 2. incidents: one draw per type in the printed order; if it fires, one draw picks the table, then its magnitude mult = STORM[2] if (storm and STORM[0] <= day <= STORM[1]) else 1.0 today = [{} for _ in TABLES] for k, (kind, applicable, _by) in enumerate(INCIDENTS): if rng.random() < probs[k] * mult: ti = applicable[int(rng.random() * len(applicable))] inc = {"day": day, "table": TABLES[ti]["name"], "type": kind} if kind == "SCHEMA_NULL": inc["col"] = int(rng.random() * len(TABLES[ti]["cols"])); inc["share"] = 0.1 + rng.random() * 0.9 elif kind == "OUTAGE": inc["minutes"] = 420 + rng.random() * 240 elif kind == "FX_SHIFT": inc["shift"] = 0.15 + rng.random() * 0.45 elif kind == "ORPHANS": inc["rate"] = 0.003 + rng.random() * 0.05 today[ti][kind] = inc; incidents.append(inc) # 3. each table lands its partition (draw order: volume, nulls per column, duplicates, orphans, arrival, amounts), then the checks run score = 0 for ti, t in enumerate(TABLES): inc = today[ti] vol = rint(t["rows"] * (1 + rng.normal(0, VOL_JITTER))) if "PARTIAL_LOAD" in inc: vol = rint(vol * 0.6) if "RERUN" in inc: vol = vol * 2 nulls = [] for col, base in t["cols"]: nulls.append(min(1.0, max(0.0, base + rng.normal(0, NULL_JITTER)))) if "SCHEMA_NULL" in inc: nulls[inc["SCHEMA_NULL"]["col"]] = inc["SCHEMA_NULL"]["share"] dup = max(0.0, rng.normal(DUP_BASE, DUP_SD)) if "RERUN" in inc: dup = 0.5 orphan = None if t["name"] == "payments": orphan = max(0.0, rng.normal(ORPH_BASE, ORPH_SD)) if "ORPHANS" in inc: orphan = inc["ORPHANS"]["rate"] arrival = rng.normal(ARRIVE_MU, ARRIVE_SD) if "OUTAGE" in inc: arrival = inc["OUTAGE"]["minutes"] psi_v = None if t["amount"] is not None: mu, sd = t["amount"] sh = inc["FX_SHIFT"]["shift"] if "FX_SHIFT" in inc else 0.0 counts = [0] * 10 for _ in range(SAMPLE): counts[bucket(math.exp(rng.normal(mu + sh, sd)), edges[t["name"]])] += 1 psi_v = psi([c / float(SAMPLE) for c in counts], base_shares[t["name"]]) psi_series[t["name"]].append(rnd(psi_v, 4)) # the checks, in order; each yields 0 OK / 1 WARN / 2 CRITICAL and a one-line detail sev = [0] * 6; detail = [""] * 6; value = [None] * 6 med = sorted(hist[ti])[3] delta = (vol - med) / float(med) sev[0] = 2 if abs(delta) > VOL_CRIT else 1 if abs(delta) > vol_warn_f else 0 value[0] = rnd(delta * 100, 1); detail[0] = "rows %d vs 7-day median %d (%s%%)" % (vol, med, f1(delta * 100)) hist[ti] = hist[ti][1:] + [vol] worst = 0; worst_d = -1.0 for ci, (col, base) in enumerate(t["cols"]): d = nulls[ci] - base if d > worst_d: worst_d = d; worst = ci sev[1] = 2 if worst_d > NULL_CRIT else 1 if worst_d > NULL_WARN else 0 value[1] = rnd(worst_d * 100, 1); detail[1] = "%s null %s%% vs baseline %s%% (%s pp)" % (t["cols"][worst][0], f1(nulls[worst] * 100), f1(t["cols"][worst][1] * 100), f1(worst_d * 100)) sev[2] = 2 if dup > DUP_CRIT else 1 if dup > DUP_WARN else 0 value[2] = rnd(dup * 100, 2); detail[2] = "duplicate keys %s%%" % f2(dup * 100) if orphan is not None: sev[3] = 2 if orphan > ORPH_CRIT else 1 if orphan > ORPH_WARN else 0 value[3] = rnd(orphan * 100, 2); detail[3] = "payments without an invoice %s%%" % f2(orphan * 100) else: detail[3] = "n/a" sev[4] = 2 if arrival > FRESH_CRIT else 1 if arrival > FRESH_WARN else 0 value[4] = rint(arrival); detail[4] = "partition landed %s" % hhmm(arrival) if arrival <= FRESH_CRIT else "partition missing at 09:00 (landed %s)" % hhmm(arrival) if psi_v is not None: sev[5] = 2 if psi_v > PSI_CRIT else 1 if psi_v > PSI_WARN else 0 value[5] = rnd(psi_v, 4); detail[5] = "PSI %s vs 30-day baseline" % ("%.3f" % rnd(psi_v, 3)) else: detail[5] = "n/a" # escalation: a WARN on a check that also warned within the previous 2 days becomes CRITICAL has_inc = len(inc) > 0 for c in range(6): if sev[c] == 0: continue esc = False if sev[c] == 1: if last_warn[ti][c] >= day - ESCALATE_WINDOW: sev[c] = 2; esc = True; escalations += 1 last_warn[ti][c] = day alerts.append({"day": day, "table": t["name"], "check": CHECKS[c], "severity": "CRITICAL" if sev[c] == 2 else "WARN", "escalated": esc, "value": value[c], "detail": detail[c], "incident": has_inc}) # the policy if max(sev) == 2: status = "CRITICAL"; quarantined[ti] = True; pages += 1 elif quarantined[ti]: status = "QUARANTINED" elif max(sev) == 1: status = "WARN"; digest += 1 else: status = "OK" grid[ti].append(status); grid_sev[ti].append(sev) score += sum(SEV_WEIGHT[s] for s in sev) # was each injected incident caught by the check meant to catch it? for kind, i in inc.items(): by = INCIDENTS[INCIDENT_INDEX[kind]][2] i["caught"] = any(sev[CHECKS.index(c)] > 0 for c in by) i["severity"] = "CRITICAL" if max(sev[CHECKS.index(c)] for c in by) == 2 else "WARN" if i["caught"] else "MISSED" what = {"PARTIAL_LOAD": "load stopped at 60%", "RERUN": "day loaded twice", "OUTAGE": "feed late, landed " + hhmm(i.get("minutes", 0)), "SCHEMA_NULL": "%s arrives %s%% null" % (t["cols"][i.get("col", 0)][0], f1(i.get("share", 0) * 100)), "FX_SHIFT": "amounts shift, log-mean +%s" % f2(i.get("shift", 0)), "ORPHANS": "%s%% of payments without an invoice" % f2(i.get("rate", 0) * 100)}[kind] i["what"] = what log.append("day %02d %-16s %-12s %-40s -> %s%s" % (day, t["name"], kind, what, "/".join(c for c in by if sev[CHECKS.index(c)] > 0) or "MISSED", " -> " + status if i["caught"] else "")) for c in range(6): if sev[c] > 0 and not has_inc: log.append("day %02d %-16s %-12s %-40s -> %s %s (false alarm)" % (day, t["name"], "no incident", detail[c][:40], CHECKS[c], "CRITICAL" if sev[c] == 2 else "WARN")) fever.append(score) anyq = any(grid[ti][day] in ("CRITICAL", "QUARANTINED") for ti in range(len(TABLES))) quarantine_days_marks.append(1 if anyq else 0) # downstream: a report is prevented from reading a bad table when any table it reads is out today for name, reads in REPORTS: if any(grid[ti][day] in ("CRITICAL", "QUARANTINED") for ti in reads): reports_prevented += 1; reports_by_name[name] += 1 return summarise(params, grid, grid_sev, alerts, incidents, fever, quarantine_days_marks, psi_series, edges, base_shares, pages, digest, escalations, clears, reports_prevented, reports_by_name, log) def summarise(params, grid, grid_sev, alerts, incidents, fever, qmarks, psi_series, edges, base_shares, pages, digest, escalations, clears, reports_prevented, reports_by_name, log): n_tables = len(TABLES) per_table = [] for ti, t in enumerate(TABLES): row = grid[ti] counts = {s: sum(1 for x in row if x == s) for s in ("OK", "WARN", "CRITICAL", "QUARANTINED")} ta = [a for a in alerts if a["table"] == t["name"]] ti_inc = [i for i in incidents if i["table"] == t["name"]] per_table.append({"table": t["name"], "rows": t["rows"], "ok": counts["OK"], "warn": counts["WARN"], "critical": counts["CRITICAL"], "quarantined": counts["QUARANTINED"], "quarantine_days": counts["CRITICAL"] + counts["QUARANTINED"], "alerts": len(ta), "incidents": len(ti_inc), "caught": sum(1 for i in ti_inc if i["caught"])}) by_check = [] for c in CHECKS: ca = [a for a in alerts if a["check"] == c] by_check.append({"check": c, "name": CHECK_NAME[c], "warn": sum(1 for a in ca if a["severity"] == "WARN"), "critical": sum(1 for a in ca if a["severity"] == "CRITICAL"), "total": len(ca), "false_alarms": sum(1 for a in ca if not a["incident"]), "escalated": sum(1 for a in ca if a["escalated"])}) by_type = [] for kind, applicable, by in INCIDENTS: ki = [i for i in incidents if i["type"] == kind] by_type.append({"type": kind, "caught_by": "/".join(by), "injected": len(ki), "caught": sum(1 for i in ki if i["caught"]), "critical": sum(1 for i in ki if i["severity"] == "CRITICAL"), "missed": sum(1 for i in ki if not i["caught"])}) injected = len(incidents); caught = sum(1 for i in incidents if i["caught"]) status_totals = {s: sum(r[s] for r in per_table) for s in ("ok", "warn", "critical", "quarantined")} quarantine_total = status_totals["critical"] + status_totals["quarantined"] return { "engine": "data-sentinel", "seed": params["seed"], "days": DAYS, "params": params, "tables": [{"name": t["name"], "rows": t["rows"], "key_columns": [{"col": c, "null_baseline_pct": rnd(b * 100, 1)} for c, b in t["cols"]], "amount": None if t["amount"] is None else {"log_mean": t["amount"][0], "log_sd": t["amount"][1], "bucket_edges": [rint(e) for e in edges[t["name"]]], "baseline_shares": [rnd(s, 4) for s in base_shares[t["name"]]]}} for t in TABLES], "reports": [{"name": n, "reads": [TABLES[i]["name"] for i in r], "prevented": reports_by_name[n]} for n, r in REPORTS], "table_days": n_tables * DAYS, "status_totals": status_totals, "quarantine_days": quarantine_total, "alerts_total": len(alerts), "warn_total": sum(1 for a in alerts if a["severity"] == "WARN"), "critical_total": sum(1 for a in alerts if a["severity"] == "CRITICAL"), "escalations": escalations, "pages": pages, "digest_notes": digest, "clears": clears, "days_out_per_page": rnd(quarantine_total / float(pages), 2) if pages else 0.0, "incidents_injected": injected, "incidents_caught": caught, "incidents_missed": injected - caught, "recall_pct": rnd(100.0 * caught / injected, 1) if injected else 0.0, "false_alarms": sum(1 for a in alerts if not a["incident"]), "mttd_days": 0, "mttd_note": "every check runs on the partition the morning it lands, so detection is same-day by construction", "reports_prevented": reports_prevented, "by_check": by_check, "by_type": by_type, "per_table": per_table, "grid": grid, "grid_sev": grid_sev, "fever": fever, "quarantine_marks": qmarks, "psi": psi_series, "alerts": alerts, "incidents": incidents, "log_sample": log[:16], } if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--seed", type=int, default=42) ap.add_argument("--partial", type=float, default=0.05); ap.add_argument("--schema", type=float, default=0.04) ap.add_argument("--rerun", type=float, default=0.03); ap.add_argument("--outage", type=float, default=0.06) ap.add_argument("--shift", type=float, default=0.04); ap.add_argument("--orphans", type=float, default=0.05) ap.add_argument("--vol-warn", type=int, default=15, help="volume drift WARN threshold, percent (CRITICAL is fixed at 35)") ap.add_argument("--storm", type=int, default=0, help="1 = the bad fortnight: days 20-33, every incident probability x4") ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "results.json")) a = ap.parse_args() res = run(a.seed, a.partial, a.schema, a.rerun, a.outage, a.shift, a.orphans, a.vol_warn, a.storm) with open(a.out, "w") as f: json.dump(res, f, indent=1) st = res["status_totals"] print("%d table-days: OK %d · WARN %d · CRITICAL %d · QUARANTINED %d (quarantine days %d)" % (res["table_days"], st["ok"], st["warn"], st["critical"], st["quarantined"], res["quarantine_days"])) print("alerts %d (WARN %d, CRITICAL %d, of which escalated %d) · pages %d · digest notes %d · false alarms %d" % (res["alerts_total"], res["warn_total"], res["critical_total"], res["escalations"], res["pages"], res["digest_notes"], res["false_alarms"])) for c in res["by_check"]: print(" %s %-22s WARN %3d CRITICAL %3d false alarms %d" % (c["check"], c["name"], c["warn"], c["critical"], c["false_alarms"])) print("incidents injected %d, caught %d (recall %s%%), missed %d · MTTD %d days (%s)" % (res["incidents_injected"], res["incidents_caught"], res["recall_pct"], res["incidents_missed"], res["mttd_days"], "same-day")) for t in res["by_type"]: print(" %-13s injected %2d caught %2d missed %d (by %s)" % (t["type"], t["injected"], t["caught"], t["missed"], t["caught_by"])) for t in res["per_table"]: print(" %-17s OK %2d WARN %2d CRIT %2d QUAR %2d · alerts %2d · incidents %d caught %d" % (t["table"], t["ok"], t["warn"], t["critical"], t["quarantined"], t["alerts"], t["incidents"], t["caught"])) print("wrong reports prevented: %d (6 reports read these tables) · days out per page %.2f" % (res["reports_prevented"], res["days_out_per_page"])) print("wrote", a.out)