"""Validate the Promo Tool's cannibalization model against historical data. For a given promo (SKU + period), reconstructs the observed sibling cannibalization on retail+web channel using the clean baseline methodology, then prints the predicted vs. observed delta. Use this to fine-tune cann_rate constants in PromoTool/page_planner.py as more real promo data accumulates. Usage: py validate_promo_model.py POL12880 2026 19 2 # single SKU py validate_promo_model.py POL12880,POL12881 2026 19 2 # basket py validate_promo_model.py --auto # auto-detect # recent promos """ from __future__ import annotations import argparse import sys from pathlib import Path import pandas as pd ROOT = Path(__file__).parent DATA = ROOT / "data" # Channel mix — RCM + WSA/B/C cover retail + web (per CM clarification). # Wholesale is excluded since it's a different ERP feed entirely. DISC_PCT_THRESHOLD = 5.0 BASELINE_LAST_N = 20 BASELINE_LOOKBACK = 52 OOS_FLOOR_MULT = 0.30 def load_data(): plan = pd.read_csv(DATA / "sku_plan_list.csv") prices = pd.read_csv(DATA / "sku_prices.csv") costs = pd.read_csv(DATA / "sku_costs.csv") sales = pd.read_csv(DATA / "sales_clean.csv") sales["yw"] = sales["year"].astype(int) * 100 + sales["week"].astype(int) erp = pd.read_csv(DATA / "erp_promo_calendar.csv") erp["yw"] = erp["year"].astype(int) * 100 + erp["week"].astype(int) return { "plan": plan, "prices": prices, "costs": costs, "sales": sales, "erp": erp, "price_map": dict(zip(prices["sku"].astype(str), pd.to_numeric(prices["normal_retail_ppp"], errors="coerce").fillna(0))), "cost_map": dict(zip(costs["sku"].astype(str), pd.to_numeric(costs["cost_price"], errors="coerce").fillna(0))), "name_map": dict(zip(plan["sku"].astype(str), plan["name"])), "cat_map": dict(zip(plan["sku"].astype(str), plan["cat"])), "erp_promo_yws": { (str(r["sku"]), int(r["yw"])) for _, r in erp[erp["is_erp_promo"] == 1].iterrows() }, } def clean_baseline(d: dict, sku: str, before_yw: int) -> dict: """Same logic as updated base_run_rate in PromoTool: filter out ERP promo / bleed / flag / discount-anomaly / zero / OOS weeks. Returns retail+web mean per week from up to BASELINE_LAST_N clean.""" sub = d["sales"][(d["sales"]["sku"] == sku) & (d["sales"]["yw"] < before_yw)].sort_values("yw", ascending=False) sub = sub.head(BASELINE_LOOKBACK) # Bleed window bleed_yws = set() for (s, yw) in d["erp_promo_yws"]: if s == sku: bleed_yws.add(yw + 1) bleed_yws.add(yw + 2) candidates = [] excluded = {"erp": 0, "bleed": 0, "flag": 0, "disc": 0, "zero": 0} for _, r in sub.iterrows(): yw = int(r["yw"]) if (sku, yw) in d["erp_promo_yws"]: excluded["erp"] += 1 continue if yw in bleed_yws: excluded["bleed"] += 1 continue if int(r.get("is_any_promo", 0) or 0): excluded["flag"] += 1 continue rd = abs(float(r.get("retail_discount_pct", 0) or 0)) wd = abs(float(r.get("webshop_discount_pct", 0) or 0)) if rd > DISC_PCT_THRESHOLD or wd > DISC_PCT_THRESHOLD: excluded["disc"] += 1 continue q = float(r["qty_retail"] or 0) + float(r["qty_webshop"] or 0) if q <= 0: excluded["zero"] += 1 continue candidates.append((yw, q)) if len(candidates) >= BASELINE_LAST_N * 2: break # OOS filter on remaining if candidates: prov_med = pd.Series([q for _, q in candidates]).median() threshold = max(1.0, OOS_FLOOR_MULT * prov_med) clean = [(yw, q) for (yw, q) in candidates if q >= threshold][:BASELINE_LAST_N] else: clean = [] qty = pd.Series([q for _, q in clean]) return { "avg": float(qty.mean()) if len(qty) else 0.0, "median": float(qty.median()) if len(qty) else 0.0, "n": len(clean), "excluded": excluded, } def observed_actuals(d: dict, sku: str, period_yws: list[int]) -> dict: """Actual retail+web quantity and revenue during the promo period. Uses sales_clean (which already has the channel split).""" sub = d["sales"][(d["sales"]["sku"] == sku) & (d["sales"]["yw"].isin(period_yws))] qty = float((sub["qty_retail"].fillna(0) + sub["qty_webshop"].fillna(0)).sum()) # Revenue: use observed avg_ppp × qty for each channel rev_retail = float((sub["qty_retail"].fillna(0) * sub["avg_ppp_retail"].fillna(0)).sum()) rev_web = float((sub["qty_webshop"].fillna(0) * sub["avg_ppp_webshop"].fillna(0)).sum()) return {"qty": qty, "rev": rev_retail + rev_web} def predict_via_model(d: dict, basket: list[str], period_yws: list[int], discount_pct: float) -> dict: """Replicate the Promo Tool's cannibalization model for this promo so we can compare predicted vs observed. NOTE: this is a SIMPLIFIED standalone version of what page_planner does — same rates, same weighting principle, but doesn't pull in proximity/flavor/etc (which would require the full planner state). For directional validation it's enough. """ # Calibrated rates (mirrors page_planner.py post-fix) if discount_pct <= 15: cann_rate = 0.05 elif discount_pct <= 25: cann_rate = 0.08 elif discount_pct <= 35: cann_rate = 0.12 elif discount_pct <= 45: cann_rate = 0.18 else: cann_rate = 0.25 # Assume price-disruptor for simplicity (would be derived in full model) cann_rate += 0.05 n_weeks = len(period_yws) total_incremental = 0.0 for sku in basket: bl = clean_baseline(d, sku, min(period_yws)) actual = observed_actuals(d, sku, period_yws) expected = bl["avg"] * n_weeks total_incremental += max(0.0, actual["qty"] - expected) total_cann_units_predicted = total_incremental * cann_rate return { "cann_rate": cann_rate, "total_incremental": total_incremental, "predicted_cann_units": total_cann_units_predicted, } def find_siblings(d: dict, basket_skus: list[str], top_n: int = 15) -> list[str]: """Same sub-cat-based sibling identification as planner. Pulls from sku_subcat_map.csv. Filters out basket SKUs.""" subcat_path = DATA / "sku_subcat_map.csv" if not subcat_path.exists(): return [] sm = pd.read_csv(subcat_path) sm["sku"] = sm["sku"].astype(str) if "sub_cat" not in sm.columns: return [] sub_cats = set() for sku in basket_skus: m = sm[sm["sku"] == sku] if not m.empty: sub_cats.add(str(m.iloc[0]["sub_cat"])) if not sub_cats: return [] sib = sm[sm["sub_cat"].isin(sub_cats) & ~sm["sku"].isin(basket_skus)] return list(sib["sku"].astype(str).head(top_n)) def measure_observed_cannibalization(d: dict, sibling_skus: list[str], period_yws: list[int]) -> dict: """Sum up observed sibling drop on retail+web. Per sibling: (baseline × n_weeks) - actual qty = lost units Negative lost_units (i.e. went UP) is allowed; we report it. """ n_weeks = len(period_yws) total_lost_units = 0.0 total_lost_rev = 0.0 rows = [] for sku in sibling_skus: bl = clean_baseline(d, sku, min(period_yws)) if bl["n"] < 3: continue actual = observed_actuals(d, sku, period_yws) expected_qty = bl["avg"] * n_weeks expected_rev_per_unit = d["price_map"].get(sku, 0.0) # Use sales_clean's channel-blended revenue per unit if available sub = d["sales"][(d["sales"]["sku"] == sku) & (d["sales"]["yw"].isin(period_yws))] delta_qty = expected_qty - actual["qty"] delta_rev = delta_qty * expected_rev_per_unit total_lost_units += delta_qty total_lost_rev += delta_rev rows.append({ "sku": sku, "name": d["name_map"].get(sku, "")[:40], "baseline_wk": round(bl["avg"], 1), "expected": round(expected_qty, 1), "actual": round(actual["qty"], 1), "delta_qty": round(delta_qty, 1), "delta_rev_eur": round(delta_rev, 0), }) return { "total_lost_units": total_lost_units, "total_lost_rev_eur": total_lost_rev, "per_sibling": rows, } def parse_args(): ap = argparse.ArgumentParser( description="Validate promo cannibalization model vs real history.") ap.add_argument("basket", nargs="?", default="", help="comma-separated SKU codes (e.g. POL12880,POL12881)") ap.add_argument("start_year", nargs="?", type=int, default=0) ap.add_argument("start_week", nargs="?", type=int, default=0) ap.add_argument("n_weeks", nargs="?", type=int, default=2) ap.add_argument("--discount", type=float, default=33.0, help="discount percent applied during promo (default 33)") ap.add_argument("--auto", action="store_true", help="auto-detect recent ERP promos and validate each") return ap.parse_args() def run_validation(d, basket_skus, period_yws, discount): print(f"\n{'='*78}") print(f"PROMO: {','.join(basket_skus)}") print(f"PERIOD: {[f'CW{y%100}' for y in period_yws]} " f"({len(period_yws)} weeks, year {period_yws[0]//100})") print(f"DISCOUNT: {discount}%") print('='*78) # Sanity check — do we have data for this period in sales_clean? max_yw_in_data = int(d["sales"]["yw"].max()) if min(period_yws) > max_yw_in_data: print(f"\n⚠️ WARN: sales_clean.csv has data only up to CW" f"{max_yw_in_data%100}/{max_yw_in_data//100}. " f"The requested period {period_yws[0]//100}.W{period_yws[0]%100} " "is in the future relative to that file. ") print(" For a CURRENT-week promo, validate against a separate ERP " "export (like analyze_nextgen_promo.py does with " "Rekapitulacijazaakciju.xlsx + akcijaaut.xlsx + akcijaslo.xlsx).") print(" For HISTORICAL promos (where post-promo data exists), pick " "a period further back. Try --auto to list recent windows.") return None # 1. Basket actuals for sku in basket_skus: bl = clean_baseline(d, sku, min(period_yws)) actual = observed_actuals(d, sku, period_yws) expected = bl["avg"] * len(period_yws) print(f"\n PROMO SKU: {sku} — {d['name_map'].get(sku, '')[:50]}") print(f" baseline/wk: {bl['avg']:6.1f} (n_clean={bl['n']})") print(f" expected qty: {expected:6.1f} actual: {actual['qty']:6.1f} " f"INCREMENTAL: {(actual['qty']-expected):+6.1f}") # 2. Model prediction pred = predict_via_model(d, basket_skus, period_yws, discount) print(f"\n MODEL PREDICTS:") print(f" incremental units (basket): {pred['total_incremental']:.0f}") print(f" cann rate applied: {pred['cann_rate']*100:.0f}%") print(f" predicted cann units: {pred['predicted_cann_units']:.0f}") # 3. Observed sibling cannibalization sibs = find_siblings(d, basket_skus, top_n=20) obs = measure_observed_cannibalization(d, sibs, period_yws) print(f"\n OBSERVED on {len(obs['per_sibling'])} sub-cat siblings:") print(f" {'SKU':10s} {'Name':42s} {'BL/wk':>6s} {'Exp':>5s} {'Act':>5s} " f"{'ΔQ':>5s} {'Δ€':>8s}") for r in sorted(obs["per_sibling"], key=lambda x: -x["delta_rev_eur"])[:15]: print(f" {r['sku']:10s} {r['name']:42s} " f"{r['baseline_wk']:>6.1f} {r['expected']:>5.1f} {r['actual']:>5.1f} " f"{r['delta_qty']:>+5.1f} {r['delta_rev_eur']:>+8,.0f}") print(f"\n Total observed lost units: {obs['total_lost_units']:.1f}") print(f" Total observed lost rev €: {obs['total_lost_rev_eur']:+,.0f}") # 4. Verdict obs_units = obs["total_lost_units"] pred_units = pred["predicted_cann_units"] if pred_units > 0: err = (pred_units - obs_units) / pred_units * 100 flag = ("MODEL OVER-PREDICTS" if err > 30 else ("MODEL UNDER-PREDICTS" if err < -30 else "MODEL OK")) else: err = 0.0 flag = "n/a" print(f"\n VERDICT: predicted {pred_units:.0f}u vs observed {obs_units:.0f}u " f"(error {err:+.0f}% → {flag})") print() return {"predicted": pred_units, "observed": obs_units, "flag": flag} def auto_detect_recent_promos(d: dict, n: int = 5) -> list[tuple]: """Find the most recent N distinct promo windows in erp_promo_calendar.""" erp = d["erp"][d["erp"]["is_erp_promo"] == 1].copy() erp["promo_id"] = erp["promo_types"].fillna("?") + "_" + erp["year"].astype(str) out = [] for pid, grp in erp.groupby("promo_id"): yws = sorted(set(grp["yw"].astype(int))) if not yws: continue skus = list(grp["sku"].astype(str).unique())[:3] out.append({ "id": pid, "yws": yws, "skus": skus, "n_skus_total": int(grp["sku"].nunique()), }) out.sort(key=lambda x: -max(x["yws"])) return out[:n] # ============================================================ def main(): args = parse_args() d = load_data() if args.auto: recent = auto_detect_recent_promos(d, n=5) print(f"Auto-detected {len(recent)} recent promo windows:") for p in recent: print(f"\n Promo: {p['id']} · weeks: {[w%100 for w in p['yws']]} " f"· SKUs: {p['n_skus_total']} (showing 3: {p['skus']})") return if not args.basket or not args.start_year: print(__doc__) sys.exit(1) basket = [s.strip() for s in args.basket.split(",") if s.strip()] period_yws = [args.start_year * 100 + args.start_week + i for i in range(args.n_weeks)] run_validation(d, basket, period_yws, args.discount) if __name__ == "__main__": main()