"""03_Simulated - Jeddah labour model (version 3, 22 Oct 2026: Binzagr's mappings applied; head count from employee ids). 1. Loaded labour cost per role tier (Researched / Assumption -> Simulated SAR per worked shift and per hour): operative = tally clerk, and logins that are not in Binzagr's user list storekeeper = storekeeper menu in the WMS user list supervisory = warehouse manager menu 2. Work content per event ("cycle time") from each badge's scan timeline (Inferred from Actual) 3. Standard time per activity AND line size (cases per line, from Binzagr's conversion factors) = percentile of badge-month mean cycle times (Inferred). Line size matters for physical moves (e.g. picking 60 s for a 1-case line, ~140 s for 11-25 cases), so a per-line standard alone would make shifts with big lines look slow. 4. Paid shifts per PERSON session (badges of the same payroll number merged, 6 h rule) under Saudi Labour Law, sessions > 12 h counted as two shifts (Simulated) 5. Shift performance vs Binzagr's own top-quartile shift in the same activity role (Inferred). Not benchmarked: supervisory staff, system logins, short / indirect sessions, and bulk-posting sessions (> 90% of scans in the same second as the previous scan: confirmations posted in batches, e.g. badge BHAS210004 - paid hours still counted) 6. Recoverable hours valued at the role's hourly cost (Simulated) 6b. Head count from employee ids (Mehul, 22 Oct): each person counts from the first to the last day they scan. Payroll- confirmed people (Binzagr's user list) are salaried, so every working day in that spell is paid - also days without a single scan. Logins the list does not know (desktop WA.AD.nnn logins, merged with their plain-number twin; shared or device logins) count only on the days they scan unless Binzagr confirms them. Three paid-time bases: scan days (paid only on days with scans; v2) <= payroll-confirmed head count (base) <= every employee id salaried 7. Task-based cost posting: each session's paid cost is spread over its events in proportion to cycle time (unscanned time pro rata) -> labour cost per activity, per case and per pallet (Simulated). This is how Binzagr posts labour cost (wage x task x duration, per user), with the time the scans cannot see spread pro rata. Outputs (03_Simulated/data): sim_labour_cost_model loaded monthly cost by role tier and scenario + per-shift / per-hour rates sim_standard_times standard seconds per activity x line-size band (3 percentile variants x 3 idle caps) + evidence sim_session_attendance one row per PERSON work session: paid shifts / hours, earned hours, performance, cost sim_wh_month_labour warehouse x month roll-up (base scenario) sim_activity_volumes wh x month x activity x client type: events, cases, pallets, hours, posted labour cost sim_person_month_labour one row per person (or unconfirmed login) x month: employed days, scan days, paid days without a scan, cost inf_jeddah_headcount per month: head count from the user list and from employee ids, FTE and labour cost on the three bases audit/labour_variants.npz per-session arrays used by the Monte Carlo """ import glob, os import numpy as np import pandas as pd from sim_common import CLEAN, write_table, save_stats, AUDIT from parameters import tables, get, pdict RAMADAN = (pd.Timestamp('2026-02-18'), pd.Timestamp('2026-03-19')) SYSTEM = {'SAPIMPORT', 'ADMIN', 'SHORTPICK'} SYSTEM_PERSONS = {f'B:{s}' for s in SYSTEM} BANDS = [(-np.inf, 1, '<=1'), (1, 4, '2-4'), (4, 10, '5-10'), (10, 25, '11-25'), (25, 60, '26-60'), (60, np.inf, '>60')] TIERS = ('operative', 'storekeeper', 'supervisory') def tier_of(role): r = str(role or '').lower() if 'manager' in r: return 'supervisory' if 'storekeeper' in r or 'inventory control' in r: return 'storekeeper' return 'operative' # tally clerk, or login not in the user list def _expat(B, pv): H = B * pv['housing_pct_of_basic']; Tr = pv['transport']; F = pv['food_allowance'] return {'basic': B, 'housing': H, 'transport': Tr, 'food': F, 'gosi_employer': pv['gosi_employer_rate_expat'] * (B + H), 'expat_levy': pv['expat_levy'], 'iqama': pv['iqama_fee'] / 12, 'medical': pv['medical_insurance'] / 12, 'eosb_accrual': pv['eosb_months_per_year'] * (B + H) / 12, 'air_ticket': pv['air_ticket_biennial'] / 24, 'recruitment_visa': pv['recruitment_visa_amortised'] / 24} def loaded_cost(pv): """pv: dict parameter -> value. Monthly loaded cost per role tier + per worked shift / per paid hour / overtime hour.""" W = pv['total_wage'] saudi = {'wage': W, 'gosi_employer': pv['gosi_employer_rate_saudi'] * W, 'medical': pv['medical_insurance'] / 12, 'eosb_accrual': pv['eosb_months_per_year'] * W / 12} saudi_month = sum(saudi.values()) mix = pv['saudi_share_of_operatives'] days_year = pv['working_days_per_month'] * 12 - pv['annual_leave_days'] - pv['public_holiday_days'] out = {'saudi': saudi, 'saudi_month': saudi_month, 'mix_saudi': mix, 'worked_days_per_year': days_year} for tier, B in (('operative', pv['basic_salary']), ('storekeeper', pv['storekeeper_basic_salary'])): e = _expat(B, pv) em = sum(e.values()) blended = (1 - mix) * em + mix * saudi_month day = blended * 12 / days_year # one worked shift (paid leave / holidays spread over worked days) basic_blend = (1 - mix) * B + mix * W * pv['basic_share_of_wage'] cash_blend = (1 - mix) * (B + e['housing'] + e['transport'] + e['food']) + mix * W out[tier] = {'expat_components': e, 'expat_month': em, 'blended_month': blended, 'cost_per_worked_shift': day, 'cost_per_paid_hour': day / pv['daily_hours'], 'overtime_cost_per_hour': cash_blend / 240 + pv['overtime_premium_on_basic'] * basic_blend / 240} # Art. 107 mgr_month = pv['manager_monthly_wage'] * (1 + pv['manager_oncost_share']) mday = mgr_month * 12 / days_year out['supervisory'] = {'blended_month': mgr_month, 'cost_per_worked_shift': mday, 'cost_per_paid_hour': mday / pv['daily_hours'], 'overtime_cost_per_hour': mday / pv['daily_hours']} # managerial posts: no overtime premium assumed # backwards-compatible keys (operative tier) for k in ('blended_month', 'cost_per_worked_shift', 'cost_per_paid_hour', 'overtime_cost_per_hour'): out[k] = out['operative'][k] return out def headcount(sess, par, costs): """Head count from employee ids. Returns person x month table, month summary, stats (see 6b in the module notes).""" s = sess[['session_id', 'person_id', 'payroll_number', 'badges', 'cost_tier', 'main_wh', 'session_start', 'month', 'paid_shifts', 'paid_hours', 'observed_h', 'labour_cost_base_sar']].copy() s['day'] = s.session_start.dt.normalize() s['confirmed'] = s.payroll_number.notna() s['head_id'] = np.where(s.confirmed, s.person_id, 'L:' + s.badges.str.replace('WA.AD.', '', regex=False)) cal = pd.DataFrame({'day': pd.date_range('2026-02-01', '2026-09-30')}) cal = cal[cal.day.dt.dayofweek != 4].copy() # Friday = weekly rest day (the scans show ~3% of activity on Fridays) cal['month'] = cal.day.dt.strftime('%Y-%m') cal['legal_h'] = np.where((cal.day >= RAMADAN[0]) & (cal.day <= RAMADAN[1]), get(par, 'daily_hours_ramadan'), get(par, 'daily_hours')) wd_month = cal.groupby('month').size() wdm = get(par, 'working_days_per_month') f_worked = costs['base']['worked_days_per_year'] / (wdm * 12) # share of working days actually worked (annual leave, holidays) sp = s.groupby('head_id').agg(first_day=('day', 'min'), last_day=('day', 'max'), confirmed=('confirmed', 'first'), cost_tier=('cost_tier', lambda x: x.mode().iloc[0]), spell_main_wh=('main_wh', lambda x: x.mode().iloc[0]), person_ids=('person_id', lambda x: ' | '.join(sorted(set(x)))), payroll_number=('payroll_number', 'first'), logins=('badges', lambda x: ' | '.join(sorted(set('|'.join(x).split('|')))))) # the extract starts on 1 Feb and ends on 30 Sep: someone seen in its first / last week is counted from 1 Feb / to 30 Sep start, end, edge = pd.Timestamp('2026-02-01'), pd.Timestamp('2026-09-30'), pd.Timedelta(days=7) sp['first_scan'], sp['last_scan'] = sp.first_day, sp.last_day sp['first_day'] = sp.first_day.where(sp.first_day > start + edge, start) sp['last_day'] = sp.last_day.where(sp.last_day < end - edge, end) rows = [] for h, r in sp.iterrows(): c = cal[(cal.day >= r.first_day) & (cal.day <= r.last_day)] g = c.groupby('month').agg(n=('day', 'size'), lh=('legal_h', 'sum')) rows += [(h, m, int(v.n), float(v.lh)) for m, v in g.iterrows()] PM = pd.DataFrame(rows, columns=['head_id', 'month', 'employed_working_days', 'employed_legal_hours']) act = s.groupby(['head_id', 'month']).agg(scan_days=('day', 'nunique'), sessions=('session_id', 'size'), paid_shifts=('paid_shifts', 'sum'), paid_hours_on_scan_days=('paid_hours', 'sum'), observed_h=('observed_h', 'sum'), labour_cost_on_scan_days_sar=('labour_cost_base_sar', 'sum'), main_wh_month=('main_wh', lambda x: x.mode().iloc[0])).reset_index() PM = PM.merge(act, on=['head_id', 'month'], how='outer') # outer: a login that only ever scanned on a Friday keeps its sessions for c in ('scan_days', 'sessions', 'paid_shifts', 'paid_hours_on_scan_days', 'observed_h', 'labour_cost_on_scan_days_sar', 'employed_working_days', 'employed_legal_hours'): PM[c] = PM[c].fillna(0) PM = PM.merge(sp, left_on='head_id', right_index=True) PM['main_wh'] = PM.main_wh_month.fillna(PM.spell_main_wh) PM['login_group'] = np.where(PM.confirmed, 'payroll-confirmed person', 'login not in the user list') PM['employed_share_of_month'] = PM.employed_working_days / PM.month.map(wd_month) PM['expected_worked_days'] = PM.employed_working_days * f_worked PM['paid_days_without_scan'] = (PM.expected_worked_days - PM.paid_shifts).clip(lower=0) PM['paid_hours_without_scan'] = np.where(PM.employed_working_days > 0, PM.paid_days_without_scan * PM.employed_legal_hours / PM.employed_working_days.where(PM.employed_working_days > 0, 1), 0.0) PM['labour_cost_without_scan_sar'] = PM.paid_days_without_scan * PM.cost_tier.map({t: costs['base'][t]['cost_per_worked_shift'] for t in TIERS}) PM['in_headcount_base'] = PM.confirmed b = PM.in_headcount_base PM['paid_hours_headcount_base'] = PM.paid_hours_on_scan_days + np.where(b, PM.paid_hours_without_scan, 0) PM['labour_cost_headcount_base_sar'] = PM.labour_cost_on_scan_days_sar + np.where(b, PM.labour_cost_without_scan_sar, 0) PM['fte_scan_days'] = PM.paid_shifts / wdm PM['fte_headcount_base'] = np.where(b, PM.employed_share_of_month, PM.fte_scan_days) PM['fte_all_ids'] = PM.employed_share_of_month PM['basis'] = ('Inferred (employment spell = first to last scan day, extended to 1 Feb / 30 Sep when seen in the first / last week; ' 'payroll from Binzagr user list) / Simulated (expected worked days, ' 'paid days without a scan, cost, base cost parameters)') PM = PM.drop(columns=['main_wh_month', 'spell_main_wh']) # month summary reg = pd.read_parquet(os.path.join(CLEAN, 'ref_wms_user_headcount_month.parquet')) rA = reg[(reg.scope == 'Jeddah sites in the WMS log') & (reg.menu_level == 'ALL')].set_index('month').active_people rJ = reg[(reg.scope == 'all Jeddah sites (2100-xx)') & (reg.menu_level == 'ALL')].set_index('month').active_people g = PM.groupby('month') scan = PM[PM.sessions > 0].groupby('month') HC = pd.DataFrame({ 'registered_active_people_log_sites': rA, 'registered_active_people_all_jeddah': rJ, 'employee_ids_scanning': scan.size(), 'payroll_people_scanning': scan.confirmed.sum(), 'other_logins_scanning': scan.size() - scan.confirmed.sum(), 'payroll_people_employed': g.confirmed.sum(), 'fte_scan_days': g.fte_scan_days.sum(), 'fte_headcount_base': g.fte_headcount_base.sum(), 'fte_all_ids': g.fte_all_ids.sum(), 'paid_days_on_scan_days': g.paid_shifts.sum(), 'paid_days_without_scan_base': PM[b].groupby('month').paid_days_without_scan.sum(), 'paid_days_without_scan_all_ids': g.paid_days_without_scan.sum(), 'paid_hours_scan_days': g.paid_hours_on_scan_days.sum(), 'paid_hours_headcount_base': g.paid_hours_headcount_base.sum(), 'paid_hours_all_ids': g.paid_hours_on_scan_days.sum() + g.paid_hours_without_scan.sum(), 'observed_hours': g.observed_h.sum(), 'labour_cost_scan_days_sar': g.labour_cost_on_scan_days_sar.sum(), 'labour_cost_headcount_base_sar': g.labour_cost_headcount_base_sar.sum(), 'labour_cost_all_ids_sar': g.labour_cost_on_scan_days_sar.sum() + g.labour_cost_without_scan_sar.sum()}).reindex(sorted(PM.month.unique())) HC.index.name = 'month' for k in ('scan_days', 'headcount_base', 'all_ids'): HC[f'unobserved_share_{k}'] = 1 - HC.observed_hours / HC[f'paid_hours_{k}'] HC = HC.reset_index() HC['basis'] = ('Actual (registered people: Binzagr user list; employee ids scanning: WMS log) / Inferred (employment spells) / ' 'Simulated (FTE, paid hours and labour cost on each basis; base cost parameters)') tot = HC.drop(columns=['month', 'basis']).sum() conf = PM[PM.confirmed] hst = {'headcount': { 'people_and_logins_total': int(len(sp)), 'payroll_people_total': int(sp.confirmed.sum()), 'other_logins_total': int((~sp.confirmed).sum()), 'desktop_logins_merged_with_twin': int(len(set(s[~s.confirmed & s.badges.str.startswith('WA.AD.')].head_id) & set(s[~s.confirmed & ~s.badges.str.startswith('WA.AD.')].head_id))), 'avg_per_month': {c: float(HC[c].mean()) for c in ('registered_active_people_log_sites', 'registered_active_people_all_jeddah', 'employee_ids_scanning', 'payroll_people_scanning', 'other_logins_scanning', 'payroll_people_employed', 'fte_scan_days', 'fte_headcount_base', 'fte_all_ids')}, 'labour_cost_8m': {k: float(tot[f'labour_cost_{k}_sar']) for k in ('scan_days', 'headcount_base', 'all_ids')}, 'paid_hours_8m': {k: float(tot[f'paid_hours_{k}']) for k in ('scan_days', 'headcount_base', 'all_ids')}, 'unobserved_share': {k: float(1 - tot.observed_hours / tot[f'paid_hours_{k}']) for k in ('scan_days', 'headcount_base', 'all_ids')}, 'paid_days_without_scan_8m': {'headcount_base': float(tot.paid_days_without_scan_base), 'all_ids': float(tot.paid_days_without_scan_all_ids)}, 'payroll_people_scan_share_of_expected_days': float(conf.paid_shifts.sum() / conf.expected_worked_days.sum()), 'worked_share_of_working_days': float(f_worked)}} return PM, HC, hst def cycle_seconds(T, cap_s): """Work content per event: time since the badge's previous event ended (+ own duration). Idle gaps longer than cap are treated as breaks (not work). Durations > 1h (tasks left open) capped at 1h.""" dur = T.dur_s.fillna(0).clip(upper=3600).to_numpy() idle = T.idle_secs_before.to_numpy() start = T.start_ts.to_numpy(); end = T.end_ts.fillna(T.start_ts).to_numpy() prev_end = start - pd.to_timedelta(np.nan_to_num(idle), unit='s').to_numpy() overlap_inc = np.maximum(0, (end - prev_end) / np.timedelta64(1, 's')) cyc = np.where(np.isnan(idle), dur, np.where(idle < 0, overlap_inc, np.where(idle <= cap_s, idle + dur, dur))) return np.clip(cyc, 0, 3600) def size_band(cases): c = cases.to_numpy(dtype='float64') out = np.full(len(c), 'unknown', dtype=object) for lo, hi, lab in BANDS: out[(c > lo) & (c <= hi)] = lab out[~np.isfinite(c) | (c <= 0)] = 'unknown' return out def main(): src, par = tables() st = {} # ---------------- 1. loaded cost per tier rows = [] costs = {s: loaded_cost(pdict(par, s)) for s in ('low', 'base', 'high')} for s, c in costs.items(): for tier in ('operative', 'storekeeper'): for k, v in c[tier]['expat_components'].items(): rows.append({'scenario': s, 'role': f'{tier} (expat)', 'component': k, 'sar_per_month': round(v, 2), 'basis': 'Simulated'}) rows.append({'scenario': s, 'role': f'{tier} (expat)', 'component': 'TOTAL', 'sar_per_month': round(c[tier]['expat_month'], 2), 'basis': 'Simulated'}) for k, v in c['saudi'].items(): rows.append({'scenario': s, 'role': 'saudi operative', 'component': k, 'sar_per_month': round(v, 2), 'basis': 'Simulated'}) rows.append({'scenario': s, 'role': 'saudi operative', 'component': 'TOTAL', 'sar_per_month': round(c['saudi_month'], 2), 'basis': 'Simulated'}) for tier in TIERS: for k in ('blended_month', 'cost_per_worked_shift', 'cost_per_paid_hour', 'overtime_cost_per_hour'): rows.append({'scenario': s, 'role': f'{tier} (blended)', 'component': k, 'sar_per_month': round(c[tier][k], 4), 'basis': 'Simulated'}) rows.append({'scenario': s, 'role': 'all', 'component': 'mix_saudi', 'sar_per_month': c['mix_saudi'], 'basis': 'Simulated'}) rows.append({'scenario': s, 'role': 'all', 'component': 'worked_days_per_year', 'sar_per_month': c['worked_days_per_year'], 'basis': 'Simulated'}) lc = pd.DataFrame(rows) st['loaded_cost'] = {s: {t: {k: round(v, 2) for k, v in c[t].items() if not isinstance(v, dict)} for t in TIERS} for s, c in costs.items()} # ---------------- 2. events + cycle time cols = ['tran_type', 'tran_desc', 'tran_category', 'wh_id', 'client_code', 'start_ts', 'end_ts', 'dur_s', 'employee_id_norm', 'badge_type', 'idle_secs_before', 'session_id', 'person_session_id', 'month', 'is_billing_event', 'qty_cases', 'qty_pallets', 'movement_type', 'cost_or_revenue', 'case_or_pallet'] T = pd.concat([pd.read_parquet(p, columns=cols) for p in sorted(glob.glob(os.path.join(CLEAN, 'jeddah_wms_transactions', '*.parquet')))], ignore_index=True) for c in ('movement_type', 'cost_or_revenue', 'case_or_pallet'): T[c] = T[c].astype(str) T = T[~T.employee_id_norm.isin(SYSTEM)] T = T.sort_values(['employee_id_norm', 'start_ts', 'end_ts'], kind='mergesort').reset_index(drop=True) caps = {15: 900, 30: 1800, 45: 2700} for cm, cs in caps.items(): T[f'cyc{cm}'] = cycle_seconds(T, cs).astype('float32') T['size_band'] = size_band(T.qty_cases) st['events_by_size_band'] = T.size_band.value_counts().to_dict() # ---------------- 3. standard times per activity x line-size band pcts = {'lenient_p60': 0.60, 'median_p50': 0.50, 'tight_p40': 0.40} std_rows = [] for cm in caps: c = f'cyc{cm}' bm_d = T.groupby(['tran_desc', 'employee_id_norm', 'month'])[c].agg(['mean', 'size']).reset_index() bm_d = bm_d[bm_d['size'] >= 30] pooled_d = T.groupby('tran_desc')[c].mean() n_d = T.groupby('tran_desc').size() desc_std = {} for d in n_d.index: b = bm_d[bm_d.tran_desc == d] desc_std[d] = {nm: (float(b['mean'].quantile(q)) if len(b) >= 5 else float(pooled_d[d])) for nm, q in pcts.items()} desc_std[d]['_method'] = 'percentile_of_badge_month_means' if len(b) >= 5 else 'pooled_mean (too few badge-months)' bm_b = T.groupby(['tran_desc', 'size_band', 'employee_id_norm', 'month'])[c].agg(['mean', 'size']).reset_index() bm_b = bm_b[bm_b['size'] >= 30] pooled_b = T.groupby(['tran_desc', 'size_band'])[c].agg(['mean', 'size']) for (d, band), pb in pooled_b.iterrows(): b = bm_b[(bm_b.tran_desc == d) & (bm_b.size_band == band)] r = {'tran_desc': d, 'size_band': band, 'cap_minutes': cm, 'events': int(pb['size']), 'badge_months_used': int(len(b)), 'pooled_mean_s': float(pb['mean'])} if band != 'unknown' and len(b) >= 5: for nm, q in pcts.items(): r[f'std_s_{nm}'] = float(b['mean'].quantile(q)) r['method'] = 'percentile_of_badge_month_means (activity x size band)' elif band != 'unknown' and pb['size'] >= 30: # pooled band mean, scaled to each percentile variant by the activity-level ratio to its median base = desc_std[d]['median_p50'] for nm in pcts: ratio = desc_std[d][nm] / base if base > 0 else 1.0 r[f'std_s_{nm}'] = float(pb['mean']) * ratio r['method'] = 'pooled_band_mean (few badge-months) x activity percentile ratio' else: for nm in pcts: r[f'std_s_{nm}'] = desc_std[d][nm] r['method'] = 'activity-level standard (' + desc_std[d]['_method'] + ')' std_rows.append(r) stdt = pd.DataFrame(std_rows) stdt['basis'] = 'Inferred' st['standards_rows'] = int(len(stdt)) st['standards_method_counts_cap30'] = stdt[stdt.cap_minutes == 30].method.value_counts().to_dict() # ---------------- 4. person sessions: earned hours per variant sess = pd.read_parquet(os.path.join(CLEAN, 'jeddah_person_sessions.parquet')) sess = sess[~sess.person_id.isin(SYSTEM_PERSONS) & ~sess.any_system_badge].reset_index(drop=True) sess = sess.rename(columns={'person_session_id': 'session_id'}) T = T[T.person_session_id.isin(sess.session_id)].reset_index(drop=True) variants = {} key = T.tran_desc + '|' + T.size_band for cm in caps: s_c = stdt[stdt.cap_minutes == cm].assign(k=lambda d: d.tran_desc + '|' + d.size_band).set_index('k') for nm in pcts: earned = key.map(s_c[f'std_s_{nm}']).astype('float64').groupby(T.person_session_id).sum() / 3600 variants[(cm, nm)] = earned.reindex(sess.session_id).fillna(0).to_numpy() if cm == 30: T[f'std_s_{nm}'] = key.map(s_c[f'std_s_{nm}']).astype('float32') del key sess['scan_work_h'] = (T.groupby('person_session_id').cyc30.sum() / 3600).reindex(sess.session_id).fillna(0).to_numpy() # ---------------- 5. paid shifts (simulated) under labour law, per person session in_ram = (sess.session_start >= RAMADAN[0]) & (sess.session_start < RAMADAN[1] + pd.Timedelta(days=1)) legal = np.where(in_ram, get(par, 'daily_hours_ramadan'), get(par, 'daily_hours')) brk = get(par, 'break_hours'); maxh = get(par, 'max_hours_at_workplace') split = sess.elapsed_h > get(par, 'badge_sharing_split_hours') sess['ramadan'] = in_ram sess['legal_hours'] = legal sess['badge_sharing_split'] = split sess['paid_shifts'] = np.where(split, 2, 1) sess['overtime_h'] = np.where(split, 0, np.clip(sess.elapsed_h - (legal + brk), 0, maxh - legal - brk)) sess['paid_hours'] = sess.paid_shifts * legal + sess.overtime_h sess['wms_role'] = sess.wms_role.fillna('Not in user list') sess['cost_tier'] = sess.wms_role.map(tier_of) cnt = T.groupby(['person_session_id', 'tran_category']).size().unstack(fill_value=0) sess['role'] = cnt.idxmax(axis=1).reindex(sess.session_id).to_numpy() sess['role_share'] = (cnt.max(axis=1) / cnt.sum(axis=1)).reindex(sess.session_id).to_numpy() del cnt sess['is_direct'] = (sess.n_events >= get(par, 'direct_shift_min_events')) for (cm, nm), arr in variants.items(): if cm == 30: sess[f'earned_h_{nm}'] = arr pfd = get(par, 'pfd_allowance') sess['observed_h'] = np.minimum(sess.elapsed_h, sess.paid_hours).clip(upper=maxh) sess['unobserved_paid_h'] = (sess.paid_hours - sess.observed_h).clip(lower=0) sess['performance_vs_paid'] = sess.earned_h_median_p50 / (sess.paid_hours * (1 - pfd)) sess['span_productivity'] = sess.earned_h_median_p50 / sess.observed_h.clip(lower=0.25) MIN_OBS = 4.0 bulk_thr = get(par, 'bulk_posting_same_second_share') sess['is_bulk_posting'] = sess.same_second_share > bulk_thr sess['eligible_for_benchmark'] = sess.is_direct & (sess.elapsed_h >= MIN_OBS) & ~sess.is_bulk_posting & sess.cost_tier.ne('supervisory') E = sess[sess.eligible_for_benchmark] q = get(par, 'target_performance_percentile') targets = E.groupby('role').span_productivity.quantile(q) n_role = E.groupby('role').size() targets = targets[n_role >= 30] # roles with too few full shifts get no target (no savings claimed) sess['target_span_productivity'] = sess.role.map(targets) need = sess.earned_h_median_p50 / sess.target_span_productivity gap = np.maximum(0, sess.observed_h - need) ok = sess.eligible_for_benchmark & sess.target_span_productivity.notna() sess['recoverable_h_base'] = np.where(ok, gap * get(par, 'savings_capture_rate'), 0.0) c = costs['base'] shift_cost = sess.cost_tier.map({t: c[t]['cost_per_worked_shift'] for t in TIERS}) ot_cost = sess.cost_tier.map({t: c[t]['overtime_cost_per_hour'] for t in TIERS}) sess['labour_cost_base_sar'] = sess.paid_shifts * shift_cost + sess.overtime_h * ot_cost sess['recoverable_cost_base_sar'] = sess.recoverable_h_base * (shift_cost / sess.legal_hours) sess['month'] = sess.session_start.dt.strftime('%Y-%m') sess['basis'] = 'Simulated (paid hours, cost, recoverable) / Inferred (earned, productivity, role) / Actual (person, WMS role)' # ---------------- 5b. head count from employee ids: salaried people are paid on every working day, not only on days they scan PM, HC, hst = headcount(sess, par, costs) st.update(hst) st['benchmark_min_observed_hours'] = MIN_OBS st['bulk_posting_threshold'] = bulk_thr st['role_targets_base'] = {k: round(float(v), 4) for k, v in targets.items()} st['role_eligible_sessions'] = {k: int(v) for k, v in n_role.items()} st['sessions'] = int(len(sess)); st['direct_sessions'] = int(sess.is_direct.sum()); st['eligible_sessions'] = int(ok.sum()) st['bulk_posting_sessions'] = int(sess.is_bulk_posting.sum()) st['bulk_posting_sessions_by_badge_top'] = sess[sess.is_bulk_posting].badges.value_counts().head(8).to_dict() st['bulk_posting_paid_hours'] = float(sess.loc[sess.is_bulk_posting, 'paid_hours'].sum()) st['sessions_by_cost_tier'] = sess.cost_tier.value_counts().to_dict() st['paid_hours_by_cost_tier'] = {k: float(v) for k, v in sess.groupby('cost_tier').paid_hours.sum().items()} st['paid_hours_by_wms_role'] = {k: float(v) for k, v in sess.groupby('wms_role').paid_hours.sum().items()} st['badge_sharing_splits'] = int(split.sum()) st['multi_badge_sessions'] = int((sess.n_badges > 1).sum()) st['concurrent_badge_hours'] = float(sess.concurrent_badge_h.sum()) st['paid_hours_total'] = float(sess.paid_hours.sum()); st['overtime_hours_total'] = float(sess.overtime_h.sum()) st['observed_hours_total'] = float(sess.observed_h.sum()); st['unobserved_paid_hours_total'] = float(sess.unobserved_paid_h.sum()) st['earned_hours_base'] = float(sess.earned_h_median_p50.sum()); st['scan_work_hours'] = float(sess.scan_work_h.sum()) st['recoverable_hours_base'] = float(sess.recoverable_h_base.sum()) st['recoverable_share_of_paid_base'] = float(sess.recoverable_h_base.sum() / sess.paid_hours.sum()) st['labour_cost_base_total'] = float(sess.labour_cost_base_sar.sum()); st['recoverable_cost_base_total'] = float(sess.recoverable_cost_base_sar.sum()) st['fte_equivalent_avg'] = float(sess.paid_shifts.sum() / 8 / get(par, 'working_days_per_month')) st['span_productivity_quantiles_eligible'] = {str(qq): float(E.span_productivity.quantile(qq)) for qq in (0.1, 0.25, 0.5, 0.75, 0.9)} # ---------------- 6. warehouse x month roll-up (base) wm = sess.groupby(['main_wh', 'month']).agg(sessions=('session_id', 'size'), direct_sessions=('is_direct', 'sum'), paid_shifts=('paid_shifts', 'sum'), paid_hours=('paid_hours', 'sum'), overtime_h=('overtime_h', 'sum'), earned_hours=('earned_h_median_p50', 'sum'), scan_work_h=('scan_work_h', 'sum'), recoverable_hours=('recoverable_h_base', 'sum'), labour_cost_sar=('labour_cost_base_sar', 'sum'), recoverable_cost_sar=('recoverable_cost_base_sar', 'sum'), people=('person_id', 'nunique'), badge_sharing_splits=('badge_sharing_split', 'sum'), bulk_posting_sessions=('is_bulk_posting', 'sum')).reset_index().rename(columns={'main_wh': 'wh_id'}) wm['median_span_productivity_eligible'] = sess[sess.eligible_for_benchmark].groupby(['main_wh', 'month']).span_productivity.median().reindex( pd.MultiIndex.from_frame(wm[['wh_id', 'month']])).to_numpy() wm['unobserved_paid_h'] = sess.groupby(['main_wh', 'month']).unobserved_paid_h.sum().reindex(pd.MultiIndex.from_frame(wm[['wh_id', 'month']])).to_numpy() wm['fte_equivalent'] = wm.paid_shifts / get(par, 'working_days_per_month') pw = PM.groupby(['main_wh', 'month']).agg(employee_ids=('head_id', 'size'), fte_headcount_base=('fte_headcount_base', 'sum'), paid_days_without_scan_base=('paid_days_without_scan', lambda x: float(x[PM.loc[x.index, 'in_headcount_base']].sum())), paid_hours_headcount_base=('paid_hours_headcount_base', 'sum'), observed_hours_people=('observed_h', 'sum'), labour_cost_headcount_base_sar=('labour_cost_headcount_base_sar', 'sum')).reset_index().rename(columns={'main_wh': 'wh_id'}) wm = wm.merge(pw, on=['wh_id', 'month'], how='outer') for c in ('sessions', 'paid_shifts', 'paid_hours', 'labour_cost_sar', 'unobserved_paid_h'): wm[c] = wm[c].fillna(0) wm['observed_hours'] = wm.paid_hours - wm.unobserved_paid_h wm['unobserved_paid_h_headcount_base'] = (wm.paid_hours_headcount_base - wm.observed_hours_people).clip(lower=0) # both by the person's main site wm['basis'] = 'Simulated (scan-day columns as v2; *_headcount_base columns: payroll-confirmed people paid on every working day of their spell)' # ---------------- 7. task-based cost posting: session paid cost / hours spread over its events by cycle time w = T.cyc30.astype('float64') tot = w.groupby(T.person_session_id).transform('sum') n_ev = T.groupby('person_session_id').cyc30.transform('size') share = np.where(tot > 0, w / tot.where(tot > 0, 1), 1.0 / n_ev) sidx = sess.set_index('session_id') T['posted_cost_sar'] = share * T.person_session_id.map(sidx.labour_cost_base_sar).to_numpy() T['posted_paid_h'] = share * T.person_session_id.map(sidx.paid_hours).to_numpy() T['client_type'] = np.where(T.client_code.str.startswith('3', na=False), 'external_3pl', np.where(T.client_code.notna(), 'binzagr_business_unit', 'unknown')) vol = T.groupby(['wh_id', 'month', 'tran_type', 'tran_desc', 'movement_type', 'cost_or_revenue', 'case_or_pallet', 'client_type'], observed=True).agg( events=('cyc30', 'size'), cases=('qty_cases', 'sum'), pallets=('qty_pallets', 'sum'), events_with_case_factor=('qty_cases', 'count'), cycle_hours=('cyc30', lambda x: float(x.sum()) / 3600), std_hours=('std_s_median_p50', lambda x: float(x.sum()) / 3600), posted_paid_hours=('posted_paid_h', 'sum'), posted_labour_cost_sar=('posted_cost_sar', 'sum')).reset_index() vol['labour_cost_per_event_sar'] = vol.posted_labour_cost_sar / vol.events vol['labour_cost_per_case_sar'] = np.where(vol.cases > 0, vol.posted_labour_cost_sar / vol.cases.where(vol.cases > 0, 1), np.nan) vol['basis'] = 'Actual (events, cases, pallets) / Inferred (cycle, standard hours) / Simulated (posted paid hours and cost, base)' st['posted_cost_reconciles'] = float(vol.posted_labour_cost_sar.sum() - sess.labour_cost_base_sar.sum()) mv = vol.groupby('movement_type').posted_labour_cost_sar.sum() st['posted_cost_by_movement_type'] = {k: float(v) for k, v in mv.items()} st['posted_cost_share_by_movement_type'] = {k: float(v / mv.sum()) for k, v in mv.items()} cr = vol.groupby('cost_or_revenue').posted_labour_cost_sar.sum() st['posted_cost_share_by_cost_or_revenue'] = {k: float(v / cr.sum()) for k, v in cr.items()} top = vol.groupby('tran_desc').posted_labour_cost_sar.sum().sort_values(ascending=False) st['posted_cost_top_activities'] = {k: float(v) for k, v in top.head(12).items()} # ---------------- 8. MC arrays np.savez_compressed(os.path.join(AUDIT, 'labour_variants.npz'), **{f'earned_{cm}_{nm}': v for (cm, nm), v in variants.items()}, paid_shifts=sess.paid_shifts.to_numpy(), overtime_h=sess.overtime_h.to_numpy(), legal_hours=sess.legal_hours.to_numpy(), is_direct=sess.is_direct.to_numpy(), observed_h=sess.observed_h.to_numpy(), eligible=sess.eligible_for_benchmark.to_numpy(), role=sess.role.fillna('none').to_numpy().astype(str), tier=sess.cost_tier.to_numpy().astype(str), month=sess.month.to_numpy().astype(str), wh=sess.main_wh.to_numpy().astype(str), pm_noscan_days=PM.paid_days_without_scan.to_numpy(), pm_tier=PM.cost_tier.to_numpy().astype(str), pm_confirmed=PM.confirmed.to_numpy(), pm_month=PM.month.to_numpy().astype(str)) out_cols = ['session_id', 'person_id', 'payroll_number', 'badges', 'n_badges', 'wms_role', 'cost_tier', 'main_wh', 'month', 'session_start', 'session_end', 'elapsed_h', 'n_events', 'n_warehouses', 'n_fast_wh_switch', 'concurrent_badge_h', 'same_second_share', 'is_bulk_posting', 'ramadan', 'legal_hours', 'badge_sharing_split', 'paid_shifts', 'overtime_h', 'paid_hours', 'is_direct', 'role', 'role_share', 'observed_h', 'unobserved_paid_h', 'scan_work_h', 'earned_h_lenient_p60', 'earned_h_median_p50', 'earned_h_tight_p40', 'performance_vs_paid', 'span_productivity', 'eligible_for_benchmark', 'target_span_productivity', 'recoverable_h_base', 'labour_cost_base_sar', 'recoverable_cost_base_sar', 'basis'] files = [] files += write_table(lc, 'sim_labour_cost_model') files += write_table(stdt, 'sim_standard_times') files += write_table(sess[out_cols], 'sim_session_attendance') files += write_table(wm, 'sim_wh_month_labour') files += write_table(vol, 'sim_activity_volumes') pm_cols = ['head_id', 'login_group', 'person_ids', 'payroll_number', 'logins', 'cost_tier', 'main_wh', 'month', 'first_scan', 'last_scan', 'first_day', 'last_day', 'employed_working_days', 'employed_share_of_month', 'expected_worked_days', 'scan_days', 'sessions', 'paid_shifts', 'paid_days_without_scan', 'paid_hours_on_scan_days', 'paid_hours_without_scan', 'observed_h', 'labour_cost_on_scan_days_sar', 'labour_cost_without_scan_sar', 'in_headcount_base', 'paid_hours_headcount_base', 'labour_cost_headcount_base_sar', 'fte_scan_days', 'fte_headcount_base', 'fte_all_ids', 'basis'] files += write_table(PM[pm_cols], 'sim_person_month_labour') files += write_table(HC, 'inf_jeddah_headcount') st['files'] = files save_stats('labour', st) print({k: v for k, v in st.items() if k != 'files'}) if __name__ == '__main__': main()