"""Stage 4: modern trade, discounters, wholesale and e-com operations:
PO (F04), fulfilment (F05), promo execution (F09), sell-out (F03), customer stock (F17), trade spend (F10),
merchandiser visits (F11), shelf audits (F12), price observations (F06), competitor activity (F07)."""
import numpy as np, pandas as pd
from common import *
import warnings; warnings.filterwarnings('ignore')

r = rng(4)
FI = load('fact_sell_in'); FI['doc_date'] = pd.to_datetime(FI.doc_date)
outl = load('dim_outlet'); acc = load('dim_account').set_index('account_id'); prod = load('dim_product_ext').set_index('sku')
PR = load('fact_promotion'); emp = load('dim_employee'); comp = load('dim_competitor_product')
NG = FI[FI.channel_id != 'GR'].copy()
city = outl.set_index('outlet_id').city

# ---------------- F05 fulfilment (all non-grocery order lines) + rejected / unshipped lines
F = NG[['channel_id', 'account_id', 'outlet_id', 'doc_no', 'doc_line_seq', 'sku', 'qty', 'conv_factor', 'doc_date', 'ship_to_city']].copy()
F = F.rename(columns={'doc_no': 'order_id', 'doc_line_seq': 'order_line', 'qty': 'shipped_cases', 'doc_date': 'ship_date'})
peak = F.ship_date.between('2026-03-15', '2026-03-31') | F.ship_date.between('2026-08-20', '2026-09-10')
short = r.random(len(F)) < np.where(peak, 0.10, 0.05)
F['ordered_cases'] = F.shipped_cases
F.loc[short, 'ordered_cases'] = np.ceil(F.loc[short, 'shipped_cases'] / r.uniform(0.5, 0.92, short.sum())).astype(int)
F['allocated_cases'] = F.shipped_cases
F['cut_reason'] = None
F.loc[short, 'cut_reason'] = r.choice(['no stock', 'quality hold', 'truck capacity', 'late picking'], short.sum(), p=[0.55, 0.1, 0.2, 0.15])
lead = F.channel_id.map({'MT': 2, 'DS': 3, 'WS': 1, 'EC': 2})
F['order_date'] = F.ship_date - pd.to_timedelta(lead + r.integers(0, 2, len(F)), unit='D')
transit = np.where(F.ship_to_city == 'Riyadh', r.integers(0, 2, len(F)), r.integers(1, 4, len(F)))
F['promised_date'] = F.order_date + pd.to_timedelta(lead + 2 + np.where(F.ship_to_city == 'Riyadh', 0, 2), unit='D')
late = r.random(len(F)) < np.where(peak, 0.12, 0.05)
F['delivery_date'] = F.ship_date + pd.to_timedelta(transit + np.where(late, r.integers(1, 4, len(F)), 0), unit='D')
F['delivered_cases'] = F.shipped_cases - np.where(r.random(len(F)) < 0.004, 1, 0)   # rare refusals at the door
F.loc[F.delivered_cases < F.shipped_cases, 'cut_reason'] = 'customer refusal'
F['on_time_flag'] = F.delivery_date <= F.promised_date
F['in_full_flag'] = F.delivered_cases >= F.ordered_cases
F['otif_flag'] = F.on_time_flag & F.in_full_flag
F['warehouse_id'] = np.where(F.ship_to_city == 'Riyadh', 'RYD-DC', np.where(r.random(len(F)) < 0.7, '2100-04', '2100-07'))
for c in ['order_date', 'ship_date', 'promised_date', 'delivery_date']: F[c] = F[c].dt.date
F['basis'] = 'Simulated'
save(F, 'fact_order_fulfilment')
print('  line fill', round(F.shipped_cases.sum() / F.ordered_cases.sum(), 3), '| OTIF lines', round(F.otif_flag.mean(), 3),
      F.groupby('channel_id').otif_flag.mean().round(3).to_dict())

# ---------------- F04 purchase orders (MT, DS, EC accounts) incl. rejected lines that never shipped
po = F[F.channel_id.isin(['MT', 'DS', 'EC'])].copy()
po['po_id'] = 'PO-' + po.order_id
po['expected_cases'] = np.maximum(1, np.round(po.ordered_cases * r.normal(1, 0.15, len(po)))).astype(int)
po['status'] = np.where(po.shipped_cases >= po.ordered_cases, 'shipped in full', 'partially shipped')
po['rejection_reason'] = None
n_rej = int(len(po) * 0.035)
rej = po.sample(n_rej, random_state=SEED).copy()
rej['sku'] = r.choice(po.sku.unique(), n_rej)
rej['order_line'] = rej.order_line + 1000
rej['ordered_cases'] = np.maximum(1, r.poisson(4, n_rej)); rej['shipped_cases'] = 0
rej['status'] = 'rejected'
rej['rejection_reason'] = r.choice(['price mismatch', 'item not listed', 'PO expired', 'no stock', 'duplicate PO'], n_rej, p=[0.35, 0.25, 0.15, 0.15, 0.1])
po = pd.concat([po, rej], ignore_index=True)
po['appointment_date'] = po.promised_date
po['delay_days'] = (pd.to_datetime(po.delivery_date) - pd.to_datetime(po.promised_date)).dt.days.clip(lower=0)
po.loc[po.status == 'rejected', ['delay_days', 'delivery_date']] = [None, None]
PO = po[['po_id', 'order_line', 'channel_id', 'account_id', 'outlet_id', 'sku', 'order_date', 'promised_date', 'expected_cases', 'ordered_cases',
         'shipped_cases', 'status', 'rejection_reason', 'appointment_date', 'delivery_date', 'delay_days']].rename(columns={'order_line': 'po_line',
         'order_date': 'po_date', 'promised_date': 'requested_delivery_date', 'outlet_id': 'ship_to_outlet_id'})
PO['basis'] = 'Simulated'
save(PO, 'fact_purchase_order')

# ---------------- F23 merchandiser absences (used for visits; saved again in the team stage)
mer = emp[emp.role == 'Merchandiser']
cal = calendar()
ab = []
for _, m in mer.iterrows():
    for d in cal.date:
        if m.status == 'vacant' and d >= pd.Timestamp('2026-08-01'):
            ab.append((m.employee_id, d, 'vacant'))
        elif d.day_name() != 'Friday' and r.random() < 0.045:
            ab.append((m.employee_id, d, r.choice(['sick', 'absent', 'annual leave'], p=[0.4, 0.25, 0.35])))
AB = pd.DataFrame(ab, columns=['employee_id', 'date', 'status'])
# annual leave blocks of 10-14 days for 30% of merchandisers
for m in r.choice(mer.employee_id, 12, replace=False):
    s0 = START + pd.Timedelta(days=int(r.integers(60, 200)))
    AB = pd.concat([AB, pd.DataFrame({'employee_id': m, 'date': pd.date_range(s0, periods=int(r.integers(10, 15))), 'status': 'annual leave'})])
AB = AB.drop_duplicates(['employee_id', 'date'])
AB.to_parquet(f'{OUT}/_tmp_merch_absence.parquet', index=False)
absent = set(zip(AB.employee_id, AB.date))

# ---------------- F11 merchandiser visits (MT, DS stores)
st = outl[outl.channel_id.isin(['MT', 'DS'])].copy()
freq = st.size_band.map({'XL': 3, 'L': 3, 'M': 2, 'S': 1, 'XS': 1})
pat = {1: ['Sunday'], 2: ['Sunday', 'Wednesday'], 3: ['Saturday', 'Monday', 'Wednesday']}
vis = []
for (_, s), f in zip(st.iterrows(), freq):
    for d in cal.date[cal.date.dt.day_name().isin(pat[f])]:
        if d < pd.Timestamp(s.open_date) or (pd.notna(s.close_date) and d > pd.Timestamp(s.close_date)): continue
        vis.append((s.outlet_id, s.account_id, s.channel_id, s.merchandiser_id, d, s.size_band))
MV = pd.DataFrame(vis, columns=['outlet_id', 'account_id', 'channel_id', 'merchandiser_id', 'planned_date', 'size_band'])
ab_status = AB.set_index(['employee_id', 'date']).status
MV['absn'] = [ab_status.get((m, d)) for m, d in zip(MV.merchandiser_id, MV.planned_date)]
MV['status'] = np.where(MV.absn.notna(), np.where(MV.absn == 'vacant', 'missed - vacant', 'missed - absent'),
                        np.where(r.random(len(MV)) < 0.06, 'missed', 'done'))
MV['check_in'] = MV.planned_date + pd.to_timedelta(r.integers(8 * 60, 17 * 60, len(MV)), unit='m')
MV['check_out'] = MV.check_in + pd.to_timedelta(MV.size_band.map({'XL': 75, 'L': 60, 'M': 45, 'S': 30, 'XS': 25}) * r.uniform(0.7, 1.3, len(MV)), unit='m')
MV['gps_distance_m'] = np.round(np.abs(r.normal(0, 40, len(MV))))
fake = MV.index[(MV.status == 'done')].to_numpy(); fake = r.choice(fake, 150, replace=False)
MV.loc[fake, 'gps_distance_m'] = r.integers(400, 3000, 150)
MV['anomaly_label'] = None; MV.loc[fake, 'anomaly_label'] = 'checkin_far_from_store'
MV['photo_count'] = np.where(MV.status == 'done', r.integers(3, 15, len(MV)), 0)
MV.loc[MV.status != 'done', ['check_in', 'check_out', 'gps_distance_m']] = [pd.NaT, pd.NaT, None]
MV.insert(0, 'visit_id', [f'MV{i+1:06d}' for i in range(len(MV))])
MV['planned_date'] = MV.planned_date.dt.date
save(MV.drop(columns=['absn', 'size_band']), 'fact_merch_visit')

# ---------------- F09 promo execution (store × promotion × weekly check)
PH = PR.drop_duplicates('promotion_id')[['promotion_id', 'account_id', 'channel_id', 'start_date', 'end_date', 'display_flag']]
comp_rate = {a: v for a, v in zip(acc.index, r.uniform(0.75, 0.92, len(acc)))}
pe = []
for _, p in PH[PH.channel_id.isin(['MT', 'DS'])].iterrows():
    stores = st[st.account_id == p.account_id].outlet_id.values
    checks = pd.date_range(pd.Timestamp(p.start_date) + pd.Timedelta(days=1), pd.Timestamp(p.end_date), freq='7D')
    for s in stores:
        live_store = r.random() < comp_rate[p.account_id]
        for d in checks:
            pc = live_store or r.random() < 0.3
            disp = (not p.display_flag) or (live_store and r.random() < 0.85)
            pos = r.random() < (0.9 if live_store else 0.5)
            pe.append((p.promotion_id, s, p.account_id, d.date(), pc, disp, pos, bool(pc and disp)))
PE = pd.DataFrame(pe, columns=['promotion_id', 'outlet_id', 'account_id', 'check_date', 'price_correct', 'display_built', 'pos_material', 'live_flag'])
PE.insert(0, 'check_id', [f'PC{i+1:06d}' for i in range(len(PE))])
save(PE, 'fact_promo_execution')
print('  promo live rate', round(PE.live_flag.mean(), 3))

# ---------------- F03 sell-out (EPOS shared by KA01–KA03 for their top 150 SKUs) + F17 customer stock (month-end snapshots)
ep_acc = ['KA01', 'KA02', 'KA03']
x = NG[NG.account_id.isin(ep_acc)].copy()
x['week_end'] = x.doc_date + pd.to_timedelta((1 - x.doc_date.dt.weekday) % 7, unit='D')   # Tuesday-closing weeks
topsku = x.groupby(['account_id', 'sku']).net_amt.sum().groupby(level=0, group_keys=False).nlargest(150).reset_index()[['account_id', 'sku']]
x = x.merge(topsku, on=['account_id', 'sku'])
wk = pd.date_range(x.week_end.min(), x.week_end.max(), freq='7D')
si = x.groupby(['outlet_id', 'sku', 'week_end']).qty_base_ea.sum()
nw = si.groupby(level=[0, 1]).size(); si = si[si.index.droplevel(2).isin(nw[nw >= 8].index)]   # regularly stocked lines only
live = PE.merge(PR[['promotion_id', 'sku', 'sim_true_uplift_pct']], on='promotion_id')
live['week_end'] = pd.to_datetime(live.check_date) + pd.to_timedelta((1 - pd.to_datetime(live.check_date).dt.weekday) % 7, unit='D')
upl = live[live.live_flag].groupby(['outlet_id', 'sku', 'week_end']).sim_true_uplift_pct.max()
season = x.groupby('week_end').qty_base_ea.sum().reindex(wk).fillna(0); season = (season.rolling(3, center=True, min_periods=1).mean() / season.mean()).values
out_rows = []
sim = si.unstack('week_end').reindex(columns=wk, fill_value=0).fillna(0)
for (o, s), row in sim.iterrows():
    v = row.values; base = v.mean() * 0.97
    u = np.array([upl.get((o, s, w), 0.0) for w in wk])
    dem = np.maximum(0, base * season * (1 + u) * r.lognormal(0, 0.2, len(wk)))
    stock = base * 3.0; price = prod.list_price_sar.get(s, 10) * 1.2
    for k, w in enumerate(wk):
        avail = stock + v[k]; sold = min(dem[k], avail); stock = avail - sold
        daily = dem[k] / 7 if dem[k] > 0 else 0
        oos = int(min(7, round((dem[k] - sold) / daily))) if daily > 0 and sold < dem[k] else 0
        out_rows.append((w.date(), o, s, round(sold), round(sold * price, 2), round(stock), round(stock / daily, 1) if daily else None, oos))
SO = pd.DataFrame(out_rows, columns=['week_end', 'outlet_id', 'sku', 'units_sold', 'sales_value_sar', 'on_hand_units', 'days_of_cover', 'oos_days'])
SO['account_id'] = SO.outlet_id.map(outl.set_index('outlet_id').account_id); SO['source'] = 'EPOS'
save(SO, 'fact_sell_out')
print('  sell-out OOS share of store-SKU-weeks', round((SO.oos_days > 0).mean(), 3), '| sell-through', round(SO.units_sold.sum() / si.sum(), 3))
SO['month'] = pd.to_datetime(SO.week_end).dt.to_period('M')
snap = SO.sort_values('week_end').groupby(['outlet_id', 'sku', 'month']).tail(1).copy()
life = snap.sku.map(prod.shelf_life_days).fillna(365)
age = (snap.days_of_cover.fillna(0) * r.uniform(0.6, 1.4, len(snap))) + 0.2 * life * r.random(len(snap))
snap['near_expiry_units'] = np.where(age > 0.7 * life, np.round(snap.on_hand_units * r.uniform(0.1, 0.6, len(snap))), 0)
snap['return_requested_units'] = np.where(snap.near_expiry_units > 0, np.round(snap.near_expiry_units * r.uniform(0.3, 1.0, len(snap))), 0)
snap['return_authorised_units'] = np.round(snap.return_requested_units * np.where(r.random(len(snap)) < 0.8, 1, 0.5))
CS = snap[['week_end', 'outlet_id', 'account_id', 'sku', 'on_hand_units', 'near_expiry_units', 'return_requested_units', 'return_authorised_units']]
save(CS, 'fact_customer_stock')

# ---------------- F10 trade spend (claims per promotion and account)
pv = NG.dropna(subset=['promotion_id']).groupby('promotion_id').agg(promo_units=('qty_base_ea', 'sum'), discount_sar=('discount_sar', 'sum'))
TS = PH.merge(PR.groupby('promotion_id').budget_sar.sum(), on='promotion_id').merge(pv, on='promotion_id', how='left').fillna({'discount_sar': 0})
TS['accrual_sar'] = (TS.discount_sar * TS.promotion_id.map(PR.groupby('promotion_id').discount_pct.first()) / (1 - P['channel_discount']['MT'] + 0.12) + np.where(TS.display_flag, 2500, 0)).round(2)
TS['claim_date'] = pd.to_datetime(TS.end_date) + pd.to_timedelta(r.integers(15, 61, len(TS)), unit='D')
TS = TS[TS.claim_date <= END].copy()
over = r.random(len(TS)) < 0.10
TS['claim_sar'] = (TS.accrual_sar * np.where(over, r.uniform(1.1, 1.4, len(TS)), r.uniform(0.97, 1.03, len(TS)))).round(2)
age = (END - TS.claim_date).dt.days
TS['status'] = np.select([over & (r.random(len(TS)) < 0.6), age < 20, over], ['disputed', 'pending validation', 'rejected'], 'approved')
TS['approved_sar'] = np.where(TS.status == 'approved', TS.claim_sar, 0)
TS['reason'] = np.where(TS.status.isin(['disputed', 'rejected']), r.choice(['claimed above agreed rate', 'volume not supported by sell-in', 'display not verified'], len(TS)), None)
TS.insert(0, 'claim_id', [f'CL{i+1:05d}' for i in range(len(TS))])
TS['claim_date'] = TS.claim_date.dt.date
save(TS.drop(columns=['display_flag']), 'fact_trade_spend')

# ---------------- F12 shelf audits (done visits × top-25 SKUs of the store's account)
done = MV[MV.status == 'done'][['visit_id', 'outlet_id', 'account_id', 'channel_id', 'planned_date', 'size_band']]
acc_top = NG.groupby(['account_id', 'sku']).net_amt.sum().groupby(level=0, group_keys=False).nlargest(25).reset_index()[['account_id', 'sku']]
SA = done.merge(acc_top, on='account_id')
SA['facings'] = np.maximum(1, np.round(SA.size_band.map({'XL': 6, 'L': 4, 'M': 3, 'S': 2, 'XS': 1}) * r.lognormal(0, 0.35, len(SA)))).astype(int)
SA['fair_share_pct'] = 0.18
SA['shelf_share_pct'] = np.clip(r.normal(0.17, 0.05, len(SA)), 0.03, 0.4).round(3)
SA['oos_flag'] = r.random(len(SA)) < np.where(SA.channel_id == 'DS', 0.09, 0.06)
SA['planogram_ok'] = r.random(len(SA)) < 0.86
lp = SA.sku.map(prod.list_price_sar)
SA['shelf_price_sar'] = (lp * np.where(SA.channel_id == 'DS', 1.10, 1.20) * r.normal(1, 0.03, len(SA))).round(2)
dsv = set(prod[prod.channel_variant_of.notna()].index)
SA['wrong_pack_flag'] = False
mt_wrong = SA.index[SA.channel_id == 'MT'].to_numpy(); mt_wrong = r.choice(mt_wrong, 180, replace=False)
SA.loc[mt_wrong, 'wrong_pack_flag'] = True                       # discounter-only pack found on an MT shelf
ds_wrong = SA.index[SA.channel_id == 'DS'].to_numpy(); ds_wrong = r.choice(ds_wrong, 120, replace=False)
SA.loc[ds_wrong, 'wrong_pack_flag'] = True                       # MT pack found where the discounter pack is required
SA['photo_id'] = 'IMG-' + SA.visit_id + '-' + (SA.groupby('visit_id').cumcount() + 1).astype(str)
SA['recognition_confidence'] = np.clip(r.normal(0.9, 0.06, len(SA)), 0.5, 0.99).round(2)
SA['anomaly_label'] = np.where(SA.wrong_pack_flag, 'channel_pack_breach', None)
save(SA.drop(columns=['size_band', 'planned_date']), 'fact_shelf_audit')

# ---------------- F06 price observations: wholesale invoice prices (weekly) + MT/DS shelf prices + competitor shelf prices
ws = NG[NG.channel_id.isin(['WS', 'MT', 'DS'])].copy()
ws['week'] = ws.doc_date.dt.to_period('W').dt.start_time
inv = ws.groupby(['channel_id', 'account_id', 'sku', 'week']).agg(price_sar=('unit_net_price_per_ea', 'median'),
                                                                    flag=('anomaly_label', lambda s: s.eq('price_below_floor').any())).reset_index()
inv = inv[inv.channel_id == 'WS']
inv['price_type'] = 'invoice price'; inv['source'] = 'invoice'; inv['outlet_id'] = inv.account_id.map(outl.groupby('account_id').outlet_id.first())
shelf = SA[['outlet_id', 'channel_id', 'sku', 'shelf_price_sar', 'visit_id']].merge(MV[['visit_id', 'planned_date']], on='visit_id')
shelf = shelf.sample(frac=0.35, random_state=SEED).rename(columns={'shelf_price_sar': 'price_sar', 'planned_date': 'week'})
shelf['price_type'] = 'shelf price'; shelf['source'] = 'merchandiser'; shelf['flag'] = False
ds_ix = shelf.index[shelf.channel_id == 'DS'].to_numpy(); ds_ix = r.choice(ds_ix, 60, replace=False)
shelf.loc[ds_ix, 'price_sar'] = (shelf.loc[ds_ix, 'sku'].map(prod.list_price_sar) * r.uniform(0.70, 0.80, 60)).round(2)
shelf.loc[ds_ix, 'flag'] = True
cmap = comp[['competitor_sku_id', 'mapped_our_sku', 'competitor_list_price_sar']]
csh = shelf[['outlet_id', 'channel_id', 'sku', 'week']].merge(cmap, left_on='sku', right_on='mapped_our_sku').sample(frac=0.5, random_state=SEED)
csh['price_sar'] = (csh.competitor_list_price_sar * np.where(csh.channel_id == 'DS', 1.10, 1.20) * r.normal(1, 0.04, len(csh))).round(2)
csh['price_type'] = 'competitor shelf price'; csh['source'] = 'merchandiser'; csh['flag'] = False
PO_ = pd.concat([inv.assign(competitor_sku_id=None), shelf.assign(competitor_sku_id=None), csh], ignore_index=True)
PO_['obs_date'] = pd.to_datetime(PO_.week).dt.date
PO_['promo_flag'] = False
PO_['anomaly_label'] = np.where(PO_.flag & (PO_.price_type == 'invoice price'), 'ws_below_floor', np.where(PO_.flag, 'ds_below_mt_price', None))
PO_ = PO_[['obs_date', 'outlet_id', 'channel_id', 'sku', 'competitor_sku_id', 'price_type', 'price_sar', 'promo_flag', 'source', 'anomaly_label']]
PO_.insert(0, 'observation_id', [f'OB{i+1:07d}' for i in range(len(PO_))])
save(PO_, 'fact_price_observation')

# ---------------- F07 competitor activity reported by the field
cats = comp.category.dropna().unique()
reps = emp[emp.role.isin(['Merchandiser', 'Van rep'])].employee_id.values
ev = []
for c in cats:
    for m in pd.period_range(START, END, freq='M'):
        for _ in range(r.poisson(2.2)):
            d = m.start_time + pd.Timedelta(days=int(r.integers(0, 28)))
            if d > END: continue
            cs_ = comp[comp.category == c].sample(1, random_state=int(r.integers(1e9))).iloc[0]
            t = r.choice(['price cut', 'promotion', 'display', 'launch'], p=[0.35, 0.4, 0.15, 0.1])
            ev.append((d.date(), r.choice(reps), r.choice(st.outlet_id.values), cs_.competitor_sku_id, c, t,
                       round(float(r.uniform(0.05, 0.30)), 2) if t in ('price cut', 'promotion') else None, bool(r.random() < 0.7)))
for _, cl in comp[comp.launch_date.notna()].iterrows():
    ev.append((cl.launch_date, r.choice(reps), r.choice(st.outlet_id.values), cl.competitor_sku_id, cl.category, 'launch', None, True))
CA = pd.DataFrame(ev, columns=['event_date', 'reported_by', 'outlet_id', 'competitor_sku_id', 'category', 'event_type', 'depth_pct', 'photo_flag'])
CA.insert(0, 'event_id', [f'CE{i+1:05d}' for i in range(len(CA))])
save(CA, 'fact_competitor_activity')