""" Polleo XYZ Classification — v3.6 ================================= Computes demand-variability classification (XYZ) per SKU, specifically on the wholesale channel AND on total demand, over the trailing 26 weeks. Classification rule (coefficient of variation on non-zero weeks): X: CV < 0.5 → stable / predictable Y: 0.5 <= CV < 1.0 → variable Z: CV >= 1.0 → highly variable / lumpy Writes two new columns to sku_plan_list.csv: ws_xyz — wholesale-channel XYZ class (most relevant for KAM attention) total_xyz — total-demand XYZ class (retail + webshop + wholesale) SKUs with no wholesale activity → ws_xyz = '-' (not applicable) SKUs with fewer than 4 non-zero weeks → ws_xyz = 'Z' (treat as lumpy by default) Usage: python compute_xyz.py [data_dir] Default: data/ """ import pandas as pd, numpy as np, sys, os from datetime import datetime import io if sys.stdout.encoding and sys.stdout.encoding.lower() not in ('utf-8', 'utf8'): try: sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', errors='replace') except Exception: pass def classify_cv(cv): """Return XYZ letter given a coefficient of variation.""" if cv is None or np.isnan(cv): return 'Z' if cv < 0.5: return 'X' if cv < 1.0: return 'Y' return 'Z' def compute_xyz(data_dir='data', window_weeks=26): print(f'\n{"="*60}') print(f' POLLEO XYZ CLASSIFICATION (wholesale + total)') print(f' {datetime.now().strftime("%Y-%m-%d %H:%M")}') print(f'{"="*60}') sales_path = os.path.join(data_dir, 'sales_clean.csv') plan_path = os.path.join(data_dir, 'sku_plan_list.csv') if not os.path.exists(sales_path): print(f' ERROR: {sales_path} not found') return None if not os.path.exists(plan_path): print(f' ERROR: {plan_path} not found') return None sc = pd.read_csv(sales_path) plan = pd.read_csv(plan_path) plan_skus = set(plan['sku']) print(f' Loaded {len(sc):,} rows from sales_clean.csv') print(f' Classifying {len(plan_skus)} planned SKUs (window: last {window_weeks} weeks)') # Keep only the trailing window (vectorized, handles ~100k rows fast) all_yw = sc[['year', 'week']].drop_duplicates().sort_values(['year', 'week']) window_df = all_yw.tail(window_weeks).copy() window_df['_window'] = 1 sc_w = sc.merge(window_df, on=['year', 'week'], how='inner') sc_w = sc_w[sc_w['sku'].isin(plan_skus)] results = [] for sku, grp in sc_w.groupby('sku'): ws = grp['qty_wholesale'].values tot = grp['qty_total'].values ws_nz = ws[ws > 0] tot_nz = tot[tot > 0] # Wholesale CV (on non-zero weeks; penalizes SKUs with sparse wholesale) if len(ws_nz) == 0: ws_cv, ws_class = None, '-' # no wholesale activity elif len(ws_nz) < 4: # Too few observations to trust CV; default to Z ws_cv = float(np.std(ws_nz) / np.mean(ws_nz)) if np.mean(ws_nz) > 0 else 1.5 ws_class = 'Z' else: mean = np.mean(ws_nz) std = np.std(ws_nz) ws_cv = float(std / mean) if mean > 0 else float('nan') ws_class = classify_cv(ws_cv) # Total CV if len(tot_nz) == 0: tot_cv, tot_class = None, '-' elif len(tot_nz) < 4: tot_cv = float(np.std(tot_nz) / np.mean(tot_nz)) if np.mean(tot_nz) > 0 else 1.5 tot_class = 'Z' else: mean = np.mean(tot_nz) std = np.std(tot_nz) tot_cv = float(std / mean) if mean > 0 else float('nan') tot_class = classify_cv(tot_cv) # Also compute wholesale share + non-zero week count (diagnostic) ws_share = float(ws.sum() / (grp['qty_retail'].sum() + grp['qty_webshop'].sum() + ws.sum())) if ( grp['qty_retail'].sum() + grp['qty_webshop'].sum() + ws.sum() > 0 ) else 0.0 results.append({ 'sku': sku, 'ws_xyz': ws_class, 'ws_cv': round(ws_cv, 3) if ws_cv is not None else None, 'ws_nz_weeks': int(len(ws_nz)), 'total_xyz': tot_class, 'total_cv': round(tot_cv, 3) if tot_cv is not None else None, 'total_nz_weeks': int(len(tot_nz)), 'ws_share_26w': round(ws_share, 3), }) xyz_df = pd.DataFrame(results) # Add SKUs that have no sales data in the window (treat as Z / unknown) missing = plan_skus - set(xyz_df['sku']) if missing: fill = pd.DataFrame([ {'sku': s, 'ws_xyz': '-', 'ws_cv': None, 'ws_nz_weeks': 0, 'total_xyz': '-', 'total_cv': None, 'total_nz_weeks': 0, 'ws_share_26w': 0.0} for s in missing ]) xyz_df = pd.concat([xyz_df, fill], ignore_index=True) # Merge back into sku_plan_list.csv # Preserve all existing columns; overwrite only the XYZ fields xyz_cols = ['ws_xyz', 'ws_cv', 'ws_nz_weeks', 'total_xyz', 'total_cv', 'total_nz_weeks', 'ws_share_26w'] plan_enriched = plan.copy() for c in xyz_cols: if c in plan_enriched.columns: plan_enriched = plan_enriched.drop(columns=[c]) plan_enriched = plan_enriched.merge( xyz_df[['sku'] + xyz_cols], on='sku', how='left' ) # Write atomically — to a tmp file first, then rename tmp_path = plan_path + '.tmp' plan_enriched.to_csv(tmp_path, index=False) os.replace(tmp_path, plan_path) # ---- Summary ---- print(f'\n Results:') print(f'\n Wholesale XYZ distribution (planning SKUs):') for cls in ['X', 'Y', 'Z', '-']: n = (xyz_df['ws_xyz'] == cls).sum() pct = n / len(xyz_df) * 100 if len(xyz_df) else 0 label = {'X': 'stable', 'Y': 'variable', 'Z': 'lumpy', '-': 'no wholesale'}[cls] print(f' {cls} ({label:13s}): {n:4d} SKUs ({pct:.1f}%)') print(f'\n Total-demand XYZ distribution:') for cls in ['X', 'Y', 'Z', '-']: n = (xyz_df['total_xyz'] == cls).sum() pct = n / len(xyz_df) * 100 if len(xyz_df) else 0 label = {'X': 'stable', 'Y': 'variable', 'Z': 'lumpy', '-': 'no sales'}[cls] print(f' {cls} ({label:13s}): {n:4d} SKUs ({pct:.1f}%)') # SKUs that need the most KAM attention: Z-wholesale + high ws_share priority = xyz_df[(xyz_df['ws_xyz'] == 'Z') & (xyz_df['ws_share_26w'] >= 0.5)] print(f'\n Priority KAM-attention SKUs (Z-wholesale + ws_share >= 50%): ' f'{len(priority)} SKUs') print(f'\n Saved: {plan_path}') print(f'{"="*60}\n') return xyz_df if __name__ == '__main__': data_dir = sys.argv[1] if len(sys.argv) > 1 else 'data' try: compute_xyz(data_dir) except Exception as e: print(f'\n ERROR: {e}') import traceback; traceback.print_exc() try: input('\n Press Enter to close...') except (EOFError, OSError): pass