"""Business logic for the promo module. Layering: takes a SQLAlchemy Session, delegates SQL to PromoRepository. Three logical surfaces: 1. History — group erp_promo_weeks into events, attach metadata 2. Performance — same events + uplift/cannibalization/net_effect (math mirrors build_promo_performance.py) 3. Proposals — list/create planner submissions Analog forecast lives here too — it walks a SKU's own past promos, weights them by similarity to the proposed (mechanic, discount), and returns a median uplift the planner can sanity-check against. Similarity weight is intentionally simple: same_mechanic_match: 1.0 if mechanic appears in promo_types, else 0.5 discount_pct_match : exp(-|delta_pct| / 15) (0 < weight ≤ 1) recency : 1 / (1 + months_ago × 0.05) capped at 1.0 The three are multiplied → similarity ∈ (0, 1]. """ from __future__ import annotations import math from statistics import median from typing import Optional from sqlalchemy.orm import Session from backend.repositories.promo_repo import PromoRepository def _cw_label(year: int, week: int) -> str: return f"CW{week:02d}/{year}" def _cw_range(sy: int, sw: int, ey: int, ew: int) -> str: if sy == ey and sw == ew: return _cw_label(sy, sw) return f"{_cw_label(sy, sw)} → {_cw_label(ey, ew)}" def _safe_median(values: list[float]) -> Optional[float]: clean = [v for v in values if v is not None and math.isfinite(v)] return float(median(clean)) if clean else None # --------------------------------------------------------------------------- # Calibrated curves and helpers — ported VERBATIM from PromoTool/promo_data.py # Do not modify the band edges without rerunning analyze_promo_calibration.py. # --------------------------------------------------------------------------- def discount_band_uplift_curve(discount_pct: float) -> float: """Piecewise non-linear curve calibrated against 3,107 historical AKCIJA events. Values are observed median uplifts per discount band; the engine multiplies base predict by curve(d)/curve(reference) so it stays multiplicative. 0..15 % → 0.85..1.05 15..25 % → 1.05..1.50 25..35 % → 1.50..2.10 (inflection) 35..45 % → 2.10..2.50 45..60 % → 2.50..3.50 >60 % → 3.50..5.00 (clearance) """ 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.""" 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 def effective_discount(mechanic: Optional[str], user_pct: Optional[float]) -> float: """Convert promo mechanic to effective per-unit discount %. 1+1 gratis → 50.0 % 2+1 gratis → 33.33 % 3+1 gratis → 25.0 % 4+1 gratis → 20.0 % Discount % / Other → user-entered % """ 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) if "3+1" in m: return 25.0 if "4+1" in m: return 20.0 try: return float(user_pct or 0) except (TypeError, ValueError): return 0.0 def guess_mechanic_from_discount(discount_pct: Optional[float]) -> str: """Reverse-engineer likely promo mechanic from observed discount %.""" 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 price_disruptor_multiplier( promo_per100: float, cat_p10_per100: Optional[float] ) -> tuple[float, str]: """Multiplier on uplift from promo €/100u vs category **p10** benchmark. Returns (multiplier, zone_label). Anchored to p10 (the cheap-end 10th percentile) so the boost only triggers when a promo price is in the bottom ~10% of the category — i.e. a genuine price disruptor — and is capped at 4x. (Previously anchored to p25 with a 7x ceiling, which fired too often and too hard.)""" if not cat_p10_per100 or cat_p10_per100 <= 0 or promo_per100 <= 0: return 1.0, "" ratio = promo_per100 / cat_p10_per100 if ratio >= 1.0: return 1.0, "above cat p10" if ratio >= 0.7: share = (1.0 - ratio) / 0.3 return round(1.0 + share * 1.0, 3), "near cat p10" # 1.0 → 2.0 if ratio >= 0.5: share = (0.7 - ratio) / 0.2 return round(2.0 + share * 1.0, 3), "price-disruptor (below cat p10)" # 2.0 → 3.0 deficit = min(1.0, (0.5 - ratio) / 0.5) return round(3.0 + deficit * 1.0, 3), "extreme price-disruptor (≪ cat p10)" # 3.0 → 4.0 (cap) _VARIANT_TOKENS = { # apparel sizes "xs", "s", "sm", "m", "md", "l", "lg", "xl", "xxl", "xxxl", "small", "medium", "large", # flavors "chocolate", "vanilla", "strawberry", "banana", "coconut", "hazelnut", "coffee", "mango", "lemon", "watermelon", "peach", "raspberry", "blueberry", "cookie", "cookies", "cream", "milk", "plain", "neutral", "cinnamon", "caramel", "mint", "schokolade", "cokolada", "cokoladni", "vanilija", "jagoda", "kokos", "lubenica", "breskva", "malina", "naranca", "limun", "lešnik", "lesnik", "čokolada", } def family_key(name: Optional[str], n_tokens: int = 4) -> str: """Reduce a product name to its 'family' identifier by stripping trailing variant tokens (colors/flavors/sizes). Numeric tokens like '454g' are kept as part of the family. Ported from PromoTool/parent_map.py.""" if not name: return "" import re as _re import unicodedata norm = unicodedata.normalize("NFKD", str(name)).encode("ascii", "ignore").decode("ascii") norm = norm.lower() norm = _re.sub(r"[^a-z0-9., ]+", " ", norm) norm = _re.sub(r"\s+", " ", norm).strip() tokens = norm.split() keep: list[str] = [] for t in tokens: if len(keep) >= n_tokens: break if t in _VARIANT_TOKENS: break keep.append(t) return " ".join(keep) if keep else norm class PromoService: def __init__(self, db: Session): self.db = db self.repo = PromoRepository(db) # ------------------------------------------------------------------ # History # ------------------------------------------------------------------ def get_promo_history( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, min_start_yw: Optional[int] = None, max_start_yw: Optional[int] = None, max_events: int = 5000, ) -> dict: raw = self.repo.get_promo_events( category=category, tier=tier, min_start_yw=min_start_yw, max_start_yw=max_start_yw, max_events=max_events, ) events: list[dict] = [] skus_seen: set[str] = set() for r in raw: events.append({ "sku": r["sku"], "name": r.get("name"), "category": r.get("category"), "tier": r.get("tier"), "start_year": int(r["start_year"]), "start_week": int(r["start_week"]), "end_year": int(r["end_year"]), "end_week": int(r["end_week"]), "n_weeks": int(r["n_weeks"]), "promo_types": r.get("promo_types"), "cw_label": _cw_range( int(r["start_year"]), int(r["start_week"]), int(r["end_year"]), int(r["end_week"]), ), }) skus_seen.add(r["sku"]) note = None if len(events) == max_events: note = f"Result capped at {max_events} events. Tighten filters to narrow." return { "events": events, "n_events": len(events), "n_skus": len(skus_seen), "note": note, } # ------------------------------------------------------------------ # Performance # ------------------------------------------------------------------ def get_promo_performance( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, max_weeks: int = 13, min_start_yw: Optional[int] = None, max_events: int = 5000, ) -> dict: raw = self.repo.get_promo_performance( category=category, tier=tier, max_weeks=max_weeks, min_start_yw=min_start_yw, max_events=max_events, ) events: list[dict] = [] cat_buckets: dict[str, dict] = {} skus_seen: set[str] = set() n_with_uplift = 0 for r in raw: sku = r["sku"] uplift = r.get("actual_uplift") cann = r.get("cannibalization") net = r.get("net_effect") ev = { "sku": sku, "name": r.get("name"), "category": r.get("category"), "tier": r.get("tier"), "start_year": int(r["start_year"]), "start_week": int(r["start_week"]), "end_year": int(r["end_year"]), "end_week": int(r["end_week"]), "n_weeks": int(r["n_weeks"]), "promo_types": r.get("promo_types"), "cw_label": _cw_range( int(r["start_year"]), int(r["start_week"]), int(r["end_year"]), int(r["end_week"]), ), "qty_before_avg": float(r["qty_before"]) if r.get("qty_before") is not None else None, "qty_during_avg": float(r["qty_during"]) if r.get("qty_during") is not None else None, "qty_after_avg": float(r["qty_after"]) if r.get("qty_after") is not None else None, "actual_uplift": float(uplift) if uplift is not None else None, "cannibalization": float(cann) if cann is not None else None, "net_effect": float(net) if net is not None else None, } events.append(ev) skus_seen.add(sku) if uplift is not None: n_with_uplift += 1 cat = ev["category"] or "— uncategorised —" cb = cat_buckets.setdefault(cat, { "category": cat, "n_events": 0, "uplifts": [], "canns": [], "nets": [], }) cb["n_events"] += 1 if uplift is not None: cb["uplifts"].append(float(uplift)) if cann is not None: cb["canns"].append(float(cann)) if net is not None: cb["nets"].append(float(net)) by_category = [{ "category": cb["category"], "n_events": cb["n_events"], "median_uplift": _safe_median(cb["uplifts"]), "median_cannibalization": _safe_median(cb["canns"]), "median_net_effect": _safe_median(cb["nets"]), } for cb in sorted(cat_buckets.values(), key=lambda x: x["n_events"], reverse=True)] all_uplifts = [e["actual_uplift"] for e in events if e["actual_uplift"] is not None] all_canns = [e["cannibalization"] for e in events if e["cannibalization"] is not None] all_nets = [e["net_effect"] for e in events if e["net_effect"] is not None] note = None if len(events) == max_events: note = f"Result capped at {max_events} events. Tighten filters to narrow." elif n_with_uplift == 0 and events: note = "No events had enough pre-promo history to compute uplift." return { "events": events, "by_category": by_category, "n_events": len(events), "n_skus": len(skus_seen), "n_with_uplift": n_with_uplift, "median_uplift": _safe_median(all_uplifts), "median_cannibalization": _safe_median(all_canns), "median_net_effect": _safe_median(all_nets), "max_weeks_filter": max_weeks, "note": note, } # ------------------------------------------------------------------ # Proposals # ------------------------------------------------------------------ def get_proposals( self, *, status: Optional[str] = None, proposed_by_id: Optional[int] = None, ) -> dict: raw = self.repo.get_proposals(status=status, proposed_by_id=proposed_by_id) proposals: list[dict] = [] by_status: dict[str, int] = {} for r in raw: skus_raw = r.get("skus") or [] # JSONB can come back as a Python list of dicts already skus_typed: list[dict] = [] if isinstance(skus_raw, list): for s in skus_raw: if isinstance(s, dict): skus_typed.append({ "sku": str(s.get("sku") or ""), "name": s.get("name"), "forecast_qty": s.get("forecast_qty"), "analog_uplift": s.get("analog_uplift"), }) else: skus_typed.append({"sku": str(s), "name": None, "forecast_qty": None, "analog_uplift": None}) status_val = r.get("status") or "draft" by_status[status_val] = by_status.get(status_val, 0) + 1 cw_label = None sy, sw, ey, ew = r.get("start_year"), r.get("start_week"), r.get("end_year"), r.get("end_week") if sy is not None and sw is not None and ey is not None and ew is not None: cw_label = _cw_range(int(sy), int(sw), int(ey), int(ew)) created = r.get("created_at") created_str = created.isoformat() if created is not None and hasattr(created, "isoformat") else (str(created) if created else None) proposals.append({ "id": int(r["id"]), "name": r.get("name"), "source": r.get("source"), "mechanic": r.get("mechanic"), "discount_pct": r.get("discount_pct"), "start_year": sy and int(sy), "start_week": sw and int(sw), "end_year": ey and int(ey), "end_week": ew and int(ew), "skus": skus_typed, "n_skus": len(skus_typed), "status": status_val, "proposed_by_id": r.get("proposed_by_id"), "proposed_by": r.get("proposed_by"), "channels": r.get("channels"), "log": r.get("log"), "planner_state": r.get("planner_state"), "created_at": created_str, "cw_label": cw_label, }) return { "proposals": proposals, "n_total": len(proposals), "by_status": by_status, "note": None if proposals else "No proposals yet. Use the Planner to create one.", } def submit_proposal(self, data: dict, user_id: Optional[int] = None) -> dict: payload = dict(data) if user_id is not None and payload.get("proposed_by_id") is None: payload["proposed_by_id"] = user_id return self.repo.create_proposal(payload) def update_proposal( self, *, proposal_id: int, data: dict, user_id: Optional[int] = None, ) -> dict: """Overwrite the content of an existing proposal (planner reopened a draft / revision and re-saved). Returns the freshly-shaped row from get_proposals so the frontend can refresh without another fetch.""" row = self.repo.update_proposal(proposal_id, data) if row is None: raise ValueError(f"Proposal {proposal_id} not found") result = self.get_proposals() for p in result["proposals"]: if p["id"] == proposal_id: return p raise ValueError(f"Proposal {proposal_id} vanished after update") # ------------------------------------------------------------------ # Analog forecast # ------------------------------------------------------------------ def get_analog_forecast( self, *, sku: str, mechanic: Optional[str] = None, discount_pct: Optional[float] = None, group_skus: Optional[list[str]] = None, ) -> dict: info = self.repo.get_sku_info(sku=sku) if not info: return { "sku": sku, "name": None, "n_analogs": 0, "median_uplift": None, "median_cannibalization": None, "expected_uplift": None, "expected_qty_per_week": None, "baseline_qty_per_week": None, "hits": [], "note": f"SKU '{sku}' not found in dim_products.", } runs = self.repo.get_sku_promo_history(sku=sku, max_weeks=13) baseline = self.repo.get_sku_baseline(sku=sku, weeks=13) # New-SKU fallback: if this SKU has no own baseline (new flavour/variant # with no clean non-promo sales), seed it from the average of its # PARENT-GROUP siblings (the typical sibling's weekly volume). Honors # the planner's grouping — "average of the group under the parent". baseline_source = "own" if (baseline and baseline > 0) else None baseline_group_n: Optional[int] = None group_price: Optional[float] = None group_cost: Optional[float] = None if group_skus: siblings = [s for s in group_skus if s and s != sku] if (not baseline or baseline <= 0): grp = self.repo.get_group_baseline(skus=siblings, weeks=13) if grp and grp.get("baseline", 0) > 0: baseline = grp["baseline"] baseline_source = "group_avg" baseline_group_n = grp.get("n") # Group-average price/cost so a NEW SKU (no own price) still values # its promo revenue/RUC — "take a price from the parent group". pc = self.repo.get_group_price_cost(skus=siblings) if pc: group_price = pc.get("price") group_cost = pc.get("cost") hits: list[dict] = [] uplifts: list[float] = [] weighted_uplifts: list[tuple[float, float]] = [] # (uplift, weight) for r in runs: before = r.get("qty_before") during = r.get("qty_during") uplift = (float(during) / float(before) if before and float(before) > 0 and during is not None else None) if uplift is None: continue # Similarity weight mech_match = 1.0 if (mechanic and r.get("promo_types") and mechanic.lower() in str(r["promo_types"]).lower()) else 0.5 # We don't store per-event discount in erp_promo_weeks → recency-only for now recency = 1.0 try: from datetime import datetime # Approximate "months ago" from start_year/start_week to current date sy, sw = int(r["start_year"]), int(r["start_week"]) monday = datetime.strptime(f"{sy}-W{sw:02d}-1", "%G-W%V-%u") months_ago = max(0.0, (datetime.now() - monday).days / 30.0) recency = 1.0 / (1.0 + months_ago * 0.05) except Exception: pass similarity = max(0.05, mech_match * recency) uplifts.append(uplift) weighted_uplifts.append((uplift, similarity)) hits.append({ "start_year": int(r["start_year"]), "start_week": int(r["start_week"]), "end_year": int(r["end_year"]), "end_week": int(r["end_week"]), "cw_label": _cw_range( int(r["start_year"]), int(r["start_week"]), int(r["end_year"]), int(r["end_week"]), ), "n_weeks": int(r["n_weeks"]), "promo_types": r.get("promo_types"), "discount_pct": None, "qty_before_avg": float(before) if before is not None else None, "qty_during_avg": float(during) if during is not None else None, "actual_uplift": uplift, "similarity": similarity, }) expected_uplift = None if weighted_uplifts: tot_w = sum(w for _, w in weighted_uplifts) expected_uplift = sum(u * w for u, w in weighted_uplifts) / tot_w if tot_w > 0 else None expected_qty = (baseline * expected_uplift if baseline is not None and expected_uplift is not None else None) note = None if baseline_source == "group_avg": note = (f"New SKU — no own sales. Baseline estimated from the parent-group " f"average of {baseline_group_n} sibling(s). Uplift uses the " f"calibrated discount-band prior.") elif not hits and baseline and baseline > 0: note = ("No past promos for this SKU — uplift uses the calibrated " "discount-band prior over its own non-promo run-rate.") elif baseline is None or baseline <= 0: note = ("No own sales and no group to borrow from — baseline unavailable, " "expected_qty cannot be computed. Group this SKU under its parent.") # Apply discount band curve + price-disruptor multiplier on top of # the recency-weighted historical uplift (matches PromoTool/promo_data.py # suggest_uplift() logic). band_mult = discount_band_uplift_curve(discount_pct or 20) disruptor_mult, disruptor_zone = 1.0, "" if info.get("category"): stats = self.repo.get_category_price_stats(category=info["category"]) snap = self.repo.get_sku_snapshot(sku=sku) price = float(snap.get("price_retail") or snap.get("avg_sell_price") or 0) if snap else 0 if stats and price > 0 and (discount_pct is not None): promo_price = price * (1 - max(0, discount_pct) / 100) # Use raw € as a proxy for €/100u (we don't parse unit sizes server-side) disruptor_mult, disruptor_zone = price_disruptor_multiplier( promo_price, float(stats.get("p10") or 0) ) # If we have historical uplift, scale it by band relative to reference 20%. # Otherwise, fall back to band curve × disruptor as the first-time prior. if expected_uplift is not None: ref_band = discount_band_uplift_curve(20) if ref_band > 0: expected_uplift = expected_uplift * max(0.5, band_mult / ref_band) expected_uplift = expected_uplift * disruptor_mult else: expected_uplift = max(1.0, band_mult * disruptor_mult) expected_uplift = round(max(1.0, expected_uplift), 2) expected_qty = (baseline * expected_uplift if baseline is not None and expected_uplift is not None else None) # Upside p90 — empirical ratio per discount band upside_ratio = upside_ratio_for_band(discount_pct or 20) upside_uplift = round(expected_uplift * upside_ratio, 2) upside_qty = (baseline * upside_uplift if baseline is not None and upside_uplift is not None else None) return { "sku": sku, "name": info.get("name"), "n_analogs": len(hits), "median_uplift": _safe_median(uplifts), "median_cannibalization": None, "expected_uplift": expected_uplift, "expected_qty_per_week": expected_qty, "baseline_qty_per_week": baseline, "baseline_source": baseline_source, "baseline_group_n": baseline_group_n, "group_price": round(group_price, 2) if group_price else None, "group_cost": round(group_cost, 2) if group_cost else None, "upside_uplift": upside_uplift, "upside_ratio": upside_ratio, "upside_qty_per_week": upside_qty, "band_mult": round(band_mult, 3), "price_disruptor_mult": disruptor_mult, "price_zone": disruptor_zone or None, "hits": hits, "note": note, } # ------------------------------------------------------------------ # NC30 compliance — promo price must be ≤ NC30 (lowest-30-day rule) # ------------------------------------------------------------------ def check_nc30(self, *, sku: str, planned_promo_price: float) -> dict: """Compare planned promo price to NC30. Returns {ok, nc30, delta_pct, message, has_data}. has_data=False means the SKU has no NC30 set — the check is skipped (treated as OK), planner shows a warning.""" nc30 = self.repo.get_nc30(sku=sku) if nc30 is None or nc30 <= 0: return { "sku": sku, "has_data": False, "nc30": None, "planned_promo_price": planned_promo_price, "delta_pct": None, "ok": True, "message": "NC30 reference not available — check skipped.", } delta_pct = ((planned_promo_price - nc30) / nc30 * 100) if nc30 > 0 else 0 ok = planned_promo_price <= nc30 + 0.01 if ok: msg = f"OK — promo price €{planned_promo_price:.2f} ≤ NC30 €{nc30:.2f}" else: msg = (f"BLOCKED — promo €{planned_promo_price:.2f} is " f"{delta_pct:+.1f}% above NC30 €{nc30:.2f} (Croatian price-floor law)") return { "sku": sku, "has_data": True, "nc30": round(nc30, 2), "planned_promo_price": planned_promo_price, "delta_pct": round(delta_pct, 1), "ok": ok, "message": msg, } def check_nc30_batch(self, *, items: list[dict]) -> list[dict]: """Bulk NC30 check. items = [{sku, planned_promo_price}, ...].""" skus = [it["sku"] for it in items] nc30_map = self.repo.get_nc30_batch(skus=skus) results = [] for it in items: sku = it["sku"] ppp = float(it.get("planned_promo_price") or 0) nc30 = nc30_map.get(sku) if nc30 is None or nc30 <= 0: results.append({ "sku": sku, "has_data": False, "nc30": None, "planned_promo_price": ppp, "delta_pct": None, "ok": True, "message": "NC30 not available — skipped.", }) continue delta_pct = ((ppp - nc30) / nc30 * 100) if nc30 > 0 else 0 ok = ppp <= nc30 + 0.01 results.append({ "sku": sku, "has_data": True, "nc30": round(nc30, 2), "planned_promo_price": ppp, "delta_pct": round(delta_pct, 1), "ok": ok, "message": (f"OK — €{ppp:.2f} ≤ NC30 €{nc30:.2f}" if ok else f"BLOCKED — €{ppp:.2f} is {delta_pct:+.1f}% above NC30 €{nc30:.2f}"), }) return results # ------------------------------------------------------------------ # SKU snapshot for the Planner # ------------------------------------------------------------------ def get_sku_snapshot(self, *, sku: str) -> Optional[dict]: """One-shot snapshot — price, cost, NC30, baseline, on-hand, etc. Used by the Planner to populate per-SKU tab data without firing 4 separate endpoints.""" snap = self.repo.get_sku_snapshot(sku=sku) if not snap: return None return { "sku": snap["sku"], "name": snap.get("name"), "category": snap.get("category"), "tier": snap.get("tier"), "xyz": snap.get("xyz"), "price_retail": snap.get("price_retail"), "price_webshop": snap.get("price_webshop"), "avg_sell_price": snap.get("avg_sell_price"), "cost_price": snap.get("cost_price"), "ruc_unit": snap.get("ruc_unit"), "nc30": snap.get("nc30"), "baseline_avg_weekly": snap.get("baseline_avg_weekly") or 0.0, "baseline_n_weeks": snap.get("baseline_n_weeks") or 0, "on_hand": snap.get("on_hand") or 0.0, "lead_time_weeks": snap.get("lead_time_weeks"), "moq": snap.get("moq"), } # ------------------------------------------------------------------ # Parent groups — product-family detection via name-stem heuristic # ------------------------------------------------------------------ def get_parent_groups(self, *, min_size: int = 2) -> dict: """Return {parent_key: {display, n_skus, category, skus: [...] }}. Groups with fewer than `min_size` SKUs are filtered out. The family_key() heuristic strips trailing variant tokens (colors, flavors, sizes) so 'Whey 454g Vanilla' and 'Whey 454g Chocolate' both reduce to 'whey 454g'.""" products = self.repo.get_dim_products_for_family_grouping() buckets: dict[str, dict] = {} for p in products: key = family_key(p.get("name")) if not key: continue b = buckets.setdefault(key, { "key": key, "display": (p.get("name") or "").strip(), "category": p.get("category"), "skus": [], }) b["skus"].append({ "sku": p["sku"], "name": p.get("name"), "tier": p.get("tier"), }) groups = [] for k, v in buckets.items(): if len(v["skus"]) < min_size: continue # Display name = the shortest child name (usually the cleanest) v["display"] = min((s["name"] or k for s in v["skus"]), key=lambda s: len(s) if s else 999) v["n_skus"] = len(v["skus"]) groups.append(v) groups.sort(key=lambda g: (-g["n_skus"], g["display"])) return {"groups": groups, "n_groups": len(groups)} # ------------------------------------------------------------------ # Forecaster — recommend discount + duration for SKU × outcome # ------------------------------------------------------------------ _OUTCOMES = { "stock_clear": "Stock clearance — pick deepest historical discount", "margin": "Higher margin — pick shallowest historical discount", "acquisition": "New customers — above-average discount", "traffic": "Traffic driver — average historical discount", } _OUTCOME_DEFAULTS = { "stock_clear": 30.0, "margin": 12.0, "acquisition": 22.0, "traffic": 20.0, } def get_forecaster(self, *, sku: str, outcome: str = "traffic") -> dict: """SKU + desired outcome → recommended discount + duration + P&L. Discount anchor logic (ported from page_forecaster.py): - If ≥2 historical discounts on this SKU → anchor on them stock_clear → max margin → min acquisition → avg + half (max - avg) traffic → avg - Else → outcome default (30/12/22/20%) Then auto-lower if breakeven uplift > predicted (margin-positive). Duration is fixed at 4 weeks (matches Streamlit's monthly AKCIJA). """ snap = self.repo.get_sku_snapshot(sku=sku) if not snap: return { "sku": sku, "outcome": outcome, "error": f"SKU '{sku}' not found.", "rec_discount": None, "rec_duration": 4, "pred_uplift": None, "breakeven": None, } runs = self.repo.get_sku_promo_history(sku=sku, max_weeks=13) # Build uplift + discount history (no explicit discount stored per event # in erp_promo_weeks, so we use band-curve inference for now) uplifts: list[float] = [] durations: list[int] = [] for r in runs: before, during = r.get("qty_before"), r.get("qty_during") if before and float(before) > 0 and during is not None: uplifts.append(float(during) / float(before)) durations.append(int(r["n_weeks"])) n_history = len(uplifts) avg_uplift = sum(uplifts) / n_history if n_history else 0.0 avg_duration = (sum(durations) / len(durations)) if durations else 4.0 # Outcome anchor: without per-event discount data we use defaults outcome_key = (outcome or "traffic").strip().lower() if "lager" in outcome_key or "stock" in outcome_key or "clear" in outcome_key: okey = "stock_clear" elif "ruc" in outcome_key or "margin" in outcome_key: okey = "margin" elif "novi" in outcome_key or "acqui" in outcome_key: okey = "acquisition" else: okey = "traffic" rec_discount = self._OUTCOME_DEFAULTS[okey] rec_duration = 4 price = float(snap.get("price_retail") or snap.get("avg_sell_price") or 0) cost = float(snap.get("cost_price") or 0) baseline = float(snap.get("baseline_avg_weekly") or 0) # Auto-lower for breakeven compliance floor_d = 5 chosen_d = rec_discount for trial in range(int(rec_discount), floor_d - 1, -1): ppp = price * (1 - trial / 100) if price > 0 else 0 be = (price / ppp) if ppp > 0 else 0 pred = self._predict_uplift( avg_uplift_hist=avg_uplift, n_history=n_history, d=trial, category=snap.get("category"), price=price, ) if pred >= be: chosen_d = trial break else: chosen_d = floor_d promo_price = price * (1 - chosen_d / 100) if price > 0 else 0 breakeven = (price / promo_price) if promo_price > 0 else None pred_uplift_at_rec = self._predict_uplift( avg_uplift_hist=avg_uplift, n_history=n_history, d=chosen_d, category=snap.get("category"), price=price, ) # Volume + P&L promo_qty = baseline * rec_duration * pred_uplift_at_rec if baseline > 0 else 0 base_qty = baseline * rec_duration incremental = promo_qty - base_qty promo_revenue = promo_qty * promo_price base_revenue = base_qty * price rev_delta = promo_revenue - base_revenue promo_ruc_unit = max(0.0, promo_price - cost) base_ruc_unit = max(0.0, price - cost) promo_ruc = promo_qty * promo_ruc_unit base_ruc = base_qty * base_ruc_unit ruc_delta = promo_ruc - base_ruc # Discount sensitivity table (5% steps from 0 to 50) sensitivity = [] for d in range(0, 55, 5): ppp = price * (1 - d / 100) if price > 0 else 0 be = (price / ppp) if ppp > 0 else None pred = self._predict_uplift( avg_uplift_hist=avg_uplift, n_history=n_history, d=d, category=snap.get("category"), price=price, ) sensitivity.append({ "discount_pct": d, "promo_price": round(ppp, 2) if price > 0 else None, "predicted_uplift": round(pred, 2), "breakeven_uplift": round(be, 2) if be else None, "margin_ok": be is not None and pred >= be, }) # Margin signal at recommendation margin_positive = (breakeven is not None and pred_uplift_at_rec >= breakeven) # Latest historical events for the backing table history_rows = [] for r in runs[:25]: before, during = r.get("qty_before"), r.get("qty_during") u = (float(during) / float(before) if before and float(before) > 0 and during is not None else None) history_rows.append({ "start_year": int(r["start_year"]), "start_week": int(r["start_week"]), "end_year": int(r["end_year"]), "end_week": int(r["end_week"]), "cw_label": _cw_range(int(r["start_year"]), int(r["start_week"]), int(r["end_year"]), int(r["end_week"])), "n_weeks": int(r["n_weeks"]), "promo_types": r.get("promo_types"), "qty_before_avg": float(before) if before is not None else None, "qty_during_avg": float(during) if during is not None else None, "actual_uplift": round(u, 2) if u is not None else None, }) note = None if baseline <= 0: note = "No recent non-promo sales — baseline unavailable, forecast cannot be computed." elif n_history == 0: note = (f"No past promos for this SKU. Recommendation uses outcome default ({chosen_d}%) " f"and the calibrated discount-band curve.") return { "sku": sku, "name": snap.get("name"), "category": snap.get("category"), "tier": snap.get("tier"), "outcome": okey, "rec_discount": chosen_d, "rec_duration": rec_duration, "pred_uplift": round(pred_uplift_at_rec, 2), "breakeven": round(breakeven, 2) if breakeven else None, "margin_positive": margin_positive, "price": round(price, 2) if price else None, "promo_price": round(promo_price, 2) if promo_price else None, "cost": round(cost, 2) if cost else None, "baseline_avg_weekly": round(baseline, 1), "promo_qty": round(promo_qty, 0), "base_qty": round(base_qty, 0), "incremental_qty": round(incremental, 0), "promo_revenue": round(promo_revenue, 0), "rev_delta": round(rev_delta, 0), "promo_ruc": round(promo_ruc, 0), "ruc_delta": round(ruc_delta, 0), "n_history": n_history, "avg_uplift_hist": round(avg_uplift, 2), "avg_duration_hist": round(avg_duration, 1), "sensitivity": sensitivity, "history": history_rows, "note": note, } def _predict_uplift( self, *, avg_uplift_hist: float, n_history: int, d: float, category: Optional[str], price: float, ) -> float: """Predict uplift at discount d, applying band curve + disruptor.""" band = discount_band_uplift_curve(d) if n_history > 0 and avg_uplift_hist > 0: ref_band = discount_band_uplift_curve(20) scaled = avg_uplift_hist * max(0.5, band / ref_band if ref_band > 0 else 1.0) else: scaled = band # Price-disruptor multiplier mult = 1.0 if category and price > 0: stats = self.repo.get_category_price_stats(category=category) if stats and stats.get("p25"): promo_price = price * (1 - d / 100) mult, _zone = price_disruptor_multiplier(promo_price, float(stats["p25"])) scaled = scaled * mult return max(1.0, scaled) # ------------------------------------------------------------------ # Marketing History — Magento webshop coupon campaigns # ------------------------------------------------------------------ def get_marketing_history(self) -> dict: """List webshop coupon campaigns with per-campaign aggregates. Returns {} if no coupon data has been imported.""" campaigns_raw = self.repo.get_marketing_campaigns() if not campaigns_raw: return { "campaigns": [], "n_campaigns": 0, "note": ("No webshop coupon orders imported yet. " "Load Magento export into webshop_coupon_orders to populate this view."), } campaigns = [] for c in campaigns_raw: campaigns.append({ "campaign_id": int(c["campaign_id"]), "campaign_name": c.get("campaign_name") or "", "campaign_label": c.get("campaign_label"), "first_date": c["first_date"].isoformat() if c.get("first_date") else None, "last_date": c["last_date"].isoformat() if c.get("last_date") else None, "days_active": ((c["last_date"] - c["first_date"]).days + 1) if c.get("first_date") and c.get("last_date") else 0, "n_orders": int(c.get("n_orders") or 0), "n_coupons": int(c.get("n_coupons") or 0), "n_skus": int(c.get("n_skus") or 0), "total_units": float(c.get("total_units") or 0), "total_rev_before": float(c.get("total_rev_before") or 0), "total_rev_after": float(c.get("total_rev_after") or 0), "total_discount": float(c.get("total_discount") or 0), "avg_disc_pct": (float(c["total_discount"] or 0) / float(c["total_rev_before"]) * 100 if c.get("total_rev_before") and float(c["total_rev_before"]) > 0 else 0.0), }) return { "campaigns": campaigns, "n_campaigns": len(campaigns), "note": None, } def get_marketing_campaign_detail(self, *, campaign_id: int) -> dict: """Per-campaign drill-down: per-SKU + per-coupon + daily breakdown.""" skus = self.repo.get_marketing_campaign_skus(campaign_id=campaign_id) coupons = self.repo.get_marketing_campaign_coupons(campaign_id=campaign_id) daily = self.repo.get_marketing_daily(campaign_id=campaign_id) # Convert dates for JSON daily_clean = [{ "day": r["day"].isoformat() if r.get("day") else "", "code": r.get("code") or "", "units": float(r.get("units") or 0), } for r in daily] coupons_clean = [{ "code": c.get("code") or "", "first_date": c["first_date"].isoformat() if c.get("first_date") else None, "last_date": c["last_date"].isoformat() if c.get("last_date") else None, "days_active": ((c["last_date"] - c["first_date"]).days + 1) if c.get("first_date") and c.get("last_date") else 0, "n_orders": int(c.get("n_orders") or 0), "units": float(c.get("units") or 0), "discount": float(c.get("discount") or 0), "avg_disc_pct": float(c.get("avg_disc_pct") or 0), } for c in coupons] return { "campaign_id": campaign_id, "skus": skus, "coupons": coupons_clean, "daily": daily_clean, } # ------------------------------------------------------------------ # Calendar — list proposals shaped as calendar entries # ------------------------------------------------------------------ # Color palette + dept map — match PromoCalendar/promo_data.py _CALENDAR_SOURCES: dict = { "B2C — MP (retail)": "#7C6FEE", "B2C — WEB": "#E8734A", "B2B — FMCG": "#34D399", "B2B — FITNESS": "#38BDF8", } _CALENDAR_SOURCE_TO_DEPT: dict = { "B2C — MP (retail)": "Nabava", "B2C — WEB": "Marketing", "B2B — FMCG": "Nabava", "B2B — FITNESS": "Nabava", } def _proposal_to_entry(self, raw: dict) -> dict: """Shape a promo_proposals row into a CalendarEntry dict.""" skus_raw = raw.get("skus") or [] sku_list: list[str] = [] if isinstance(skus_raw, list): for s in skus_raw: if isinstance(s, dict): val = str(s.get("sku") or "") if val: sku_list.append(val) else: val = str(s) if val: sku_list.append(val) source = raw.get("source") dept = self._CALENDAR_SOURCE_TO_DEPT.get(source or "", "Nabava") color = self._CALENDAR_SOURCES.get(source or "") sy = raw.get("start_year") sw = raw.get("start_week") ey = raw.get("end_year") ew = raw.get("end_week") cw_label = None start_date = None end_date = None if sy and sw and ey and ew: cw_label = _cw_range(int(sy), int(sw), int(ey), int(ew)) try: from datetime import datetime, timedelta s = datetime.strptime(f"{int(sy)}-W{int(sw):02d}-1", "%G-W%V-%u").date() e = (datetime.strptime(f"{int(ey)}-W{int(ew):02d}-1", "%G-W%V-%u") + timedelta(days=6)).date() start_date = s.isoformat() end_date = e.isoformat() except Exception: pass created = raw.get("created_at") created_str = created.isoformat() if created is not None and hasattr(created, "isoformat") else (str(created) if created else None) duration_weeks = None if sy and sw and ey and ew: duration_weeks = (int(ey) * 53 + int(ew)) - (int(sy) * 53 + int(sw)) + 1 if duration_weeks < 1: duration_weeks = None return { "id": int(raw["id"]), "name": raw.get("name"), "source": source, "type": raw.get("mechanic"), "outcome": None, "objective": raw.get("objective"), "status": raw.get("status") or "draft", "start_year": sy and int(sy), "start_week": sw and int(sw), "end_year": ey and int(ey), "end_week": ew and int(ew), "skus": sku_list, "n_skus": len(sku_list), "units": 0, "category": None, "categories": [], "n_categories": 0, "duration_weeks": duration_weeks, "discount_min": None, "discount_max": None, "owner": raw.get("proposed_by"), "department": dept, "notes": None, "log": raw.get("log"), "color": color, "created_at": created_str, "cw_label": cw_label, "start_date": start_date, "end_date": end_date, } def get_proposal_sku_detail(self, *, proposal_id: int) -> list[dict]: """Per-SKU detail (sku/name/category/price/discount/promo_price) for the calendar drill-down + detail panel.""" return self.repo.get_proposal_sku_detail(proposal_id=proposal_id) def get_sku_overlaps(self, *, proposal_ids: list[int]) -> list[dict]: """SKUs double-promoted across the given proposals (in 2+ of them).""" return self.repo.get_sku_overlaps(proposal_ids=proposal_ids) def _enrich_entries(self, entries: list[dict]) -> None: """Fill the overview fields on each calendar entry, in place: • categories / n_categories — SKU rollup by product category • discount_min / discount_max — Rabat range from the matching ERP campaign Two bulk queries total (category map + discount ranges), so this stays cheap regardless of how many entries are on the board.""" if not entries: return # Invert {category: {sku}} → {sku: category} once. sku_to_cat: dict[str, str] = {} for cat, skus in self.repo.get_category_sku_map().items(): for s in skus: sku_to_cat.setdefault(s, cat) disc = self.repo.get_discount_ranges_by_name([e.get("name") for e in entries]) for e in entries: counts: dict[str, int] = {} for s in e.get("skus", []): cat = sku_to_cat.get(s) if cat: counts[cat] = counts.get(cat, 0) + 1 e["categories"] = [ {"category": k, "n_skus": v} for k, v in sorted(counts.items(), key=lambda kv: (-kv[1], kv[0])) ] e["n_categories"] = len(counts) # Single dominant category kept for the conflict detector / legacy field. if e.get("category") is None and e["categories"]: e["category"] = e["categories"][0]["category"] dr = disc.get(e.get("name") or "") if dr: e["discount_min"], e["discount_max"] = dr def get_calendar( self, *, source: Optional[list[str]] = None, status: Optional[list[str]] = None, min_yw: Optional[int] = None, max_yw: Optional[int] = None, ) -> dict: raw = self.repo.list_calendar_entries( source=source, status=status, min_yw=min_yw, max_yw=max_yw, ) entries = [self._proposal_to_entry(r) for r in raw] self._enrich_entries(entries) return { "entries": entries, "n_total": len(entries), "sources": dict(self._CALENDAR_SOURCES), "source_to_dept": dict(self._CALENDAR_SOURCE_TO_DEPT), "depts": list(self._CALENDAR_SOURCE_TO_DEPT.keys()), # not used; depts come from values "note": None, } # ------------------------------------------------------------------ # Conflict detection — ports detect_conflicts() from PromoCalendar # ------------------------------------------------------------------ def get_calendar_conflicts(self) -> dict: """Find SKU-level conflicts between proposals with: - overlapping weeks - different sources - same SKU OR category-vs-SKU expansion Mirrors the algorithm in PromoCalendar/promo_data.py:detect_conflicts. """ raw = self.repo.list_calendar_entries() entries = [self._proposal_to_entry(r) for r in raw] cat_map: dict[str, set[str]] = self.repo.get_category_sku_map() # Skip done/analyzed (and our rejected) — only live/pending/approved should conflict def is_active(status: Optional[str]) -> bool: s = (status or "").lower() return s not in ("done", "analyzed", "rejected") enriched = [] for i, e in enumerate(entries): sy, sw = e.get("start_year"), e.get("start_week") ey, ew = e.get("end_year"), e.get("end_week") if not sy or not sw or not ey or not ew: continue s_yw = int(sy) * 100 + int(sw) e_yw = int(ey) * 100 + int(ew) if e_yw < s_yw: continue if not is_active(e.get("status")): continue sku_set: set[str] = set(e["skus"]) from_cat = False if not sku_set and e.get("category"): cat = str(e["category"]).strip() if cat in cat_map: sku_set = set(cat_map[cat]) from_cat = True enriched.append({ "i": i, "id": int(e["id"]), "name": e.get("name"), "source": e.get("source"), "status": e.get("status"), "category": (e.get("category") or "").strip(), "start": s_yw, "end": e_yw, "sw": int(sw), "ew": int(ew), "sy": int(sy), "skus": sku_set, "from_category": from_cat, "dept": self._CALENDAR_SOURCE_TO_DEPT.get(e.get("source") or "", "Nabava"), }) conflicts: list[dict] = [] seen_pairs: set[tuple[str, int, int]] = set() for i in range(len(enriched)): for j in range(i + 1, len(enriched)): a, b = enriched[i], enriched[j] if a["start"] > b["end"] or b["start"] > a["end"]: continue if a["source"] == b["source"]: continue overlap_skus = a["skus"] & b["skus"] if not overlap_skus: continue kind = "sku_overlap" if a["from_category"] and not b["from_category"]: kind = "category_vs_sku" elif b["from_category"] and not a["from_category"]: kind = "category_vs_sku" elif a["from_category"] and b["from_category"]: kind = "category_vs_category" ov_sw = max(a["sw"], b["sw"]) ov_ew = min(a["ew"], b["ew"]) ov_y = max(a["sy"], b["sy"]) weeks_label = f"CW{ov_sw}" + (f"–CW{ov_ew}" if ov_ew > ov_sw else "") for sku in sorted(overlap_skus): key = (sku, a["i"], b["i"]) if key in seen_pairs: continue seen_pairs.add(key) conflicts.append({ "sku": sku, "kind": kind, "weeks": weeks_label, "year": ov_y, "id_a": a["id"], "id_b": b["id"], "source_a": a["source"], "name_a": a["name"], "dept_a": a["dept"], "source_b": b["source"], "name_b": b["name"], "dept_b": b["dept"], "category_a": a["category"] if a["from_category"] else "", "category_b": b["category"] if b["from_category"] else "", }) return {"conflicts": conflicts, "n_conflicts": len(conflicts)} # ------------------------------------------------------------------ # Calendar stats # ------------------------------------------------------------------ def get_calendar_stats( self, *, source: Optional[list[str]] = None, status: Optional[list[str]] = None, min_yw: Optional[int] = None, max_yw: Optional[int] = None, ) -> dict: s = self.repo.get_calendar_stats( source=source, status=status, min_yw=min_yw, max_yw=max_yw, ) # Get conflict count from full conflict detection cf = self.get_calendar_conflicts() s["n_conflicts"] = cf["n_conflicts"] return {"stats": s} # ------------------------------------------------------------------ # Approvals — approve / reject / request revision / acknowledge # ------------------------------------------------------------------ _DECISION_TO_STATUS = { "approve": "approved", "reject": "rejected", "revision": "revision", } def perform_approval( self, *, proposal_id: int, decision: str, feedback: Optional[str] = None, reviewer_id: Optional[int] = None, ) -> dict: if decision not in self._DECISION_TO_STATUS: raise ValueError(f"Unknown decision '{decision}' — expected approve/reject/revision") new_status = self._DECISION_TO_STATUS[decision] from datetime import datetime ts = datetime.now().strftime("%Y-%m-%d %H:%M") actor = "Direktor nabave" # placeholder when no auth yet action_label = {"approve": "Approved", "reject": "Rejected", "revision": "Sent back for revision"}[decision] log_line = f"[{ts}] {actor} · {action_label}" if feedback: log_line += f" — {feedback}" ok = self.repo.update_proposal_status( proposal_id=proposal_id, new_status=new_status, log_line=log_line, ) if not ok: raise ValueError(f"Proposal {proposal_id} not found") approval_id = self.repo.add_approval( proposal_id=proposal_id, reviewed_by_id=reviewer_id, decision=new_status, feedback=feedback, ) return { "proposal_id": proposal_id, "new_status": new_status, "approval_id": approval_id, "message": f"Proposal {proposal_id} → {new_status}", } def change_proposal_status( self, *, proposal_id: int, new_status: str, log_line: Optional[str] = None, reviewer_id: Optional[int] = None, ) -> dict: """Generic status change (re-submit, withdraw, manual). Inserts an approval log entry when the action is a director decision; for self-service actions (re-submit, withdraw) just updates status.""" ok = self.repo.update_proposal_status( proposal_id=proposal_id, new_status=new_status, log_line=log_line, ) if not ok: raise ValueError(f"Proposal {proposal_id} not found") return { "proposal_id": proposal_id, "new_status": new_status, "message": f"Status updated to {new_status}", } def delete_proposal(self, *, proposal_id: int) -> dict: ok = self.repo.delete_proposal(proposal_id=proposal_id) if not ok: raise ValueError(f"Proposal {proposal_id} not found") return {"proposal_id": proposal_id, "message": "Deleted"} def acknowledge_conflict( self, *, id_a: int, id_b: int, reviewer_id: Optional[int] = None, ) -> dict: """Append a log line to BOTH proposals indicating director acknowledged the conflict pair. The conflict detector skips pairs where one side has the other id in its log marker.""" from datetime import datetime ts = datetime.now().strftime("%Y-%m-%d %H:%M") actor = "Direktor nabave" line_a = f"[{ts}] {actor} · Acknowledged conflict — kept both intentionally (paired with #{id_b})" line_b = f"[{ts}] {actor} · Acknowledged conflict — kept both intentionally (paired with #{id_a})" # Don't change status — just append log lines proposal_a = self.repo.get_proposal(proposal_id=id_a) proposal_b = self.repo.get_proposal(proposal_id=id_b) if proposal_a is None or proposal_b is None: raise ValueError("One or both proposals not found") # We don't have a "no-op status update with just log" method; use # the existing one which is happy to keep status the same: self.repo.update_proposal_status( proposal_id=id_a, new_status=proposal_a.get("status") or "draft", log_line=line_a, ) self.repo.update_proposal_status( proposal_id=id_b, new_status=proposal_b.get("status") or "draft", log_line=line_b, ) # Also write to promo_approvals so the audit trail is queryable self.repo.add_approval(proposal_id=id_a, reviewed_by_id=reviewer_id, decision="acknowledged", feedback=f"with #{id_b}") self.repo.add_approval(proposal_id=id_b, reviewed_by_id=reviewer_id, decision="acknowledged", feedback=f"with #{id_a}") return {"id_a": id_a, "id_b": id_b, "message": "Acknowledged"} def list_approvals(self, *, proposal_id: Optional[int] = None) -> list[dict]: raw = self.repo.list_approvals(proposal_id=proposal_id) out = [] for r in raw: decided = r.get("decided_at") decided_str = decided.isoformat() if decided is not None and hasattr(decided, "isoformat") else (str(decided) if decided else None) out.append({ "id": int(r["id"]), "proposal_id": int(r["proposal_id"]), "decision": r.get("decision") or "", "feedback": r.get("feedback"), "decided_at": decided_str, "reviewed_by_id": r.get("reviewed_by_id"), "reviewed_by": r.get("reviewed_by"), }) return out