#!/usr/bin/env python3 """ The Invoice Validator — an invoice is checked against the customer's own rules before it is sent, so the customer's AP portal has nothing to reject. Part of "The Engine Room" on simranjaiswal.in. python3 engine.py # reference run, seed 42, writes results.json python3 engine.py --seed 7 --doc 1.5 --po 2 --tax 0.5 --portal 0.9 python3 engine.py --no-validator # every invoice goes out as drafted; the portal decides Everything here is deterministic. The synthetic book 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. Stdlib only. """ import argparse, json, math, os, re # ----------------------------------------------------------------------------- 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): return int(math.floor(x + 0.5)) # Math.round for x >= 0 def r1(x): return math.floor(x * 10 + 0.5) / 10 # printed means to 1 dp, same op as the JS N_CUST = 40 N_INV = 800 DAYS = 60 TERMS = 30 # every customer pays on 30-day terms DATE_WINDOW = 7 # invoice date within 7 days of delivery AMOUNT_TOL = 0.005 # amount may exceed the PO line by 0.5% GSTIN_RE = re.compile(r"^[0-9]{2}[A-Z0-9]{13}$") # format only, no checksum; state code 01–37 LETTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" ALNUM = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZ" # The ten checks, in the order the validator runs them. (code, group, base defect rate, which customers require it) CHECKS = [ ("PO_MISSING", "po", 0.08, "needs_po"), ("PO_EXHAUSTED", "po", 0.04, "needs_po"), ("GSTIN_INVALID", "tax", 0.03, "all"), ("STATE_MISMATCH", "tax", 0.05, "all"), ("ADDRESS_DRIFT", "doc", 0.06, "all"), ("HSN_MISSING", "doc", 0.07, "hsn"), ("DATE_WINDOW", "doc", 0.06, "date_win"), ("DN_MISSING", "doc", 0.09, "dn"), ("AMOUNT_VS_PO", "doc", 0.04, "po_cap"), ("DUPLICATE_NUMBER", "doc", 0.01, "all"), ] CODES = [c[0] for c in CHECKS] # ----------------------------------------------------------------------------- the synthetic book def make_gstin(rng, state): """15 chars: 2-digit state, a PAN-shaped 10 (5 letters, 4 digits, 1 letter), entity '1', 'Z', a check char.""" s = "%02d" % state for _ in range(5): s += LETTERS[int(rng.random() * 26)] for _ in range(4): s += str(int(rng.random() * 10)) s += LETTERS[int(rng.random() * 26)] s += "1Z" + ALNUM[int(rng.random() * 36)] return s def build_customers(rng): """40 customers, each with a printed requirements profile. Draw order matters — the page port mirrors it.""" customers = [] for c in range(N_CUST): state = 1 + int(rng.random() * 37) # bill-to state, 01–37 gstin = make_gstin(rng, state) cust = { "id": "C%02d" % (c + 1), "state": state, "gstin": gstin, "address": "MASTER-%02d" % (c + 1), "needs_po": rng.random() < 0.70, # needs a PO number; if so the PO must be open with remaining value "po_cap": rng.random() < 0.30, # max invoice value per PO line "hsn": rng.random() < 0.50, # line-level HSN codes "date_win": rng.random() < 0.40, # invoice date within 7 days of delivery "dn": rng.random() < 0.35, # delivery note number on the invoice "wdays": rng.random() < 0.25, # portal accepts submissions on working days only } customers.append(cust) return customers def build_invoices(rng, customers, rates): """800 invoices over 60 days. Defects are injected independently at the printed rates; every random number a world will need later (fix delay, portal delay, rejection roll, payment lag) is drawn HERE, so both worlds see the same draws in the same order.""" invoices = [] for i in range(N_INV): cust = int(rng.random() * N_CUST) C = customers[cust] delivery = int(rng.random() * DAYS) quote = rnd(math.exp(rng.normal(10.2, 0.8))) # the PO line / quote value quote = min(300000, max(3000, quote)) gap = int(rng.random() * 4) # invoice raised 0–3 days after delivery defects = [rng.random() < rates[k] for k in range(len(CHECKS))] d = dict(zip(CODES, defects)) inv = {"number": "INV-%04d" % (i + 1), "cust": cust, "delivery": delivery, "quote": quote, "amount": quote, "po": {"no": "PO-%s-%03d" % (C["id"], i + 1), "remaining": quote}, "gstin": C["gstin"], "bill_to": C["address"], "hsn": True, "dn": True, "defects": [CODES[k] for k in range(len(CHECKS)) if defects[k]]} # the defects, applied to the document in check order if d["PO_MISSING"]: inv["po"] = None if d["PO_EXHAUSTED"] and inv["po"] is not None: inv["po"]["remaining"] = 0 # the PO is closed / fully billed if d["STATE_MISMATCH"]: # the wrong registration's GSTIN s2 = 1 + int(rng.random() * 37) if s2 == C["state"]: s2 = s2 % 37 + 1 inv["gstin"] = "%02d" % s2 + inv["gstin"][2:] if d["GSTIN_INVALID"]: mode = int(rng.random() * 3) inv["gstin"] = "" if mode == 0 else inv["gstin"][:14] if mode == 1 else "99" + inv["gstin"][2:] if d["ADDRESS_DRIFT"]: inv["bill_to"] = "OLD-%02d" % (cust + 1) if d["HSN_MISSING"]: inv["hsn"] = False if d["DATE_WINDOW"]: gap = 8 + int(rng.random() * 10) # 8–17 days after delivery if d["DN_MISSING"]: inv["dn"] = False if d["AMOUNT_VS_PO"]: inv["amount"] = rnd(quote * (1 + 0.01 + rng.random() * 0.09)) # 1–10% over the PO line if d["DUPLICATE_NUMBER"] and i > 0: inv["number"] = invoices[int(rng.random() * i)]["number"] inv["issue"] = delivery + gap # the draws both worlds will use inv["fix_delay"] = 1 + int(rng.random() * 2) # fixing a held invoice takes 1–2 days inv["reject_delay"] = 4 + int(rng.random() * 10) # the portal takes 4–13 days to reject inv["roll"] = rng.random() # one roll against the portal's rejection probability inv["lag"] = int(round(rng.normal(8, 6))) # days after terms the customer actually pays (may be negative) invoices.append(inv) return invoices # ----------------------------------------------------------------------------- the validator def validate(inv, C, seen): """Run every check the customer requires, in order; return the list of failures (all of them, not the first).""" fails = [] if C["needs_po"] and inv["po"] is None: fails.append("PO_MISSING") if C["needs_po"] and inv["po"] is not None and inv["po"]["remaining"] <= 0: fails.append("PO_EXHAUSTED") # closed or fully billed g = inv["gstin"] gstin_ok = bool(GSTIN_RE.match(g)) and 1 <= int(g[:2]) <= 37 if not gstin_ok: fails.append("GSTIN_INVALID") if gstin_ok and int(g[:2]) != C["state"]: fails.append("STATE_MISMATCH") # place of supply: GSTIN state = bill-to state if inv["bill_to"] != C["address"]: fails.append("ADDRESS_DRIFT") if C["hsn"] and not inv["hsn"]: fails.append("HSN_MISSING") if C["date_win"] and inv["issue"] - inv["delivery"] > DATE_WINDOW: fails.append("DATE_WINDOW") if C["dn"] and not inv["dn"]: fails.append("DN_MISSING") if C["po_cap"] and inv["po"] is not None and inv["amount"] > inv["quote"] * (1 + AMOUNT_TOL): fails.append("AMOUNT_VS_PO") if inv["number"] in seen: fails.append("DUPLICATE_NUMBER") return fails def slide(day, wdays): """Portals that only take submissions on working days: a weekend send lands on Monday. Day 0 is a Monday.""" if wdays: while day % 7 in (5, 6): day += 1 return day def settle(inv, C, mode, portal_p): """One invoice's journey in one world. mode 'validator': held and fixed before sending. mode 'portal': sent as drafted; the customer's portal rejects with p per applicable failure, the clock restarts.""" k = len(inv["fails"]) send0 = slide(inv["issue"], C["wdays"]) base_cash = send0 + TERMS + inv["lag"] out = {"held": False, "rejected": False, "slipped": False} if mode == "validator": if k: out["held"] = True send = slide(inv["issue"] + inv["fix_delay"], C["wdays"]) else: send = send0 cash = send + TERMS + inv["lag"] else: q = 1.0 for _ in range(k): q *= (1 - portal_p) # P(slips through) = (1 - p)^k, multiplied out the same way in JS if k and inv["roll"] < 1 - q: out["rejected"] = True reissue = slide(send0 + inv["reject_delay"] + inv["fix_delay"], C["wdays"]) cash = reissue + TERMS + inv["lag"] else: if k: out["slipped"] = True cash = base_cash out["cash"] = cash out["delay"] = cash - base_cash # days added by this world, vs a clean send out["days_to_cash"] = cash - inv["issue"] return out HIST_LO, HIST_STEP, HIST_N = 15, 5, 14 # 15–19 … 80–84, outsiders clamped into the end bins def world(invoices, customers, mode, portal_p): n = len(invoices); held = rejected = slipped = 0 rupee_days = 0; sum_days = 0; total = 0 hist = [0] * HIST_N per_cust = {} for inv in invoices: C = customers[inv["cust"]] o = settle(inv, C, mode, portal_p) inv[mode] = o held += o["held"]; rejected += o["rejected"]; slipped += o["slipped"] rupee_days += inv["amount"] * o["delay"]; sum_days += o["days_to_cash"]; total += inv["amount"] b = (o["days_to_cash"] - HIST_LO) // HIST_STEP hist[max(0, min(HIST_N - 1, b))] += 1 pc = per_cust.setdefault(C["id"], {"cust": C["id"], "invoices": 0, "held": 0, "rejected": 0, "rupee_days": 0}) pc["invoices"] += 1; pc["held"] += o["held"]; pc["rejected"] += o["rejected"]; pc["rupee_days"] += inv["amount"] * o["delay"] clean = n - held - rejected - slipped return { "mode": mode, "invoices": n, "clean": clean, "held": held, "rejected": rejected, "slipped": slipped, "rupee_days": int(rupee_days), # Σ ₹ × days delayed; ÷ 1e5 = "lakh-days" "mean_days_to_cash": r1(sum_days / n), "days_added_to_dso": r1(rupee_days / total), # value-weighted delay, in days "days_to_cash_hist": hist, "per_customer": per_cust, } def fmt(v): return "₹{:,}".format(int(round(v))) # ----------------------------------------------------------------------------- the run def run(seed=42, doc_mult=1.0, po_mult=1.0, tax_mult=1.0, portal_p=0.85, validator=True): params = {"seed": seed, "doc_mult": doc_mult, "po_mult": po_mult, "tax_mult": tax_mult, "portal_p": portal_p, "validator": bool(validator), "terms_days": TERMS, "date_window_days": DATE_WINDOW, "amount_tolerance": AMOUNT_TOL, "fix_delay_days": "1-2", "portal_reject_delay_days": "4-13", "pay_lag": "normal(8, 6)"} mult = {"doc": doc_mult, "po": po_mult, "tax": tax_mult} rates = [min(1.0, c[2] * mult[c[1]]) for c in CHECKS] rng = RNG(seed) customers = build_customers(rng) invoices = build_invoices(rng, customers, rates) # validate every invoice once — the customer's rules are the same whoever runs them seen = set() for inv in invoices: inv["fails"] = validate(inv, customers[inv["cust"]], seen) seen.add(inv["number"]) with_v = world(invoices, customers, "validator" if validator else "portal", portal_p) without = world(invoices, customers, "portal", portal_p) return summarise(params, rates, customers, invoices, with_v, without) def summarise(params, rates, customers, invoices, with_v, without): wm = with_v["mode"] # the ten checks: who requires them, how often the defect was injected, how often it actually failed, what it cost by_reason = [] for k, (code, group, base, req) in enumerate(CHECKS): requiring = sum(1 for C in customers if req == "all" or C[req]) injected = sum(1 for inv in invoices if code in inv["defects"]) failed = sum(1 for inv in invoices if code in inv["fails"]) caused = sum(1 for inv in invoices if code in inv["fails"] and inv["portal"]["rejected"]) by_reason.append({"check": code, "group": group, "base_rate": base, "rate": rates[k], "customers_requiring": requiring, "defects_injected": injected, "failures": failed, "rejections_without": caused}) defective = sum(1 for inv in invoices if inv["fails"]) multi = sum(1 for inv in invoices if len(inv["fails"]) > 1) profiles = [{"id": C["id"], "state": C["state"], "needs_po": C["needs_po"], "po_cap": C["po_cap"], "hsn": C["hsn"], "date_win": C["date_win"], "dn": C["dn"], "wdays": C["wdays"], "invoices": without["per_customer"].get(C["id"], {"invoices": 0})["invoices"], "held": with_v["per_customer"].get(C["id"], {"held": 0})["held"], "rejected": without["per_customer"].get(C["id"], {"rejected": 0})["rejected"]} for C in customers] top = sorted(without["per_customer"].values(), key=lambda r: (-r["rejected"], -r["rupee_days"], r["cust"]))[:10] top = [{"cust": r["cust"], "invoices": r["invoices"], "rejected": r["rejected"], "rupee_days": int(r["rupee_days"])} for r in top] holds = [{"number": inv["number"], "cust": customers[inv["cust"]]["id"], "amount": inv["amount"], "fails": inv["fails"]} for inv in invoices if inv["fails"]][:12] total_value = sum(inv["amount"] for inv in invoices) avoided = without["rupee_days"] - with_v["rupee_days"] # value-weighted mean days to cash per world, then the improvement, all rounded the same way mean_with = sum(inv[wm]["days_to_cash"] for inv in invoices) / len(invoices) mean_without = sum(inv["portal"]["days_to_cash"] for inv in invoices) / len(invoices) for w in (with_v, without): del w["per_customer"] return { "engine": "invoice-validator", "seed": params["seed"], "params": params, "gstin_regex": GSTIN_RE.pattern, "book": {"customers": N_CUST, "invoices": N_INV, "days": DAYS, "value": int(total_value), "defective": defective, "multi_failure": multi}, "with_validator": with_v, "without_validator": without, "headline": {"rupee_days_avoided": int(avoided), "mean_days_improvement": r1(mean_without - mean_with), "rejections_avoided": without["rejected"] - with_v["rejected"], "held": with_v["held"]}, "by_reason": by_reason, "profiles": profiles, "rejections_by_customer": top, "sample_holds": holds, "tie": {"with": with_v["clean"] + with_v["held"] + with_v["rejected"] + with_v["slipped"], "without": without["clean"] + without["rejected"] + without["slipped"], "held_equals_defective": with_v["held"] == defective if params["validator"] else None}, } if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--seed", type=int, default=42) ap.add_argument("--doc", type=float, default=1.0, help="multiplier on the document-defect rates (address, HSN, date, delivery note, amount, duplicate)") ap.add_argument("--po", type=float, default=1.0, help="multiplier on the PO-defect rates (missing, exhausted)") ap.add_argument("--tax", type=float, default=1.0, help="multiplier on the tax-defect rates (GSTIN, state)") ap.add_argument("--portal", type=float, default=0.85, help="the portal's rejection probability per applicable failure") ap.add_argument("--no-validator", action="store_true") 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.doc, a.po, a.tax, a.portal, not a.no_validator) with open(a.out, "w") as f: json.dump(res, f, indent=1) b, W, X, H = res["book"], res["with_validator"], res["without_validator"], res["headline"] print("book: %d invoices to %d customers over %d days, %s billed · %d defective (%d with more than one failure)" % (b["invoices"], b["customers"], b["days"], fmt(b["value"]), b["defective"], b["multi_failure"])) print("with the validator%s: clean %d · held %d · rejected %d · slipped %d · %s lakh-days delayed · mean %.1f days to cash · +%.1f days on DSO" % ("" if res["params"]["validator"] else " OFF", W["clean"], W["held"], W["rejected"], W["slipped"], "{:,}".format(W["rupee_days"] // 100000), W["mean_days_to_cash"], W["days_added_to_dso"])) print("without : clean %d · rejected %d · slipped %d · %s lakh-days delayed · mean %.1f days to cash · +%.1f days on DSO" % (X["clean"], X["rejected"], X["slipped"], "{:,}".format(X["rupee_days"] // 100000), X["mean_days_to_cash"], X["days_added_to_dso"])) for r in res["by_reason"]: print(" %-17s require %2d/40 injected %3d failed %3d caused %3d rejections" % (r["check"], r["customers_requiring"], r["defects_injected"], r["failures"], r["rejections_without"])) print("avoided: %s lakh-days of cash delay · %.1f days off the mean days-to-cash · %d rejections" % ("{:,}".format(H["rupee_days_avoided"] // 100000), H["mean_days_improvement"], H["rejections_avoided"])) print("tie-out: with %d · without %d · held = defective: %s" % (res["tie"]["with"], res["tie"]["without"], res["tie"]["held_equals_defective"])) print("wrote", a.out)