"""Past Promotions browser — same dark dashboard view as the static HTML
preview, but live in Streamlit (no rebuild needed)."""
from __future__ import annotations
from datetime import datetime
import pandas as pd
import streamlit as st
import plotly.graph_objects as go
from promo_data import load_all
def _build_campaigns(erp, sales, plan, prices_map, ruc_unit_map,
name_map, cat_map, tier_map):
"""Build past-campaign list. Revenue & RUC are computed from the
ACTUAL sales_clean numbers for those weeks (real prices booked,
not current list price). Retail + webshop channels only —
wholesale is excluded per CM direction."""
iso = datetime.now().isocalendar()
cur_y, cur_w = int(iso[0]), int(iso[1])
erp_p = erp[(erp["year"].astype(int) * 100 + erp["week"].astype(int))
< (cur_y * 100 + cur_w)].copy()
if len(erp_p) == 0:
return []
erp_p["primary"] = erp_p["promo_types"].astype(str).str.split(";").str[0].str.strip()
sales = sales.copy()
sales["yw"] = sales["year"].astype(int) * 100 + sales["week"].astype(int)
out = []
for (ptype, yr), g in erp_p.groupby(["primary", "year"]):
weeks = sorted(g["week"].astype(int).unique())
if not weeks:
continue
blocks = [[weeks[0]]]
for w in weeks[1:]:
(blocks[-1].append(w) if w - blocks[-1][-1] <= 6 else blocks.append([w]))
for blk in blocks:
sw, ew = blk[0], blk[-1]
duration = ew - sw + 1
# Removed min-duration filter — keep all campaigns regardless
# of length so single-week / short campaigns also appear.
skus = sorted(g[g["week"].astype(int).isin(blk)]["sku"].unique().tolist())
promo_yws = {yr * 100 + w for w in range(sw, ew + 1)}
base_yws = {yr * 100 + w for w in range(max(1, sw - 4), sw)}
sub_d = sales[sales["sku"].isin(skus) & sales["yw"].isin(promo_yws)]
sub_b = sales[sales["sku"].isin(skus) & sales["yw"].isin(base_yws)]
if sub_d.empty:
continue
sku_rows = []
for s in skus:
d = sub_d[sub_d["sku"] == s]
b = sub_b[sub_b["sku"] == s]
if d.empty:
continue
# Retail + web only — wholesale excluded
d_qty_retail = d["qty_retail"].fillna(0).astype(float)
d_qty_web = d["qty_webshop"].fillna(0).astype(float)
d_qty = float((d_qty_retail + d_qty_web).sum())
d_n = max(1, d["yw"].nunique())
b_qty_retail = b["qty_retail"].fillna(0).astype(float)
b_qty_web = b["qty_webshop"].fillna(0).astype(float)
b_qty = float((b_qty_retail + b_qty_web).sum()) if not b.empty else 0
b_n = max(1, b["yw"].nunique()) if not b.empty else 1
avg_d = d_qty / d_n
avg_b = b_qty / b_n if b_qty > 0 else 0
upl = (avg_d / avg_b) if avg_b > 0 else 0
# Revenue from sales_clean at HISTORICAL prices in those
# weeks — what actually got booked (post-discount).
d_rev_retail = float((d_qty_retail
* d["avg_ppp_retail"].fillna(0)).sum())
d_rev_web = float((d_qty_web
* d["avg_ppp_webshop"].fillna(0)).sum())
d_rev = d_rev_retail + d_rev_web
# RUC for those weeks — already pre-computed in sales_clean
d_ruc = float(d.get("ruc_retail", 0).fillna(0).sum()
+ d.get("ruc_webshop", 0).fillna(0).sum())
# Effective price per unit during promo (for transparency)
eff_price = d_rev / d_qty if d_qty > 0 else 0.0
sku_rows.append({
"sku": s, "name": (name_map.get(s, "") or "")[:60],
"cat": cat_map.get(s, "") or "",
"tier": tier_map.get(s, "") or "",
"weeks": int(d["yw"].nunique()),
"qty": int(d_qty),
"avg_base": round(avg_b, 1),
"avg_promo": round(avg_d, 1),
"uplift": round(upl, 2),
"rev": int(d_rev),
"ruc": int(d_ruc),
"eff_price": round(eff_price, 2),
})
if not sku_rows:
continue
sku_rows.sort(key=lambda r: r["rev"], reverse=True)
ups = [r["uplift"] for r in sku_rows if r["uplift"] > 0]
avg_upl = round(sum(ups) / len(ups), 2) if ups else 0
out.append({
"id": f"{yr}-{sw:02d}-{ptype[:30]}",
"type": ptype, "year": int(yr),
"start_week": sw, "end_week": ew,
"duration": duration,
"n_skus": len(sku_rows),
"total_qty": sum(r["qty"] for r in sku_rows),
"total_rev": sum(r["rev"] for r in sku_rows),
"total_ruc": sum(r["ruc"] for r in sku_rows),
"avg_uplift": avg_upl,
"skus": sku_rows,
})
out.sort(key=lambda c: (-c["year"], -c["end_week"]))
return out
@st.cache_data(ttl=300)
def _campaigns_cached(_data_keys):
"""Compute campaigns. _data_keys is a stable hash key for the input data."""
data = load_all()
return _build_campaigns(
data["erp"], data["sales"], data["plan"],
data["price_map"], data["ruc_unit_map"],
data["name_map"], data["cat_map"], data["tier_map"],
)
def render():
data = load_all()
if data["erp"].empty or data["sales"].empty:
st.error("Missing erp_promo_calendar.csv or sales_clean.csv in ../data/.")
return
campaigns = _campaigns_cached("v1")
# ---- SIDEBAR FILTERS ----
with st.sidebar:
st.markdown("### 📜 Past Promotions")
st.caption(f"{len(campaigns)} campaigns")
years = sorted({c["year"] for c in campaigns}, reverse=True)
year_pick = st.selectbox("Year", ["All"] + years, key="hist_year")
search = st.text_input("Search campaign", "", key="hist_search")
# ---- FILTER ----
flt = []
for c in campaigns:
if year_pick != "All" and c["year"] != year_pick:
continue
if search and search.lower() not in c["type"].lower():
continue
flt.append(c)
# ---- HEADER ----
st.markdown("## 📜 Past Promotions browser")
st.caption(f"{len(flt)} campaigns matching filters · pick one to inspect SKU-level performance")
if not flt:
st.info("No campaigns match the current filters.")
return
# ---- CAMPAIGN PICKER ----
labels = [
f"{c['type']} · CW{c['start_week']}–CW{c['end_week']} ({c['year']}) · "
f"{c['n_skus']} SKUs · {c['avg_uplift']}× upl · €{c['total_ruc']:,.0f} RUC"
for c in flt
]
pick_idx = st.selectbox("Campaign", range(len(flt)),
format_func=lambda i: labels[i], key="hist_pick")
c = flt[pick_idx]
# ---- CAMPAIGN DETAIL ----
st.markdown(f"### 📌 {c['type']} — {c['year']}")
st.caption(f"CW{c['start_week']} → CW{c['end_week']} · {c['duration']} weeks · "
f"{c['n_skus']} SKUs")
k1, k2, k3, k4, k5 = st.columns(5)
k1.metric("SKUs", f"{c['n_skus']:,}")
k2.metric("Total units", f"{c['total_qty']:,}")
k3.metric("Revenue", f"€{c['total_rev']:,.0f}")
k4.metric("RUC", f"€{c['total_ruc']:,.0f}")
k5.metric("Avg uplift", f"{c['avg_uplift']}×")
# SKU table — sorted by Revenue €, columns ordered by reporting importance
st.markdown("#### 📦 SKUs (sortirano po Revenue)")
df = pd.DataFrame(c["skus"])
df = df[["sku", "name", "cat", "tier", "weeks", "qty",
"avg_base", "avg_promo", "uplift", "eff_price", "rev", "ruc"]]
df.columns = [
"SKU", "Name", "Category", "Tier", "Weeks", "Units (retail+web)",
"Avg base/wk", "Avg promo/wk", "Uplift",
"Effective €/u", "Revenue €", "RUC €",
]
st.dataframe(df, use_container_width=True, hide_index=True)
# ---- Weekly performance line chart — all SKUs (filter down) ----
# Default = all SKUs shown as separate lines. Multi-select to
# narrow down to specific SKU(s). Time range = 4 weeks pre-promo +
# promo period + 2 weeks post-promo.
st.markdown("#### 📈 Weekly performance per SKU")
sku_labels_map = {
r["sku"]: f"{r['sku']} · {r['name'][:50]} · €{r['rev']:,.0f} rev"
for r in c["skus"]
}
all_sku_codes = [r["sku"] for r in c["skus"]]
selected_skus = st.multiselect(
"Odaberi artikle (default sve, suzi izborom)",
all_sku_codes,
default=all_sku_codes,
format_func=lambda s: sku_labels_map.get(s, s),
key=f"hist_sku_multi_{c['id']}",
)
if not selected_skus:
st.info("Odaberi barem jedan artikl.")
else:
sales = data["sales"].copy()
sales["yw"] = sales["year"].astype(int) * 100 + sales["week"].astype(int)
yr = c["year"]; sw = c["start_week"]; ew = c["end_week"]
pre_weeks = 4
post_weeks = 2
start_yw = yr * 100 + max(1, sw - pre_weeks)
end_yw = yr * 100 + ew + post_weeks
# Build union of all weeks across selected SKUs in the period
sub_all = sales[(sales["sku"].isin(selected_skus))
& (sales["yw"] >= start_yw)
& (sales["yw"] <= end_yw)]
if sub_all.empty:
st.info("Nema podataka za odabrane SKU-ove u periodu.")
else:
all_yws = sorted(sub_all["yw"].unique())
week_labels = [f"CW{int(yw) % 100}" for yw in all_yws]
from plotly.subplots import make_subplots
fig = make_subplots(specs=[[{"secondary_y": False}]])
PALETTE = ["#FF6B6B", "#38BDF8", "#FBBF24", "#4ADE80", "#A78BFA",
"#F472B6", "#34D399", "#FB923C", "#60A5FA", "#E8734A",
"#22D3EE", "#FCD34D", "#C084FC", "#10B981", "#F87171"]
# Solo mode (only 1 SKU) gets richer detail
solo = len(selected_skus) == 1
sku_row_by_id = {r["sku"]: r for r in c["skus"]}
for i, sku in enumerate(selected_skus):
sku_sales = sub_all[sub_all["sku"] == sku].sort_values("yw")
# Build full week-aligned series (fill missing weeks with 0)
qty_by_yw = {}
for _, r in sku_sales.iterrows():
yw = int(r["yw"])
q = float(r.get("qty_retail", 0) or 0) + float(r.get("qty_webshop", 0) or 0)
qty_by_yw[yw] = q
ys = [qty_by_yw.get(int(yw), 0) for yw in all_yws]
color = PALETTE[i % len(PALETTE)]
name_label = sku_row_by_id[sku]["name"][:30]
trace = go.Scatter(
x=week_labels, y=ys,
mode=("lines+markers+text" if solo else "lines+markers"),
line=dict(color=color, width=(3 if solo else 1.8)),
marker=dict(size=(9 if solo else 6)),
text=([f"{int(q)}" for q in ys] if solo else None),
textposition="top center",
textfont=dict(size=10, color="#E8E9ED"),
name=f"{sku} {name_label}",
hovertemplate=(f"{sku} {name_label}
"
"%{x}: %{y:.0f} u"),
)
fig.add_trace(trace)
# Baseline reference (only when solo — multiple baselines = clutter)
if solo:
base = sku_row_by_id[selected_skus[0]]["avg_base"]
if base > 0:
fig.add_hline(y=base, line_dash="dash", line_color="#9CA3AF",
annotation_text=f"Baseline {base:.0f}/wk",
annotation_position="top left",
annotation_font=dict(color="#9CA3AF", size=10))
# Shade promo period
try:
p_start = f"CW{sw}"
p_end = f"CW{ew}"
if p_start in week_labels and p_end in week_labels:
fig.add_vrect(
x0=p_start, x1=p_end,
fillcolor="rgba(255, 107, 107, 0.10)",
line_width=0,
annotation_text="🎯 PROMO PERIOD",
annotation_position="top",
annotation_font=dict(color="#FF6B6B", size=11),
)
except Exception:
pass
fig.update_layout(
plot_bgcolor="#1A1D27", paper_bgcolor="#1A1D27",
font=dict(color="#E8E9ED"),
height=max(420, 380 + (len(selected_skus) // 6) * 40),
margin=dict(l=10, r=10, t=40, b=30),
xaxis=dict(showgrid=False),
yaxis=dict(title="Units / week (retail + web)", gridcolor="#2A2D3A"),
legend=dict(orientation="h", y=-0.20, x=0.5, xanchor="center",
font=dict(size=10)),
hovermode="x unified",
)
st.plotly_chart(fig, use_container_width=True)
# Stats card — solo SKU only (multiselect would clutter)
if solo:
sel_sku_row = sku_row_by_id[selected_skus[0]]
qty_total = sum(qty_by_yw.values())
cs1, cs2, cs3, cs4 = st.columns(4)
peak_yw = max(qty_by_yw, key=qty_by_yw.get) if qty_by_yw else 0
peak_qty = qty_by_yw.get(peak_yw, 0)
cs1.metric("Peak week", f"CW{peak_yw % 100}",
f"{int(peak_qty)} u")
cs2.metric("Period units", f"{int(qty_total):,}")
cs3.metric("Uplift", f"{sel_sku_row['uplift']}×")
cs4.metric("Period RUC", f"€{sel_sku_row['ruc']:,.0f}")
# CSV export
csv = df.to_csv(index=False).encode("utf-8")
st.download_button("⬇️ Export this campaign as CSV", csv,
file_name=f"campaign_{c['id']}.csv",
mime="text/csv")