"""Dashboard prediction CSVs for the 30 operating tasks.
Actuals run to 20 Oct 2026 (end of data). Predictions cover 21 Oct – 30 Nov 2026.
One CSV per task (t01 … t30), small and dashboard-ready. Every file has a `basis` column
(Actual / Predicted) and `as_of` = 2026-10-20."""
import os, numpy as np, pandas as pd
from common import *
import warnings; warnings.filterwarnings('ignore')

PO_ = '/mnt/user-data/outputs/Binzagr_Data/11_Dashboard_Data'
os.makedirs(PO_, exist_ok=True)
AS_OF = pd.Timestamp('2026-10-20'); H_END = pd.Timestamp('2026-11-30')
FUT = pd.date_range(AS_OF + pd.Timedelta(days=1), H_END)
r = rng(11)
MAN = []

def out(df, name, task, desc, grain):
    df = df.copy(); df['as_of'] = AS_OF.date()
    for c in df.columns:
        if df[c].dtype.kind == 'f': df[c] = df[c].round(4 if df[c].abs().max() < 2 else 2)
    df.to_csv(f'{PO_}/{name}.csv', index=False)
    MAN.append((task, name + '.csv', grain, len(df), desc)); print(f'  {name}: {len(df):,} rows')

def wk(d): d = pd.to_datetime(d); return (d - pd.to_timedelta((d.dt.dayofweek + 1) % 7, 'D')).dt.normalize()   # week starts Sunday
LAST_FULL = pd.Timestamp('2026-10-11')        # last complete Sun–Sat week
FWEEKS = pd.date_range('2026-10-18', '2026-11-29', freq='7D')

def wk_fc(s, n=len(FWEEKS)):
    """s: weekly values for complete weeks. Level = mean of last 6 weeks, damped trend, 80% band."""
    s = s.astype(float).dropna()
    if len(s) == 0: return np.zeros(n), np.zeros(n), np.zeros(n)
    last = s.iloc[-6:]; lvl = last.mean()
    tr = np.clip(s.iloc[-3:].mean() / max(s.iloc[-6:-3].mean(), 1e-9), 0.9, 1.1) if len(s) >= 6 else 1.0
    p = np.array([lvl * tr ** (0.5 * (k + 1) / n) for k in range(n)])
    sd = s.iloc[-8:].std() if len(s) > 2 else lvl * 0.15
    return p, np.maximum(p - 1.28 * sd, 0), p + 1.28 * sd

# ---------- load
FI = load('fact_sell_in'); FI['doc_date'] = pd.to_datetime(FI.doc_date)
prod = load('dim_product_ext').set_index('sku'); acc = load('dim_account').set_index('account_id')
FI['division'] = FI.sku.map(prod.division)
T = load('fact_sales_target'); T['period_month'] = pd.to_datetime(T.period_month)
CH = ['GR', 'MT', 'WS', 'DS', 'EC']

# ================= Cluster performance
# T01 sell-in vs daily target (channel × day)
d = FI.groupby(['doc_date', 'channel_id']).net_amt.sum().unstack().reindex(pd.date_range(START, AS_OF)).fillna(0)
mt = T.groupby(['period_month', 'channel_id']).target_value_sar.sum().unstack()
mt.loc[pd.Timestamp('2026-10-01')] *= 31 / 20          # Oct plan in data covers 1–20 Oct only
mt.loc[pd.Timestamp('2026-11-01')] = mt.loc[pd.Timestamp('2026-10-01')] * 1.02          # Nov target = Oct + 2% (no Nov plan in data)
dow = d.iloc[-56:].groupby(d.index[-56:].dayofweek).mean()
dow_w = dow / dow.mean()
rows = []
allD = pd.date_range(START, H_END)
for ch in CH:
    recent = d[ch].iloc[-56:]; lvl = (recent / dow_w[ch].reindex(recent.index.dayofweek).values).mean()
    tr = np.clip(d[ch].iloc[-28:].sum() / d[ch].iloc[-56:-28].sum(), 0.92, 1.08)
    resid = (recent / (lvl * dow_w[ch].reindex(recent.index.dayofweek).values)).replace([np.inf], np.nan).std()
    for day in allD:
        m = day.to_period('M').to_timestamp(); mdays = pd.date_range(m, m + pd.offsets.MonthEnd(0))
        tgt = mt.loc[m, ch] * dow_w[ch][day.dayofweek] / dow_w[ch].reindex(mdays.dayofweek).sum() if m in mt.index else np.nan
        if day <= AS_OF:
            rows.append((day.date(), ch, d.loc[day, ch], tgt, None, None, None, 'Actual'))
        else:
            k = (day - AS_OF).days
            p = lvl * dow_w[ch][day.dayofweek] * tr ** (k / 41 * 0.5)
            if ch == 'EC' and pd.Timestamp('2026-11-19') <= day <= pd.Timestamp('2026-11-29'): p *= 1.35     # White Friday season on platforms
            rows.append((day.date(), ch, None, tgt, p, p * (1 - 1.28 * resid), p * (1 + 1.28 * resid), 'Predicted'))
T01 = pd.DataFrame(rows, columns=['date', 'channel_id', 'actual_sar', 'target_sar', 'predicted_sar', 'predicted_low_sar', 'predicted_high_sar', 'basis'])
T01['predicted_low_sar'] = T01.predicted_low_sar.clip(lower=0)
out(T01, 't01_sellin_vs_daily_target', 1, 'Daily sell-in vs phased target by channel; forecast to 30 Nov with 80% band', 'date × channel')

# T02 MTD run rate vs target (month × channel × division)
divshare = FI[FI.doc_date > AS_OF - pd.Timedelta(days=56)].groupby(['channel_id', 'division']).net_amt.sum()
divshare = divshare / divshare.groupby(level=0).transform('sum')
tdiv = T.groupby(['period_month', 'channel_id', 'division']).target_value_sar.sum()
tdiv[tdiv.index.get_level_values(0) == pd.Timestamp('2026-10-01')] *= 31 / 20
act = FI.groupby([FI.doc_date.dt.to_period('M').dt.to_timestamp(), 'channel_id', 'division']).net_amt.sum()
fut = T01[T01.basis == 'Predicted'].assign(m=lambda x: pd.to_datetime(x.date).dt.to_period('M').dt.to_timestamp()).groupby(['m', 'channel_id']).predicted_sar.sum()
rows = []
for (m, ch, dv), a in act.items():
    tg = tdiv.get((m, ch, dv), np.nan)
    if m == pd.Timestamp('2026-10-01'):
        pe = a + fut.get((m, ch), 0) * divshare.get((ch, dv), 0)
        rows.append((m.date(), ch, dv, tg, a, 20, 31, a / 20 * 31, pe, 'Actual MTD + predicted'))
    else:
        rows.append((m.date(), ch, dv, tg, a, None, None, None, a, 'Actual'))
for (ch, dv), sh in divshare.items():
    tg = tdiv.get((pd.Timestamp('2026-10-01'), ch, dv), np.nan) * 1.02
    rows.append(('2026-11-01', ch, dv, tg, 0.0, 0, 30, None, fut.get((pd.Timestamp('2026-11-01'), ch), 0) * sh, 'Predicted'))
T02 = pd.DataFrame(rows, columns=['month', 'channel_id', 'division', 'target_sar', 'actual_to_date_sar', 'days_elapsed', 'days_in_month',
                                  'straight_line_run_rate_sar', 'predicted_month_end_sar', 'basis'])
T02['predicted_attainment_pct'] = T02.predicted_month_end_sar / T02.target_sar
T02['gap_to_target_sar'] = T02.predicted_month_end_sar - T02.target_sar
out(T02, 't02_mtd_run_rate', 2, 'Month to date vs target with straight-line and model month-end projection', 'month × channel × division')

# T03 channel mix vs plan
plan = load('dim_channel').set_index('channel_id').planned_share_of_sales
mm = FI.groupby([FI.doc_date.dt.to_period('M').dt.to_timestamp(), 'channel_id']).net_amt.sum().unstack()
mm.loc[pd.Timestamp('2026-11-01')] = fut.xs(pd.Timestamp('2026-11-01'))
mm.loc[pd.Timestamp('2026-10-01')] += fut.xs(pd.Timestamp('2026-10-01'))
T03 = mm.stack().rename('sales_sar').reset_index().rename(columns={'level_0': 'month', 'doc_date': 'month'})
T03.columns = ['month', 'channel_id', 'sales_sar']
T03['share'] = T03.sales_sar / T03.groupby('month').sales_sar.transform('sum')
T03['plan_share'] = T03.channel_id.map(plan); T03['gap_pts'] = (T03.share - T03.plan_share) * 100
T03['basis'] = np.select([T03.month == pd.Timestamp('2026-11-01'), T03.month == pd.Timestamp('2026-10-01')], ['Predicted', 'Actual + predicted'], 'Actual')
T03['month'] = T03.month.dt.date
out(T03, 't03_channel_mix', 3, 'Channel share of sales vs plan by month; Oct completed and Nov predicted', 'month × channel')

# T04 sell-out (account × division × week)
SO = load('fact_sell_out'); SO['wk'] = wk(SO.week_end) - pd.Timedelta(days=7) if False else wk(SO.week_end)
SO['division'] = SO.sku.map(prod.division)
g = SO.groupby(['account_id', 'division', 'wk']).agg(units=('units_sold', 'sum'), value=('sales_value_sar', 'sum'), oos=('oos_days', 'mean')).reset_index()
rows = []
for (a, dv), x in g.groupby(['account_id', 'division']):
    x = x.set_index('wk').sort_index(); full = x[x.index <= LAST_FULL]
    for w, rr in full.iterrows(): rows.append((w.date(), a, dv, rr.units, rr.value, rr.oos / 7, None, None, None, 'Actual'))
    p, lo, hi = wk_fc(full.value)
    pu, _, _ = wk_fc(full.units)
    for k, w in enumerate(FWEEKS): rows.append((w.date(), a, dv, None, None, full.oos.iloc[-6:].mean() / 7, pu[k], p[k], (lo[k], hi[k]), 'Predicted'))
T04 = pd.DataFrame(rows, columns=['week_start', 'account_id', 'division', 'units_sold', 'sales_value_sar', 'oos_day_share', 'predicted_units', 'predicted_value_sar', 'band', 'basis'])
T04[['predicted_low_sar', 'predicted_high_sar']] = pd.DataFrame(T04.band.apply(lambda b: b if isinstance(b, tuple) else (None, None)).tolist(), index=T04.index)
T04 = T04.drop(columns='band')
out(T04, 't04_sell_out', 4, 'EPOS sell-out from key accounts 01–03 with 6-week forecast', 'week × account × division')

# ================= Modern trade
# T05 POs received vs expected, rejected, delayed
PO = load('fact_purchase_order'); PO['wk'] = wk(PO.po_date)
g = PO[PO.channel_id == 'MT'].groupby(['account_id', 'wk']).agg(po_count=('po_id', 'nunique'), lines=('po_line', 'size'), expected_cases=('expected_cases', 'sum'),
        ordered_cases=('ordered_cases', 'sum'), rejected_lines=('status', lambda s: (s == 'rejected').sum()), delayed_lines=('delay_days', lambda s: (s > 0).sum())).reset_index()
exp_po = g.groupby('account_id').po_count.apply(lambda s: s.iloc[-9:-1].median())
rows = []
for a, x in g.groupby('account_id'):
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    for w, rr in full.iterrows():
        rows.append((w.date(), a, exp_po[a], rr.po_count, rr.lines, rr.expected_cases, rr.ordered_cases, rr.rejected_lines / rr.lines, rr.delayed_lines / rr.lines, 'Actual'))
    pc, _, _ = wk_fc(full.po_count); pcs, _, _ = wk_fc(full.ordered_cases)
    rj = full.rejected_lines.iloc[-8:].sum() / full.lines.iloc[-8:].sum(); dl = full.delayed_lines.iloc[-8:].sum() / full.lines.iloc[-8:].sum()
    for k, w in enumerate(FWEEKS): rows.append((w.date(), a, exp_po[a], round(pc[k]), None, None, pcs[k], rj, dl, 'Predicted'))
T05 = pd.DataFrame(rows, columns=['week_start', 'account_id', 'expected_po_count', 'po_count', 'po_lines', 'expected_cases', 'ordered_cases', 'rejected_line_share', 'delayed_line_share', 'basis'])
T05['po_vs_expected'] = T05.po_count - T05.expected_po_count
T05['account_name'] = T05.account_id.map(acc.account_name)
out(T05, 't05_po_tracker', 5, 'MT purchase orders received vs expected, rejected and delayed lines; next 6 weeks predicted', 'week × account')
rr = PO[(PO.channel_id == 'MT') & (PO.status == 'rejected')].groupby(['account_id', 'rejection_reason']).size().rename('rejected_lines').reset_index()
rr['basis'] = 'Actual'; out(rr, 't05b_po_rejection_reasons', 5, 'Rejected MT PO lines by reason', 'account × reason')

# T06 merchandiser photo reports (store level, last 28 days + next-visit risk)
MV = load('fact_merch_visit'); MV['planned_date'] = pd.to_datetime(MV.planned_date)
SA = load('fact_shelf_audit'); vd = MV.set_index('visit_id').planned_date; SA['date'] = SA.visit_id.map(vd)
cut = AS_OF - pd.Timedelta(days=28)
m4 = MV[MV.planned_date > cut]; s4 = SA[SA.date > cut]
st = m4.groupby(['outlet_id', 'account_id', 'channel_id', 'merchandiser_id']).agg(planned_visits=('visit_id', 'size'), done_visits=('status', lambda s: (s == 'done').sum()),
        photos=('photo_count', 'sum'), gps_flags=('anomaly_label', lambda s: s.notna().sum()), last_visit=('planned_date', 'max')).reset_index()
au = s4.groupby('outlet_id').agg(skus_audited=('sku', 'size'), oos_rate=('oos_flag', 'mean'), planogram_ok_rate=('planogram_ok', 'mean'),
        share_of_shelf=('shelf_share_pct', 'mean'), fair_share=('fair_share_pct', 'mean'), wrong_pack=('wrong_pack_flag', 'sum'))
st = st.join(au, on='outlet_id')
acct_oos = st.groupby('account_id').oos_rate.transform('mean')
st['visit_compliance'] = st.done_visits / st.planned_visits
st['predicted_oos_next_visit'] = 0.6 * st.oos_rate.fillna(acct_oos) + 0.4 * acct_oos + 0.02 * (1 - st.visit_compliance)
st['risk'] = pd.cut(st.predicted_oos_next_visit, [-1, 0.07, 0.1, 1], labels=['low', 'medium', 'high']).astype(str)
st['last_visit'] = st.last_visit.dt.date; st['basis'] = 'Actual 28 days + predicted'
out(st, 't06_merch_photo_reports', 6, 'Store photo-audit results (last 28 days) and predicted out-of-stock at next visit', 'store')

# T07 promotions live
PR = load('fact_promotion'); PR['start_date'] = pd.to_datetime(PR.start_date); PR['end_date'] = pd.to_datetime(PR.end_date)
PE = load('fact_promo_execution')
nxt = PR[PR.start_date >= pd.Timestamp('2026-09-22')].copy()            # Nov calendar: repeat the last 4 weeks' promotions 5 weeks later
nxt['promotion_id'] = 'PRN' + nxt.promotion_id.str[2:]; nxt['start_date'] += pd.Timedelta(days=35); nxt['end_date'] += pd.Timedelta(days=35)
nxt = nxt[nxt.start_date > AS_OF]
PR = pd.concat([PR, nxt], ignore_index=True)
ph = PR.groupby('promotion_id').agg(account_id=('account_id', 'first'), channel_id=('channel_id', 'first'), start_date=('start_date', 'first'), end_date=('end_date', 'first'),
        mechanic=('mechanic', 'first'), discount_pct=('discount_pct', 'first'), skus=('sku', 'nunique'), planned_uplift_pct=('planned_uplift_pct', 'first'), budget_sar=('budget_sar', 'sum'))
ex = PE.groupby('promotion_id').agg(stores_checked=('outlet_id', 'nunique'), checks=('check_id', 'size'), live_rate=('live_flag', 'mean'),
        price_correct_rate=('price_correct', 'mean'), display_built_rate=('display_built', 'mean'), pos_rate=('pos_material', 'mean'))
ph = ph.join(ex)
acct_live = ph.groupby('account_id').live_rate.mean()
ph['status'] = np.select([ph.start_date > AS_OF, ph.end_date < AS_OF], ['upcoming', 'ended'], 'running')
ph['predicted_live_rate'] = np.where(ph.status != 'ended', ph.account_id.map(acct_live), np.nan)
ph = ph[ph.end_date >= pd.Timestamp('2026-08-01')].reset_index()
ph['basis'] = np.where(ph.status == 'ended', 'Actual', np.where(ph.status == 'running', 'Actual + predicted', 'Predicted'))
ph[['start_date', 'end_date']] = ph[['start_date', 'end_date']].apply(lambda c: c.dt.date)
out(ph, 't07_promotions_live', 7, 'Promotions Aug–Nov: in-store execution (live, price, display) and predicted live rate for running/upcoming', 'promotion')

# ================= Wholesale
# T08 wholesale price levels (top 40 WS SKUs × week)
ws = FI[(FI.channel_id == 'WS') & ~FI.is_foc].copy(); ws['wk'] = wk(ws.doc_date)
top = ws.groupby('sku').net_amt.sum().nlargest(40).index
ws = ws[ws.sku.isin(top)]
g = ws.groupby(['sku', 'wk']).apply(lambda x: pd.Series({'avg_price_ea': x.net_amt.sum() / x.qty_base_ea.sum(), 'min_price_ea': x.unit_net_price_per_ea.min(),
        'buyers': x.account_id.nunique(), 'below_floor_lines': x.anomaly_label.eq('price_below_floor').sum()})).reset_index()
rows = []
for s, x in g.groupby('sku'):
    lp = prod.list_price_sar[s]; floor = lp * (1 - P['channel_discount']['WS'] - 0.05)
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    for w, rr in full.iterrows(): rows.append((w.date(), s, prod.sku_name.get(s), prod.division[s], lp, floor, rr.avg_price_ea, rr.min_price_ea, rr.buyers, rr.below_floor_lines, None, 'Actual'))
    ew = full.avg_price_ea.ewm(halflife=3).mean().iloc[-1]
    drift = np.clip((full.avg_price_ea.iloc[-4:].mean() / full.avg_price_ea.iloc[-12:-4].mean()) - 1, -0.03, 0.03) if len(full) > 12 else 0
    for k, w in enumerate(FWEEKS): rows.append((w.date(), s, prod.sku_name.get(s), prod.division[s], lp, floor, None, None, None, None, ew * (1 + drift * (k + 1) / 6), 'Predicted'))
T08 = pd.DataFrame(rows, columns=['week_start', 'sku', 'sku_name', 'division', 'list_price_ea', 'floor_price_ea', 'avg_price_ea', 'min_price_ea', 'buyers', 'below_floor_lines', 'predicted_price_ea', 'basis'])
T08['price_index_vs_list'] = T08.avg_price_ea.fillna(T08.predicted_price_ea) / T08.list_price_ea
out(T08, 't08_wholesale_price_levels', 8, 'Wholesale net price per unit vs list and floor for top 40 SKUs; 6-week price outlook', 'week × SKU')

# T09 abnormal / irregular orders (line rules rolled up to the order document)
docs = FI[FI.channel_id.isin(['WS', 'DS', 'MT']) & ~FI.is_foc].copy()
docs['pidx'] = np.where(docs.promotion_id.isna(), docs.unit_net_price_per_ea / docs.list_price_sar, np.nan)
docs['qty_ratio'] = docs.qty_base_ea / docs.groupby(['account_id', 'sku']).qty_base_ea.transform('median')
D9 = docs.groupby(['doc_no', 'account_id', 'channel_id']).agg(doc_date=('doc_date', 'first'), value_sar=('net_amt', 'sum'), lines=('sku', 'size'),
        min_price_index=('pidx', 'min'), max_qty_ratio=('qty_ratio', 'max'), ship_to_city=('ship_to_city', lambda c: next((v for v in c.dropna() if v not in ('Riyadh', 'Jeddah', 'Dammam')), c.dropna().iloc[0] if c.notna().any() else None)),
        true_label=('anomaly_label', lambda s: s.dropna().iloc[0] if s.notna().any() else None)).reset_index()
D9['city_flag'] = D9.ship_to_city.notna() & ~D9.ship_to_city.isin(['Riyadh', 'Jeddah', 'Dammam'])
D9['price_flag'] = D9.min_price_index < 0.73
D9['size_flag'] = (D9.channel_id == 'WS') & (D9.max_qty_ratio >= 12)
D9['anomaly_score'] = (0.6 * D9.city_flag + 0.6 * D9.price_flag + np.clip(D9.max_qty_ratio / 25, 0, 0.6)).clip(upper=1)
D9['predicted_flag'] = D9.city_flag | D9.price_flag | D9.size_flag
D9['reason'] = np.select([D9.city_flag, D9.price_flag, D9.size_flag], ['ships outside normal regions (possible parallel trade)', 'non-promo price below floor', 'line quantity 12x+ the usual for this SKU'], '')
T09 = D9[D9.predicted_flag | D9.true_label.notna()].sort_values('anomaly_score', ascending=False)
T09['basis'] = 'Predicted (scored)'; T09['doc_date'] = T09.doc_date.dt.date
prec = T09[T09.predicted_flag].true_label.notna().mean(); rec = T09[T09.true_label.notna()].predicted_flag.mean()
print(f'    irregular orders: precision {prec:.2f} recall {rec:.2f}')
out(T09, 't09_irregular_orders', 9, f'Flagged MT/WS/DS orders with reason (precision {prec:.0%}, recall {rec:.0%} vs planted labels)', 'order document')

# ================= Discounters
PX = load('fact_price_observation'); PX['wk'] = wk(PX.obs_date)
sp = PX[PX.price_type == 'shelf price'].copy()
sp['sku'] = sp.sku.map(lambda k: prod.channel_variant_of.get(k) if str(k).startswith('DSV') else k)   # discounter pack → base SKU, price per unit
mtp = sp[sp.channel_id == 'MT'].groupby(['sku', 'wk']).price_sar.median()
dso = sp[sp.channel_id == 'DS'].copy(); dso['mt'] = dso.set_index(['sku', 'wk']).index.map(mtp); dso = dso.dropna(subset=['mt'])
dso['undercut'] = dso.price_sar / dso.mt < 0.75
g = dso.groupby(['sku', 'wk']).agg(DS=('price_sar', 'median'), MT=('mt', 'first'), stores=('outlet_id', 'nunique'), flagged=('undercut', 'sum')).reset_index()
g['gap_pct'] = g.DS / g.MT - 1
g = g[g.sku.isin(dso.groupby('sku').size().nlargest(80).index)]
rows = []
for s, x in g.groupby('sku'):
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    if len(full) < 4: continue
    for w, rr in full.iterrows(): rows.append((w.date(), s, rr.DS, rr.MT, rr.gap_pct, rr.stores, rr.flagged, None, 'Actual'))
    pg = full.gap_pct.ewm(halflife=3).mean().iloc[-1]
    for w in FWEEKS[:4]: rows.append((w.date(), s, None, None, None, None, None, pg, 'Predicted'))
T10 = pd.DataFrame(rows, columns=['week_start', 'sku', 'ds_shelf_price_per_unit', 'mt_shelf_price_per_unit', 'gap_pct', 'ds_stores_seen', 'stores_undercutting', 'predicted_gap_pct', 'basis'])
T10['division'] = T10.sku.map(prod.division); T10['sku_name'] = T10.sku.map(prod.sku_name)
T10['agreed_gap_pct'] = -0.15
T10['alert'] = (T10.stores_undercutting.fillna(0) > 0) | (T10.gap_pct.fillna(T10.predicted_gap_pct) < -0.20)
out(T10, 't10_ds_vs_mt_price', 10, 'Discounter vs MT shelf price per unit (top 80 SKUs); stores more than 25% below MT flagged; 4-week outlook', 'week × SKU')

# T11 pack and SKU compliance
dsa = SA[SA.channel_id == 'DS']
T11 = dsa.groupby(['account_id', 'outlet_id']).agg(audit_lines=('sku', 'size'), wrong_pack_lines=('wrong_pack_flag', 'sum'), breaches=('anomaly_label', lambda s: s.notna().sum()),
        planogram_ok_rate=('planogram_ok', 'mean'), oos_rate=('oos_flag', 'mean'), last_audit=('date', 'max')).reset_index()
T11['wrong_pack_rate'] = T11.wrong_pack_lines / T11.audit_lines
acct_wp = T11.groupby('account_id').wrong_pack_rate.transform('mean')
T11['predicted_breach_prob_next_audit'] = 1 - (1 - (T11.wrong_pack_lines + acct_wp * 20) / (T11.audit_lines + 20)) ** 25    # 25 SKUs per audit
listed = PX[(PX.channel_id == 'DS') & (PX.price_type == 'shelf price')].groupby('outlet_id').sku.nunique()
T11['skus_seen_on_shelf'] = T11.outlet_id.map(listed)
T11['last_audit'] = T11.last_audit.dt.date; T11['basis'] = 'Actual + predicted'
out(T11, 't11_pack_compliance', 11, 'Discounter stores: wrong-pack / channel-pack breaches, planogram and predicted breach chance at next audit', 'store')

# ================= Grocery
RV = load('fact_route_visit'); RV['visit_date'] = pd.to_datetime(RV.visit_date); RV['wk'] = wk(RV.visit_date)
g = RV.groupby(['van_id', 'wk']).agg(planned_calls=('visit_id', 'size'), visited=('outcome', lambda s: (s != 'not visited').sum()),
        productive=('outcome', lambda s: (s == 'sale').sum()), value_sar=('drop_value_sar', 'sum'), days=('visit_date', 'nunique')).reset_index()
vdiv = load('dim_vehicle').set_index('van_id').division
rows = []
for v, x in g.groupby('van_id'):
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    if full.empty: continue
    full = full.reindex(pd.date_range(full.index.min(), LAST_FULL, freq='7D'), fill_value=0)
    for w, rr in full[full.days > 0].iterrows():
        rows.append((w.date(), v, vdiv.get(v), rr.days, rr.planned_calls, rr.visited, rr.productive, rr.value_sar, None, None, 'Actual'))
    if full.days.iloc[-3:].sum() == 0: continue     # van inactive for 3 weeks: no forecast
    pv, _, _ = wk_fc(full.visited); pp, _, _ = wk_fc(full.productive); pval, lo, hi = wk_fc(full.value_sar)
    for k, w in enumerate(FWEEKS): rows.append((w.date(), v, vdiv.get(v), None, None, round(pv[k]), round(pp[k]), pval[k], lo[k], hi[k], 'Predicted'))
T12 = pd.DataFrame(rows, columns=['week_start', 'van_id', 'division', 'working_days', 'planned_calls', 'calls_made', 'productive_calls', 'sales_sar', 'predicted_low_sar', 'predicted_high_sar', 'basis'])
T12['strike_rate'] = T12.productive_calls / T12.calls_made
T12['drop_size_sar'] = T12.sales_sar / T12.productive_calls
T12['calls_per_day'] = T12.calls_made / T12.working_days
T12['strike_target'] = 0.80
out(T12, 't12_salesman_kpis', 12, 'Calls, strike rate, drop size per van (codes only, no names) with 6-week forecast', 'week × van')

# T13 new and lost outlets, must-sell coverage
gr = FI[FI.channel_id == 'GR'].copy(); gr['m'] = gr.doc_date.dt.to_period('M').dt.to_timestamp()
first = gr.groupby('outlet_id').doc_date.min(); lastd = gr.groupby('outlet_id').doc_date.max()
main_van = gr.groupby(['outlet_id', 'slsman_cd']).net_amt.sum().reset_index().sort_values('net_amt').drop_duplicates('outlet_id', keep='last').set_index('outlet_id').slsman_cd.astype(str)
must = set(prod.index[prod.must_sell_GR == True])
rows = []
months = sorted(gr.m.unique())
for m in months:
    me = min(m + pd.offsets.MonthEnd(0), AS_OF)
    x = gr[gr.m == m]
    act_o = x.groupby('outlet_id').size().index
    lost = lastd.index[(lastd >= m - pd.Timedelta(days=60)) & (lastd < me - pd.Timedelta(days=60))]   # went 60 days silent during this month
    new = first[(first >= m) & (first <= me) & (first >= START + pd.Timedelta(days=60))].index
    ms = x[x.sku.isin(must)].groupby('outlet_id').sku.nunique()
    for v in main_van.unique():
        ov = main_van.index[main_van == v]
        a_ = act_o.intersection(ov)
        rows.append((pd.Timestamp(m).date(), v, len(a_), len(new.intersection(ov)), len(lost.intersection(ov)),
                     (ms.reindex(a_).fillna(0) >= 3).mean() if len(a_) else np.nan, 'Actual' if me < AS_OF else 'Actual (to 20 Oct)'))
T13 = pd.DataFrame(rows, columns=['month', 'van_id', 'active_outlets', 'new_outlets', 'lost_outlets', 'must_sell_coverage', 'basis'])
# predicted lost in Nov: outlets whose gap since last order is > 1.5× their usual interval
iv = gr.groupby('outlet_id').doc_date.apply(lambda s: s.drop_duplicates().sort_values().diff().dt.days.median())
since = (AS_OF - lastd).dt.days
risk = (since > 1.5 * iv.fillna(30)) & (since <= 60)
pr_ = pd.DataFrame({'van_id': main_van, 'risk': risk}).groupby('van_id').risk.sum()
nov = T13[T13.month == pd.Timestamp('2026-10-01').date()].copy()
cand = lastd.index[(lastd >= pd.Timestamp('2026-09-02')) & (lastd < pd.Timestamp('2026-10-01'))]     # will hit 60 days silent in November
hist_conv = 0.2                                                                                   # measured: 15–32% of outlets silent 20–48 days never came back within 41 days (Jun–Aug cut-offs)
nov['month'] = pd.Timestamp('2026-11-01').date(); nov['lost_outlets'] = nov.van_id.map(pd.Series(main_van.reindex(cand).values).value_counts()).fillna(0) * hist_conv
nov['new_outlets'] = T13.groupby('van_id').new_outlets.apply(lambda s: s.iloc[-4:].mean()).reindex(nov.van_id).values
sep = T13[T13.month == pd.Timestamp('2026-09-01').date()].set_index('van_id').active_outlets
nov['active_outlets'] = (nov.van_id.map(sep).fillna(nov.active_outlets) - nov.lost_outlets + nov.new_outlets).clip(lower=0); nov['basis'] = 'Predicted'
T13 = pd.concat([T13, nov]); T13['must_sell_target'] = 0.6
out(T13, 't13_outlet_movement', 13, 'Active, new and lost grocery outlets and must-sell coverage (≥3 must-sell SKUs) per van; Nov predicted', 'month × van')
atr = pd.DataFrame({'outlet_id': risk[risk].index}); atr['van_id'] = atr.outlet_id.map(main_van); atr['days_since_last_order'] = atr.outlet_id.map(since)
atr['usual_interval_days'] = atr.outlet_id.map(iv); atr['last_order_date'] = atr.outlet_id.map(lastd).dt.date
atr['predicted_loss_prob'] = np.clip(0.3 + 0.25 * (atr.days_since_last_order / atr.usual_interval_days.clip(lower=3) - 1.5), 0.3, 0.9)
atr['basis'] = 'Predicted'
out(atr.sort_values('predicted_loss_prob', ascending=False), 't13b_outlets_at_risk', 13, 'Grocery outlets overdue for an order: likely to be lost in November', 'outlet')

# T14 van stock, cash and credit reconciliation (van × week)
VS = load('fact_van_stock'); VS['wk'] = wk(VS.stock_date); VC = load('fact_van_cash'); VC['wk'] = wk(VC.cash_date)
VS['val'] = VS.sku.map(prod.list_price_sar).fillna(5)
s = VS.groupby(['van_id', 'wk']).agg(loaded_units=('loaded_qty', 'sum'), sold_units=('sold_qty', 'sum'), returned_units=('returned_qty', 'sum'), damaged_units=('damaged_qty', 'sum'),
        stock_variance_units=('stock_variance_qty', 'sum'), stock_loss_events=('anomaly_label', lambda s: s.notna().sum()))
VS['var_val'] = VS.stock_variance_qty * VS.val
s['stock_variance_sar'] = VS.groupby(['van_id', 'wk']).var_val.sum()
c = VC.groupby(['van_id', 'wk']).agg(cash_sales_sar=('cash_sales_sar', 'sum'), card_sales_sar=('card_sales_sar', 'sum'), credit_sales_sar=('credit_sales_sar', 'sum'),
        cash_deposited_sar=('cash_deposited_sar', 'sum'), cash_variance_sar=('cash_variance_sar', 'sum'), cash_short_days=('anomaly_label', lambda s: s.notna().sum()))
T14 = s.join(c, how='outer').reset_index(); T14 = T14[T14.wk > AS_OF - pd.Timedelta(days=84)]
T14['basis'] = 'Actual'
vr = T14.groupby('van_id').agg(ev=('stock_loss_events', 'sum'), cs=('cash_short_days', 'sum'), n=('wk', 'size'))
pv = pd.DataFrame({'van_id': vr.index, 'wk': FWEEKS[1], 'basis': 'Predicted',
                   'predicted_issue_prob_next_week': 1 - np.exp(-((vr.ev + vr.cs + 0.5) / (vr.n + 2)).values)})
T14 = pd.concat([T14, pv]).rename(columns={'wk': 'week_start'}); T14['week_start'] = pd.to_datetime(T14.week_start).dt.date
T14['division'] = T14.van_id.map(vdiv)
out(T14, 't14_van_reconciliation', 14, 'Van stock, cash and credit reconciliation (last 12 weeks) and predicted chance of a variance next week', 'week × van')

# ================= E-commerce
EC = load('fact_ecom_listing_daily'); EC['date'] = pd.to_datetime(EC.date); EC['wk'] = wk(EC.date)
g = EC.groupby(['platform_id', 'wk']).agg(orders=('orders', 'sum'), units=('units', 'sum'), gmv_sar=('gmv_sar', 'sum'),
        listing_days=('listing_id', 'size'), oos_days=('listing_status', lambda s: (s == 'out of stock').sum()), inactive_days=('listing_status', lambda s: (s == 'inactive').sum()),
        ad_spend_sar=('ad_spend_sar', 'sum'), ad_sales_sar=('ad_sales_sar', 'sum')).reset_index()
rows = []
for p_, x in g.groupby('platform_id'):
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    for w, rr in full.iterrows(): rows.append((w.date(), p_, rr.orders, rr.units, rr.gmv_sar, rr.oos_days / rr.listing_days, rr.inactive_days / rr.listing_days, rr.ad_spend_sar, rr.ad_sales_sar, None, None, None, 'Actual'))
    po_, lo, hi = wk_fc(full.orders); pg, _, _ = wk_fc(full.gmv_sar)
    for k, w in enumerate(FWEEKS):
        wf = 1.35 if w in (pd.Timestamp('2026-11-15'), pd.Timestamp('2026-11-22')) else 1.0     # White Friday weeks
        rows.append((w.date(), p_, None, None, None, full.oos_days.iloc[-4:].sum() / full.listing_days.iloc[-4:].sum(), full.inactive_days.iloc[-4:].sum() / full.listing_days.iloc[-4:].sum(),
                     None, None, po_[k] * wf, pg[k] * wf, (lo[k] * wf, hi[k] * wf), 'Predicted'))
T15 = pd.DataFrame(rows, columns=['week_start', 'platform_id', 'orders', 'units', 'gmv_sar', 'oos_listing_share', 'inactive_listing_share', 'ad_spend_sar', 'ad_sales_sar',
                                  'predicted_orders', 'predicted_gmv_sar', 'band', 'basis'])
T15[['predicted_orders_low', 'predicted_orders_high']] = pd.DataFrame(T15.band.apply(lambda b: b if isinstance(b, tuple) else (None, None)).tolist(), index=T15.index)
T15 = T15.drop(columns='band'); T15['platform_name'] = T15.platform_id.map(acc.account_name)
out(T15, 't15_ecom_orders', 15, 'Orders and GMV by platform with out-of-stock and inactive listing share; White Friday uplift in Nov forecast', 'week × platform')
last = EC[EC.date == AS_OF]
lst = last[last.listing_status != 'active'][['platform_id', 'listing_id', 'sku', 'listing_status', 'stock_units']]
streak = EC[EC.date > AS_OF - pd.Timedelta(days=30)].assign(bad=lambda x: x.listing_status != 'active').groupby('listing_id').bad.sum()
lst['bad_days_last_30'] = lst.listing_id.map(streak); lst['sku_name'] = lst.sku.map(prod.sku_name)
lst['est_lost_gmv_per_day_sar'] = lst.listing_id.map(EC[EC.listing_status == 'active'].groupby('listing_id').gmv_sar.mean())
lst['basis'] = 'Actual'
out(lst.sort_values('est_lost_gmv_per_day_sar', ascending=False), 't15b_ecom_listings_down', 15, 'Listings out of stock or inactive on 20 Oct with estimated GMV lost per day', 'listing')

# T16 search rank, ratings, ad return (listing)
l28 = EC[EC.date > AS_OF - pd.Timedelta(days=28)]
T16 = l28.groupby(['platform_id', 'listing_id', 'sku']).agg(avg_rank_28d=('search_rank', 'mean'), rating=('rating', 'last'), content_score=('content_score', 'last'),
        gmv_28d_sar=('gmv_sar', 'sum'), ad_spend_28d_sar=('ad_spend_sar', 'sum'), ad_sales_28d_sar=('ad_sales_sar', 'sum')).reset_index()
r7 = EC[EC.date > AS_OF - pd.Timedelta(days=7)].groupby('listing_id').search_rank.mean(); r28 = EC[(EC.date > AS_OF - pd.Timedelta(days=28)) & (EC.date <= AS_OF - pd.Timedelta(days=21))].groupby('listing_id').search_rank.mean()
T16['rank_last_7d'] = T16.listing_id.map(r7); T16['rank_trend_3w'] = T16.listing_id.map(r7) - T16.listing_id.map(r28)
T16['predicted_rank_next_week'] = (T16.rank_last_7d + T16.rank_trend_3w / 3).clip(1, 400)
T16['roas'] = T16.ad_sales_28d_sar / T16.ad_spend_28d_sar.replace(0, np.nan)
T16['sku_name'] = T16.sku.map(prod.sku_name); T16['basis'] = 'Actual 28 days + predicted'
T16['action'] = np.select([T16.rating < 3.8, T16.content_score < 60, T16.roas < 3, T16.rank_trend_3w > 15], ['fix reviews / quality', 'improve content', 'cut or retarget ads', 'rank falling: check stock and price'], '')
out(T16, 't16_ecom_rank_rating_roas', 16, 'Per listing: search rank and its trend, rating, content score and ad return (ROAS); next-week rank predicted', 'listing')

# ================= Pricing and promotions
# T17 promotion uplift vs plan (promotion × account × sku → header)
pi = FI[FI.promotion_id.notna()]
base = FI[FI.promotion_id.isna() & FI.channel_id.isin(['MT', 'DS', 'WS', 'EC'])].groupby(['account_id', 'sku', 'doc_date']).qty_base_ea.sum()
up = []
for p_, x in PR.groupby('promotion_id'):
    a = x.account_id.iat[0]; s0, s1 = x.start_date.iat[0], x.end_date.iat[0]
    if s0 > AS_OF: continue
    days = (min(s1, AS_OF) - s0).days + 1
    pu = pi[(pi.promotion_id == p_)].qty_base_ea.sum() / days
    try:
        b = base.loc[a]; b = b[b.index.get_level_values(0).isin(x.sku)]
        bd = b[(b.index.get_level_values(1) < s0) & (b.index.get_level_values(1) >= s0 - pd.Timedelta(days=28))].sum() / 28
    except KeyError: bd = np.nan
    up.append((p_, pu / bd - 1 if bd and bd > 0 else np.nan))
mu = pd.Series(dict(up))
T17 = ph.set_index('promotion_id')[['account_id', 'channel_id', 'start_date', 'end_date', 'mechanic', 'discount_pct', 'skus', 'planned_uplift_pct', 'budget_sar', 'status']].copy()
T17['measured_uplift_pct'] = mu.reindex(T17.index).clip(-0.3, 2.0)
lift_by = T17.groupby(['account_id', 'mechanic']).measured_uplift_pct.median(); lift_m = T17.groupby('mechanic').measured_uplift_pct.median()
T17['predicted_uplift_pct'] = [lift_by.get((a, m_), lift_m.get(m_)) if s_ != 'ended' else np.nan for a, m_, s_ in zip(T17.account_id, T17.mechanic, T17.status)]
T17['uplift_vs_plan_pts'] = (T17.measured_uplift_pct.fillna(T17.predicted_uplift_pct) - T17.planned_uplift_pct) * 100
T17['basis'] = np.where(T17.status == 'upcoming', 'Predicted', 'Actual (measured vs 4-week baseline)')
out(T17.reset_index(), 't17_promo_uplift', 17, 'Promotion uplift measured against a 4-week pre-promo baseline vs plan; predicted for running/upcoming promos', 'promotion')

# T18 trade spend vs budget and claims (account × month)
TS = load('fact_trade_spend'); TS['m'] = pd.to_datetime(TS.start_date).dt.to_period('M').dt.to_timestamp()
g = TS.groupby(['account_id', 'm']).agg(budget_sar=('budget_sar', 'sum'), accrual_sar=('accrual_sar', 'sum'), claimed_sar=('claim_sar', 'sum'), approved_sar=('approved_sar', 'sum'),
        disputed_claims=('status', lambda s: s.isin(['disputed', 'rejected']).sum()), claims=('claim_id', 'size')).reset_index()
g['basis'] = 'Actual'
nb = PR[PR.start_date >= pd.Timestamp('2026-11-01')].groupby('account_id').budget_sar.sum()
ratio = g.groupby('account_id').apply(lambda x: x.accrual_sar.sum() / x.budget_sar.sum())
avg3 = g[g.m >= pd.Timestamp('2026-08-01')].groupby('account_id').accrual_sar.mean()
nv = pd.DataFrame({'account_id': ratio.index, 'm': pd.Timestamp('2026-11-01'), 'basis': 'Predicted'})
nv['budget_sar'] = nv.account_id.map(nb).fillna(nv.account_id.map(g.groupby('account_id').budget_sar.mean()))
nv['accrual_sar'] = (0.5 * nv.budget_sar * nv.account_id.map(ratio) + 0.5 * nv.account_id.map(avg3)).values
T18 = pd.concat([g, nv]); T18['spend_vs_budget'] = T18.accrual_sar / T18.budget_sar
T18['month'] = T18.m.dt.date; T18 = T18.drop(columns='m'); T18['account_name'] = T18.account_id.map(acc.account_name)
out(T18, 't18_trade_spend', 18, 'Trade spend budget, accruals, claims and disputes by account and month; Nov spend predicted', 'month × account')

# T19 competitor launches and price cuts
CA = load('fact_competitor_activity'); cp = load('dim_competitor_product').set_index('competitor_sku_id')
CA['brand'] = CA.competitor_sku_id.map(cp.competitor_brand); CA['m'] = pd.to_datetime(CA.event_date).dt.to_period('M').dt.to_timestamp()
g = CA.groupby(['brand', 'category', 'event_type']).agg(events=('event_id', 'size'), avg_depth=('depth_pct', 'mean'), last_event=('event_date', 'max')).reset_index()
rate = CA[CA.m >= pd.Timestamp('2026-07-01')].groupby(['brand', 'category', 'event_type']).size() / 3.66
g['events_per_month_recent'] = g.set_index(['brand', 'category', 'event_type']).index.map(rate).fillna(0)
g['predicted_prob_next_30d'] = 1 - np.exp(-g.events_per_month_recent)
g['basis'] = 'Actual + predicted'
out(g.sort_values('predicted_prob_next_30d', ascending=False), 't19_competitor_activity', 19, 'Competitor launches, price cuts, promotions by brand and category; chance of another in next 30 days', 'brand × category × event type')
rec = CA[pd.to_datetime(CA.event_date) > AS_OF - pd.Timedelta(days=45)].copy(); rec['our_sku'] = rec.competitor_sku_id.map(cp.mapped_our_sku); rec['basis'] = 'Actual'
out(rec.drop(columns='m'), 't19b_competitor_events_recent', 19, 'Field-reported competitor events in the last 45 days', 'event')

# ================= Supply and stock
INV = load('fact_inventory_weekly'); INV['snapshot_week_end'] = pd.to_datetime(INV.snapshot_week_end)
ls = INV[INV.snapshot_week_end == INV.snapshot_week_end.max()]
inv = ls.groupby(['warehouse_id', 'sku']).agg(on_hand_units=('on_hand_units', 'sum'), in_transit_units=('in_transit_units', 'sum'), allocated_units=('allocated_units', 'sum'),
        safety_stock_units=('safety_stock_units', 'first'), reorder_point_units=('reorder_point_units', 'first'), lead_time_days=('lead_time_days', 'first'),
        near_expiry_units=('on_hand_units', lambda s: s[ls.loc[s.index, 'days_to_expiry'] < 30].sum())).reset_index()
FI['warehouse_id'] = np.where(FI.channel_id == 'GR', 'RYD-DC', None)
OF = load('fact_order_fulfilment'); whm = OF.drop_duplicates('order_id').set_index('order_id').warehouse_id
FI.loc[FI.channel_id != 'GR', 'warehouse_id'] = FI.loc[FI.channel_id != 'GR', 'doc_no'].map(whm).fillna('RYD-DC')
dem = FI[FI.doc_date > AS_OF - pd.Timedelta(days=56)].groupby(['warehouse_id', 'sku']).qty_base_ea.sum() / 8
DP = load('fact_demand_plan')
inv['weekly_demand_forecast'] = inv.set_index(['warehouse_id', 'sku']).index.map(dem).fillna(0)
inv['cover_days'] = np.where(inv.weekly_demand_forecast > 0, (inv.on_hand_units - inv.allocated_units) / (inv.weekly_demand_forecast / 7), np.nan)
inv['cover_incl_transit_days'] = np.where(inv.weekly_demand_forecast > 0, (inv.on_hand_units + inv.in_transit_units - inv.allocated_units) / (inv.weekly_demand_forecast / 7), np.nan)
inv['projected_stockout_date'] = ['out now' if pd.notna(c) and c <= 0 else ((AS_OF + pd.Timedelta(days=int(c))).date() if pd.notna(c) and c < 42 else None) for c in inv.cover_incl_transit_days]
inv['status'] = np.select([inv.cover_incl_transit_days < inv.lead_time_days, inv.cover_days > 90], ['stock-out risk', 'overstock'], 'ok')
inv['status'] = np.where(inv.weekly_demand_forecast == 0, 'no recent demand', inv.status)
inv['sku_name'] = inv.sku.map(prod.sku_name); inv['division'] = inv.sku.map(prod.division); inv['stock_value_sar'] = inv.on_hand_units * inv.sku.map(prod.cost_price_sar)
inv['basis'] = 'Actual stock + predicted demand'
out(inv, 't20_stock_cover', 20, 'Warehouse stock cover vs demand forecast (last 8 weeks run rate) with projected stock-out date', 'warehouse × SKU')

# T21 fill rate and OTIF with cut reasons
OF['wk'] = wk(OF.order_date)
g = OF.groupby(['channel_id', 'wk']).agg(lines=('order_line', 'size'), ordered_cases=('ordered_cases', 'sum'), delivered_cases=('delivered_cases', 'sum'),
        otif=('otif_flag', 'mean'), on_time=('on_time_flag', 'mean'), in_full=('in_full_flag', 'mean')).reset_index()
cr = OF.dropna(subset=['cut_reason']).groupby(['channel_id', 'wk', 'cut_reason']).size().unstack(fill_value=0)
g = g.join(cr, on=['channel_id', 'wk']).fillna({c: 0 for c in cr.columns})
g['case_fill_rate'] = g.delivered_cases / g.ordered_cases; g['basis'] = 'Actual'
fw = []
for ch_, x in g.groupby('channel_id'):
    x = x.set_index('wk'); full = x[x.index <= LAST_FULL]
    for w in FWEEKS: fw.append({'channel_id': ch_, 'wk': w, 'otif': full.otif.iloc[-8:].mean(), 'case_fill_rate': full.case_fill_rate.iloc[-8:].mean(), 'lines': full.lines.iloc[-6:].mean(), 'basis': 'Predicted'})
T21 = pd.concat([g[g.wk <= LAST_FULL], pd.DataFrame(fw)]).rename(columns={'wk': 'week_start'})
T21['week_start'] = pd.to_datetime(T21.week_start).dt.date; T21['otif_target'] = T21.channel_id.map(load('dim_channel').set_index('channel_id').otif_target_pct)
out(T21, 't21_fill_otif', 21, 'Weekly OTIF, on-time, in-full and case fill by channel with cut-reason counts; 6-week outlook', 'week × channel')

# T22 customer stock, near-expiry and pending returns
CS = load('fact_customer_stock'); CS['week_end'] = pd.to_datetime(CS.week_end)
lw = CS[CS.week_end == CS.week_end.max()]
T22 = lw.groupby(['account_id', 'sku']).agg(stores=('outlet_id', 'nunique'), on_hand_units=('on_hand_units', 'sum'), near_expiry_units=('near_expiry_units', 'sum'),
        return_requested_units=('return_requested_units', 'sum'), return_authorised_units=('return_authorised_units', 'sum')).reset_index()
conv = CS.return_requested_units.sum() / max(CS.near_expiry_units.sum(), 1)
T22['pending_return_units'] = T22.return_requested_units - T22.return_authorised_units
T22['predicted_returns_next_30d_units'] = T22.near_expiry_units * min(conv, 1)
T22['predicted_returns_next_30d_sar'] = T22.predicted_returns_next_30d_units * T22.sku.map(prod.list_price_sar) * 0.85
T22['sku_name'] = T22.sku.map(prod.sku_name); T22['division'] = T22.sku.map(prod.division)
T22['snapshot'] = CS.week_end.max().date(); T22['basis'] = 'Actual stock + predicted returns'
out(T22[(T22.near_expiry_units > 0) | (T22.pending_return_units > 0) | (T22.on_hand_units > 0)], 't22_customer_stock_expiry', 22, 'Key-account stock, near-expiry units, pending returns and predicted returns in next 30 days', 'account × SKU')

# T23 demand shifts to share with planning (Nov)
nov_plan = DP[(pd.to_datetime(DP.target_month) == pd.Timestamp('2026-11-01'))].sort_values('cycle_month').drop_duplicates(['sku', 'location'], keep='last')
T23 = nov_plan[['sku', 'location', 'cycle_month', 'final_plan_units']].copy()
run = FI[FI.doc_date > AS_OF - pd.Timedelta(days=56)].groupby(['warehouse_id', 'sku']).qty_base_ea.sum() / 56 * 30
prom = PR[(PR.start_date >= pd.Timestamp('2026-11-01')) & (PR.start_date <= pd.Timestamp('2026-11-30'))].groupby('sku').planned_volume_units.sum()
T23['new_forecast_units'] = T23.set_index(['location', 'sku']).index.map(run).fillna(0) + T23.sku.map(prom).fillna(0) * 0.5
T23['shift_pct'] = T23.new_forecast_units / T23.final_plan_units.replace(0, np.nan) - 1
T23['driver'] = np.select([T23.sku.isin(prom.index) & (T23.shift_pct > 0), T23.shift_pct > 0], ['November promotion', 'recent run rate above plan'], 'recent run rate below plan')
T23['unit_change'] = T23.new_forecast_units - T23.final_plan_units
T23['share_with_planning'] = (T23.shift_pct.abs() > 0.30) & (T23.unit_change.abs() >= T23.unit_change.abs().quantile(0.75))
T23['sku_name'] = T23.sku.map(prod.sku_name); T23['division'] = T23.sku.map(prod.division); T23['basis'] = 'Predicted'
out(T23.sort_values('shift_pct', key=abs, ascending=False), 't23_demand_shifts', 23, 'November: last S&OP plan vs new forecast (8-week run rate + promo volume); shifts over 30% and in the top quarter by units flagged', 'SKU × location')

# ================= Customers and finance
OI = load('fact_open_invoice'); RC = load('fact_receipt'); CB = load('fact_credit_block')
lat = RC[RC.days_late.notna() & (pd.to_datetime(RC.receipt_date) > AS_OF - pd.Timedelta(days=120))].groupby('customer_id').days_late.agg(['mean', 'std', 'count'])
gl = RC.days_late.mean()
cust = OI.groupby(['customer_id', 'channel_id']).agg(open_sar=('open_sar', 'sum'), invoices=('invoice_id', 'size'), oldest_due=('due_date', 'min')).reset_index()
for b in ['not due', '1-30', '31-60', '61-90', '90+']:
    cust[f'bucket_{b}'] = cust.customer_id.map(OI[OI.aging_bucket == b].groupby('customer_id').open_sar.sum()).fillna(0)
cust['overdue_sar'] = cust.open_sar - cust['bucket_not due']
cust['avg_days_late_120d'] = cust.customer_id.map(lat['mean']).fillna(gl)
blocked = CB[CB.release_date.isna()].customer_id.unique()
cust['credit_blocked'] = cust.customer_id.isin(blocked)
cust['credit_limit_sar'] = cust.customer_id.map(acc.credit_limit_sar)
due_in = (pd.to_datetime(cust.oldest_due) - AS_OF).dt.days
cust['predicted_pay_date'] = [(AS_OF + pd.Timedelta(days=int(max(1, d_ + l_)))).date() for d_, l_ in zip(due_in, cust.avg_days_late_120d)]
sd = cust.customer_id.map(lat['std']).fillna(10).clip(lower=5)
from scipy.stats import norm
OI['due_in'] = (pd.to_datetime(OI.due_date) - AS_OF).dt.days
OI['p30'] = norm.cdf((30 - (OI.due_in + OI.customer_id.map(lat['mean']).fillna(gl))) / OI.customer_id.map(lat['std']).fillna(10).clip(lower=5))
cust['expected_collection_30d_sar'] = cust.customer_id.map((OI.open_sar * OI.p30).groupby(OI.customer_id).sum())
cust['prob_pay_within_30d'] = cust.expected_collection_30d_sar / cust.open_sar
cust['customer_name'] = cust.customer_id.map(acc.account_name).fillna(cust.customer_id)
cust['basis'] = 'Actual balance + predicted payment'
out(cust.sort_values('overdue_sar', ascending=False), 't24_receivables', 24, 'Open AR at 20 Oct by customer: ageing buckets, credit block, predicted pay date and 30-day collection', 'customer')
cbm = CB.assign(m=pd.to_datetime(CB.block_date).dt.to_period('M').dt.to_timestamp()).groupby(['m', 'reason']).agg(blocks=('block_id', 'size'), blocked_order_value_sar=('blocked_order_value_sar', 'sum'),
        overrides=('release_type', lambda s: (s == 'manager override').sum())).reset_index()
cbm['month'] = cbm.m.dt.date; cbm['basis'] = 'Actual'
out(cbm.drop(columns='m'), 't24b_credit_blocks_monthly', 24, 'Credit blocks per month by reason, value of orders held and overrides', 'month × reason')

# T25 claims, deductions and disputes
DD = load('fact_deduction_dispute')
closed = DD[DD.status != 'open']; acc_rate = (closed.status == 'accepted (credit note)').groupby(closed.reason).mean()
DD['predicted_valid_prob'] = DD.reason.map(acc_rate).fillna(0.5)
DD['predicted_recovery_sar'] = np.where(DD.status == 'open', DD.deduction_sar * (1 - DD.predicted_valid_prob) * 0.8, DD.recovered_sar)
DD['predicted_resolution_date'] = np.where(DD.status == 'open', (pd.to_datetime(DD.opened_date) + pd.to_timedelta(closed.days_open.median() * 1.3, 'D')).dt.date, None)
DD['customer_name'] = DD.customer_id.map(acc.account_name)
DD['basis'] = np.where(DD.status == 'open', 'Actual + predicted', 'Actual')
out(DD, 't25_deductions_disputes', 25, 'Deduction and dispute cases with predicted validity, recovery and resolution date for open cases', 'case')

# T26 key account escalations + risk next 30 days
ES = load('fact_account_escalation'); ES['basis'] = 'Actual'; ES['account_name'] = ES.account_id.map(acc.account_name)
out(ES, 't26_escalations', 26, 'Key account escalation cases with severity, owner and status', 'escalation')
kas = acc.index[acc.channel_id.isin(['MT', 'DS', 'WS', 'EC'])]
OF['m'] = pd.to_datetime(OF.order_date)
o30 = OF[OF.m > AS_OF - pd.Timedelta(days=30)].groupby('account_id').otif_flag.mean()
rk = pd.DataFrame(index=kas)
rk['otif_30d'] = o30; rk['otif_target'] = acc.otif_target_pct
rk['open_disputes_sar'] = DD[DD.status == 'open'].groupby('customer_id').deduction_sar.sum()
rk['overdue_sar'] = cust.set_index('customer_id').overdue_sar
rk['open_escalations'] = ES[ES.status.str.startswith('open')].groupby('account_id').size()
rk['escalations_6m'] = ES[pd.to_datetime(ES.opened_date) > AS_OF - pd.Timedelta(days=180)].groupby('account_id').size()
rk = rk.fillna(0)
z = (2.5 * np.clip(rk.otif_target - rk.otif_30d, 0, 0.3) * 10 + np.log1p(rk.open_disputes_sar) / 6 + np.log1p(rk.overdue_sar) / 8 + rk.open_escalations * 0.8 + rk.escalations_6m * 0.25 - 5.5)
rk['predicted_escalation_prob_30d'] = 1 / (1 + np.exp(-z))
rk['main_driver'] = np.select([rk.otif_30d < rk.otif_target - 0.03, rk.open_disputes_sar > 20000, rk.overdue_sar > 100000], ['service (OTIF below target)', 'open disputes', 'overdue balance'], 'history')
rk['account_name'] = acc.account_name; rk['channel_id'] = acc.channel_id; rk['basis'] = 'Predicted'
out(rk.reset_index(names='account_id').sort_values('predicted_escalation_prob_30d', ascending=False), 't26b_account_risk', 26, 'Chance each key account escalates in the next 30 days, and the main driver', 'account')

# ================= Team
TA = load('fact_team_attendance'); TA['date'] = pd.to_datetime(TA.date)
g = TA.groupby(['date', 'role']).agg(headcount=('employee_id', 'size'), present=('status', lambda s: (s == 'present').sum()),
        absent=('status', lambda s: s.isin(['sick', 'absent', 'annual leave']).sum()), vacant=('status', lambda s: (s == 'vacant').sum()),
        uncovered_routes=('route_uncovered', 'sum'), covered=('cover_employee_id', lambda s: s.notna().sum())).reset_index()
g['uncovered_merch_days'] = np.where(g.role == 'Merchandiser', g.absent + g.vacant - g.covered, 0)
g = g[g.role.isin(['Van rep', 'Merchandiser'])]; g['basis'] = 'Actual'
fut = []
for role_, x in g.groupby('role'):
    x = x[x.date > AS_OF - pd.Timedelta(days=56)]
    rate = x.groupby(x.date.dt.dayofweek)[['absent', 'uncovered_routes', 'uncovered_merch_days', 'vacant']].mean()
    hc = x.headcount.iloc[-5:].mean()
    for d_ in pd.date_range(AS_OF + pd.Timedelta(days=1), AS_OF + pd.Timedelta(days=21)):
        if d_.dayofweek == 4 or d_.dayofweek not in rate.index: continue
        rr_ = rate.loc[d_.dayofweek]
        fut.append({'date': d_, 'role': role_, 'headcount': round(hc), 'absent': rr_.absent, 'vacant': rr_.vacant, 'uncovered_routes': rr_.uncovered_routes,
                    'uncovered_merch_days': rr_.uncovered_merch_days, 'basis': 'Predicted'})
T27 = pd.concat([g[g.date > AS_OF - pd.Timedelta(days=90)], pd.DataFrame(fut)]); T27['date'] = T27.date.dt.date
out(T27, 't27_coverage', 27, 'Daily van-rep and merchandiser absence, vacancies, uncovered routes and stores (last 90 days) and next 3 weeks predicted', 'date × role')

MT_ = load('fact_meeting'); MT_['date'] = pd.to_datetime(MT_.date); MT_['wk'] = wk(MT_.date)
emp = load('dim_employee').set_index('employee_id')
MT_['role'] = MT_.organiser_id.map(emp.role)
g = MT_.groupby(['wk', 'role', 'type']).agg(meetings=('meeting_id', 'size'), hours=('duration_min', lambda s: s.sum() / 60), value_at_stake_sar=('value_at_stake_sar', 'sum'),
        agreed=('outcome', lambda s: s.isin(['agreed', 'partly agreed']).sum()), cancelled=('outcome', lambda s: (s == 'cancelled').sum())).reset_index()
g['basis'] = 'Actual'
fut = g[g.wk > LAST_FULL - pd.Timedelta(days=56)].groupby(['role', 'type'])[['meetings', 'hours', 'value_at_stake_sar']].mean().reset_index()
fut = pd.concat([fut.assign(wk=w) for w in FWEEKS[:4]]).assign(basis='Predicted')
T28 = pd.concat([g[g.wk <= LAST_FULL], fut]).rename(columns={'wk': 'week_start'}); T28['week_start'] = pd.to_datetime(T28.week_start).dt.date
out(T28, 't28_meetings', 28, 'KAM, supervisor and manager meetings per week by type: hours, value at stake, outcomes; next 4 weeks predicted', 'week × role × meeting type')

AL = load('fact_action_log'); AL['created_date'] = pd.to_datetime(AL.created_date); AL['due_date'] = pd.to_datetime(AL.due_date)
stale = AL[AL.status.isin(['open', 'overdue']) & (AL.created_date < AS_OF - pd.Timedelta(days=90))]
print('    stale actions (open > 90 days, treated as expired):', len(stale))
op = AL[AL.status.isin(['open', 'overdue']) & (AL.created_date >= AS_OF - pd.Timedelta(days=90))].copy()
med_close = AL[AL.closed_date.notna()].groupby(AL.source.str.split(':').str[0]).waiting_days.median()
op['src'] = op.source.str.split(':').str[0]
op['predicted_close_date'] = (AS_OF + pd.to_timedelta(np.maximum(1, op.src.map(med_close).fillna(5) - op.waiting_days).clip(lower=1) + op.waiting_days * 0.3, 'D')).dt.date
op['prob_miss_due_date'] = np.where(op.due_date < AS_OF, 1.0, 1 - np.exp(-(op.waiting_days + 1) / (op.due_date - op.created_date).dt.days.clip(lower=1)))
op['owner_role'] = op.owner_id.map(emp.role); op['basis'] = 'Actual + predicted'
out(op.drop(columns='src'), 't29_action_log_open', 29, 'Open and overdue actions/approvals with waiting days, predicted close date and chance of missing due date', 'action')
AL['closed_eff'] = pd.to_datetime(AL.closed_date).fillna(AL.created_date + pd.Timedelta(days=90))
AL.loc[AL.closed_eff > AS_OF, 'closed_eff'] = pd.NaT
AL['wk'] = wk(AL.created_date); cl = AL.dropna(subset=['closed_eff']).assign(wk=lambda x: wk(x.closed_eff)).groupby('wk').size()
s29 = pd.DataFrame({'created': AL.groupby('wk').size(), 'closed': cl}).fillna(0); s29 = s29[s29.index <= LAST_FULL]
s29['backlog'] = (s29.created - s29.closed).cumsum(); s29['basis'] = 'Actual'
cr_, cl_ = s29.created.iloc[-6:].mean(), s29.closed.iloc[-6:].mean()
fb = pd.DataFrame({'created': cr_, 'closed': cl_, 'basis': 'Predicted'}, index=FWEEKS); fb['backlog'] = s29.backlog.iloc[-1] + (cr_ - cl_) * np.arange(1, len(FWEEKS) + 1)
s29 = pd.concat([s29, fb]).reset_index(names='week_start'); s29['week_start'] = s29.week_start.dt.date
out(s29, 't29b_action_backlog', 29, 'Actions created and closed per week and open backlog; 6-week outlook', 'week')

MTm = load('fact_manager_time'); MTm['wk'] = wk(MTm.date)
g = MTm.groupby(['employee_id', 'wk', 'activity']).hours.sum().reset_index(); g['basis'] = 'Actual'
g['role'] = g.employee_id.map(emp.role)
target_mix = {'key account meetings': 0.25, 'field visits': 0.20, 'approvals & escalations': 0.12, 'planning & forecasting': 0.13, 'internal reviews / reporting': 0.15, 'admin / email': 0.15}
fut = g[g.wk > LAST_FULL - pd.Timedelta(days=42)].groupby(['employee_id', 'role', 'activity']).hours.mean().reset_index()
fut = pd.concat([fut.assign(wk=w) for w in FWEEKS[:4]]).assign(basis='Predicted (no change)')
T30 = pd.concat([g[g.wk <= LAST_FULL], fut]).rename(columns={'wk': 'week_start'}); T30['week_start'] = pd.to_datetime(T30.week_start).dt.date
T30['share'] = T30.hours / T30.groupby(['employee_id', 'week_start']).hours.transform('sum')
T30['target_share'] = T30.activity.map(target_mix)
out(T30, 't30_manager_time', 30, 'Channel managers: hours per activity per week vs a suggested target mix; next 4 weeks if nothing changes', 'week × manager × activity')

M = pd.DataFrame(MAN, columns=['task', 'file', 'grain', 'rows', 'what_it_shows'])
M.to_csv(f'{PO_}/_manifest.csv', index=False)
print(M.groupby('task').size().shape, M.rows.sum())