"""Calibration analysis for the Promo Tool uplift engine. Pulls real data from data/ to answer two questions: Q1. What does the €/100g distribution look like inside each category? → calibrates the price_disruptor_multiplier thresholds. Q2. What discount bands historically produce what uplift multiples? → calibrates discount_band_uplift_curve. → identifies "magic prices" (promo €/100g that triggered top-decile uplifts). Run: py analyze_promo_calibration.py Output: stdout only — no files written. """ from __future__ import annotations import re from pathlib import Path import numpy as np import pandas as pd DATA = Path(__file__).parent / "data" # ============================================================ # Helpers # ============================================================ _SIZE_PATTERNS = [ (re.compile(r"(\d+(?:[.,]\d+)?)\s*kg\b", re.I), 1000.0, "g"), (re.compile(r"(\d+(?:[.,]\d+)?)\s*g\b", re.I), 1.0, "g"), (re.compile(r"(\d+(?:[.,]\d+)?)\s*l\b", re.I), 1000.0, "ml"), (re.compile(r"(\d+(?:[.,]\d+)?)\s*ml\b", re.I), 1.0, "ml"), ] def parse_size(name: str) -> tuple[float, str]: if not isinstance(name, str): return 0.0, "" for pat, scale, unit in _SIZE_PATTERNS: m = pat.search(name) if m: try: return float(m.group(1).replace(",", ".")) * scale, unit except ValueError: continue return 0.0, "" def pct(x): return f"{x:.2f}" # ============================================================ # Load # ============================================================ plan = pd.read_csv(DATA / "sku_plan_list.csv") prices = pd.read_csv(DATA / "sku_prices.csv") sales = pd.read_csv(DATA / "sales_clean.csv") erp = pd.read_csv(DATA / "erp_promo_calendar.csv") plan["sku"] = plan["sku"].astype(str) prices["sku"] = prices["sku"].astype(str) sales["sku"] = sales["sku"].astype(str) erp["sku"] = erp["sku"].astype(str) # Parse size per plan SKU plan["size_val"], plan["size_unit"] = zip(*plan["name"].map(parse_size)) # Attach list price (prefer normal_retail_ppp, fall back to vpc) price_map = dict(zip(prices["sku"], prices["normal_retail_ppp"].fillna(0))) plan["price"] = plan["sku"].map(price_map).fillna(plan["vpc"]).astype(float) plan["per100"] = np.where( (plan["size_val"] > 0) & (plan["price"] > 0), plan["price"] * 100.0 / plan["size_val"], np.nan, ) print("=" * 78) print("Q1 · CATEGORY €/100g (or €/100ml) BENCHMARKS — list-price basis") print("=" * 78) print("(only counts SKUs in plan list whose name yields a parseable size)\n") # Group by (cat, size_unit) — solids vs liquids are separate cohorts groups = [] for (cat, unit), sub in plan.groupby(["cat", "size_unit"]): pts = sub["per100"].dropna() if len(pts) < 5: continue groups.append({ "cat": cat, "unit": unit, "n_skus": len(pts), "min": float(pts.min()), "p25": float(pts.quantile(0.25)), "median": float(pts.median()), "p75": float(pts.quantile(0.75)), "max": float(pts.max()), }) grp_df = pd.DataFrame(groups).sort_values(["unit", "cat"]) print(grp_df.to_string( index=False, formatters={c: (lambda v: f"{v:7.2f}") for c in ["min", "p25", "median", "p75", "max"]}, )) # Save for downstream use cat_bench = {(r["cat"], r["unit"]): r for r in groups} # ============================================================ # Q2 · Discount band → uplift # Build per (sku, contiguous promo window) events from erp_promo_calendar # and join sales_clean to compute uplift. # ============================================================ print("\n" + "=" * 78) print("Q2 · HISTORICAL UPLIFT BY DISCOUNT BAND") print("=" * 78) # Compute year-week int sales["yw"] = sales["year"].astype(int) * 100 + sales["week"].astype(int) erp["yw"] = erp["year"].astype(int) * 100 + erp["week"].astype(int) # Each (sku, year) contiguous promo cluster → one event window. events = [] for sku, sdf in erp[erp["is_erp_promo"] == 1].sort_values(["sku", "yw"]).groupby("sku"): yws = sorted(set(zip(sdf["year"].astype(int), sdf["week"].astype(int)))) if not yws: continue cluster = [yws[0]] clusters = [] for prev, cur in zip(yws, yws[1:]): # contiguous if same year + consecutive week, OR year boundary if (cur[0] == prev[0] and cur[1] == prev[1] + 1) or \ (cur[0] == prev[0] + 1 and prev[1] >= 52 and cur[1] == 1): cluster.append(cur) else: clusters.append(cluster) cluster = [cur] clusters.append(cluster) for c in clusters: if len(c) < 2 or len(c) > 6: # AKCIJA = 2-6 weeks; skip 1-week noise / long stuff continue events.append({"sku": sku, "weeks": c}) print(f"Found {len(events):,} candidate AKCIJA windows (2-6 wk contiguous).") # For each event, compute baseline & promo uplift rows = [] sales_by_sku = {sku: g for sku, g in sales.groupby("sku")} for ev in events: sku = ev["sku"] if sku not in sales_by_sku: continue s = sales_by_sku[sku] weeks = ev["weeks"] yws_promo = [y * 100 + w for (y, w) in weeks] yw_start = min(yws_promo) yw_end = max(yws_promo) promo = s[s["yw"].isin(yws_promo)] if promo.empty: continue promo_qty = float((promo["qty_retail"].fillna(0) + promo["qty_webshop"].fillna(0)).sum()) promo_weeks = len(promo) if promo_qty <= 0 or promo_weeks == 0: continue avg_promo = promo_qty / promo_weeks # Baseline: 8 weeks BEFORE promo, non-promo only (use is_any_promo flag) pre = s[(s["yw"] < yw_start)].sort_values("yw").tail(12) pre_clean = pre[pre["is_any_promo"] == 0] if len(pre_clean) < 3: continue pre_qty_w = (pre_clean["qty_retail"].fillna(0) + pre_clean["qty_webshop"].fillna(0)) if pre_qty_w.median() <= 0: continue base_avg = float(pre_qty_w.median()) uplift = avg_promo / base_avg # Discount during promo (avg of retail discount pct over promo weeks) disc = float(promo["retail_discount_pct"].replace([np.inf, -np.inf], np.nan).fillna(0).mean()) rows.append({ "sku": sku, "uplift": uplift, "discount": disc, "promo_weeks": promo_weeks, "avg_promo": avg_promo, "base_avg": base_avg, "yw_start": yw_start, }) evdf = pd.DataFrame(rows) print(f"Measurable events: {len(evdf):,} (uplift computable + clean baseline)") print() # Bin by discount bins = [0, 10, 15, 25, 35, 45, 60, 200] labels = ["0-10%", "10-15%", "15-25%", "25-35%", "35-45%", "45-60%", "60%+"] evdf["band"] = pd.cut(evdf["discount"], bins=bins, labels=labels, right=False) print("Discount band → uplift distribution:") band_stats = evdf.groupby("band", observed=True).agg( n=("uplift", "size"), p25=("uplift", lambda x: x.quantile(0.25)), median=("uplift", "median"), p75=("uplift", lambda x: x.quantile(0.75)), p90=("uplift", lambda x: x.quantile(0.90)), max=("uplift", "max"), ).round(2) print(band_stats.to_string()) # ============================================================ # Q2b · Top-decile uplift events → "magic prices" # ============================================================ print("\n" + "=" * 78) print("Q2b · TOP UPLIFT EVENTS — their promo €/100u") print("=" * 78) # Attach plan info evdf = evdf.merge( plan[["sku", "name", "cat", "size_val", "size_unit", "price"]], on="sku", how="left" ) # Compute promo €/100unit evdf["promo_price"] = evdf["price"] * (1 - evdf["discount"] / 100) evdf["promo_per100"] = np.where( (evdf["size_val"] > 0) & (evdf["promo_price"] > 0), evdf["promo_price"] * 100 / evdf["size_val"], np.nan, ) # Attach category benchmark def _bench_for(row): k = (row["cat"], row["size_unit"]) if k in cat_bench: b = cat_bench[k] return b["p25"], b["median"] return np.nan, np.nan evdf[["cat_p25", "cat_median"]] = evdf.apply( lambda r: pd.Series(_bench_for(r)), axis=1) evdf["price_vs_p25"] = evdf["promo_per100"] / evdf["cat_p25"] # Top decile uplift threshold = evdf["uplift"].quantile(0.90) top10 = evdf[evdf["uplift"] >= threshold].copy().sort_values("uplift", ascending=False) print(f"Top decile threshold: uplift ≥ {threshold:.2f}× ({len(top10)} events)") print() print("Top 20 events:") show = top10.head(20)[["sku", "cat", "name", "discount", "uplift", "size_val", "size_unit", "promo_per100", "cat_p25", "cat_median", "price_vs_p25"]] show = show.assign( discount=show["discount"].round(1), uplift=show["uplift"].round(2), size_val=show["size_val"].astype(int, errors="ignore"), promo_per100=show["promo_per100"].round(2), cat_p25=show["cat_p25"].round(2), cat_median=show["cat_median"].round(2), price_vs_p25=show["price_vs_p25"].round(2), ) print(show.to_string(index=False)) # ============================================================ # Q2c · Aggregate by price-vs-p25 zone # ============================================================ print("\n" + "=" * 78) print("Q2c · UPLIFT vs PRICE POSITION (€/100u relative to cat p25)") print("=" * 78) z_evdf = evdf.dropna(subset=["price_vs_p25"]).copy() zone_bins = [0, 0.5, 0.7, 1.0, 1.3, 1.7, 10] zone_labels = ["<0.5× p25", "0.5–0.7×", "0.7–1.0×", "1.0–1.3×", "1.3–1.7×", "≥1.7×"] z_evdf["zone"] = pd.cut(z_evdf["price_vs_p25"], bins=zone_bins, labels=zone_labels, right=False) zone_stats = z_evdf.groupby("zone", observed=True).agg( n=("uplift", "size"), median_uplift=("uplift", "median"), p75_uplift=("uplift", lambda x: x.quantile(0.75)), p90_uplift=("uplift", lambda x: x.quantile(0.90)), avg_discount=("discount", "mean"), ).round(2) print(zone_stats.to_string()) # ============================================================ # Q3 · Magic price thresholds per category # ============================================================ print("\n" + "=" * 78) print("Q3 · CATEGORY MAGIC-PRICE THRESHOLDS") print("=" * 78) print("(Looking at events with uplift ≥ 2× — what promo €/100u they sit at)\n") ABOVE = evdf[evdf["uplift"] >= 2.0].dropna(subset=["promo_per100"]).copy() mag = ABOVE.groupby(["cat", "size_unit"]).agg( n=("uplift", "size"), median_per100=("promo_per100", "median"), p25_per100=("promo_per100", lambda x: x.quantile(0.25)), median_disc=("discount", "median"), median_uplift=("uplift", "median"), ).round(2).sort_values("n", ascending=False) print(mag.to_string()) # ============================================================ # Q4 · Calibration suggestions # ============================================================ print("\n" + "=" * 78) print("Q4 · MODEL CALIBRATION READOUT") print("=" * 78) print(""" Current promo_data.py constants vs. observed data: discount_band_uplift_curve(d): Observed medians (n>5) should align with the curve below; if median is materially higher, raise the curve in that band. """) print("Discount band → observed median uplift → current curve value") for band, low, high in [("0-15%", 0, 15), ("15-25%", 15, 25), ("25-35%", 25, 35), ("35-45%", 35, 45), ("45-60%", 45, 60)]: sub = evdf[(evdf["discount"] >= low) & (evdf["discount"] < high)] if len(sub) < 5: continue obs = sub["uplift"].median() # current calibrated curve midpoint (mirror of promo_data.py) mid = (low + high) / 2 if mid <= 15: cur = 0.85 + (mid / 15) * 0.20 elif mid <= 25: cur = 1.05 + ((mid - 15) / 10) * 0.45 elif mid <= 35: cur = 1.50 + ((mid - 25) / 10) * 0.60 elif mid <= 45: cur = 2.10 + ((mid - 35) / 10) * 0.40 elif mid <= 60: cur = 2.50 + ((mid - 45) / 15) * 1.00 else: cur = min(5.00, 3.50 + (mid - 60) / 40 * 1.50) flag = "" if obs / cur < 1.6 else " ← model UNDER-predicts" print(f" {band:8s} n={len(sub):3d} obs median {obs:.2f}× curve {cur:.2f}×{flag}") print(""" price_disruptor_multiplier observed effect: """) if not z_evdf.empty: deep = z_evdf[z_evdf["zone"].astype(str) == "<0.5× p25"] if len(deep) >= 3: print(f" Events at < 0.5× cat p25: n={len(deep)}, " f"median uplift {deep['uplift'].median():.2f}×, " f"p90 {deep['uplift'].quantile(0.9):.2f}×") deep2 = z_evdf[z_evdf["zone"].astype(str) == "0.5–0.7×"] if len(deep2) >= 3: print(f" Events at 0.5–0.7× cat p25: n={len(deep2)}, " f"median uplift {deep2['uplift'].median():.2f}×, " f"p90 {deep2['uplift'].quantile(0.9):.2f}×") print("\nDone.")