""" build_promo_performance.py - Track historical promo performance per (SKU, campaign). ================================================================================== For each contiguous promo period in erp_promo_calendar.csv, computes: - BEFORE = average weekly qty_total over the 4 weeks immediately preceding, excluding any week that was a promo week for the SAME SKU. - DURING = average weekly qty_total across the promo weeks. - AFTER = average weekly qty_total over the 4 weeks immediately following, excluding any week that was a promo week for the SAME SKU. - actual_uplift = DURING / BEFORE - cannibalization = AFTER / BEFORE (<1 = post-promo dip, >=1 = no dip) - net_effect = (sum_during + sum_after) / (BEFORE * (n_promo + n_after_weeks)) i.e. "did the campaign earn back vs the run-rate baseline?" Inputs: data/erp_promo_calendar.csv (ground-truth promo windows), data/sales_clean.csv (qty_total for the math), data/sku_plan_list.csv (optional, just attaches cat + oznaka). Output: data/promo_performance.csv Usage: python build_promo_performance.py Also chained at end of update_sales.py so this refreshes weekly. Edge cases handled: - Promo at start of history → BEFORE is None (no 4-week clean window) - Consecutive promos for same SKU → exclude other promo weeks from BEFORE/AFTER - SKUs not in sku_plan_list → still tracked, cat/oznaka empty - Missing qty in sales_clean for an asked week → treated as no data (drops from avg) """ import io import os import sys from datetime import datetime, timedelta from pathlib import Path import pandas as pd # Fix Windows console encoding for Croatian characters 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') sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding='utf-8', errors='replace') except Exception: pass BASE_DIR = Path(os.path.dirname(os.path.abspath(__file__))) DATA_DIR = BASE_DIR / 'data' if not DATA_DIR.exists() or not (DATA_DIR / 'sales_clean.csv').exists(): DATA_DIR = BASE_DIR OUT_PATH = DATA_DIR / 'promo_performance.csv' WINDOW_WEEKS = 4 # BEFORE / AFTER window length OUTPUT_COLS = [ 'sku', 'promo_start_year', 'promo_start_week', 'promo_end_year', 'promo_end_week', 'n_promo_weeks', 'qty_before_avg', 'qty_during_avg', 'qty_after_avg', 'actual_uplift', 'cannibalization', 'net_effect', 'promo_types', 'cat', 'oznaka', ] def week_to_monday(y: int, w: int) -> datetime: """ISO (year, week) -> Monday datetime of that week.""" return datetime.strptime(f'{y}-W{w:02d}-1', '%G-W%V-%u') def monday_to_yw(d: datetime) -> tuple[int, int]: iso = d.isocalendar() return int(iso[0]), int(iso[1]) def _contiguous_runs(weeks_sorted): """Given a date-sorted list of (year, week, promo_types) tuples for one SKU, return list-of-lists where each inner list is a contiguous promo run (7-day gap = contiguous, anything more = new run).""" if not weeks_sorted: return [] runs = [] current = [weeks_sorted[0]] prev_d = week_to_monday(weeks_sorted[0][0], weeks_sorted[0][1]) for entry in weeks_sorted[1:]: d = week_to_monday(entry[0], entry[1]) if (d - prev_d).days == 7: current.append(entry) else: runs.append(current) current = [entry] prev_d = d runs.append(current) return runs def main(): print(f"\n{'='*60}") print(f' POLLEO PROMO PERFORMANCE BUILD') print(f' {datetime.now().strftime("%Y-%m-%d %H:%M")}') print(f'{"="*60}') erp_path = DATA_DIR / 'erp_promo_calendar.csv' sales_path = DATA_DIR / 'sales_clean.csv' plan_path = DATA_DIR / 'sku_plan_list.csv' if not erp_path.exists(): print(f' ERROR: {erp_path} not found. Run build_erp_promo.py first.') return if not sales_path.exists(): print(f' ERROR: {sales_path} not found. Run update_sales.py first.') return print(f' Loading ERP promo calendar...') erp = pd.read_csv(erp_path) if 'is_erp_promo' in erp.columns: erp = erp[erp['is_erp_promo'] == 1] erp['sku'] = erp['sku'].astype(str) erp['year'] = erp['year'].astype(int) erp['week'] = erp['week'].astype(int) if 'promo_types' not in erp.columns: erp['promo_types'] = '' erp['promo_types'] = erp['promo_types'].fillna('').astype(str) print(f' {len(erp):,} promo-week rows | {erp["sku"].nunique():,} SKUs') print(f' Loading sales_clean...') sales = pd.read_csv(sales_path) sales['sku'] = sales['sku'].astype(str) sales['year'] = sales['year'].astype(int) sales['week'] = sales['week'].astype(int) # Dict lookup is ~50x faster than per-row merge across 100k+ promo periods. qty_map = {(s, y, w): float(q) for s, y, w, q in zip(sales['sku'], sales['year'], sales['week'], sales['qty_total'])} print(f' {len(sales):,} sales-week rows') cat_map, ozn_map = {}, {} if plan_path.exists(): plan = pd.read_csv(plan_path) plan['sku'] = plan['sku'].astype(str) cat_map = dict(zip(plan['sku'], plan.get('cat', pd.Series([''] * len(plan))))) ozn_map = dict(zip(plan['sku'], plan.get('oznaka', pd.Series([''] * len(plan))))) print(f' plan_list: {len(plan):,} SKUs (for cat/oznaka)') else: print(f' [no sku_plan_list.csv — cat/oznaka columns will be blank]') out_rows = [] # Group promo rows per SKU once, then per-SKU detect contiguous runs. print(f'\n Computing promo performance per (SKU, campaign)...') for sku, grp in erp.groupby('sku', sort=False): # Unique (year, week) with promo_types collected per week weeks_for_sku = (grp.sort_values(['year', 'week']) .drop_duplicates(subset=['year', 'week'], keep='first')) weeks_sorted = list(zip(weeks_for_sku['year'].tolist(), weeks_for_sku['week'].tolist(), weeks_for_sku['promo_types'].tolist())) sku_promo_set = set((y, w) for (y, w, _) in weeks_sorted) for period in _contiguous_runs(weeks_sorted): start_y, start_w, _ = period[0] end_y, end_w, _ = period[-1] n_promo = len(period) promo_types_combined = '; '.join(sorted({pt.strip() for (_, _, pt) in period if pt and pt.strip()})) # BEFORE / AFTER windows — skip weeks that are themselves promo # weeks for this SKU (could be a separate campaign right next door). start_d = week_to_monday(start_y, start_w) before_keys = [] for off in range(1, WINDOW_WEEKS + 1): y, w = monday_to_yw(start_d - timedelta(weeks=off)) if (y, w) not in sku_promo_set: before_keys.append((y, w)) end_d = week_to_monday(end_y, end_w) after_keys = [] for off in range(1, WINDOW_WEEKS + 1): y, w = monday_to_yw(end_d + timedelta(weeks=off)) if (y, w) not in sku_promo_set: after_keys.append((y, w)) during_keys = [(y, w) for (y, w, _) in period] def _avg(keys): vals = [qty_map.get((sku, y, w)) for (y, w) in keys] vals = [v for v in vals if v is not None] return (sum(vals) / len(vals)) if vals else None def _sum(keys): vals = [qty_map.get((sku, y, w), 0.0) for (y, w) in keys] return sum(v for v in vals if v is not None) qty_before = _avg(before_keys) qty_during = _avg(during_keys) qty_after = _avg(after_keys) uplift = (qty_during / qty_before if qty_before and qty_before > 0 and qty_during is not None else None) cann = (qty_after / qty_before if qty_before and qty_before > 0 and qty_after is not None else None) # net_effect: actual volume over (promo + 4-week-after) horizon # vs expected volume at the BEFORE run-rate. >1 = campaign added # genuine demand; <1 = it cannibalized future weeks instead. actual_volume = _sum(during_keys) + _sum(after_keys) expected = (qty_before * (n_promo + len(after_keys)) if qty_before and qty_before > 0 else None) net = (actual_volume / expected if expected and expected > 0 else None) out_rows.append({ 'sku': sku, 'promo_start_year': start_y, 'promo_start_week': start_w, 'promo_end_year': end_y, 'promo_end_week': end_w, 'n_promo_weeks': n_promo, 'qty_before_avg': round(qty_before, 2) if qty_before is not None else None, 'qty_during_avg': round(qty_during, 2) if qty_during is not None else None, 'qty_after_avg': round(qty_after, 2) if qty_after is not None else None, 'actual_uplift': round(uplift, 3) if uplift is not None else None, 'cannibalization': round(cann, 3) if cann is not None else None, 'net_effect': round(net, 3) if net is not None else None, 'promo_types': promo_types_combined, 'cat': cat_map.get(sku, ''), 'oznaka': ozn_map.get(sku, ''), }) out_df = pd.DataFrame(out_rows, columns=OUTPUT_COLS) out_df = out_df.sort_values(['promo_start_year', 'promo_start_week', 'sku']).reset_index(drop=True) out_df.to_csv(OUT_PATH, index=False, encoding='utf-8') # Quick summary print n_events = len(out_df) n_skus = out_df['sku'].nunique() if n_events else 0 n_with_uplift = int(out_df['actual_uplift'].notna().sum()) if n_events else 0 print(f'\n Written: {OUT_PATH.name}') print(f' {n_events:,} promo events | {n_skus:,} SKUs') print(f' {n_with_uplift:,} events with computable uplift (BEFORE window had data)') if n_with_uplift > 0: sub = out_df.dropna(subset=['actual_uplift']) print(f' Median uplift across all events: {sub["actual_uplift"].median():.2f}') print(f' Median cannibalization: {sub["cannibalization"].dropna().median():.2f}') print(f' Median net_effect: {sub["net_effect"].dropna().median():.2f}') if __name__ == '__main__': try: main() except Exception as e: print(f'\n ERROR: {e}') import traceback traceback.print_exc()