"""Promo Performance โ€” past-campaign uplift / cannibalization / net effect. Reads `../data/promo_performance.csv` (rebuilt weekly by `build_promo_performance.py`, which is chained into update_sales.py). This page shows what already happened โ€” not predictions. """ from __future__ import annotations from pathlib import Path import pandas as pd import plotly.express as px import streamlit as st DATA_DIR = Path(__file__).parent.parent / "data" PROMO_PERF_FILE = DATA_DIR / "promo_performance.csv" @st.cache_data(ttl=300) def _load() -> pd.DataFrame | None: if not PROMO_PERF_FILE.exists(): return None df = pd.read_csv(PROMO_PERF_FILE, dtype={"sku": str}) if len(df) == 0: return df # ISO-week int key for sorting / range filters df["promo_start_yw"] = (df["promo_start_year"].astype(int) * 100 + df["promo_start_week"].astype(int)) return df def render(): st.title("๐Ÿ“Š Promo Performance") st.caption( "What past promotions actually did โ€” uplift during, cannibalization " "after, and net effect over the campaign + 4-week recovery window. " "Source: `data/promo_performance.csv` (rebuilt weekly from " "`erp_promo_calendar` + `sales_clean`)." ) df = _load() if df is None: st.warning( "`data/promo_performance.csv` not found yet. Run " "`python build_promo_performance.py` in the project root, " "or wait for the next weekly `update_sales.py` cycle." ) return if len(df) == 0: st.info("No promo events in the file yet.") return # ---- Filters ---- f1, f2, f3, f4 = st.columns([2, 2, 2, 2]) all_cats = sorted(df["cat"].dropna().astype(str).unique().tolist()) cat_pick = f1.multiselect("Category", all_cats, default=all_cats, key="pp_cat") all_ozn = sorted(df["oznaka"].dropna().astype(str).unique().tolist()) ozn_pick = f2.multiselect("Tier (oznaka)", all_ozn, default=all_ozn, key="pp_ozn") yw_min = int(df["promo_start_yw"].min()) yw_max = int(df["promo_start_yw"].max()) yw_pick = f3.slider( "Promo start range (year*100 + week)", min_value=yw_min, max_value=yw_max, value=(yw_min, yw_max), key="pp_yw", help="Filter campaigns by their start (year*100 + ISO week).", ) max_weeks = f4.slider( "Max promo duration (weeks)", min_value=1, max_value=52, value=13, key="pp_max_weeks", help="ERP marks loyalty/manufacturer permanent prices as 'promo' too โ€” " "those show up as long contiguous runs (often 40+ weeks). Default " "13 keeps a quarter's worth of real campaigns and hides the rest.", ) view = df[ (df["cat"].astype(str).isin(cat_pick) if cat_pick else True) & (df["oznaka"].astype(str).isin(ozn_pick) if ozn_pick else True) & (df["promo_start_yw"] >= yw_pick[0]) & (df["promo_start_yw"] <= yw_pick[1]) & (df["n_promo_weeks"] <= max_weeks) ].copy() if len(view) == 0: st.info("No campaigns match the current filters.") return # ---- Summary metrics ---- n_events = len(view) n_skus = view["sku"].nunique() sub = view.dropna(subset=["actual_uplift"]) med_up = sub["actual_uplift"].median() if len(sub) else float("nan") med_cann = view["cannibalization"].dropna().median() med_net = view["net_effect"].dropna().median() m1, m2, m3, m4, m5 = st.columns(5) m1.metric("Campaigns", f"{n_events:,}") m2.metric("SKUs", f"{n_skus:,}") m3.metric("Median uplift", f"{med_up:.2f}x" if pd.notna(med_up) else "โ€”", help="DURING / BEFORE. 2.0x = doubled the weekly run-rate.") m4.metric("Median cannibal.", f"{med_cann:.2f}x" if pd.notna(med_cann) else "โ€”", help="AFTER / BEFORE. <1.0 = post-promo dip. ~1.0 = no dip.") m5.metric("Median net effect", f"{med_net:.2f}x" if pd.notna(med_net) else "โ€”", help="(volume during + 4w after) / (BEFORE * total weeks). " ">1.0 = campaign added real demand. <1.0 = net cannibalization.") st.divider() # ---- Category-level breakdown ---- st.subheader("By category") cat_brk = (view.groupby("cat", dropna=False) .agg(n_campaigns=("sku", "size"), n_skus=("sku", "nunique"), median_uplift=("actual_uplift", "median"), median_cannibalization=("cannibalization", "median"), median_net_effect=("net_effect", "median")) .reset_index() .sort_values("n_campaigns", ascending=False)) cat_brk.columns = ["Category", "Campaigns", "SKUs", "Median uplift", "Median cannibalization", "Median net effect"] for c in ("Median uplift", "Median cannibalization", "Median net effect"): cat_brk[c] = cat_brk[c].round(2) st.dataframe(cat_brk, use_container_width=True, hide_index=True) st.divider() # ---- Scatter: uplift vs cannibalization, one dot per campaign ---- st.subheader("Uplift vs Cannibalization") st.caption( "One dot = one campaign. Top-right quadrant = high uplift WITH " "no post-promo dip (the unicorns). Bottom-left = small uplift " "and post-promo dip (avoid). Hover for SKU + dates." ) plot_df = view.dropna(subset=["actual_uplift", "cannibalization"]).copy() if len(plot_df) == 0: st.info("Not enough data with both uplift and cannibalization to plot.") else: plot_df["campaign"] = (plot_df["sku"].astype(str) + " ยท CW" + plot_df["promo_start_week"].astype(str) + "/" + plot_df["promo_start_year"].astype(str)) fig = px.scatter( plot_df, x="actual_uplift", y="cannibalization", color="oznaka" if plot_df["oznaka"].notna().any() else None, size="n_promo_weeks", hover_data=["sku", "cat", "promo_start_year", "promo_start_week", "promo_end_year", "promo_end_week", "n_promo_weeks", "qty_before_avg", "qty_during_avg", "qty_after_avg", "net_effect", "promo_types"], labels={ "actual_uplift": "Uplift (DURING / BEFORE)", "cannibalization": "Cannibalization (AFTER / BEFORE)", "n_promo_weeks": "Weeks", }, title=None, ) # Reference lines at 1.0 (= no change) fig.add_hline(y=1.0, line_dash="dot", line_color="gray", opacity=0.5, annotation_text="no post-promo dip", annotation_position="right") fig.add_vline(x=1.0, line_dash="dot", line_color="gray", opacity=0.5, annotation_text="no uplift", annotation_position="top") fig.update_layout(height=520) st.plotly_chart(fig, use_container_width=True) st.divider() # ---- Full table + download ---- st.subheader("All campaigns in scope") show = view.sort_values(["promo_start_yw", "sku"], ascending=[False, True]).copy() show["campaign_start"] = ("CW" + show["promo_start_week"].astype(str) + "/" + show["promo_start_year"].astype(str)) show["campaign_end"] = ("CW" + show["promo_end_week"].astype(str) + "/" + show["promo_end_year"].astype(str)) cols = [ "sku", "cat", "oznaka", "campaign_start", "campaign_end", "n_promo_weeks", "qty_before_avg", "qty_during_avg", "qty_after_avg", "actual_uplift", "cannibalization", "net_effect", "promo_types", ] st.dataframe(show[cols], use_container_width=True, hide_index=True) st.download_button( "โฌ‡๏ธ Download CSV", view.to_csv(index=False).encode("utf-8"), file_name="promo_performance_filtered.csv", mime="text/csv", key="pp_dl", )