#!/usr/bin/env python3 """ The Dunning Ladder Engine — a six-rung reminder ladder as a per-invoice state machine, raced against no ladder and against shouting. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --tempo 5 --promise 10 --no-weekend-rule --no-batching --seed 7 Everything here is deterministic. The synthetic book (50 customers x 8 invoices) and the day-by-day simulation are driven by 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 whole thing hangs on this; the JavaScript port follows it exactly) book: per customer: 1 draw for the segment; per invoice: 1 draw (issue day) + 2 draws (lognormal amount); for DISPUTED customers 1 more draw (dispute flag) and, if flagged, 1 more (resolution). a day: (1) ONE draw per unpaid invoice, in invoice order, whatever its state (disputed, on hold, handed over) — pays if the draw is below today's hazard; (2) state moves, in invoice order: dispute resolved -> rejoin at the rung its age implies; promise expired unpaid -> broken, resume at R6 today; past the handover day -> handed over; (3) contacts, in customer order then invoice order; ONE draw per contact that lands on R3/R4 (the promise-to-pay chance), only when the contact actually happens. The three policies replay the same book from the same simulation seed; their streams diverge only where their decisions differ. 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_CUST = 50 PER_CUST = 8 TERMS = 30 ISSUE_SPAN = 61 # invoices issued on days 0..60 HORIZON = 150 # the simulation runs day 1..150 SIM_SALT = 1000003 # the simulation stream is RNG(seed + SIM_SALT); the book is RNG(seed) SEGMENTS = ["PROMPT", "NUDGE", "CHRONIC", "DISPUTED"] SEG_CUT = [0.40, 0.75, 0.95] # 40% / 35% / 20% / 5% BASE = {"PROMPT": 0.20, "NUDGE": 0.03, "CHRONIC": 0.012, "DISPUTED": 0.06} # hazard per day once it starts BASE_FROM = {"PROMPT": -2, "NUDGE": 0, "CHRONIC": 0, "DISPUTED": 0} # days from due (DISPUTED: from resolution if disputed) UPLIFT = { # added to the hazard for UPLIFT_DAYS after a contact, by rung "PROMPT": {1: 0.02, 2: 0.02, 3: 0.02, 4: 0.02, 6: 0.02}, "NUDGE": {1: 0.18, 2: 0.18, 3: 0.10, 4: 0.10, 6: 0.10}, "CHRONIC": {1: 0.02, 2: 0.02, 3: 0.12, 4: 0.12, 6: 0.20}, "DISPUTED": {1: 0.18, 2: 0.18, 3: 0.10, 4: 0.10, 6: 0.10}, # once resolved they behave like NUDGE (not in the brief; chosen here) } UPLIFT_DAYS = 5 RUNGS = [(1, -3, "friendly nudge"), (2, 3, "statement"), (3, 10, "call"), (4, 20, "escalation to account owner"), (6, 35, "final notice")] RUNG_NO = [r[0] for r in RUNGS] HANDOVER_AT = 50 # days after due: the invoice goes on the handover list, in-house contact stops PROMISE_P = 0.35 # chance a customer promises to pay at an R3 / R4 contact SHOUT_EVERY = 2 # SHOUTING: a contact every 2 days from due+1 FATIGUE_AFTER = 4 # SHOUTING: each contact after the 4th to a customer halves the uplift (compounding) POLICIES = ["LADDER", "NO_LADDER", "SHOUTING"] # ----------------------------------------------------------------------------- the synthetic book def build_book(seed): """50 customers x 8 invoices, issued over days 0-60, terms 30. Segments and disputes drawn as printed.""" rng = RNG(seed) customers, invoices = [], [] for c in range(N_CUST): r = rng.random() seg = "PROMPT" if r < SEG_CUT[0] else "NUDGE" if r < SEG_CUT[1] else "CHRONIC" if r < SEG_CUT[2] else "DISPUTED" customers.append({"id": "C%03d" % (c + 1), "seg": seg}) for k in range(PER_CUST): issue = int(rng.random() * ISSUE_SPAN) amount = int(round(math.exp(rng.normal(10.0, 0.7)))) amount = min(250000, max(3000, amount)) inv = {"id": "INV-%04d" % (len(invoices) + 1), "cust": c, "issue": issue, "due": issue + TERMS, "amount": amount, "dispute": False, "resolve": None} if seg == "DISPUTED" and rng.random() < 0.5: # half of a disputed customer's invoices are disputed inv["dispute"] = True inv["resolve"] = inv["due"] + 10 + int(rng.random() * 20) invoices.append(inv) return customers, invoices # ----------------------------------------------------------------------------- integer-safe rounding (the port does the same) def r1(num, den): return (num * 20 + den) // (2 * den) / 10 if den else 0.0 def r2(num, den): return (num * 200 + den) // (2 * den) / 100 if den else 0.0 # ----------------------------------------------------------------------------- the engine def simulate(customers, invoices, policy, seed, tempo=0, promise_days=7, weekend_rule=True, batching=True): rng = RNG(seed + SIM_SALT) def slide(d): if weekend_rule: while d % 7 in (5, 6): d += 1 # day 0 is a Monday; 5 and 6 are the weekend return d def is_weekend(d): return weekend_rule and d % 7 in (5, 6) def implied_rung(inv, t): # the last rung whose (tempo-shifted) day has passed r = 0 for no, off, _ in RUNGS: if inv["due"] + off + tempo <= t: r = no return r def prev_rung(no): i = RUNG_NO.index(no) return RUNG_NO[i - 1] if i > 0 else 0 # per-invoice state S = [] for inv in invoices: S.append({"paid": None, "last_sent": 0, "last_contact": None, "last_rung": 0, "fatigue": 1.0, "promise_until": None, "handed": None, "shout_last": None, "reached": set(), "sched": {no: slide(inv["due"] + off + tempo) for no, off, _ in RUNGS}}) by_cust = [[] for _ in customers] for i, inv in enumerate(invoices): by_cust[inv["cust"]].append(i) cust_contacts = [0] * len(customers) events, seq = [], [0] def ev(t, i, kind, rung=0, amount=0): events.append({"day": t, "inv": invoices[i]["id"], "c": invoices[i]["cust"], "ev": kind, "rung": rung, "amount": amount, "seq": seq[0]}); seq[0] += 1 for i, inv in enumerate(invoices): ev(inv["issue"], i, "issued", 0, inv["amount"]) if inv["dispute"]: ev(inv["issue"], i, "dispute", 0, 0) K = {"contacts": 0, "promises": 0, "kept": 0, "broken": 0, "handover_n": 0, "handover_v": 0, "paid_n": 0, "paid_by_due7": 0, "collected": 0, "after_handover": 0, "days_to_pay": 0} reach = {no: 0 for no in RUNG_NO}; reach[5] = 0 cash_by_day, st_paid, st_open, st_handed, st_disputed = [], [], [], [], [] def hazard(i, t): inv, s = invoices[i], S[i]; seg = customers[inv["cust"]]["seg"] if inv["dispute"] and t < inv["resolve"]: return 0.0 if t < inv["due"] + BASE_FROM[seg]: return 0.0 h = BASE[seg] if policy != "NO_LADDER" and s["last_contact"] is not None and 1 <= t - s["last_contact"] <= UPLIFT_DAYS: u = UPLIFT[seg][s["last_rung"]] if policy == "SHOUTING": u = u * s["fatigue"] h += u if s["promise_until"] is not None and t <= s["promise_until"]: h *= 2 return min(1.0, h) def pay(i, t): inv, s = invoices[i], S[i] s["paid"] = t; K["paid_n"] += 1; K["collected"] += inv["amount"]; K["days_to_pay"] += t - inv["issue"] if t <= inv["due"] + 7: K["paid_by_due7"] += 1 if s["handed"] is not None: K["after_handover"] += inv["amount"] if s["promise_until"] is not None: K["kept"] += 1; s["promise_until"] = None; ev(t, i, "kept", 5, inv["amount"]) ev(t, i, "paid", 0, inv["amount"]) for t in range(1, HORIZON + 1): # (1) one draw per unpaid invoice, in invoice order for i, inv in enumerate(invoices): if S[i]["paid"] is not None: continue u = rng.random() if u < hazard(i, t): pay(i, t) # (2) state moves for i, inv in enumerate(invoices): s = S[i] if s["paid"] is not None: continue if inv["dispute"] and inv["resolve"] == t: ev(t, i, "resolved", 0, 0) imp = implied_rung(inv, t) s["last_sent"] = prev_rung(imp) if imp else 0 if s["promise_until"] is not None and t > s["promise_until"]: s["promise_until"] = None; K["broken"] += 1; ev(t, i, "broken", 6, 0) s["last_sent"] = 4; s["sched"][6] = slide(t) # resume at R6 immediately (next weekday) if policy != "NO_LADDER" and s["handed"] is None and s["promise_until"] is None and not (inv["dispute"] and t < inv["resolve"]) and t >= inv["due"] + HANDOVER_AT + tempo: s["handed"] = t; K["handover_n"] += 1; K["handover_v"] += inv["amount"]; ev(t, i, "handover", 0, inv["amount"]) # (3) contacts if policy == "LADDER": for c in range(len(customers)): batch = [] for i in by_cust[c]: inv, s = invoices[i], S[i] if s["paid"] is not None or s["handed"] is not None or s["promise_until"] is not None or (inv["dispute"] and t < inv["resolve"]): continue due_rungs = [no for no in RUNG_NO if no > s["last_sent"] and s["sched"][no] <= t] if not due_rungs: continue for no in due_rungs: reach[no] += 1; s["reached"].add(no) top = due_rungs[-1] s["last_sent"] = top batch.append((i, top)) if not batch: continue if batching: # one message per customer per day K["contacts"] += 1; cust_contacts[c] += 1 top_c = max(r for _, r in batch) for i, r in batch: S[i]["last_contact"] = t; S[i]["last_rung"] = r; ev(t, i, "contact", r, 0) if top_c in (3, 4) and rng.random() < PROMISE_P: # one draw per contact that lands on R3/R4 for i, r in batch: if r in (3, 4): S[i]["promise_until"] = t + promise_days; K["promises"] += 1; reach[5] += 1; ev(t, i, "promise", 5, 0) else: # one message per invoice for i, r in batch: K["contacts"] += 1; cust_contacts[c] += 1 S[i]["last_contact"] = t; S[i]["last_rung"] = r; ev(t, i, "contact", r, 0) if r in (3, 4) and rng.random() < PROMISE_P: S[i]["promise_until"] = t + promise_days; K["promises"] += 1; reach[5] += 1; ev(t, i, "promise", 5, 0) elif policy == "SHOUTING": for c in range(len(customers)): batch = [] for i in by_cust[c]: inv, s = invoices[i], S[i] if s["paid"] is not None or s["handed"] is not None or (inv["dispute"] and t < inv["resolve"]): continue if t < inv["due"] + 1 + tempo or is_weekend(t): continue if s["shout_last"] is not None and t - s["shout_last"] < SHOUT_EVERY: continue r = implied_rung(inv, t) or 1 s["shout_last"] = t if r not in s["reached"]: s["reached"].add(r); reach[r] += 1 batch.append((i, r)) if not batch: continue if batching: cust_contacts[c] += 1; K["contacts"] += 1 fat = 0.5 ** max(0, cust_contacts[c] - FATIGUE_AFTER) for i, r in batch: S[i]["last_contact"] = t; S[i]["last_rung"] = r; S[i]["fatigue"] = fat; ev(t, i, "contact", r, 0) else: for i, r in batch: cust_contacts[c] += 1; K["contacts"] += 1 fat = 0.5 ** max(0, cust_contacts[c] - FATIGUE_AFTER) S[i]["last_contact"] = t; S[i]["last_rung"] = r; S[i]["fatigue"] = fat; ev(t, i, "contact", r, 0) # end of day: the four states (the tie-out is checked every day) paid = open_ = handed = disputed = 0 for i, inv in enumerate(invoices): s = S[i] if s["paid"] is not None: paid += 1 elif inv["dispute"] and t < inv["resolve"]: disputed += 1 elif s["handed"] is not None: handed += 1 else: open_ += 1 assert paid + open_ + handed + disputed == len(invoices) cash_by_day.append(K["collected"]); st_paid.append(paid); st_open.append(open_); st_handed.append(handed); st_disputed.append(disputed) billed = sum(inv["amount"] for inv in invoices) paid_scan = sum(inv["amount"] for i, inv in enumerate(invoices) if S[i]["paid"] is not None) segs = [] for seg in SEGMENTS: cs = [c for c in range(len(customers)) if customers[c]["seg"] == seg] ivs = [i for c in cs for i in by_cust[c]] n = len(ivs) adh = sum(1 for i in ivs if S[i]["paid"] is not None and S[i]["paid"] <= invoices[i]["due"] + 7) col = sum(invoices[i]["amount"] for i in ivs if S[i]["paid"] is not None) bil = sum(invoices[i]["amount"] for i in ivs) segs.append({"segment": seg, "customers": len(cs), "invoices": n, "paid": sum(1 for i in ivs if S[i]["paid"] is not None), "adherence_pct": r1(adh * 100, n), "collected_pct": r1(col * 100, bil), "contacts": sum(cust_contacts[c] for c in cs)}) return { "collected": K["collected"], "collected_pct": r1(K["collected"] * 100, billed), "paid_scan": paid_scan, "paid": st_paid[-1], "open": st_open[-1], "handed_open": st_handed[-1], "disputed_open": st_disputed[-1], "adherence_pct": r1(K["paid_by_due7"] * 100, len(invoices)), "paid_by_due7": K["paid_by_due7"], "dso_days": r1(K["days_to_pay"], K["paid_n"]), "contacts": K["contacts"], "contacts_per_customer_month": r2(K["contacts"], len(customers) * (HORIZON // 30)), "handover": {"count": K["handover_n"], "value": K["handover_v"], "collected_after": K["after_handover"]}, "promises": {"made": K["promises"], "kept": K["kept"], "broken": K["broken"], "kept_pct": r1(K["kept"] * 100, K["kept"] + K["broken"])}, "rung_reach": {"R1": reach[1], "R2": reach[2], "R3": reach[3], "R4": reach[4], "R5": reach[5], "R6": reach[6], "HANDOVER": K["handover_n"]}, "contacts_by_customer": cust_contacts, "cash_by_day": cash_by_day, "states_by_day": {"paid": st_paid, "open": st_open, "handed": st_handed, "disputed": st_disputed}, "segments": segs, "_events": events, } def journey(customers, invoices, events): """The sample customer: the first (by id) with a promise on the LADDER; their whole story, in order.""" with_promise = sorted(set(e["c"] for e in events if e["ev"] == "promise")) c = with_promise[0] if with_promise else 0 rows = sorted([e for e in events if e["c"] == c], key=lambda e: (e["day"], e["seq"])) return {"id": customers[c]["id"], "segment": customers[c]["seg"], "events": [{"day": e["day"], "inv": e["inv"], "ev": e["ev"], "rung": e["rung"], "amount": e["amount"]} for e in rows]} def run(seed=42, tempo=0, promise_days=7, weekend_rule=True, batching=True): params = {"seed": seed, "tempo": tempo, "rung_offsets": {"R%d" % no: off + tempo for no, off, _ in RUNGS}, "handover_at": HANDOVER_AT + tempo, "promise_days": promise_days, "promise_p": PROMISE_P, "weekend_rule": weekend_rule, "batching": batching, "uplift_days": UPLIFT_DAYS, "shout_every": SHOUT_EVERY, "fatigue_after": FATIGUE_AFTER} customers, invoices = build_book(seed) pol = {} for p in POLICIES: pol[p] = simulate(customers, invoices, p, seed, tempo, promise_days, weekend_rule, batching) sample = journey(customers, invoices, pol["LADDER"]["_events"]) for p in POLICIES: del pol[p]["_events"] seg_counts = {s: sum(1 for c in customers if c["seg"] == s) for s in SEGMENTS} billed = sum(i["amount"] for i in invoices) L, N, SH = pol["LADDER"], pol["NO_LADDER"], pol["SHOUTING"] return { "engine": "dunning-engine", "seed": seed, "horizon_day": HORIZON, "params": params, "book": {"customers": len(customers), "invoices": len(invoices), "billed": billed, "terms": TERMS, "segments": seg_counts, "disputed_invoices": sum(1 for i in invoices if i["dispute"])}, "policies": pol, "comparison": {"extra_cash_ladder_vs_none": L["collected"] - N["collected"], "extra_pct_points_ladder_vs_none": r1((L["collected"] - N["collected"]) * 100, billed), "extra_cash_shouting_vs_none": SH["collected"] - N["collected"], "contacts_ladder": L["contacts"], "contacts_shouting": SH["contacts"]}, "sample_customer": 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("--tempo", type=int, default=0, help="days added to every rung offset (-5..10)") ap.add_argument("--promise", type=int, default=7, help="promise-to-pay window in days (3..14)") ap.add_argument("--no-weekend-rule", action="store_true", help="allow contacts on weekends") ap.add_argument("--no-batching", action="store_true", help="one message per invoice instead of one per customer per day") 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.tempo, a.promise, not a.no_weekend_rule, not a.no_batching) with open(a.out, "w") as f: json.dump(res, f, indent=1) b = res["book"] print("book: %d customers, %d invoices, billed %s, %d disputed · segments %s" % (b["customers"], b["invoices"], fmt(b["billed"]), b["disputed_invoices"], b["segments"])) print("%-10s %14s %6s %8s %7s %9s %9s %10s %12s %8s" % ("policy", "collected", "%", "adher%", "dso", "contacts", "c/cust/mo", "handovers", "hand value", "kept%")) for p in POLICIES: r = res["policies"][p] print("%-10s %14s %6.1f %8.1f %7.1f %9d %9.2f %10d %12s %8s" % (p, fmt(r["collected"]), r["collected_pct"], r["adherence_pct"], r["dso_days"], r["contacts"], r["contacts_per_customer_month"], r["handover"]["count"], fmt(r["handover"]["value"]), ("%.1f" % r["promises"]["kept_pct"]) if p == "LADDER" else "-")) L = res["policies"]["LADDER"] print("ladder rungs reached:", " ".join("%s %d" % kv for kv in L["rung_reach"].items())) print("promises: made %d, kept %d, broken %d" % (L["promises"]["made"], L["promises"]["kept"], L["promises"]["broken"])) print("tie-out day %d: %d = paid %d + open %d + handed over %d + disputed-open %d" % (HORIZON, b["invoices"], L["paid"], L["open"], L["handed_open"], L["disputed_open"])) print("extra cash, ladder vs no ladder: %s (%.1f pts of billed) · shouting vs no ladder: %s" % (fmt(res["comparison"]["extra_cash_ladder_vs_none"]), res["comparison"]["extra_pct_points_ladder_vs_none"], fmt(res["comparison"]["extra_cash_shouting_vs_none"]))) print("sample customer: %s (%s), %d events" % (res["sample_customer"]["id"], res["sample_customer"]["segment"], len(res["sample_customer"]["events"]))) print("wrote", a.out)