"""Synthetic process model for a fictional card issuer's dispute workflow. Every random draw for a case is made up front (common random numbers), so the current-state run and each future-state scenario see the same customers, the same amounts and the same luck. Only the process rules change between runs. All parameters are illustrative assumptions, documented in docs/assumptions.md. """ import math import random from datetime import date, datetime, timedelta SEED = 2027 N_CASES = 1200 START = date(2026, 1, 5) # Monday, week 1 WEEKS = 26 SPECIAL_CAUSE_WEEK = 14 # simulated analyst shortage, documented in the data dictionary CHANNELS = [("phone", 0.46), ("app", 0.30), ("web", 0.16), ("branch", 0.08)] # name, weight, median amount (CAD), amount spread, median investigation touch (min), # share decided in the cardholder's favour DISPUTE_TYPES = [ ("unauthorized_fraud", 0.30, 180, 0.9, 20, 0.88), ("not_received", 0.20, 95, 0.8, 32, 0.70), ("not_as_described", 0.17, 120, 0.8, 42, 0.52), ("duplicate_charge", 0.12, 45, 0.7, 14, 0.90), ("incorrect_amount", 0.11, 60, 0.8, 18, 0.80), ("cancelled_recurring", 0.10, 30, 0.6, 22, 0.62), ] INCOMPLETE_RATE = {"phone": 0.45, "app": 0.24, "web": 0.30, "branch": 0.36} MISCATEGORIZED_RATE = {"phone": 0.18, "app": 0.09, "web": 0.11, "branch": 0.15} SECOND_DOC_REQUEST_RATE = 0.30 REKEY_ERROR_RATE = 0.06 UNREACHABLE_RATE = 0.05 WITHDRAWN_RATE = 0.07 SPECIAL_CAUSE_FACTOR = 2.2 FAST_TRACK_LIMIT = 50.0 TEAMS = [ ("contact_centre", "Contact Centre"), ("disputes_triage", "Disputes Triage"), ("disputes_ops", "Disputes Operations"), ("dispute_analysts", "Dispute Analysts"), ("chargeback_team", "Chargeback Team"), ("customer_resolution", "Customer Resolution"), ("customer", "Customer (external)"), ("network_merchant", "Card network / merchant (external)"), ] CURRENT_STATE = { "digital_intake": False, "assisted_triage": False, "fast_track": False, "visual_management": False, "prefilled_chargeback": False, } # Effect of each future-state lever when fully realised (effect = 1.0). LEVER_EFFECTS = { "incomplete_mult_digital": 0.40, # app / web "incomplete_mult_assisted": 0.65, # phone / branch agent checklist "second_request_mult": 0.50, "phone_intake_extra_min": 1.5, "miscategorized_mult": 0.40, "queue_wait_mult": 0.75, "special_cause_factor": 1.4, "rekey_error_mult": 0.25, "chargeback_touch_saving_min": 4.0, } def _pick(weighted, u): total = 0.0 for item in weighted: total += item[1] if u < total: return item return weighted[-1] def _lognormal_hours(median, sigma, z): return median * math.exp(sigma * z) def _scaled(mult, effect): """Move a multiplier part of the way from 1.0 towards its full effect.""" return 1.0 - effect * (1.0 - mult) def draw_cases(seed=SEED, n=N_CASES): rng = random.Random(seed) draws = [] for i in range(n): d = {"case_no": i + 1} for key in ["channel", "type", "tenure", "day", "hour", "incomplete", "second_doc", "miscat", "withdraw", "rekey", "outcome", "unreachable", "merchant_days"]: d["u_" + key] = rng.random() for key in ["amount", "intake_touch", "triage_wait", "triage_touch", "doc_wait", "doc_wait2", "cust_wait", "cust_wait2", "inv_wait", "inv_touch", "retriage_wait", "reinv_wait", "credit_wait", "cb_wait", "cb_touch", "refile_wait", "final_wait", "final_touch", "retry_wait", "ft_wait", "ft_final_wait"]: d["z_" + key] = rng.gauss(0.0, 1.0) draws.append(d) return draws def run(draws, levers=None, effect=1.0): """Return (disputes, steps, chargebacks) for one scenario.""" levers = dict(CURRENT_STATE, **(levers or {})) e = LEVER_EFFECTS disputes, steps, chargebacks = [], [], [] step_id = 0 for d in draws: case_id = f"DSP-{d['case_no']:05d}" channel = _pick(CHANNELS, d["u_channel"])[0] dtype, _, amt_median, amt_sigma, inv_median, win_rate = _pick(DISPUTE_TYPES, d["u_type"]) amount = round(max(5.0, amt_median * math.exp(amt_sigma * 0.6 * d["z_amount"])), 2) tenure_years = round(d["u_tenure"] * 15, 1) opened = datetime.combine(START, datetime.min.time()) + timedelta( days=int(d["u_day"] * WEEKS * 7), hours=8 + int(d["u_hour"] * 10)) week_no = (opened.date() - START).days // 7 + 1 surge = SPECIAL_CAUSE_FACTOR if week_no == SPECIAL_CAUSE_WEEK else 1.0 if levers["visual_management"] and surge > 1.0: surge = 1.0 + effect * (e["special_cause_factor"] - 1.0) + (1 - effect) * (SPECIAL_CAUSE_FACTOR - 1.0) queue_mult = _scaled(e["queue_wait_mult"], effect) if levers["visual_management"] else 1.0 incomplete_rate = INCOMPLETE_RATE[channel] second_rate = SECOND_DOC_REQUEST_RATE if levers["digital_intake"]: mult = e["incomplete_mult_digital"] if channel in ("app", "web") else e["incomplete_mult_assisted"] incomplete_rate *= _scaled(mult, effect) second_rate *= _scaled(e["second_request_mult"], effect) miscat_rate = MISCATEGORIZED_RATE[channel] if levers["assisted_triage"]: miscat_rate *= _scaled(e["miscategorized_mult"], effect) rekey_rate = REKEY_ERROR_RATE if levers["prefilled_chargeback"]: rekey_rate *= _scaled(e["rekey_error_mult"], effect) incomplete = d["u_incomplete"] < incomplete_rate second_doc = incomplete and d["u_second_doc"] < second_rate miscategorized = d["u_miscat"] < miscat_rate withdrawn = d["u_withdraw"] < WITHDRAWN_RATE rekey_error = d["u_rekey"] < rekey_rate unreachable = d["u_unreachable"] < UNREACHABLE_RATE fast_track = (levers["fast_track"] and amount < FAST_TRACK_LIMIT and dtype != "unauthorized_fraud" and tenure_years >= 1.0 and not withdrawn) case_steps = [] def add(step, team, wait_type, wait_hours, touch_minutes, rework_reason=None): case_steps.append({ "step_name": step, "team_id": team, "wait_type": wait_type, "wait_hours": round(max(0.0, wait_hours), 2), "touch_minutes": round(max(0.0, touch_minutes), 1), "is_rework": 1 if rework_reason else 0, "rework_reason": rework_reason or "", }) intake_touch = 9 + 2 * d["z_intake_touch"] if levers["digital_intake"] and channel in ("phone", "branch"): intake_touch += e["phone_intake_extra_min"] * effect add("intake", "contact_centre", "none", 0, max(3, intake_touch)) if fast_track: add("triage", "disputes_triage", "internal_queue", _lognormal_hours(2, 0.5, d["z_ft_wait"]), 2) add("final_decision_notification", "customer_resolution", "internal_queue", _lognormal_hours(4, 0.5, d["z_ft_final_wait"]), 5) outcome = "fast_track_credit" else: add("triage", "disputes_triage", "internal_queue", _lognormal_hours(20, 0.6, d["z_triage_wait"]) * queue_mult, max(2, 6 + 1.5 * d["z_triage_touch"])) if incomplete: add("document_request", "disputes_ops", "internal_queue", _lognormal_hours(8, 0.5, d["z_doc_wait"]) * queue_mult, 7) add("await_customer_documents", "customer", "customer", _lognormal_hours(96, 0.5, d["z_cust_wait"]), 0) if second_doc: add("document_request", "disputes_ops", "internal_queue", _lognormal_hours(8, 0.5, d["z_doc_wait2"]) * queue_mult, 7, "incomplete_documents") add("await_customer_documents", "customer", "customer", _lognormal_hours(72, 0.5, d["z_cust_wait2"]), 0, "incomplete_documents") if withdrawn: outcome = "withdrawn" else: inv_touch = inv_median * math.exp(0.3 * d["z_inv_touch"]) add("investigation", "dispute_analysts", "internal_queue", _lognormal_hours(40, 0.6, d["z_inv_wait"]) * queue_mult * surge, inv_touch) if miscategorized: add("triage", "disputes_triage", "internal_queue", _lognormal_hours(12, 0.5, d["z_retriage_wait"]) * queue_mult, 6, "wrong_category") add("investigation", "dispute_analysts", "internal_queue", _lognormal_hours(24, 0.6, d["z_reinv_wait"]) * queue_mult * surge, 0.6 * inv_touch, "wrong_category") add("provisional_credit_decision", "dispute_analysts", "internal_queue", _lognormal_hours(6, 0.5, d["z_credit_wait"]) * queue_mult, 4) cb_touch = 14 + 3 * d["z_cb_touch"] if levers["prefilled_chargeback"]: cb_touch -= e["chargeback_touch_saving_min"] * effect add("chargeback_filing", "chargeback_team", "internal_queue", _lognormal_hours(24, 0.6, d["z_cb_wait"]) * queue_mult, max(4, cb_touch)) if rekey_error: add("chargeback_filing", "chargeback_team", "internal_queue", _lognormal_hours(16, 0.5, d["z_refile_wait"]) * queue_mult, 10, "rekeying_error") low, high = (5, 30) if dtype == "unauthorized_fraud" else (8, 45) merchant_days = low + d["u_merchant_days"] * (high - low) add("await_merchant_response", "network_merchant", "merchant_network", merchant_days * 24, 0) outcome = "cardholder_favour" if d["u_outcome"] < win_rate else "merchant_favour" chargebacks.append({ "case_id": case_id, "merchant_response_days": round(merchant_days, 1), "network_outcome": "accepted" if outcome == "cardholder_favour" else "represented", "refiled_after_error": 1 if rekey_error else 0, }) add("final_decision_notification", "customer_resolution", "internal_queue", _lognormal_hours(16, 0.6, d["z_final_wait"]) * queue_mult, max(4, 9 + 2 * d["z_final_touch"])) if unreachable: add("final_decision_notification", "customer_resolution", "internal_queue", _lognormal_hours(24, 0.4, d["z_retry_wait"]), 5, "customer_unreachable") clock = opened for seq, s in enumerate(case_steps, start=1): step_id += 1 start = clock + timedelta(hours=s["wait_hours"]) end = start + timedelta(minutes=s["touch_minutes"]) steps.append(dict(s, step_event_id=step_id, case_id=case_id, step_seq=seq, started_at=start.isoformat(timespec="minutes"), completed_at=end.isoformat(timespec="minutes"))) clock = end disputes.append({ "case_id": case_id, "opened_at": opened.isoformat(timespec="minutes"), "week_no": week_no, "channel": channel, "dispute_type": dtype, "amount_cad": amount, "customer_tenure_years": tenure_years, "intake_complete": 0 if (incomplete and not fast_track) else 1, "initially_miscategorized": 1 if (miscategorized and not fast_track and not withdrawn) else 0, "fast_tracked": 1 if fast_track else 0, "outcome": outcome, "closed_at": clock.isoformat(timespec="minutes"), }) return disputes, steps, chargebacks