Wiki:Packs/Migration Service-Level Simulation/Modules
From WFM Labs
Modules for the Migration Service-Level Simulation pack (CP-WFM-019). Save each block under the filename in its heading, keep both files in one folder (the intake imports the simulator), and upload them as project knowledge with the rest of the pack. Dependencies: numpy, pandas, scipy, openpyxl; optionally pyyaml and matplotlib.
The pack page carries the method, the levers, the interpretation and a worked example; this page carries the intake schema and the simulator they describe. Read Wiki:Packs/Migration Service-Level Simulation first.
Block 4 — 04-intake-schema.md
# 04 — Intake schema and the gap register
*Figures marked `[estimated]` are judgment, not measurement.*
Migration models are usually built in whatever workbook the program office already has: one
tab listing clients with their contracts and volumes, one tab listing teams and headcount, and
one tab reconciling the end state. This module turns that shape into a simulator configuration
**without silently assuming anything**. Every default applied to a client becomes a row in the
gap register, weighted by the share of volume it touches, so the room can see which
assumptions carry the answer.
Three inputs. Two carry observations, one carries the model.
| File | One row per | Required |
|---|---|---|
| Client table (`.xlsx` or `.csv`) | client | yes |
| Staffing roster (`.xlsx` or `.csv`) | team × role × location | recommended |
| `params.yaml` | — | yes: pools, levers, world, and an `intake:` block |
Column names do not need to match. The `intake:` block maps your headers onto the canonical
names below, so the source workbook is never edited.
## Client table
| Canonical column | Type | Required | Notes |
|---|---|---|---|
| `client_id` | string | **yes** | Unique; a duplicate id is a blocking gap |
| `pool` | string | one of these two | The queue that serves the client, named as in `params.pools` |
| `team_type` | string | | Mapped to a pool through `intake.pool_map`. Service-model labels such as "dedicated", "shared" or "integrated" usually live here |
| `annual_volume` | number | one of these two | Transactions or contacts per year |
| `volume_ytd` | number | | Year-to-date. Annualised as `× 12 / intake.months_ytd` |
| `adoption` | share | no | Self-service or online share. With `use_offline_share`, only `1 − adoption` of volume reaches agents. Accepts `0.33`, `"0.33"`, `33`, `"33"` or `"33%"` (up to 1 is a fraction, above 1 a percentage); above 100% is capped and flagged |
| `health` | text | no | Relationship health. Classified by first letter: G / A / R |
| `contract_type` | text | no | Classified by `intake.contract_rules` into `fixed`, `evergreen` or `evergreen_tfc` |
| `expiry_date` | date | for fixed | Converted to a week relative to `intake.start_date`. Text dates like `05/03/2027` are read month-first unless `intake.dayfirst: true`; ambiguous ones get a `check` row |
| `fte` | number | no | FTE currently serving the client. Used for reconciliation and for carving dedicated clients into their own pools |
| `wave` | string | no | Overrides the pool's default wave |
| `p_transfer`, `p_exit`, `p_replatform` | probability | no | All three or none. Override the health prior when an account owner has a view |
Anything else in the file (revenue, segment, booking channel, location) is ignored by the
model but kept in the source. Revenue in particular is the right lens for *which* clients to
fight for, but it is not a service input.
## Staffing roster
| Canonical column | Type | Required | Notes |
|---|---|---|---|
| `role` | text | **yes** | Only roles listed in `intake.frontline_roles` are simulated |
| `heads` | number | **yes** | Heads today |
| `pool` or `team_type` | text | yes | As for clients |
| `team`, `location`, `client` | text | no | Used to detect duplicate rows |
**Frontline versus overhead.** Team leaders, support functions and management belong in the
program's headcount waterfall, but they do not answer contacts. The intake simulates frontline
roles only and reports every other role on an *Overhead roles* sheet, so the totals still
reconcile to the program's population without inflating capacity.
## Validation before any run
| Check | Severity | What it catches |
|---|---|---|
| Unmapped team type or unknown pool | **block** | Work with nobody to serve it |
| Blank or non-numeric volume | check | A client that silently carries no work |
| Roster rows sharing team / role / location | note | Double counting: the most common workbook error in headcount reconciliations |
| Roster frontline vs summed client FTE, per pool, differs by more than 10% | check | Either the roster includes non-frontline roles or client FTE allocations are stale |
| Week-0 utilisation outside 0.50–0.95 | check | Almost always a units problem: transactions entered as contacts, or annual volume treated as year-to-date |
A **block** stops the run. Everything else runs and is reported.
## The definitions that decide the answer
- **Transactions are not contacts.** Client tables usually count transactions (bookings,
orders, cases). Interactive pools need contacts. Set `contacts_per_unit` on each interactive
pool in `params.yaml`. It is removed from the config before the run. It defaults to 1.0,
which is almost never true, so measure it.
- **Offline share.** If `adoption` means "share of transactions completed without an agent",
keep `use_offline_share: true`. If the volume column is already agent-handled work, set it to
`false` or adoption will be subtracted twice.
- **Headcount is heads, not budget.** Use who is on the roll today. Vacancies are not supply.
- **Expired fixed-term contracts** are treated as rolling (`evergreen`). A contract that ended
before the start date and is still trading is on some form of tacit renewal.
- **Dedicated clients.** A client served by its own team is its own queue. Pooling it with
other dedicated clients overstates the pooling benefit. With `split_dedicated`, any client
in the template pool with at least `min_fte` FTE is carved into a pool of its own, sized to
its FTE; the template pool's headcount (roster or params) is reduced by the same FTE, and any
overflow capacity stays with the template pool. On the demonstration inputs, carving out a 12-FTE client dropped that client's
week-0 service level from about 0.85 to about 0.65 at the *same* 74% utilisation. The risk
was there all along; the combined pool hid it.
## The gap register
Every rule the intake applies to fill a hole is logged:
| Column | Meaning |
|---|---|
| `field` | The input that was missing or unreadable |
| `rule_applied` | What the intake did instead |
| `clients` | How many clients it touched |
| `volume_share` | The share of total volume those clients carry: the number that matters |
| `severity` | `block`, `check` or `assumed` |
| `examples` | Up to five client IDs, to start the chase |
Sort by `volume_share`. A default applied to 1% of volume is housekeeping. A default applied
to 30% of volume is the answer. Close the largest gaps first, and do not present results while
a gap above about 10% of volume is still open `[estimated]`.
## Reconciling to the program's plan
Programs usually carry an end-state waterfall: start, minus attrition, minus transfers, minus
releases. Put its numbers in `intake.plan_waterfall` and the intake compares them with the
simulated mean flows, per pool or in total. Large differences mean one of two things, and
both are worth knowing:
- **The plan and the model disagree on a mechanism.** For example, the plan nets all natural
attrition against the pool being released, whereas the model lets attrition fall on everyone,
transferees included.
- **The plan is a point and the model is a distribution.** The waterfall's attrition range
(P10–P90) shows how much of the plan's single number is luck.
## `params.yaml` template
```yaml
horizon_weeks: 52
draws: 2000
calendar:
paid_hours_week: 37.5
open_hours_week: 55
interval_seconds: 1800
schedule_fit: 0.85
buckets: {volume_share: [0.45, 0.40, 0.15], hour_share: [0.30, 0.45, 0.25]}
seasonality: null
service: # interactive pools; any pool can override with service_model, patience_seconds, ...
model: A # A: callers abandon (Erlang A). C: nobody abandons (Erlang C, version 1.0)
patience_seconds: 120 # [estimated] MEASURE THIS: average queue time of all callers / abandon rate
redial_rate: 0.4 # [estimated] share of extra abandonment that calls back next week
abandon_cap: 0.10 # abandon rate that counts as a breach in the KPIs
freeze: {start_week: 0, end_week: [8, 14, 26]}
attrition:
base_annual: 0.15
freeze_multiplier: [1.0, 1.4, 2.2]
post_multiplier_release: [1.5, 3.0, 5.0]
post_multiplier_transfer: [0.8, 1.2, 2.0]
retention_effect: 0.0
shrinkage: {post_announcement_add: [0.0, 0.04, 0.10]}
release: {notice_weeks: 6, lag_weeks: 2, buffer: 0.05, allowed: true}
scoring: {min_share: 0.10}
destination: {backfill: true, top_up_to_plan: true}
transfer:
synergy: 0.25
synergy_realisation: [0.4, 0.8, 1.0]
acceptance: 1.0
training_hours: 24
training_weeks: 6
aht_uplift: [1.03, 1.10, 1.25]
ramp_weeks: 12
waves: {W1: 6, W2: 12, W3: 18}
exit: {notice_weeks: {evergreen: [8, 13, 26], evergreen_tfc: [4, 6, 13]}}
replatform: {offset_weeks: [4, 8, 16]}
fate_priors:
green: {transfer: 0.85, exit: 0.10, replatform: 0.05}
amber: {transfer: 0.65, exit: 0.25, replatform: 0.10}
red: {transfer: 0.35, exit: 0.55, replatform: 0.10}
pools:
DED: # headcount omitted: taken from the roster
mode: interactive
shrinkage: 0.32
aht_seconds: 540
asa_seconds: 60
contacts_per_unit: 1.0 # MEASURE THIS
service_floor: 0.70
wave: W1
volume_error: [0.92, 1.08]
aht_error: [0.95, 1.05]
SHR: {mode: interactive, shrinkage: 0.32, aht_seconds: 480, asa_seconds: 60,
contacts_per_unit: 1.0, service_floor: 0.70, wave: W2}
# a chat pool sets channel: chat and usually its own patience, e.g. {..., channel: chat, patience_seconds: 300};
# the shape-card exporter then carries chat handle time and patience separately from voice
BO: {mode: deferrable, shrinkage: 0.30, tx_per_fte_year: 4400, target_days: 2,
service_floor: 0.80, wave: W3}
intake:
start_date: "2026-10-05" # week 0
months_ytd: 5
use_offline_share: true
client_columns: # canonical: header in your file
client_id: Account
team_type: Team Type
volume_ytd: Txn YTD
adoption: Adoption
health: Health
contract_type: Contract
expiry_date: Expiry
fte: FTE
roster_columns: {team: Team, team_type: Team Type, role: Role, location: Location, heads: Heads}
pool_map: {Dedicated: DED, Shared: SHR, Back office: BO}
frontline_roles: [Agent]
default_health: amber
split_dedicated: {template_pool: DED, min_fte: 12}
plan_waterfall: {TOTAL: {transferred: 80, released: 40, attrition: 15}}
# contract_rules: [[regex, fixed|evergreen|evergreen_tfc], ...] first match wins
```
## Running the intake
| Command | What happens |
|---|---|
| `python mig_intake.py --demo` | Writes synthetic `clients_demo.xlsx`, `roster_demo.xlsx` and `params_demo.yaml` in a realistic messy shape, then resolves and runs them |
| `python mig_intake.py clients.xlsx roster.xlsx params.yaml` | Resolve, report, run |
| `... --no-run` | Resolve and report only: use this first on real data |
Outputs: `config_resolved.yaml` (the exact configuration run), `intake_report.xlsx` (Notes,
Gap register, Headcount, Baseline, Overhead roles, Roster duplicates, Plan check) and, when
it runs, `migration_sl_results.xlsx`, the simulator's workbook from `05-simulator.md`.
The intake imports `mig_sl_sim.py`, so keep both files in the same folder.
---
## `mig_intake.py`
```python
"""Intake: turn a client table, a staffing roster and params.yaml into a simulator config.
python mig_intake.py clients.xlsx roster.xlsx params.yaml # resolve, report, run
python mig_intake.py clients.csv roster.csv params.yaml --no-run # resolve and report only
python mig_intake.py --demo # write synthetic inputs, then run them
Writes `config_resolved.yaml` (the exact config the simulator ran) and `intake_report.xlsx`
(gap register, headcount reconciliation, baseline utilisation, overhead roles, plan check).
Nothing is silently assumed: every default applied to a client is a row in the gap register.
"""
from __future__ import annotations
import copy
import re
import sys
from dataclasses import dataclass
import numpy as np
import pandas as pd
import mig_sl_sim as sim
# First match wins; case-insensitive regex against the contract-type text.
DEFAULT_CONTRACT_RULES = [
[r"no\s+(termination\s+for\s+)?convenience", "fixed"],
[r"convenience|\btfc\b", "evergreen_tfc"],
[r"evergreen|rolling|auto.?renew", "evergreen"],
[r"expir|fixed|\bterm\b", "fixed"],
]
DEFAULT_HEALTH_RULES = [[r"^\s*g", "green"], [r"^\s*a", "amber"], [r"^\s*r", "red"]]
@dataclass
class IntakeReport:
gaps: pd.DataFrame
reconciliation: pd.DataFrame
baseline: pd.DataFrame
overhead: pd.DataFrame
duplicates: pd.DataFrame
notes: list
def ok(self):
return not (self.gaps["severity"] == "block").any() if len(self.gaps) else True
# --------------------------------------------------------------------------
# readers and parsers
# --------------------------------------------------------------------------
def read_table(path, sheet=None):
if str(path).lower().endswith((".xlsx", ".xlsm", ".xls")):
return pd.read_excel(path, sheet_name=sheet or 0)
return pd.read_csv(path)
def _rename(df, colmap):
"""colmap is canonical -> source column. Unmapped canonical names are used as-is."""
missing = [src for src in (colmap or {}).values() if src not in df.columns]
if missing:
raise ValueError(f"column_map names columns not in the file: {missing}")
return df.rename(columns={src: canon for canon, src in (colmap or {}).items()})
def _blank(v):
return v is None or (isinstance(v, float) and np.isnan(v)) or (isinstance(v, str) and not v.strip())
def parse_share(v):
"""0.33, '0.33', '33%', 33, '33' -> 0.33; '1' or 1 -> 1.0 (a value up to 1 is a fraction,
above 1 a percentage, '%' always a percentage). Blank -> None. Values above 100% are returned
as given (> 1) so the caller can flag them."""
if _blank(v):
return None
if isinstance(v, str):
t = v.strip()
pct = t.endswith("%")
t = t.rstrip("%").strip()
if not t:
return None
x = float(t)
return x / 100 if pct or x > 1 else x
x = float(v)
return x / 100 if x > 1 else x
_AMBIGUOUS = re.compile(r"^\s*(\d{1,2})[/.-](\d{1,2})[/.-](\d{2,4})\s*$")
def ambiguous_date(v):
"""True for text dates like 05/03/2027 that read differently day-first and month-first."""
if not isinstance(v, str):
return False
m = _AMBIGUOUS.match(v)
return bool(m) and int(m.group(1)) <= 12 and int(m.group(2)) <= 12 and m.group(1) != m.group(2)
def classify(text, rules):
if _blank(text):
return None
for pattern, label in rules:
if re.search(pattern, str(text), flags=re.I):
return label
return None
def to_week(date, start, dayfirst=False):
if _blank(date):
return None
d = pd.to_datetime(date, errors="coerce", dayfirst=dayfirst)
if pd.isna(d):
return None
return int((d - pd.Timestamp(start)).days // 7)
# --------------------------------------------------------------------------
# the resolver
# --------------------------------------------------------------------------
def build_config(clients_df, roster_df, params):
params = copy.deepcopy(params)
if "intake" not in params:
raise ValueError("params.yaml needs an 'intake' section (see 04-intake-schema.md)")
intake = params.pop("intake")
dayfirst = bool(intake.get("dayfirst", False))
start = intake["start_date"]
months = float(intake.get("months_ytd", 12))
pool_map = intake.get("pool_map", {})
contract_rules = intake.get("contract_rules", DEFAULT_CONTRACT_RULES)
health_rules = intake.get("health_rules", DEFAULT_HEALTH_RULES)
use_offline = bool(intake.get("use_offline_share", True))
gaps, notes = [], []
df = _rename(clients_df.copy(), intake.get("client_columns"))
if "client_id" not in df.columns:
raise ValueError("clients file needs a client_id column (map it in intake.client_columns)")
if "annual_volume" not in df.columns and "volume_ytd" not in df.columns:
raise ValueError("clients file needs annual_volume or volume_ytd")
total_units = None
def gap(field, rule, rows, severity="assumed"):
if len(rows):
share = float(df.loc[rows, "_units"].sum() / total_units) if total_units else np.nan
gaps.append({"field": field, "rule_applied": rule, "clients": len(rows),
"volume_share": share, "severity": severity,
"examples": ", ".join(map(str, df.loc[rows, "client_id"].head(5)))})
# volume first, so every later gap can be weighted by the work it touches
if "annual_volume" in df.columns:
units = pd.to_numeric(df["annual_volume"], errors="coerce")
else:
units = pd.to_numeric(df["volume_ytd"], errors="coerce") * 12 / months
notes.append(f"annualised volume_ytd x 12/{months:g}")
df["_units"] = units.fillna(0.0)
total_units = float(df["_units"].sum())
gap("volume", "blank or non-numeric -> 0 (client carries no work)", df.index[units.isna()], "check")
dups = df.index[df["client_id"].duplicated(keep=False)]
gap("client_id", "duplicate client_id; merge or rename the rows", dups, "block")
adoption = df["adoption"].map(parse_share) if "adoption" in df.columns else pd.Series(None, index=df.index, dtype=float)
if use_offline:
gap("adoption", "blank -> 0 (all volume assisted)", df.index[adoption.isna()])
gap("adoption", "above 100% -> capped at 100% (no assisted volume)", df.index[adoption > 1], "check")
df["_units"] = df["_units"] * (1 - adoption.fillna(0.0).clip(0, 1))
# pool
pool = df["pool"] if "pool" in df.columns else pd.Series(None, index=df.index, dtype=object)
if "team_type" in df.columns:
mapped = df["team_type"].map(lambda t: pool_map.get(str(t).strip()) if not _blank(t) else None)
pool = pool.where(~pool.map(_blank), mapped) if pool.notna().any() else mapped
unmapped = df.index[pool.map(_blank)]
if len(unmapped) and intake.get("default_pool"):
gap("pool", f"unmapped team type -> {intake['default_pool']}", unmapped)
pool = pool.where(~pool.map(_blank), intake["default_pool"])
elif len(unmapped):
gap("pool", "unmapped team type; add it to intake.pool_map", unmapped, "block")
df["_pool"] = pool
unknown = df.index[~df["_pool"].isin(params["pools"]) & ~df["_pool"].map(_blank)]
gap("pool", "pool name not defined in params.pools", unknown, "block")
# health and contract
health = df["health"].map(lambda v: classify(v, health_rules)) if "health" in df.columns else pd.Series(None, index=df.index)
dh = intake.get("default_health", "amber")
gap("health", f"blank or unrecognised -> {dh}", df.index[health.isna()])
df["_health"] = health.fillna(dh)
contract = df["contract_type"].map(lambda v: classify(v, contract_rules)) if "contract_type" in df.columns else pd.Series(None, index=df.index)
gap("contract_type", "blank or unrecognised -> evergreen", df.index[contract.isna()])
df["_contract"] = contract.fillna("evergreen")
expiry = df["expiry_date"].map(lambda d: to_week(d, start, dayfirst)) if "expiry_date" in df.columns else pd.Series(None, index=df.index)
fixed = df["_contract"] == "fixed"
if "expiry_date" in df.columns:
amb = df.index[fixed & df["expiry_date"].map(ambiguous_date)]
gap("expiry_date", f"day/month ambiguous text date read {'day' if dayfirst else 'month'}-first (intake.dayfirst)", amb, "check")
no_date = df.index[fixed & expiry.isna()]
gap("expiry_date", "fixed-term with no usable date -> evergreen", no_date)
lapsed = df.index[fixed & expiry.notna() & (expiry < 0)]
gap("expiry_date", "expired before start_date -> evergreen (rolling)", lapsed)
df.loc[no_date.union(lapsed), "_contract"] = "evergreen"
df["_expiry"] = expiry.where(df["_contract"] == "fixed")
# dedicated clients large enough to be their own queue
pools = copy.deepcopy(params["pools"])
fte = pd.to_numeric(df["fte"], errors="coerce") if "fte" in df.columns else pd.Series(np.nan, index=df.index)
split = intake.get("split_dedicated")
carved, carved_names = {}, set()
if split:
tmpl = split["template_pool"]
big = df.index[(df["_pool"] == tmpl) & (fte >= float(split.get("min_fte", 3)))]
for i in big:
name = f"{tmpl}-{df.at[i, 'client_id']}"
spec = {k: v for k, v in copy.deepcopy(pools[tmpl]).items() if not k.startswith("overflow_")}
pools[name] = {**spec, "headcount": float(fte[i])} # overflow stays with the template pool
carved[tmpl] = carved.get(tmpl, 0.0) + float(fte[i])
df.at[i, "_pool"] = name
carved_names.add(name)
if len(big):
notes.append(f"split {len(big)} dedicated client(s) out of {tmpl} into their own pools")
# headcount: roster frontline heads by pool, else summed client FTE
overhead = pd.DataFrame()
duplicates = pd.DataFrame()
roster_heads = {}
if roster_df is not None:
rs = _rename(roster_df.copy(), intake.get("roster_columns"))
for col in ("role", "heads"):
if col not in rs.columns:
raise ValueError(f"roster needs a '{col}' column (map it in intake.roster_columns)")
key = [c for c in ("team", "role", "location", "client") if c in rs.columns]
dup = rs.duplicated(subset=key, keep=False) if key else pd.Series(False, index=rs.index)
duplicates = rs[dup]
if len(duplicates):
notes.append(f"{len(duplicates)} roster rows share team/role/location: check for double counting")
rp = rs["pool"] if "pool" in rs.columns else rs.get("team_type", pd.Series(None, index=rs.index)).map(
lambda t: pool_map.get(str(t).strip()) if not _blank(t) else None)
rs["_pool"] = rp
front = rs["role"].astype(str).str.strip().isin(intake.get("frontline_roles", []))
overhead = rs[~front].groupby("role", dropna=False)["heads"].sum().reset_index()
roster_heads = rs[front & rs["_pool"].notna()].groupby("_pool")["heads"].sum().to_dict()
lost = rs[front & rs["_pool"].isna()]
if len(lost):
notes.append(f"{float(lost['heads'].sum()):g} frontline roster heads have no pool mapping and are excluded")
recon = []
client_fte = df.assign(_fte=fte).groupby("_pool")["_fte"].sum(min_count=1).to_dict()
for p, spec in pools.items():
fixed_hc = spec.get("headcount")
r = roster_heads.get(p)
if r is not None and p in carved:
r = r - carved[p]
c = client_fte.get(p)
if p in carved_names:
r = None
if fixed_hc is None:
if r is not None:
spec["headcount"], source = float(r), "roster"
elif c is not None and not np.isnan(c):
spec["headcount"], source = float(c), "client FTE"
else:
raise ValueError(f"pool {p}: no headcount in params, roster or client FTE")
elif p in carved_names:
source = "client FTE (carved)"
elif p in carved:
spec["headcount"], source = float(fixed_hc) - carved[p], "params less carved FTE"
if spec["headcount"] <= 0:
raise ValueError(f"pool {p}: carved dedicated FTE ({carved[p]:g}) exceeds its headcount ({fixed_hc})")
else:
source = "params"
diff = (r - c) / c if (r is not None and c and not np.isnan(c)) else np.nan
recon.append({"pool": p, "headcount_used": spec["headcount"], "source": source,
"roster_frontline": r, "client_fte_sum": c, "roster_vs_clients": diff,
"flag": "check" if not np.isnan(diff) and abs(diff) > 0.10 else ""})
# clients -> config rows, converting units to contacts where a pool is interactive
clients = []
bad_wave, partial_p = [], []
for i, row in df.iterrows():
if _blank(row["_pool"]) or row["_pool"] not in pools:
continue
cpu = float(pools[row["_pool"]].get("contacts_per_unit", 1.0))
c = {"id": str(row["client_id"]), "pool": row["_pool"],
"annual_volume": float(row["_units"]) * cpu,
"health": row["_health"], "contract": row["_contract"],
"expiry_week": None if pd.isna(row["_expiry"]) else int(row["_expiry"])}
if "wave" in df.columns and not _blank(row["wave"]):
if str(row["wave"]) in params["waves"]:
c["wave"] = str(row["wave"])
else:
bad_wave.append(i)
probs = [row.get(f"p_{f}") for f in sim.FATES]
if all(not _blank(v) for v in probs):
c.update({f"p_{f}": float(v) for f, v in zip(sim.FATES, probs)})
elif any(not _blank(v) for v in probs):
partial_p.append(i)
clients.append(c)
gap("wave", "wave name not in waves; add it or clear the cell", pd.Index(bad_wave), "block")
gap("p_transfer/p_exit/p_replatform", "incomplete override -> health prior used", pd.Index(partial_p), "check")
cfg = {**params, "pools": pools, "clients": clients}
for p in pools.values():
p.pop("contacts_per_unit", None)
if not any(g["severity"] == "block" for g in gaps):
sim.validate(cfg)
baseline = baseline_utilisation(cfg)
rep = IntakeReport(pd.DataFrame(gaps, columns=["field", "rule_applied", "clients", "volume_share", "severity", "examples"]),
pd.DataFrame(recon), baseline, overhead, duplicates, notes)
return cfg, rep
def baseline_utilisation(cfg):
"""Week-0 utilisation per pool, before any simulation. Outside 0.5-0.95 usually means a units problem."""
cal = cfg["calendar"]
rows = []
for p, spec in cfg["pools"].items():
vol = sum(c["annual_volume"] for c in cfg["clients"] if c["pool"] == p) / 52
prod = spec["headcount"] * (1 - spec["shrinkage"])
if spec["mode"] == "interactive":
util = vol * spec["aht_seconds"] / 3600 / max(prod * cal["paid_hours_week"], 1e-9)
else:
util = vol / max(spec["headcount"] * spec["tx_per_fte_year"] / 52, 1e-9)
flag = "" if 0.5 <= util <= 0.95 else ("over capacity" if util > 1 else "check units")
rows.append({"pool": p, "mode": spec["mode"], "headcount": spec["headcount"],
"clients": sum(c["pool"] == p for c in cfg["clients"]),
"weekly_volume": vol, "utilisation": util, "flag": flag})
return pd.DataFrame(rows)
def compare_to_plan(res, plan):
"""Simulated mean waterfall against a plan's end-state numbers. plan: {pool|TOTAL: {flow: value}}."""
wf = sim.waterfall(res).set_index("pool")
wf["attrition"] = wf["attrition_freeze"] + wf["attrition_post"]
rows = []
for p, flows in (plan or {}).items():
for k, v in flows.items():
got = float(wf.at[p, k]) if p in wf.index and k in wf.columns else np.nan
rows.append({"pool": p, "flow": k, "plan": v, "simulated_mean": got, "difference": got - v})
return pd.DataFrame(rows)
def write_report(rep, path, plan_check=None):
with pd.ExcelWriter(path, engine="openpyxl") as xw:
pd.DataFrame({"note": rep.notes or ["none"]}).to_excel(xw, sheet_name="Notes", index=False)
rep.gaps.to_excel(xw, sheet_name="Gap register", index=False)
rep.reconciliation.to_excel(xw, sheet_name="Headcount", index=False)
rep.baseline.to_excel(xw, sheet_name="Baseline", index=False)
rep.overhead.to_excel(xw, sheet_name="Overhead roles", index=False)
rep.duplicates.to_excel(xw, sheet_name="Roster duplicates", index=False)
if plan_check is not None:
plan_check.to_excel(xw, sheet_name="Plan check", index=False)
return path
# --------------------------------------------------------------------------
# demonstration inputs, written the way a real workbook arrives
# --------------------------------------------------------------------------
def write_demo_inputs(outdir="."):
"""Synthetic client + roster files with messy real-world labels, plus a params.yaml."""
import yaml
base = sim.demo_config()
rng = np.random.default_rng(11)
team_of = {"DED": "Dedicated", "SHR": "Shared", "BO": "Back office"}
contract_text = {"fixed": "Contract with expiration date", "evergreen": "Evergreen contract",
"evergreen_tfc": "Evergreen contract (inc termination for convenience clause)"}
rows = []
for c in base["clients"]:
adopt = float(rng.choice([0, 0.2, 0.35, 0.5]))
ytd = c["annual_volume"] / (1 - adopt) * 5 / 12
expiry = (pd.Timestamp("2026-10-05") + pd.Timedelta(weeks=c["expiry_week"])).date() if c["expiry_week"] is not None else None
rows.append({"Account": c["id"], "Team Type": team_of[c["pool"]], "Txn YTD": round(ytd),
"Adoption": f"{adopt:.0%}", "Health": c["health"].title(),
"Contract": contract_text[c["contract"]], "Expiry": expiry,
"FTE": round(base["pools"][c["pool"]]["headcount"] * c["annual_volume"]
/ sum(x["annual_volume"] for x in base["clients"] if x["pool"] == c["pool"]), 1)})
clients = pd.DataFrame(rows)
roster = pd.DataFrame([
{"Team": "T-DED", "Team Type": "Dedicated", "Role": "Agent", "Location": "Site 1", "Heads": 40},
{"Team": "T-SHR", "Team Type": "Shared", "Role": "Agent", "Location": "Site 1", "Heads": 45},
{"Team": "T-SHR", "Team Type": "Shared", "Role": "Agent", "Location": "Site 2", "Heads": 25},
{"Team": "T-BO", "Team Type": "Back office", "Role": "Agent", "Location": "Site 1", "Heads": 25},
{"Team": "T-SHR", "Team Type": "Shared", "Role": "Team leader", "Location": "Site 1", "Heads": 7},
{"Team": "T-DED", "Team Type": "Dedicated", "Role": "Team leader", "Location": "Site 1", "Heads": 4},
{"Team": "Support", "Team Type": "", "Role": "Support", "Location": "Site 1", "Heads": 12},
])
params = {k: v for k, v in base.items() if k != "clients"}
for p in params["pools"].values():
p.pop("headcount")
params["intake"] = {
"start_date": "2026-10-05", "months_ytd": 5, "use_offline_share": True,
"client_columns": {"client_id": "Account", "team_type": "Team Type", "volume_ytd": "Txn YTD",
"adoption": "Adoption", "health": "Health", "contract_type": "Contract",
"expiry_date": "Expiry", "fte": "FTE"},
"roster_columns": {"team": "Team", "team_type": "Team Type", "role": "Role",
"location": "Location", "heads": "Heads"},
"pool_map": {"Dedicated": "DED", "Shared": "SHR", "Back office": "BO"},
"frontline_roles": ["Agent"],
"default_health": "amber",
"split_dedicated": {"template_pool": "DED", "min_fte": 12},
"plan_waterfall": {"TOTAL": {"transferred": 80, "released": 40, "attrition": 15}},
}
paths = (f"{outdir}/clients_demo.xlsx", f"{outdir}/roster_demo.xlsx", f"{outdir}/params_demo.yaml")
clients.to_excel(paths[0], index=False)
roster.to_excel(paths[1], index=False)
with open(paths[2], "w") as f:
yaml.safe_dump(params, f, sort_keys=False)
return paths
def main(clients_path, roster_path, params_path, do_run=True, draws=None):
import yaml
with open(params_path) as f:
params = yaml.safe_load(f)
plan = params.get("intake", {}).get("plan_waterfall")
cfg, rep = build_config(read_table(clients_path), read_table(roster_path) if roster_path else None, params)
with open("config_resolved.yaml", "w") as f:
yaml.safe_dump(cfg, f, sort_keys=False)
pd.set_option("display.width", 160)
print("NOTES:", *rep.notes, sep="\n ")
print("\nGAP REGISTER\n", rep.gaps.round(3).to_string(index=False) if len(rep.gaps) else " none")
print("\nHEADCOUNT\n", rep.reconciliation.round(2).to_string(index=False))
print("\nBASELINE\n", rep.baseline.round(3).to_string(index=False))
if not rep.ok():
print("\nBLOCKED: resolve the 'block' rows in the gap register before running.")
write_report(rep, "intake_report.xlsx")
return cfg, rep, None
res, check = None, None
if do_run:
res = sim.run(cfg, draws)
check = compare_to_plan(res, plan) if plan else None
print("\nKPIs\n", sim.kpis(res).round(3).to_string(index=False))
print("\nWATERFALL (means)\n", sim.waterfall(res).round(1).to_string(index=False))
if check is not None:
print("\nPLAN CHECK\n", check.round(1).to_string(index=False))
sim.to_excel(res, "migration_sl_results.xlsx")
write_report(rep, "intake_report.xlsx", check)
return cfg, rep, res
if __name__ == "__main__":
a = sys.argv[1:]
draws = int(a[a.index("--draws") + 1]) if "--draws" in a else None
if "--demo" in a:
main(*write_demo_inputs("."), do_run="--no-run" not in a, draws=draws or 1000)
else:
files = [x for x in a if not x.startswith("--") and not x.isdigit()]
if len(files) == 2:
files = [files[0], None, files[1]]
main(*files[:3], do_run="--no-run" not in a, draws=draws)
```
Block 5, part 1 of 2 — 05-simulator.md
Paste the 2 parts one after another, in order, into a single file named 05-simulator.md.
# 05 — Simulator
The whole model in one file: distribution helpers, vectorised Erlang C and exact Erlang A
(M/M/N+M, with redials), the weekly simulation, summaries, scenario runner, Excel export and
charts. The method it implements is
`02-method.md`; every parameter it reads is documented in `03-levers-and-parameters.md`.
Save the fenced block below as `mig_sl_sim.py`. Dependencies: `numpy`, `pandas`, `scipy`,
`openpyxl`. Optional: `pyyaml` to read a YAML configuration, `matplotlib` for charts. All
are present in Claude's analysis environment.
## Running it
| Command | What happens |
|---|---|
| `python mig_sl_sim.py` | Self-test on the synthetic demonstration configuration |
| `python mig_sl_sim.py params.yaml` | Run a real configuration (schema in `04-intake-schema.md`) |
| `... --draws 4000 --xlsx results.xlsx` | More draws; name the workbook |
From Python:
```python
import mig_sl_sim as m
cfg = m.demo_config() # or m.load_config("params.yaml")
res = m.run(cfg) # one Monte Carlo run
m.kpis(res) # trough, P(breach), weeks below floor, peak utilisation, peak abandonment
m.waterfall(res) # start -> attrition, transferred, released, remaining
m.weekly(res) # P10/P50/P90 service and abandonment per pool, side and week
sc = m.scenarios(cfg, {"overflow": {"pools.SHR.overflow_fte": 10},
"erlang_c": {"service.model": "C"}}) # compare with nobody abandoning
m.erlang_a(agents, load, aht / patience, threshold / aht) # (SL, abandon rate) for any arrays
m.to_excel(res, "results.xlsx", sc) # the file colleagues open
m.charts(res) # one fan chart per pool
```
## The Excel output
| Sheet | Contents |
|---|---|
| About | Draws, horizon, median freeze end, which service metric and abandonment model each mode uses |
| KPIs | One row per pool and side, including the service model, peak abandonment (P50, P90) and the probability and weeks of abandonment above the cap |
| Waterfall | Mean headcount flows per pool (attrition in the freeze, after it, transferred, released, remaining) that add up exactly, plus the attrition P10–P90 |
| Scenarios | The KPI table for each named scenario |
| Weekly | Long table of the weekly service and abandonment bands, redials, utilisation, heads and workload, ready for a pivot chart |
| Clients | Each client's inputs plus its share of draws by fate and median departure week |
| Inputs | Every parameter as run, flattened, so the file is self-documenting |
## Self-test
On the demonstration configuration the default run (Erlang A) prints a KPI table in which every
interactive source pool starts at or above 0.95, troughs between roughly 0.75 and 0.9 at P50,
and peaks at 10–15% abandonment. With `service.model: C` the same pools trough at roughly
0.1–0.3 and abandonment is zero. The deferrable pool reads the same under both. Then run the neutral case from `02-method.md`: the shared pool's source P50
trough must equal its week-0 value and no redials may appear. Each run takes a few seconds at
2,000 draws.
## Design notes
- **Common random numbers.** Every uncertain input consumes the same random draws whether it
is given as a range or a fixed value, so scenario differences are caused by the lever and
not by sampling noise. A lever that touches only one pool leaves the others identical to base.
- **Waves are offsets from the freeze end,** so a longer freeze slips every wave with it.
- **Releases follow the work,** never lead it: nobody is released before the dismissal notice
has run, or while their work is still in the building (plus the lag).
- **Erlang A in closed form.** The M/M/N+M sums are evaluated through the regularised incomplete
gamma function, so a whole draws × weeks grid is one vectorised call. Where that under- or
overflows (very large pools with very patient callers) the direct series is used instead. The
run is about twice as slow as Erlang C alone.
- **Redials are sequential.** A week's redials depend on the previous week's abandonment, so
weeks that receive redials are recomputed in order; weeks without redials are not.
- **Residual slivers are not scored.** A pool whose remaining work is below
`scoring.min_share` of its baseline is treated as wound down, so a single late-leaving client
does not register as a collapse of the whole pool.
---
## `mig_sl_sim.py`
```python
"""Migration service-level simulator: weekly Monte Carlo over a work-migration transition.
Work leaves a source site in steps (client transfers, exits, re-platforming) while the staff
serving it decay through attrition that cannot be backfilled during a consultation freeze.
Each draw simulates both sides on the same weekly grid, then converts the gap into service:
Erlang A (callers abandon, default) or Erlang C service level for interactive pools, and a
backlog timeliness index for deferrable ones.
python mig_sl_sim.py # self-test on demonstration data
python mig_sl_sim.py params.yaml # run a real configuration
python mig_sl_sim.py params.yaml --xlsx out.xlsx --draws 4000
python mig_sl_sim.py params.yaml --scenarios scenarios.yaml # {name: {dotted.key: value}} override sets
Dependencies: numpy, pandas, scipy, openpyxl. Optional: pyyaml (YAML configs), matplotlib (charts).
"""
from __future__ import annotations
import copy
import sys
import warnings
from dataclasses import dataclass
import numpy as np
import pandas as pd
from scipy import special, stats
SEED = 20260926
Z90 = 1.6448536269514722
FATES = ("transfer", "exit", "replatform")
T_TRANSFER, T_EXIT, T_REPLATFORM = 0, 1, 2
# --------------------------------------------------------------------------
# distribution helpers
# --------------------------------------------------------------------------
def pert(rng, spec, size):
"""PERT draw from [lo, mode, hi]; a scalar spec is a fixed value.
Always consumes the same random draws, so switching a lever between a range and a fixed
value leaves every other input's stream untouched (common random numbers across scenarios).
"""
u = rng.beta(2.0, 2.0, size)
if np.isscalar(spec):
return np.full(size, float(spec))
lo, mode, hi = map(float, spec)
if hi <= lo:
return np.full(size, lo)
a = 1 + 4 * (mode - lo) / (hi - lo)
b = 1 + 4 * (hi - mode) / (hi - lo)
return lo + (hi - lo) * stats.beta.ppf(stats.beta.cdf(u, 2.0, 2.0), a, b)
def lognormal_mult(rng, spec, size):
"""Multiplier whose 90% interval is [lo, hi]; None or a scalar is fixed."""
z = rng.standard_normal(size)
if spec is None:
return np.ones(size)
if np.isscalar(spec):
return np.full(size, float(spec))
lo, hi = map(float, spec)
mu = (np.log(lo) + np.log(hi)) / 2
sigma = (np.log(hi) - np.log(lo)) / (2 * Z90)
return np.exp(mu + sigma * z)
def losses(u, heads, p):
"""Binomial leavers from a (fractional) headcount, never more than are there.
`u` is a block of uniforms drawn once per week per pool (the same shape as `heads`), and the
binomial is inverted through its quantile function. The draws consumed never depend on the
headcount or the rate, so a lever that changes either leaves every other week's and pool's
attrition draw identical: common random numbers hold inside the loop, not only for the
per-draw drivers.
"""
n = np.maximum(np.floor(heads + 0.5), 0).astype(np.int64)
k = stats.binom.ppf(np.clip(u, 0, 1 - 1e-12), n, np.clip(p, 0, 1))
return np.minimum(np.nan_to_num(k), np.maximum(heads, 0))
# --------------------------------------------------------------------------
# Erlang C, vectorised, with fractional-agent interpolation
# --------------------------------------------------------------------------
def _sl_integer(n, a, aht, t):
n = n.astype(np.int64)
bn = np.ones(a.shape)
b = np.ones(a.shape)
for k in range(1, int(n.max(initial=0)) + 1):
b = a * b / (k + a * b)
hit = n == k
bn[hit] = b[hit]
out = np.zeros(a.shape)
ok = (n > a) & (n > 0)
c = n[ok] * bn[ok] / (n[ok] - a[ok] * (1 - bn[ok]))
out[ok] = 1 - c * np.exp(-(n[ok] - a[ok]) * t / aht[ok])
out[a <= 0] = 1.0
return np.clip(out, 0, 1)
def service_level(n, a, aht, t):
"""P(wait <= t) for n agents (fractional allowed) and a Erlangs of offered load."""
n = np.maximum(n, 0.0)
lo = np.floor(n)
w = n - lo
return (1 - w) * _sl_integer(lo, a, aht, t) + w * _sl_integer(lo + 1, a, aht, t)
# --------------------------------------------------------------------------
# Erlang A (M/M/N+M: waiting callers abandon), vectorised, exact
# --------------------------------------------------------------------------
#
# N agents, offered load A Erlangs, h = AHT / mean patience, tau = answer threshold / AHT.
# Relative to "all agents busy, nobody waiting": q_0 = 1, q_k = q_{k-1} A / (N + k h),
# free-agent states sum to 1/B - 1 (B = Erlang B), Z = 1/B - 1 + sum q_k. Then
# P(wait) = sum q_k / Z
# P(abandon) = sum q_k (k+1)h / (N + (k+1)h) / Z
# SL = answered within tau / ALL offered (abandoners count as misses)
# = [1/B - 1 + sum q_k N/(N+(k+1)h) (1 - F_k)] / Z, F_k a negative-binomial CDF.
# The sums have closed forms in the regularised incomplete gamma function (y = A/h, c = N/h,
# G = e^y y^-c Gamma(c+1), u0 = e^{-h tau}):
# sum q_k = G P(c, y); sum q_k N/(N+(k+1)h) = (N/A)(sum q_k - 1);
# N * (answered-in-time integral) = G (c/y) [P(c+1, y) - P(c+1, y u0)].
# That makes the whole grid one vectorised call. Where the closed form under- or overflows
# (huge pools with very patient callers), the direct series is used for those elements.
# The same maths as the WFM Labs Migration Scenario Modeler (src/lib/erlangA.ts); the
# series and closed form agree with it to about 1e-12. Very patient callers (h -> 0)
# reproduce Erlang C; very impatient ones reproduce Erlang B.
H_MIN = 1e-6 # below this AHT/patience ratio, callers are treated as infinitely patient (Erlang C)
def _erlang_c_metrics(n, a, t_over_aht):
"""Erlang C at integer n for the infinitely patient limit: (P(wait), abandon, SL)."""
ni = n.astype(np.int64)
bn = np.ones(a.shape)
b = np.ones(a.shape)
for k in range(1, int(ni.max(initial=0)) + 1):
b = a * b / (k + a * b)
bn[ni == k] = b[ni == k]
over = n <= a
with np.errstate(divide="ignore", invalid="ignore"):
pw = np.where(over, 1.0, n * bn / (n - a * (1 - bn)))
sl = np.where(over, 0.0, 1 - pw * np.exp(-(n - a) * t_over_aht))
ab = np.where(over, 1 - n / a, 0.0)
return pw, ab, sl
def _erlang_a_series(n, a, h, tau):
"""Direct series (the reference algorithm), vectorised with per-element stopping."""
B = np.ones(a.shape)
b = np.ones(a.shape)
ni = n.astype(np.int64)
for k in range(1, int(ni.max(initial=0)) + 1):
b = a * b / (k + a * b)
B[ni == k] = b[ni == k]
x = -np.expm1(-h * tau)
bb = n / h + 1
q = np.ones(a.shape)
Z0 = 1 / B - 1
Sw, Sab, Ssl = (np.zeros(a.shape) for _ in range(3))
tS, Fs = np.ones(a.shape), np.zeros(a.shape)
logS = -(n + h) * tau
kstar = np.maximum(0, (a - n) / h)
kmax = np.minimum(2_000_000, np.ceil(kstar + 9 * np.sqrt(a / h) + 200))
act = np.ones(a.shape, bool)
k = 0
while act.any():
Fs = np.where(act, Fs + tS, Fs)
cdf = np.maximum(0, 1 - Fs * np.exp(logS))
denom = n + (k + 1) * h
ps = n / denom
Sw = np.where(act, Sw + q, Sw)
Ssl = np.where(act, Ssl + q * ps * cdf, Ssl)
Sab = np.where(act, Sab + q * (1 - ps), Sab)
tS = np.where(act, tS * (x * (bb + k)) / (k + 1), tS)
big = act & (tS > 1e200)
tS, Fs = np.where(big, tS * 1e-200, tS), np.where(big, Fs * 1e-200, Fs)
logS = np.where(big, logS + 200 * np.log(10), logS)
r = a / denom
q = np.where(act, q * r, q)
with np.errstate(divide="ignore", invalid="ignore"):
done = (k + 1 > kstar) & (r < 1) & (q * r / (1 - r) < 1e-16 * (Z0 + Sw))
act &= ~done & (k + 1 <= kmax)
big = act & (q > 1e250)
for arr in (q, Z0, Sw, Ssl, Sab):
arr[big] *= 1e-250
k += 1
Z = Z0 + Sw
return np.minimum(1, Sw / Z), np.clip(Sab / Z, 0, 1), np.clip((Z0 + Ssl) / Z, 0, 1)
def _log_inv_b_minus_1(n, a):
"""log(1/B - 1) for Erlang B at integer n: log P(Poisson(a) <= n-1) - log P(Poisson(a) = n)."""
with np.errstate(divide="ignore", invalid="ignore"):
return np.log(special.gammaincc(n, a)) - (n * np.log(a) - a - special.gammaln(n + 1))
def erlang_a_int(n, a, h, tau, lr=None):
"""Exact Erlang A at integer agents n: (P(wait), abandon rate, SL). Arrays broadcast.
lr, if given, is log(1/B - 1) for (n, a); it lets callers reuse the Erlang B recurrence.
"""
n, a, h, tau = (np.asarray(v, float) for v in np.broadcast_arrays(n, a, h, tau))
pw, ab, sl = np.zeros(a.shape), np.zeros(a.shape), np.ones(a.shape)
empty = (a > 0) & (n <= 0)
pw[empty], ab[empty], sl[empty] = 1.0, 1.0, 0.0
live = (a > 0) & (n > 0)
pat = live & (h < H_MIN)
if pat.any():
pw[pat], ab[pat], sl[pat] = _erlang_c_metrics(n[pat], a[pat], tau[pat])
m = live & ~pat
if not m.any():
return pw, ab, sl
N, A, hh, tt = n[m], a[m], h[m], tau[m]
with np.errstate(divide="ignore", invalid="ignore", over="ignore", under="ignore"):
lr = _log_inv_b_minus_1(N, A) if lr is None else np.broadcast_to(lr, a.shape)[m]
y, c = A / hh, N / hh
logG = y - c * np.log(y) + special.gammaln(c + 1)
P = special.gammainc(c, y) # sum q_k = G P
logSw = logG + np.log(P)
ratio = np.exp(lr - logSw) # (1/B - 1) / sum q_k
inv_g = np.exp(-logG)
P1 = P - inv_g # P(c+1, y) = P(c, y) - y^c e^-y / Gamma(c+1)
yu = y * np.exp(-hh * tt)
dP = np.empty_like(P)
hi = P1 > 0.5 # difference of upper tails is the accurate one
dP[hi] = special.gammaincc(c[hi] + 1, yu[hi]) - (1 - P1[hi])
dP[~hi] = P1[~hi] - special.gammainc(c[~hi] + 1, yu[~hi])
o_pw = 1 / (1 + ratio)
o_ab = (1 - (N / A) * (1 - np.exp(-logSw))) / (1 + ratio)
o_sl = (ratio + (c / y) * dP / P) / (1 + ratio)
no_wait = lr > np.log(1e300) # B < 1e-300: nobody waits
o_pw[no_wait], o_ab[no_wait], o_sl[no_wait] = 0.0, 0.0, 1.0
bad = ~no_wait & ~(np.isfinite(o_pw) & np.isfinite(o_ab) & np.isfinite(o_sl) & (P > 0) & np.isfinite(logG))
if bad.any():
o_pw[bad], o_ab[bad], o_sl[bad] = _erlang_a_series(N[bad], A[bad], hh[bad], tt[bad])
pw[m], ab[m], sl[m] = np.clip(o_pw, 0, 1), np.clip(o_ab, 0, 1), np.clip(o_sl, 0, 1)
return pw, ab, sl
def erlang_a(n, a, h, tau):
"""Erlang A at fractional agents (linear between integers): (SL, abandon rate)."""
n, a, h, tau = (np.asarray(v, float) for v in np.broadcast_arrays(np.maximum(n, 0.0), a, h, tau))
sl, ab = np.ones(a.shape), np.zeros(a.shape)
m = a > 0 # no load: full service, nobody abandons
if not m.any():
return sl, ab
nn, aa, hh, tt = n[m], a[m], h[m], tau[m]
lo = np.floor(nn)
w = nn - lo
with np.errstate(divide="ignore", invalid="ignore"):
lr0 = _log_inv_b_minus_1(lo, aa)
lr1 = np.log((lo + 1) / aa) + np.logaddexp(lr0, 0.0) # 1/B(n+1) = 1 + (n+1)/a * 1/B(n)
k = aa.size
_, ab2, sl2 = erlang_a_int(np.concatenate([lo, lo + 1]), np.tile(aa, 2), np.tile(hh, 2), np.tile(tt, 2),
lr=np.concatenate([lr0, lr1]))
sl[m] = (1 - w) * sl2[:k] + w * sl2[k:]
ab[m] = (1 - w) * ab2[:k] + w * ab2[k:]
return sl, ab
def service_model(cfg, pool_name):
"""Service settings for an interactive pool: pool keys override the config's `service` block."""
base = {"model": "A", "patience_seconds": 120.0, "redial_rate": 0.4, "abandon_cap": 0.10}
base.update(cfg.get("service") or {})
p = cfg["pools"][pool_name]
for k, pk in (("model", "service_model"), ("patience_seconds", "patience_seconds"),
("redial_rate", "redial_rate"), ("abandon_cap", "abandon_cap")):
if pk in p:
base[k] = p[pk]
base["model"] = str(base["model"]).upper()
return base
def interactive_service_a(D, F, pool, cal, aht_mult, svc, base_rate=None):
"""Erlang A weekly SL and abandon rate, with redials carried into the next week.
Forecast volumes already contain today's redials, so only abandonment above the week-0 rate
(base_rate, per draw) creates extra contacts. A redial share of them arrives the following
week, scaled by the share of the pool's book still there. Returns sl, abandon, util, retries_in.
"""
aht = pool["aht_seconds"] * np.broadcast_to(aht_mult, D.shape)
interval = cal["interval_seconds"]
vs = np.asarray(cal["buckets"]["volume_share"], float)
hs = np.asarray(cal["buckets"]["hour_share"], float)
alloc = cal["schedule_fit"] * vs + (1 - cal["schedule_fit"]) * hs
patience = max(float(svc["patience_seconds"]), 1e-9)
t = float(pool["asa_seconds"])
def week_metrics(Dw, Fw, ahtw):
"""SL and abandon rate for arrays of draws (any shape); buckets stacked in one call."""
hours = cal["open_hours_week"] * hs[:, None]
agents = Fw[None] * cal["paid_hours_week"] * alloc[:, None] / hours
lam = Dw[None] * vs[:, None] / (hours * 3600 / interval)
s, a = erlang_a(agents, lam * ahtw[None] / interval, ahtw[None] / patience, t / ahtw[None])
return (vs[:, None] * s).sum(0), (vs[:, None] * a).sum(0)
shape = D.shape
sl, ab = week_metrics(D.reshape(-1), F.reshape(-1), aht.reshape(-1))
sl, ab = sl.reshape(shape), ab.reshape(shape)
Deff = D.copy()
retries_in = np.zeros(shape)
redial = float(svc["redial_rate"])
if redial > 0 and shape[1] > 1:
r0 = np.nan_to_num(ab[:, 0]) if base_rate is None else base_rate
out = np.zeros(shape[0])
for w in range(shape[1]):
if w > 0:
prev = D[:, w - 1]
stay = np.where(prev > 1e-9, np.minimum(1.0, D[:, w] / np.where(prev > 1e-9, prev, 1)), 0.0)
rin = out * stay
hit = rin > 0
if hit.any():
retries_in[hit, w] = rin[hit]
Deff[hit, w] = D[hit, w] + rin[hit]
s2, a2 = week_metrics(Deff[hit, w], F[hit, w], aht[hit, w])
sl[hit, w], ab[hit, w] = s2, a2
aw = np.where(Deff[:, w] > 1e-9, ab[:, w], 0.0)
out = redial * np.maximum(0.0, (aw - r0) * Deff[:, w])
util = Deff * aht / 3600 / np.maximum(F * cal["paid_hours_week"], 1e-9)
empty = D <= 1e-9
sl[empty], ab[empty], util[empty] = np.nan, np.nan, np.nan
return sl, ab, util, retries_in
def interactive_service(D, F, pool, cal, aht_mult):
"""Volume-weighted weekly Erlang C SL across intraday buckets.
D: contacts per week (N, W). F: productive FTE, already net of shrinkage (N, W).
"""
aht = pool["aht_seconds"] * np.broadcast_to(aht_mult, D.shape)
interval = cal["interval_seconds"]
vs = np.asarray(cal["buckets"]["volume_share"], float)
hs = np.asarray(cal["buckets"]["hour_share"], float)
alloc = cal["schedule_fit"] * vs + (1 - cal["schedule_fit"]) * hs
sl = np.zeros(D.shape)
for b in range(len(vs)):
hours_b = cal["open_hours_week"] * hs[b]
agents = F * cal["paid_hours_week"] * alloc[b] / hours_b
lam = D * vs[b] / (hours_b * 3600 / interval)
sl += vs[b] * service_level(agents, lam * aht / interval, aht, pool["asa_seconds"])
util = D * aht / 3600 / np.maximum(F * cal["paid_hours_week"], 1e-9)
sl[D <= 1e-9] = np.nan
util[D <= 1e-9] = np.nan
return sl, util
def deferrable_service(D, F, pool, aht_mult):
"""Backlog timeliness index (1 = within target, target/days when late), utilisation, backlog days."""
base_shrink = pool["shrinkage"]
cap = F / (1 - base_shrink) * pool["tx_per_fte_year"] / 52 / np.broadcast_to(aht_mult, D.shape)
backlog = np.zeros(D.shape)
b = np.zeros(D.shape[0])
for w in range(D.shape[1]):
b = np.maximum(0.0, b + D[:, w] - cap[:, w])
backlog[:, w] = b
with np.errstate(divide="ignore", invalid="ignore"):
days = np.where(D > 1e-9, backlog / (D / 5), np.nan)
idx = np.where(days <= pool["target_days"], 1.0, pool["target_days"] / days)
idx[np.isnan(days)] = np.nan
util = np.where(D > 1e-9, D / np.maximum(cap, 1e-9), np.nan)
return idx, util, days
# --------------------------------------------------------------------------
# configuration
# --------------------------------------------------------------------------
def set_path(cfg, dotted, value):
node = cfg
keys = dotted.split(".")
for k in keys[:-1]:
node = node.setdefault(k, {})
node[keys[-1]] = value
def apply_overrides(cfg, overrides):
out = copy.deepcopy(cfg)
for k, v in (overrides or {}).items():
set_path(out, k, v)
return out
def _check_range(name, spec, n=3, lo_ok=None):
"""A PERT triple [low, likely, high] (or a [low, high] pair) must be finite and ordered."""
if spec is None or np.isscalar(spec):
if spec is not None and not np.isfinite(float(spec)):
raise ValueError(f"{name}: not a number ({spec!r})")
return
vals = list(spec)
if len(vals) != n or not all(np.isfinite(float(v)) for v in vals):
raise ValueError(f"{name}: expected {n} finite numbers [low{', likely' if n == 3 else ''}, high], got {spec!r}")
if any(float(a) > float(b) for a, b in zip(vals, vals[1:])):
raise ValueError(f"{name}: values must be ordered low <= {'likely <= ' if n == 3 else ''}high, got {spec!r}")
def validate(cfg):
for key in ("horizon_weeks", "calendar", "freeze", "attrition", "release", "transfer",
"waves", "exit", "replatform", "fate_priors", "pools", "clients"):
if key not in cfg:
raise ValueError(f"config missing '{key}'")
b = cfg["calendar"]["buckets"]
for k in ("volume_share", "hour_share"):
if abs(sum(b[k]) - 1) > 1e-6:
raise ValueError(f"calendar.buckets.{k} must sum to 1")
for name, p in cfg["pools"].items():
if p["mode"] not in ("interactive", "deferrable"):
raise ValueError(f"pool {name}: mode must be interactive or deferrable")
need = ["aht_seconds", "asa_seconds"] if p["mode"] == "interactive" else ["tx_per_fte_year", "target_days"]
for k in ["headcount", "shrinkage", "wave", "service_floor"] + need:
if k not in p:
raise ValueError(f"pool {name}: missing '{k}'")
if p["wave"] not in cfg["waves"]:
raise ValueError(f"pool {name}: wave '{p['wave']}' not in waves")
if p["mode"] == "interactive":
sv = service_model(cfg, name)
if sv["model"] not in ("A", "C"):
raise ValueError(f"pool {name}: service model must be A or C")
if not (float(sv["patience_seconds"]) > 0 and 0 <= float(sv["redial_rate"]) <= 1 and 0 < float(sv["abandon_cap"]) < 1):
raise ValueError(f"pool {name}: patience_seconds > 0, redial_rate 0-1 and abandon_cap 0-1 required")
for c in cfg["clients"]:
if c["pool"] not in cfg["pools"]:
raise ValueError(f"client {c['id']}: unknown pool '{c['pool']}'")
if c.get("wave") and c["wave"] not in cfg["waves"]:
raise ValueError(f"client {c['id']}: unknown wave '{c['wave']}'")
if c.get("contract", "evergreen") not in ("fixed", "evergreen", "evergreen_tfc"):
raise ValueError(f"client {c['id']}: contract must be fixed, evergreen or evergreen_tfc")
s = cfg.get("seasonality")
if s is not None and len(s) < cfg["horizon_weeks"]:
raise ValueError("seasonality shorter than horizon_weeks")
att, tr = cfg["attrition"], cfg["transfer"]
for name, spec in (("freeze.end_week", cfg["freeze"]["end_week"]),
("attrition.freeze_multiplier", att["freeze_multiplier"]),
("attrition.post_multiplier_release", att["post_multiplier_release"]),
("attrition.post_multiplier_transfer", att["post_multiplier_transfer"]),
("shrinkage.post_announcement_add", cfg.get("shrinkage", {}).get("post_announcement_add", 0.0)),
("transfer.synergy_realisation", tr.get("synergy_realisation", 1.0)),
("transfer.aht_uplift", tr["aht_uplift"]),
("replatform.offset_weeks", cfg["replatform"]["offset_weeks"]),
*((f"exit.notice_weeks.{k}", v) for k, v in cfg["exit"]["notice_weeks"].items())):
_check_range(name, spec)
for name, p in cfg["pools"].items():
for k in ("volume_error", "aht_error"):
_check_range(f"pools.{name}.{k}", p.get(k), n=2)
if not 0 <= float(att.get("retention_effect", 0.0)) <= 1:
raise ValueError("attrition.retention_effect must be 0-1")
return cfg
def fate_probs(cfg, client):
if all(f"p_{f}" in client for f in FATES):
p = np.array([client[f"p_{f}"] for f in FATES], float)
else:
pri = cfg["fate_priors"][str(client.get("health", "green")).lower()]
p = np.array([pri[f] for f in FATES], float)
return p / p.sum()
# --------------------------------------------------------------------------
# the simulation
# --------------------------------------------------------------------------
@dataclass
class Result:
cfg: dict
pools: list
freeze_end: np.ndarray # (N,)
fate: np.ndarray # (N, C)
dep: np.ndarray # (N, C) week the work leaves the source
H: np.ndarray # (N, P, W) source heads
F: np.ndarray # (N, P, W) source productive FTE
D: np.ndarray # (N, P, W) source workload per week
S: np.ndarray # (N, P, W) source service (SL or timeliness index)
Hd: np.ndarray # destination heads
Dd: np.ndarray # destination workload, AHT-adjusted
Sd: np.ndarray # destination service
U: np.ndarray # source utilisation: work offered / productive capacity
Ud: np.ndarray # destination utilisation
flows: dict # (N, P) source headcount flows; they add up exactly per draw
A: np.ndarray = None # (N, P, W) source abandon rate (interactive Erlang A pools; else NaN)
Ad: np.ndarray = None # destination abandon rate
R: np.ndarray = None # (N, P, W) source redials arriving this week (contacts)
Rd: np.ndarray = None # destination redials
train_undelivered_hours: np.ndarray = None # (N, P) training hours that could not fit before the wave (normally 0)
def run(cfg, draws=None, seed=None):
cfg = validate(cfg)
N = int(draws or cfg.get("draws", 2000))
W = int(cfg["horizon_weeks"])
base_seed = int(seed if seed is not None else cfg.get("seed", SEED))
rng = np.random.default_rng(base_seed)
rng_dest = np.random.default_rng([base_seed, 1]) # destination attrition: its own stream, so it never shifts the source's draws
cal, att, rel, tr = cfg["calendar"], cfg["attrition"], cfg["release"], cfg["transfer"]
pools = list(cfg["pools"])
pidx = {p: i for i, p in enumerate(pools)}
P, clients = len(pools), cfg["clients"]
C = len(clients)
base = np.array([c["annual_volume"] / 52 for c in clients], float)
M = np.zeros((C, P))
M[np.arange(C), [pidx[c["pool"]] for c in clients]] = 1
season = np.asarray(cfg.get("seasonality") or [1.0] * W, float)[:W]
# per-draw drivers
fz_start = int(cfg["freeze"].get("start_week", 0))
fz_end = np.maximum(np.round(pert(rng, cfg["freeze"]["end_week"], N)).astype(int), fz_start)
m_freeze = pert(rng, att["freeze_multiplier"], N)
m_post_r = pert(rng, att["post_multiplier_release"], N) * (1 - att.get("retention_effect", 0.0))
m_post_t = pert(rng, att["post_multiplier_transfer"], N)
shrink_add = pert(rng, cfg.get("shrinkage", {}).get("post_announcement_add", 0.0), N)
vol_err = np.stack([lognormal_mult(rng, cfg["pools"][p].get("volume_error"), N) for p in pools], 1)
aht_err = np.stack([lognormal_mult(rng, cfg["pools"][p].get("aht_error"), N) for p in pools], 1)
aht_up = pert(rng, tr["aht_uplift"], N)
realised = pert(rng, tr.get("synergy_realisation", 1.0), N)
# client fates and the week each client's work leaves the source
fate = np.empty((N, C), int)
for j, c in enumerate(clients):
fate[:, j] = rng.choice(3, size=N, p=fate_probs(cfg, c))
wave_wk = np.stack([fz_end + int(cfg["waves"][c.get("wave") or cfg["pools"][c["pool"]]["wave"]])
for c in clients], 1)
dep = wave_wk.copy()
for j, c in enumerate(clients):
# each client's timing draws come from its own stream, so adding, removing or re-classifying
# one client never reshuffles another's departure week
rng_c = np.random.default_rng([base_seed, 2, j])
repl = fz_end + np.round(pert(rng_c, cfg["replatform"]["offset_weeks"], N)).astype(int)
is_r = fate[:, j] == T_REPLATFORM
dep[:, j] = np.where(is_r, np.minimum(repl, wave_wk[:, j]), dep[:, j]) # re-platformed work cannot outlive its wave
kind = c.get("contract", "evergreen")
spec = cfg["exit"]["notice_weeks"]["evergreen_tfc" if kind == "evergreen_tfc" else "evergreen"]
drawn = fz_end + np.round(pert(rng_c, spec, N)).astype(int) # drawn for fixed-term clients too, then discarded
ex = np.full(N, int(c["expiry_week"])) if kind == "fixed" and c.get("expiry_week") is not None else drawn
is_exit = fate[:, j] == T_EXIT
late = is_exit & (ex >= wave_wk[:, j]) # contract outlives the wave: work transfers
fate[late, j] = T_TRANSFER
dep[:, j] = np.where(is_exit & ~late, ex, dep[:, j])
dep = np.maximum(dep, 0)
is_t = fate == T_TRANSFER
# staff state
H0 = np.array([cfg["pools"][p]["headcount"] for p in pools], float)
shrink0 = np.array([cfg["pools"][p]["shrinkage"] for p in pools], float)
D0 = base @ M
syn, acc = float(tr["synergy"]), float(tr.get("acceptance", 1.0))
tw, th = int(tr["training_weeks"]), float(tr["training_hours"])
# each transferee's training fits between the announcement and their wave: a wave closer than
# training_weeks compresses the window rather than dropping hours. A window is never shorter
# than the weeks the hours need at one productive week per head per week (so 80 hours are not
# asked of a 25-hour week); it can only stretch back as far as the announcement.
shr_c = shrink0[[pidx[c["pool"]] for c in clients]]
fit = np.ceil(th / np.maximum(cal["paid_hours_week"] * (1 - shr_c), 1e-9))[None, :]
span = np.maximum(1, dep - fz_end[:, None])
tw_eff = np.minimum(np.maximum(np.minimum(tw, span), fit), span).astype(float)
lag, buf = int(rel["lag_weeks"]), float(rel.get("buffer", 0.0))
ov_fte = np.array([cfg["pools"][p].get("overflow_fte", 0.0) for p in pools], float)
ov_start = np.array([cfg["pools"][p].get("overflow_start_week", 0) for p in pools])
ov_mult = np.array([cfg["pools"][p].get("overflow_aht_mult", 1.0) for p in pools], float)
p_week = att["base_annual"] / 52
R = np.tile(H0, (N, 1)) # everyone, until the split at announcement
T = np.zeros((N, P)) # transferees, after the split
Hd = np.tile(np.array([cfg["pools"][p].get("dest_extra_fte", 0.0) for p in pools], float), (N, 1))
split = np.zeros(N, bool)
carry = np.zeros((N, P)) # training FTE-weeks that did not fit into a week, spilled forward
undelivered = np.zeros((N, P)) # training hours still owed when the wave took the transferees away
req_hist = []
out = {k: np.zeros((N, P, W)) for k in ("H", "F", "D", "Hd", "Dd")}
flows = {k: np.zeros((N, P)) for k in ("attrition_freeze", "attrition_post", "transferred", "released")}
for w in range(W):
active = dep > w
load = base * season[w]
D = ((active * load) @ M) * vol_err
DT_plan = ((active & is_t) * base) @ M
DR_plan = ((active & ~is_t) * base) @ M
post = w >= fz_end
in_freeze = (w >= fz_start) & ~post
new = post & ~split
if new.any():
tot = DT_plan + DR_plan
share = np.where(tot > 0, DT_plan * (1 - syn) / np.where(tot > 0, tot, 1), 0.0)
T[new] = R[new] * share[new]
R[new] = R[new] - T[new]
split |= new
mult_r = np.where(post, m_post_r, np.where(in_freeze, m_freeze, 1.0))[:, None]
mult_t = np.where(post, m_post_t, np.where(in_freeze, m_freeze, 1.0))[:, None]
u_r, u_t = rng.random((N, P)), rng.random((N, P)) # fixed consumption per week: CRN inside the loop
lost = losses(u_r, R, p_week * mult_r)
R = R - lost
lost_t = losses(u_t, T, p_week * mult_t)
T = T - lost_t
flows["attrition_post"] += (lost + lost_t) * post[:, None]
flows["attrition_freeze"] += (lost + lost_t) * ~post[:, None]
# transfers: staff follow their work in proportion to transfer workload leaving now
leave_plan = (((dep == w) & is_t) * base) @ M
denom = DT_plan + leave_plan
frac = np.where(denom > 0, leave_plan / np.where(denom > 0, denom, 1), 0.0)
moved = T * frac
T = T - moved
flows["transferred"] += moved
if cfg.get("destination", {}).get("top_up_to_plan", True):
# the destination hires to plan; source leakage is not inherited
Hd = Hd + H0 * leave_plan / np.where(D0 > 0, D0, 1) * (1 - syn)
else:
Hd = Hd + moved * acc
u_d = rng_dest.random((N, P)) # always drawn, so the switch changes nothing else
if not cfg.get("destination", {}).get("backfill", True):
Hd = Hd - losses(u_d, Hd, np.full((N, 1), p_week))
# releases: only after the dismissal notice, only once the work is gone (+ lag)
req = H0 * (DR_plan + syn * DT_plan) / np.where(D0 > 0, D0, 1) * (1 + buf)
req_hist.append(req)
req_eff = np.max(np.stack(req_hist[-(lag + 1):]), axis=0)
can_release = (w >= fz_end + int(rel["notice_weeks"])) & bool(rel.get("allowed", True))
released = np.clip(R - req_eff, 0, None) * can_release[:, None]
R = R - released
flows["released"] += released
# training pulls upcoming transferees off the floor before their wave
win = (dep > w) & (dep <= w + tw_eff) & is_t
up = (win * base * (tw / tw_eff)) @ M # each client's work, spread over its own window
train = T * np.where(DT_plan > 0, up / np.where(DT_plan > 0, DT_plan, 1), 0.0) * th / (tw * cal["paid_hours_week"])
shrink = shrink0[None, :] + np.where(post, shrink_add, 0.0)[:, None]
ov = np.where(w >= ov_start, ov_fte / ov_mult, 0.0)
# a compressed window can ask for more training time than the pool has in a week: what does
# not fit spills into the following weeks rather than being dropped (until the wave takes the
# transferees with it, after which nothing is left to train)
avail = (R + T) * (1 - shrink)
undelivered += np.where(T > 1e-9, 0.0, carry) * cal["paid_hours_week"]
need = train + np.where(T > 1e-9, carry, 0.0)
taken = np.minimum(need, avail)
carry = need - taken
F = np.maximum(avail - taken + ov * (1 - shrink0), 0.0)
landed = is_t & (dep <= w)
age = w - dep
steady = 1 - syn * realised[:, None] # the productivity the synergy assumes, as realised
ramp = steady * (1 + (aht_up[:, None] - 1) * np.clip(1 - age / max(int(tr["ramp_weeks"]), 1), 0, 1))
Dd = (((landed * ramp) * load) @ M) * vol_err
out["H"][:, :, w], out["F"][:, :, w], out["D"][:, :, w] = R + T, F, D
out["Hd"][:, :, w], out["Dd"][:, :, w] = Hd, Dd
S, Sd, U, Ud, Ab, Abd = (np.full((N, P, W), np.nan) for _ in range(6))
Rs, Rds = np.zeros((N, P, W)), np.zeros((N, P, W))
min_share = float(cfg.get("scoring", {}).get("min_share", 0.10))
for i, p in enumerate(pools):
pool = cfg["pools"][p]
Fd = out["Hd"][:, i, :] * (1 - pool["shrinkage"])
e = aht_err[:, i:i + 1]
if pool["mode"] == "interactive" and service_model(cfg, p)["model"] == "A":
svc = service_model(cfg, p)
S[:, i, :], Ab[:, i, :], U[:, i, :], Rs[:, i, :] = interactive_service_a(
out["D"][:, i, :], out["F"][:, i, :], pool, cal, e, svc)
r0 = np.nan_to_num(Ab[:, i, 0]) # the destination inherits today's abandonment as its baseline
Sd[:, i, :], Abd[:, i, :], Ud[:, i, :], Rds[:, i, :] = interactive_service_a(
out["Dd"][:, i, :], Fd, pool, cal, e, svc, base_rate=r0)
elif pool["mode"] == "interactive":
S[:, i, :], U[:, i, :] = interactive_service(out["D"][:, i, :], out["F"][:, i, :], pool, cal, e)
Sd[:, i, :], Ud[:, i, :] = interactive_service(out["Dd"][:, i, :], Fd, pool, cal, e)
Ab[:, i, :] = np.where(out["D"][:, i, :] > 1e-9, 0.0, np.nan)
Abd[:, i, :] = np.where(out["Dd"][:, i, :] > 1e-9, 0.0, np.nan)
else:
S[:, i, :], U[:, i, :], _ = deferrable_service(out["D"][:, i, :], out["F"][:, i, :], pool, e)
Sd[:, i, :], Ud[:, i, :], _ = deferrable_service(out["Dd"][:, i, :], Fd, pool, e)
# residual slivers (a client left behind after its pool's wave) are not scored
tiny = out["D"][:, i, :] < min_share * D0[i]
S[:, i, :][tiny] = np.nan
U[:, i, :][tiny] = np.nan
Ab[:, i, :][tiny] = np.nan
# the destination is scored only once at least min_share of what will eventually land has
# landed AND its team has reached an agent floor: max(1, ceil(min_share × eventual heads)).
# A single early client is a sliver there too, and a handful of agents is an Erlang
# small-pool artefact, not a queue the program will run. min_share is a cliff: just above it
# the small landed team is scored at full weight (see 06-interpretation).
full_d = np.max(out["Dd"][:, i, :], axis=1, keepdims=True)
heads_full = np.max(out["Hd"][:, i, :], axis=1, keepdims=True)
floor_h = np.maximum(1.0, np.ceil(min_share * heads_full))
tiny_d = (out["Dd"][:, i, :] < min_share * np.where(full_d > 0, full_d, np.inf)) | (out["Hd"][:, i, :] < floor_h)
Sd[:, i, :][tiny_d] = np.nan
Ud[:, i, :][tiny_d] = np.nan
Abd[:, i, :][tiny_d] = np.nan
return Result(cfg, pools, fz_end, fate, dep, out["H"], out["F"], out["D"], S,
out["Hd"], out["Dd"], Sd, U, Ud, flows, Ab, Abd, Rs, Rds, undelivered)
# --------------------------------------------------------------------------
# summaries
# --------------------------------------------------------------------------
def _nanmin(x):
m = np.where(np.isnan(x), np.inf, x).min(axis=1)
return np.where(np.isinf(m), np.nan, m)
def _abandon_kpis(ab, svc):
"""Peak weekly abandonment and weeks above the cap (Erlang A); zeros under Erlang C."""
cap = float(svc["abandon_cap"])
peak = np.where(np.isnan(ab), -np.inf, ab).max(axis=1)
peak = np.where(np.isinf(peak), np.nan, peak)
over = (np.nan_to_num(ab) > cap).sum(axis=1)
return {"service_model": svc["model"], "abandon_peak_p50": float(np.nanmedian(peak)),
"abandon_peak_p90": float(np.nanpercentile(peak, 90)),
"p_abandon_over_cap": float(np.mean(np.nan_to_num(peak) > cap)),
"weeks_over_cap_p50": float(np.median(over))}
Block 5, part 2 of 2 — 05-simulator.md
def kpis(res):
"""One row per pool and side: baseline, trough, breach probability, weeks below floor."""
rows = []
for i, p in enumerate(res.pools):
floor = res.cfg["pools"][p]["service_floor"]
interactive = res.cfg["pools"][p]["mode"] == "interactive"
svc = service_model(res.cfg, p) if interactive else None
for side, s, u, ab in (("source", res.S[:, i, :], res.U[:, i, :], res.A[:, i, :]),
("destination", res.Sd[:, i, :], res.Ud[:, i, :], res.Ad[:, i, :])):
live = ~np.isnan(s).all(axis=1)
if not live.any():
continue
s, u, ab = s[live], u[live], ab[live]
lo = _nanmin(s)
below = (s < floor).sum(axis=1)
trough_wk = np.argmin(np.where(np.isnan(s), np.inf, s), axis=1)
first_wk = np.argmax(~np.isnan(s), axis=1) # the week the side becomes scorable (0 at the source)
with warnings.catch_warnings():
warnings.simplefilter("ignore", RuntimeWarning)
rows.append({
"pool": p, "side": side, "floor": floor,
"week0_p50": np.nanmedian(s[:, 0]) if side == "source" else np.nan,
"trough_p50": np.nanmedian(lo),
"trough_p10": np.nanpercentile(lo, 10),
"trough_week_p50": float(np.median(trough_wk)),
"scorable_week_p50": float(np.median(first_wk)),
"p_breach": float(np.mean(lo < floor)),
"weeks_below_p50": float(np.median(below)),
"weeks_below_p90": float(np.percentile(below, 90)),
"peak_util_p50": float(np.nanmedian(np.nanmax(u, axis=1))),
**(_abandon_kpis(ab, svc) if interactive else
{"service_model": "", "abandon_peak_p50": np.nan, "abandon_peak_p90": np.nan,
"p_abandon_over_cap": np.nan, "weeks_over_cap_p50": np.nan}),
})
return pd.DataFrame(rows)
def weekly(res):
"""Long table: pool, side, week, service P10/P50/P90, heads and workload P50."""
rows = []
with warnings.catch_warnings():
warnings.simplefilter("ignore", RuntimeWarning)
for i, p in enumerate(res.pools):
for side, s, h, d, u, ab, rr in (("source", res.S, res.H, res.D, res.U, res.A, res.R),
("destination", res.Sd, res.Hd, res.Dd, res.Ud, res.Ad, res.Rd)):
q = np.nanpercentile(s[:, i, :], [10, 50, 90], axis=0)
um = np.nanmedian(u[:, i, :], axis=0)
qa = np.nanpercentile(ab[:, i, :], [10, 50, 90], axis=0)
rm = np.median(rr[:, i, :], axis=0)
for w in range(s.shape[2]):
rows.append({"pool": p, "side": side, "week": w,
"service_p10": q[0, w], "service_p50": q[1, w], "service_p90": q[2, w],
"abandon_p10": qa[0, w], "abandon_p50": qa[1, w], "abandon_p90": qa[2, w],
"redials_p50": rm[w],
"util_p50": um[w], "heads_p50": np.median(h[:, i, w]),
"workload_p50": np.median(d[:, i, w])})
return pd.DataFrame(rows)
def waterfall(res):
"""Mean source headcount flows per pool; means are used so each row adds up exactly.
start = attrition_freeze + attrition_post + transferred + released + remaining.
Attrition in the freeze counts everything before the announcement.
"""
rows = []
start = np.array([res.cfg["pools"][p]["headcount"] for p in res.pools], float)
remaining = res.H[:, :, -1]
for i, p in enumerate(res.pools):
rows.append({"pool": p, "start": start[i],
**{k: float(v[:, i].mean()) for k, v in res.flows.items()},
"remaining": float(remaining[:, i].mean()),
"attrition_p10": float(np.percentile(res.flows["attrition_freeze"][:, i] + res.flows["attrition_post"][:, i], 10)),
"attrition_p90": float(np.percentile(res.flows["attrition_freeze"][:, i] + res.flows["attrition_post"][:, i], 90))})
df = pd.DataFrame(rows)
total = df.drop(columns="pool").sum(numeric_only=True)
total[["attrition_p10", "attrition_p90"]] = np.nan # percentiles do not add across pools
return pd.concat([df, pd.DataFrame([{"pool": "TOTAL", **total}])], ignore_index=True)
def client_table(res):
df = pd.DataFrame(res.cfg["clients"])
for k, f in enumerate(FATES):
df[f"share_{f}"] = (res.fate == k).mean(axis=0)
df["dep_week_p50"] = np.median(res.dep, axis=0)
return df
def scenarios(cfg, variants, draws=None, seed=None):
"""KPIs for a base case plus named override sets. Same seed = common random numbers."""
frames = []
for name, ov in {"base": {}, **variants}.items():
k = kpis(run(apply_overrides(cfg, ov), draws, seed))
k.insert(0, "scenario", name)
frames.append(k)
return pd.concat(frames, ignore_index=True)
def _flatten(d, prefix=""):
for k, v in d.items():
key = f"{prefix}{k}"
if isinstance(v, dict):
yield from _flatten(v, key + ".")
else:
yield key, str(v)
def to_excel(res, path, scen=None):
inputs = pd.DataFrame([kv for kv in _flatten({k: v for k, v in res.cfg.items() if k != "clients"})],
columns=["parameter", "value"])
models = sorted({service_model(res.cfg, p)["model"] for p in res.pools
if res.cfg["pools"][p]["mode"] == "interactive"})
about = pd.DataFrame({"item": ["draws", "horizon_weeks", "freeze_end_week_p50",
"service metric (interactive)", "abandonment (interactive)",
"service metric (deferrable)"],
"value": [res.S.shape[0], res.S.shape[2], float(np.median(res.freeze_end)),
" / ".join({"A": "Erlang A: answered within threshold / all offered",
"C": "Erlang C service level (no abandonment)"}[m] for m in models) or "none",
"Erlang A abandon rate; redials of abandonment above week 0 added to next week" if "A" in models else "not modelled (Erlang C)",
"timeliness index = min(1, target_days / backlog_days)"]})
with pd.ExcelWriter(path, engine="openpyxl") as xw:
about.to_excel(xw, sheet_name="About", index=False)
kpis(res).to_excel(xw, sheet_name="KPIs", index=False)
waterfall(res).to_excel(xw, sheet_name="Waterfall", index=False)
if scen is not None:
scen.to_excel(xw, sheet_name="Scenarios", index=False)
weekly(res).to_excel(xw, sheet_name="Weekly", index=False)
client_table(res).to_excel(xw, sheet_name="Clients", index=False)
inputs.to_excel(xw, sheet_name="Inputs", index=False)
return path
def charts(res, outdir="."):
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
wk = weekly(res)
paths = []
for p in res.pools:
fig, ax = plt.subplots(figsize=(10, 4))
for side, col in (("source", "#1f4e79"), ("destination", "#b35c00")):
d = wk[(wk.pool == p) & (wk.side == side)]
ax.fill_between(d.week, d.service_p10, d.service_p90, color=col, alpha=0.18)
ax.plot(d.week, d.service_p50, color=col, label=f"{side} P50 (band P10-P90)")
if d.abandon_p50.fillna(0).max() > 0:
ax.plot(d.week, d.abandon_p50, color=col, lw=1, ls=":", label=f"{side} abandonment P50")
fe = np.percentile(res.freeze_end, [10, 50, 90])
ax.axvspan(res.cfg["freeze"].get("start_week", 0), fe[1], color="grey", alpha=0.10, label="freeze (P50 end)")
ax.axhline(res.cfg["pools"][p]["service_floor"], color="red", lw=0.8, ls="--", label="floor")
ax.set_ylim(0, 1.02)
ax.set_xlabel("week")
ax.set_ylabel("service / abandonment")
ax.set_title(f"Pool {p}")
ax.legend(loc="lower left", fontsize=8)
path = f"{outdir}/service_{p}.png"
fig.tight_layout()
fig.savefig(path, dpi=120)
plt.close(fig)
paths.append(path)
return paths
# --------------------------------------------------------------------------
# demonstration configuration (synthetic, no real data)
# --------------------------------------------------------------------------
def demo_config():
rng = np.random.default_rng(7)
cal = {"paid_hours_week": 37.5, "open_hours_week": 55, "interval_seconds": 1800, "schedule_fit": 0.85,
"buckets": {"volume_share": [0.45, 0.40, 0.15], "hour_share": [0.30, 0.45, 0.25]}}
pools = {
"DED": {"mode": "interactive", "headcount": 40, "shrinkage": 0.32, "aht_seconds": 540, "asa_seconds": 60,
"service_floor": 0.70, "wave": "W1", "volume_error": [0.92, 1.08], "aht_error": [0.95, 1.05]},
"SHR": {"mode": "interactive", "headcount": 70, "shrinkage": 0.32, "aht_seconds": 480, "asa_seconds": 60,
"service_floor": 0.70, "wave": "W2", "volume_error": [0.92, 1.08], "aht_error": [0.95, 1.05]},
"BO": {"mode": "deferrable", "headcount": 25, "shrinkage": 0.30, "tx_per_fte_year": 4400, "target_days": 2,
"service_floor": 0.80, "wave": "W3", "volume_error": [0.92, 1.08], "aht_error": [0.95, 1.05]},
}
def weekly_capacity(p, occ):
q = pools[p]
if q["mode"] == "interactive":
return q["headcount"] * (1 - q["shrinkage"]) * cal["paid_hours_week"] * 3600 / q["aht_seconds"] * occ
return q["headcount"] * q["tx_per_fte_year"] / 52 * occ
clients = []
for p, n, occ in (("DED", 4, 0.74), ("SHR", 40, 0.74), ("BO", 12, 0.88)):
w = rng.lognormal(0, 0.9 if p != "DED" else 0.4, n)
vols = w / w.sum() * weekly_capacity(p, occ) * 52
for k in range(n):
contract = rng.choice(["fixed", "evergreen", "evergreen_tfc"], p=[0.5, 0.3, 0.2])
clients.append({"id": f"{p}-{k + 1:02d}", "pool": p, "annual_volume": round(float(vols[k])),
"health": str(rng.choice(["green", "amber", "red"], p=[0.6, 0.3, 0.1])),
"contract": str(contract),
"expiry_week": int(rng.integers(0, 90)) if contract == "fixed" else None})
return {
"horizon_weeks": 52, "draws": 2000, "seed": SEED, "calendar": cal, "seasonality": None,
"service": {"model": "A", "patience_seconds": 120, "redial_rate": 0.4, "abandon_cap": 0.10},
"freeze": {"start_week": 0, "end_week": [8, 14, 26]},
"attrition": {"base_annual": 0.15, "freeze_multiplier": [1.0, 1.4, 2.2],
"post_multiplier_release": [1.5, 3.0, 5.0], "post_multiplier_transfer": [0.8, 1.2, 2.0],
"retention_effect": 0.0},
"shrinkage": {"post_announcement_add": [0.0, 0.04, 0.10]},
"release": {"notice_weeks": 6, "lag_weeks": 2, "buffer": 0.05, "allowed": True},
"scoring": {"min_share": 0.10},
"destination": {"backfill": True, "top_up_to_plan": True},
"transfer": {"synergy": 0.25, "synergy_realisation": [0.4, 0.8, 1.0], "acceptance": 1.0, "training_hours": 24, "training_weeks": 6,
"aht_uplift": [1.03, 1.10, 1.25], "ramp_weeks": 12},
"waves": {"W1": 6, "W2": 12, "W3": 18},
"exit": {"notice_weeks": {"evergreen": [8, 13, 26], "evergreen_tfc": [4, 6, 13]}},
"replatform": {"offset_weeks": [4, 8, 16]},
"fate_priors": {"green": {"transfer": 0.85, "exit": 0.10, "replatform": 0.05},
"amber": {"transfer": 0.65, "exit": 0.25, "replatform": 0.10},
"red": {"transfer": 0.35, "exit": 0.55, "replatform": 0.10}},
"pools": pools, "clients": clients,
}
def load_config(path):
import yaml
with open(path) as f:
return yaml.safe_load(f)
if __name__ == "__main__":
args = sys.argv[1:]
draws = int(args[args.index("--draws") + 1]) if "--draws" in args else None
xlsx = args[args.index("--xlsx") + 1] if "--xlsx" in args else "migration_sl_results.xlsx"
cfg_path = next((a for a in args if a.endswith((".yaml", ".yml"))), None)
cfg = load_config(cfg_path) if cfg_path else demo_config()
res = run(cfg, draws)
if "--scenarios" in args:
# a YAML file of {name: {dotted.key: value, ...}} override sets, e.g. the recipes in 03-levers
variants = load_config(args[args.index("--scenarios") + 1]) or {}
else:
variants = {
"short_freeze": {"freeze.end_week": [6, 8, 12]},
"retention_bonus": {"attrition.retention_effect": 0.4},
"overflow_20_SHR": {"pools.SHR.overflow_fte": 20, "pools.SHR.overflow_aht_mult": 1.2},
}
scen = scenarios(cfg, variants, draws=res.S.shape[0]) # the base row must match the KPI table above
pd.set_option("display.width", 160)
print(kpis(res).round(3).to_string(index=False))
print()
print(scen.round(3).to_string(index=False))
to_excel(res, xlsx, scen)
try:
print("charts:", charts(res))
except ImportError:
pass
print("workbook:", xlsx)
```
Shape-card scripts
Optional. They move a scenario's shape (timing, rates, ranges and their status, the book of business and the team split as shares; never volumes, headcount or names) between this pack and the Migration Scenario Modeler, in both directions. Save each under its filename next to the simulator.
Block 8 — export_shape_card.py
"""Export a pack run as a shape card for the WFM Labs Migration Scenario Modeler.
A shape card is the scenario without its scale: timing, rates, multipliers, handle times,
ranges and how sure each one is, plus week-0 occupancy and the workload mix instead of
volumes and headcount. No client ids, pool names, wave names, volumes or headcount leave in
it: the card is built from an explicit allow-list (SCHEMA below); anything else is refused.
The modeler has one blended team (voice, chat, email); the pack has several pools. The export
collapses the pools: interactive pools count as voice (or chat, if the pool sets
``channel: chat``), deferrable pools as email. Per-client departures become the modeler's waves
(transferred work, which takes its staff with it) and step-downs (work that exits or
re-platforms), each expressed as a share of the work still at the source when it happens, so
the modeler's multiplicative rules reproduce the pack's expected departure curve. Events that
the pack schedules relative to the freeze end are placed at the median freeze end.
The departure curve is the EXPECTED (mean) staircase over the pack's draws: the pack's P50
trough is deeper than the modeler's most-likely line. The card says so in its `note`.
Status of each assumption: from an optional ``status:`` block in the config
(``status: {attrition.base_annual: confirmed, freeze.end_week: {status: estimated, owner: HR}}``).
Without an entry, a value that still equals the pack's built-in default is "default" (not
asked); a value someone changed is "estimated". Nothing is ever exported as confirmed unless the
block says so.
Format 2 (the default) also carries the BOOK itself: contract-type and relationship-health
mixes as workload shares, the fate priors, the exit-notice and re-platform ranges and the waves
as shares of transferring work, plus the transfer/release split of the team. The modeler then
derives the departure curve from those shares and draws real staircases in its Monte Carlo
instead of the fixed expected staircase (which is still in the card for format-1 readers).
python export_shape_card.py config.yaml --out card.json [--draws 500] [--title "..."] [--format 1|2]
python export_shape_card.py --demo --out card.json
"""
import copy
import json
import re
import sys
from datetime import date
import numpy as np
from scipy import stats
import mig_sl_sim as sim
FORMAT_1 = "wfm-migration-shape-card/1"
FORMAT_2 = "wfm-migration-shape-card/2"
FORMATS = (FORMAT_1, FORMAT_2)
FORMAT = FORMAT_2
VERSION = "1.2" # the pack version; the card format is separate (wfm-migration-shape-card/2)
GRANULARITY = 40
QUANTILE_GRID = 64
MAX_STEPS = 12
MAX_WAVES = 4
TOOL_TEXT = set() # the exporter's own notes and labels; the text guard checks everything else
def _t(text):
TOOL_TEXT.add(text)
return text
DEFAULT_TITLE = _t("Migration scenario (from the desktop pack)")
CARD_NOTE = _t("Departures are the expected (mean) staircase over the pack's draws; the pack's P50 "
"trough is deeper. Turn on bands and treat this as exploration.")
CARD_NOTE_2 = _t("The book travels as shares (contract and health mix, fate priors, notice ranges, waves); "
"the modeler draws real staircases from them. One blended team stands in for the pack's pools.")
# Everything a card may contain, with the type of each leaf. Unknown keys never get written.
NUM, INT, BOOL, STR = "num", "int", "bool", "str"
SCHEMA = {
"format": STR, "title": STR, "created": STR, "note": STR,
"occupancy": NUM, "mix": {"voice": NUM, "chat": NUM, "email": NUM}, "borrowedShare": NUM,
"stepDownsRelativeToFreezeEnd": BOOL,
"scenario": {
"horizonWeeks": INT,
"channels": {
"voice": {"aht": NUM, "slTarget": NUM, "slSeconds": INT},
"chat": {"aht": NUM, "slTarget": NUM, "slSeconds": INT},
"email": {"targetDays": NUM},
},
"pool": {"shrinkage": NUM, "paidHours": NUM, "openHours": NUM},
"profile": {"volumeShare": [NUM], "hourShare": [NUM], "scheduleFit": NUM},
"attrition": {"annual": NUM, "tensionMult": NUM, "postMult": NUM},
"freeze": {"startWeek": INT, "endWeek": INT, "backfillBefore": BOOL},
"demand": {"runoffPctWeek": NUM, "stepDowns": [{"week": INT, "pct": NUM}], "intakeOn": BOOL},
"after": {"waves": [{"weeksAfterFreeze": INT, "pct": NUM}], "trainingHours": NUM, "trainingWeeks": INT,
"surgePts": NUM, "surgeWeeks": INT, "releasesOn": BOOL, "noticeWeeks": INT,
"releaseBuffer": NUM, "lookaheadWeeks": INT},
"borrowed": {"startWeek": INT, "endWeek": INT, "ahtPenalty": NUM},
"service": {"model": STR, "patience": {"voice": NUM, "chat": NUM}, "redialRate": NUM, "abandonCap": NUM},
# format 2: the book and the team split (shares, weeks and multipliers only)
"people": {"split": BOOL, "postMultTransfer": NUM, "postMultRelease": NUM, "retentionEffect": NUM, "retentionTarget": STR},
"book": {
"mode": STR,
"contractMix": {"fixed": NUM, "evergreen": NUM, "tfc": NUM},
"fixedExpiry": [INT],
"healthMix": {"green": NUM, "amber": NUM, "red": NUM},
"priors": {"green": {"transfer": NUM, "exit": NUM, "replatform": NUM},
"amber": {"transfer": NUM, "exit": NUM, "replatform": NUM},
"red": {"transfer": NUM, "exit": NUM, "replatform": NUM}},
"exitNotice": {"evergreen": [NUM], "tfc": [NUM]},
"replatformOffset": [NUM],
"waves": [{"weeksAfterFreeze": INT, "pct": NUM}],
"waveSlip": [NUM],
"granularity": INT,
},
},
"assumptions": {"*": {"range": [NUM], "status": STR, "owner": STR, "note": STR}},
"questions": [{"id": STR, "text": STR, "sets": [STR]}],
"source": {"tool": STR, "version": STR, "date": STR},
}
STATUSES = ("default", "estimated", "confirmed")
MAX_TEXT = 200
# pack config key → modeler register path, for the register and for the importer
KEY_TO_PATH = {
"freeze.start_week": "freeze.startWeek",
"freeze.end_week": "freeze.length",
"attrition.base_annual": "attrition.annual",
"attrition.freeze_multiplier": "attrition.tensionMult",
"attrition.post_multiplier_release": "attrition.postMult",
"shrinkage.post_announcement_add": "after.surgePts",
"transfer.training_hours": "after.trainingHours",
"release.notice_weeks": "after.noticeWeeks",
"service.patience_seconds": "service.patience.voice",
"service.patience_chat_seconds": "service.patience.chat",
"service.redial_rate": "service.redialRate",
# format 2
"attrition.post_multiplier_transfer": "people.postMultTransfer",
"attrition.retention_effect": "people.retentionEffect",
}
# format-2 structured blocks: register path (status only) → pack key
BOOK_KEYS = {
"book.priors": "fate_priors",
"book.exitNotice": "exit.notice_weeks",
"book.replatformOffset": "replatform.offset_weeks",
}
CONTRACT_KIND = {"fixed": "fixed", "evergreen": "evergreen", "evergreen_tfc": "tfc"}
HEALTHS = ("green", "amber", "red")
FATES = ("transfer", "exit", "replatform")
# --------------------------------------------------------------------------
# helpers
# --------------------------------------------------------------------------
def _get(cfg, dotted, default=None):
node = cfg
for k in dotted.split("."):
if not isinstance(node, dict) or k not in node:
return default
node = node[k]
return node
def _triple(spec, scale=1.0, shift=0.0):
"""A pack PERT spec (number or [low, likely, high]) as a modeler range."""
if isinstance(spec, (int, float)):
v = float(spec) * scale + shift
return [v, v, v]
lo, mode, hi = (float(x) * scale + shift for x in spec)
return [lo, mode, hi]
def _default_config():
return sim.demo_config()
def status_of(cfg, key, defaults=None):
"""(status, owner, note) for a pack key: the status block first, else default vs changed."""
blk = cfg.get("status") or {}
ent = blk.get(key)
if isinstance(ent, str):
st, owner, note = ent, None, None
elif isinstance(ent, dict):
st, owner, note = ent.get("status", "estimated"), ent.get("owner"), ent.get("note")
else:
defaults = defaults or _default_config()
st = "default" if _get(cfg, key) == _get(defaults, key) else "estimated"
owner, note = None, None
if st not in STATUSES:
raise ValueError(f"status[{key}] must be one of {STATUSES}, got {st!r}")
return st, owner, note
def _entry(cfg, key, spec, defaults, scale=1.0, shift=0.0, note=None):
st, owner, blk_note = status_of(cfg, key, defaults)
rng = _triple(spec, scale, shift)
e = {"range": rng, "status": st}
if owner:
e["owner"] = str(owner)[:MAX_TEXT]
text = blk_note or note
if text:
e["note"] = str(text)[:MAX_TEXT]
return e
def _channel(pool):
if pool["mode"] == "deferrable":
return "email"
return "chat" if pool.get("channel") == "chat" else "voice"
def _merge_events(events, max_n):
"""[(week, share)] merged into at most max_n by joining the smallest adjacent pairs (share-weighted week)."""
ev = [[int(w), float(s)] for w, s in sorted(events) if s > 1e-6]
while len(ev) > max_n:
k = min(range(len(ev) - 1), key=lambda i: ev[i][1] + ev[i + 1][1])
a, b = ev[k], ev[k + 1]
tot = a[1] + b[1]
ev[k] = [int(round((a[0] * a[1] + b[0] * b[1]) / tot)), tot]
del ev[k + 1]
return ev
def _conditional(steps, waves, fz_end):
"""Original-book shares → the modeler's inputs, in its order (steps, then waves, by week).
steps: [(absolute week, share)] → step-down pct = share of the work still there (the modeler
multiplies its existing-book factor by 1 − pct).
waves: [(weeks after freeze end, share)] → wave pct = share of the whole book σ_j, because the
modeler itself converts σ to a conditional share π_j = σ_j ÷ (1 − Σ_{i<j} σ_i) at run time.
Both are chosen so that existing × Π(1 − π) equals the pack's remaining share after each event.
"""
order = sorted([(w, 0, i) for i, (w, _) in enumerate(steps)] + [(fz_end + w, 1, i) for i, (w, _) in enumerate(waves)])
left, out_s, out_w, sigma_taken = 1.0, [0.0] * len(steps), [0.0] * len(waves), 0.0
for w, kind, i in order:
share = (steps if kind == 0 else waves)[i][1]
pi = min(0.999, share / max(left, 1e-9)) # share of what is still there
if kind == 0:
out_s[i] = pi
else:
sigma = pi * (1 - sigma_taken) # what the modeler must be told to get π back
out_w[i] = sigma
sigma_taken += sigma
left = max(0.0, left - share)
return out_s, out_w
# --------------------------------------------------------------------------
# format 2: the book and the people split
# --------------------------------------------------------------------------
def _client_hours(cfg, res):
"""Base workload hours per client at week 0 (the exporter's hours-per-unit convention)."""
pidx = {p: i for i, p in enumerate(res.pools)}
hpu = hours_per_unit(cfg, res)
return np.array([c["annual_volume"] / 52 * hpu[pidx[c["pool"]]] for c in cfg["clients"]])
def _share_mix(keys, weights, key_of):
mix = {k: 0.0 for k in keys}
for w, k in zip(weights, key_of):
if k in mix:
mix[k] += w
tot = sum(mix.values()) or 1.0
out = {k: round(v / tot, 4) for k, v in mix.items()}
big = max(out, key=out.get)
out[big] = round(out[big] + (1.0 - sum(out.values())), 4) # rounding residual, so the shares sum to exactly 1
return out
def book_block(cfg, res):
"""The book as shares, weeks and probabilities: nothing per client survives."""
base_h = _client_hours(cfg, res)
clients = cfg["clients"]
contract_mix = _share_mix(("fixed", "evergreen", "tfc"), base_h,
[CONTRACT_KIND.get(c.get("contract", "evergreen"), "evergreen") for c in clients])
health_mix = _share_mix(HEALTHS, base_h, [str(c.get("health", "green")) for c in clients])
expiries = [int(c["expiry_week"]) for c in clients if c.get("contract") == "fixed" and c.get("expiry_week") is not None]
fixed_expiry = [min(expiries), max(max(expiries), min(expiries) + 1)] if expiries else [0, int(cfg["horizon_weeks"])]
priors = {h: {f: float(cfg["fate_priors"][h][f]) for f in FATES} for h in HEALTHS}
# waves as shares of transferring work: expected transfers per wave over the pack's draws
# (the pack's fate array already includes exits that arrive too late and transfer instead)
waves = {}
for j, c in enumerate(clients):
off = int(cfg["waves"][c.get("wave") or cfg["pools"][c["pool"]]["wave"]])
waves[off] = waves.get(off, 0.0) + base_h[j] * float((res.fate[:, j] == sim.T_TRANSFER).mean())
tot = sum(waves.values()) or 1.0
wave_list = _merge_events([(off, s / tot) for off, s in waves.items()], MAX_WAVES)
ex = cfg["exit"]["notice_weeks"]
return {
"mode": "book",
"contractMix": contract_mix,
"fixedExpiry": [int(fixed_expiry[0]), int(fixed_expiry[1])],
"healthMix": health_mix,
"priors": priors,
"exitNotice": {"evergreen": _triple(ex["evergreen"]), "tfc": _triple(ex.get("evergreen_tfc", ex["evergreen"]))},
"replatformOffset": _triple(cfg["replatform"]["offset_weeks"]),
"waves": [{"weeksAfterFreeze": int(off), "pct": round(float(p), 4)} for off, p in wave_list],
"waveSlip": [0.0, 0.0, 0.0],
"granularity": GRANULARITY,
}
def people_block(cfg):
att = cfg["attrition"]
return {
"split": True,
"postMultTransfer": max(0.5, _triple(att.get("post_multiplier_transfer", att["post_multiplier_release"]))[1]),
"postMultRelease": _triple(att["post_multiplier_release"])[1],
"retentionEffect": float(att.get("retention_effect", 0.0)),
"retentionTarget": "release",
}
def _status_only(cfg, key, defaults, note=None):
st, owner, blk_note = status_of(cfg, key, defaults)
e = {"status": st}
if owner:
e["owner"] = str(owner)[:MAX_TEXT]
if blk_note or note:
e["note"] = str(blk_note or note)[:MAX_TEXT]
return e
def _book_assumptions(cfg, defaults, book, people):
"""Register entries for the book and the split: mixes as estimated ranges (they come from the client
list, but each client's fate does not), priors and ranges status-only, multipliers ranged."""
att = cfg["attrition"]
out = {}
for grp, keys in (("contractMix", ("fixed", "evergreen", "tfc")), ("healthMix", ("green", "amber", "red"))):
for k in keys:
v = float(book[grp][k])
out[f"book.{grp}.{k}"] = {"range": [round(max(0.0, v - 0.1), 4), round(v, 4), round(min(1.0, v + 0.1), 4)],
"status": "estimated", "note": _t("Workload share from the client list")}
out["book.fixedExpiry"] = {"status": "estimated", "note": _t("Expiry weeks of the fixed-term contracts in the client list")}
out["book.priors"] = _status_only(cfg, "fate_priors", defaults, note=_t("Fate probabilities by relationship health"))
out["book.exitNotice"] = _status_only(cfg, "exit.notice_weeks", defaults)
out["book.replatformOffset"] = _status_only(cfg, "replatform.offset_weeks", defaults)
out["book.waveSlip"] = {"status": "default", "note": _t("The pack schedules waves at fixed offsets; slip is the modeler's lever")}
pt = _triple(att.get("post_multiplier_transfer", att["post_multiplier_release"]))
pt = [max(0.5, x) for x in pt] # the modeler accepts a transfer group calmer than baseline, down to 0.5
out["people.postMultTransfer"] = _entry(cfg, "attrition.post_multiplier_transfer", pt, defaults,
note=_t("Transfer group (values below 1 in the pack are floored at 1 here)"))
out["people.postMultRelease"] = _entry(cfg, "attrition.post_multiplier_release", att["post_multiplier_release"], defaults, note=_t("Release group"))
ret = float(att.get("retention_effect", 0.0))
out["people.retentionEffect"] = {"range": [ret, ret, ret], "status": "default" if ret == 0 else "estimated",
"note": _t("Lever, not a fact: the reduction a retention offer buys")}
return out
def _pert_grid(spec, n=QUANTILE_GRID):
"""PERT(lo, mode, hi) on a midpoint quantile grid (the modeler uses the same grid)."""
lo, mode, hi = (float(x) for x in spec)
q = (np.arange(n) + 0.5) / n
if hi <= lo:
return np.full(n, lo)
a = 1 + 4 * (mode - lo) / (hi - lo)
b = 1 + 4 * (hi - mode) / (hi - lo)
return lo + (hi - lo) * stats.beta.ppf(q, a, b)
def expected_curve_from_book(book, fz_end, W, slip=0.0):
"""The modeler's closed-form expected share of the book still at the source, by week.
Cells (contract × health) with workload share s_c and fate probabilities from the health prior.
Every client-equivalent belongs to a wave j with probability pct_j (share of transferring work),
cutting over at fz_end + offset_j + slip. Transfer leaves at the wave; an exit leaves at its
notice week, or transfers at the wave if the notice runs past it (the pack's late-exit rule);
re-platformed work leaves at min(fz_end + offset, wave). So every departure is ≤ the wave week
and P(dep > w) = P(event > w) · 1[w < wave].
"""
weeks = np.arange(W)
waves = book["waves"]
tot = sum(x["pct"] for x in waves) or 1.0
wave_p = [x["pct"] / tot for x in waves]
wave_wk = [int(round(fz_end + x["weeksAfterFreeze"] + slip)) for x in waves]
ex_grid = {"evergreen": np.round(fz_end + _pert_grid(book["exitNotice"]["evergreen"])),
"tfc": np.round(fz_end + _pert_grid(book["exitNotice"]["tfc"]))}
lo, hi = book["fixedExpiry"]
ex_grid["fixed"] = np.round(lo + (hi - lo) * (np.arange(QUANTILE_GRID) + 0.5) / QUANTILE_GRID)
rp_grid = np.round(fz_end + _pert_grid(book["replatformOffset"]))
out = np.zeros(W)
for k, sk in book["contractMix"].items():
for h, sh in book["healthMix"].items():
s = sk * sh
if s <= 0:
continue
pr = book["priors"][h]
p_exit_gt = (ex_grid[k][None, :] > weeks[:, None]).mean(1) # P(exit week > w)
p_rp_gt = (rp_grid[None, :] > weeks[:, None]).mean(1)
for pj, wk in zip(wave_p, wave_wk):
before = (weeks < wk).astype(float)
out += s * pj * before * (pr["transfer"] + pr["exit"] * p_exit_gt + pr["replatform"] * p_rp_gt)
return out
# --------------------------------------------------------------------------
# the card
# --------------------------------------------------------------------------
def departures(cfg, res):
"""Expected departure shares of the week-0 book: waves [(offset_after_freeze, share)] and step-downs [(week, share)].
Freeze-relative events (transfers, evergreen exits, re-platforming) are placed relative to the
median freeze end; fixed-term expiries keep their absolute week. Shares are hours-weighted.
"""
W = int(cfg["horizon_weeks"])
N = res.fate.shape[0]
pools = res.pools
pidx = {p: i for i, p in enumerate(pools)}
hpu = hours_per_unit(cfg, res)
base_h = np.array([c["annual_volume"] / 52 * hpu[pidx[c["pool"]]] for c in cfg["clients"]])
share = base_h / max(base_h.sum(), 1e-12)
fz_med = int(np.median(res.freeze_end))
waves, drops = {}, np.zeros(W)
for j, c in enumerate(cfg["clients"]):
is_t = res.fate[:, j] == sim.T_TRANSFER
wave = c.get("wave") or cfg["pools"][c["pool"]]["wave"]
off = int(cfg["waves"][wave])
waves[off] = waves.get(off, 0.0) + share[j] * is_t.mean()
fixed = c.get("contract", "evergreen") == "fixed" and c.get("expiry_week") is not None
gone = ~is_t
if not gone.any():
continue
if fixed:
wk = res.dep[gone, j]
else:
wk = res.dep[gone, j] - res.freeze_end[gone] + fz_med
for w in np.clip(wk, 0, W - 1):
drops[int(w)] += share[j] / N
wave_list = _merge_events(list(waves.items()), MAX_WAVES)
step_list = _merge_events([(w, d) for w, d in enumerate(drops)], MAX_STEPS)
return wave_list, step_list, fz_med
def hours_per_unit(cfg, res):
"""Workload hours per unit of volume per pool: exporter convention (utilisation × productive hours)."""
paid = float(cfg["calendar"]["paid_hours_week"])
F0 = np.nanmedian(res.F[:, :, 0], axis=0)
U0 = np.nanmedian(res.U[:, :, 0], axis=0)
work0 = np.nan_to_num(U0 * F0 * paid)
D0 = np.nanmedian(res.D[:, :, 0], axis=0)
return np.where(D0 > 0, work0 / np.maximum(D0, 1e-12), 0.0)
def expected_curve(cfg, res):
"""The pack's expected share of week-0 workload still at the source, by week (hours-weighted)."""
W = int(cfg["horizon_weeks"])
pidx = {p: i for i, p in enumerate(res.pools)}
hpu = hours_per_unit(cfg, res)
base_h = np.array([c["annual_volume"] / 52 * hpu[pidx[c["pool"]]] for c in cfg["clients"]])
active = res.dep[:, :, None] > np.arange(W)[None, None, :]
return (active * base_h[None, :, None]).sum(1).mean(0) / max(base_h.sum(), 1e-12)
def fidelity(cfg, res, card, draws=None):
"""How well the card's departures reproduce the pack, with the freeze held at its median draw
(the pack's own mean over drawn freeze ends differs by the freeze smear, which the modeler
reproduces by drawing freeze.length itself). Returns a dict of max absolute differences as
shares of the week-0 book."""
fz_med = int(np.median(res.freeze_end))
cfg_fixed = copy.deepcopy(cfg)
cfg_fixed["freeze"]["end_week"] = fz_med
res_fixed = sim.run(cfg_fixed, draws or res.fate.shape[0])
exp_fixed = expected_curve(cfg_fixed, res_fixed)
exp_mean = expected_curve(cfg, res)
out = {"freezeMedian": fz_med,
"staircaseVsFixed": float(np.abs(rebuild_curve(card) - exp_fixed).max()),
"freezeSmear": float(np.abs(exp_mean - exp_fixed).max()),
"endResidual": float(rebuild_curve(card)[-1])}
if "book" in card["scenario"]:
book = expected_curve_from_book(card["scenario"]["book"], fz_med, int(cfg["horizon_weeks"]))
out["bookVsFixed"] = float(np.abs(book - exp_fixed).max())
return out
def rebuild_curve(card):
"""What the modeler's demand pass makes of a card: existing × Π(1−step) × Π(1−wave)."""
sc = card["scenario"]
W = sc["horizonWeeks"]
fz_end = sc["freeze"]["endWeek"]
# waves: shares of the book σ, converted as the engine does (scheduledWaves) to conditional π
waves, taken = {}, 0.0
for w in sorted(sc["after"]["waves"], key=lambda x: x["weeksAfterFreeze"]):
s = min(w["pct"], 1 - taken)
pi = s / (1 - taken) if 1 - taken > 1e-12 else 0.0
taken += s
wk = fz_end + w["weeksAfterFreeze"]
waves[wk] = 1 - (1 - waves.get(wk, 0.0)) * (1 - pi)
steps = {}
for s in sc["demand"]["stepDowns"]:
steps[s["week"]] = 1 - (1 - steps.get(s["week"], 0.0)) * (1 - s["pct"])
existing, wm, out = 1.0, 1.0, []
for w in range(W):
existing *= 1 - steps.get(w, 0.0)
wm *= 1 - waves.get(w, 0.0)
out.append(existing * wm)
return np.array(out)
def shape_card(cfg, res, title=DEFAULT_TITLE, fmt=2):
if fmt not in (1, 2):
raise ValueError("format must be 1 or 2")
pools = res.pools
pcfg = [cfg["pools"][p] for p in pools]
paid = float(cfg["calendar"]["paid_hours_week"])
W = int(cfg["horizon_weeks"])
defaults = _default_config()
_check_title(title, cfg)
# week 0 workload hours per pool: utilisation × productive hours (median over draws)
F0 = np.nanmedian(res.F[:, :, 0], axis=0)
U0 = np.nanmedian(res.U[:, :, 0], axis=0)
work0 = np.nan_to_num(U0 * F0 * paid)
mix = {"voice": 0.0, "chat": 0.0, "email": 0.0}
for i, p in enumerate(pcfg):
mix[_channel(p)] += work0[i]
total_work = sum(mix.values()) or 1.0
mix = {k: v / total_work for k, v in mix.items()}
occupancy = float(work0.sum() / max(F0.sum() * paid, 1e-9))
def wavg(ch, key, default):
idx = [i for i, p in enumerate(pcfg) if _channel(p) == ch and key in p]
if not idx:
return default
return float(np.average([pcfg[i][key] for i in idx], weights=np.array([work0[i] for i in idx]) + 1e-12))
# departures as conditional shares
wave_list, step_list, fz_med = departures(cfg, res)
s_pct, w_pct = _conditional(step_list, wave_list, fz_med)
waves = [{"weeksAfterFreeze": int(off), "pct": round(float(p), 4)} for (off, _), p in zip(wave_list, w_pct)]
step_downs = [{"week": int(w), "pct": round(float(p), 4)} for (w, _), p in zip(step_list, s_pct)]
att, tr, rel, svc = cfg["attrition"], cfg["transfer"], cfg["release"], cfg.get("service", {})
start = int(cfg["freeze"].get("start_week", 0))
heads = np.array([p["headcount"] for p in pcfg], float)
shrink = float(np.average([p["shrinkage"] for p in pcfg], weights=heads + 1e-12))
ov = [i for i, p in enumerate(pcfg) if p.get("overflow_fte", 0) > 0]
borrowed = float(sum(pcfg[i]["overflow_fte"] for i in ov))
surge = _get(cfg, "shrinkage.post_announcement_add", 0.0)
# service settings are workload-weighted over the pools of each channel, so per-pool overrides
# (a chat pool's patience, a pool on Erlang C) reach the card
svc_by_pool = {p: sim.service_model(cfg, p) for p in pools if cfg["pools"][p]["mode"] == "interactive"}
def svc_avg(ch, key, default):
idx = [i for i, p in enumerate(pools) if p in svc_by_pool and _channel(cfg["pools"][p]) == ch]
if not idx:
return default
return round(float(np.average([float(svc_by_pool[pools[i]][key]) for i in idx], weights=np.array([work0[i] for i in idx]) + 1e-12)), 6)
pat_v = svc_avg("voice", "patience_seconds", float(svc.get("patience_seconds", 120)))
pat_c = float(svc.get("patience_chat_seconds", svc_avg("chat", "patience_seconds", 300.0)))
work_a = sum(work0[i] for i, p in enumerate(pools) if p in svc_by_pool and svc_by_pool[p]["model"] == "A")
work_c = sum(work0[i] for i, p in enumerate(pools) if p in svc_by_pool and svc_by_pool[p]["model"] == "C")
model = "C" if work_c > work_a else "A" # the model most of the interactive workload runs on
redial = svc_avg("voice", "redial_rate", float(svc.get("redial_rate", 0.4)))
cap = svc_avg("voice", "abandon_cap", float(svc.get("abandon_cap", 0.10)))
scenario = {
"horizonWeeks": int(min(78, max(13, W))),
"channels": {
"voice": {"aht": round(wavg("voice", "aht_seconds", 360.0)), "slTarget": 0.8,
"slSeconds": int(round(wavg("voice", "asa_seconds", 20.0)))},
"chat": {"aht": round(wavg("chat", "aht_seconds", 600.0)), "slTarget": 0.8,
"slSeconds": int(round(wavg("chat", "asa_seconds", 60.0)))},
"email": {"targetDays": float(wavg("email", "target_days", 1.0))},
},
"pool": {"shrinkage": round(shrink, 4), "paidHours": paid, "openHours": float(cfg["calendar"]["open_hours_week"])},
"profile": {"volumeShare": [float(x) for x in cfg["calendar"]["buckets"]["volume_share"]],
"hourShare": [float(x) for x in cfg["calendar"]["buckets"]["hour_share"]],
"scheduleFit": float(cfg["calendar"]["schedule_fit"])},
"attrition": {"annual": float(att["base_annual"]), "tensionMult": _triple(att["freeze_multiplier"])[1],
"postMult": _triple(att["post_multiplier_release"])[1]},
"freeze": {"startWeek": start, "endWeek": int(fz_med), "backfillBefore": False},
"demand": {"runoffPctWeek": 0.0, "stepDowns": step_downs, "intakeOn": False},
"after": {"waves": waves, "trainingHours": float(tr["training_hours"]), "trainingWeeks": int(tr["training_weeks"]),
"surgePts": _triple(surge)[1], "surgeWeeks": 52,
"releasesOn": bool(rel.get("allowed", True)), "noticeWeeks": int(rel["notice_weeks"]),
"releaseBuffer": float(rel.get("buffer", 0.05)), "lookaheadWeeks": int(rel.get("lag_weeks", 2))},
"borrowed": {"startWeek": int(min(pcfg[i].get("overflow_start_week", 0) for i in ov)) if ov else 8,
"endWeek": int(min(77, W - 1)),
"ahtPenalty": float(np.average([pcfg[i].get("overflow_aht_mult", 1.0) for i in ov],
weights=[pcfg[i]["overflow_fte"] for i in ov])) if ov else 1.2},
"service": {"model": model, "patience": {"voice": pat_v, "chat": pat_c}, "redialRate": redial, "abandonCap": cap},
}
if fmt == 2:
scenario["book"] = book_block(cfg, res)
scenario["people"] = people_block(cfg)
end_rng = _triple(cfg["freeze"]["end_week"])
assumptions = {
"freeze.startWeek": _entry(cfg, "freeze.start_week", start, defaults),
# the register's likely value must equal the scenario's (the median draw), so the modeler keeps the range
"freeze.length": _entry(cfg, "freeze.end_week",
[max(0, min(end_rng[0], fz_med) - start), fz_med - start, max(0, max(end_rng[2], fz_med) - start)], defaults,
note=_t("From the pack's freeze end range; the likely value is the median draw")),
"attrition.annual": _entry(cfg, "attrition.base_annual", att["base_annual"], defaults),
"attrition.tensionMult": _entry(cfg, "attrition.freeze_multiplier", att["freeze_multiplier"], defaults),
"attrition.postMult": _entry(cfg, "attrition.post_multiplier_release", att["post_multiplier_release"], defaults,
note=_t("Release group; the transfer group is milder in the pack")),
"after.surgePts": _entry(cfg, "shrinkage.post_announcement_add", surge, defaults),
"after.trainingHours": _entry(cfg, "transfer.training_hours", tr["training_hours"], defaults),
"after.noticeWeeks": _entry(cfg, "release.notice_weeks", rel["notice_weeks"], defaults),
"service.patience.voice": _entry(cfg, "service.patience_seconds", pat_v, defaults),
"service.patience.chat": _entry(cfg, "service.patience_chat_seconds", pat_c, defaults),
"service.redialRate": _entry(cfg, "service.redial_rate", redial, defaults),
"after.waves": {"status": "estimated", "note": _t("Transferred work by wave, expected over the pack's draws")},
"demand.stepDowns": {"status": "estimated", "note": _t("Exits and re-platforming, expected over the pack's draws")},
}
if fmt == 2:
assumptions.update(_book_assumptions(cfg, defaults, scenario["book"], scenario["people"]))
# forecast error: the pools' 90% volume intervals, workload-weighted per channel
for ch in ("voice", "chat", "email"):
idx = [i for i, p in enumerate(pcfg) if _channel(p) == ch and isinstance(p.get("volume_error"), (list, tuple))]
if idx and mix[ch] > 0:
w = np.array([work0[i] for i in idx]) + 1e-12
lo = float(np.average([pcfg[i]["volume_error"][0] for i in idx], weights=w))
hi = float(np.average([pcfg[i]["volume_error"][1] for i in idx], weights=w))
assumptions[f"channels.{ch}.volume"] = {"range": [lo, 1.0, hi], "status": "default",
"note": _t("Forecast error from the pack's volume_error (relative to the rebuilt volume)")}
card = {
"format": FORMAT_2 if fmt == 2 else FORMAT_1,
"title": title[:120],
"created": date.today().isoformat(),
"note": CARD_NOTE if fmt == 1 else CARD_NOTE_2,
"occupancy": round(occupancy, 4),
"mix": {k: round(v, 4) for k, v in mix.items()},
"borrowedShare": round(borrowed / max(heads.sum(), 1e-9), 4),
"stepDownsRelativeToFreezeEnd": False,
"scenario": scenario,
"assumptions": assumptions,
"questions": [],
"source": {"tool": _t("CP-WFM-019 pack"), "version": VERSION, "date": date.today().isoformat()},
}
validate_card(card, cfg)
return card
# --------------------------------------------------------------------------
# leak guard: allow-list schema, then a sweep for anything identifying
# --------------------------------------------------------------------------
# free-text leaves of a card (everything else is a number, a flag, a format string or a date)
FREE_TEXT = ("title", "note", "source.tool")
FREE_TEXT_UNDER = ("assumptions", "questions")
def name_tokens(cfg):
"""Letter runs of three or more from every client id, pool name and wave name (lower case)."""
names = [str(c.get("id", "")) for c in cfg.get("clients", [])] + list(cfg.get("pools", {})) + list(cfg.get("waves", {}))
return {t for n in names for t in re.findall(r"[a-z]+", n.lower()) if len(t) >= 3}
def check_text(text, tokens, where="text"):
"""Refuses free text that could identify the program: a run of three or more digits (weeks and
small figures pass), or any client/pool/wave name token as a substring once everything but
letters is stripped ("AcmeCorp", "Acme_Corp" and "ACME" all match "acme")."""
if not isinstance(text, str):
return
if re.search(r"\d{3,}", text):
raise ValueError(f"{where} contains a figure ({text[:40]!r}); refusing")
letters = re.sub(r"[^a-z]", "", text.lower())
for tok in tokens:
if tok in letters:
raise ValueError(f"{where} contains a client, pool or wave name ({tok!r}); refusing")
def _free_text_leaves(card):
for key in FREE_TEXT:
v = _get(card, key)
if v is not None:
yield key, v
for path, a in (card.get("assumptions") or {}).items():
for k in ("owner", "note"):
if isinstance(a, dict) and a.get(k) is not None:
yield f"assumptions.{path}.{k}", a[k]
for i, q in enumerate(card.get("questions") or []):
for k in ("id", "text"):
if isinstance(q, dict) and q.get(k) is not None:
yield f"questions[{i}].{k}", q[k]
def _check_title(title, cfg):
if title == DEFAULT_TITLE:
return
check_text(title, name_tokens(cfg), "title")
def _walk(value, schema, path, lenient=False):
if isinstance(schema, dict):
if not isinstance(value, dict):
raise ValueError(f"{path}: expected an object")
wild = schema.get("*")
for k, v in value.items():
sub = schema.get(k, wild)
if sub is None:
if lenient:
continue # incoming cards (e.g. from the online modeler) carry keys the pack does not read
raise ValueError(f"{path}.{k}: not in the card schema; refusing to write it")
_walk(v, sub, f"{path}.{k}", lenient)
elif isinstance(schema, list):
if not isinstance(value, list):
raise ValueError(f"{path}: expected a list")
for i, v in enumerate(value):
_walk(v, schema[0], f"{path}[{i}]", lenient)
elif schema == NUM:
if isinstance(value, bool) or not isinstance(value, (int, float)) or not np.isfinite(value):
raise ValueError(f"{path}: expected a finite number")
elif schema == INT:
if isinstance(value, bool) or not isinstance(value, int):
raise ValueError(f"{path}: expected an integer")
elif schema == BOOL:
if not isinstance(value, bool):
raise ValueError(f"{path}: expected true/false")
elif schema == STR:
if not isinstance(value, str) or len(value) > MAX_TEXT:
raise ValueError(f"{path}: expected text of at most {MAX_TEXT} characters")
def _all_strings(value, path="card"):
if isinstance(value, str):
yield path, value
elif isinstance(value, dict):
for k, v in value.items():
yield from _all_strings(v, f"{path}.{k}")
elif isinstance(value, list):
for i, v in enumerate(value):
yield from _all_strings(v, f"{path}[{i}]")
def validate_card(card, cfg=None, incoming=False):
"""Refuses a card that carries anything outside the allow-list or anything identifying.
incoming=True (the importer): keys the pack does not read are ignored rather than refused, since
cards exported by the online modeler carry its whole scenario; every string in the card is still
checked against the local names, and the known keys are still type-checked."""
if not isinstance(card, dict):
raise ValueError("a shape card is a JSON object")
for key in ("format", "scenario", "assumptions"):
if key not in card:
raise ValueError(f"not a shape card: missing '{key}'")
_walk(card, SCHEMA, "card", lenient=incoming)
if incoming: # lenient on keys the pack doesn't read, never on scale
sc = card.get("scenario") or {}
for ch in ("voice", "chat", "email"):
if isinstance((sc.get("channels") or {}).get(ch), dict) and "volume" in sc["channels"][ch]:
raise ValueError(f"card.scenario.channels.{ch}.volume: a shape card carries no volumes; refusing")
for blk in ("pool", "borrowed"):
if isinstance(sc.get(blk), dict) and "fte" in sc[blk]:
raise ValueError(f"card.scenario.{blk}.fte: a shape card carries no headcount; refusing")
if card.get("format") not in FORMATS:
raise ValueError(f"unknown card format {card.get('format')!r}")
if "book" in card["scenario"]:
_check_book(card["scenario"]["book"])
if "people" in card["scenario"]:
pp = card["scenario"]["people"]
if pp["retentionTarget"] not in ("release", "transfer", "both") or not 0 <= pp["retentionEffect"] <= 1:
raise ValueError("people: retentionTarget must be release/transfer/both and retentionEffect 0-1")
if pp["postMultTransfer"] < 0.5 or pp["postMultRelease"] < 1:
raise ValueError("people: the transfer-group multiplier must be >= 0.5 and the release-group >= 1")
for path, a in card["assumptions"].items():
if a["status"] not in STATUSES:
raise ValueError(f"assumptions.{path}: bad status {a['status']!r}")
if cfg is not None:
tokens = name_tokens(cfg)
leaves = _all_strings(card) if incoming else _free_text_leaves(card)
for where, text in leaves:
if incoming and (where.endswith(".created") or where.endswith(".date")) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
continue # dates stamped on the card
if incoming and (where == "card.format" or where.endswith(".status") or where.endswith(".mode")
or where.endswith(".model") or where.endswith(".retentionTarget")
or where.startswith("card.scenario.balance.order")):
continue # enum values, not free text
if text not in TOOL_TEXT: # the exporter's own labels are generic; anything typed by a person is checked
check_text(text, tokens, where)
text = json.dumps(card).lower()
names = [str(c.get("id", "")) for c in cfg.get("clients", [])] + list(cfg.get("pools", {})) + list(cfg.get("waves", {}))
for n in names:
if n and len(n) > 1 and f'"{n.lower()}"' in text:
raise ValueError(f"card would carry the name {n!r}; refusing")
# second line behind the schema: no headcount or volume figure of this configuration may
# appear as a value. Leaves that hold weeks, seconds, hours, shares or counts of steps are
# skipped: a headcount of 120 must not be confused with a 120-second patience.
figures = {float(p["headcount"]) for p in cfg.get("pools", {}).values() if "headcount" in p} | \
{float(c["annual_volume"]) for c in cfg.get("clients", []) if "annual_volume" in c} | \
{round(float(c["annual_volume"]) / 52, 4) for c in cfg.get("clients", []) if "annual_volume" in c}
for path, v in _leaf_numbers(card):
if v in figures and not UNIT_LEAF.search(path) and v > 1:
raise ValueError(f"card would carry the figure {v} ({path}); refusing")
return card
def _check_book(book):
if book["mode"] not in ("manual", "book"):
raise ValueError("book.mode must be manual or book")
for grp in ("contractMix", "healthMix"):
vals = list(book[grp].values())
if any(v < 0 or v > 1 for v in vals) or abs(sum(vals) - 1) > 1e-3:
raise ValueError(f"book.{grp}: shares must be 0-1 and sum to 1")
for h, pr in book["priors"].items():
vals = list(pr.values())
if any(v < 0 or v > 1 for v in vals) or abs(sum(vals) - 1) > 1e-3:
raise ValueError(f"book.priors.{h}: probabilities must be 0-1 and sum to 1")
for name, t in (("exitNotice.evergreen", book["exitNotice"]["evergreen"]), ("exitNotice.tfc", book["exitNotice"]["tfc"]),
("replatformOffset", book["replatformOffset"]), ("waveSlip", book["waveSlip"])):
if len(t) != 3 or not all(np.isfinite(t)) or not (t[0] <= t[1] <= t[2]):
raise ValueError(f"book.{name}: expected [low, likely, high] with low <= likely <= high")
if len(book["fixedExpiry"]) != 2 or book["fixedExpiry"][0] > book["fixedExpiry"][1]:
raise ValueError("book.fixedExpiry: expected [first, last] weeks")
if not 5 <= book["granularity"] <= 200:
raise ValueError("book.granularity must be 5-200")
if len(book["waves"]) > MAX_WAVES or any(w["pct"] < 0 or w["pct"] > 1 for w in book["waves"]):
raise ValueError("book.waves: at most 4 waves with shares 0-1")
# leaves whose values are durations, timings or handle times, never headcount or volume
UNIT_LEAF = re.compile(r"(week|seconds|aht|patience|hours|expiry|granularity|slip|notice|offset|horizon|lookahead)", re.I)
def _leaf_numbers(x, path="card"):
"""(dotted path, value) for every numeric leaf."""
if isinstance(x, bool):
return []
if isinstance(x, (int, float)):
return [(path, float(x))]
if isinstance(x, dict):
return [v for k, y in x.items() for v in _leaf_numbers(y, f"{path}.{k}")]
if isinstance(x, list):
return [v for i, y in enumerate(x) for v in _leaf_numbers(y, f"{path}[{i}]")]
return []
# every literal the exporter ever writes into a free-text field is registered at import, so the
# importer (which never runs shape_card) exempts the same labels
for _lit in re.findall(r'_t\("([^"]*)"\)', open(__file__, encoding="utf-8").read()):
TOOL_TEXT.add(_lit)
if __name__ == "__main__":
args = sys.argv[1:]
out = args[args.index("--out") + 1] if "--out" in args else "shape_card.json"
draws = int(args[args.index("--draws") + 1]) if "--draws" in args else 500
title = args[args.index("--title") + 1] if "--title" in args else DEFAULT_TITLE
fmt = int(args[args.index("--format") + 1]) if "--format" in args else 2
cfg_path = next((a for a in args if a.endswith((".yaml", ".yml"))), None)
cfg = sim.load_config(cfg_path) if cfg_path else sim.demo_config()
res = sim.run(cfg, draws)
card = shape_card(cfg, res, title=title, fmt=fmt)
with open(out, "w") as f:
json.dump(card, f, indent=2)
fid = fidelity(cfg, res, card)
print(f"wrote {out} ({card['format']}): occupancy {card['occupancy']:.2f}, mix { {k: round(float(v), 3) for k, v in card["mix"].items()} }, "
f"{len(card['scenario']['after']['waves'])} waves, {len(card['scenario']['demand']['stepDowns'])} step-downs")
print(f"fidelity, freeze held at its median week {fid['freezeMedian']}: the card's staircase is within "
f"{fid['staircaseVsFixed']:.3f} of the book of the pack's expected departures"
+ (f"; the book block's closed form within {fid['bookVsFixed']:.3f}" if "bookVsFixed" in fid else "")
+ f"; end residual {fid['endResidual']:.3f}.")
print(f"(The pack's mean over drawn freeze ends differs by up to {fid['freezeSmear']:.2f}: that is the freeze-length "
f"smear, which the modeler reproduces by drawing freeze.length in its own bands.)")
Block 9 — import_shape_card.py
"""Bring answers recorded in the WFM Labs Migration Scenario Modeler back into a pack config.
The modeler's assumption register (ranges with a status, owner and note) travels in a shape
card. This script applies those entries as OVERRIDES onto an existing params.yaml: it never
creates pools or clients, never touches headcount or volumes, and never overwrites a key the
config marks ``confirmed`` in its ``status:`` block unless ``--force`` is given.
python import_shape_card.py card.json params.yaml [--out params_updated.yaml] [--force] [--apply-aht]
Rules per register entry:
confirmed → the likely value is written as a fixed number
estimated → the [low, likely, high] range is written (a point range as a fixed number)
default → the key is left untouched (nobody answered)
The freeze length is re-based on the card's own freeze start (written too), whatever the order
of the card's keys. Note the one-time shift: the exporter carries the freeze end's *median draw*
as the likely value, so a round trip can move the mode by a week (14 → 15); it does not ratchet.
Incoming cards pass the same allow-list and text guard as outgoing ones (names and figures in
owner or note fields are refused), and `service.model` must be A or C.
Every write is listed with the previous value, the status, owner and note, and recorded in the
config's ``status:`` block so the next export carries it.
Mapped: freeze start and length, base attrition, freeze and post-announcement multipliers
(release and, from a format-2 card, transfer group), retention effect, absence surge, training
hours, release notice, patience, redial rate, and the service model and abandonment cap as
settings. From a format-2 card's book: the fate priors, the exit-notice and re-platform ranges
(each written only when its register entry is estimated or confirmed). With ``--apply-aht`` the
card's handle times are written to every interactive pool of the same channel. Reported but
never written (the workbook or the pack's own inputs are the truth): volumes, headcount, the
contract and health mixes, wave shares and slip, step-downs, per-client fates, destination
parameters.
"""
import copy
import json
import sys
import yaml
from export_shape_card import BOOK_KEYS, KEY_TO_PATH, STATUSES, validate_card
PATH_TO_KEY = {v: k for k, v in KEY_TO_PATH.items()}
SETTINGS = {"service.model": "service.model", "service.abandonCap": "service.abandon_cap"}
NEVER = ("channels.voice.volume", "channels.chat.volume", "channels.email.volume", "borrowed.fte", "pool.fte")
def _get(d, dotted, default=None):
node = d
for k in dotted.split("."):
if not isinstance(node, dict) or k not in node:
return default
node = node[k]
return node
def _set(d, dotted, value):
keys = dotted.split(".")
node = d
for k in keys[:-1]:
node = node.setdefault(k, {})
node[keys[-1]] = value
def _num(x):
return int(x) if float(x).is_integer() else float(x)
def _value_for(path, entry, start):
"""The pack value for a register entry, or None when the entry does not apply."""
if entry["status"] == "default" or "range" not in entry:
return None
lo, mode, hi = (float(x) for x in entry["range"])
if path == "freeze.length":
lo, mode, hi = lo + start, mode + start, hi + start
if entry["status"] == "confirmed" or lo == hi:
return _num(mode) # confirmed: the likely value is a fixed number; a point range likewise
return [_num(lo), _num(mode), _num(hi)]
def _book_value(path, book):
"""A pack value for a structured book block, in the pack's own shape."""
if path == "book.priors":
return {h: {f: _num(v) for f, v in pr.items()} for h, pr in book["priors"].items()}
if path == "book.exitNotice":
ex = book["exitNotice"]
return {"evergreen": [_num(x) for x in ex["evergreen"]], "evergreen_tfc": [_num(x) for x in ex["tfc"]]}
if path == "book.replatformOffset":
return [_num(x) for x in book["replatformOffset"]]
return None
def import_card(card, params, force=False, apply_aht=False):
"""Returns (updated params, provenance rows). Raises on a card that fails the allow-list or the
text guard (the local pools, waves and clients supply the names it must not carry)."""
validate_card(card, cfg=params, incoming=True)
model = _get(card, "scenario.service.model")
if model is not None and model not in ("A", "C"):
raise ValueError(f"scenario.service.model must be A or C, got {model!r}")
out = copy.deepcopy(params)
status_blk = dict(out.get("status") or {}) # written back only if something is recorded
rows = []
src = f"card {card.get('title', '')[:40]} ({card.get('created', '')})"
def guard(key, new_status):
cur = status_blk.get(key)
cur = cur.get("status") if isinstance(cur, dict) else cur
return cur == "confirmed" and new_status != "confirmed" and not force
# The freeze length is re-based on the freeze start that will be in force after the import: the
# card's start only if the card answers that question (estimated or confirmed) and it is written,
# otherwise the pack's own start. Independent of the order of the register's keys.
card_start = _get(card, "scenario.freeze.startWeek")
st_entry = (card.get("assumptions") or {}).get("freeze.startWeek") or {}
local_start = int(_get(params, "freeze.start_week", 0))
writes_start = card_start is not None and st_entry.get("status") in ("estimated", "confirmed") and not guard("freeze.start_week", st_entry.get("status"))
start = int(card_start) if writes_start else local_start
entries = sorted(card.get("assumptions", {}).items(), key=lambda kv: kv[0] != "freeze.startWeek")
for path, entry in entries:
if path in NEVER or entry.get("status") not in STATUSES:
continue
key = PATH_TO_KEY.get(path)
if key is None:
continue
value = _value_for(path, entry, start)
if value is None:
rows.append((key, _get(out, key), None, "default", "", "untouched"))
continue
if guard(key, entry["status"]):
rows.append((key, _get(out, key), value, entry["status"], entry.get("owner", ""), "kept: locally confirmed (use --force)"))
continue
old = _get(out, key)
_set(out, key, value)
status_blk[key] = {"status": entry["status"], **({"owner": entry["owner"]} if entry.get("owner") else {}),
**({"note": entry["note"]} if entry.get("note") else {}), "source": src}
rows.append((key, old, value, entry["status"], entry.get("owner", ""), entry.get("note", "")))
# format 2: structured book blocks, written when their register entry says someone answered
book = card.get("scenario", {}).get("book")
if isinstance(book, dict):
for path, key in BOOK_KEYS.items():
entry = card.get("assumptions", {}).get(path, {"status": "default"})
st = entry.get("status", "default")
value = _book_value(path, book)
if value is None:
continue
if st == "default":
rows.append((key, _get(out, key), None, "default", "", "untouched"))
continue
if guard(key, st):
rows.append((key, _get(out, key), value, st, entry.get("owner", ""), "kept: locally confirmed (use --force)"))
continue
old = _get(out, key)
_set(out, key, value)
status_blk[key] = {"status": st, **({"owner": entry["owner"]} if entry.get("owner") else {}),
**({"note": entry["note"]} if entry.get("note") else {}), "source": src}
rows.append((key, old, value, st, entry.get("owner", ""), entry.get("note", "")))
# the mixes, waves and slip are reported so a disagreement with the workbook is visible
for grp in ("contractMix", "healthMix"):
rows.append((f"book.{grp}", None, json.dumps(book.get(grp)), "report", "", "not written: the client list is the truth"))
rows.append(("book.waves", None, json.dumps(book.get("waves")), "report", "", "shares of transferring work; the pack keeps its wave offsets"))
if book.get("waveSlip") and any(book["waveSlip"]):
rows.append(("book.waveSlip", None, json.dumps(book["waveSlip"]), "report", "", "the pack has no wave slip; schedule the waves later instead"))
for path, key in SETTINGS.items():
value = _get(card.get("scenario", {}), path)
if value is None:
continue
old = _get(out, key)
if old != value:
_set(out, key, value)
rows.append((key, old, value, "setting", "", "from the card's scenario"))
if apply_aht:
for ch in ("voice", "chat"):
aht = _get(card.get("scenario", {}), f"channels.{ch}.aht")
if aht is None:
continue
for name, pool in out.get("pools", {}).items():
if pool.get("mode") != "interactive":
continue
pool_ch = "chat" if pool.get("channel") == "chat" else "voice"
if pool_ch == ch and pool.get("aht_seconds") != aht:
rows.append((f"pools.{name}.aht_seconds", pool.get("aht_seconds"), aht, "setting", "", "--apply-aht"))
pool["aht_seconds"] = aht
if status_blk:
out["status"] = status_blk
return out, rows
def provenance_table(rows):
lines = ["| Key | Before | After | Status | Owner | Note |", "|---|---|---|---|---|---|"]
for key, old, new, st, owner, note in rows:
lines.append(f"| `{key}` | {old} | {new if new is not None else '—'} | {st} | {owner or ''} | {note or ''} |")
return "\n".join(lines)
if __name__ == "__main__":
args = sys.argv[1:]
if len(args) < 2 or args[0].startswith("--"):
print(__doc__)
sys.exit(2)
card_path, params_path = args[0], args[1]
out_path = args[args.index("--out") + 1] if "--out" in args else "params_updated.yaml"
try:
with open(card_path) as f:
card = json.load(f)
with open(params_path) as f:
params = yaml.safe_load(f) or {}
updated, rows = import_card(card, params, force="--force" in args, apply_aht="--apply-aht" in args)
except (OSError, json.JSONDecodeError, yaml.YAMLError, ValueError, KeyError, TypeError) as e:
print(f"cannot import {card_path}: {e}")
sys.exit(1)
with open(out_path, "w") as f:
yaml.safe_dump(updated, f, sort_keys=False)
changed = [r for r in rows if r[2] is not None and r[3] != "report" and r[5] != "kept: locally confirmed (use --force)"]
print(f"wrote {out_path}: {len(changed)} key(s) changed, {len(rows) - len(changed)} left as they were\n")
print(provenance_table(rows))
See also
- Wiki:Packs/Migration Service-Level Simulation — the pack these modules belong to
- Service Level During Work Migration — the concepts
- Wiki:Packs — the pack index
