#!/usr/bin/env python3 """ The Dispute Clock — every dispute gets an owner, a cause and a deadline, and the cash it unlocks is measured. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --triage 12 --sla-mult 0.8 --no-escalation --seed 7 Everything here is deterministic. The 300 synthetic disputes are drawn once from the printed rules 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 customer data is involved. THE ORDER OF RANDOM DRAWS (the JavaScript port follows it exactly) per dispute, in id order: 1 draw (day raised) + 2 draws (lognormal amount) + 1 draw (cause) + 1 draw (triage effort, 1 or 2 days) + 2 draws (the owner's raw duration, lognormal around the SLA) + 1 draw (outcome) + 2 draws (credit share, only when the outcome is CREDIT). Every draw happens at generation time, so the two policies (THE CLOCK and NO CLOCK) replay exactly the same disputes with the same durations; they differ only in the triage desk's capacity, the multiplier applied to the drawn duration, the deadline, and whether a missed deadline escalates. 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) # ----------------------------------------------------------------------------- the printed constants N_DISPUTES = 300 RAISE_SPAN = 120 # disputes are raised on days 0..119 HORIZON = 150 # the cash series runs day 1..150 SNAPSHOT = 120 # the day the open pile and the big number are measured CAUSES = ["PRICE", "QUANTITY", "QUALITY", "PO_MISMATCH", "TAX", "OTHER"] CAUSE_SHARE = [0.30, 0.20, 0.15, 0.20, 0.10, 0.05] OWNER = {"PRICE": "PRICING", "QUANTITY": "OPS", "QUALITY": "QA", "PO_MISMATCH": "SALES", "TAX": "FINANCE", "OTHER": "TRIAGE"} SLA = {"PRICE": 3, "QUANTITY": 5, "QUALITY": 7, "PO_MISMATCH": 4, "TAX": 3, "OTHER": 5} # working days with the owner TEAMS = ["PRICING", "OPS", "QA", "SALES", "FINANCE", "TRIAGE"] OUTCOMES = ["CREDIT", "RE_INVOICE", "UPHELD"] OUTCOME_P = { # P(CREDIT), P(RE_INVOICE); the rest is UPHELD (the customer pays in full) "PRICE": (0.55, 0.15), "QUANTITY": (0.60, 0.20), "QUALITY": (0.50, 0.10), "PO_MISMATCH": (0.10, 0.70), "TAX": (0.20, 0.60), "OTHER": (0.30, 0.30), } DURATION_SD = 0.45 # the owner's duration is exp(normal(log(SLA), 0.45)) working days, then x the multiplier NO_CLOCK = {"triage_capacity": 2, "duration_multiplier": 2.2} # the world without a clock, fixed AGE_BANDS = [("0–7", 0, 7), ("8–14", 8, 14), ("15–30", 15, 30), ("31–60", 31, 60), ("61+", 61, 10**6)] DAY_BINS = [("0–5", 0, 5), ("6–10", 6, 10), ("11–15", 11, 15), ("16–20", 16, 20), ("21–30", 21, 30), ("31–45", 31, 45), ("46–60", 46, 60), ("61–90", 61, 90), ("91+", 91, 10**6)] # ----------------------------------------------------------------------------- integer-safe helpers (the port does the same) def rhu(x): return int(math.floor(x + 0.5)) # round half up; JS Math.floor(x + 0.5) def r1(num, den): return (num * 20 + den) // (2 * den) / 10 if den else 0.0 def wd(d): return d % 7 not in (5, 6) # day 0 is a Monday; 5 and 6 are the weekend def add_wd(d, n): """The calendar day n working days after d.""" for _ in range(n): d += 1 while not wd(d): d += 1 return d def pct_rank(sorted_vals, q): """Nearest-rank percentile on an already-sorted list.""" n = len(sorted_vals) return sorted_vals[min(n - 1, int(math.ceil(q * n)) - 1)] if n else 0 def median(sorted_vals): n = len(sorted_vals) if not n: return 0.0 return r1(sorted_vals[n // 2] * 2, 2) if n % 2 else r1(sorted_vals[n // 2 - 1] + sorted_vals[n // 2], 2) # ----------------------------------------------------------------------------- the synthetic disputes def build_disputes(seed): """300 disputes, every random draw made here, in this order.""" rng = RNG(seed) disputes = [] for i in range(N_DISPUTES): day = int(rng.random() * RAISE_SPAN) amount = min(400000, max(5000, rhu(math.exp(rng.normal(10.4, 0.8))))) r = rng.random(); acc = 0.0; cause = CAUSES[-1] for c, s in zip(CAUSES, CAUSE_SHARE): acc += s if r < acc: cause = c; break triage_days = 1 + int(rng.random() * 2) # 1 or 2 working days of triage effort raw = math.exp(rng.normal(math.log(SLA[cause]), DURATION_SD)) # the owner's duration, before any multiplier r = rng.random(); pc, pr = OUTCOME_P[cause] outcome = "CREDIT" if r < pc else "RE_INVOICE" if r < pc + pr else "UPHELD" share = 0.0; credit = 0 if outcome == "CREDIT": share = min(1.0, max(0.05, rng.normal(0.4, 0.25))) # a partial credit, 5–100% of the invoice credit = rhu(amount * share) disputes.append({"id": "D-%03d" % (i + 1), "day": day, "amount": amount, "cause": cause, "owner": OWNER[cause], "triage_days": triage_days, "raw": raw, "outcome": outcome, "credit": credit}) return disputes # ----------------------------------------------------------------------------- the engine def simulate(disputes, clock, triage_capacity, sla_mult, escalation): """Walk every dispute through LOGGED -> TRIAGED -> WITH_OWNER -> RESOLVED under one policy. clock=True : THE CLOCK. The triage desk opens up to `triage_capacity` new disputes per working day (first logged, first opened); the owner's deadline is SLA x sla_mult working days and the drawn duration is scaled by the same multiplier (a shorter clock is a faster owner); a dispute still open at its deadline escalates to the owner's manager and the remaining duration is halved, rounded up (a resolution hazard x2 from the deadline on). clock=False : NO CLOCK. Desk capacity 2/day, no deadline, no escalation, the drawn duration x 2.2. Its "within SLA" is measured against the clock's base SLA it never had, so the two compare. """ cap = triage_capacity if clock else NO_CLOCK["triage_capacity"] mult = sla_mult if clock else NO_CLOCK["duration_multiplier"] order = sorted(range(len(disputes)), key=lambda i: (disputes[i]["day"], i)) opened = {} # working day -> disputes the desk opened that day rows = [None] * len(disputes) for i in order: d = disputes[i] s = d["day"] while not (wd(s) and opened.get(s, 0) < cap): s += 1 # queue for the desk opened[s] = opened.get(s, 0) + 1 w = add_wd(s, d["triage_days"]) # handed to the owner dur = max(1, rhu(d["raw"] * mult)) # the owner's duration, working days sla = max(1, rhu(SLA[d["cause"]] * sla_mult)) if clock else SLA[d["cause"]] esc = clock and escalation and dur > sla esc_day = add_wd(w, sla) if esc else None actual = sla + (dur - sla + 1) // 2 if esc else dur # after escalation the remainder is halved, rounded up r = add_wd(w, actual) rows[i] = {"id": d["id"], "cause": d["cause"], "owner": d["owner"], "amount": d["amount"], "logged": d["day"], "opened": s, "with_owner": w, "sla": sla, "owner_days": dur, "escalated": esc, "escalated_day": esc_day, "resolved": r, "days": r - d["day"], "within_sla": dur <= sla, "outcome": d["outcome"], "credit": d["credit"], "cash": d["amount"] - d["credit"]} return rows def summarise(rows): n = len(rows) days = sorted(x["days"] for x in rows) by_day = [0] * (HORIZON + 1) for x in rows: if x["resolved"] <= HORIZON: by_day[x["resolved"]] += x["cash"] series = []; acc = 0 for d in range(1, HORIZON + 1): acc += by_day[d]; series.append(acc) resolved = [x for x in rows if x["resolved"] <= SNAPSHOT] in_triage = [x for x in rows if x["with_owner"] > SNAPSHOT] with_owner = [x for x in rows if x["with_owner"] <= SNAPSHOT < x["resolved"]] open_ = in_triage + with_owner by_age = [{"band": b, "n": sum(1 for x in open_ if lo <= SNAPSHOT - x["logged"] <= hi), "value": sum(x["amount"] for x in open_ if lo <= SNAPSHOT - x["logged"] <= hi)} for b, lo, hi in AGE_BANDS] hist = [{"bin": b, "n": sum(1 for x in rows if lo <= x["days"] <= hi)} for b, lo, hi in DAY_BINS] outcomes = [{"outcome": o, "n": sum(1 for x in resolved if x["outcome"] == o), "cash": sum(x["cash"] for x in resolved if x["outcome"] == o), "credit": sum(x["credit"] for x in resolved if x["outcome"] == o)} for o in OUTCOMES] by_cause = [] for c in CAUSES: cs = [x for x in rows if x["cause"] == c]; cd = sorted(x["days"] for x in cs) by_cause.append({"cause": c, "owner": OWNER[c], "sla": SLA[c], "n": len(cs), "median_days": median(cd), "p90_days": pct_rank(cd, 0.9), "within_sla_pct": r1(sum(1 for x in cs if x["within_sla"]) * 100, len(cs)), "credit_share_pct": r1(sum(x["credit"] for x in cs) * 100, sum(x["amount"] for x in cs)), "cash_120": sum(x["cash"] for x in cs if x["resolved"] <= SNAPSHOT)}) by_owner = [] for t in TEAMS: ts = [x for x in rows if x["owner"] == t]; od = sorted(x["owner_days"] for x in ts) by_owner.append({"team": t, "n": len(ts), "open_120": sum(1 for x in ts if x["with_owner"] <= SNAPSHOT < x["resolved"]), "escalated": sum(1 for x in ts if x["escalated"]), "median_owner_days": median(od), "sla_hit_pct": r1(sum(1 for x in ts if x["within_sla"]) * 100, len(ts))}) return { "median_days": median(days), "p90_days": pct_rank(days, 0.9), "mean_days": r1(sum(days) * 10, n * 10) if n else 0.0, "within_sla_pct": r1(sum(1 for x in rows if x["within_sla"]) * 100, n), "escalations": sum(1 for x in rows if x["escalated"]), "escalations_by_120": sum(1 for x in rows if x["escalated"] and x["escalated_day"] <= SNAPSHOT), "resolved_120": len(resolved), "with_owner_120": len(with_owner), "in_triage_120": len(in_triage), "cash_120": sum(x["cash"] for x in resolved), "credit_120": sum(x["credit"] for x in resolved), "cash_150": sum(x["cash"] for x in rows if x["resolved"] <= HORIZON), "open_value_120": sum(x["amount"] for x in open_), "outcomes_120": outcomes, "open_by_age_120": by_age, "days_histogram": hist, "by_cause": by_cause, "by_owner": by_owner, "cash_by_day": series, } def journey(x): return {"logged": x["logged"], "opened": x["opened"], "with_owner": x["with_owner"], "sla": x["sla"], "owner_days": x["owner_days"], "escalated_day": x["escalated_day"], "resolved": x["resolved"], "days": x["days"], "within_sla": x["within_sla"]} def run(seed=42, triage_capacity=6, sla_mult=1.0, escalation=True): params = {"seed": seed, "triage_capacity": triage_capacity, "sla_multiplier": sla_mult, "escalation": escalation, "no_clock": dict(NO_CLOCK), "duration_sd": DURATION_SD, "snapshot_day": SNAPSHOT, "horizon_day": HORIZON} disputes = build_disputes(seed) C = simulate(disputes, True, triage_capacity, sla_mult, escalation) N = simulate(disputes, False, triage_capacity, sla_mult, escalation) SC, SN = summarise(C), summarise(N) total = sum(d["amount"] for d in disputes) book = {"disputes": len(disputes), "value": total, "credit_value": sum(d["credit"] for d in disputes), "by_cause": [{"cause": c, "n": sum(1 for d in disputes if d["cause"] == c), "value": sum(d["amount"] for d in disputes if d["cause"] == c)} for c in CAUSES], "outcomes": [{"outcome": o, "n": sum(1 for d in disputes if d["outcome"] == o)} for o in OUTCOMES]} order = sorted(range(len(disputes)), key=lambda i: (disputes[i]["day"], i)) picks = order[:3] first_esc = next((i for i in order if C[i]["escalated"]), None) if first_esc is not None and first_esc not in picks: picks.append(first_esc) sample = [{"id": disputes[i]["id"], "cause": disputes[i]["cause"], "owner": disputes[i]["owner"], "amount": disputes[i]["amount"], "outcome": disputes[i]["outcome"], "credit": disputes[i]["credit"], "CLOCK": journey(C[i]), "NO_CLOCK": journey(N[i])} for i in picks] return { "engine": "dispute-clock", "seed": seed, "params": params, "book": book, "policies": {"CLOCK": SC, "NO_CLOCK": SN}, "comparison": {"extra_cash_120": SC["cash_120"] - SN["cash_120"], "extra_cash_150": SC["cash_150"] - SN["cash_150"], "extra_cash_120_pct_of_book": r1((SC["cash_120"] - SN["cash_120"]) * 100, total), "median_days_saved": rhu((SN["median_days"] - SC["median_days"]) * 10) / 10, "p90_days_saved": SN["p90_days"] - SC["p90_days"], "fewer_open_120": (SN["with_owner_120"] + SN["in_triage_120"]) - (SC["with_owner_120"] + SC["in_triage_120"]), "within_sla_gain_pts": rhu((SC["within_sla_pct"] - SN["within_sla_pct"]) * 10) / 10}, "sample": sample, } def fmt(v): return "₹{:,}".format(int(round(v))) if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--seed", type=int, default=42) ap.add_argument("--triage", type=int, default=6, help="triage desk capacity, disputes opened per working day (the clock)") ap.add_argument("--sla-mult", type=float, default=1.0, help="multiplier on every SLA (and on the drawn owner duration)") ap.add_argument("--no-escalation", action="store_true", help="switch off escalation at the deadline") 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.triage, a.sla_mult, not a.no_escalation) with open(a.out, "w") as f: json.dump(res, f, indent=1) b = res["book"]; P = res["policies"] print("book: %d disputes worth %s over %d days; credits drawn %s" % (b["disputes"], fmt(b["value"]), RAISE_SPAN, fmt(b["credit_value"]))) for k in ("CLOCK", "NO_CLOCK"): p = P[k] print("%-9s median %.1f d · p90 %d d · within SLA %.1f%% · escalations %d · by day %d: resolved %d, with owner %d, in triage %d · cash unlocked %s" % ( k, p["median_days"], p["p90_days"], p["within_sla_pct"], p["escalations"], SNAPSHOT, p["resolved_120"], p["with_owner_120"], p["in_triage_120"], fmt(p["cash_120"]))) c = res["comparison"] print("the clock unlocks %s more by day %d (%.1f%% of the book); median %.1f days sooner, p90 %d days sooner, %d fewer open" % ( fmt(c["extra_cash_120"]), SNAPSHOT, c["extra_cash_120_pct_of_book"], c["median_days_saved"], c["p90_days_saved"], c["fewer_open_120"])) for r in P["CLOCK"]["by_cause"]: print(" %-12s %-8s SLA %d n %3d median %5.1f d within %5.1f%% credit %5.1f%% cash by 120 %s" % (r["cause"], r["owner"], r["sla"], r["n"], r["median_days"], r["within_sla_pct"], r["credit_share_pct"], fmt(r["cash_120"]))) print("wrote", a.out)