"""03_Simulated - commercial layer (Riyadh) + calendar events. sim_receivables_open_items Simulated: each credit customer's outstanding balance allocated to its most recent invoices (FIFO payment assumption: payments clear the oldest invoices first), due dates from payment terms sim_receivables_customer_ageing Simulated: ageing buckets per customer at the snapshot date (2026-10-20) inf_outlet_activity Inferred: purchase recency, visit cadence, churn proxy per customer inf_salesman_route_profile Inferred: drops per day, active days, weekday pattern, revenue per drop per van rep ref_calendar_events Researched: Saudi 2026 events/holidays (cited) + observed activity index (Actual) as validation """ import os import numpy as np import pandas as pd from sim_common import CLEAN, SNAPSHOT_DATE, write_table, save_stats SNAP = pd.Timestamp(SNAPSHOT_DATE) def due_date(inv_date, anchor, net_days): if anchor == 'end_of_month': return (inv_date + pd.offsets.MonthEnd(0)) + pd.Timedelta(days=net_days) if anchor in ('invoice_date', 'immediate'): return inv_date + pd.Timedelta(days=net_days) return pd.NaT def main(): st = {} cust = pd.read_parquet(os.path.join(CLEAN, 'dim_customer.parquet')) terms = pd.read_parquet(os.path.join(CLEAN, 'dim_payment_terms.parquet')).set_index('invterm_cd') s = pd.read_parquet(os.path.join(CLEAN, 'riyadh_sales_lines.parquet')) r = pd.read_parquet(os.path.join(CLEAN, 'riyadh_return_lines.parquet')) s['doc_date'] = pd.to_datetime(s.doc_date); r['doc_date'] = pd.to_datetime(r.doc_date) # ---------------- 1. receivables (FIFO allocation of the outstanding balance) inv = s[~s.is_foc].groupby(['dist_cd', 'cust_cd', 'doc_no', 'doc_date']).net_amt.sum().reset_index() cred = cust[cust.has_credit_terms & (cust.outstanding_bal > 0)][['dist_cd', 'cust_cd', 'cust_name', 'invterm_cd', 'outstanding_bal', 'cust_crdlmt']] rows, cust_rows = [], [] invg = {k: g.sort_values(['doc_date', 'doc_no'], ascending=False) for k, g in inv.groupby(['dist_cd', 'cust_cd'])} for c in cred.itertuples(index=False): t = terms.loc[c.invterm_cd] if c.invterm_cd in terms.index else None anchor, nd = (t.anchor, t.net_days) if t is not None else (None, np.nan) remaining = float(c.outstanding_bal) g = invg.get((c.dist_cd, c.cust_cd)) if g is not None: for i in g.itertuples(index=False): if remaining <= 0.005: break alloc = min(remaining, float(i.net_amt)) remaining -= alloc dd = due_date(i.doc_date, anchor, nd) if pd.notna(nd) else pd.NaT dpd = (SNAP - dd).days if pd.notna(dd) else np.nan rows.append({'dist_cd': c.dist_cd, 'cust_cd': c.cust_cd, 'doc_no': i.doc_no, 'invoice_date': i.doc_date.date(), 'invoice_amount_sar': float(i.net_amt), 'open_amount_sar': alloc, 'invterm_cd': c.invterm_cd, 'due_date': dd.date() if pd.notna(dd) else None, 'days_past_due': dpd}) cust_rows.append({'dist_cd': c.dist_cd, 'cust_cd': c.cust_cd, 'cust_name': c.cust_name, 'invterm_cd': c.invterm_cd, 'outstanding_bal_sar': float(c.outstanding_bal), 'allocated_to_invoices_sar': float(c.outstanding_bal) - max(remaining, 0), 'unexplained_by_sfa_invoices_sar': max(remaining, 0), 'credit_limit_sar': c.cust_crdlmt, 'credit_limit_is_placeholder': c.cust_crdlmt == 9999999}) OI = pd.DataFrame(rows) def bucket(d): if pd.isna(d): return 'unknown_terms' if d <= 0: return 'not_due' if d <= 30: return '1-30' if d <= 60: return '31-60' if d <= 90: return '61-90' return '90+' OI['ageing_bucket'] = OI.days_past_due.map(bucket) OI['basis'] = 'Simulated (FIFO allocation of Actual balance to Actual invoices; due dates from Actual terms)' CA = pd.DataFrame(cust_rows) piv = OI.pivot_table(index=['dist_cd', 'cust_cd'], columns='ageing_bucket', values='open_amount_sar', aggfunc='sum', fill_value=0) for b in ['not_due', '1-30', '31-60', '61-90', '90+', 'unknown_terms']: if b not in piv.columns: piv[b] = 0.0 CA = CA.merge(piv.add_prefix('open_').reset_index(), on=['dist_cd', 'cust_cd'], how='left').fillna({f'open_{b}': 0 for b in piv.columns}) # sales velocity -> balance days (Inferred, deterministic) last90 = inv[inv.doc_date > SNAP - pd.Timedelta(days=90)].groupby(['dist_cd', 'cust_cd']).net_amt.sum() / 90 CA['avg_daily_credit_sales_90d_sar'] = last90.reindex(pd.MultiIndex.from_frame(CA[['dist_cd', 'cust_cd']])).to_numpy() CA['balance_days'] = CA.outstanding_bal_sar / CA.avg_daily_credit_sales_90d_sar CA['overdue_share'] = (CA[[c for c in CA.columns if c.startswith('open_') and c not in ('open_not_due', 'open_unknown_terms')]].sum(axis=1) / CA.outstanding_bal_sar.where(CA.outstanding_bal_sar > 0)) CA['basis'] = 'Simulated (ageing) / Actual (balance, limit)' st['receivables'] = {'credit_customers_with_balance': int(len(CA)), 'outstanding_total_sar': float(CA.outstanding_bal_sar.sum()), 'allocated_sar': float(CA.allocated_to_invoices_sar.sum()), 'unexplained_sar': float(CA.unexplained_by_sfa_invoices_sar.sum()), 'open_items': int(len(OI)), 'buckets_sar': OI.groupby('ageing_bucket').open_amount_sar.sum().round(0).to_dict(), 'reconciles': bool(abs(CA.outstanding_bal_sar.sum() - CA.allocated_to_invoices_sar.sum() - CA.unexplained_by_sfa_invoices_sar.sum()) < 1)} # ---------------- 2. outlet activity (all customers in master) docs = s.groupby(['dist_cd', 'cust_cd', 'doc_no']).agg(doc_date=('doc_date', 'first'), amt=('net_amt', 'sum')).reset_index() a = docs.sort_values('doc_date').groupby(['dist_cd', 'cust_cd']).agg(first_purchase=('doc_date', 'min'), last_purchase=('doc_date', 'max'), invoices=('doc_no', 'nunique'), purchase_days=('doc_date', 'nunique'), sales_sar=('amt', 'sum')).reset_index() gaps = docs.drop_duplicates(['dist_cd', 'cust_cd', 'doc_date']).sort_values('doc_date') gaps['gap'] = gaps.groupby(['dist_cd', 'cust_cd']).doc_date.diff().dt.days med_gap = gaps.groupby(['dist_cd', 'cust_cd']).gap.median() a['median_days_between_purchases'] = med_gap.reindex(pd.MultiIndex.from_frame(a[['dist_cd', 'cust_cd']])).to_numpy() O = cust[['dist_cd', 'cust_cd', 'cust_name', 'cust_type', 'cust_hier2', 'cust_hier3', 'cust_status', 'block_ind', 'latitude', 'longitude']].merge( a, on=['dist_cd', 'cust_cd'], how='left') O['days_since_last_purchase'] = (SNAP - O.last_purchase).dt.days O['churn_threshold_days'] = np.maximum(60, 3 * O.median_days_between_purchases.fillna(20)) O['activity_status'] = np.where(O.last_purchase.isna(), 'no_purchase_in_window (Mar-Oct 2026)', np.where(O.days_since_last_purchase > O.churn_threshold_days, 'lapsed (churn proxy)', np.where(O.days_since_last_purchase > 30, 'slowing', 'active'))) O['basis'] = 'Inferred (recency/cadence rules on Actual invoices)' st['outlets'] = O.activity_status.value_counts().to_dict() # ---------------- 3. van-rep route profile sm = pd.read_parquet(os.path.join(CLEAN, 'dim_salesman.parquet')) rd = s.groupby(['dist_cd', 'slsman_cd', 'doc_date']).agg(drops=('cust_cd', 'nunique'), invoices=('doc_no', 'nunique'), sales=('net_amt', 'sum'), lines=('doc_no', 'size')).reset_index() rp = rd.groupby(['dist_cd', 'slsman_cd']).agg(active_days=('doc_date', 'nunique'), first_day=('doc_date', 'min'), last_day=('doc_date', 'max'), median_drops_per_day=('drops', 'median'), p90_drops_per_day=('drops', lambda x: x.quantile(.9)), median_sales_per_day_sar=('sales', 'median'), total_sales_sar=('sales', 'sum'), total_drops=('drops', 'sum')).reset_index() rp['sales_per_drop_sar'] = rp.total_sales_sar / rp.total_drops wd = rd.assign(wd=rd.doc_date.dt.day_name()).groupby(['dist_cd', 'slsman_cd', 'wd']).size().unstack(fill_value=0) rp = rp.merge(wd.add_prefix('days_').reset_index(), on=['dist_cd', 'slsman_cd'], how='left') rp['distinct_customers'] = rp.set_index(['dist_cd', 'slsman_cd']).index.map(s.groupby(['dist_cd', 'slsman_cd']).cust_cd.nunique()) rp = rp.merge(sm[['dist_cd', 'slsman_cd', 'is_active', 'branch_cd']], on=['dist_cd', 'slsman_cd'], how='left') rp['basis'] = 'Inferred (from Actual invoices; true route plans / van loads are a gap)' st['reps'] = {'reps_with_sales': int(len(rp)), 'median_drops_per_day': float(rp.median_drops_per_day.median()), 'median_sales_per_drop_sar': float(rp.sales_per_drop_sar.median())} # ---------------- 4. calendar events (Researched) + observed activity (Actual) ev = [('2026-02-18', '2026-03-19', 'Ramadan 1447', 'reduced working hours (6h/day for Muslims, Art. 98)', 'S30;S07;S33'), ('2026-02-22', '2026-02-22', 'Founding Day', 'public holiday', 'S29'), ('2026-03-19', '2026-03-22', 'Eid al-Fitr private-sector holiday', '4 days from close of business 29 Ramadan (18 Mar); Eid day = Fri 20 Mar', 'S25;S26'), ('2026-05-26', '2026-05-29', 'Arafat Day + Eid al-Adha private-sector holiday', 'from Arafat Day 26 May, minimum 4 days; Eid day = Wed 27 May', 'S27;S28'), ('2026-09-23', '2026-09-23', 'National Day', 'public holiday', 'S29')] cal = pd.read_parquet(os.path.join(CLEAN, 'dim_calendar.parquet')) cal['date'] = pd.to_datetime(cal.date) for c in ['event', 'event_note', 'event_sources']: cal[c] = None for a0, b0, n, note, srcs in ev: m = (cal.date >= a0) & (cal.date <= b0) cal.loc[m, 'event'] = np.where(cal.loc[m, 'event'].isna(), n, cal.loc[m, 'event'] + ' + ' + n) cal.loc[m, 'event_note'] = note; cal.loc[m, 'event_sources'] = srcs cal['is_ramadan'] = cal.event.fillna('').str.contains('Ramadan') cal['is_public_holiday'] = cal.event.fillna('').str.contains('Founding|Eid|National|Arafat') # observed activity index (Actual): warehouse rows and sales lines per day vs median of same weekday import glob wmsd = pd.concat([pd.read_parquet(p, columns=['event_date']) for p in glob.glob(os.path.join(CLEAN, 'jeddah_wms_transactions', '*.parquet'))]).event_date.value_counts() wmsd.index = pd.to_datetime(wmsd.index) sd = s.groupby('doc_date').size() cal['wms_rows'] = cal.date.map(wmsd).fillna(0) cal['sales_lines'] = cal.date.map(sd).fillna(0) in_w = (cal.date >= '2026-02-01') & (cal.date <= '2026-09-30'); in_s = (cal.date >= '2026-03-01') & (cal.date <= '2026-10-20') med_w = cal[in_w & ~cal.is_public_holiday].groupby('weekday').wms_rows.median() med_s = cal[in_s & ~cal.is_public_holiday].groupby('weekday').sales_lines.median() cal['wms_activity_index'] = np.where(in_w, cal.wms_rows / cal.weekday.map(med_w), np.nan) cal['sales_activity_index'] = np.where(in_s, cal.sales_lines / cal.weekday.map(med_s), np.nan) cal['basis'] = 'Researched (events, cited) / Actual (activity index)' cal['wms_closed_observed'] = in_w & (cal.weekday != 'Friday') & (cal.wms_activity_index < 0.15) hv = {} for a0, b0, n, _, _ in ev: w = cal[(cal.date >= a0) & (cal.date <= b0 + ' 23:59') if False else (cal.date >= a0) & (cal.date <= b0)] nf = w[w.weekday != 'Friday'] near = cal[(cal.date >= pd.Timestamp(a0) - pd.Timedelta(days=1)) & (cal.date <= pd.Timestamp(b0) + pd.Timedelta(days=2))] hv[n] = {'wms_index_mean_excl_fridays': float(nf.wms_activity_index.mean()) if len(nf) else None, 'sales_index_mean_excl_fridays': float(nf.sales_activity_index.mean()) if len(nf) else None, 'observed_warehouse_closure_days': [str(d.date()) for d in near[near.wms_closed_observed].date]} st['holiday_validation'] = hv st['ramadan_vs_normal_wms_excl_fridays'] = float(cal[cal.is_ramadan & (cal.weekday != 'Friday') & in_w].wms_activity_index.median()) files = [] files += write_table(OI, 'sim_receivables_open_items') files += write_table(CA, 'sim_receivables_customer_ageing') files += write_table(O, 'inf_outlet_activity') files += write_table(rp, 'inf_salesman_route_profile') cal['date'] = cal.date.dt.date files += write_table(cal, 'ref_calendar_events') st['files'] = files save_stats('commercial_sim', st) print({k: v for k, v in st.items() if k != 'files'}) if __name__ == '__main__': main()