"""Write every dashboard CSV except the forecast files (p02), plus value levers, reconciliation and the KPI tile sheet."""
import json, pickle, numpy as np, pandas as pd
from pcommon import *

S = pd.read_parquet(f'{WORK}/sales.parquet'); R = pd.read_parquet(f'{WORK}/returns.parquet'); P = pd.read_parquet(f'{WORK}/products.parquet').set_index('sku')
O = pd.read_parquet(f'{WORK}/outlets.parquet'); OK = pickle.load(open(f'{WORK}/outlets_kpi.pkl', 'rb'))['kpi']
CH = pickle.load(open(f'{WORK}/churn.pkl', 'rb')); VD = pickle.load(open(f'{WORK}/van_day.pkl', 'rb')); VL = pickle.load(open(f'{WORK}/van_load.pkl', 'rb'))
RT = pickle.load(open(f'{WORK}/reco_tuned.pkl', 'rb')); RS = pickle.load(open(f'{WORK}/reco_select.pkl', 'rb')); RECO = pd.read_parquet(f'{WORK}/reco_final.parquet')
RC = pickle.load(open(f'{WORK}/receivables.pkl', 'rb')); WH = pickle.load(open(f'{WORK}/warehouse.pkl', 'rb')); FC = pickle.load(open(f'{WORK}/forecast_results.pkl', 'rb'))
LAB = json.load(open(f'{ROOT}/03_Simulated/audit/stats_labour.json')); PNL = json.load(open(f'{ROOT}/03_Simulated/audit/stats_pnl.json'))
C = pd.read_parquet(f'{CLEAN}/dim_customer.parquet'); CS = json.load(open(f'{ROOT}/02_Cleaned/audit/stats_commercial.json')); JS = json.load(open(f'{ROOT}/02_Cleaned/audit/stats_jeddah.json'))
tiles = []
def tile(screen, name, tile_, value, display, unit, label, note=''):
    if isinstance(value, float): value = round(value, 6)
    tiles.append({'screen': screen, 'screen_name': name, 'tile': tile_, 'value': value, 'display': display, 'unit': unit, 'label': label, 'note': note})

# ================================================================== 01 reconciliation
B = S[S.kind == 'sale']
rec = pd.DataFrame([
    ('Net sales (SAR)', CS['sales']['net_amt_sum'], round(B.net_amt.sum(), 2), 'Actual'),
    ('Returns (SAR)', CS['returns']['net_amt_sum'], round(R.net_amt.sum(), 2), 'Actual'),
    ('Sales invoice lines', CS['sales']['raw_rows'], len(B), 'Actual'),
    ('Invoices', CS['sales']['docs'], B.doc_no.nunique(), 'Actual'),
    ('Buying outlets', B.outlet_id.nunique(), O.shape[0], 'Actual'),
    ('Vans (sales reps)', CS['salesman_raw_rows'], B.van.nunique(), 'Actual'),
    ('Customers in master', CS['customer_raw_rows'], CS['customer_rows'], 'Corrected'),
    ('WMS scan rows', JS['raw_rows_total'], JS['clean_rows'], 'Corrected')], columns=['measure', 'binzagr_file', 'loaded', 'label'])
rec['difference'] = (rec.loaded - rec.binzagr_file).round(2)
rec['status'] = np.where(rec.difference.abs() < 0.01, 'Match', '')
rec.loc[rec.measure == 'Customers in master', 'status'] = 'Explained: +174 customers found only in the Lipton customer file'
rec.loc[rec.measure == 'WMS scan rows', 'status'] = 'Explained: 65,141 rows were exact copies (each month file repeated day 1 of the next)'
rec['source'] = 'Binzagr file = recomputed from the raw export (smoke tests C01, C02, J03, J04); loaded = 02_Cleaned'
write_csv(rec, '01_reconciliation', 'reconciliation_checks.csv')
tile(1, 'Reconciliation', 'Van sales', round(B.net_amt.sum(), 2), f"SAR {B.net_amt.sum():,.0f}", 'SAR', 'Actual', 'identical to the raw file')
tile(1, 'Reconciliation', 'Returns', round(R.net_amt.sum(), 2), f"SAR {R.net_amt.sum():,.0f}", 'SAR', 'Actual')
tile(1, 'Reconciliation', 'Invoice lines', len(B), f"{len(B):,}", 'lines', 'Actual')
tile(1, 'Reconciliation', 'WMS scans', JS['clean_rows'], f"{JS['clean_rows']:,}", 'rows', 'Corrected', f"{JS['removed_boundary_rows']:,} duplicates removed")

# ================================================================== 03 forecast tiles (files written by p02)
fm = {lv: FC[lv]['m'] for lv in FC}
tile(3, 'Forecast proof', 'SKU weekly error', fm['sku']['holdout_wape_model'], f"{fm['sku']['holdout_wape_model']:.1%}", 'WAPE', 'Actual', f"vs {fm['sku']['holdout_wape_naive']:.1%} repeat last week; hold-out 23 Sep - 20 Oct 2026")
tile(3, 'Forecast proof', 'Less error (forecast value added)', fm['sku']['forecast_value_added'], f"{fm['sku']['forecast_value_added']:.0%}", 'share', 'Actual', 'SKU level, 4 unseen weeks')
tile(3, 'Forecast proof', 'Van weekly error', fm['van']['holdout_wape_model'], f"{fm['van']['holdout_wape_model']:.1%}", 'WAPE', 'Actual', f"vs {fm['van']['holdout_wape_naive']:.1%} repeat last week")
tile(3, 'Forecast proof', 'Principal weekly error', fm['principal']['holdout_wape_model'], f"{fm['principal']['holdout_wape_model']:.1%}", 'WAPE', 'Actual', f"vs {fm['principal']['holdout_wape_naive']:.1%} repeat last week")
tile(3, 'Forecast proof', 'Safety stock', fm['sku']['safety_stock_change'], f"{fm['sku']['safety_stock_change']:+.0%}", 'share', 'Actual', 'same 95% service level, one-week cover, SKU level')
tile(3, 'Forecast proof', 'Prediction band coverage', fm['sku']['band_80_coverage_holdout'], f"{fm['sku']['band_80_coverage_holdout']:.0%}", 'share', 'Actual', 'share of hold-out weeks inside the p10-p90 band (target 80%)')

# ================================================================== 04 warehouse labour
ca = WH['cost_act']
write_csv(ca, '04_warehouse_labour', 'warehouse_cost_per_activity.csv')
risk = pd.read_parquet(f'{SIM}/inf_badge_session_risk.parquet'); att = pd.read_parquet(f'{SIM}/sim_session_attendance.parquet')
fl = risk[risk.badge_sharing_risk.isin(['high', 'medium']) | risk.rule_gt_12h | risk.rule_concurrent_badges].merge(
    att[['session_id', 'payroll_number', 'session_end', 'paid_shifts', 'paid_hours', 'badge_sharing_split', 'legal_hours']], on='session_id', how='left')
def why(r):
    w = []
    if r.rule_gt_16h: w.append(f'span {r.elapsed_h:.1f} h (over 16 h)')
    elif r.rule_gt_12h: w.append(f'span {r.elapsed_h:.1f} h (over the 12 h legal limit)')
    if r.rule_concurrent_badges: w.append(f'two badges of one person active together ({r.concurrent_badge_h:.1f} h)')
    if r.rule_fast_wh_switch: w.append('site switch within 2 minutes')
    if r.rule_extreme_rate: w.append('extreme scan rate')
    if r.anomaly_flag_ml: w.append('unusual pattern (model)')
    return '; '.join(w)
fl['why_flagged'] = fl.apply(why, axis=1)
fl['recorded_hours_on_badge'] = fl.elapsed_h.round(2)
# split sessions: person 1 gets half (max 12 h), person 2 gets the rest (max 12 h), so the two never add up to more than the badge recorded
fl['corrected_hours_person_1'] = np.where(fl.badge_sharing_split, (fl.recorded_hours_on_badge / 2).clip(upper=12), fl.recorded_hours_on_badge.clip(upper=12)).round(2)
fl['corrected_hours_person_2'] = np.where(fl.badge_sharing_split, (fl.recorded_hours_on_badge - fl.corrected_hours_person_1).clip(upper=12), 0).round(2)
bs = fl[['session_id', 'person_id', 'payroll_number', 'badges', 'n_badges', 'wms_role', 'main_wh', 'session_start', 'session_end', 'recorded_hours_on_badge',
         'corrected_hours_person_1', 'corrected_hours_person_2', 'paid_shifts', 'concurrent_badge_h', 'badge_sharing_risk', 'why_flagged', 'n_events']].copy()
bs['concurrent_badge_h'] = bs.concurrent_badge_h.round(2)
bs['date'] = bs.session_start.dt.normalize(); bs = bs.sort_values(['badge_sharing_risk', 'recorded_hours_on_badge'], ascending=[True, False])
bs['session_start'] = bs.session_start.dt.strftime('%Y-%m-%d %H:%M'); bs['session_end'] = pd.to_datetime(bs.session_end).dt.strftime('%Y-%m-%d %H:%M')
bs = bs.rename(columns={'session_id': 'person_session_id', 'n_events': 'scans'})
bs['drill_to_scans'] = 'jeddah_wms_transactions where person_session_id = this id (02_Cleaned)'
write_csv(bs, '04_warehouse_labour', 'warehouse_badge_sharing_shifts.csv')
pr = WH['prof'].rename(columns={'wh_id': 'site', 'hod': 'hour_of_day', 'on_shift': 'heads_on_shift', 'need50': 'heads_needed_median_rate', 'need75': 'heads_needed_top_quartile_rate', 'earned_h': 'workload_standard_hours'})
pr['weekday_order'] = pr.weekday.map({'Saturday': 1, 'Sunday': 2, 'Monday': 3, 'Tuesday': 4, 'Wednesday': 5, 'Thursday': 6, 'Friday': 7})
for c in ('heads_on_shift', 'heads_needed_median_rate', 'heads_needed_top_quartile_rate', 'workload_standard_hours'): pr[c] = pr[c].round(2)
pr['period'] = '2026-08-05 to 2026-09-30 (average day)'
write_csv(pr[['site', 'weekday', 'weekday_order', 'hour_of_day', 'heads_on_shift', 'heads_needed_median_rate', 'heads_needed_top_quartile_rate', 'workload_standard_hours', 'n_days', 'period']],
          '04_warehouse_labour', 'warehouse_heads_profile_by_hour.csv')
hh = WH['HH']; hh = hh[hh.date > pd.Timestamp('2026-09-02')].copy()
hh = hh.rename(columns={'wh_id': 'site', 'on_shift': 'heads_on_shift', 'need50': 'heads_needed_median_rate', 'need75': 'heads_needed_top_quartile_rate', 'earned_h': 'workload_standard_hours'})
hh['hour'] = hh.hour.dt.strftime('%Y-%m-%d %H:00')
for c in ('heads_on_shift', 'heads_needed_median_rate', 'heads_needed_top_quartile_rate', 'workload_standard_hours'): hh[c] = hh[c].round(2)
write_csv(hh[['site', 'date', 'hour', 'heads_on_shift', 'heads_needed_median_rate', 'heads_needed_top_quartile_rate', 'workload_standard_hours']], '04_warehouse_labour', 'warehouse_heads_hourly_last_4_weeks.csv')
W = WH['W']; ft = WH['ftest']; fo = WH['foct']; tg = WH['targets']; tga = WH['targets_all']
mix = (WH['HH'].assign(m=WH['HH'].hour.dt.to_period('M')).query("m >= '2026-08'").groupby('wh_id')[['earned_h', 'need50', 'need75']].sum())
r50 = (mix.need50 / mix.earned_h).to_dict(); r75 = (mix.need75 / mix.earned_h).to_dict()          # required observed hours per standard hour, by site
rows = []
for d in ft.index:
    for s in ft.columns:
        rows.append({'site': s, 'date': d, 'split': 'backtest (Sep 2026, made on 31 Aug)', 'actual_workload_standard_hours': round(W.loc[d, s], 2), 'forecast_workload_standard_hours': round(ft.loc[d, s], 2)})
for d in fo.index:
    for s in fo.columns:
        rows.append({'site': s, 'date': d, 'split': 'forecast (Oct 2026, made on 30 Sep)', 'actual_workload_standard_hours': np.nan, 'forecast_workload_standard_hours': round(fo.loc[d, s], 2)})
wf = pd.DataFrame(rows)
wf['heads_needed_median_rate_8h'] = (wf.forecast_workload_standard_hours * wf.site.map(r50).fillna(1 / tga[0.5]) / 8).round(2)
wf['heads_needed_top_quartile_rate_8h'] = (wf.forecast_workload_standard_hours * wf.site.map(r75).fillna(1 / tga[0.75]) / 8).round(2)
wf['weekday'] = pd.to_datetime(wf.date).dt.day_name(); wf['method'] = WH['wf_metrics']['method_chosen_on_jun_aug']
write_csv(wf, '04_warehouse_labour', 'warehouse_workload_forecast_daily.csv')
# October hourly plan: daily forecast x the site's weekday hour-of-day profile (Aug-Sep)
sh = WH['prof'].copy(); sh['share'] = sh.earned_h / sh.groupby(['wh_id', 'weekday']).earned_h.transform('sum')
rows = []
for d in fo.index:
    for s in fo.columns:
        p = sh[(sh.wh_id == s) & (sh.weekday == d.day_name())]
        for r in p.itertuples():
            eh = fo.loc[d, s] * (r.share if r.share == r.share else 0)
            rows.append({'site': s, 'date': d, 'hour_of_day': r.hod, 'forecast_workload_standard_hours': round(eh, 3),
                         'heads_needed_median_rate': round(eh * r50.get(s, 1 / tga[0.5]), 2), 'heads_needed_top_quartile_rate': round(eh * r75.get(s, 1 / tga[0.75]), 2),
                         'typical_heads_on_shift': round(r.on_shift, 2)})
write_csv(pd.DataFrame(rows), '04_warehouse_labour', 'warehouse_heads_forecast_hourly_oct.csv')
ra = WH['ra'].rename(columns={'wms_role': 'wms_role_menu_level', 'tran_category': 'activity_family'}); ra['scan_work_hours'] = ra.scan_work_hours.round(1); ra['share_of_role_hours'] = ra.share_of_role_hours.round(4)
write_csv(ra, '04_warehouse_labour', 'warehouse_role_activity_hours.csv')
hk = WH['hh_kpi']; gk = WH['gap_kpi']; hc = LAB['headcount']['avg_per_month']
tile(4, 'Warehouse labour', 'Paid FTE (head-count base)', hc['fte_headcount_base'], f"{hc['fte_headcount_base']:.0f}", 'FTE', 'Simulated', f"{hc['fte_scan_days']:.0f} FTE on scan days; {hc['employee_ids_scanning']:.0f} employee ids scan a month")
tile(4, 'Warehouse labour', 'Labour cost a year', PNL['pnl_annualised_base']['labour_cost_sar'], f"SAR {PNL['pnl_annualised_base']['labour_cost_sar'] / 1e6:.2f}M", 'SAR', 'Simulated', 'Monte Carlo P10-P90 SAR 6.84M-11.42M')
tile(4, 'Warehouse labour', 'Gap to median', gk['gap_to_median_sar_yr'], f"SAR {gk['gap_to_median_sar_yr'] / 1e3:,.0f}K", 'SAR / yr', 'Simulated', f"{gk['gap_to_median_share_of_labour']:.1%} of labour; every full shift at its role's median rate (100% of the gap)")
tile(4, 'Warehouse labour', 'Gap to top quartile', gk['gap_to_top_quartile_sar_yr'], f"SAR {gk['gap_to_top_quartile_sar_yr'] / 1e3:,.0f}K", 'SAR / yr', 'Simulated', f"{gk['gap_to_top_quartile_share_of_labour']:.1%} of labour; base case claims 70% of it: SAR {gk['recoverable_base_sar_yr'] / 1e3:,.0f}K")
nflag = int((bs.badge_sharing_risk.isin(['high', 'medium'])).sum())
tile(4, 'Warehouse labour', 'Shared-badge shifts', nflag, f"{nflag}", 'shifts', 'Corrected', f"{nflag / len(att):.1%} of {len(att):,} shifts are high or medium risk; {int(att.badge_sharing_split.sum())} shifts over 12 h were split into two")
tile(4, 'Warehouse labour', 'Excess head-hours', hk['excess_head_hours_vs_median_rate_last8w'], f"{hk['excess_head_hours_vs_median_rate_last8w']:.0%}", 'share', 'Corrected', f"heads on shift above workload at median rates, Aug-Sep; {hk['excess_head_hours_vs_top_quartile_rate_last8w']:.0%} at top-quartile rates")

# ================================================================== 05 van rep day
prod = VD['prod'].reset_index().rename(columns={'index': 'van'}); ps = VD['prod_stats']
risk_p = O.churn_risk_8w
nx = VD['visit_next']; nx = nx[nx.p >= 0.15].copy()
nx = nx.merge(O[['outlet_name', 'outlet_name_ar', 'churn_risk_tier', 'churn_risk_8w', 'status', 'spend_band', 'balance_sar', 'customer_type'] + [f'offer_{i}' for i in range(1, 6)] + [f'offer_{i}_value_sar' for i in range(1, 6)]],
              left_on='outlet_id', right_index=True, how='left')
order = {'High': 0, 'Lapsed': 1, 'Medium': 2, 'Low': 3}
nx['risk_order'] = nx.churn_risk_tier.map(order).fillna(4)
nx = nx.sort_values(['van', 'date', 'risk_order', 'p'], ascending=[True, True, True, False])
nx['visit_rank_in_day'] = nx.groupby(['van', 'date']).cumcount() + 1
nx['expected_visits_van_day'] = nx.groupby(['van', 'date']).p.transform('sum').round(1)
nx['offer_value_8w_sar'] = nx[[f'offer_{i}_value_sar' for i in range(1, 6)]].sum(axis=1).round(2)
vp = nx[['van', 'date', 'visit_rank_in_day', 'outlet_id', 'outlet_name', 'outlet_name_ar', 'p', 'expected_visits_van_day', 'days_since_last', 'churn_risk_tier', 'churn_risk_8w',
         'spend_band', 'customer_type', 'balance_sar'] + [f'offer_{i}' for i in range(1, 6)] + ['offer_value_8w_sar']].rename(columns={'p': 'visit_probability', 'days_since_last': 'days_since_last_visit_by_this_van'})
vp['visit_probability'] = vp.visit_probability.round(3); vp['churn_risk_8w'] = vp.churn_risk_8w.round(3); vp['weekday'] = pd.to_datetime(vp.date).dt.day_name()
write_csv(vp, '05_van_rep_day', 'van_visit_plan_next_week.csv')
ld = VL['next'].copy(); ld['sku_name'] = ld.sku.map(P.prd_desc_en); ld['principal'] = ld.sku.map(P.l1_desc)
ld = ld.rename(columns={'p50': 'forecast_p50_cases', 'p90': 'suggested_load_p90_cases', 'p10': 'forecast_p10_cases'})
for c in ('forecast_p10_cases', 'forecast_p50_cases', 'suggested_load_p90_cases', 'last_week_sold_cases', 'avg_weekly_cases_8w', 'returns_last_4w_cases'): ld[c] = ld[c].round(2)
# same selling-status rule as the SKU order on screen 3: p90 for active SKUs, p50 for slowing ones, nothing for dormant ones
SST = pd.read_csv(f'{OUT}/03_forecast/forecast_series.csv', dtype={'series_id': str}).query("level == 'sku'").set_index('series_id').selling_status
ld['sku_selling_status'] = ld.sku.map(SST).fillna('new')
ld['suggested_load_cases_rounded'] = np.select([ld.sku_selling_status == 'dormant', ld.sku_selling_status == 'slowing'],
                                               [0, np.ceil(ld.forecast_p50_cases - 1e-9)], np.ceil(ld.suggested_load_p90_cases - 1e-9)).astype(int)
ld['load_rule'] = ld.sku_selling_status.map({'active': 'p90 of demand', 'new': 'p90 of demand (SKU new since 22 Sep)', 'slowing': 'p50 only: SKU not sold for longer than usual',
                                             'dormant': 'no load: SKU not sold anywhere for 8+ weeks'})
ld['week'] = '2026-10-21 to 2026-10-27'
write_csv(ld[['van', 'week', 'sku', 'sku_name', 'principal', 'sku_selling_status', 'forecast_p10_cases', 'forecast_p50_cases', 'suggested_load_p90_cases', 'suggested_load_cases_rounded', 'load_rule',
              'last_week_sold_cases', 'avg_weekly_cases_8w', 'returns_last_4w_cases']].sort_values(['van', 'forecast_p50_cases'], ascending=[True, False]), '05_van_rep_day', 'van_load_suggestion_next_week.csv')
rc = RECO.merge(O[['status', 'outlet_name', 'outlet_name_ar']], left_on='outlet_id', right_index=True, how='left').rename(columns={'status': 'outlet_status'})
write_csv(rc[['outlet_id', 'outlet_name', 'outlet_name_ar', 'outlet_status', 'main_van', 'rank', 'sku', 'sku_name', 'principal', 'p_buy_8w', 'typical_8w_spend_sar', 'expected_value_8w_sar', 'why', 'picks_made_on']],
          '05_van_rep_day', 'next_best_sku_by_outlet.csv')
tr = RT['test_res']; te_alt = RS['test']
bt = pd.DataFrame([
    {'method': 'Our next-best SKU (picks made 25 Aug 2026)', 'precision_at_5': tr['precision_at_5'], 'outlets_with_at_least_one_pick_bought': tr['outlets_with_hit'], 'note': 'weights chosen on the 30 Jun back-test'},
    {'method': 'Push best-sellers (top 5 not yet bought)', 'precision_at_5': te_alt['push_best_sellers']['precision_at_5'], 'outlets_with_at_least_one_pick_bought': te_alt['push_best_sellers']['outlets_with_hit'], 'note': 'baseline'},
    {'method': "Van best-sellers (what the outlet's van sells to others)", 'precision_at_5': te_alt['van_best_sellers']['precision_at_5'], 'outlets_with_at_least_one_pick_bought': te_alt['van_best_sellers']['outlets_with_hit'], 'note': 'simple rule'},
    {'method': 'Random new SKUs', 'precision_at_5': tr['random_precision'], 'outlets_with_at_least_one_pick_bought': np.nan, 'note': 'expected value of 5 random picks'}])
bt['lift_vs_best_sellers'] = bt.precision_at_5 / te_alt['push_best_sellers']['precision_at_5']; bt['lift_vs_random'] = bt.precision_at_5 / tr['random_precision']
bt['test_window'] = '26 Aug - 20 Oct 2026 (8 weeks the model never saw)'; bt = bt.round(4)
write_csv(bt, '05_van_rep_day', 'next_best_sku_backtest.csv')
hbr = RT['hit_by_rank'].rename('share_bought_8w').reset_index().rename(columns={'rank': 'pick_rank'}); write_csv(hbr.round(4), '05_van_rep_day', 'next_best_sku_hit_rate_by_rank.csv')
prod = prod.round(3)
write_csv(prod, '05_van_rep_day', 'van_list.csv')
vm = VD['visit_metrics']; lm = VL['metrics']
mets = [{'area': 'visit plan', 'metric': k, 'value': v if isinstance(v, str) else round(float(v), 4)} for k, v in vm.items()]
mets += [{'area': 'van load', 'metric': k, 'value': round(float(v), 4)} for k, v in lm.items()]
mets += [{'area': 'van productivity', 'metric': k, 'value': round(float(v), 4)} for k, v in ps.items()]
write_csv(pd.DataFrame(mets), '05_van_rep_day', 'van_metrics.csv')
tile(5, 'Van rep day', 'Picks bought unprompted (next-best SKU)', tr['precision_at_5'], f"{tr['precision_at_5']:.1%}", 'share', 'Actual', f"vs {te_alt['push_best_sellers']['precision_at_5']:.1%} push best-sellers and {tr['random_precision']:.2%} random; 8 unseen weeks after 25 Aug")
tile(5, 'Van rep day', 'Lift vs best-sellers', tr['precision_at_5'] / te_alt['push_best_sellers']['precision_at_5'], f"{tr['precision_at_5'] / te_alt['push_best_sellers']['precision_at_5']:.1f}x", 'x', 'Actual', f"{tr['precision_at_5'] / tr['random_precision']:.0f}x random")
tile(5, 'Van rep day', 'Outlets that bought a pick', tr['outlets_with_hit'], f"{tr['outlets_with_hit']:.0%}", 'share', 'Actual')
tile(5, 'Van rep day', 'Outlets per van-day', ps['median_outlets_per_van_day_8w'], f"{ps['median_outlets_per_van_day_8w']:.0f}", 'outlets', 'Actual', f"best quartile {ps['best_quartile_outlets_per_van_day_8w']:.0f}; last 8 weeks")
tile(5, 'Van rep day', 'Outlets served by 2+ vans', ps['share_outlets_served_by_2plus_vans'], f"{ps['share_outlets_served_by_2plus_vans']:.0%}", 'share', 'Actual')
tile(5, 'Van rep day', 'Visit prediction', vm['auc_model'], f"AUC {vm['auc_model']:.2f}", 'AUC', 'Actual', f"vs {vm['auc_same_weekday_rule']:.2f} same-weekday rule; week 14-20 Oct")

# ================================================================== 06 outlets
cols = ['outlet_name', 'outlet_name_ar', 'customer_type', 'payment_term', 'channel_code', 'channel_inferred', 'spend_band', 'status', 'churn_risk_tier', 'churn_risk_8w',
        'first_purchase', 'last_purchase', 'days_since_last', 'usual_gap_days', 'purchase_days', 'invoices', 'sales_window_sar', 'annual_value_sar', 'balance_sar',
        'main_van', 'last_van', 'vans_in_window', 'lat', 'lon', 'last_sku_1', 'last_sku_2', 'last_sku_3'] + [f'offer_{i}' for i in range(1, 6)] + ['nearest_active_van', 'nearest_active_km']
ot = O[cols].copy(); ot.index.name = 'outlet_id'; ot = ot.reset_index()
for c in ('churn_risk_8w',): ot[c] = ot[c].round(4)
for c in ('sales_window_sar', 'annual_value_sar', 'balance_sar'): ot[c] = ot[c].round(2)
write_csv(ot, '06_outlets', 'outlets.csv')
Cm = C.copy(); Cm['outlet_id'] = outlet_id(Cm.dist_cd, Cm.cust_cd); Cm = Cm[Cm.latitude.notna() & Cm.longitude.notna()]
mp = Cm[['outlet_id', 'cust_name', 'latitude', 'longitude', 'cust_type', 'cust_hier2']].rename(columns={'cust_name': 'outlet_name', 'latitude': 'lat', 'longitude': 'lon', 'cust_hier2': 'channel_code'})
mp = mp.merge(O[['status', 'churn_risk_tier', 'spend_band', 'main_van']], left_on='outlet_id', right_index=True, how='left')
mp['status'] = mp.status.fillna('no purchase Mar-Oct 2026'); mp['churn_risk_tier'] = mp.churn_risk_tier.fillna(''); mp['spend_band'] = mp.spend_band.fillna('')
mp['ring_lapsed'] = mp.status == 'lapsed'; mp['customer_type'] = np.where(mp.cust_type == 'Q', 'Credit', 'Cash'); mp = mp.drop(columns='cust_type')
write_csv(mp, '06_outlets', 'outlet_map_points.csv')
ms = B[B.sellable].groupby(['outlet_id', 'month']).net_amt.sum().round(2).rename('net_sales_sar').reset_index()
write_csv(ms, '06_outlets', 'outlet_monthly_sales.csv')
wb = O[O.status == 'lapsed'].sort_values('annual_value_sar', ascending=False)
wb = wb[['outlet_name', 'outlet_name_ar', 'last_van', 'last_purchase', 'days_since_last', 'annual_value_sar', 'nearest_active_van', 'nearest_active_km', 'spend_band', 'customer_type', 'balance_sar']
        + [f'offer_{i}' for i in range(1, 6)]].reset_index().rename(columns={'days_since_last': 'days_quiet', 'offer_1': 'next_best_sku_1', 'offer_2': 'next_best_sku_2', 'offer_3': 'next_best_sku_3', 'offer_4': 'next_best_sku_4', 'offer_5': 'next_best_sku_5'})
wb['annual_value_sar'] = wb.annual_value_sar.round(2)
write_csv(wb, '06_outlets', 'winback_list.csv')
cr = CH['res']
cm = [{'metric': k, 'value': v if isinstance(v, str) else round(float(v), 4)} for k, v in cr.items()]
cm += [{'metric': f'test_quintile_{int(i) + 1}_mean_score', 'value': round(float(r.score), 4)} for i, r in CH['quintiles'].iterrows()]
cm += [{'metric': f'test_quintile_{int(i) + 1}_went_quiet_rate', 'value': round(float(r.went_quiet), 4)} for i, r in CH['quintiles'].iterrows()]
cm += [{'metric': f'feature_importance::{k}', 'value': int(v)} for k, v in CH['imp'].head(10).items()]
write_csv(pd.DataFrame(cm), '06_outlets', 'churn_model_metrics.csv')
ch_ = cr['chosen_variant']
tile(6, 'Outlets', 'Churn model AUC', cr[f'auc_{ch_}'], f"{cr[f'auc_{ch_}']:.2f}", 'AUC', 'Actual', f"vs {cr['auc_recency_rule']:.2f} recency rule; tested at 25 Aug on {int(cr['test_outlets'])} outlets the model never saw")
tile(6, 'Outlets', 'Riskiest 20% catch', cr[f'top20_capture_{ch_}'], f"{cr[f'top20_capture_{ch_}']:.0%}", 'share', 'Actual', f"of the outlets that went quiet in the next 8 weeks; sorting by days since last purchase catches {cr['top20_capture_recency_rule']:.0%}")
tile(6, 'Outlets', 'Lapsed outlets', OK['lapsed'], f"{OK['lapsed']}", 'outlets', 'Actual', f"worth SAR {OK['lapsed_annual_value_sar'] / 1e3:,.0f}K a year (average since first purchase)")
tile(6, 'Outlets', 'High-risk active outlets', OK['high_risk_outlets'], f"{OK['high_risk_outlets']}", 'outlets', 'Actual', 'chance of no purchase in the next 8 weeks >= 25%')

# ================================================================== 07 receivables
L = RC['L'].reset_index()
cl = L[['outlet_id', 'outlet_name', 'outlet_name_ar', 'priority', 'van', 'outstanding_bal_sar', 'past_due_sar', 'past_due_over_60_sar', 'unmatched_sar', 'exposure_sar', 'max_days_past_due',
        'monthly_buying_sar', 'days_since_last_purchase', 'return_rate_13w', 'credit_limit_sar', 'credit_limit_is_placeholder', 'over_limit', 'churn_risk_8w', 'risk_points', 'why_flagged',
        'open_not_due', 'open_1-30', 'open_31-60', 'open_61-90', 'open_90+', 'invterm_cd']].rename(columns={'outstanding_bal_sar': 'balance_sar', 'max_days_past_due': 'days_late', 'invterm_cd': 'payment_term',
        'open_not_due': 'not_due_sar', 'open_1-30': 'due_1_30_sar', 'open_31-60': 'due_31_60_sar', 'open_61-90': 'due_61_90_sar', 'open_90+': 'due_90_plus_sar'})
cl['credit_limit_sar'] = np.where(cl.credit_limit_is_placeholder, np.nan, cl.credit_limit_sar)
# who can collect: the van rep only if the account's van is still on the road; otherwise credit control
active_vans = set(prod.loc[prod.active_now.astype(bool), 'van'])
cl['collect_via'] = np.select([cl.van.isna(), ~cl.van.isin(active_vans)], ['Credit control: no purchase since March', 'Credit control: van no longer active'], 'Van rep on route')
cl.insert(cl.columns.get_loc('van') + 1, 'collect_via', cl.pop('collect_via'))
for c in cl.columns:
    if cl[c].dtype.kind == 'f': cl[c] = cl[c].round(2)
cl = cl.sort_values(['priority', 'exposure_sar'], ascending=[True, False])
write_csv(cl, '07_receivables', 'receivables_collection_list.csv')
bk = RC['buckets'].copy(); bk['amount_sar'] = bk.amount_sar.round(2); bk['share_of_book'] = bk.share_of_book.round(4)
write_csv(bk, '07_receivables', 'receivables_ageing_buckets.csv')
tm = RC['terms'].copy(); tm['balance_sar'] = tm.balance_sar.round(2); tm['customer_type'] = np.where(tm.cust_type == 'Q', 'Credit (Q)', 'Cash (C)'); tm = tm.drop(columns='cust_type')
write_csv(tm, '07_receivables', 'receivables_by_term.csv')
du = RC['due'].rename(columns={'when': 'due_week_ending', 'open_amount_sar': 'amount_sar'}); du['amount_sar'] = du.amount_sar.round(2)
write_csv(du, '07_receivables', 'receivables_cash_due_by_week.csv')
k = RC['kpi']
tile(7, 'Receivables', 'Credit book', k['credit_book_gross_sar'], f"SAR {k['credit_book_gross_sar'] / 1e6:.2f}M", 'SAR', 'Actual', f"{k['customers_with_balance']:,} credit customers with a balance; snapshot 20 Oct 2026")
tile(7, 'Receivables', 'Past due', k['past_due_sar'], f"SAR {k['past_due_sar'] / 1e3:,.0f}K", 'SAR', 'Corrected', f"{k['past_due_share_of_balance']:.0%} of the book; plus SAR {k['unmatched_sar'] / 1e3:,.0f}K not matched to invoices since March")
tile(7, 'Receivables', 'DSO', k['dso_days'], f"{k['dso_days']:.0f} days", 'days', 'Corrected', 'credit balance / credit sales of the last 90 days per day')
p1 = cl[cl.priority == 'P1']; p1r = p1[p1.collect_via == 'Van rep on route']; p1c = p1[p1.collect_via != 'Van rep on route']
tile(7, 'Receivables', 'P1 accounts', k['p1_accounts'], f"{k['p1_accounts']}", 'accounts', 'Corrected',
     f"SAR {k['p1_exposure_sar'] / 1e3:,.0f}K exposure (past due + unmatched): {len(p1r)} on active van routes (SAR {p1r.exposure_sar.sum() / 1e3:,.0f}K), "
     f"{len(p1c)} for credit control (SAR {p1c.exposure_sar.sum() / 1e3:,.0f}K), mostly no purchase since March")

# ================================================================== 08 value roll-up (levers, assumptions, data asks)
lev0 = pd.read_parquet(f'{SIM}/sim_opportunity_levers.parquet').pivot(index='lever', columns='case', values='profit_annual_sar')
A = pd.DataFrame([
    ('A01', 'Gross margin on extra sales', 'share', 0.12, 0.16, 0.20, 'Researched: Almunajem Foods FY2025 gross margin 16% (another Saudi distributor; S38)'),
    ('A02', 'Extra conversion when reps offer the pick (vs unprompted)', 'share of unprompted rate', 0.10, 0.25, 0.50, 'Assumption: unprompted buying was measured; prompting effect needs a pilot'),
    ('A03', 'Lapsed outlets won back', 'share', 0.10, 0.20, 0.30, 'Assumption'),
    ('A04', 'Cost of capital', 'share per year', 0.06, 0.08, 0.10, 'Assumption: SAR financing cost'),
    ('A05', 'Inventory carrying cost', 'share per year', 0.15, 0.20, 0.25, 'Assumption: capital + storage + obsolescence'),
    ('A06', 'Past-due balance collected', 'share', 0.30, 0.50, 0.70, 'Assumption'),
    ('A07', 'Cost of one van-day (rep + van + fuel)', 'SAR per day', 350, 450, 550, 'Assumption: rep loaded wage + van lease + fuel'),
    ('A08', 'Route productivity gap captured', 'share', 0.25, 0.50, 0.75, 'Assumption'),
    ('A09', 'Warehouse operative wage (expat, loaded)', 'SAR per month', 2859, 4101, 5358, 'Researched build-up (03_Simulated sim_labour_cost_model)'),
    ('A10', 'Storekeeper vs tally clerk wage difference', 'SAR per paid hour', 0.0, 0.57, 3.0, 'Researched: no real premium found in postings; high case = an assumed premium')],
    columns=['assumption_id', 'assumption', 'unit', 'low', 'base', 'high', 'source'])
a = A.set_index('assumption_id')
active = O[(O.status != 'lapsed') & (O.days_since_last <= 56)].index
ws_driver = RECO[RECO.outlet_id.isin(active)].expected_value_8w_sar.sum() * 365 / 56          # unprompted pick sales a year
lapsed_val = OK['lapsed_annual_value_sar']
ssm = fm['sku']; ss_red_price = ssm['safety_stock_value_naive_sar'] - ssm['safety_stock_value_model_sar']
pdue = k['past_due_sar']
vdy = VD['prod'].active_days_8w.sum() * 365 / 56; gap_v = 1 - ps['median_outlets_per_van_day_8w'] / ps['best_quartile_outlets_per_van_day_8w']
rar = WH['ra'][WH['ra'].wms_role.eq('Storekeeper')]
sk_share = rar[rar.flag_senior_role_on_floor_work].scan_work_hours.sum() / rar.scan_work_hours.sum()      # share of storekeeper work time spent on floor work
role_hours_yr = sk_share * LAB['paid_hours_by_wms_role']['Storekeeper'] * 1.5                          # storekeeper paid hours a year on floor work (scan-day basis)
rows = []
def lever(i, area, func, name, driver_desc, driver, unit, f_low, f_base, f_high, oneoff=(0, 0, 0), label='Simulated', proof='', include=True, overlap='', formula='', evidence='Actual'):
    rows.append({'lever_id': i, 'level_2_area': area, 'level_3_function': func, 'level_4_lever': name, 'driver': driver_desc, 'driver_value': round(driver, 2), 'driver_unit': unit,
                 'formula': formula, 'low_sar_yr': round(f_low), 'base_sar_yr': round(f_base), 'high_sar_yr': round(f_high), 'one_off_cash_low_sar': round(oneoff[0]),
                 'one_off_cash_base_sar': round(oneoff[1]), 'one_off_cash_high_sar': round(oneoff[2]), 'value_label': label, 'evidence_label': evidence,
                 'proof_screen': proof, 'include_in_total': include, 'overlap_note': overlap})
lever('V01', 'Warehouse', 'Labour', 'Productivity to own top quartile', 'recoverable labour on worked shifts (Monte Carlo)', gk['recoverable_base_sar_yr'], 'SAR a year',
      PNL['mc_summary']['recoverable_labour_annual_sar']['p10'], gk['recoverable_base_sar_yr'], PNL['mc_summary']['recoverable_labour_annual_sar']['p90'], proof='4 Warehouse labour',
      formula='Monte Carlo P10 / base / P90 of recoverable labour (03_Simulated)')
lever('V02', 'Warehouse', 'Labour', 'Role alignment (storekeepers on floor work)', 'storekeeper paid hours on loading, picking and moves a year', role_hours_yr, 'paid hours a year',
      role_hours_yr * a.loc['A10', 'low'], role_hours_yr * a.loc['A10', 'base'], role_hours_yr * a.loc['A10', 'high'], proof='4 Warehouse labour', formula='hours x A10',
      overlap='small because researched wages show no storekeeper premium; needs HR job titles and pay (gap G22)')
lever('V03', 'Retail (B2B)', 'Van sales', 'Route productivity (fleet right-sizing)', 'van-days a year at the current pace', vdy, 'van-days a year',
      vdy * gap_v * a.loc['A08', 'low'] * a.loc['A07', 'low'], vdy * gap_v * a.loc['A08', 'base'] * a.loc['A07', 'base'], vdy * gap_v * a.loc['A08', 'high'] * a.loc['A07', 'high'],
      proof='5 Van rep day', formula=f'van-days x (1 - median/best-quartile outlets per day = {gap_v:.0%}) x A08 x A07', overlap='needs route plans to confirm; distances not modelled')
lever('V04', 'Retail (B2B)', 'Van sales', 'Returns above the median rep rate', 'excess returns (03_Simulated lever)', lev0.loc['Returns reduction', 'base'], 'SAR a year',
      lev0.loc['Returns reduction', 'low'], lev0.loc['Returns reduction', 'base'], lev0.loc['Returns reduction', 'high'], proof='5 Van rep day', formula='03_Simulated sim_opportunity_levers')
lever('V05', 'Retail (B2B)', 'Outlet lifecycle', 'Win back lapsed outlets', 'annual value of lapsed outlets', lapsed_val, 'SAR sales a year',
      lapsed_val * a.loc['A03', 'low'] * a.loc['A01', 'low'], lapsed_val * a.loc['A03', 'base'] * a.loc['A01', 'base'], lapsed_val * a.loc['A03', 'high'] * a.loc['A01', 'high'],
      proof='6 Outlets', formula='lapsed annual value x A03 x A01')
lever('V06', 'Distribution', 'Advanced sales', 'Whitespace: next-best SKU', 'sales that active outlets make unprompted on our picks, a year', ws_driver, 'SAR sales a year',
      ws_driver * a.loc['A02', 'low'] * a.loc['A01', 'low'], ws_driver * a.loc['A02', 'base'] * a.loc['A01', 'base'], ws_driver * a.loc['A02', 'high'] * a.loc['A01', 'high'],
      proof='5 Van rep day', formula='unprompted pick sales x A02 x A01', overlap='overlaps the van-rep productivity view (V09), which is not added')
lever('V07', 'Distribution', 'Receivables', 'Collections', 'past-due balance', pdue, 'SAR', pdue * a.loc['A06', 'low'] * a.loc['A04', 'low'], pdue * a.loc['A06', 'base'] * a.loc['A04', 'base'],
      pdue * a.loc['A06', 'high'] * a.loc['A04', 'high'], oneoff=(pdue * a.loc['A06', 'low'], pdue * a.loc['A06', 'base'], pdue * a.loc['A06', 'high']), label='Corrected',
      proof='7 Receivables', formula='past due x A06 = one-off cash; x A04 = yearly financing saving', overlap='bad-debt write-offs avoided not valued (needs the ledger)')
lever('V08', 'Distribution', 'Demand', 'Safety stock from better forecasts', 'safety-stock reduction at selling price (SKU level)', ss_red_price, 'SAR',
      ss_red_price * (1 - a.loc['A01', 'high']) * a.loc['A05', 'low'], ss_red_price * (1 - a.loc['A01', 'base']) * a.loc['A05', 'base'], ss_red_price * (1 - a.loc['A01', 'low']) * a.loc['A05', 'high'],
      oneoff=(ss_red_price * (1 - a.loc['A01', 'high']), ss_red_price * (1 - a.loc['A01', 'base']), ss_red_price * (1 - a.loc['A01', 'low'])),
      proof='3 Forecast proof', formula='reduction at cost (1 - A01) = one-off cash; x A05 = yearly carrying saving', overlap='stock on hand not in the data; safety stock is a model figure')
lever('V09', 'Retail (B2B)', 'Van sales', 'Van-rep productivity (alternative view)', 'sales gap of below-median reps (03_Simulated lever)', lev0.loc['Van-rep productivity', 'base'], 'SAR a year',
      lev0.loc['Van-rep productivity', 'low'], lev0.loc['Van-rep productivity', 'base'], lev0.loc['Van-rep productivity', 'high'], proof='5 Van rep day', include=False,
      overlap='overlaps whitespace (V06) and win-back (V05): shown, not added')
lever('V10', 'Retail (B2B)', 'Pricing', 'Price-spread proxy', 'lines sold > 5% below the monthly median price (03_Simulated lever)', lev0.loc['Price leakage', 'base'], 'SAR a year',
      lev0.loc['Price leakage', 'low'], lev0.loc['Price leakage', 'base'], lev0.loc['Price leakage', 'high'], proof='-', include=False,
      overlap='a proxy only: real price leakage needs trade-spend and promotion data; shown, not added')
LV = pd.DataFrame(rows)
tot = LV[LV.include_in_total][['low_sar_yr', 'base_sar_yr', 'high_sar_yr', 'one_off_cash_low_sar', 'one_off_cash_base_sar', 'one_off_cash_high_sar']].sum()
LV = pd.concat([LV, pd.DataFrame([{'lever_id': 'TOTAL', 'level_4_lever': 'Total a year (levers with include_in_total = True)', **tot.round().to_dict()}])], ignore_index=True)
write_csv(LV, '08_value_rollup', 'value_levers.csv')
used = {'A01': 'V05, V06, V08', 'A02': 'V06', 'A03': 'V05', 'A04': 'V07', 'A05': 'V08', 'A06': 'V07', 'A07': 'V03', 'A08': 'V03', 'A09': 'V01 (via 03_Simulated)', 'A10': 'V02'}
A['used_by'] = A.assumption_id.map(used)
write_csv(A, '08_value_rollup', 'value_assumptions.csv')
asks = pd.DataFrame([('Pay rates, roster, attendance (Jeddah)', 'V01, V02', 'Warehouse labour'), ('Van load / unload by SKU, route plans', 'V03, V04', 'Van rep day'),
                     ('Customer ledger, real credit limits', 'V07', 'Receivables'), ('Warehouse-served invoices with channel', 'V06', 'Forecast proof'),
                     ('Promotions and trade-spend agreements', 'V10', 'Forecast proof'), ('Stock on hand and cost price by SKU', 'V08', 'Forecast proof'),
                     ('12-24 months of sales (Hijri dates)', 'V08', 'Forecast proof'), ('Outlet visit logs / geo-fence flags', 'V05', 'Outlets')], columns=['data_needed', 'levers_moved_to_actual', 'screen'])
write_csv(asks, '08_value_rollup', 'value_data_asks.csv')
tile(2, 'Impact tree', 'Estimated annual saving (base)', tot.base_sar_yr, f"SAR {tot.base_sar_yr / 1e6:.2f}M", 'SAR / yr', 'Mixed', f"range SAR {tot.low_sar_yr / 1e6:.2f}M - {tot.high_sar_yr / 1e6:.2f}M; levers that overlap are not added")
tile(2, 'Impact tree', 'One-off cash', tot.one_off_cash_base_sar, f"SAR {tot.one_off_cash_base_sar / 1e6:.2f}M", 'SAR', 'Mixed', 'collections + safety stock')
write_csv(pd.DataFrame(tiles), '', '00_KPI_TILES.csv')
print(LV[['lever_id', 'level_4_lever', 'low_sar_yr', 'base_sar_yr', 'high_sar_yr', 'one_off_cash_base_sar', 'include_in_total']].to_string())