"""03_Simulated - inferred dimensions (rules / models on client data; no outside numbers). inf_client_principal_map client code -> principal(s) behind its SKUs, client type (Binzagr BU vs external 3PL) inf_warehouse_profile per warehouse: activity timeline, structural changes, client mix, cold-chain indicator inf_customer_channel_map customer hierarchy code -> inferred channel label from customer names (EN + AR keywords) inf_badge_session_risk per work session: rule flags + IsolationForest anomaly score -> badge-sharing risk tier """ import glob, os, re import numpy as np import pandas as pd from sklearn.ensemble import IsolationForest from sim_common import CLEAN, DATA, write_table, save_stats TX = sorted(glob.glob(os.path.join(CLEAN, 'jeddah_wms_transactions', '*.parquet'))) CHANNEL_KW = { 'baqala / small grocery': [r'\bBAQ+AL', r'\bBAK+AL', r'GROCER', r'بقال'], 'tamweenat (grocery store)': [r'TAMW', r'\bTMW', r'TAMEEN', r'^TAM\b', r'\bTAM\.?\s', r'تموين'], 'mini / supermarket': [r'MINI ?MARKET', r'SUPER ?MARKET', r'\bMARKETS?\b', r'MART\b', r'ASWA[QK]', r'ماركت', r'سوبر', r'اسواق|أسواق'], 'key account / hypermarket': [r'HYPER', r'PANDA', r'OTHAIM', r'DANUBE', r'TAMIMI', r'CARREFOUR', r'LULU', r'BIN ?DAWOOD', r'NESTO', r'FARM SUPER', r'هايبر', r'بنده|باندا', r'العثيم', r'الدانوب'], 'HoReCa (food service)': [r'RESTAURANT', r'\bREST\b\.?', r'MATA+M', r'BU+F+I+A|BOOFIA|BUFFET', r'CAFE|CAFÉ|COFFEE', r'BAKER(Y|IES)', r'HOTEL', r'KITCHEN', r'CATERING', r'FAST ?FOOD', r'SHAWARMA|BROAST|BURGER|PIZZA', r'HALAWIY|SWEETS', r'مطعم', r'بوفي', r'كافي|كوفي', r'مخبز|مخابز', r'فندق', r'مطابخ|مطبخ', r'حلويات'], 'pharmacy': [r'PHARMAC', r'\bPHAR\b', r'صيدلي'], 'perfumery / cosmetics': [r'PERFUM', r'COSMETIC', r'عطور'], 'wholesale / trading': [r'WHOLESALE', r'TRADING', r'جملة|الجملة', r'تجارة|للتجارة'], 'fuel station / automotive': [r'STATION', r'PETROL', r'AUTOMOTIVE SERVICES', r'SASCO', r'محطة'], } COLD_KW = r'FROZEN|ICE ?CREAM|CHILLED|FRIES|NUGGET|CHEESE|BUTTER|YOGH?URT|MCCAIN|FRESH MILK|مجمد|ايس كريم|آيس|جبن|زبد' def main(): st = {} prod = pd.read_parquet(os.path.join(CLEAN, 'dim_product.parquet'), columns=['prd_cd', 'prd_desc_en', 'prd_desc_combined', 'l1_code', 'l1_desc', 'l2_desc', 'l4_desc']) prod['cold_kw'] = prod.prd_desc_combined.fillna('').str.upper().str.contains(COLD_KW, regex=True) cols = ['wh_id', 'month', 'client_code', 'item_client', 'item_code', 'item_kind', 'tran_category', 'employee_id_norm'] T = pd.concat([pd.read_parquet(p, columns=cols) for p in TX], ignore_index=True) # ---------------- 1. client -> principal sku = T[T.item_kind == 'sku'].merge(prod[['prd_cd', 'l1_code', 'l1_desc', 'l2_desc', 'cold_kw']], left_on='item_code', right_on='prd_cd', how='left') rows = [] for cl, g in sku.groupby('item_client'): matched = g.prd_cd.notna() pr = g[matched].groupby(['l1_code', 'l1_desc']).size().sort_values(ascending=False) top = pr.head(3) rows.append({'client_code': cl, 'sku_events': int(len(g)), 'distinct_items': int(g.item_code.nunique()), 'share_items_matching_product_master': float(g[['item_code']].drop_duplicates().item_code.isin(prod.prd_cd).mean()), 'client_type_inferred': 'binzagr_business_unit' if matched.mean() > 0.8 else ('external_3pl_client' if matched.mean() < 0.05 else 'mixed'), 'top_principal_code': top.index[0][0] if len(top) else None, 'top_principal': top.index[0][1] if len(top) else None, 'top_principal_share': float(top.iloc[0] / pr.sum()) if len(top) else None, 'principals_top3': '; '.join(f'{n} ({v / pr.sum():.0%})' for (c, n), v in top.items()) if len(top) else '', 'n_principals': int(len(pr)), 'cold_chain_item_share': float(g[matched].cold_kw.mean()) if matched.any() else None}) cp = pd.DataFrame(rows).sort_values('sku_events', ascending=False) cp['confidence'] = np.where(cp.client_type_inferred == 'external_3pl_client', 'high (own item codes, 0% match)', np.where(cp.top_principal_share >= 0.8, 'high', np.where(cp.top_principal_share >= 0.5, 'medium', 'low (multi-principal unit)'))) cp['basis'] = 'Inferred (item code -> product master principal)' st['clients'] = int(len(cp)); st['client_types'] = cp.client_type_inferred.value_counts().to_dict() # ---------------- 2. warehouse profile ctype = cp.set_index('client_code').client_type_inferred T['client_type'] = T.client_code.map(ctype).fillna('unknown') wm = T.groupby(['wh_id', 'month']).size().unstack(fill_value=0) wp = [] for wh, row in wm.iterrows(): active = row[row > 0] chg = (row / row.shift(1).replace(0, np.nan) - 1) breaks = [f'{m}: {v:+.0%}' for m, v in chg.items() if pd.notna(v) and abs(v) >= 0.5 and max(row[m], row.shift(1)[m]) >= 2000] g = T[T.wh_id == wh] topc = g.client_code.value_counts(normalize=True).head(3) cat = g.tran_category.value_counts(normalize=True).head(3) s = sku[sku.wh_id == wh] cold = float(s[s.prd_cd.notna()].cold_kw.mean()) if s.prd_cd.notna().any() else np.nan wp.append({'wh_id': wh, 'rows_total': int(row.sum()), 'first_month': active.index.min(), 'last_month': active.index.max(), 'months_active': int((row > 0).sum()), 'rows_by_month': '; '.join(f'{m}:{int(v):,}' for m, v in row.items()), 'structural_changes': '; '.join(breaks), 'top_clients': '; '.join(f'{c} ({v:.0%})' for c, v in topc.items()), 'external_3pl_share': float((g.client_type == 'external_3pl_client').mean()), 'top_activities': '; '.join(f'{c} ({v:.0%})' for c, v in cat.items()), 'badges': int(g.employee_id_norm.nunique()), 'cold_chain_keyword_share_of_sku_events': cold, 'temperature_regime_inferred': 'likely cold-chain' if cold >= 0.5 else ('mixed' if cold >= 0.15 else 'likely ambient') if pd.notna(cold) else 'unknown'}) WP = pd.DataFrame(wp).sort_values('rows_total', ascending=False) WP['basis'] = 'Inferred (from Actual activity; temperature from product-name keywords - low confidence)' # ---------------- 3. customer channel inference cust = pd.read_parquet(os.path.join(CLEAN, 'dim_customer.parquet')) sales = pd.read_parquet(os.path.join(CLEAN, 'riyadh_sales_lines.parquet'), columns=['dist_cd', 'cust_cd', 'net_amt']) sv = sales.groupby(['dist_cd', 'cust_cd']).net_amt.sum() cust['sales_sar'] = sv.reindex(pd.MultiIndex.from_frame(cust[['dist_cd', 'cust_cd']])).fillna(0).to_numpy() names = (cust.cust_name.fillna('') + ' ' + cust.cust_name2.fillna('')).str.upper() for ch, pats in CHANNEL_KW.items(): cust[f'kw_{ch}'] = names.str.contains('|'.join(pats), regex=True) kwcols = [f'kw_{ch}' for ch in CHANNEL_KW] out = [] for lvl in ('cust_hier1', 'cust_hier2', 'cust_hier3'): for code, g in cust.groupby(lvl): shares = g[kwcols].mean() any_kw = g[kwcols].any(axis=1).mean() top = shares.sort_values(ascending=False) lab = top.index[0][3:] if top.iloc[0] >= 0.25 and (len(top) < 2 or top.iloc[0] >= 1.5 * top.iloc[1]) else 'mixed / unclear' out.append({'level': lvl, 'code': code, 'customers': int(len(g)), 'active_customers': int((g.cust_status == '1').sum()), 'credit_customers': int(g.has_credit_terms.sum()), 'sales_sar_mar_oct_2026': float(g.sales_sar.sum()), 'share_names_with_any_keyword': float(any_kw), **{f'share_{k[3:]}': float(v) for k, v in shares.items()}, 'inferred_channel': lab, 'confidence': 'medium' if top.iloc[0] >= 0.5 else ('low' if lab != 'mixed / unclear' else 'none'), 'example_names': ' | '.join(g.cust_name.dropna().sample(min(5, len(g)), random_state=1).tolist())}) CH = pd.DataFrame(out) CH['basis'] = 'Inferred (customer-name keywords, English + Arabic)' st['channel_codes'] = int(len(CH)); st['channel_labels_hier2'] = CH[CH.level == 'cust_hier2'].set_index('code').inferred_channel.to_dict() # ---------------- 4. badge-sharing risk per PERSON session (rules + IsolationForest) S = pd.read_parquet(os.path.join(DATA, 'sim_session_attendance.parquet')) S['events_per_hour'] = S.n_events / S.elapsed_h.clip(lower=0.25) feats = ['elapsed_h', 'n_events', 'events_per_hour', 'n_warehouses', 'n_fast_wh_switch', 'n_badges', 'concurrent_badge_h', 'same_second_share'] X = S[feats].fillna(0).to_numpy(dtype='float64') iso = IsolationForest(n_estimators=300, contamination=0.03, random_state=7).fit(X) S['anomaly_score'] = -iso.score_samples(X) # higher = more unusual S['anomaly_flag_ml'] = iso.predict(X) == -1 S['rule_gt_12h'] = S.elapsed_h > 12 S['rule_gt_16h'] = S.elapsed_h > 16 S['rule_fast_wh_switch'] = S.n_fast_wh_switch > 0 S['rule_extreme_rate'] = S.events_per_hour > S.events_per_hour.quantile(0.995) S['rule_concurrent_badges'] = S.concurrent_badge_h > 0.5 # two badges of one payroll number active together > 30 min nrules = S[['rule_gt_12h', 'rule_fast_wh_switch', 'rule_extreme_rate', 'rule_concurrent_badges']].sum(axis=1) S['badge_sharing_risk'] = np.where(S.rule_gt_16h | (S.rule_gt_12h & S.anomaly_flag_ml) | (S.rule_concurrent_badges & S.anomaly_flag_ml), 'high', np.where(S.rule_gt_12h | S.rule_concurrent_badges | (nrules >= 1) & S.anomaly_flag_ml, 'medium', np.where(S.anomaly_flag_ml | (nrules >= 1), 'low', 'none'))) R = S[['session_id', 'person_id', 'badges', 'n_badges', 'wms_role', 'main_wh', 'month', 'session_start', 'elapsed_h', 'n_events', 'events_per_hour', 'n_warehouses', 'n_fast_wh_switch', 'concurrent_badge_h', 'same_second_share', 'is_bulk_posting', 'rule_gt_12h', 'rule_gt_16h', 'rule_fast_wh_switch', 'rule_extreme_rate', 'rule_concurrent_badges', 'anomaly_score', 'anomaly_flag_ml', 'badge_sharing_risk']].copy() R['basis'] = 'Inferred (rules + IsolationForest on scan behaviour; person = payroll number from Binzagr user list; needs attendance data to confirm)' st['risk_tiers'] = R.badge_sharing_risk.value_counts().to_dict() st['risk_by_badge_top'] = R[R.badge_sharing_risk.isin(['high', 'medium'])].badges.value_counts().head(10).to_dict() st['risk_concurrent_badge_sessions'] = int(R.rule_concurrent_badges.sum()) files = [] files += write_table(cp, 'inf_client_principal_map') files += write_table(WP, 'inf_warehouse_profile') files += write_table(CH, 'inf_customer_channel_map') files += write_table(R, 'inf_badge_session_risk') st['files'] = files save_stats('inferred', st) print({k: v for k, v in st.items() if k != 'files'}) if __name__ == '__main__': main()