"""Data loaders & uplift engine for the Promo Tool. All caches use Streamlit's @st.cache_data so files only get re-read when they change on disk. Looks for inputs in ../data/ (the main project's data folder) so the tool stays in sync with the production data. """ from __future__ import annotations from datetime import datetime from pathlib import Path import numpy as np import pandas as pd import streamlit as st # Parent-project data folder. PromoTool sits next to it. DATA_DIR = Path(__file__).parent.parent / "data" PROMO_DIR = Path(__file__).parent / "data" PROMO_DIR.mkdir(exist_ok=True) CM_PROMO_FILE = PROMO_DIR / "cm_promotions.csv" PROMO_MECHANICS = [ "Discount %", "1+1 gratis", "2+1 gratis", "3+1 gratis", "Other", ] # ============================================================ # Unit-size parsing + price-point benchmarking # # Forecast model is multiplicative on baseline — for first-time promos # on hero/anchor SKUs at attractive PRICE POINTS, this systematically # under-predicts. We add a price-disruptor multiplier driven by €/100g # (or €/100ml) vs. the SKU's category benchmark. # ============================================================ import re as _re _SIZE_PATTERNS = [ (_re.compile(r"(\d+(?:[.,]\d+)?)\s*kg\b", _re.I), 1000.0, "g"), (_re.compile(r"(\d+(?:[.,]\d+)?)\s*g\b", _re.I), 1.0, "g"), (_re.compile(r"(\d+(?:[.,]\d+)?)\s*l\b", _re.I), 1000.0, "ml"), (_re.compile(r"(\d+(?:[.,]\d+)?)\s*ml\b", _re.I), 1.0, "ml"), ] def parse_flavor(name: str) -> str | None: """Extract the primary flavor token from a product name. Returns the first matching token from _FLAVOUR_TOKENS or None if none found. Lowercased. Skips structural suffix tokens (LG/MD/SM, pack/bundle…) that are in _FLAVOUR_TOKENS but aren't real flavors. Examples: 'Polleo HydroX Whey 908g Chocolate Gourmet' → 'chocolate'. 'Optimum Whey 2 kg' → None.""" if not name: return None # Real flavor tokens — subset of _FLAVOUR_TOKENS without size/pack tags. _SKIP = {"lg", "md", "sm", "xl", "xxl", "xs", "pack", "bundle", "set", "edition", "limited", "double", "wild", "mixed", "creamy", "smooth", "chunky", "natural", "plain", "salted", "crunchy", "gourmet", "choc", "birthday", "cake", "cream"} import re as _re_local words = _re_local.findall(r"[a-zA-ZčćšžđČĆŠŽĐ]+", str(name).lower()) for w in words: if w in _FLAVOUR_TOKENS and w not in _SKIP: return w return None def flavor_compatibility(a: str | None, b: str | None) -> float: """Switching probability between two flavors. Returns 0..1. Premise: customers strongly prefer their flavor. Discount-driven switching across flavors is much rarer than switching across brands within the same flavor. same known flavor → 1.00 (full substitute) different known → 0.40 (some switching, not full) unknown / flavorless → 0.70 (neutral, can't tell) """ if not a or not b: return 0.70 if a == b: return 1.00 return 0.40 def parse_unit_size(name: str) -> tuple[float, str]: """Extract pack size from a product name. Returns (value_in_base_unit, unit) where unit is "g" or "ml". Returns (0, "") if not found. Examples: "Polleo Premium HydroX Whey 908g …" → (908, "g"). "Polleo NextGen Protein 2 kg Chocolate" → (2000, "g"). "Abe Can Orange Burst 330ml" → (330, "ml").""" if not name: return 0.0, "" for pat, scale, unit in _SIZE_PATTERNS: m = pat.search(str(name)) if m: try: return float(m.group(1).replace(",", ".")) * scale, unit except ValueError: continue return 0.0, "" @st.cache_data(ttl=300) def category_price_benchmark(_plan: pd.DataFrame, _prices: dict, cat: str, unit: str) -> dict: """Compute €/100g (or €/100ml) distribution for all SKUs in this category that share the given unit. Cached because plan and prices rarely change within a session. Returns: {p25, median, p75, n_skus, unit}. Empty dict if not enough samples (< 5 SKUs) or no parseable sizes.""" if _plan is None or _plan.empty or not cat or not unit: return {} if "cat" not in _plan.columns or "name" not in _plan.columns: return {} sub = _plan[_plan["cat"].astype(str) == str(cat)] if sub.empty: return {} pts = [] for _, r in sub.iterrows(): sku = str(r["sku"]) size_val, size_unit = parse_unit_size(str(r.get("name", ""))) if size_val <= 0 or size_unit != unit: continue price = float(_prices.get(sku, 0) or 0) if price <= 0: continue per100 = price * 100.0 / size_val # €/100g or €/100ml pts.append(per100) if len(pts) < 5: return {} arr = np.array(sorted(pts)) return { "p25": float(np.percentile(arr, 25)), "median": float(np.percentile(arr, 50)), "p75": float(np.percentile(arr, 75)), "min": float(arr[0]), "max": float(arr[-1]), "n_skus": int(len(arr)), "unit": unit, } def price_disruptor_multiplier(promo_per100: float, bench: dict) -> tuple[float, str]: """Multiplier on predicted uplift from promo €/100u position vs category benchmark. Tuned aggressively because the historical data is thin in the disruptor zone (n=25 events <0.5× cat p25) and under-represents the actual scale of price-driven demand jumps that Polleo's CMs report seeing (e.g. 2 kg whey at €40 selling 400 in 12 days vs. baseline 14/wk). Zones (relative to cat p25): ≥ 1.0× p25 → 1.00× (above p25 — no kicker) 0.7–1.0× → 1.0..1.5× (mild) 0.5–0.7× → 3.0..4.0× (real disruptor zone) < 0.5× → 4.0..7.0× (extreme price-point disruption) No upper cap — caller handles ceiling if needed. """ if not bench or promo_per100 <= 0: return 1.0, "" p25 = bench["p25"] if p25 <= 0: return 1.0, "" ratio = promo_per100 / p25 if ratio >= 1.0: return 1.0, "above cat p25" if ratio >= 0.7: share = (1.0 - ratio) / 0.3 # 0..1 return round(1.0 + share * 0.5, 3), "near cat p25" if ratio >= 0.5: # 0.5–0.7× — real disruptor zone share = (0.7 - ratio) / 0.2 return round(3.0 + share * 1.0, 3), "price-disruptor (below cat p25)" # < 0.5× p25 — extreme price-point disruption deficit = (0.5 - ratio) / 0.5 # 0..1, clamp at 1 deficit = min(1.0, deficit) return round(4.0 + deficit * 3.0, 3), "extreme price-disruptor (≪ cat p25)" def discount_band_uplift_curve(discount_pct: float) -> float: """Piecewise non-linear curve calibrated against 3,107 historical AKCIJA events (see analyze_promo_calibration.py). Values are the *observed median uplifts* per discount band; the engine multiplies base predict by `curve(d) / curve(reference_discount)` so it stays multiplicative. 0..15 % → 0.85..1.05 (shallow — obs median 1.02) 15..25 % → 1.05..1.50 (obs median 1.50) 25..35 % → 1.50..2.10 (obs median 2.00 — inflection) 35..45 % → 2.10..2.50 (obs median 1.77, sparse n=30) 45..60 % → 2.50..3.50 (obs median 3.36) >60 % → 3.50..5.00 (clearance / liquidation) """ d = max(0.0, float(discount_pct)) if d <= 15: return 0.85 + (d / 15) * 0.20 if d <= 25: return 1.05 + ((d - 15) / 10) * 0.45 if d <= 35: return 1.50 + ((d - 25) / 10) * 0.60 if d <= 45: return 2.10 + ((d - 35) / 10) * 0.40 if d <= 60: return 2.50 + ((d - 45) / 15) * 1.00 return min(5.00, 3.50 + (d - 60) / 40 * 1.50) def upside_ratio_for_band(discount_pct: float) -> float: """Empirical p90/median uplift ratio per discount band. Used to surface the 'realistic upside' alongside the median forecast — so a 2× median prediction with 2.8× ratio shows users that 5.6× is a plausible top-of-range outcome (not promise). Calibrated from same dataset: 0-15% → 2.00 (median 1.00, p90 2.00) 15-25% → 2.33 (1.50 → 3.50) 25-35% → 2.84 (2.00 → 5.68) 35-45% → 3.15 (1.77 → 5.58 — sparse) 45-60% → 1.84 (3.36 → 6.18) >60% → 4.72 (6.40 → 30.21) """ d = max(0.0, float(discount_pct)) if d <= 15: return 2.00 if d <= 25: return 2.33 if d <= 35: return 2.84 if d <= 45: return 3.15 if d <= 60: return 1.84 return 4.72 @st.cache_data(ttl=3600) def historical_sibling_drop_pct(_sales: pd.DataFrame, _erp: pd.DataFrame, focus_sku: str, sibling_sku: str, fallback_pct: float = 0.15) -> dict: """How much did `sibling_sku` retail+web sales drop during past promos of `focus_sku`? For each ERP-flagged promo week of `focus_sku`, compute the sibling's qty_retail + qty_webshop during that week, then compare to the sibling's CLEAN baseline (excluding its own promo / bleed / discount-anomaly weeks). Returns: {drop_pct, source, n_observations, weeks: [(yw, drop_pct)]} Where `drop_pct` is the median observed drop (0.25 = 25 % drop). Falls back to `fallback_pct` if no usable history. Negative drops (sibling went UP during focus promo — co-promo or unrelated lift) are kept in the median calculation so the signal stays honest. Wholesale is excluded — we only compare retail+web channels. """ if _sales is None or _erp is None or len(_sales) == 0 or len(_erp) == 0: return {"drop_pct": fallback_pct, "source": "fallback (no data)", "n_observations": 0, "weeks": []} # Past promo weeks of focus SKU from ERP calendar focus_promos = _erp[(_erp["sku"] == focus_sku) & (_erp["is_erp_promo"] == 1)] promo_yws = sorted(set(zip(focus_promos["year"].astype(int), focus_promos["week"].astype(int)))) if not promo_yws: return {"drop_pct": fallback_pct, "source": f"fallback {fallback_pct*100:.0f}% (no past promos of focus SKU)", "n_observations": 0, "weeks": []} # Sibling's clean baseline (same logic as base_run_rate but inline # so we don't depend on it computing for this specific sibling) sib_sales = _sales[_sales["sku"] == sibling_sku] if len(sib_sales) == 0: return {"drop_pct": fallback_pct, "source": "fallback (no sibling sales data)", "n_observations": 0, "weeks": []} sib_erp_promos = set() sib_erp_rows = _erp[(_erp["sku"] == sibling_sku) & (_erp["is_erp_promo"] == 1)] if len(sib_erp_rows): sib_erp_promos = set(zip(sib_erp_rows["year"].astype(int), sib_erp_rows["week"].astype(int))) sib_bleed = set() for (y, w) in sib_erp_promos: sib_bleed.add((y, w + 1)) sib_bleed.add((y, w + 2)) sib_bleed -= sib_erp_promos clean_qtys = [] for _, r in sib_sales.iterrows(): yw = (int(r["year"]), int(r["week"])) if yw in sib_erp_promos or yw in sib_bleed: continue if int(r.get("is_any_promo", 0) or 0): 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 > 5.0 or wd > 5.0: continue q = float(r.get("qty_retail", 0) or 0) + float(r.get("qty_webshop", 0) or 0) if q > 0: clean_qtys.append(q) if not clean_qtys: return {"drop_pct": fallback_pct, "source": "fallback (no clean baseline weeks for sibling)", "n_observations": 0, "weeks": []} baseline = sum(clean_qtys) / len(clean_qtys) if baseline <= 0: return {"drop_pct": fallback_pct, "source": "fallback (sibling baseline is 0)", "n_observations": 0, "weeks": []} # For each focus-promo week, measure sibling drop weeks_observed = [] for (y, w) in promo_yws: row = sib_sales[(sib_sales["year"] == y) & (sib_sales["week"] == w)] if len(row) == 0: continue r = row.iloc[0] sib_qty_during = float(r.get("qty_retail", 0) or 0) + float(r.get("qty_webshop", 0) or 0) # Drop = (baseline - actual) / baseline ; positive = sales dropped drop = (baseline - sib_qty_during) / baseline weeks_observed.append((y * 100 + w, drop)) if not weeks_observed: return {"drop_pct": fallback_pct, "source": f"fallback {fallback_pct*100:.0f}% (no sibling sales during focus promo weeks)", "n_observations": 0, "weeks": []} drops_sorted = sorted([d for (_, d) in weeks_observed]) median_drop = drops_sorted[len(drops_sorted) // 2] return { "drop_pct": float(median_drop), "source": f"historical median ({len(weeks_observed)} past promo weeks)", "n_observations": len(weeks_observed), "weeks": weeks_observed, "sibling_baseline_wk": baseline, } def first_time_uplift_cap(n_history: int, discount_pct: float) -> float: """Adaptive ceiling for predicted uplift. • SKU has rich history (≥3) → 1.5× max historical (current rule) • SKU has 1-2 events → 2× max historical (looser) • SKU has NO history + aggressive depth (≥30%) → 15× absolute (first-time price-disruptor case) • SKU has no history + shallow → 4× absolute """ if n_history >= 3: return -1.0 # caller uses max_hist × 1.5 if n_history >= 1: return -2.0 # caller uses max_hist × 2.0 # No history at all — absolute cap return 15.0 if discount_pct >= 30 else 4.0 # ---------- Mechanic detection from raw RCM transactions ---------- # # When data/sales_transactions.csv is available, the tool can read the # distribution of basket sizes (qty per RCM document) during a promo # period and infer the mechanic with high confidence: # # Dominant qty=2 → 1+1 gratis # Dominant qty=3 → 2+1 gratis # Dominant qty=4 → 3+1 gratis # # Required schema for data/sales_transactions.csv (case-insensitive): # sku, date (YYYY-MM-DD), qty, tip (RCM, WSA, WSB, WSC, WSD) # Only RCM + webshop tips are used for the bundle detection (those are # the only channels that run bundle mechanics — wholesale is separate). # # Without that file, the tool falls back to discount-band heuristic. TX_FILE = Path(__file__).parent.parent / "data" / "sales_transactions.csv" @st.cache_data(ttl=3600) def load_transactions() -> pd.DataFrame: """Load raw RCM + webshop transactions if available. Returns empty DataFrame if file missing.""" if not TX_FILE.exists(): return pd.DataFrame() try: df = pd.read_csv(TX_FILE) df.columns = [str(c).strip().lower() for c in df.columns] if "date" in df.columns: df["date"] = pd.to_datetime(df["date"], errors="coerce") df = df.dropna(subset=["date"]) iso = df["date"].dt.isocalendar() df["year"] = iso["year"].astype(int) df["week"] = iso["week"].astype(int) # Filter to bundle-eligible channels: RCM (retail) + WSA-WSD (web) if "tip" in df.columns: df = df[df["tip"].astype(str).str.upper().isin( ["RCM", "WSA", "WSB", "WSC", "WSD"])] return df except Exception: return pd.DataFrame() def detect_mechanic_from_transactions(_tx: pd.DataFrame, sku: str, start_yw: int, end_yw: int) -> dict: """Look at RCM/web baskets for this SKU during the promo period and infer the mechanic from dominant qty-per-document. Returns {'mechanic', 'confidence_pct', 'n_docs', 'qty_distribution'}. Confidence is the % of documents matching the inferred mechanic. If <40 % of docs match any bundle pattern → 'Straight discount %'. Empty result if no transactions or too few documents. """ if _tx is None or _tx.empty: return {} sub = _tx[(_tx["sku"].astype(str) == str(sku))] if sub.empty or "qty" not in sub.columns: return {} yw = sub["year"].astype(int) * 100 + sub["week"].astype(int) sub = sub[(yw >= start_yw) & (yw <= end_yw)] if len(sub) < 5: return {} qtys = pd.to_numeric(sub["qty"], errors="coerce").dropna().astype(int) qtys = qtys[qtys > 0] if len(qtys) < 5: return {} n_docs = len(qtys) counts = qtys.value_counts(normalize=True) dist = {int(k): float(v) for k, v in counts.items()} # Bundle inference — dominant qty must be > 40 % of docs to qualify for bundle_qty, mech_label in [(4, "3+1 gratis"), (3, "2+1 gratis"), (2, "1+1 gratis")]: share = dist.get(bundle_qty, 0) if share >= 0.40: return { "mechanic": mech_label, "confidence_pct": round(share * 100, 1), "n_docs": n_docs, "qty_distribution": {k: round(v * 100, 1) for k, v in sorted(dist.items())[:6]}, } return { "mechanic": "Discount % (straight)", "confidence_pct": round(dist.get(1, 0) * 100, 1), "n_docs": n_docs, "qty_distribution": {k: round(v * 100, 1) for k, v in sorted(dist.items())[:6]}, } def guess_mechanic_from_discount(discount_pct: float) -> str: """Reverse-engineer the likely promo mechanic from the observed discount %, since ERP promo calendar only carries campaign names not mechanic tags. Returns a tag like '1+1 gratis (likely)' or 'Discount %' when none of the bundle bands match. """ try: d = float(discount_pct or 0) except (TypeError, ValueError): return "" if d <= 0: return "" if 47 <= d <= 53: return "1+1 gratis (likely)" if 30 <= d <= 36: return "2+1 gratis (likely)" if 22 <= d <= 28: return "3+1 gratis (likely)" if 18 <= d < 22: return "4+1 gratis (likely)" return f"Discount {d:.0f}% (straight)" def effective_discount(mechanic: str, user_pct: float) -> float: """Convert a promo mechanic to its effective per-unit discount %. Bundle math (all units in the basket get the effective discount when qualified — but historical uplift covers the whole period including non-qualifying baskets, so we apply the effective % as the average discount across all units sold during the campaign): 1+1 gratis → 50.0 % (1 free out of 2 = 50%) 2+1 gratis → 33.33 % (1 free out of 3) 3+1 gratis → 25.0 % (1 free out of 4) For "Discount %", returns user-entered %. For "Other", returns user-entered % (CM enters manually). """ m = (mechanic or "").strip().lower() if "1+1" in m: return 50.0 if "2+1" in m: return round(100.0 / 3.0, 2) # 33.33 if "3+1" in m: return 25.0 if "4+1" in m: return 20.0 # Discount %, Other, or anything else → use user's number try: return float(user_pct or 0) except (TypeError, ValueError): return 0.0 # ---------- NC30 (najniža cijena u zadnjih 30 dana) ---------- NC30_FILE = Path(__file__).parent.parent / "data" / "nc30.csv" @st.cache_data(ttl=3600) def load_nc30() -> dict: """Return {sku: nc30_price} from data/nc30.csv. Required columns: sku, nc30_price (any case).""" if not NC30_FILE.exists(): return {} try: df = pd.read_csv(NC30_FILE) df.columns = [str(c).strip().lower() for c in df.columns] if "sku" not in df.columns or "nc30_price" not in df.columns: return {} return dict(zip(df["sku"].astype(str), pd.to_numeric(df["nc30_price"], errors="coerce").fillna(0))) except Exception: return {} def check_nc30(sku: str, planned_promo_price: float) -> dict: """Compare planned promo price to NC30. Returns dict with `ok` flag and a human-readable message. If NC30 data is missing for the SKU, returns has_data=False and the check is skipped (treated as OK).""" nc30_map = load_nc30() nc30 = float(nc30_map.get(sku, 0) or 0) if nc30 <= 0: return {"has_data": False, "nc30": 0, "ok": True, "delta_pct": 0, "message": "NC30 reference not available — upload data/nc30.csv to enable check."} delta_pct = ((planned_promo_price - nc30) / nc30 * 100) if nc30 > 0 else 0 ok = planned_promo_price <= nc30 + 0.01 # 1c tolerance for rounding return { "has_data": True, "nc30": round(nc30, 2), "ok": ok, "delta_pct": round(delta_pct, 1), "message": ( f"✅ Promo price €{planned_promo_price:.2f} ≤ NC30 €{nc30:.2f}. OK to run." if ok else f"❌ Promo price €{planned_promo_price:.2f} is **{delta_pct:+.1f}%** " f"above NC30 €{nc30:.2f} — cannot run this promo (lowest-30d rule)." ), } def _read_csv(name: str) -> pd.DataFrame: p = DATA_DIR / name if not p.exists(): return pd.DataFrame() df = pd.read_csv(p) df.columns = [str(c).strip().lower() for c in df.columns] return df @st.cache_data(ttl=3600) def load_all(): """Return a dict of all source dataframes + lookup maps.""" out = {} out["sales"] = _read_csv("sales_clean.csv") out["plan"] = _read_csv("sku_plan_list.csv") out["prices"] = _read_csv("sku_prices.csv") out["costs"] = _read_csv("sku_costs.csv") out["erp"] = _read_csv("erp_promo_calendar.csv") out["subcat"] = _read_csv("sku_subcat_map.csv") out["stock"] = _read_csv("stock.csv") out["incoming"] = _read_csv("incoming_supply.csv") out["supply_master"] = _read_csv("supply_master.csv") out["uplift"] = _read_csv("sku_uplift.csv") # Maps plan = out["plan"] out["name_map"] = dict(zip(plan["sku"], plan.get("name", ""))) if "sku" in plan.columns else {} out["cat_map"] = dict(zip(plan["sku"], plan.get("cat", ""))) if "sku" in plan.columns else {} out["tier_map"] = dict(zip(plan["sku"], plan.get("oznaka", ""))) if "sku" in plan.columns else {} out["xyz_map"] = dict(zip(plan["sku"], plan.get("total_xyz", ""))) if "sku" in plan.columns else {} out["wsshare_map"] = dict(zip(plan["sku"], plan.get("ws_share_26w", 0))) if "sku" in plan.columns else {} sub = out["subcat"] out["subcat_map"] = dict(zip(sub["sku"], sub.get("sub_cat", ""))) if "sku" in sub.columns else {} prices = out["prices"] # Use NORMAL retail PPP (regular MP price), not blended avg_sell_price. # Promotions are applied as discount off the normal retail price. if "sku" in prices.columns and "normal_retail_ppp" in prices.columns: out["price_map"] = dict(zip(prices["sku"], prices["normal_retail_ppp"])) elif "sku" in prices.columns: out["price_map"] = dict(zip(prices["sku"], prices.get("avg_sell_price", 0))) else: out["price_map"] = {} # Webshop normal price kept separately if needed if "sku" in prices.columns and "normal_webshop_ppp" in prices.columns: out["price_web_map"] = dict(zip(prices["sku"], prices["normal_webshop_ppp"])) else: out["price_web_map"] = {} costs = out["costs"] out["cost_map"] = dict(zip(costs["sku"], costs.get("cost_price", 0))) if "sku" in costs.columns else {} out["ruc_unit_map"] = dict(zip(costs["sku"], costs.get("ruc", 0))) if "sku" in costs.columns else {} return out def _retail_qty(row_or_df): """Sum of qty_retail + qty_webshop only (NOT qty_wholesale). These are the channels promotions actually run on (RCM + WEB).""" r = row_or_df.get("qty_retail", 0) w = row_or_df.get("qty_webshop", 0) if "qty_webshop" in row_or_df else 0 return r + w @st.cache_data(ttl=3600) def base_run_rate(_sales: pd.DataFrame, _erp: pd.DataFrame, sku: str, n_weeks: int = 4, lookback: int = 13) -> dict: """Baseline (RCM + WEB) = simple average of the most recent `n_weeks` non-promo weeks within the last `lookback` weeks. ONLY exclusion: ERP promo calendar (`is_erp_promo=1` for this SKU). Zero-sale weeks are still skipped so OOS doesn't pull baseline to 0, but no statistical / discount / post-promo-bleed filters are applied — the user's rule is "anything not in ERP promo calendar is fair game, and 3 months back is all that's relevant." Defaults: 4 weeks of average, 13-week lookback. """ if _sales is None or len(_sales) == 0 or "sku" not in _sales.columns: return {"avg": 0, "weeks": 0, "yws": [], "excluded": {}} sub = _sales[_sales["sku"] == sku].sort_values(["year", "week"], ascending=False) if len(sub) == 0: return {"avg": 0, "weeks": 0, "yws": [], "excluded": {}} sub = sub.head(lookback) # ERP promo weeks for this SKU promo_yws = set() if _erp is not None and len(_erp) > 0 and "sku" in _erp.columns: e = _erp[_erp["sku"] == sku] if len(e): promo_yws = set(zip(e["year"].astype(int), e["week"].astype(int))) excluded_promo = 0 excluded_zero = 0 # Walk most-recent first, collect up to n_weeks non-promo weeks candidates = [] for _, r in sub.iterrows(): y, w = int(r["year"]), int(r["week"]) if (y, w) in promo_yws: excluded_promo += 1 continue rq = float(r.get("qty_retail", 0)) + float(r.get("qty_webshop", 0)) if rq <= 0: excluded_zero += 1 continue candidates.append((y, w, rq)) if len(candidates) >= n_weeks: break if not candidates: return {"avg": 0, "weeks": 0, "yws": [], "excluded": {"promo": excluded_promo, "bleed": 0, "flag": 0, "disc": 0, "zero": excluded_zero, "oos": 0}} baseline = sum(q for (_, _, q) in candidates) / len(candidates) yws_sorted = [(y, w) for (y, w, _) in candidates] # already most-recent first return { "avg": baseline, "weeks": len(candidates), "yws": yws_sorted, "upper_half_pre_oos": baseline, # legacy field — caller may reference "excluded": { "promo": excluded_promo, "bleed": 0, "flag": 0, "disc": 0, "zero": excluded_zero, "oos": 0, }, } def _is_monthly_akcija(ptype: str, duration: int) -> bool: """Monthly campaign filter: - duration 3 or 4 weeks (typical AKCIJA), AND/OR - primary type contains 'AKCIJA' / 'AKCIJE' Multi-week (>4) campaigns and weekly flash promos are excluded — those don't represent the monthly AKCIJA pattern CMs plan around. """ if duration in (3, 4): return True label = (ptype or "").upper() if "AKCIJA" in label or "AKCIJE" in label: return True return False @st.cache_data(ttl=3600) def promo_pattern_for_sku(_sales: pd.DataFrame, _erp: pd.DataFrame, sku: str, target_duration: int, current_baseline: float, target_discount: float, plan_df: pd.DataFrame) -> dict: """Analog-method weekly qty prediction for an upcoming promo. Picks the most recent past AKCIJA (3-4 wk) for this SKU. If none, falls back to the most recent AKCIJA from any sub-category sibling. Returns: - `weekly_qty`: list of length target_duration, each week's predicted qty (RCM+WEB, integer). Uses the analog's weekly SHAPE, scaled by (current_baseline / analog_baseline) and adjusted for discount delta. - `analog`: metadata about which past event was used. - `pattern_norm`: normalised analog shape (each week / analog_avg). - `derived_uplift`: implied total uplift = sum(weekly_qty) / ( target_duration × current_baseline). If no analog exists at all, falls back to a typical AKCIJA ramp-peak- decline shape × 1.5 default uplift. This is the industry-standard 'analog method' used by big retail — each week is predicted differently to capture build-up, peak, saturation and end-push instead of a flat line. """ own = past_promos_for_sku(_sales, _erp, sku) own = [e for e in own if e["uplift"] > 0] def _weekly_qty_from_event(ev: dict) -> list: """Return analog's per-week qty with OOS weeks REMOVED. OOS weeks (zero-sale or sharp-drop) are excluded so the shape reflects real demand, not stockout-constrained sales. """ per_q = ev.get("per_week_qty", []) per_oos = ev.get("per_week_oos", [False] * len(per_q)) clean = [q for q, oos in zip(per_q, per_oos) if not oos] return clean analog = None analog_weekly = [] source = "" # 1. Own SKU's most recent AKCIJA — try most recent first, then fall # back to older ones if the most recent had too many OOS weeks. if own: for candidate in own: cand_weekly = _weekly_qty_from_event(candidate) if len(cand_weekly) >= 2: analog = candidate analog_weekly = cand_weekly oos_note = "" if candidate.get("oos_suspected"): n_oos = candidate.get("oos_weeks", 0) oos_note = f" — OOS in {n_oos} week(s) dropped" source = (f"Analog: own SKU's CW{analog['start_week']}-CW{analog['end_week']} " f"({analog['year']}) AKCIJA{oos_note}") break # 2. Category sibling fallback if not analog_weekly and plan_df is not None and len(plan_df): sib_cat = "" m = plan_df[plan_df["sku"].astype(str) == sku] if not m.empty and "cat" in plan_df.columns: sib_cat = str(m.iloc[0]["cat"]) if sib_cat: sib_skus = plan_df[plan_df["cat"] == sib_cat]["sku"].astype(str).tolist() best = None for s in sib_skus[:50]: if s == sku: continue evs = past_promos_for_sku(_sales, _erp, s) evs = [e for e in evs if e["uplift"] > 0] # Pick siblings with usable shape (≥ 2 non-OOS weeks) evs = [e for e in evs if len(_weekly_qty_from_event(e)) >= 2] if evs and (best is None or evs[0]["year"] * 100 + evs[0]["end_week"] > best[1]["year"] * 100 + best[1]["end_week"]): best = (s, evs[0]) if best: analog = best[1] analog_weekly = _weekly_qty_from_event(analog) source = (f"Analog: category sibling {best[0]} CW{analog['start_week']}-" f"CW{analog['end_week']} ({analog['year']}) AKCIJA") # 3. No analog — generic ramp-peak-decline shape, calibrated uplift if not analog_weekly: if target_duration >= 4: shape = [0.80, 1.20, 1.10, 0.90] elif target_duration == 3: shape = [0.90, 1.20, 0.90] elif target_duration == 2: shape = [1.10, 0.90] else: shape = [1.0] * target_duration # Pull calibrated uplift from suggest_uplift — single source of # truth, picks up band curve + price-disruptor + adaptive cap. upl_calc = suggest_uplift(_sales, _erp, sku, target_discount) predicted_uplift = upl_calc["uplift"] promo_avg_per_week = current_baseline * predicted_uplift # Normalise shape to avg ~1.0 so SHAPE × LEVEL is clean s_avg = sum(shape) / len(shape) if shape else 1.0 pat = [s / s_avg for s in shape[:target_duration]] if s_avg > 0 else shape[:target_duration] weekly_qty = [round(promo_avg_per_week * p) for p in pat] return { "weekly_qty": weekly_qty, "source": (f"No analog — calibrated prior: {upl_calc['source']}"), "analog": None, "pattern_norm": [round(p, 2) for p in pat], "derived_uplift": round(sum(weekly_qty) / max(1, target_duration * current_baseline), 2) if current_baseline > 0 else 0, "predicted_uplift": round(predicted_uplift, 2), "promo_avg_per_week": round(promo_avg_per_week, 1), "analog_baseline": 0, "analog_discount": None, "scale_baseline": 1.0, "scale_discount": 1.0, } # ---- Decompose: LEVEL from history, SHAPE generic ---- # Per CM feedback: copying weekly shape from the most recent analog is # NOT robust — OOS / shadow-promo weeks in the analog distort the shape. # Instead, use a GENERIC normalised ramp-peak-decline shape (industry # default for monthly AKCIJA: build-up → peak → saturation → end-push). # The LEVEL (per-week average) still comes from past history via # baseline × predicted_uplift, so per-SKU magnitude is right. analog_avg = sum(analog_weekly) / len(analog_weekly) if analog_weekly else 0 if target_duration >= 4: # 4-week AKCIJA: classic ramp → peak → saturation → end-push generic_shape = [0.80, 1.20, 1.10, 0.90] elif target_duration == 3: generic_shape = [0.90, 1.20, 0.90] elif target_duration == 2: generic_shape = [1.10, 0.90] else: generic_shape = [1.0] * target_duration # Always renormalise so avg ≈ 1.0 (LEVEL × shape stays clean) s_avg = sum(generic_shape) / len(generic_shape) pattern_norm = [s / s_avg for s in generic_shape] if s_avg > 0 else generic_shape # Predicted uplift — pull from suggest_uplift to keep ONE source of # truth. The chart / per-week qty and the Uplift metric in the UI now # always show consistent numbers; calibration (band curve, price- # disruptor, adaptive cap) is applied uniformly. upl_calc = suggest_uplift(_sales, _erp, sku, target_discount) predicted_uplift = upl_calc["uplift"] recent_disc = analog.get("discount_pct", 0) or 20.0 # LEVEL = baseline × predicted_uplift (per-week average across promo) promo_avg_per_week = current_baseline * predicted_uplift # Apply SHAPE (pattern_norm averages to ~1.0) weekly_qty = [round(promo_avg_per_week * p) for p in pattern_norm] derived_uplift = (sum(weekly_qty) / (target_duration * current_baseline)) if current_baseline > 0 else 0 return { "weekly_qty": weekly_qty, "source": source, "analog": analog, "pattern_norm": [round(p, 2) for p in pattern_norm], "derived_uplift": round(derived_uplift, 2), "predicted_uplift": round(predicted_uplift, 2), "promo_avg_per_week": round(promo_avg_per_week, 1), "analog_baseline": round(analog["avg_base"], 1) if analog["avg_base"] else 0, "analog_discount": round(recent_disc, 1) if recent_disc else None, "scale_baseline": 1.0, # decoupled now — level is from baseline×uplift "scale_discount": 1.0, # baked into calibrated uplift already } def _detect_shadow_promo_weeks(_sales: pd.DataFrame, sku: str, spike_mult: float = 2.5): """Find weeks where retail+web qty spiked ≥ spike_mult × trailing 4-week median. Returns list of (year, week, spike_x, discount_pct) — these look like promos that aren't in the ERP calendar. """ if _sales is None or len(_sales) == 0: return [] sub = _sales[_sales["sku"] == sku].sort_values(["year", "week"]) if len(sub) < 8: return [] rows = sub.to_dict("records") out = [] import numpy as _np for i, r in enumerate(rows): qty = float(r.get("qty_retail", 0)) + float(r.get("qty_webshop", 0)) if qty <= 0: continue # Trailing 4 weeks (excluding current) prev = [ float(rows[j].get("qty_retail", 0)) + float(rows[j].get("qty_webshop", 0)) for j in range(max(0, i - 4), i) ] prev_pos = [q for q in prev if q > 0] if len(prev_pos) < 2: continue med = float(_np.median(prev_pos)) if med <= 0: continue spike = qty / med if spike >= spike_mult: disc = float(r.get("retail_discount_pct", 0) or 0) out.append({ "year": int(r["year"]), "week": int(r["week"]), "qty": int(qty), "median_prev": round(med, 1), "spike_x": round(spike, 2), "discount_pct": round(disc, 1), }) return out @st.cache_data(ttl=3600) def past_promos_for_sku(_sales: pd.DataFrame, _erp: pd.DataFrame, sku: str) -> list: """Return list of past MONTHLY AKCIJA events for this SKU. Filters: - duration 3-4 weeks OR primary type contains 'AKCIJA'. - Baseline & promo qty use RETAIL + WEBSHOP only (RCM+WEB), no wholesale. - Detects OOS weeks: if any week mid-campaign has retail+web qty == 0 while neighbours sold > 0, flag as suspected stockout. """ if _erp is None or len(_erp) == 0 or _sales is None or len(_sales) == 0: return [] erp_s = _erp[_erp["sku"] == sku].copy() if len(erp_s) == 0: return [] iso = datetime.now().isocalendar() cur_y, cur_w = int(iso[0]), int(iso[1]) erp_s = erp_s[(erp_s["year"].astype(int) * 100 + erp_s["week"].astype(int)) < (cur_y * 100 + cur_w)] if len(erp_s) == 0: return [] erp_s["primary"] = erp_s["promo_types"].astype(str).str.split(";").str[0].str.strip() out = [] for (ptype, yr), g in erp_s.groupby(["primary", "year"]): weeks = sorted(g["week"].astype(int).unique().tolist()) 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 # Monthly AKCIJA filter if not _is_monthly_akcija(ptype, duration): continue 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)} post_yws = {yr * 100 + w for w in range(ew + 1, ew + 5)} yw_key = _sales["year"].astype(int) * 100 + _sales["week"].astype(int) sub_d = _sales[(_sales["sku"] == sku) & yw_key.isin(promo_yws)] sub_b = _sales[(_sales["sku"] == sku) & yw_key.isin(base_yws)] sub_p = _sales[(_sales["sku"] == sku) & yw_key.isin(post_yws)] if len(sub_d) == 0: continue # Retail + webshop only (NO wholesale) def _ret_sum(df): r = df["qty_retail"].sum() if "qty_retail" in df.columns else 0 w = df["qty_webshop"].sum() if "qty_webshop" in df.columns else 0 return float(r) + float(w) d_total = _ret_sum(sub_d) b_total = _ret_sum(sub_b) p_total = _ret_sum(sub_p) avg_d = d_total / max(1, len(sub_d)) avg_b = b_total / max(1, len(sub_b)) if len(sub_b) else 0 avg_p = p_total / max(1, len(sub_p)) if len(sub_p) else 0 # OOS detection — two heuristics: # (1) zero-sale weeks mid-campaign # (2) sharp-drop weeks: qty < 50 % of the campaign's max # AND < 2 × baseline (baseline×2 is a soft uplift floor; if # a promo week barely lifts at all while peers do 5-7×, it's # almost certainly stockout-constrained). d_sorted = sub_d.sort_values(["year", "week"]).copy() weekly_q = [] for _, rrow in d_sorted.iterrows(): rq = float(rrow.get("qty_retail", 0)) + float(rrow.get("qty_webshop", 0)) weekly_q.append(rq) # Detect OOS using: # (a) zero-sale week, or # (b) sharp drop: qty < 40 % of the campaign's median non-zero # week. A monthly AKCIJA on a healthy SKU should not have a # single week 60 %+ below the rest unless stock ran out. non_zero = [q for q in weekly_q if q > 0] median_q = sorted(non_zero)[len(non_zero) // 2] if non_zero else 0 per_week_oos = [] for q in weekly_q: if q == 0: per_week_oos.append(True) elif median_q > 0 and q < 0.4 * median_q: per_week_oos.append(True) else: per_week_oos.append(False) oos_weeks = sum(per_week_oos) oos_suspected = oos_weeks > 0 and not all(per_week_oos) # Recompute uplift only on non-OOS weeks clean_qs = [q for q, oos in zip(weekly_q, per_week_oos) if not oos] if oos_suspected and clean_qs: avg_d_clean = sum(clean_qs) / len(clean_qs) else: avg_d_clean = avg_d uplift = (avg_d_clean / avg_b) if avg_b > 0 else 0 post_dip = ((avg_p / avg_b) - 1) * 100 if avg_b > 0 else 0 # Average retail discount % during the promo (from sales_clean). avg_disc = 0.0 if "retail_discount_pct" in sub_d.columns: disc_vals = pd.to_numeric(sub_d["retail_discount_pct"], errors="coerce").fillna(0) disc_pos = disc_vals[disc_vals > 0] if len(disc_pos) > 0: avg_disc = float(disc_pos.mean()) # Try to detect mechanic from transactional data (if available). tx_detect = {} try: _tx = load_transactions() if not _tx.empty: tx_detect = detect_mechanic_from_transactions( _tx, sku, yr * 100 + sw, yr * 100 + ew) or {} except Exception: tx_detect = {} out.append({ "type": ptype, "year": int(yr), "start_week": sw, "end_week": ew, "duration": duration, "avg_base": round(avg_b, 1), "avg_promo": round(avg_d, 1), "avg_promo_clean": round(avg_d_clean, 1), "uplift": round(uplift, 2), "post_dip_pct": round(post_dip, 1), "oos_weeks": oos_weeks, "oos_suspected": bool(oos_suspected), "per_week_qty": [int(q) for q in weekly_q], "per_week_oos": per_week_oos, "discount_pct": round(avg_disc, 1), # avg of weekly retail discount % "mechanic_detected": tx_detect.get("mechanic", ""), "mechanic_confidence": tx_detect.get("confidence_pct", 0), "mechanic_n_docs": tx_detect.get("n_docs", 0), }) out.sort(key=lambda x: (-x["year"], -x["end_week"])) return out @st.cache_data(ttl=3600) def suggest_uplift(_sales: pd.DataFrame, _erp: pd.DataFrame, sku: str, discount_pct: float, outcome: str = "") -> dict: """Suggest expected uplift at given discount. Priority: 1. SKU's own past AKCIJA history (≥1 events) — weighted by historical discount: scale base uplift by the ratio between the new and historical discount level. Higher historical uplift → higher predicted uplift, no artificial cap. 2. Category siblings' AKCIJA history if SKU has none. 3. Flat fallback 1.35× scaled to discount depth. Also surfaces the historical discount range so CMs can sanity-check. """ own = past_promos_for_sku(_sales, _erp, sku) own_valid = [x for x in own if x["uplift"] > 0] # ---- Pull SKU's master data once for benchmarking ---- plan = _read_csv("sku_plan_list.csv") prices = _read_csv("sku_prices.csv") sku_cat = "" sku_name = "" sku_price = 0.0 if not plan.empty and "sku" in plan.columns: m = plan[plan["sku"].astype(str) == sku] if not m.empty: sku_cat = str(m.iloc[0].get("cat", "") or "") sku_name = str(m.iloc[0].get("name", "") or "") if not prices.empty and "sku" in prices.columns: m = prices[prices["sku"].astype(str) == sku] if not m.empty: price_col = next((c for c in ["normal_retail_ppp", "avg_sell_price", "price", "price_eur", "retail_price", "mp_price", "cijena"] if c in prices.columns), None) if price_col: try: sku_price = float(m.iloc[0][price_col] or 0) except (ValueError, TypeError): sku_price = 0.0 # Compute price-disruptor multiplier (€/100unit vs category benchmark) size_val, size_unit = parse_unit_size(sku_name) promo_price = sku_price * (1 - max(0.0, discount_pct) / 100.0) disruptor_mult = 1.0 disruptor_label = "" promo_per100 = 0.0 bench = {} if size_val > 0 and size_unit and sku_price > 0: promo_per100 = promo_price * 100.0 / size_val prices_map = {} if not prices.empty and "sku" in prices.columns: pc = next((c for c in ["normal_retail_ppp", "avg_sell_price", "price", "price_eur", "retail_price", "mp_price", "cijena"] if c in prices.columns), None) if pc: prices_map = dict(zip(prices["sku"].astype(str), pd.to_numeric(prices[pc], errors="coerce").fillna(0))) bench = category_price_benchmark(plan, prices_map, sku_cat, size_unit) if bench: disruptor_mult, disruptor_label = price_disruptor_multiplier( promo_per100, bench) band_mult = discount_band_uplift_curve(discount_pct) def _from_events(events: list, source: str) -> dict: # `events` is already sorted most-recent-first by past_promos_for_sku. upl = [e["uplift"] for e in events] disc_vals = [e.get("discount_pct", 0) for e in events if e.get("discount_pct", 0) > 0] avg_disc_hist = (sum(disc_vals) / len(disc_vals)) if disc_vals else 20.0 disc_min = min(disc_vals) if disc_vals else None disc_max = max(disc_vals) if disc_vals else None upl_sorted = sorted(upl) median_upl = upl_sorted[len(upl_sorted) // 2] max_upl = max(upl) avg_upl = sum(upl) / len(upl) # ---- Recency bias (60% recent / 40% median) ---- recent = events[0] recent_upl = recent["uplift"] recent_disc = (recent.get("discount_pct") if recent.get("discount_pct", 0) > 0 else avg_disc_hist) base_predict = 0.6 * recent_upl + 0.4 * median_upl # Discount scaling — anchor to most recent event's discount. delta_pp = discount_pct - recent_disc scale = max(0.3, 1.0 + delta_pp * 0.03) predicted = base_predict * scale # Apply non-linear discount-band shaping (multiplicative on top of # the linear scale, gives the proper kink at 30–35 %). # Normalise: band_mult at recent_disc would be ~1.0 already, so we # only apply the *delta* shift from recent_disc to discount_pct. recent_band = discount_band_uplift_curve(recent_disc) if recent_band > 0: predicted *= max(0.5, band_mult / recent_band) # Apply price-disruptor multiplier (the big lever for our case). predicted *= disruptor_mult # Floor only — NO upper cap. CMs reported real uplifts that # exceed any cap based on historical maxima (e.g. 17× on 2 kg # whey at €40 on first promo, no history). Capping hides those. predicted = max(1.0, predicted) ceiling = 999.0 # sentinel: "uncapped" # Surface realistic upside (p90 outcome) — empirical ratio from # 3,107 historical events. upside_ratio = upside_ratio_for_band(discount_pct) upside = round(predicted * upside_ratio, 2) return { "uplift": round(predicted, 2), "uplift_p90": upside, "upside_ratio": round(upside_ratio, 2), "source": source, "n_history": len(events), "median_uplift": round(median_upl, 2), "avg_uplift": round(avg_upl, 2), "max_uplift": round(max_upl, 2), "recent_uplift": round(recent_upl, 2), "recent_year": recent["year"], "recent_start": recent["start_week"], "recent_end": recent["end_week"], "recent_discount": round(recent_disc, 1) if recent_disc else None, "avg_disc_hist": round(avg_disc_hist, 1), "disc_min": round(disc_min, 1) if disc_min is not None else None, "disc_max": round(disc_max, 1) if disc_max is not None else None, "price_disruptor_mult": disruptor_mult, "price_zone": disruptor_label, "promo_per100": round(promo_per100, 3) if promo_per100 else None, "cat_p25_per100": round(bench["p25"], 3) if bench else None, "cat_median_per100": round(bench["median"], 3) if bench else None, "size_unit": size_unit or None, "band_mult": round(band_mult, 3), "ceiling_applied": round(ceiling, 2), } if own_valid: return _from_events(own_valid, f"SKU history ({len(own_valid)} AKCIJA campaigns)") # ---- Category fallback ---- sib_cat = sku_cat if sib_cat and not plan.empty: sib_skus = plan[plan["cat"] == sib_cat]["sku"].astype(str).tolist() sib_events = [] for s in sib_skus[:50]: if s == sku: continue sib_events.extend(past_promos_for_sku(_sales, _erp, s)) sib_valid = [e for e in sib_events if e["uplift"] > 0] if sib_valid: return _from_events( sib_valid, f"Category fallback — {sib_cat} ({len(sib_valid)} sibling AKCIJA)", ) # ---- Flat fallback with band curve + disruptor multiplier baked in ---- # No SKU history, no category history. The 1.35 prior gets shaped by # the discount band and the price-disruptor multiplier so a first-time # aggressive promo at a competitive price point still produces a # plausible volume forecast. # Flat fallback: calibrated band curve × disruptor — no upper cap. predicted_flat = max(1.0, band_mult * disruptor_mult) fallback_cap = 999.0 # sentinel: "uncapped" upside_ratio = upside_ratio_for_band(discount_pct) upside_flat = round(predicted_flat * upside_ratio, 2) return { "uplift": round(predicted_flat, 2), "uplift_p90": upside_flat, "upside_ratio": round(upside_ratio, 2), "source": ("first-time prior (calibrated band curve × " f"price-zone {disruptor_label or 'unscored'})"), "n_history": 0, "median_uplift": round(band_mult, 2), "avg_uplift": round(band_mult, 2), "max_uplift": round(band_mult, 2), "avg_disc_hist": 20.0, "disc_min": None, "disc_max": None, "price_disruptor_mult": disruptor_mult, "price_zone": disruptor_label, "promo_per100": round(promo_per100, 3) if promo_per100 else None, "cat_p25_per100": round(bench["p25"], 3) if bench else None, "cat_median_per100": round(bench["median"], 3) if bench else None, "size_unit": size_unit or None, "band_mult": round(band_mult, 3), "ceiling_applied": round(fallback_cap, 2), } # Flavour / suffix keywords (HR + EN). When these appear at the END of a # product name, they're variant info — stripping them collapses the SKU # to its product-family key. _FLAVOUR_TOKENS = { # English "vanilla", "chocolate", "strawberry", "banana", "cookies", "cookie", "caramel", "cinnamon", "hazelnut", "coconut", "mint", "mocha", "peanut", "raspberry", "toffee", "unflavoured", "unflavored", "neutral", "mango", "apple", "lemon", "cherry", "grape", "watermelon", "kiwi", "pineapple", "orange", "tropical", "citrus", "cream", "yogurt", "yoghurt", "tiramisu", "cheesecake", "bubblegum", "cola", "pear", "peach", "berry", "berries", "birthday", "cake", "salted", "crunchy", "choc", "gourmet", "wild", "mixed", "double", "creamy", "smooth", "chunky", "natural", "plain", # HR "vanilija", "čokolada", "cokolada", "jagoda", "limun", "naranča", "naranca", "borovnica", "malina", "banana", "kava", "kokos", "menta", "kruška", "kruska", "trešnja", "tresnja", "lubenica", "ananas", "breskva", "kiwi", # Common size suffix tags often appear after flavour ("LG", "MD", "SM") "lg", "md", "sm", "xl", "xxl", "xs", # Generic "pack", "bundle", "set", "edition", "limited", } def _family_key(name: str) -> str: """Reduce a product name to its 'family' identifier — i.e. strip trailing flavour / size-variant tokens. Examples: 'Polleo Premium HydroX Whey 454g Vanilla Gourmet' → 'Polleo Premium HydroX Whey 454g' 'Shirt Fitness Gym Kit SS Carbon Heather LG' → 'Shirt Fitness Gym Kit SS Carbon Heather' (LG stripped) """ if not isinstance(name, str) or not name.strip(): return "" # Tokenise — keep punctuation in tokens, but lower-case for matching. parts = name.strip().split() # Walk from the END, dropping tokens that match any flavour keyword. while parts: tail = parts[-1].strip(",.;:()[]{}").lower() # Strip tokens that are pure flavour words OR look like a flavour # variant (capitalised single word at the end with no digit). if tail in _FLAVOUR_TOKENS: parts.pop() continue # Don't strip size markers (1kg, 454g, 250ml, 60caps, etc.) # Stop if the trailing token has a digit (size/quantity). if any(ch.isdigit() for ch in tail): break # Stop on conjunctions / typical non-flavour words break return " ".join(parts).strip() @st.cache_data(ttl=300) def build_family_map(_plan: pd.DataFrame) -> dict: """Return {family_key: [sku, sku, ...]} grouping SKUs that share a product-family name (only families with ≥ 2 variants).""" if _plan is None or _plan.empty or "sku" not in _plan.columns: return {} by_family = {} for _, r in _plan.iterrows(): sku = str(r["sku"]) name = r.get("name", "") or "" fk = _family_key(name) if not fk: continue by_family.setdefault(fk, []).append(sku) # Keep only families with > 1 SKU return {k: sorted(v) for k, v in by_family.items() if len(v) > 1} def cw_to_label(year: int, week: int) -> str: return f"CW{week} ({year})" def label_to_cw(label: str) -> tuple[int, int]: """'CW20 (2026)' -> (2026, 20)""" parts = label.replace("CW", "").replace("(", "").replace(")", "").strip().split() return int(parts[1]), int(parts[0]) def horizon_weeks(n: int = 26) -> list[tuple[int, int]]: """Return (year, week) for current week + n future weeks.""" iso = datetime.now().isocalendar() out = [(int(iso[0]), int(iso[1]))] from datetime import timedelta d = datetime.strptime(f"{iso[0]}-W{iso[1]:02d}-1", "%G-W%V-%u") for _ in range(n): d = d + timedelta(days=7) i = d.isocalendar() out.append((int(i[0]), int(i[1]))) return out def _read_promotions() -> pd.DataFrame: if CM_PROMO_FILE.exists(): return pd.read_csv(CM_PROMO_FILE) return pd.DataFrame(columns=[ "id", "name", "source", "promo_type", "outcome", "target_units", "start_year", "start_week", "end_year", "end_week", "discount_pct", "channels", "sku", "sku_name", "weekly_qty_csv", "total_units", "revenue_eur", "ruc_eur", "created_at", ]) def append_promotion(rows: list[dict]) -> int: """Append a list of promo SKU-rows to cm_promotions.csv.""" df_new = pd.DataFrame(rows) df = _read_promotions() df = pd.concat([df, df_new], ignore_index=True) df.to_csv(CM_PROMO_FILE, index=False) return len(df_new) def detect_conflicts(start_y: int, start_w: int, end_y: int, end_w: int, skus: list[str]) -> list[dict]: """Return list of overlapping past promos for the given SKUs + pending entries in cm_promotions.csv.""" conflicts = [] df = _read_promotions() if df.empty: return conflicts s_key = start_y * 100 + start_w e_key = end_y * 100 + end_w for _, r in df.iterrows(): try: rs = int(r["start_year"]) * 100 + int(r["start_week"]) re = int(r["end_year"]) * 100 + int(r["end_week"]) except (TypeError, ValueError): continue if r.get("sku", "") not in skus: continue # Overlap? if rs <= e_key and re >= s_key: conflicts.append({ "sku": r["sku"], "name": r.get("sku_name", ""), "campaign": r.get("name", ""), "period": f"CW{r['start_week']}-CW{r['end_week']} ({r['start_year']})", }) return conflicts