"""Promo Planner — pick SKUs, set period & discount, see impact, save."""
from __future__ import annotations
from datetime import datetime
from uuid import uuid4
import numpy as np
import pandas as pd
import plotly.graph_objects as go
import streamlit as st
from promo_data import (
load_all, base_run_rate, past_promos_for_sku, suggest_uplift,
promo_pattern_for_sku, cw_to_label, label_to_cw, horizon_weeks,
append_promotion, detect_conflicts, _detect_shadow_promo_weeks,
build_family_map, PROMO_MECHANICS, check_nc30, effective_discount,
guess_mechanic_from_discount, parse_unit_size,
parse_flavor, flavor_compatibility, historical_sibling_drop_pct,
)
from parent_map import build_parent_map
def render():
data = load_all()
if data["sales"].empty:
st.error("No sales_clean.csv found in ../data/. Run Update Sales in the main tool first.")
return
# ---------- SIDEBAR ----------
with st.sidebar:
st.markdown("### 🎯 Promotion Planner")
cy_now, cw_now = datetime.now().isocalendar()[:2]
st.caption(f"CW{cw_now}, {cy_now}")
promo_name = st.text_input("Promotion name", value="Summer Protein Sale",
key="planner_name")
source = st.selectbox(
"Source",
["B2C — MP (retail)", "B2C — WEB", "B2B — FMCG", "B2B — FITNESS"],
key="planner_source",
)
promo_type = st.selectbox(
"Promo type",
["Univerzalna", "Partner / Passport", "Rok istek",
"Otvaranje", "Dani centra", "VISA"],
key="planner_type",
)
# Default mechanic + discount — used to auto-fill new SKUs when
# added to basket. Per-SKU override happens inside each SKU tab.
default_mechanic = st.selectbox(
"Default promo mechanic",
PROMO_MECHANICS,
key="planner_mechanic_default",
help="Auto-fills mechanic for newly added SKUs. Each SKU can "
"override its own mechanic in its tab — different SKUs in "
"the same campaign can have different mechanics and "
"different discount depths.",
)
outcome = st.radio(
"Desired outcome",
["📦 Rješavanje lagera", "💰 Veći RUC",
"👥 Novi kupci", "🚀 Traffic driver"],
key="planner_outcome",
)
target = st.number_input("Measurable target (units or EUR)",
min_value=0, value=500, step=50, key="planner_target")
st.divider()
# ---- Period ----
all_weeks = horizon_weeks(13)
labels = [cw_to_label(y, w) for (y, w) in all_weeks]
c1, c2 = st.columns(2)
start_lbl = c1.selectbox("Start week", labels, index=2, key="planner_start")
end_lbl = c2.selectbox("End week", labels, index=5, key="planner_end")
sy, sw_ = label_to_cw(start_lbl)
ey, ew_ = label_to_cw(end_lbl)
n_weeks = max(1, (ey * 53 + ew_) - (sy * 53 + sw_) + 1)
st.metric("Duration", f"{n_weeks} weeks")
default_discount = st.slider("Default discount depth (%)",
0, 80, 20, 5,
key="planner_disc_default",
help="Auto-fills discount for newly added SKUs. "
"Override per SKU in each tab. Only used "
"when mechanic = 'Discount %' or 'Other'.")
mode = st.radio("Planning mode", ["Normal", "Target mode"],
horizontal=True, key="planner_mode")
st.divider()
# ---- Article picker ----
plan = data["plan"]
all_cats = ["All categories"] + sorted(plan["cat"].dropna().unique().tolist()) if "cat" in plan.columns else ["All categories"]
cat_pick = st.selectbox("Category", all_cats, key="planner_cat")
# Session state init
if "basket" not in st.session_state:
st.session_state["basket"] = []
if "groups" not in st.session_state:
# {"group_name": ["SKU1", "SKU2", ...]}
st.session_state["groups"] = {}
# ── Parent (proizvod) shortcut ──
# CMs typically run a promo on a "product" with multiple SKU variants
# (colors / flavors / sizes). Heuristic grouping until a proper
# sku_parent_map.csv is supplied: SKU-pattern split for hyphenated
# codes (VENUM-03813-449-{M,L,XL}), name-stem fallback for the rest.
parent_map = build_parent_map(plan)
if cat_pick != "All categories" and "cat" in plan.columns:
cat_skus = set(plan[plan["cat"] == cat_pick]["sku"].astype(str))
pmap = {k: v for k, v in parent_map.items()
if len(v["skus"]) > 1
and any(s in cat_skus for s in v["skus"])}
else:
pmap = {k: v for k, v in parent_map.items() if len(v["skus"]) > 1}
parent_keys = sorted(pmap.keys(),
key=lambda k: pmap[k]["display"].lower())
st.markdown(
"**🧩 Proizvod (parent) — heuristically grouped** "
"*(placeholder until sku_parent_map.csv arrives)*"
)
sel_parent = st.selectbox(
"Pick a parent to add all its variants at once",
options=[None] + parent_keys,
format_func=lambda k: (
"(none — pick SKUs individually below)" if k is None
else f"{pmap[k]['display'][:70]} · {len(pmap[k]['skus'])} variants"
),
key="planner_parent_pick",
)
if sel_parent is not None:
info = pmap[sel_parent]
name_lookup = dict(zip(plan["sku"].astype(str), plan["name"].astype(str)))
already_basket = set(st.session_state["basket"])
preview_rows = [
{
"SKU": s,
"Name": name_lookup.get(s, ""),
"In basket": "🛒" if s in already_basket else "",
}
for s in info["skus"]
]
st.dataframe(
pd.DataFrame(preview_rows),
hide_index=True, use_container_width=True,
height=min(220, 40 + 36 * len(preview_rows)),
)
n_new = sum(1 for s in info["skus"] if s not in already_basket)
if st.button(
f"➕ Add {n_new} new SKU(s) from this parent to basket",
type="primary", use_container_width=True,
disabled=n_new == 0,
key="planner_add_parent_btn",
):
for s in info["skus"]:
if s not in st.session_state["basket"]:
st.session_state["basket"].append(s)
# Auto-create a group for this parent. Group name is the
# parent's display (truncated). Members of the group are
# all child SKUs of the parent — clears any prior group
# assignments for those SKUs so the parent group becomes
# the single home for them.
group_name = info["display"][:60]
groups = st.session_state.setdefault("groups", {})
# Strip these SKUs from any other group (single-membership rule).
for gn in list(groups.keys()):
if gn == group_name:
continue
groups[gn] = [s for s in groups[gn] if s not in info["skus"]]
if not groups[gn]:
del groups[gn]
groups.setdefault(group_name, [])
for s in info["skus"]:
if s not in groups[group_name]:
groups[group_name].append(s)
# Persist parent metadata so the group tab can render the
# parent-overview block. Tied to the group name, not the
# parent key, so manual renames don't break the link
# (rename = lose overview; recreate from parent picker
# to get it back).
st.session_state.setdefault("group_parent_meta", {})[group_name] = {
"parent_key": sel_parent,
"skus": list(info["skus"]),
}
st.toast(f"Added {n_new} SKU(s) to group '{group_name[:30]}'.")
st.rerun()
st.divider()
# Search by name/SKU code (text filter)
search_q = st.text_input(
"🔎 Search article (name or SKU code)",
value="",
key="planner_search",
placeholder="e.g. Garmin Sapphire, POL12880, Fenix…",
).strip().lower()
# Filter SKU list
sku_options = plan.copy()
if cat_pick != "All categories" and "cat" in sku_options.columns:
sku_options = sku_options[sku_options["cat"] == cat_pick]
if search_q:
sku_options = sku_options[
sku_options["sku"].astype(str).str.lower().str.contains(search_q, na=False)
| sku_options.get("name", "").astype(str).str.lower().str.contains(search_q, na=False)
]
st.caption(f"**{len(sku_options)} matches**" + (f" (filtered by '{search_q}')" if search_q else ""))
# Multi-select checkbox table (only show top 50 to keep UI snappy)
already_in_basket = set(st.session_state["basket"])
pick_df_src = sku_options.head(50).copy() if len(sku_options) else sku_options.copy()
if len(pick_df_src) == 0:
st.info("No SKUs match. Adjust search or category.")
else:
pick_df = pd.DataFrame({
"✓": [False] * len(pick_df_src),
"SKU": pick_df_src["sku"].astype(str).tolist(),
"Name": pick_df_src.get("name", "").astype(str).tolist(),
"In basket": ["🛒" if s in already_in_basket else ""
for s in pick_df_src["sku"].astype(str)],
})
edited = st.data_editor(
pick_df,
column_config={
"✓": st.column_config.CheckboxColumn(width="small", default=False),
"SKU": st.column_config.TextColumn(width="small", disabled=True),
"Name": st.column_config.TextColumn(disabled=True),
"In basket": st.column_config.TextColumn(width="small", disabled=True),
},
hide_index=True,
use_container_width=True,
num_rows="fixed",
height=min(400, 40 + 36 * len(pick_df)),
key=f"planner_picker_{cat_pick}_{search_q}", # reset state on filter change
)
selected_skus = edited[edited["✓"]]["SKU"].tolist()
ba1, ba2 = st.columns([2, 1])
if ba1.button(f"➕ Add {len(selected_skus)} to basket",
use_container_width=True, type="primary",
key="planner_add_multi",
disabled=not selected_skus):
added = 0
for s in selected_skus:
if s not in st.session_state["basket"]:
st.session_state["basket"].append(s)
added += 1
if added:
st.toast(f"Added {added} SKU(s) to basket.")
st.rerun()
if ba2.button("Clear filter", use_container_width=True,
key="planner_clear_search"):
st.session_state["planner_search"] = ""
st.rerun()
channels_default = ["Retail", "Webshop"]
channels = st.multiselect("Channels",
["Retail", "Webshop", "Wholesale"],
default=channels_default, key="planner_chans")
st.divider()
# ============================================================
# BASKET — organize into groups
# ============================================================
basket = st.session_state["basket"]
groups = st.session_state["groups"]
# Ungrouped = basket SKUs not in any group
in_any_group = {s for skus in groups.values() for s in skus}
ungrouped = [s for s in basket if s not in in_any_group]
# Also clean orphan group members (SKUs removed from basket)
for gn in list(groups.keys()):
groups[gn] = [s for s in groups[gn] if s in basket]
if not groups[gn]:
del groups[gn]
st.caption(f"**Basket ({len(basket)}) · Groups ({len(groups)}) · "
f"Ungrouped ({len(ungrouped)})**")
# ---- Create / move-to group controls ----
with st.expander("➕ Create group / assign SKUs"):
if not basket:
st.caption("Add SKUs to basket first.")
else:
# 1. Pick SKUs to move (ungrouped + currently-in-other-group)
move_options = [
f"{s} — {data['name_map'].get(s, '')[:35]}"
for s in basket
]
picked = st.multiselect(
"Pick SKUs to assign",
options=basket,
format_func=lambda s: f"{s} — {data['name_map'].get(s, '')[:35]}",
key="planner_group_picker",
)
# 2. Choose existing group OR create new
existing = sorted(groups.keys())
gc1, gc2 = st.columns([2, 1])
target_group = gc1.selectbox(
"Move to group",
options=["(new group…)"] + existing,
key="planner_group_target",
)
new_group_name = ""
if target_group == "(new group…)":
new_group_name = gc1.text_input(
"New group name",
placeholder="e.g. Garmin Sapphire, Bučice 1–5kg",
key="planner_new_group_name",
).strip()
if gc2.button("Apply", use_container_width=True,
disabled=not picked or
(target_group == "(new group…)" and not new_group_name),
key="planner_group_apply"):
final_name = new_group_name if target_group == "(new group…)" else target_group
# Remove picked SKUs from any other group first (single membership)
for gn in list(groups.keys()):
groups[gn] = [s for s in groups[gn] if s not in picked]
if not groups[gn]:
del groups[gn]
groups.setdefault(final_name, []).extend(
[s for s in picked if s not in groups.get(final_name, [])]
)
st.toast(f"Moved {len(picked)} SKU(s) to '{final_name}'.")
st.rerun()
# ---- Display groups + ungrouped ----
for gn in sorted(groups.keys()):
with st.expander(f"📁 **{gn}** ({len(groups[gn])})", expanded=True):
for s in groups[gn]:
cA, cB = st.columns([5, 1])
cA.caption(f"`{s}` — {data['name_map'].get(s, '')[:40]}")
if cB.button("✕", key=f"rm_g_{gn}_{s}",
help="Remove from basket"):
if s in st.session_state["basket"]:
st.session_state["basket"].remove(s)
groups[gn] = [x for x in groups[gn] if x != s]
if not groups[gn]:
del groups[gn]
st.rerun()
if st.button(f"🗑️ Dissolve group", key=f"diss_{gn}",
help="Unassigns SKUs back to Ungrouped (keeps them in basket)"):
del groups[gn]
st.rerun()
if ungrouped:
with st.expander(f"📦 **Ungrouped** ({len(ungrouped)})", expanded=True):
for s in ungrouped:
cA, cB = st.columns([5, 1])
cA.caption(f"`{s}` — {data['name_map'].get(s, '')[:40]}")
if cB.button("✕", key=f"rm_u_{s}"):
st.session_state["basket"].remove(s)
st.rerun()
if basket:
if st.button("Clear all basket", use_container_width=True,
key="planner_clear"):
st.session_state["basket"] = []
st.session_state["groups"] = {}
st.rerun()
# ---------- MAIN ----------
st.markdown(f"## 🎯 {promo_name}")
st.caption(f"{source} · {promo_type} · {outcome} · "
f"CW{sw_} → CW{ew_} ({n_weeks} weeks) · "
f"default {default_discount}% / mechanic '{default_mechanic}' "
f"(can be overridden per SKU)")
with st.expander("ℹ️ How forecasting works (analog method)"):
st.markdown("""
**Per-week prediction (industry-standard analog method):**
The tool finds the **most recent past AKCIJA** for this SKU (3–4 wk
duration). It copies that campaign's actual weekly shape (RCM + WEB qty
per week), then scales it for today's conditions:
1. **Baseline ratio** — `current 4-week non-promo avg / analog's pre-promo avg`.
If sales have grown since the analog, this scales the pattern up.
2. **Discount delta** — each +1pp of discount above analog adds ~3 % to qty.
New discount lower than analog → predicted qty lower.
3. **OOS-cleaning** — analog weeks where qty was zero are dropped (they
represent stockouts, not real demand signal).
If the SKU has no past AKCIJA, the tool falls back to **category siblings**
(most recent AKCIJA from the same `cat`). If nothing matches at all, it
uses a generic ramp/peak/decline shape × 1.5 default uplift.
**Why this beats flat × uplift:**
A monthly AKCIJA isn't flat — typical shape is **week 1 build-up →
week 2 peak → week 3 saturation → week 4 end-push**. Predicting each
week separately gets stockout timing, in-store allocation and PO
deadlines right. Walmart, Tesco and Coop all use analog method as the
core for promo demand.
---
**Target mode:** flips the question. Instead of "what qty given X%
discount", you set a unit/EUR target and the tool back-solves the
required uplift × estimated discount. Useful for sanity-checking
ambitious targets — if required uplift is far above your max
historical, the target is unrealistic.
**Normal mode** (default) is the forward direction: set discount → see
predicted qty + P&L impact.
""")
if not st.session_state["basket"]:
st.info("Add at least one SKU from the sidebar to start planning.")
return
# Conflicts
conflicts = detect_conflicts(sy, sw_, ey, ew_, st.session_state["basket"])
if conflicts:
st.warning(f"⚠️ **{len(conflicts)} conflict(s)** with existing promos:")
for c in conflicts:
st.caption(f"• {c['sku']} ({c['name']}) — already in '{c['campaign']}' "
f"({c['period']})")
# ============================================================
# Tabs — one per GROUP + one per UNGROUPED SKU + Summary
# ============================================================
groups = st.session_state.get("groups", {})
in_any_group = {s for skus in groups.values() for s in skus}
ungrouped = [s for s in st.session_state["basket"] if s not in in_any_group]
# Build ordered list of (label, kind, payload) where kind=group|sku
tab_specs = []
for gn in sorted(groups.keys()):
tab_specs.append((f"📁 {gn} ({len(groups[gn])})", "group",
{"name": gn, "skus": groups[gn]}))
for sku in ungrouped:
tab_specs.append((f"📦 {sku}", "sku", {"sku": sku}))
tab_specs.append(("📊 Summary", "summary", None))
tabs = st.tabs([t[0] for t in tab_specs])
summary_data = []
for idx, (label, kind, payload) in enumerate(tab_specs):
with tabs[idx]:
if kind == "summary":
_render_summary(summary_data, promo_name, source, promo_type, outcome,
channels, sy, sw_, ey, ew_, n_weeks)
elif kind == "group":
gn = payload["name"]
gskus = payload["skus"]
is_parent_group = gn in st.session_state.get("group_parent_meta", {})
# Group-level discount override
header_prefix = "🧩 Parent" if is_parent_group else "📁 Group"
st.markdown(f"### {header_prefix}: {gn}")
gd1, gd2 = st.columns([1, 3])
disc_label = ("Parent discount %" if is_parent_group
else "Group discount %")
group_disc = gd1.number_input(
disc_label, min_value=0.0, max_value=80.0,
value=float(default_discount), step=5.0,
key=f"group_disc_{gn}",
help=("Discount koji se primjenjuje na cijeli parent "
"(sve child SKU-ove istovremeno). Vidi se odmah u "
"parent overview projekciji ispod. Per-SKU override "
"i dalje moguć unutar svakog SKU sub-tab-a."
if is_parent_group else
"Postavlja default discount za sve SKU-ove u grupi. "
"Per-SKU override ostaje moguć unutar svakog SKU sub-tab-a."),
)
# Propagate parent/group discount to each child's per-SKU
# session_state — copy, not stack. Streamlit otherwise keeps
# the old per-SKU `disc_{sku}` value once the user has
# interacted (value= arg gets ignored when key exists), so
# children visibly desync from parent.
# We only overwrite when the parent discount has actually
# changed since last render, so a user's intentional
# per-SKU override survives until they touch the parent
# control again.
_prev_key = f"group_disc_{gn}_prev"
_prev_val = st.session_state.get(_prev_key)
if _prev_val is None or float(_prev_val) != float(group_disc):
for _s in gskus:
st.session_state[f"disc_{_s}"] = float(group_disc)
st.session_state[_prev_key] = float(group_disc)
gd2.caption(
f"📦 **{len(gskus)} child SKU-ova** u parentu. "
f"Discount ide na sve istovremeno; parent overview ispod "
f"pokazuje agregiranu projekciju."
if is_parent_group else
f"📦 **{len(gskus)} SKU-ova** u grupi. "
f"Group discount se primjenjuje na sve, ali svaki "
f"SKU može pojedinačno biti override-an u svom tab-u "
f"ispod (Mechanic & discount sekcija)."
)
# Parent-level overview — only when this group is parent-derived
# AND has >1 SKU. Aggregates child SKUs into a synthetic parent
# row, then runs the same base_run_rate / suggest_uplift /
# past_promos_for_sku helpers on that synthetic SKU.
parent_meta = st.session_state.get("group_parent_meta", {}).get(gn)
if parent_meta and len(gskus) > 1:
with st.expander("📊 Parent-level analysis (aggregated)",
expanded=True):
_render_parent_overview(
parent_meta["parent_key"], gskus, data,
sy, sw_, ey, ew_, n_weeks, group_disc,
)
# SKU sub-tabs within the group
if gskus:
sub_tabs = st.tabs([f"📦 {s}" for s in gskus])
for j, s in enumerate(gskus):
with sub_tabs[j]:
sku_data = _render_sku_tab(
s, data, sy, sw_, ey, ew_, n_weeks,
group_disc, default_mechanic,
mode, channels, target,
)
sku_data["group"] = gn
summary_data.append(sku_data)
else: # kind == "sku"
sku = payload["sku"]
sku_data = _render_sku_tab(
sku, data, sy, sw_, ey, ew_, n_weeks,
default_discount, default_mechanic,
mode, channels, target,
)
sku_data["group"] = None
summary_data.append(sku_data)
def _render_sku_tab(sku, data, sy, sw_, ey, ew_, n_weeks,
default_discount, default_mechanic, mode,
channels, target):
cy_now, cw_now = datetime.now().isocalendar()[:2]
name = data["name_map"].get(sku, "")
cat = data["cat_map"].get(sku, "")
tier = data["tier_map"].get(sku, "")
xyz = data["xyz_map"].get(sku, "")
ws_share = float(data["wsshare_map"].get(sku, 0) or 0)
price = float(data["price_map"].get(sku, 0) or 0)
ruc_unit = float(data["ruc_unit_map"].get(sku, 0) or 0)
cost = float(data["cost_map"].get(sku, 0) or 0)
# ---- Per-SKU mechanic + discount (override of sidebar defaults) ----
st.markdown("##### 🎟️ Mechanic & discount for this SKU")
cm1, cm2 = st.columns([2, 1])
sku_mechanic = cm1.selectbox(
"Promo mechanic", PROMO_MECHANICS,
index=PROMO_MECHANICS.index(default_mechanic)
if default_mechanic in PROMO_MECHANICS else 0,
key=f"mech_{sku}",
help="1+1/2+1/3+1 set effective discount automatically "
"(50%, 33%, 25%). Use 'Discount %' or 'Other' to enter "
"your own number.",
)
sku_disc_input = cm2.number_input(
"Discount % (manual)", min_value=0.0, max_value=80.0,
value=float(default_discount), step=5.0,
key=f"disc_{sku}",
help="Only used when mechanic = 'Discount %' or 'Other'. "
"Bundle mechanics derive their own effective %.",
)
discount = effective_discount(sku_mechanic, sku_disc_input)
# Header — title on its own row so long product names don't squeeze the metrics
st.markdown(f"### {name or '(no name)'}")
st.caption(f"`{sku}` · {cat} · {tier} · WS share: {ws_share*100:.0f}%")
# Uplift suggestion + baseline run rate
upl = suggest_uplift(data["sales"], data["erp"], sku, discount)
uplift = upl["uplift"]
rr = base_run_rate(data["sales"], data["erp"], sku, lookback=26)
base_avg = rr["avg"]
# --- Build tooltips (methodology lives here, not in the page body) ---
uplift_help_lines = [f"Source: {upl['source']}"]
if upl.get("n_history", 0) > 0:
uplift_help_lines.append(
"Prediction = 60% most recent + 40% median across past AKCIJA campaigns."
)
rec = (f"Most recent: {upl['recent_uplift']}× "
f"(CW{upl['recent_start']}-CW{upl['recent_end']} {upl['recent_year']}")
if upl.get("recent_discount") is not None:
rec += f" @ {upl['recent_discount']}% discount"
rec += ")."
uplift_help_lines.append(rec)
uplift_help_lines.append(
f"Median: {upl['median_uplift']}× · Max: {upl['max_uplift']}×."
)
if upl.get("disc_min") is not None:
uplift_help_lines.append(
f"Historical discount range: {upl['disc_min']}–{upl['disc_max']}% "
f"(avg {upl['avg_disc_hist']}%)."
)
# Surface the new adjustment layers
if upl.get("band_mult"):
uplift_help_lines.append(
f"Discount-band shape: ×{upl['band_mult']} "
f"(non-linear curve, inflection at ~30–35%)."
)
if upl.get("price_disruptor_mult", 1.0) > 1.001:
uplift_help_lines.append(
f"Price-disruptor multiplier: ×{upl['price_disruptor_mult']} "
f"({upl.get('price_zone', '')})."
)
if upl.get("promo_per100") is not None:
uplift_help_lines.append(
f"€/100{upl.get('size_unit', '')} at promo price: "
f"{upl['promo_per100']:.2f} vs cat p25 "
f"{upl.get('cat_p25_per100', 0):.2f} / median "
f"{upl.get('cat_median_per100', 0):.2f}."
)
if upl.get("ceiling_applied"):
uplift_help_lines.append(
f"Ceiling applied: {upl['ceiling_applied']}× "
f"(adaptive — looser for first-time / aggressive promos)."
)
uplift_help = "\n\n".join(uplift_help_lines)
if rr["weeks"] > 0:
excl = rr.get("excluded", {})
excl_parts = []
if excl.get("promo"): excl_parts.append(f"{excl['promo']} ERP promo")
if excl.get("zero"): excl_parts.append(f"{excl['zero']} zero-sale (OOS)")
excl_str = ("Excluded: " + ", ".join(excl_parts) + ".") if excl_parts else ""
sample_str = ", ".join([f"CW{w} ({y})" for (y, w) in rr["yws"]])
baseline_help = (
f"Average of last {rr['weeks']} non-promo weeks (RCM + WEB) "
f"within 13-week lookback. {excl_str}\n\n"
f"Weeks used: {sample_str}.\n\n"
"Only weeks flagged in ERP promo calendar are excluded — "
"no statistical / discount filters."
)
baseline_value = f"{base_avg:.0f}"
else:
baseline_help = ("No non-promo weeks found in last 13 weeks. "
"Enter expected weekly qty manually in override section.")
baseline_value = "—"
discount_help = (f"Effective: {discount:.1f}% (mechanic: {sku_mechanic}). "
"Bundle mechanics (1+1 / 2+1 / 3+1) derive their own effective %.")
m1, m2, m3, m4, m5 = st.columns(5)
m1.metric("Price", f"€{price:.2f}")
m2.metric("Baseline / wk", baseline_value, help=baseline_help)
m3.metric("Uplift (adj.)", f"{uplift}×", help=uplift_help)
m4.metric("Discount", f"{discount}%", help=discount_help)
m5.metric("XYZ", xyz or "—")
# Inline alert for price-disruptor zone — too important to hide in a tooltip
if upl.get("price_disruptor_mult", 1.0) >= 2.0:
st.success(
f"💥 **Price-disruptor zone** — promo cijena €{price * (1 - discount/100):.2f} "
f"({upl.get('promo_per100', 0):.2f} €/100{upl.get('size_unit','') or 'u'}) je "
f"ispod kategorijskog p25 ({upl.get('cat_p25_per100', 0):.2f} €/100{upl.get('size_unit','') or 'u'}). "
f"Uplift dignut ×{upl['price_disruptor_mult']:.2f} jer ovaj price point "
"tipično otključava nove kupce, ne samo pojačanu konverziju postojećih."
)
elif upl.get("n_history", 0) == 0 and discount >= 30:
st.info(
"ℹ️ **First-time aggressive promo** — nema povijesti za ovaj SKU, "
"forecast je izveden iz kalibrirane band curve + price-disruptor. "
"Variance je široka — pogledaj 'Realistic upside (p90)' za upside "
"stranu i napravi sanity-check kroz override fields ili comparable SKU."
)
if rr["weeks"] == 0:
st.warning(
"No non-promo weeks found in last 13 weeks for this SKU — "
"baseline cannot be computed. Use override fields to enter "
"expected weekly quantities manually."
)
# ---- Analog method: compute per-week qty pattern FIRST so chart, P&L
# and coverage all use the ramp-peak-decline shape (not flat × uplift).
pattern = promo_pattern_for_sku(
data["sales"], data["erp"], sku,
target_duration=n_weeks,
current_baseline=base_avg,
target_discount=discount,
plan_df=data["plan"],
)
weekly_pattern_qty = list(pattern["weekly_qty"])
if len(weekly_pattern_qty) < n_weeks:
weekly_pattern_qty = weekly_pattern_qty + [int(round(base_avg * 1.5))] * (
n_weeks - len(weekly_pattern_qty))
# Forecast chart — pass per-week pattern, not flat × uplift
chart_fig = _build_forecast_chart(sku, data, sy, sw_, ey, ew_,
base_avg, weekly_pattern_qty)
st.plotly_chart(chart_fig, use_container_width=True)
# Total promo qty — quick callout right under the chart so CMs don't
# have to scroll to P&L for the bottom-line number.
promo_qty = sum(weekly_pattern_qty)
base_qty = base_avg * n_weeks
incremental = promo_qty - base_qty
peak_qty = max(weekly_pattern_qty) if weekly_pattern_qty else 0
peak_idx = weekly_pattern_qty.index(peak_qty) if weekly_pattern_qty else 0
peak_cw = sw_ + peak_idx
tq1, tq2, tq3, tq4, tq5 = st.columns(5)
tq1.metric(f"Total promo qty ({n_weeks}w)", f"{int(promo_qty):,}")
tq2.metric("Avg per week", f"{int(promo_qty / max(1, n_weeks)):,}")
tq3.metric("Peak week", f"{int(peak_qty):,}", f"CW{peak_cw}")
tq4.metric("Incremental vs no-promo",
f"{int(incremental):+,}",
delta=f"{(incremental / base_qty * 100 if base_qty else 0):+.0f}%")
# Realistic upside (p90 outcome — empirical from 3,107 events)
upside_uplift = upl.get("uplift_p90", uplift)
upside_qty = int(base_avg * n_weeks * upside_uplift)
tq5.metric("Realistic upside (p90)",
f"{upside_qty:,}",
delta=f"+{upside_qty - int(promo_qty):,} vs median",
help=f"Empirical p90 ratio for this discount band = "
f"{upl.get('upside_ratio', 1):.2f}×. Median forecast "
f"{int(promo_qty):,} u; upside {upside_qty:,} u is the "
"p90 historical outcome at similar depth — plausible "
"if SKU + timing + marketing align well.")
# ---- Coverage-aware clip warning ----
# If the forecast exceeds available stock + incoming during the promo
# window, surface it now so planner can order more or knows units will
# be OOS-capped. Pulls from WH stock + store stocks + incoming PO-s
# whose ETA falls within or before the promo period.
on_hand_total = 0.0
if not data["stock"].empty and "sku" in data["stock"].columns:
m = data["stock"]["sku"].astype(str) == sku
if m.any():
on_hand_total += float(data["stock"][m]["on_hand"].iloc[0] or 0)
for store_key in ("stock_stores", "stock_stores_at", "stock_stores_slo"):
st_df = data.get(store_key)
if st_df is not None and not st_df.empty and "sku" in st_df.columns:
m = st_df["sku"].astype(str) == sku
if m.any():
on_hand_total += float(st_df[m]["on_hand"].sum() or 0)
incoming_in_window = 0.0
inc_df = data.get("incoming")
if inc_df is not None and not inc_df.empty and "sku" in inc_df.columns:
sub = inc_df[inc_df["sku"].astype(str) == sku]
if not sub.empty and "week" in sub.columns:
wm = sub["week"].astype(int) <= int(ew_)
if "year" in sub.columns:
wm &= sub["year"].astype(int) <= int(ey)
incoming_in_window = float(pd.to_numeric(sub[wm].get("qty", 0),
errors="coerce").fillna(0).sum())
available = on_hand_total + incoming_in_window
if promo_qty > available and available > 0:
shortfall = promo_qty - available
st.error(
f"🚨 **OOS risk** — forecast {int(promo_qty):,} u > available "
f"{int(available):,} ({int(on_hand_total):,} on-hand + "
f"{int(incoming_in_window):,} incoming). "
f"Order ≥ **{int(shortfall):,}** units or expect cap during the promo."
)
elif promo_qty > available * 0.7 and available > 0:
st.warning(
f"⚠️ Tight stock — forecast {int(promo_qty):,} u vs available "
f"{int(available):,}. Buffer is thin; consider ordering "
f"{int(promo_qty - available * 0.7):,} more."
)
promo_price = price * (1 - discount / 100)
promo_revenue = promo_qty * promo_price
base_revenue = base_qty * price
rev_delta = promo_revenue - base_revenue
promo_ruc_unit = max(0, promo_price - cost)
promo_ruc = promo_qty * promo_ruc_unit
base_ruc = base_qty * ruc_unit
ruc_delta = promo_ruc - base_ruc
breakeven = (price / promo_price) if promo_price > 0 else 0
# ---- NC30 compliance check (lowest price in last 30 days) ----
st.markdown("#### ⚖️ NC30 compliance — najniža cijena u zadnjih 30 dana")
nc30_res = check_nc30(sku, promo_price)
if not nc30_res["has_data"]:
st.info(f"ℹ️ {nc30_res['message']}")
elif nc30_res["ok"]:
st.success(nc30_res["message"])
else:
st.error(nc30_res["message"])
st.caption("**Što napraviti:** spusti planiranu promo cijenu na ≤ NC30, "
"ili promijeni SKU za ovu kampanju. Trenutni promo cannot legally run.")
# ---- Bundle-mechanic analytical caveat (per-SKU) ----
mech_l = sku_mechanic.lower()
if "1+1" in mech_l or "2+1" in mech_l or "3+1" in mech_l or "4+1" in mech_l:
st.warning(
f"🔀 **Bundle mechanic ({sku_mechanic}):** effective discount "
f"{discount:.1f}% applies to the bundle. Historical uplift "
"covers the WHOLE promo period — baskets where customer "
"bought only 1 unit (non-qualifying) are mixed in. Reported "
"uplift is **total-period vs baseline**, not 'qty=N+1 only'. "
"For precise bundle-effect read, transaction-level data "
"would be needed."
)
st.markdown("#### 💰 P&L / margin impact")
p1, p2, p3, p4, p5 = st.columns(5)
p1.metric("Promo revenue", f"€{promo_revenue:,.0f}")
p2.metric("Rev w/o promo", f"€{base_revenue:,.0f}")
p3.metric("Revenue delta", f"€{rev_delta:+,.0f}",
delta=f"{(rev_delta/base_revenue*100 if base_revenue else 0):+.1f}%")
p4.metric("RUC delta", f"€{ruc_delta:+,.0f}")
p5.metric("Breakeven uplift", f"{breakeven:.2f}×")
if uplift >= breakeven:
st.success(f"✅ Estimated uplift ({uplift}×) exceeds breakeven ({breakeven:.2f}×) — margin-positive.")
else:
st.error(f"❌ Estimated uplift ({uplift}×) below breakeven ({breakeven:.2f}×) — margin-negative.")
# Target mode
if mode == "Target mode":
st.markdown("#### 🎯 Target mode")
t1, t2, t3 = st.columns(3)
t1.metric("Target units", f"{int(target):,}")
if base_avg > 0 and n_weeks > 0:
req_uplift = target / (base_avg * n_weeks)
est_disc = max(0, min(50, 20 + (req_uplift - 1.35) / 0.04))
t2.metric("Required uplift", f"{req_uplift:.2f}×")
t3.metric("Est. discount needed", f"~{est_disc:.0f}%")
else:
t2.metric("Required uplift", "—")
t3.metric("Est. discount needed", "—")
# Promo history
st.markdown("#### 📜 Promo history for this SKU")
history = past_promos_for_sku(data["sales"], data["erp"], sku)
if history:
for h in history[:10]:
tag = "🟢" if h["uplift"] >= 1.5 else ("🟡" if h["uplift"] >= 1.15 else "🔴")
oos_note = ""
if h.get("oos_suspected"):
oos_note = (f" · ⚠️ **OOS suspected** ({h['oos_weeks']} zero-sale week(s)) — "
f"clean avg used: {h['avg_promo_clean']}/wk")
disc_note = ""
mech_note = ""
if h.get("discount_pct", 0) > 0:
disc_note = f" · discount **{h['discount_pct']}%**"
# Prefer transaction-detected mechanic (high confidence) over
# discount-band heuristic guess.
if h.get("mechanic_detected"):
mech_note = (
f" · mechanic: **{h['mechanic_detected']}** "
f"({h['mechanic_confidence']:.0f}% confidence, "
f"{h['mechanic_n_docs']} docs)"
)
elif h.get("discount_pct", 0) > 0:
mech_guess = guess_mechanic_from_discount(h["discount_pct"])
if mech_guess:
mech_note = f" · mechanic: *{mech_guess}*"
st.caption(
f"{tag} **CW{h['start_week']}-CW{h['end_week']} ({h['year']})** "
f"· {h['type']} · {h['duration']}w "
f"· Base: {h['avg_base']}/wk → Promo: {h['avg_promo']}/wk "
f"· **{h['uplift']}× uplift**"
+ disc_note + mech_note +
f" · Post-dip: {h['post_dip_pct']:+.0f}%"
+ oos_note
)
else:
st.caption("No past **monthly AKCIJA** (3-4 weeks) found for this SKU.")
# Shadow promos — big retail+web spikes without ERP record
shadows = _detect_shadow_promo_weeks(data["sales"], sku, spike_mult=2.5)
# Drop shadows that overlap an existing ERP-recorded campaign
erp_yws = set()
for h in history:
for w in range(h["start_week"], h["end_week"] + 1):
erp_yws.add((h["year"], w))
shadows = [s for s in shadows if (s["year"], s["week"]) not in erp_yws]
if shadows:
st.markdown("##### 🔍 Shadow promos detected (qty spike, no ERP record)")
for s in shadows[:8]:
disc_str = f" · discount {s['discount_pct']}%" if s["discount_pct"] > 0 else ""
st.caption(
f"💥 **CW{s['week']} ({s['year']})** · qty {s['qty']:,} vs "
f"trailing median {s['median_prev']} → **{s['spike_x']}× spike**"
+ disc_str
)
# Coverage analysis
st.markdown("#### 📦 Coverage & stock")
on_hand = 0
if not data["stock"].empty and "sku" in data["stock"].columns:
m = data["stock"]["sku"].astype(str) == sku
if m.any():
on_hand = float(data["stock"][m]["on_hand"].iloc[0])
if base_avg > 0:
cov_now = on_hand / base_avg
# Stock at promo start (assume normal demand consumed during run-up)
weeks_until_promo = max(0, (sy * 53 + sw_) - (cy_now * 53 + cw_now))
stock_at_start = max(0, on_hand - base_avg * weeks_until_promo)
stock_at_end = max(0, stock_at_start - promo_qty)
cov_after = stock_at_end / base_avg if base_avg else 0
# Upside scenario stock projection
stock_at_end_upside = max(0, stock_at_start - upside_qty)
cov_after_upside = stock_at_end_upside / base_avg if base_avg else 0
else:
cov_now = stock_at_start = stock_at_end = cov_after = 0
stock_at_end_upside = cov_after_upside = 0
s1, s2, s3, s4, s5 = st.columns(5)
s1.metric("Coverage now", f"{cov_now:.1f}w")
s2.metric("Stock now", f"{int(on_hand):,}")
s3.metric("@ promo start", f"{int(stock_at_start):,}")
s4.metric("@ promo end", f"{int(stock_at_end):,}")
s5.metric("After promo (cov)", f"{cov_after:.1f}w")
# ---- MEDIAN scenario ----
lead_time = 8
if not data["supply_master"].empty and "sku" in data["supply_master"].columns:
m = data["supply_master"]["sku"].astype(str) == sku
if m.any():
lead_time = int(data["supply_master"][m].get("lead_time_weeks", 8).iloc[0])
st.markdown("**Median scenario** — forecast {:,} u".format(int(promo_qty)))
extra_needed = max(0, promo_qty - stock_at_start)
if extra_needed > 0:
weeks_to_order = max(0, weeks_until_promo - lead_time)
if weeks_to_order < 0:
st.error(f"🚨 **Too late** — need {int(extra_needed):,} extra units but lead time {lead_time}w "
f"exceeds {weeks_until_promo}w until promo start.")
elif weeks_to_order < 2:
st.warning(f"⚠️ **Tight** — order {int(extra_needed):,} extra units within {weeks_to_order}w "
f"(lead time {lead_time}w).")
else:
st.info(f"ℹ️ Order {int(extra_needed):,} extra units in next {weeks_to_order}w "
f"(lead time {lead_time}w).")
else:
st.success(f"✅ Stock sufficient — no extra order needed for median scenario.")
# ---- UPSIDE (p90) scenario ----
st.markdown(
f"**Realistic upside (p90) scenario** — forecast {int(upside_qty):,} u "
f"_(empirical p90 = {upl.get('upside_ratio', 1):.2f}× median)_"
)
u1, u2, u3 = st.columns(3)
u1.metric("@ promo end (upside)", f"{int(stock_at_end_upside):,}")
u2.metric("After promo (cov, upside)", f"{cov_after_upside:.1f}w")
extra_upside = max(0, upside_qty - stock_at_start)
u3.metric("Extra units to cover upside",
f"{int(extra_upside):,}",
delta=f"+{int(extra_upside - extra_needed):,} vs median order",
help="How many more units you'd need on top of stock to "
"cover the p90 (upside) scenario without running OOS.")
if extra_upside > 0:
weeks_to_order_up = max(0, weeks_until_promo - lead_time)
if weeks_to_order_up < 0:
st.error(
f"🚨 **Upside unreachable** — covering p90 forecast needs "
f"{int(extra_upside):,} extra units, ali lead time {lead_time}w "
f"prelazi {weeks_until_promo}w do starta. Akcija će biti "
f"OOS-capped ako stvarno krene gore."
)
elif weeks_to_order_up < 2:
st.warning(
f"⚠️ **Tight za upside** — order {int(extra_upside):,} extra units "
f"within {weeks_to_order_up}w da pokriješ p90 ishod "
f"(lead time {lead_time}w)."
)
else:
st.info(
f"ℹ️ Da pokriješ realistic upside: order {int(extra_upside):,} extra "
f"units u sljedećih {weeks_to_order_up}w "
f"(lead time {lead_time}w). Razlika vs median order: "
f"+{int(extra_upside - extra_needed):,} u."
)
else:
st.success(f"✅ Stock pokriva čak i upside scenarij — niskobudžet order.")
if pattern.get("analog"):
override_help = (
f"Level = baseline × predicted uplift = {base_avg:.0f}/wk × "
f"{pattern['predicted_uplift']}× = {pattern['promo_avg_per_week']:.0f}/wk avg.\n\n"
f"Shape copied from {pattern['source']} "
f"(weekly index: {pattern['pattern_norm']}, avg ≈ 1.0).\n\n"
f"Discount delta vs analog ({pattern['analog_discount']}%): "
f"×{pattern['scale_discount']}.\n\n"
f"Implied total uplift after shape: {pattern['derived_uplift']}×."
)
else:
override_help = (
f"No analog AKCIJA found. Level = baseline × 1.5× = "
f"{base_avg * 1.5:.0f}/wk avg. "
f"Shape = generic ramp-peak-decline ({pattern['pattern_norm']})."
)
st.markdown("#### ✏️ Override per-week quantities")
weekly_qty = []
cols = st.columns(min(n_weeks, 7))
for j in range(n_weeks):
wk = sw_ + j
default_q = int(weekly_pattern_qty[j]) if j < len(weekly_pattern_qty) else int(round(base_avg * 1.5))
with cols[j % len(cols)]:
# Only the first input carries the methodology tooltip — keeps
# the row visually clean while the info is still discoverable.
v = st.number_input(f"CW{wk}", min_value=0,
value=default_q, step=10,
key=f"ov_{sku}_{wk}",
help=override_help if j == 0 else None)
weekly_qty.append(v)
# Surface implied uplift from the analog pattern (sum / baseline_total)
uplift = pattern["derived_uplift"] if pattern["derived_uplift"] > 0 else uplift
# Cannibalization — conservation-of-customers model with weighted
# distribution. See review notes for rationale: incremental units
# come from somewhere, weighted across siblings by baseline ×
# retail_share × price_proximity × tier × overlap.
st.markdown("#### 🔀 Cannibalization (sub-category siblings)")
sub_cat = data["subcat_map"].get(sku, "")
siblings_raw = []
if sub_cat and not data["subcat"].empty:
sib_df = data["subcat"][(data["subcat"].get("sub_cat", "") == sub_cat)
& (data["subcat"]["sku"].astype(str) != sku)]
# Pre-compute promo-window for overlap detection
promo_yw_start = sy * 100 + sw_
promo_yw_end = ey * 100 + ew_
# Promo SKU's price-per-100 (from suggest_uplift output)
promo_per100 = float(upl.get("promo_per100") or 0)
promo_size_unit = upl.get("size_unit")
cat_p25 = float(upl.get("cat_p25_per100") or 0)
cat_med = float(upl.get("cat_median_per100") or 0)
# Distance normalization scale — half the cat p25..median span
cat_scale = max((cat_med - cat_p25) * 2.0, 1.0)
# Pre-index ws_share from sku_plan_list for retail-share weighting
ws_share_map = {}
if not data["plan"].empty and "ws_share_26w" in data["plan"].columns:
ws_share_map = dict(zip(
data["plan"]["sku"].astype(str),
pd.to_numeric(data["plan"]["ws_share_26w"],
errors="coerce").fillna(0)
))
# Promo SKU's flavor — sibling cannibalization weighted by
# flavor match (different flavors don't substitute strongly).
promo_flavor = parse_flavor(name)
# Promo SKU's size — used for size-first proximity scoring.
# Buyers of 2 kg whey aren't realistic substitutes for 4.5 kg
# whey buyers (different shopper cohort) even when €/100g looks
# similar. Size proximity gates that.
promo_size_val = float(upl.get("size_val") or 0)
if promo_size_val <= 0:
ps_val, _ps_unit = parse_unit_size(name)
promo_size_val = ps_val
def _size_proximity(a: float, b: float) -> float:
"""How close in size (as ratio of smaller / larger).
≥ 0.85 → 1.00 (within 15 %, same shopper)
≥ 0.65 → 0.75
≥ 0.40 → 0.45 (2× difference — different cohort)
≥ 0.25 → 0.30
< 0.25 → 0.20
missing → 0.50 (neutral)"""
if a <= 0 or b <= 0:
return 0.50
r = min(a, b) / max(a, b)
if r >= 0.85: return 1.00
if r >= 0.65: return 0.75
if r >= 0.40: return 0.45
if r >= 0.25: return 0.30
return 0.20
# Pre-index ERP promos per SKU for overlap check
erp_by_sku = {}
erp_df = data["erp"]
if not erp_df.empty and "sku" in erp_df.columns:
erp_by_sku = {
s: list(zip(g["year"].astype(int), g["week"].astype(int)))
for s, g in erp_df.groupby("sku")
}
# SKUs that are part of THIS promo basket — they're not cannibalized
# by the focus SKU, they're co-promoted. Drop them from siblings.
basket_skus = set(st.session_state.get("basket", []))
# ============================================================
# PERF: compute CHEAP proximity score first (name + price-map
# lookups only), filter to top 20 by proximity, THEN call the
# heavy base_run_rate. Reduces base_run_rate calls from ~200
# (full sub-cat) to ~20 (top candidates) on a typical promo.
# ============================================================
SIBLING_HARD_CAP = 20 # max siblings to fully evaluate
cheap_candidates = []
for _, r in sib_df.iterrows():
sib_sku = str(r["sku"])
if sib_sku in basket_skus:
continue
# Cheap fields — all O(1) lookups
ws_share = float(ws_share_map.get(sib_sku, 0))
sib_tier = data["tier_map"].get(sib_sku, "")
sib_name = data["name_map"].get(sib_sku, "")
sib_size_val, sib_size_unit = parse_unit_size(sib_name)
sib_price = float(data["price_map"].get(sib_sku, 0) or 0)
sib_per100 = (sib_price * 100.0 / sib_size_val) if sib_size_val > 0 else 0
# Composite proximity (size 50% / flavor 30% / price 20%)
if sib_size_unit and promo_size_unit and sib_size_unit != promo_size_unit:
size_prox = 0.10
else:
size_prox = _size_proximity(promo_size_val, sib_size_val)
sib_flavor = parse_flavor(sib_name)
if not promo_flavor or not sib_flavor:
flavor_prox = 0.25
elif promo_flavor == sib_flavor:
flavor_prox = 1.00
else:
flavor_prox = 0.40
if (sib_per100 > 0 and promo_per100 > 0
and sib_size_unit == promo_size_unit):
distance = abs(sib_per100 - promo_per100)
price_prox = max(0.10, 1.0 - min(1.0, distance / cat_scale))
else:
price_prox = 0.50
proximity = (0.50 * size_prox
+ 0.30 * flavor_prox
+ 0.20 * price_prox)
# Early bail-out by cheap filters: WS share + proximity threshold
if proximity < 0.50: # well below 0.75 final threshold
continue
if ws_share > 0.50: # well above 0.35 final threshold
continue
# Tier factor (still cheap)
if "GOLD" in sib_tier: tier_factor = 0.70
elif "BRONZE" in sib_tier: tier_factor = 1.30
else: tier_factor = 1.00
cheap_candidates.append({
"sku": sib_sku, "name": sib_name,
"ws_share": ws_share, "retail_share": max(0.0, 1.0 - ws_share),
"tier": sib_tier, "tier_factor": tier_factor,
"size_prox": size_prox, "flavor_prox": flavor_prox,
"price_prox": price_prox, "proximity": proximity,
"size_val": sib_size_val, "per100": sib_per100,
"price": sib_price, "flavor": sib_flavor,
})
# Sort by proximity descending, keep top N → only those get
# expensive base_run_rate + historical drop computation.
cheap_candidates.sort(key=lambda c: -c["proximity"])
cheap_candidates = cheap_candidates[:SIBLING_HARD_CAP]
# ---- Heavy compute only for survivors ----
for cand in cheap_candidates:
sib_sku = cand["sku"]
sib_rr = base_run_rate(data["sales"], data["erp"], sib_sku, n_weeks=4)
if sib_rr["avg"] <= 0:
continue
sib_name = cand["name"]
sib_size_val = cand["size_val"]
ws_share = cand["ws_share"]
retail_share = cand["retail_share"]
sib_tier = cand["tier"]
tier_factor = cand["tier_factor"]
size_prox = cand["size_prox"]
flavor_prox = cand["flavor_prox"]
price_prox = cand["price_prox"]
proximity = cand["proximity"]
sib_per100 = cand["per100"]
sib_price = cand["price"]
sib_flavor = cand["flavor"]
# Overlap with sibling's own promo → dampen
has_overlap = False
for (y, w) in erp_by_sku.get(sib_sku, []):
p_yw = int(y) * 100 + int(w)
if promo_yw_start <= p_yw <= promo_yw_end:
has_overlap = True
break
overlap_factor = 0.30 if has_overlap else 1.00
# Composite proximity already incorporates flavor (×0.30
# share). No separate flavor_compat factor — would double-
# count. Proximity alone drives both filter and weight.
weight = (sib_rr["avg"]
* retail_share
* proximity
* tier_factor
* overlap_factor)
siblings_raw.append({
"sku": sib_sku,
"name": sib_name[:40],
"tier": sib_tier,
"baseline_wk": sib_rr["avg"],
"ws_share": ws_share,
"retail_share": retail_share,
"sib_per100": sib_per100,
"sib_size": sib_size_val,
"size_prox": size_prox,
"flavor_prox": flavor_prox,
"price_prox": price_prox,
"proximity": proximity,
"overlap": has_overlap,
"tier_factor": tier_factor,
"flavor": sib_flavor or "—",
"weight": weight,
"price": sib_price,
"ruc_unit": float(data["ruc_unit_map"].get(sib_sku, 0) or 0),
})
# ---- Filter to REAL substitutes ----
# Three filters:
# proximity ≥ 0.75 → close enough across size/flavor/price.
# Loosened from 0.85 because that was too
# strict — concentrated budget on 1-2 SKUs.
# ws_share ≤ 0.35 → predominantly retail (65 %+)
# no overlap → sibling NOT running its own promo in our window
PROXIMITY_THRESHOLD = 0.75
WS_SHARE_MAX = 0.35
# Per-sibling sanity cap: sibling can't lose more than 85 % of its
# own period baseline (you can't cannibalize what doesn't exist).
PER_SIBLING_MAX_LOSS_PCT = 0.85
real_subs = [
s for s in siblings_raw
if s["proximity"] >= PROXIMITY_THRESHOLD
and s["ws_share"] <= WS_SHARE_MAX
and not s["overlap"]
]
excluded_n = len(siblings_raw) - len(real_subs)
excluded_by_proximity = sum(1 for s in siblings_raw
if s["proximity"] < PROXIMITY_THRESHOLD)
excluded_by_ws = sum(1 for s in siblings_raw
if s["proximity"] >= PROXIMITY_THRESHOLD
and s["ws_share"] > WS_SHARE_MAX)
excluded_by_overlap = sum(1 for s in siblings_raw
if s["proximity"] >= PROXIMITY_THRESHOLD
and s["ws_share"] <= WS_SHARE_MAX
and s["overlap"])
if real_subs:
# ---- Cannibalization = per-sibling drop %, data-driven ----
# NEW SIMPLE MODEL (replaces rate-based budget distribution):
# For each real substitute, look at its retail+web sales during
# PAST promos of THIS focus SKU. Median drop % across those
# weeks = expected drop in the upcoming promo. If no history,
# fall back to flat 15% (mid of 10-20% per CM rule of thumb).
#
# Wholesale is excluded entirely — baseline and the historical
# drop measurement both use qty_retail + qty_webshop only.
FALLBACK_DROP_PCT = 0.15
siblings_raw = real_subs
for s in siblings_raw:
hist = historical_sibling_drop_pct(
data["sales"], data["erp"],
focus_sku=sku, sibling_sku=s["sku"],
fallback_pct=FALLBACK_DROP_PCT,
)
drop_pct = max(0.0, min(0.95, float(hist["drop_pct"])))
s["drop_pct"] = drop_pct
s["drop_source"] = hist["source"]
s["n_history_obs"] = hist["n_observations"]
# Loss = baseline × n_weeks × drop_pct (capped naturally by
# the 0..0.95 clamp above — physically can't lose >95% of
# what you sell).
s["loss_units"] = s["baseline_wk"] * n_weeks * drop_pct
s["loss_revenue"] = s["loss_units"] * s["price"]
s["loss_ruc"] = s["loss_units"] * s["ruc_unit"]
# For caption — total cannibalization actually computed
total_cann = sum(s["loss_units"] for s in siblings_raw)
incremental_units = max(0.0, float(promo_qty - base_qty))
# ---- Display top 10 sorted by € loss revenue (bigger-impact first) ----
sorted_sib = sorted(siblings_raw,
key=lambda x: -x.get("loss_revenue", 0))
# Size unit suffix for display ("g" for solids, "ml" for liquids)
size_u = promo_size_unit or "g"
show_df = pd.DataFrame([{
"SKU": s["sku"],
"Name": s["name"],
"Tier": s["tier"],
"Baseline/wk": round(s["baseline_wk"], 1),
"WS share": f"{s['ws_share']*100:.0f}%",
f"Size ({size_u})": int(s["sib_size"]) if s["sib_size"] > 0 else "—",
"Flavor": s["flavor"],
"Proximity": f"{s['proximity']:.2f}",
"Drop %": f"{s['drop_pct']*100:.0f}%",
"Drop source": ("📊 history" if s.get("n_history_obs", 0) > 0
else "⚙️ default 15%"),
"List €/unit": round(s["price"], 2),
f"Loss units ({n_weeks}w)": round(s["loss_units"], 2),
"Loss revenue €": round(s["loss_revenue"], 0),
} for s in sorted_sib[:10]])
st.dataframe(show_df, use_container_width=True, hide_index=True)
# Tail summary — explicit so the Net totals below don't surprise
if len(sorted_sib) > 10:
top10_loss_rev = sum(s["loss_revenue"] for s in sorted_sib[:10])
top10_loss_units = sum(s["loss_units"] for s in sorted_sib[:10])
tail_loss_rev = sum(s["loss_revenue"] for s in sorted_sib[10:])
tail_loss_units = sum(s["loss_units"] for s in sorted_sib[10:])
st.caption(
f"📋 Prikazano top 10 od **{len(sorted_sib)} siblinga**. "
f"Top 10 zbroj: {top10_loss_units:.1f} u · €{top10_loss_rev:,.0f}. "
f"**Tail** ({len(sorted_sib) - 10} ostalih siblinga, pojedinačno mali "
f"loss): {tail_loss_units:.1f} u · €{tail_loss_rev:,.0f}. "
f"Net panel ispod zbraja **sve**."
)
# ---- NET category impact ----
total_loss_units = sum(s["loss_units"] for s in siblings_raw)
total_loss_revenue = sum(s["loss_revenue"] for s in siblings_raw)
total_loss_ruc = sum(s["loss_ruc"] for s in siblings_raw)
net_units = incremental_units - total_loss_units
net_revenue = rev_delta - total_loss_revenue
net_ruc = ruc_delta - total_loss_ruc
st.markdown("##### 📊 Net category impact (after cannibalization)")
n1, n2, n3 = st.columns(3)
n1.metric(
"Net category units",
f"{int(round(net_units)):+,}",
delta=f"-{int(round(total_loss_units)):,} from siblings",
help=f"Incremental {int(round(incremental_units)):+,} u "
f"− cannibalization {int(round(total_loss_units)):,} u "
f"= true category increment."
)
n2.metric(
"Net category revenue",
f"€{net_revenue:+,.0f}",
delta=f"-€{total_loss_revenue:,.0f} from siblings",
help="Promo revenue gain minus revenue lost on siblings "
"(sibling units × their list price)."
)
n3.metric(
"Net category RUC",
f"€{net_ruc:+,.0f}",
delta=f"-€{total_loss_ruc:,.0f} from siblings",
help="Promo RUC minus sibling RUC lost — true margin impact "
"across the category."
)
n_with_history = sum(1 for s in siblings_raw if s.get("n_history_obs", 0) > 0)
n_fallback = len(siblings_raw) - n_with_history
excl_parts = []
if excluded_by_proximity:
excl_parts.append(f"{excluded_by_proximity} too distant "
f"(proximity < {PROXIMITY_THRESHOLD})")
if excluded_by_ws:
excl_parts.append(f"{excluded_by_ws} WS-heavy "
f"(WS share > {int(WS_SHARE_MAX*100)}%)")
if excluded_by_overlap:
excl_parts.append(f"{excluded_by_overlap} on their own promo "
"(overlap excluded)")
excl_note = (" Excluded: " + " · ".join(excl_parts) + "." ) if excl_parts else ""
st.caption(
f"💡 **Drop-based model:** za svaki sibling se mjeri **historical "
f"drop %** iz prošlih akcija ovog SKU-a (retail + web only, "
f"wholesale isključen). Fallback na **15%** ako sibling nema "
f"povijest. Loss = sibling baseline × {n_weeks} weeks × drop %. \n"
f"📊 **{n_with_history}** of {len(siblings_raw)} siblings koriste "
f"historical drop · ⚙️ **{n_fallback}** koriste 15% fallback. \n"
f"Total cannibalization: **{int(round(total_cann)):,} u**.{excl_note} "
f"Filtri: proximity ≥ {PROXIMITY_THRESHOLD}, WS share ≤ "
f"{int(WS_SHARE_MAX*100)}%, no overlap. "
f"**Proximity** = 0.50 × size + 0.30 × flavor + 0.20 × price."
)
elif siblings_raw:
# Had siblings in sub-cat but none qualified
st.info(
f"ℹ️ Found {len(siblings_raw)} sub-category siblings but **none** met "
f"the filters (proximity ≥ {PROXIMITY_THRESHOLD} AND WS share ≤ "
f"{int(WS_SHARE_MAX*100)}%). No cannibalization modeled — incremental "
f"units treated as fully new retail demand."
)
else:
st.caption(f"No sibling SKUs found in sub-category '{sub_cat or '—'}'.")
return {
"sku": sku, "name": name, "cat": cat, "tier": tier,
"weekly": weekly_qty, "total": sum(weekly_qty),
"uplift": uplift, "discount": discount,
"mechanic": sku_mechanic,
"price": price, "promo_price": promo_price,
"ruc_unit_promo": promo_ruc_unit,
# Realistic upside (p90) for category summary scenario projection
"upside_uplift": upl.get("uplift_p90", uplift),
"upside_total": int(base_avg * n_weeks * upl.get("uplift_p90", uplift))
if base_avg > 0 else 0,
# Cannibalization aggregates for this SKU's external siblings
"cann_units": (sum(s.get("loss_units", 0) for s in siblings_raw)
if siblings_raw else 0),
"cann_revenue": (sum(s.get("loss_revenue", 0) for s in siblings_raw)
if siblings_raw else 0),
"cann_ruc": (sum(s.get("loss_ruc", 0) for s in siblings_raw)
if siblings_raw else 0),
}
def _build_forecast_chart(sku, data, sy, sw_, ey, ew_, base_avg, weekly_pattern):
"""`weekly_pattern` is a list of per-week qty (length = ew_-sw_+1).
Plot the ramp-peak-decline shape, not a flat line."""
fig = go.Figure()
sales = data["sales"]
sub = sales[sales["sku"] == sku].sort_values(["year", "week"]).tail(12)
if len(sub):
labels_a = [f"CW{int(r['week'])}" for _, r in sub.iterrows()]
# Show RCM + WEB only (matching what promotions affect)
vals_a = (sub["qty_retail"].astype(float)
+ sub.get("qty_webshop", pd.Series([0]*len(sub))).astype(float)).tolist()
fig.add_trace(go.Bar(x=labels_a, y=vals_a, name="Actuals (RCM+WEB)",
marker_color="rgba(139,143,163,0.4)"))
promo_weeks = list(range(sw_, ew_ + 1))
promo_qty = list(weekly_pattern[:len(promo_weeks)])
fig.add_trace(go.Scatter(
x=[f"CW{w}" for w in promo_weeks],
y=promo_qty, mode="lines+markers+text",
name="Promo estimate (analog shape)",
text=[f"{int(q):,}" for q in promo_qty],
textposition="top center",
line=dict(color="#FF6B6B", width=3),
marker=dict(size=12, color="#FF6B6B"),
))
base_weeks = list(range(max(1, sw_ - 4), ew_ + 5))
base_qty = [base_avg] * len(base_weeks)
fig.add_trace(go.Scatter(x=[f"CW{w}" for w in base_weeks],
y=base_qty, mode="lines",
name="Base forecast",
line=dict(color="#60A5FA", width=1.5, dash="dash")))
# Post-promo dip
post_weeks = list(range(ew_ + 1, ew_ + 4))
post_qty = [base_avg * 0.85] * len(post_weeks)
fig.add_trace(go.Scatter(x=[f"CW{w}" for w in post_weeks],
y=post_qty, mode="lines",
name="Post-promo dip",
line=dict(color="#F97316", width=1.5, dash="dot")))
fig.update_layout(
plot_bgcolor="#1A1D27", paper_bgcolor="#1A1D27",
font=dict(color="#E8E9ED", size=11),
height=320, margin=dict(l=10, r=10, t=10, b=10),
legend=dict(orientation="h", y=-0.2),
xaxis=dict(showgrid=False),
yaxis=dict(gridcolor="#2A2D3A", title="Units / week"),
)
return fig
def _render_parent_overview(parent_key, skus, data, sy, sw_, ey, ew_,
n_weeks, discount):
"""Parent = sum of children. Discount applies to each child via the
existing per-SKU helpers; parent metrics are arithmetic sums of those
per-child results. No synthetic aggregation."""
sales = data["sales"]
erp = data["erp"]
child_sales = (sales[sales["sku"].isin(skus)]
if "sku" in sales.columns else sales.iloc[0:0])
if len(child_sales) == 0:
st.info("No sales history for any child SKU.")
return
# Weekly totals across all children — used for raw volume + chart actuals
grouped = child_sales.groupby(["year", "week"], as_index=False).agg({
"qty_total": "sum",
"qty_retail": "sum",
"qty_webshop": "sum",
"qty_wholesale": "sum",
}).sort_values(["year", "week"])
p_sales = grouped # alias for the rest of the function
p_sorted = grouped
# ── Raw volume block — sum of children's actuals ──
last4 = p_sorted.tail(4)
last13 = p_sorted.tail(13)
last52 = p_sorted.tail(52)
def _sum(df, col):
return int(df[col].sum()) if col in df.columns else 0
st.caption(f"**Sum across {len(skus)} child SKUs** · "
f"discount applied: **{float(discount):.0f}%**")
v1, v2, v3, v4 = st.columns(4)
v1.metric("Last 4 weeks", f"{_sum(last4, 'qty_total'):,}")
v2.metric("Last 13 weeks", f"{_sum(last13, 'qty_total'):,}")
v3.metric("Last 52 weeks", f"{_sum(last52, 'qty_total'):,}")
avg_recent = (_sum(last13, "qty_total") / max(1, len(last13))) if len(last13) else 0
v4.metric("Avg/wk (last 13)", f"{avg_recent:,.0f}")
ch_total = _sum(last13, "qty_total") or 1
st.caption(
f"Channel split last 13 weeks - "
f"RCM **{_sum(last13, 'qty_retail'):,}** "
f"({_sum(last13, 'qty_retail')/ch_total*100:.0f}%) * "
f"WEB **{_sum(last13, 'qty_webshop'):,}** "
f"({_sum(last13, 'qty_webshop')/ch_total*100:.0f}%) * "
f"WS **{_sum(last13, 'qty_wholesale'):,}** "
f"({_sum(last13, 'qty_wholesale')/ch_total*100:.0f}%)"
)
st.divider()
# ── Per-child compute, then sum to parent ──
child_rows = []
total_base = 0.0
total_promo_avg = 0.0
total_past_events = 0
for s in skus:
rr = base_run_rate(sales, erp, s, lookback=26)
upl = suggest_uplift(sales, erp, s, float(discount))
hist = past_promos_for_sku(sales, erp, s)
base_avg_s = float(rr.get("avg", 0) or 0)
upl_s = float(upl.get("uplift", 0) or 0)
promo_avg_s = base_avg_s * upl_s
total_base += base_avg_s
total_promo_avg += promo_avg_s
total_past_events += len(hist)
child_rows.append({
"sku": s, "base_avg": base_avg_s, "uplift": upl_s,
"promo_avg": promo_avg_s, "past_n": len(hist),
})
parent_uplift = (total_promo_avg / total_base) if total_base else 0
promo_total_period = total_promo_avg * max(1, n_weeks)
c1, c2, c3, c4 = st.columns(4)
c1.metric("Parent base run-rate", f"{total_base:,.0f} u/wk",
help=f"Sum of {len(skus)} child base run-rates (RCM+WEB).")
c2.metric("Effective uplift",
f"{parent_uplift:.2f}x" if parent_uplift else "-",
help="Volume-weighted across children.")
c3.metric("Expected promo avg", f"{total_promo_avg:,.0f} u/wk",
help="Sum of (child base * child uplift at this discount).")
c4.metric(f"Total over {n_weeks} promo wks",
f"{promo_total_period:,.0f}",
help="Expected promo avg * promo duration.")
with st.expander(f"Per-child breakdown ({len(skus)} SKUs)"):
name_map = data.get("name_map", {})
tbl = pd.DataFrame([{
"SKU": r["sku"],
"Name": name_map.get(r["sku"], "")[:50],
"Base u/wk": f"{r['base_avg']:.0f}",
"Uplift": f"{r['uplift']:.2f}x" if r["uplift"] else "-",
"Promo u/wk": f"{r['promo_avg']:.0f}",
"Past AKCIJA": r["past_n"],
} for r in sorted(child_rows, key=lambda x: -x["promo_avg"])])
st.dataframe(tbl, use_container_width=True, hide_index=True)
# ── Parent chart: actuals (sum) + projection (sum of children) ──
actuals_tail = p_sorted.tail(12).copy()
actuals_tail["label"] = "CW" + actuals_tail["week"].astype(int).astype(str)
actuals_y = (actuals_tail["qty_retail"].astype(float)
+ actuals_tail["qty_webshop"].astype(float)).tolist()
fig = go.Figure()
fig.add_trace(go.Bar(
x=actuals_tail["label"].tolist(), y=actuals_y,
name="Actuals (RCM+WEB)",
marker_color="rgba(139,143,163,0.4)",
))
promo_weeks = list(range(sw_, ew_ + 1))
promo_qty = [total_promo_avg] * len(promo_weeks)
fig.add_trace(go.Scatter(
x=[f"CW{w}" for w in promo_weeks], y=promo_qty,
mode="lines+markers+text",
name="Promo projection (sum of children)",
text=[f"{int(q):,}" for q in promo_qty],
textposition="top center",
line=dict(color="#FF6B6B", width=3),
marker=dict(size=12, color="#FF6B6B"),
))
base_weeks = list(range(max(1, sw_ - 4), ew_ + 5))
fig.add_trace(go.Scatter(
x=[f"CW{w}" for w in base_weeks], y=[total_base] * len(base_weeks),
mode="lines", name="Base forecast (sum)",
line=dict(color="#60A5FA", width=1.5, dash="dash"),
))
post_weeks = list(range(ew_ + 1, ew_ + 4))
fig.add_trace(go.Scatter(
x=[f"CW{w}" for w in post_weeks],
y=[total_base * 0.85] * len(post_weeks),
mode="lines", name="Post-promo dip",
line=dict(color="#F97316", width=1.5, dash="dot"),
))
fig.update_layout(
plot_bgcolor="#1A1D27", paper_bgcolor="#1A1D27",
font=dict(color="#E8E9ED", size=11),
height=320, margin=dict(l=10, r=10, t=10, b=10),
legend=dict(orientation="h", y=-0.2),
xaxis=dict(showgrid=False),
yaxis=dict(gridcolor="#2A2D3A", title="Units / week (sum of children)"),
)
st.plotly_chart(fig, use_container_width=True,
key=f"parent_chart_{parent_key}")
st.caption(f"Past AKCIJA events across {len(skus)} children: "
f"{total_past_events} total. Drill into individual SKU "
f"sub-tabs below for per-child history.")
def _render_summary(items, name, source, ptype, outcome, channels,
sy, sw_, ey, ew_, n_weeks):
st.markdown(f"### 📊 Promotion summary — {name}")
st.caption(f"{source} · {ptype} · {outcome} · Channels: {', '.join(channels) or '—'}")
# ---- Median scenario totals ----
total_units = sum(it["total"] for it in items)
total_rev = sum(it["total"] * it["promo_price"] for it in items)
total_ruc = sum(it["total"] * it["ruc_unit_promo"] for it in items)
# ---- Cannibalization aggregates ----
cann_units = sum(it.get("cann_units", 0) for it in items)
cann_rev = sum(it.get("cann_revenue", 0) for it in items)
cann_ruc = sum(it.get("cann_ruc", 0) for it in items)
net_units = total_units - cann_units
net_rev = total_rev - cann_rev
net_ruc = total_ruc - cann_ruc
# ---- Upside (p90) scenario ----
upside_units = sum(it.get("upside_total", it["total"]) for it in items)
upside_rev = sum(it.get("upside_total", it["total"]) * it["promo_price"]
for it in items)
upside_ruc = sum(it.get("upside_total", it["total"]) * it["ruc_unit_promo"]
for it in items)
# ---- Top KPI row (median) ----
st.markdown("##### Median scenario")
k1, k2, k3, k4, k5 = st.columns(5)
k1.metric("Articles", len(items))
k2.metric("Period", f"CW{sw_}–CW{ew_}", f"{n_weeks} weeks")
k3.metric("Total units", f"{total_units:,}")
k4.metric("Revenue", f"€{total_rev:,.0f}")
k5.metric("RUC", f"€{total_ruc:,.0f}")
# ---- Cannibalization + Net row ----
st.markdown("##### Net category impact (after cannibalization)")
c1, c2, c3 = st.columns(3)
c1.metric("Net units",
f"{int(net_units):+,}",
delta=f"-{int(cann_units):,} cannibalized",
help="Promo gain − cannibalization of external siblings.")
c2.metric("Net revenue",
f"€{net_rev:+,.0f}",
delta=f"-€{cann_rev:,.0f}")
c3.metric("Net RUC",
f"€{net_ruc:+,.0f}",
delta=f"-€{cann_ruc:,.0f}")
# ---- Upside scenario ----
st.markdown("##### 🚀 Realistic upside (p90) scenario")
u1, u2, u3 = st.columns(3)
u1.metric("Upside units",
f"{int(upside_units):,}",
delta=f"+{int(upside_units - total_units):,} vs median")
u2.metric("Upside revenue",
f"€{upside_rev:,.0f}",
delta=f"+€{(upside_rev - total_rev):,.0f}")
u3.metric("Upside RUC",
f"€{upside_ruc:,.0f}",
delta=f"+€{(upside_ruc - total_ruc):,.0f}")
st.caption(
"p90 = empirical 90th-percentile uplift for the discount band across "
"3,107 historical AKCIJA events. Plan inventory for upside, expect median."
)
st.divider()
# ---- Group breakdown (if any items are grouped) ----
grouped_items = {}
ungrouped_items = []
for it in items:
g = it.get("group")
if g:
grouped_items.setdefault(g, []).append(it)
else:
ungrouped_items.append(it)
if grouped_items:
st.markdown("#### 📁 Grouped breakdown")
for gn, gitems in grouped_items.items():
g_units = sum(it["total"] for it in gitems)
g_rev = sum(it["total"] * it["promo_price"] for it in gitems)
g_ruc = sum(it["total"] * it["ruc_unit_promo"] for it in gitems)
with st.expander(f"📁 **{gn}** · {len(gitems)} SKU · "
f"{g_units:,} u · €{g_rev:,.0f} rev · "
f"€{g_ruc:,.0f} RUC", expanded=False):
rows_g = []
for it in gitems:
rows_g.append({
"SKU": it["sku"],
"Name": it["name"][:40],
"Discount %": it.get("discount", 0),
"Units": int(it["total"]),
"Revenue €": int(it["total"] * it["promo_price"]),
"RUC €": int(it["total"] * it["ruc_unit_promo"]),
})
st.dataframe(pd.DataFrame(rows_g),
use_container_width=True, hide_index=True)
st.markdown("#### 📦 Ungrouped & weekly detail")
# ---- Pivot table (per-SKU per-week) ----
rows = []
for it in items:
row = {"Group": it.get("group") or "—",
"SKU": it["sku"], "Name": it["name"][:40]}
for j in range(n_weeks):
row[f"CW{sw_+j}"] = it["weekly"][j] if j < len(it["weekly"]) else 0
row["TOTAL"] = it["total"]
rows.append(row)
if rows:
st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
# ---- Line chart — per-SKU + total + upside ----
weeks = [f"CW{sw_+j}" for j in range(n_weeks)]
weekly_totals = [sum(it["weekly"][j] for it in items if j < len(it["weekly"]))
for j in range(n_weeks)]
# Upside: same per-week shape but scaled to upside_total
weekly_upside = []
for j in range(n_weeks):
wk_sum = 0
for it in items:
if it["total"] > 0 and j < len(it["weekly"]):
scale = it.get("upside_total", it["total"]) / it["total"]
wk_sum += it["weekly"][j] * scale
weekly_upside.append(wk_sum)
fig = go.Figure()
# Upside band (dashed top)
fig.add_trace(go.Scatter(
x=weeks, y=weekly_upside,
mode="lines+markers",
line=dict(color="#FFA500", width=2, dash="dot"),
marker=dict(size=7),
name="Upside (p90) — total",
hovertemplate="%{x}: %{y:.0f} uUpside",
))
# Median total (solid)
fig.add_trace(go.Scatter(
x=weeks, y=weekly_totals,
mode="lines+markers",
line=dict(color="#FF6B6B", width=3),
marker=dict(size=9),
name="Median — total",
hovertemplate="%{x}: %{y:.0f} uTotal",
))
# Per-SKU thin lines
palette = ["#6C63FF", "#34D399", "#38BDF8", "#FBBF24",
"#A78BFA", "#F472B6", "#10B981", "#E8734A"]
for i, it in enumerate(items):
ys = [it["weekly"][j] if j < len(it["weekly"]) else 0
for j in range(n_weeks)]
fig.add_trace(go.Scatter(
x=weeks, y=ys,
mode="lines+markers",
line=dict(color=palette[i % len(palette)], width=1.5),
marker=dict(size=5),
name=f"{it['sku']}",
hovertemplate=f"%{{x}}: %{{y:.0f}} u{it['sku']}",
))
fig.update_layout(
plot_bgcolor="#1A1D27", paper_bgcolor="#1A1D27",
font=dict(color="#E8E9ED"), height=380,
title="Weekly units — per SKU + total (median + upside)",
margin=dict(l=10, r=10, t=50, b=40),
xaxis=dict(showgrid=False),
yaxis=dict(gridcolor="#2A2D3A", title="Units / week"),
legend=dict(orientation="h", y=-0.18),
hovermode="x unified",
)
st.plotly_chart(fig, use_container_width=True)
# Save
st.divider()
cS, cP, cI = st.columns([2, 2, 3])
with cS:
if st.button("🚀 Create & save promotion", type="primary",
use_container_width=True, key="planner_save"):
promo_id = uuid4().hex[:8]
now = datetime.now().strftime("%Y-%m-%d %H:%M")
rows_to_save = []
for it in items:
weekly_csv = ";".join(str(v) for v in it["weekly"])
rows_to_save.append({
"id": promo_id, "name": name,
"source": source, "promo_type": ptype, "outcome": outcome,
"target_units": "",
"start_year": sy, "start_week": sw_,
"end_year": ey, "end_week": ew_,
"discount_pct": it.get("discount", 0),
"mechanic": it.get("mechanic", ""),
"channels": "; ".join(channels),
"sku": it["sku"], "sku_name": it["name"],
"weekly_qty_csv": weekly_csv,
"total_units": it["total"],
"revenue_eur": round(it["total"] * it["promo_price"]),
"ruc_eur": round(it["total"] * it["ruc_unit_promo"]),
"created_at": now,
})
n = append_promotion(rows_to_save)
st.success(f"✅ Saved {n} SKU rows to data/cm_promotions.csv (promo id: {promo_id}).")
with cP:
if st.button("📤 Submit to Promo Calendar", type="secondary",
use_container_width=True, key="planner_submit_cal"):
try:
# Load the sibling Promo Calendar's promo_data under a unique
# name to avoid colliding with PromoTool's own promo_data.
import importlib.util
from pathlib import Path as _P
_cal_pd = _P(__file__).parent.parent / "promocalendar" / "promo_data.py"
_spec = importlib.util.spec_from_file_location("calendar_promo_data", _cal_pd)
_cpd = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_cpd)
_cal_add_promo = _cpd.add_promo
total_units = sum(int(it.get("total", 0) or 0) for it in items)
avg_disc = (
round(sum(float(it.get("discount", 0) or 0) for it in items) / len(items))
if items else 0
)
sku_list = ", ".join(it["sku"] for it in items)
new_id = _cal_add_promo({
"name": name,
"source": source,
"type": ptype,
"outcome": outcome,
"status": "💡 idea", # Submitted as draft / for approval
"start_year": sy, "start_week": sw_,
"end_year": ey, "end_week": ew_,
"skus": sku_list,
"units": total_units,
"category": "",
"owner": "Promo Tool",
"notes": (
f"Submitted from Promo Tool · {len(items)} SKU(s) · "
f"avg discount {avg_disc}% · channels: {'; '.join(channels)}"
),
})
st.success(
f"📤 Submitted to Promo Calendar (id `{new_id}`, status 💡 idea). "
f"Direktor nabave će ga vidjeti u Nabava queue / Unified view."
)
except Exception as e:
st.error(f"Submit failed: {e}")
with cI:
st.caption(
"**Create & save** → lokalni CSV za retro-analizu. \n"
"**Submit to Promo Calendar** → šalje prijedlog direktoru nabave u "
"Promo Calendar (status 💡 idea, čeka odobrenje)."
)