#!/usr/bin/env python3 """ The Eligibility Engine — who gets which letter. The logic that decides which customer in arrears gets which letter this morning, and which must never get one. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --off E02 --off E06 --threshold 100 --gap 30 --seed 7 Everything here is deterministic. The synthetic base is 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 customer data is involved: the shape is a supplier's arrears path, anonymised; every figure is synthetic. Money is held in integer pence throughout so that sums tie out exactly in both languages. 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) N_ACCOUNTS = 2000 FORMAL_MIN_P = 10000 # £100.00: a formal notice (C) needs at least this balance; below it, B again # The exclusions, in the order they are evaluated. FIRST match wins and is recorded on the account. RULES = [ ("E01", "deceased", "DECEASED / INSOLVENT", "hold"), ("E02", "vulnerable", "VULNERABILITY REGISTER", "specialist team, never a letter"), ("E03", "dispute", "OPEN DISPUTE", "hold until resolved"), ("E04", "plan_kept", "PAYMENT PLAN KEPT", "no letter"), ("E05", "paid14", "PAID IN LAST 14 DAYS", "wait"), ("E06", "letter_gap", "LETTER WITHIN THE GAP", "minimum gap"), ("E07", "below_threshold", "BALANCE BELOW THRESHOLD", "below threshold"), ("E08", "no_letters", "NO-LETTER PREFERENCE", "other channel"), ] RULE_IDS = [r[0] for r in RULES] SCRIPTS = [("A", "FIRST REMINDER", "7–21 d"), ("B", "SECOND REMINDER", "22–45 d, or 46–90 d under £100"), ("C", "FORMAL NOTICE", "46–90 d and ≥ £100"), ("D", "FINAL NOTICE / HANDOVER", "91+ d")] AGE_BANDS = [("0–6", 0, 6), ("7–21", 7, 21), ("22–45", 22, 45), ("46–90", 46, 90), ("91+", 91, 10**6)] def gbp(p): """Integer pence -> '£1,234.56'. Same string the page builds, digit for digit.""" return "£{:,}.{:02d}".format(p // 100, p % 100) # ----------------------------------------------------------------------------- the synthetic base def build_base(seed): """2,000 accounts in arrears. Per account the draws happen in exactly this order: balance (two draws, normal), days in arrears, vulnerable, deceased/insolvent, open dispute, payment plan, plan kept (only if on a plan), paid in the last 14 days, recent letter, days since that letter, no-letter preference.""" rng = RNG(seed) accounts = [] for i in range(N_ACCOUNTS): x = math.exp(rng.normal(5.6, 0.9)) x = min(5000.0, max(20.0, x)) pence = int(math.floor(x * 100 + 0.5)) # round to the penny, half up (Math.round in the port) days = int(rng.random() * 160) + 1 # days in arrears, 1–160 vulnerable = rng.random() < 0.06 # on the vulnerability register deceased = rng.random() < 0.008 # deceased or insolvent dispute = rng.random() < 0.04 # open dispute plan = rng.random() < 0.15 # active payment plan plan_kept = (rng.random() < 0.80) if plan else False paid14 = rng.random() < 0.12 # a payment in the last 14 days recent = rng.random() < 0.22 # a letter within the last 21 days since_letter = (1 + int(rng.random() * 21)) if recent else (22 + int(rng.random() * 60)) no_letters = rng.random() < 0.03 # contact preference: no letters accounts.append({"id": "A-%04d" % (i + 1), "pence": pence, "days": days, "vulnerable": vulnerable, "deceased": deceased, "dispute": dispute, "plan": plan, "plan_kept": plan_kept, "paid14": paid14, "since_letter": since_letter, "no_letters": no_letters}) return accounts # ----------------------------------------------------------------------------- the engine def hits(a, rules_on, threshold_p, gap): """Every exclusion that matches this account, in rule order. The first one is the one that fires; the length tells us whether that rule was the SOLE thing standing between the account and a letter.""" h = [] if rules_on["E01"] and a["deceased"]: h.append("E01") if rules_on["E02"] and a["vulnerable"]: h.append("E02") if rules_on["E03"] and a["dispute"]: h.append("E03") if rules_on["E04"] and a["plan"] and a["plan_kept"]: h.append("E04") if rules_on["E05"] and a["paid14"]: h.append("E05") if rules_on["E06"] and a["since_letter"] <= gap: h.append("E06") if rules_on["E07"] and a["pence"] < threshold_p: h.append("E07") if rules_on["E08"] and a["no_letters"]: h.append("E08") return h def route(a): """Routing by days in arrears, for accounts no exclusion caught.""" d = a["days"] if d < 7: return "NOT_YET_DUE" if d <= 21: return "A" if d <= 45: return "B" if d <= 90: return "C" if a["pence"] >= FORMAL_MIN_P else "B" return "D" def flag_list(a): f = [] if a["deceased"]: f.append("deceased/insolvent") if a["vulnerable"]: f.append("vulnerable") if a["dispute"]: f.append("dispute") if a["plan"]: f.append("plan " + ("kept" if a["plan_kept"] else "BROKEN")) if a["paid14"]: f.append("paid <14d") f.append("letter %dd ago" % a["since_letter"]) if a["no_letters"]: f.append("no letters") return f def decide(accounts, rules_on, threshold_p, gap): names = {r[0]: r[2] for r in RULES}; effects = {r[0]: r[3] for r in RULES} snames = {s[0]: s[1] for s in SCRIPTS} for a in accounts: h = hits(a, rules_on, threshold_p, gap) s = route(a) a["would"] = s in ("A", "B", "C", "D") # would receive a letter by age alone a["broken"] = a["plan"] and not a["plan_kept"] if h: a["rule"] = h[0]; a["script"] = None; a["outcome"] = h[0] a["sole"] = len(h) == 1 a["reason"] = "%s · %s → %s" % (h[0], names[h[0]].lower(), effects[h[0]]) if h[0] == "E06": a["reason"] += " (%d d ago, gap %d)" % (a["since_letter"], gap) if h[0] == "E07": a["reason"] += " (%s < %s)" % (gbp(a["pence"]), gbp(threshold_p)) else: a["rule"] = None; a["script"] = s; a["outcome"] = s; a["sole"] = False if s == "NOT_YET_DUE": a["reason"] = "%d d in arrears, %s → not yet due" % (a["days"], gbp(a["pence"])) else: a["reason"] = "%d d in arrears, %s → %s %s" % (a["days"], gbp(a["pence"]), s, snames[s].lower()) if s == "B" and a["days"] >= 46: a["reason"] += " (under £100, so not a formal notice)" if a["broken"]: a["reason"] += " · plan broken: extra line" return accounts # ----------------------------------------------------------------------------- the summary def summarise(accounts, params): n = len(accounts) waterfall = [] for rid, key, name, effect in RULES: rows = [a for a in accounts if a["outcome"] == rid] waterfall.append({"rule": rid, "name": name, "effect": effect, "on": params["rules_on"][rid], "count": len(rows), "pence": sum(a["pence"] for a in rows), "prevented": sum(1 for a in rows if a["would"]), "sole": sum(1 for a in rows if a["would"] and a["sole"])}) excluded = sum(w["count"] for w in waterfall) routing = [] for sid, name, band in SCRIPTS: rows = [a for a in accounts if a["outcome"] == sid] routing.append({"script": sid, "name": name, "band": band, "count": len(rows), "pence": sum(a["pence"] for a in rows), "mean_pence": (sum(a["pence"] for a in rows) // len(rows)) if rows else 0, "plan_broken": sum(1 for a in rows if a["broken"])}) nyd = [a for a in accounts if a["outcome"] == "NOT_YET_DUE"] letters = sum(r["count"] for r in routing) letters_no_excl = sum(1 for a in accounts if a["would"]) prevented = sum(w["prevented"] for w in waterfall) tie_sum = excluded + letters + len(nyd) one_each = sum(1 for a in accounts if a["outcome"] is not None) == n bands = [] for name, lo, hi in AGE_BANDS: rows = [a for a in accounts if lo <= a["days"] <= hi] bands.append({"band": name, "n": len(rows), "A": sum(1 for a in rows if a["outcome"] == "A"), "B": sum(1 for a in rows if a["outcome"] == "B"), "C": sum(1 for a in rows if a["outcome"] == "C"), "D": sum(1 for a in rows if a["outcome"] == "D"), "excluded": sum(1 for a in rows if a["rule"] is not None), "not_due": sum(1 for a in rows if a["outcome"] == "NOT_YET_DUE")}) # the audit sample: the first account each rule caught, then the first account sent to each script, # padded with the first accounts by id until there are twelve audit, seen = [], set() for want in RULE_IDS + ["A", "B", "C", "D"]: for a in accounts: if a["outcome"] == want and a["id"] not in seen: audit.append(a); seen.add(a["id"]); break for a in accounts: if len(audit) >= 12: break if a["id"] not in seen: audit.append(a); seen.add(a["id"]) audit = audit[:12] return { "engine": "eligibility-engine", "seed": params["seed"], "params": params, "book": {"accounts": n, "pence": sum(a["pence"] for a in accounts), "flags": {"vulnerable": sum(1 for a in accounts if a["vulnerable"]), "deceased": sum(1 for a in accounts if a["deceased"]), "dispute": sum(1 for a in accounts if a["dispute"]), "plan": sum(1 for a in accounts if a["plan"]), "plan_kept": sum(1 for a in accounts if a["plan_kept"]), "plan_broken": sum(1 for a in accounts if a["broken"]), "paid14": sum(1 for a in accounts if a["paid14"]), "letter_21d": sum(1 for a in accounts if a["since_letter"] <= 21), "no_letters": sum(1 for a in accounts if a["no_letters"])}}, "waterfall": waterfall, "excluded": excluded, "to_routing": n - excluded, "routing": routing, "not_yet_due": {"count": len(nyd), "pence": sum(a["pence"] for a in nyd)}, "letters": letters, "letters_if_no_exclusions": letters_no_excl, "letters_prevented": prevented, "letters_pence": sum(r["pence"] for r in routing), "tie_out": {"accounts": n, "sum": tie_sum, "one_outcome_each": one_each, "ok": tie_sum == n and one_each}, "age_bands": bands, "audit": [{"id": a["id"], "pence": a["pence"], "days": a["days"], "flags": flag_list(a), "rule": a["rule"], "script": a["script"], "outcome": a["outcome"], "reason": a["reason"]} for a in audit], } def run(seed=42, rules_on=None, threshold=50, gap=21): rules_on = dict(rules_on or {r: True for r in RULE_IDS}) params = {"seed": seed, "rules_on": {r: bool(rules_on.get(r, True)) for r in RULE_IDS}, "threshold_gbp": threshold, "gap_days": gap, "formal_notice_min_gbp": FORMAL_MIN_P // 100} accounts = build_base(seed) decide(accounts, params["rules_on"], threshold * 100, gap) return summarise(accounts, params) if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--seed", type=int, default=42) ap.add_argument("--off", action="append", default=[], help="switch an exclusion off, e.g. --off E06 (repeatable)") ap.add_argument("--threshold", type=int, default=50, help="E07: balance below this many pounds gets no letter") ap.add_argument("--gap", type=int, default=21, help="E06: minimum days since the last letter") ap.add_argument("--out", default=os.path.join(os.path.dirname(os.path.abspath(__file__)), "results.json")) a = ap.parse_args() rules_on = {r: (r not in a.off) for r in RULE_IDS} res = run(a.seed, rules_on, a.threshold, a.gap) with open(a.out, "w") as f: json.dump(res, f, indent=1) b = res["book"] print("base: %d accounts in arrears, %s owed · rules off: %s · threshold %s · gap %d d" % ( b["accounts"], gbp(b["pence"]), ",".join(a.off) or "none", gbp(a.threshold * 100), a.gap)) print("exclusions (first match wins):") for w in res["waterfall"]: print(" %s %-26s %4d %14s prevented %3d sole guard %3d" % (w["rule"], w["name"], w["count"], gbp(w["pence"]), w["prevented"], w["sole"])) print("to routing: %d" % res["to_routing"]) for r in res["routing"]: print(" %s %-26s %4d %14s mean %s plan-broken %d" % (r["script"], r["name"], r["count"], gbp(r["pence"]), gbp(r["mean_pence"]), r["plan_broken"])) print(" NOT YET DUE %19s %4d %14s" % ("", res["not_yet_due"]["count"], gbp(res["not_yet_due"]["pence"]))) t = res["tie_out"] print("tie-out: %d = %d %s" % (t["accounts"], t["sum"], "✓ published" if t["ok"] else "✗ REFUSED TO PUBLISH")) print("letters: %d going out (%s) · %d prevented · %d if no exclusions" % (res["letters"], gbp(res["letters_pence"]), res["letters_prevented"], res["letters_if_no_exclusions"])) print("wrote", a.out)