"""Validation checks for 03_Simulated. Writes SIMULATION_CHECKS.md + simulation_checks.json.""" import glob, json, os, datetime as dt import numpy as np import pandas as pd from sim_common import OUT, DATA, CSV, CLEAN, load_stats from parameters import tables R = [] def check(tid, name, ok, detail='', severity='FAIL'): R.append({'id': tid, 'test': name, 'status': 'PASS' if ok else severity, 'detail': detail}) print(f"[{'PASS' if ok else severity}] {tid} {name} :: {detail}") def main(): src, par = tables() # S01 basis labels missing = [] for p in glob.glob(os.path.join(DATA, '*.parquet')): t = os.path.basename(p)[:-8] if t in ('sim_parameters', 'sources'): continue if 'basis' not in pd.read_parquet(p).columns: missing.append(t) check('S01', 'Every simulated/inferred table carries a basis label column', not missing, f'missing: {missing}') # S02 parameters srcids = set(src.source_id) bad_src = [r.param_id for r in par.itertuples() if r.basis in ('Researched', 'Client') and not (set(str(r.sources).split(';')) & srcids)] bad_basis = par[~par.basis.isin(['Researched', 'Assumption', 'Client'])].param_id.tolist() order_ok = par[~par.param_id.isin(['M02'])].apply(lambda r: min(r.low, r.high) <= r.base <= max(r.low, r.high), axis=1).all() check('S02', 'Parameters: base within [low, high]; every Researched parameter cites a source in the register', not bad_src and not bad_basis and order_ok, f'researched without source: {bad_src}; bad basis: {bad_basis}; {len(par)} parameters, {len(src)} sources') # S03 billable units = Actual counts (Binzagr revenue codes; case pick in cases from Binzagr conversion factors) P = pd.read_parquet(os.path.join(DATA, 'sim_jeddah_pnl_monthly.parquet')) T = pd.concat([pd.read_parquet(p, columns=['tran_type', 'qty', 'units_per_case']) for p in glob.glob(os.path.join(CLEAN, 'jeddah_wms_transactions', '*.parquet'))]) vc = T.tran_type.value_counts() cases400 = float((T.loc[T.tran_type == 400, 'qty'] / T.loc[T.tran_type == 400, 'units_per_case']).sum()) exp = {'units_Inbound receipt': vc.get(153, 0), 'units_Damaged-goods receipt': vc.get(183, 0), 'units_Unload / unpick': vc.get(391, 0), 'units_Case pick': cases400, 'units_Pallet pick': vc.get(410, 0), 'units_Consolidation': vc.get(350, 0), 'units_Re-palletisation': vc.get(204, 0), 'units_Pallet split': vc.get(206, 0), 'units_Returns put-away': vc.get(172, 0)} diffs = {k: (float(P[k].sum()), float(v)) for k, v in exp.items() if abs(float(P[k].sum()) - float(v)) > 0.5} check('S03', 'Billable volumes in the simulated P&L equal Actual counts in 02_Cleaned (Binzagr revenue codes; case pick in cases)', not diffs, f'differences: {diffs}; cases picked {cases400:,.0f}') del T # S04 session logic S = pd.read_parquet(os.path.join(DATA, 'sim_session_attendance.parquet')) ok = (S.paid_hours >= S.legal_hours - 1e-9).all() and (S.recoverable_h_base >= 0).all() and (S.recoverable_h_base <= S.observed_h + 1e-9).all() \ and (S.loc[~S.eligible_for_benchmark, 'recoverable_h_base'] == 0).all() and (S.loc[S.badge_sharing_split, 'paid_shifts'] == 2).all() check('S04', 'Session rules: paid >= legal hours; 0 <= recoverable <= observed; savings only on eligible shifts; >12h sessions split', bool(ok)) # S05 cost reconciliation L = load_stats('labour') hcb = L['headcount']['labour_cost_8m'] check('S05', 'Monthly P&L labour cost reconciles to session cost + paid days without a scan (head-count basis), and its scan-day part to the sessions', abs(P.labour_cost_sar.sum() - hcb['headcount_base']) < 1 and abs(P.labour_cost_scan_days_sar.sum() - L['labour_cost_base_total']) < 1, f"P&L {P.labour_cost_sar.sum():,.0f} = sessions {P.labour_cost_scan_days_sar.sum():,.0f} + paid days without a scan {P.labour_cost_paid_days_without_scan_sar.sum():,.0f}") # S06 baseline margin by construction tot_m = (P.ebit_sar.sum() / P.handling_revenue_sar.sum()) check('S06', 'Baseline EBIT margin equals the benchmark margin parameter (cost-recovery construction)', abs(tot_m - 0.065) < 1e-6, f'{tot_m:.4%}') # S07 Monte Carlo MS = pd.read_parquet(os.path.join(DATA, 'sim_monte_carlo_summary.parquet')) SC = pd.read_parquet(os.path.join(DATA, 'sim_scenarios.parquet')).set_index('scenario') mono = ((MS.p10 <= MS.p50) & (MS.p50 <= MS.p90)).all() rec = MS.set_index('metric').loc['recoverable_labour_annual_sar'] inband = rec.p10 <= SC.loc['Base', 'recoverable_labour_annual_sar'] <= rec.p90 check('S07', 'Monte Carlo quantiles ordered; Base scenario inside the P10-P90 band', bool(mono and inband), f"recoverable P10 {rec.p10:,.0f} / Base {SC.loc['Base','recoverable_labour_annual_sar']:,.0f} / P90 {rec.p90:,.0f}") order = (SC.loc['Conservative', 'recoverable_labour_annual_sar'] < SC.loc['Base', 'recoverable_labour_annual_sar'] < SC.loc['Stretch', 'recoverable_labour_annual_sar']) check('S07b', 'Scenario ordering Conservative < Base < Stretch', bool(order)) # S08 rate card rc = pd.read_parquet(os.path.join(DATA, 'sim_rate_card.parquet')) ok = (rc.rate_low_sar > 0).all() and (rc.rate_low_sar <= rc.rate_base_sar).all() and (rc.rate_base_sar <= rc.rate_high_sar).all() check('S08', 'Rate card positive and ordered (low <= base <= high) for every billed activity', bool(ok), f'{len(rc)} activities') pstats = load_stats('pnl') lo_c, hi_c = [float(par.set_index('parameter').loc['client_case_rate_example', k]) for k in ('low', 'high')] cpu8 = pstats['cost_per_unit_8m'] cpl = cpu8['case_pick_worked_time_labour_per_loose_case_sar']; cpf = cpu8['case_pick_labour_per_loose_case_sar'] cover = (f'at SAR {lo_c:.2f} the price would not cover it (question V12)' if cpl > lo_c else f'every quoted price covers it') check('S08b', "Case-pick labour per loose case on the days people worked is not above the highest case rate Binzagr quoted (otherwise every client would be priced below labour)", cpl <= hi_c, f'worked-time labour SAR {cpl:.2f} per loose case vs client rates SAR {lo_c:.2f}-{hi_c:.2f}: {cover}; incl. paid days without a scan ' f'SAR {cpf:.2f} (facility basis, to compare with Binzagr cost per case, V11)') cpk = rc[rc.service == 'Case pick'].iloc[0] check('S08c', "Information: full cost-recovery case-pick rate vs Binzagr's example rates (differs because storage and other charges are not modelled)", bool(lo_c <= cpk.rate_base_sar <= hi_c), f'cost-recovery SAR {cpk.rate_base_sar:.2f} per case vs client SAR {lo_c:.2f}-{hi_c:.2f}', severity='WARN') rcv = rc[rc.service == 'Inbound receipt'].iloc[0] ext = rcv.rate_base_sar / rcv.external_check_sar - 1 check('S08d', 'Information: receiving rate vs US survey conversion (US handling fee only; ours carries inbound admin and the opex gross-up)', abs(ext) <= 0.5, f'receiving base SAR {rcv.rate_base_sar:.2f} vs external SAR {rcv.external_check_sar:.2f} ({ext:+.0%})', severity='WARN') # S09 receivables OI = pd.read_parquet(os.path.join(DATA, 'sim_receivables_open_items.parquet')) CA = pd.read_parquet(os.path.join(DATA, 'sim_receivables_customer_ageing.parquet')) rec_ok = abs(CA.outstanding_bal_sar.sum() - CA.allocated_to_invoices_sar.sum() - CA.unexplained_by_sfa_invoices_sar.sum()) < 0.01 * len(CA) items_ok = (OI.open_amount_sar <= OI.invoice_amount_sar + 1e-6).all() and (pd.to_datetime(OI.due_date) >= pd.to_datetime(OI.invoice_date)).all() alloc = OI.groupby(['dist_cd', 'cust_cd']).open_amount_sar.sum() a2 = CA.set_index(['dist_cd', 'cust_cd']).allocated_to_invoices_sar.reindex(alloc.index) check('S09', 'Receivables: balance = allocated + unexplained; open <= invoice; due >= invoice date; item totals = customer totals', bool(rec_ok and items_ok and np.allclose(alloc.values, a2.values)), f'{len(CA):,} customers, {len(OI):,} open items') # S10 outlets O = pd.read_parquet(os.path.join(DATA, 'inf_outlet_activity.parquet')) cust = pd.read_parquet(os.path.join(CLEAN, 'dim_customer.parquet')) check('S10', 'Outlet activity: one row per customer in the master, status always set', len(O) == len(cust) and O.activity_status.notna().all() and not O.duplicated(['dist_cd', 'cust_cd']).any(), f'{len(O):,} rows') # S11 calendar C = pd.read_parquet(os.path.join(DATA, 'ref_calendar_events.parquet')); C['date'] = pd.to_datetime(C.date) e = C.set_index('date').event ok = 'Eid al-Fitr' in str(e[pd.Timestamp('2026-03-20')]) and 'Eid al-Adha' in str(e[pd.Timestamp('2026-05-27')]) and 'Ramadan' in str(e[pd.Timestamp('2026-02-18')]) closed = C[C.wms_closed_observed].date.dt.strftime('%Y-%m-%d').tolist() ok2 = '2026-05-27' in closed and '2026-03-21' in closed check('S11', 'Official Eid dates fall inside the event windows and the warehouse data shows closures on them', bool(ok and ok2), f'observed closures: {closed}') # S12 client map CP = pd.read_parquet(os.path.join(DATA, 'inf_client_principal_map.parquet')) ok = (CP[CP.client_type_inferred == 'external_3pl_client'].share_items_matching_product_master < 0.05).all() and \ (CP[CP.client_type_inferred == 'binzagr_business_unit'].share_items_matching_product_master > 0.8).all() check('S12', 'Client typing consistent with item-code matching', bool(ok), CP.client_type_inferred.value_counts().to_dict().__str__()) # S13 risk tiers RK = pd.read_parquet(os.path.join(DATA, 'inf_badge_session_risk.parquet')) ok = RK.badge_sharing_risk.notna().all() and (RK.loc[RK.rule_gt_16h, 'badge_sharing_risk'] == 'high').all() and len(RK) == len(S) check('S13', 'Badge-sharing risk: every session tiered; all >16h sessions are high risk', bool(ok), RK.badge_sharing_risk.value_counts().to_dict().__str__()) # S15 task-based cost posting V = pd.read_parquet(os.path.join(DATA, 'sim_activity_volumes.parquet')) check('S15', 'Labour cost posted to activities reconciles to session labour cost (to the riyal)', abs(V.posted_labour_cost_sar.sum() - S.labour_cost_base_sar.sum()) < 1, f"posted {V.posted_labour_cost_sar.sum():,.0f} vs sessions {S.labour_cost_base_sar.sum():,.0f}") # S16 revenue only on Binzagr revenue codes tm = pd.read_parquet(os.path.join(CLEAN, 'ref_wms_tran_mapping.parquet')) revcodes = set(tm[(tm.cost_or_revenue == 'Revenue Generation') & tm.in_jeddah_log & tm.desc_vs_log.eq('same')].tran_type.astype(int)) billed = set(rc.tran_type.astype(int)) check('S16', "Only activities Binzagr marks 'Revenue Generation' are billed, and each of them has a rate", billed == revcodes, f'billed {sorted(billed)}; Binzagr revenue codes in the log {sorted(revcodes)} (code 100 excluded: log says Item Import, mapping says Picked Volume)') # S17 savings rules ok = (S.loc[S.cost_tier.eq('supervisory') | S.is_bulk_posting, 'recoverable_h_base'] == 0).all() and S.cost_tier.notna().all() check('S17', 'Every session has a cost tier from the WMS role; supervisory and bulk-posting sessions carry no savings', bool(ok), f"bulk-posting sessions {int(S.is_bulk_posting.sum()):,}; supervisory {int(S.cost_tier.eq('supervisory').sum()):,}") # S18 no personal names anywhere in the simulated layer from common import RAW names = pd.read_excel(os.path.join(RAW, 'WMS Mappings 22 Oct 2026', 'Users_Mapping.xlsx'), usecols=['name'], dtype=str).name.dropna().str.strip() import re as _re names = {n.upper() for n in names if len(n.split()) >= 2 and not _re.fullmatch(r'[A-Za-z.]*\d+', n)} # real names: 2+ words, not an id hits = [] for p in glob.glob(os.path.join(DATA, '*.parquet')): d = pd.read_parquet(p) if any(c.lower() in ('name', 'employee_name', 'emp_name', 'user_name') for c in d.columns): hits.append(os.path.basename(p) + ' has a name column') for c in d.select_dtypes(include=['object', 'string']).columns: if c in ('cust_name', 'cust_name2', 'example_names'): # Binzagr customer business names (kept by design) continue vals = set(d[c].dropna().astype(str).str.upper().unique()[:200000]) if vals & names: hits.append(f'{os.path.basename(p)}.{c}') check('S18', 'No employee names from the WMS user list appear in any simulated table', not hits, str(hits[:5])) # S19 split of outbound cost between case pick and pallet pick: billed lines vs linked pick time sp = pstats['outbound_split_check'] check('S19', 'Splitting outbound cost by billed lines matches the split by actual pick time (Picking scans linked to their Case Pick / Pallet Pick row) within 5 points', sp['linked_share'] > 0.95 and abs(sp['case_share_by_pick_time'] - sp['case_share_by_billed_lines']) <= 0.05, f"case pick share: {sp['case_share_by_billed_lines']:.1%} by lines vs {sp['case_share_by_pick_time']:.1%} by pick time; " f"{sp['linked_share']:.1%} of {sp['picking_scans']:,} Picking scans linked") # S20 head count: bases ordered every month; person-month scan-day cost = sessions; no negative paid days HC = pd.read_parquet(os.path.join(DATA, 'inf_jeddah_headcount.parquet')) PMt = pd.read_parquet(os.path.join(DATA, 'sim_person_month_labour.parquet')) ordered = ((HC.fte_scan_days <= HC.fte_headcount_base + 1e-9) & (HC.fte_headcount_base <= HC.fte_all_ids + 1e-9) & (HC.labour_cost_scan_days_sar <= HC.labour_cost_headcount_base_sar + 1e-6) & (HC.labour_cost_headcount_base_sar <= HC.labour_cost_all_ids_sar + 1e-6) & (HC.payroll_people_scanning <= HC.registered_active_people_all_jeddah)).all() recon = abs(PMt.labour_cost_on_scan_days_sar.sum() - S.labour_cost_base_sar.sum()) < 1 and (PMt.paid_days_without_scan >= 0).all() check('S20', "Head count: scan-day FTE <= payroll head count <= all employee ids each month; payroll people scanning <= active people in Binzagr's user list; person-month costs reconcile to the sessions", bool(ordered and recon), f"FTE {HC.fte_scan_days.mean():.0f} / {HC.fte_headcount_base.mean():.0f} / {HC.fte_all_ids.mean():.0f}; employee ids scanning " f"{HC.employee_ids_scanning.mean():.0f} a month; user-list active people {HC.registered_active_people_log_sites.mean():.0f}") # S14 files bad = [] for p in glob.glob(os.path.join(DATA, '**', '*.parquet'), recursive=True): rel = os.path.relpath(p, DATA)[:-8] cp = [x for x in (os.path.join(CSV, rel + '.csv'), os.path.join(CSV, rel + '.csv.gz')) if os.path.exists(x)] if not cp or len(pd.read_parquet(p)) != len(pd.read_csv(cp[0], dtype=str, keep_default_na=False)): bad.append(rel) for f in [p] + cp: if os.path.getsize(f) > 29 * 1024 * 1024: bad.append(f + ' too big') check('S14', 'Every parquet has a CSV twin with equal rows; no file over 29 MB', not bad, str(bad)) n = pd.Series([r['status'] for r in R]).value_counts().to_dict() json.dump({'run_at': dt.datetime.now().isoformat(timespec='seconds'), 'summary': n, 'results': R}, open(os.path.join(OUT, 'simulation_checks.json'), 'w'), indent=2, default=str) md = ['# Validation checks - 03_Simulated', '', f"Run: {dt.datetime.now():%Y-%m-%d %H:%M} | " + ' '.join(f'**{k}**: {v}' for k, v in n.items()), '', '| ID | Check | Result | Detail |', '|---|---|---|---|'] + [f"| {r['id']} | {r['test']} | {r['status']} | {str(r['detail']).replace('|','/')[:300]} |" for r in R] open(os.path.join(OUT, 'SIMULATION_CHECKS.md'), 'w').write('\n'.join(md) + '\n') print(n) if __name__ == '__main__': main()