"""Business logic for the demand module. Layering rules in this codebase: - takes a SQLAlchemy Session at construction - delegates SQL to a repository - never imports FastAPI FA per-row formulas (matching app.py:4172-4181 exactly): error = |F - A| fa = max(0, 1 - |F-A|/A) # 0..1 per row, ×100 at display fa_signed = F / A # 1.0 = perfect bias = (F - A) / A # not stored per row in app.py; # we expose it for FARow detail hit = (|F-A| / max(A, 1)) <= 0.30 # NOTE: denominator is .clip(lower=1) # so 0 < A < 1 weeks aren't unfairly # penalised Filter: drop rows where A <= 0 (matches `fa = fa[fa["actual"] > 0]`). FA headline aggregation — the "per-week-average" rule (matches app.py:4430-4442 `_per_week_avg_metrics`, the Streamlit default for the Forecast Accuracy weekly view): 1. Group rows by (year, week). Sum forecast and actual per week. 2. Drop weeks where sum_actual <= 0. 3. Per week: fa_w = max(0, 1 - |sumF - sumA| / sumA) × 100 fa_signed_w = sumF / sumA × 100 bias_w = (sumF - sumA) / sumA × 100 4. Headline FA / FA-signed / BIAS = simple arithmetic mean across weeks. Streamlit's code groups by `week` only (line 4434); we group by (year, week) because it's semantically more correct and matches Streamlit when the data is single-year (our backtest_results is 2026-only at the moment). The behaviour diverges only when the same ISO week from two different years is in scope, which doesn't happen for the default last-N-weeks selector. Hit rate stays as the simple arithmetic mean of per-row `hit` flags ×100 (matches app.py:4519 `float(sub["hit"].mean() * 100)`). Revenue (separate concern): revenue per channel = qty × price (per-channel price source in demand_repo.get_revenue_summary). WoW growth = (this - prev) / prev × 100. Channel mix is each channel's share of total (separate dimensions for revenue and qty since they don't always agree). """ from __future__ import annotations from collections import defaultdict from datetime import datetime from typing import Optional from sqlalchemy import text from sqlalchemy.orm import Session from backend.repositories.demand_repo import DemandRepository # Month ordering for the monthly accuracy table — matches Streamlit's # Croatian month names from _week_to_month_name. _MONTHS = [ "Siječanj", "Veljača", "Ožujak", "Travanj", "Svibanj", "Lipanj", "Srpanj", "Kolovoz", "Rujan", "Listopad", "Studeni", "Prosinac", ] def _week_to_month_name(year: int, week: int) -> str: """Thursday-of-week → Croatian month name. Thursday is the canonical "centre of the ISO week" so weeks straddling a month boundary get assigned to the month containing the majority of days. Use the day-split helper for true pro-rata splitting where exact totals matter (see backend.services.time_utils.split_iso_week_across_months).""" try: thu = datetime.strptime(f"{year}-W{week:02d}-4", "%G-W%V-%u") return _MONTHS[thu.month - 1] except Exception: return "—" class DemandService: def __init__(self, db: Session): self.repo = DemandRepository(db) # ------------------------------------------------------------------ # Sales # ------------------------------------------------------------------ def get_sales_data( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, year: Optional[int] = None, week_from: Optional[int] = None, week_to: Optional[int] = None, sort_by: Optional[str] = None, sort_dir: str = "asc", page: int = 1, page_size: int = 50, ) -> dict: page = max(1, page) page_size = max(1, page_size) offset = (page - 1) * page_size rows, total = self.repo.get_sales_weekly( tier=tier, xyz=xyz, category=category, year=year, week_from=week_from, week_to=week_to, sort_by=sort_by, sort_dir=sort_dir, limit=page_size, offset=offset, ) for r in rows: yr = r.get("year") wk = r.get("week") r["year_week"] = (int(yr) * 100 + int(wk)) if (yr is not None and wk is not None) else None total_pages = (total + page_size - 1) // page_size if page_size else 1 return { "rows": rows, "total_count": total, "page": page, "page_size": page_size, "total_pages": total_pages, } # ------------------------------------------------------------------ # Revenue # ------------------------------------------------------------------ @staticmethod def _mix(retail: float, webshop: float, wholesale: float) -> dict: total = retail + webshop + wholesale if total <= 0: return {"retail": 0.0, "webshop": 0.0, "wholesale": 0.0} return { "retail": retail / total * 100, "webshop": webshop / total * 100, "wholesale": wholesale / total * 100, } def get_revenue_data(self, *, year: int) -> dict: raw = self.repo.get_revenue_summary(year=year) weeks: list[dict] = [] prev_total: Optional[float] = None for r in raw: rev_r = float(r.get("revenue_retail") or 0) rev_w = float(r.get("revenue_webshop") or 0) rev_s = float(r.get("revenue_wholesale") or 0) rev_total = rev_r + rev_w + rev_s wow = None if prev_total is not None and prev_total > 0: wow = (rev_total - prev_total) / prev_total * 100 weeks.append({ "year": int(r["year"]), "week": int(r["week"]), "qty_retail": float(r.get("qty_retail") or 0), "qty_webshop": float(r.get("qty_webshop") or 0), "qty_wholesale": float(r.get("qty_wholesale") or 0), "qty_total": float(r.get("qty_total") or 0), "revenue_retail": rev_r, "revenue_webshop": rev_w, "revenue_wholesale": rev_s, "revenue_total": rev_total, "wow_growth_pct": wow, }) prev_total = rev_total ytd_qty = sum(w["qty_total"] for w in weeks) ytd_revenue = sum(w["revenue_total"] for w in weeks) n = max(1, len(weeks)) return { "year": year, "weeks": weeks, "ytd_qty": ytd_qty, "ytd_revenue": ytd_revenue, "avg_weekly_qty": ytd_qty / n, "avg_weekly_revenue": ytd_revenue / n, "channel_mix": self._mix( sum(w["revenue_retail"] for w in weeks), sum(w["revenue_webshop"] for w in weeks), sum(w["revenue_wholesale"] for w in weeks), ), "qty_mix": self._mix( sum(w["qty_retail"] for w in weeks), sum(w["qty_webshop"] for w in weeks), sum(w["qty_wholesale"] for w in weeks), ), "wow_growth_latest": weeks[-1]["wow_growth_pct"] if weeks else None, } # ------------------------------------------------------------------ # Categories (for filter dropdown) # ------------------------------------------------------------------ def get_categories(self) -> list[str]: return self.repo.get_categories() # ------------------------------------------------------------------ # Forecast accuracy — per-week-average rule (matches app.py) # ------------------------------------------------------------------ # # Public surface: # * get_fa_summary — legacy /forecast-accuracy (kept for compat). # * get_fa_tab — new 4-tab response for backtest/live/model_only. # * get_kam_fa_summary — per-person rollup for the KAM·CM tab. # # All three flow through the same _enrich_one / _per_week_avg_metrics # primitives, so the headline rule stays identical across tabs. def get_fa_summary(self, *, tier: Optional[str] = None) -> dict: """Legacy endpoint kept for backward-compat with the old single-page ForecastAccuracy view. Accepts a single tier string (matches the original signature). Returns the original FASummary shape.""" tier_list = [tier] if tier else None raw = self.repo.get_forecast_accuracy(tier=tier_list) rows = [r for r in (self._enrich_one(r) for r in raw) if r is not None] if not rows: return { "overall_fa": 0.0, "overall_fa_signed": 0.0, "overall_bias": 0.0, "hit_rate": 0.0, "n": 0, "by_tier": {}, "by_xyz": {}, "rows": [], } headline = self._per_week_avg_metrics(rows) hit_rate = sum(1 for r in rows if r["hit"]) / len(rows) * 100 tier_groups: dict[str, list[dict]] = defaultdict(list) xyz_groups: dict[str, list[dict]] = defaultdict(list) for r in rows: tier_groups[(r.get("tier") or "N/A")].append(r) xyz_groups[(r.get("xyz") or "N/A")].append(r) return { "overall_fa": headline["fa"], "overall_fa_signed": headline["fa_signed"], "overall_bias": headline["bias"], "hit_rate": hit_rate, "n": len(rows), "by_tier": {k: self._agg(v) for k, v in tier_groups.items()}, "by_xyz": {k: self._agg(v) for k, v in xyz_groups.items()}, "rows": rows[:500], } # ------------------------------------------------------------------ # Four-tab FA — generic builder + three call-sites # ------------------------------------------------------------------ def get_fa_tab( self, *, mode: str, # "backtest" | "live" | "model_only" tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, top_n: int = 10, ) -> dict: """Build a FATabResponse for one of three SQL-row sources. Source map: backtest → backtest_results (channel_mode varies) model_only → backtest_results minus channel_mode='wholesale' live → forecasts + v_sales_weekly_full via DISTINCT ON (run_id ASC) Aggregation is the per-week-average rule for FA/FA_signed/BIAS, and simple per-row mean for hit_rate. Tier and XYZ breakdowns reuse the same rule on each group. Monthly uses sum-then-divide (matches Streamlit "Monthly accuracy (S&OP)" semantics). Top impactors are ranked by abs error contribution. """ if mode == "backtest": raw = self.repo.get_forecast_accuracy( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, ) note = None elif mode == "model_only": raw = self.repo.get_forecast_accuracy( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, model_only=True, ) note = ( "Model-only excludes wholesale channel-mode rows from backtest. " "When on_top_inputs is populated, this view will also subtract " "planner commitments — matching Streamlit's _build_fa_dataset" "('model_only')." ) elif mode == "live": raw = self.repo.get_live_fa( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, ) note = None if raw else ( "Live FA needs the `forecasts` table populated. No rows yet — " "run the forecast pipeline (prompt 4A) to seed it." ) else: raise ValueError(f"unknown FA mode: {mode!r}") rows = [r for r in (self._enrich_one(r) for r in raw) if r is not None] # Filter to Gold/Silver/Bronze only — drop SKUs without a sku_planning # tier (the "N/A" bucket). Polleo's planning process only governs the # tiered set; long-tail unplanned SKUs shouldn't drag the FA headline. rows = [r for r in rows if (r.get("tier") or "").strip()] return self._build_fa_tab_response(mode=mode, rows=rows, note=note, top_n=top_n) def get_fa_breakdown(self, *, weeks_back: int = 4, top_n: int = 12, date_from: Optional[int] = None, date_to: Optional[int] = None) -> dict: """Per-channel forecast (baseline + on-top) vs actual, and per-SKU source of the worst misses, over the chosen closed weeks. Channels Retail / Webshop / Wholesale come from forecasts_detail (as-of = earliest run per SKU·channel·week, summed across regions); actuals from v_sales_weekly_full. Window = [date_from, date_to] if given, else the last `weeks_back` closed weeks that have both a v4 channel forecast and an actual.""" rows = self.repo.db.execute(text(""" WITH asof AS ( SELECT product_id, channel, year, week, MIN(run_id) AS run_id FROM forecasts_detail GROUP BY product_id, channel, year, week ), fd AS ( SELECT a.product_id, a.channel, a.year, a.week, SUM(COALESCE(d.total,0))::float AS total FROM asof a JOIN forecasts_detail d ON d.product_id=a.product_id AND d.channel=a.channel AND d.year=a.year AND d.week=a.week AND d.run_id=a.run_id GROUP BY a.product_id, a.channel, a.year, a.week ), ff AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, COALESCE(on_top_retail,0)::float AS ot_mp, COALESCE(on_top_wholesale,0)::float AS ot_vp FROM forecasts ORDER BY product_id, year, week, run_id ASC, id ASC ) SELECT p.sku, COALESCE(p.name, p.sku) AS name, fd.channel, fd.year, fd.week, fd.total, (CASE fd.channel WHEN 'retail' THEN COALESCE(ff.ot_mp,0) WHEN 'wholesale' THEN COALESCE(ff.ot_vp,0) ELSE 0 END)::float AS on_top, (CASE fd.channel WHEN 'retail' THEN COALESCE(v.qty_retail,0) WHEN 'webshop' THEN COALESCE(v.qty_webshop,0) WHEN 'wholesale' THEN COALESCE(v.qty_wholesale,0) ELSE 0 END)::float AS actual FROM fd JOIN dim_products p ON p.id = fd.product_id LEFT JOIN ff ON ff.product_id=fd.product_id AND ff.year=fd.year AND ff.week=fd.week JOIN v_sales_weekly_full v ON v.product_id=fd.product_id AND v.year=fd.year AND v.week=fd.week WHERE v.qty_total > 0 AND (fd.year*100 + fd.week) < (EXTRACT(isoyear FROM now())::int*100 + EXTRACT(week FROM now())::int) AND (:dfrom IS NULL OR (fd.year*100 + fd.week) >= :dfrom) AND (:dto IS NULL OR (fd.year*100 + fd.week) <= :dto) ORDER BY (fd.year*100 + fd.week) DESC """), {"dfrom": date_from, "dto": date_to}).mappings().all() if not rows: return {"weeks": [], "n_weeks": 0, "channel_summary": [], "offenders": [], "note": "No closed weeks with both a per-channel forecast and an actual yet."} all_weeks = sorted({int(r["year"]) * 100 + int(r["week"]) for r in rows}, reverse=True) weeks = all_weeks if (date_from or date_to) else all_weeks[:weeks_back] wset = set(weeks) rws = [r for r in rows if int(r["year"]) * 100 + int(r["week"]) in wset] CH = (("retail", "Retail / MP"), ("webshop", "Webshop"), ("wholesale", "Wholesale / VP")) agg = {c: {"baseline": 0.0, "ontop": 0.0, "fc": 0.0, "act": 0.0} for c, _ in CH} for r in rws: d = agg.get(r["channel"]) if d is None: continue d["baseline"] += r["total"] - r["on_top"]; d["ontop"] += r["on_top"] d["fc"] += r["total"]; d["act"] += r["actual"] channel_summary = [] for c, label in CH: d = agg[c]; act, fc = d["act"], d["fc"] channel_summary.append({ "channel": label, "fc_baseline": round(d["baseline"]), "fc_ontop": round(d["ontop"]), "fc_total": round(fc), "actual": round(act), "fa": round(max(0.0, 1 - abs(fc - act) / act) * 100, 1) if act > 0 else 0.0, "bias": round((fc - act) / act * 100, 1) if act > 0 else 0.0}) by: dict[str, dict] = {} for r in rws: s = by.setdefault(r["sku"], {"name": r["name"], "baseline": 0.0, "ot_vp": 0.0, "ot_mp": 0.0, "fc": 0.0, "a_retail": 0.0, "a_web": 0.0, "a_vp": 0.0, "a_total": 0.0}) s["baseline"] += r["total"] - r["on_top"]; s["fc"] += r["total"]; s["a_total"] += r["actual"] if r["channel"] == "wholesale": s["ot_vp"] += r["on_top"]; s["a_vp"] += r["actual"] elif r["channel"] == "retail": s["ot_mp"] += r["on_top"]; s["a_retail"] += r["actual"] elif r["channel"] == "webshop": s["a_web"] += r["actual"] offenders = sorted(({ "sku": sku, "name": s["name"], "fc_baseline": round(s["baseline"]), "fc_ontop_vp": round(s["ot_vp"]), "fc_ontop_mp": round(s["ot_mp"]), "fc_total": round(s["fc"]), "act_retail": round(s["a_retail"]), "act_web": round(s["a_web"]), "act_vp": round(s["a_vp"]), "act_total": round(s["a_total"]), "abs_error": round(abs(s["fc"] - s["a_total"])), } for sku, s in by.items()), key=lambda x: x["abs_error"], reverse=True)[:top_n] return {"weeks": [f"{w // 100}W{w % 100:02d}" for w in sorted(weeks)], "n_weeks": len(weeks), "channel_summary": channel_summary, "offenders": offenders, "note": None} def get_kam_fa_summary( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, ) -> dict: """Per-person KAM/CM forecast accuracy. Each person's rollup uses the per-week-average rule on the subset of on_top_inputs they submitted. Hit rate is the per-row mean. Headline numbers reflect the full filtered dataset (across all people).""" raw = self.repo.get_kam_fa( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, ) # KAM rows carry an extra "person" field which _enrich_one drops; # tag it on after enrichment. enriched: list[dict] = [] for r in raw: e = self._enrich_one(r) if e is None: continue e["person"] = r.get("person") or "(unassigned)" e["role"] = r.get("role") enriched.append(e) if not enriched: return { "mode": "kam", "headline": self._empty_headline(), "by_person": [], "note": ( "KAM·CM FA needs `on_top_inputs` populated. No rows yet — " "wholesale/retail planners submit on-top quantities via the " "S&OP cycle workflow." ), } # Headline = full dataset headline_metrics = self._per_week_avg_metrics(enriched) hit_rate = sum(1 for r in enriched if r["hit"]) / len(enriched) * 100 unique_weeks = {(r["year"], r["week"]) for r in enriched} unique_skus = {r["sku"] for r in enriched} # Per-person rollup people: dict[str, list[dict]] = defaultdict(list) roles: dict[str, Optional[str]] = {} for r in enriched: people[r["person"]].append(r) roles.setdefault(r["person"], r.get("role")) by_person: list[dict] = [] for person, group in people.items(): pm = self._per_week_avg_metrics(group) p_hit = sum(1 for x in group if x["hit"]) / len(group) * 100 by_person.append({ "person": person, "role": roles.get(person), "n_sku_weeks": len(group), "forecast": sum(x["forecast"] for x in group), "actual": sum(x["actual"] for x in group), "fa": pm["fa"], "fa_signed": pm["fa_signed"], "bias": pm["bias"], "hit_rate": p_hit, }) by_person.sort(key=lambda r: r["fa"], reverse=True) return { "mode": "kam", "headline": { "fa": headline_metrics["fa"], "fa_signed": headline_metrics["fa_signed"], "bias": headline_metrics["bias"], "hit_rate": hit_rate, "n_sku_weeks": len(enriched), "n_skus": len(unique_skus), "n_weeks": len(unique_weeks), }, "by_person": by_person, "note": None, } # ------------------------------------------------------------------ # Internal: build the full FATabResponse shape from enriched rows. # Shared by backtest / live / model_only — only the source differs. # ------------------------------------------------------------------ def _build_fa_tab_response( self, *, mode: str, rows: list[dict], note: Optional[str], top_n: int, ) -> dict: if not rows: return { "mode": mode, "headline": self._empty_headline(), "by_tier": [], "by_xyz": [], "weekly": [], "monthly": [], "top_impactors": [], "note": note or "No data for the selected filters.", } # Headline: per-week-avg across the entire filtered set. h = self._per_week_avg_metrics(rows) hit_rate = sum(1 for r in rows if r["hit"]) / len(rows) * 100 unique_weeks = {(r["year"], r["week"]) for r in rows} unique_skus = {r["sku"] for r in rows} # Last-week FA signed — just the most recent (year, week) in scope. # Lets the UI show "8-week aggregate" and "last week" side-by-side # without a separate query. latest_yw = max(unique_weeks) if unique_weeks else (0, 0) last_week_rows = [r for r in rows if (r["year"], r["week"]) == latest_yw] last_week_metrics = self._per_week_avg_metrics(last_week_rows) if last_week_rows else self._empty_headline() h["fa_signed_last_week"] = last_week_metrics.get("fa_signed", 0.0) h["last_week_label"] = ( f"{latest_yw[0]}W{latest_yw[1]:02d}" if latest_yw[1] else "" ) # By tier / by xyz — same rule applied to each slice. tier_groups: dict[str, list[dict]] = defaultdict(list) xyz_groups: dict[str, list[dict]] = defaultdict(list) for r in rows: tier_groups[(r.get("tier") or "N/A")].append(r) xyz_groups[(r.get("xyz") or "N/A")].append(r) by_tier = sorted( (self._breakdown_row(k, v) for k, v in tier_groups.items()), key=lambda d: d["label"], ) by_xyz = sorted( (self._breakdown_row(k, v) for k, v in xyz_groups.items()), key=lambda d: d["label"], ) # Weekly — one row per (year, week), per-week FA computed on the sums. weekly: list[dict] = [] wk_acc: dict[tuple[int, int], dict[str, float]] = defaultdict( lambda: {"fc": 0.0, "ac": 0.0} ) for r in rows: k = (r["year"], r["week"]) wk_acc[k]["fc"] += r["forecast"] wk_acc[k]["ac"] += r["actual"] for (y, w), v in wk_acc.items(): ac, fc = v["ac"], v["fc"] if ac <= 0: continue weekly.append({ "year": y, "week": w, "year_week": y * 100 + w, "fa": max(0.0, 1 - abs(fc - ac) / ac) * 100, "fa_signed": fc / ac * 100, "bias": (fc - ac) / ac * 100, "forecast": fc, "actual": ac, }) weekly.sort(key=lambda d: d["year_week"]) # Monthly — Streamlit uses sum-then-divide (totals approach) for the # monthly view, NOT per-week-avg (see app.py:4421 comment). monthly = self._monthly_rollup(rows) # Top impactors — rank by abs-error contribution (whole range). top_impactors = self._top_impactors(rows, top_n=top_n) # Same, but for the most recent closed week only — "top SKUs hurting FA # last week". last_week_impactors: list[dict] = [] last_week_label = "" if rows: last_yw = max(r["year"] * 100 + r["week"] for r in rows) lw_rows = [r for r in rows if r["year"] * 100 + r["week"] == last_yw] last_week_impactors = self._top_impactors(lw_rows, top_n=top_n) last_week_label = f"{last_yw // 100}W{last_yw % 100:02d}" return { "mode": mode, "headline": { "fa": h["fa"], "fa_signed": h["fa_signed"], "bias": h["bias"], "hit_rate": hit_rate, "n_sku_weeks": len(rows), "n_skus": len(unique_skus), "n_weeks": len(unique_weeks), "fa_signed_last_week": h.get("fa_signed_last_week", 0.0), "last_week_label": h.get("last_week_label", ""), }, "by_tier": by_tier, "by_xyz": by_xyz, "weekly": weekly, "monthly": monthly, "top_impactors": top_impactors, "last_week_impactors": last_week_impactors, "last_week_label": last_week_label, "note": note, } @staticmethod def _empty_headline() -> dict: return { "fa": 0.0, "fa_signed": 0.0, "bias": 0.0, "hit_rate": 0.0, "n_sku_weeks": 0, "n_skus": 0, "n_weeks": 0, } @staticmethod def _breakdown_row(label: str, group: list[dict]) -> dict: m = DemandService._per_week_avg_metrics(group) hr = sum(1 for r in group if r["hit"]) / len(group) * 100 if group else 0.0 return { "label": label, "n_sku_weeks": len(group), "n_skus": len({r["sku"] for r in group}), "fa": m["fa"], "fa_signed": m["fa_signed"], "bias": m["bias"], "hit_rate": hr, } @staticmethod def _monthly_rollup(rows: list[dict]) -> list[dict]: """Monthly accuracy — sum-then-divide (totals approach). Matches Streamlit's "Monthly accuracy (S&OP)" view which uses totals-based metrics (see app.py:4423 comment).""" if not rows: return [] buckets: dict[str, dict[str, float]] = defaultdict( lambda: {"fc": 0.0, "ac": 0.0, "n": 0, "hits": 0, "min_yw": 10**9} ) for r in rows: month = _week_to_month_name(r["year"], r["week"]) b = buckets[month] b["fc"] += r["forecast"] b["ac"] += r["actual"] b["n"] += 1 b["hits"] += 1 if r["hit"] else 0 yw = r["year"] * 100 + r["week"] if yw < b["min_yw"]: b["min_yw"] = yw out: list[dict] = [] for month, b in buckets.items(): ac = b["ac"] if ac <= 0: continue fc = b["fc"] out.append({ "month": month, "fa": max(0.0, 1 - abs(fc - ac) / ac) * 100, "fa_signed": fc / ac * 100, "bias": (fc - ac) / ac * 100, "hit_rate": (b["hits"] / b["n"]) * 100 if b["n"] else 0.0, "forecast": fc, "actual": ac, "n_sku_weeks": int(b["n"]), "_min_yw": b["min_yw"], }) # Order by chronology (first ISO week appearance), not alphabetical. out.sort(key=lambda d: d["_min_yw"]) for d in out: d.pop("_min_yw") return out @staticmethod def _top_impactors(rows: list[dict], *, top_n: int) -> list[dict]: """Top-N SKUs by absolute error contribution. Matches app.py:4475-4494 "Top SKUs ruining FA" table: - group by SKU, sum actual / forecast / abs_error - per-SKU FA/FA_signed/BIAS computed totals-style with actual.clip(lower=1) to avoid div-by-zero - share = sku_abs_err / total_abs_err × 100 """ if not rows: return [] by_sku: dict[str, dict] = {} meta: dict[str, dict] = {} for r in rows: sku = r["sku"] d = by_sku.setdefault(sku, {"actual": 0.0, "forecast": 0.0, "abs_error": 0.0}) d["actual"] += r["actual"] d["forecast"] += r["forecast"] d["abs_error"] += abs(r["forecast"] - r["actual"]) meta.setdefault(sku, { "name": r.get("name"), "tier": r.get("tier"), "xyz": r.get("xyz"), }) total_err = sum(d["abs_error"] for d in by_sku.values()) or 1.0 ranked = sorted(by_sku.items(), key=lambda kv: kv[1]["abs_error"], reverse=True)[:top_n] out: list[dict] = [] for sku, d in ranked: a = d["actual"] f = d["forecast"] denom = a if a >= 1.0 else 1.0 out.append({ "sku": sku, "name": meta[sku]["name"], "tier": meta[sku]["tier"], "xyz": meta[sku]["xyz"], "actual": a, "forecast": f, "abs_error": d["abs_error"], "fa": max(0.0, 1 - abs(f - a) / denom) * 100, "fa_signed": (f / denom) * 100, "bias": ((f - a) / denom) * 100, "share_of_total_error_pct": d["abs_error"] / total_err * 100, }) return out # ------------------------------------------------------------------ # Per-row enrichment # ------------------------------------------------------------------ @staticmethod def _enrich_one(r: dict) -> Optional[dict]: """Per-row metric computation. Matches app.py:4172-4181 verbatim: - Drops rows where actual <= 0 (the `fa[fa["actual"] > 0]` filter). - fa = max(0, 1 - |F-A|/A) × 100 - fa_signed = F/A × 100 - bias = (F-A)/A × 100 (exposed for row-detail; not used at headline) - hit = |F-A| / max(A, 1) <= 0.30 ↑ the .clip(lower=1) on the denominator matters for SKU-weeks where 0 < actual < 1 — without it a forecast of 0.7 vs actual of 0.5 (|err|/A = 0.4) would be a miss, but with it (|err|/1 = 0.2) it's a hit. Defensible since those rows would otherwise dominate the miss rate. """ f = r.get("forecast") a = r.get("actual") if f is None or a is None: return None f = float(f) a = float(a) if a <= 0: return None error = abs(f - a) hit_denom = a if a >= 1.0 else 1.0 # actual.clip(lower=1) return { "sku": r["sku"], "name": r.get("name"), "year": int(r["year"]), "week": int(r["week"]), "forecast": f, "actual": a, "tier": r.get("tier"), "xyz": r.get("xyz"), "category": r.get("category"), "model": r.get("model"), "channel_mode": r.get("channel_mode"), "fa": max(0.0, 1 - error / a) * 100, "fa_signed": f / a * 100, "bias": (f - a) / a * 100, "hit": (error / hit_denom) <= 0.30, } # ------------------------------------------------------------------ # Per-week-average aggregation (Streamlit's default headline rule) # ------------------------------------------------------------------ @staticmethod def _per_week_avg_metrics(rows: list[dict]) -> dict: """Matches app.py:4430-4442 `_per_week_avg_metrics`. For each (year, week) bucket: sum forecast, sum actual. Drop weeks where sum_actual <= 0. Per week: fa_w = max(0, 1 - |sumF - sumA| / sumA) × 100 fa_signed_w = sumF / sumA × 100 bias_w = (sumF - sumA) / sumA × 100 Headline = simple arithmetic mean across weeks. Note: Streamlit groups by `week` only; we use (year, week) to be safe against multi-year data. Identical when data is single-year (our current backtest_results is all 2026). """ if not rows: return {"fa": 0.0, "fa_signed": 0.0, "bias": 0.0, "n_weeks": 0} wk: dict[tuple[int, int], dict[str, float]] = defaultdict( lambda: {"fc": 0.0, "ac": 0.0} ) for r in rows: key = (r["year"], r["week"]) wk[key]["fc"] += r["forecast"] wk[key]["ac"] += r["actual"] fa_list: list[float] = [] fa_signed_list: list[float] = [] bias_list: list[float] = [] for v in wk.values(): ac = v["ac"] fc = v["fc"] if ac <= 0: continue fa_list.append(max(0.0, 1 - abs(fc - ac) / ac) * 100) fa_signed_list.append(fc / ac * 100) bias_list.append((fc - ac) / ac * 100) if not fa_list: return {"fa": 0.0, "fa_signed": 0.0, "bias": 0.0, "n_weeks": 0} n = len(fa_list) return { "fa": sum(fa_list) / n, "fa_signed": sum(fa_signed_list) / n, "bias": sum(bias_list) / n, "n_weeks": n, } @staticmethod def _agg(group: list[dict]) -> dict: """Tier / XYZ breakdown row. Matches app.py:4511-4528 `_breakdown`: FA and BIAS use the per-week-average rule; hit rate is the simple arithmetic mean of per-row `hit` flags.""" if not group: return {"fa": 0.0, "bias": 0.0, "hit_rate": 0.0, "n": 0} metrics = DemandService._per_week_avg_metrics(group) hit_rate = sum(1 for r in group if r["hit"]) / len(group) * 100 return { "fa": metrics["fa"], "bias": metrics["bias"], "hit_rate": hit_rate, "n": len(group), } # ------------------------------------------------------------------ # Demand planning — main planner grid (read-only for now) # ------------------------------------------------------------------ def get_demand_plan_data( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, ) -> dict: """Build the DemandPlanResponse for the planning grid. Pivots the (sku, year, week) rows from `forecasts` into one DemandPlanRow per SKU with a sorted weeks[] array, computes row totals across the horizon, and assembles grand totals + the on-top summary used by the page header. Empty when `forecasts` is empty — returns a stub with a note so the page can render the "no forecast yet" state without a 404. """ latest = self.repo.get_latest_forecast_run() cycle_info: Optional[dict] = None if latest: cycle_info = { "cycle_id": latest.get("cycle_id"), "cycle_year_week": latest.get("cycle_year_week"), "cycle_status": latest.get("cycle_status"), "run_id": latest.get("run_id"), "run_started_at": latest.get("started_at").isoformat() if latest.get("started_at") else None, "run_by": latest.get("run_by"), "n_skus": latest.get("n_skus"), } raw = self.repo.get_demand_plan( tier=tier, xyz=xyz, category=category, run_id=(latest or {}).get("run_id"), ) on_top = self._build_on_top_summary( cycle_id=(latest or {}).get("cycle_id") if latest else None ) if not raw: note = self._planning_note(latest is not None) return { "rows": [], "horizon": [], "cycle_info": cycle_info, "on_top_summary": on_top, "grand_total_baseline": 0.0, "grand_total_on_top_vp": 0.0, "grand_total_on_top_mp": 0.0, "grand_total_promo_uplift": 0.0, "grand_total": 0.0, "note": note, } # Pivot rows → one DemandPlanRow per SKU. dict preserves insertion # order, so the first occurrence of a SKU sets its ordering. by_sku: dict[str, dict] = {} horizon_set: set[tuple[int, int]] = set() for r in raw: sku = r["sku"] horizon_set.add((int(r["year"]), int(r["week"]))) sku_entry = by_sku.setdefault(sku, { "sku": sku, "name": r.get("name"), "category": r.get("category"), "tier": r.get("tier"), "xyz": r.get("xyz"), "weeks": [], "total_baseline": 0.0, "total_on_top_vp": 0.0, "total_on_top_mp": 0.0, "total_promo_uplift": 0.0, "total": 0.0, }) y, w = int(r["year"]), int(r["week"]) baseline = float(r.get("baseline") or 0) on_top_vp = float(r.get("on_top_vp") or 0) on_top_mp = float(r.get("on_top_mp") or 0) promo_up = float(r.get("promo_uplift") or 0) factor = float(r.get("planner_factor") or 1.0) total = float(r.get("total") or 0) sku_entry["weeks"].append({ "year": y, "week": w, "year_week": y * 100 + w, "cw_label": f"CW{w:02d}", "baseline": baseline, "on_top_vp": on_top_vp, "on_top_mp": on_top_mp, "promo_uplift": promo_up, "planner_factor": factor, "total": total, }) sku_entry["total_baseline"] += baseline sku_entry["total_on_top_vp"] += on_top_vp sku_entry["total_on_top_mp"] += on_top_mp sku_entry["total_promo_uplift"] += promo_up sku_entry["total"] += total # Sort each SKU's weeks chronologically (defensive — repo orders # but a SKU may have a gap if the run skipped weeks for it). for entry in by_sku.values(): entry["weeks"].sort(key=lambda w: w["year_week"]) rows_out = list(by_sku.values()) # Default sort: by SKU. Frontend can re-sort. rows_out.sort(key=lambda r: r["sku"]) horizon = [ f"CW{w:02d}" for (y, w) in sorted(horizon_set, key=lambda yw: yw[0] * 100 + yw[1]) ] return { "rows": rows_out, "horizon": horizon, "cycle_info": cycle_info, "on_top_summary": on_top, "grand_total_baseline": sum(r["total_baseline"] for r in rows_out), "grand_total_on_top_vp": sum(r["total_on_top_vp"] for r in rows_out), "grand_total_on_top_mp": sum(r["total_on_top_mp"] for r in rows_out), "grand_total_promo_uplift": sum(r["total_promo_uplift"] for r in rows_out), "grand_total": sum(r["total"] for r in rows_out), "note": None, } def _build_on_top_summary(self, *, cycle_id: Optional[int]) -> dict: raw = self.repo.get_on_top_summary_rows(cycle_id=cycle_id) vp_total = 0.0 mp_total = 0.0 people: dict[str, dict] = {} for r in raw: channel = (r.get("channel") or "").lower() qty = float(r.get("qty_total") or 0) if channel == "wholesale": vp_total += qty elif channel == "retail": mp_total += qty person = r["person"] p_entry = people.setdefault(person, { "person": person, "role": r.get("role"), "channel": None, "qty_total": 0.0, "n_sku_weeks": 0, "last_submitted_at": None, }) p_entry["channel"] = channel or p_entry["channel"] p_entry["qty_total"] += qty p_entry["n_sku_weeks"] += int(r.get("n_sku_weeks") or 0) submitted = r.get("last_submitted_at") iso = submitted.isoformat() if submitted else None if iso and (p_entry["last_submitted_at"] is None or iso > p_entry["last_submitted_at"]): p_entry["last_submitted_at"] = iso by_submitter = sorted(people.values(), key=lambda p: p["qty_total"], reverse=True) return { "vp_total": vp_total, "mp_total": mp_total, "n_submitters": len(by_submitter), "by_submitter": by_submitter, } # ------------------------------------------------------------------ # Demand Planning page (Streamlit page_demand_planning parity) # ------------------------------------------------------------------ def get_planning_view( self, *, category: Optional[str] = None, oznaka: Optional[str] = None, xyz: Optional[list[str]] = None, sku: Optional[str] = None, yw_from: Optional[int] = None, yw_to: Optional[int] = None, ) -> dict: """Build the response for /demand/planning-view. Filters semantics match Streamlit page_demand_planning: - category single-select (or None) - oznaka single-select (or None) - xyz multi-select (list or None) - sku optional drill-down Date range narrows historical weeks only — forecast weeks always return for the latest run's horizon. """ latest = self.repo.get_latest_forecast_run() cycle_info: Optional[dict] = None if latest: cycle_info = { "cycle_id": latest.get("cycle_id"), "cycle_year_week": latest.get("cycle_year_week"), "cycle_status": latest.get("cycle_status"), "run_id": latest.get("run_id"), "run_started_at": latest.get("started_at").isoformat() if latest.get("started_at") else None, "run_by": latest.get("run_by"), "n_skus": latest.get("n_skus"), } # SKU options for the selectbox — narrowed by category/oznaka/xyz sku_options = self.repo.get_planning_view_skus( category=category, oznaka=oznaka, xyz=xyz, ) n_skus_filter = len(sku_options) is_single_sku = bool(sku) selected: Optional[dict] = None skus_in_scope: Optional[list[str]] = None if is_single_sku: meta = self.repo.get_planning_view_sku_meta(sku) if meta is None: return { "display_title": f"SKU not found: {sku}", "caption": None, "is_single_sku": True, "n_skus": 0, "sku_options": sku_options, "historical": [], "forecast": [], "selected": None, "cycle_info": cycle_info, "note": f"SKU '{sku}' is not in dim_products.", } selected = { "sku": meta["sku"], "name": meta.get("name"), "category": meta.get("category"), "oznaka": meta.get("oznaka"), "xyz": meta.get("xyz"), } skus_in_scope = [sku] # Historical actuals hist_raw = self.repo.get_planning_view_historical( skus=skus_in_scope, category=category, oznaka=oznaka, xyz=xyz, yw_from=yw_from, yw_to=yw_to, ) historical = [ { "year": int(r["year"]), "week": int(r["week"]), "year_week": int(r["year_week"]), "cw_label": f"CW{int(r['week']):02d}", "qty_total": float(r["qty_total"] or 0), "is_promo": bool(r["is_promo"]), } for r in hist_raw ] # Forecast components fc_raw = self.repo.get_planning_view_forecast( skus=skus_in_scope, category=category, oznaka=oznaka, xyz=xyz, run_id=(latest or {}).get("run_id"), ) forecast = [ { "year": int(r["year"]), "week": int(r["week"]), "year_week": int(r["year_week"]), "cw_label": f"CW{int(r['week']):02d}", "baseline": float(r["baseline"] or 0), "planner_factor": float(r["planner_factor"] or 1.0), "stat_adjusted": float(r["stat_adjusted"] or 0), "on_top_vp": float(r["on_top_vp"] or 0), "on_top_mp": float(r["on_top_mp"] or 0), "promo_uplift": float(r["promo_uplift"] or 0), "total": float(r["total"] or 0), } for r in fc_raw ] # Display title + caption — mirrors Streamlit format if is_single_sku and selected: name = selected.get("name") or selected["sku"] display_title = f"{name} ({selected['sku']})" cap_parts = [] if selected.get("category"): cap_parts.append(selected["category"]) if selected.get("oznaka"): cap_parts.append(selected["oznaka"]) if selected.get("xyz"): cap_parts.append(f"XYZ: {selected['xyz']}") caption = " · ".join(cap_parts) if cap_parts else None n_skus = 1 else: label_parts: list[str] = [] if category: label_parts.append(category) if oznaka: label_parts.append(oznaka) if xyz: label_parts.append("XYZ: " + ",".join(xyz)) group = " · ".join(label_parts) if label_parts else "All SKUs" display_title = f"{group} ({n_skus_filter} SKUs)" caption = None n_skus = n_skus_filter note: Optional[str] = None if not forecast and not historical: note = self._planning_note(latest is not None) return { "display_title": display_title, "caption": caption, "is_single_sku": is_single_sku, "n_skus": n_skus, "sku_options": sku_options, "historical": historical, "forecast": forecast, "selected": selected, "cycle_info": cycle_info, "note": note, } def save_planner_factors(self, *, sku: str, factors: list[dict]) -> dict: n = self.repo.save_planner_factors(sku=sku, factors=factors) return {"updated": n} # ------------------------------------------------------------------ # Revenue forecast page (Streamlit page_revenue parity) # ------------------------------------------------------------------ def get_revenue_forecast( self, *, view: str = "revenue", # "revenue" | "ruc" category: Optional[str] = None, source: str = "all", # "all" | "vp" | "mp" months_filter: Optional[list[str]] = None, include_nonplanned: bool = False, ) -> dict: """Build the response for /demand/revenue-forecast. Forecast values: baseline = SUM(baseline × factor × rate_for_channel) where rate depends on `view` and `source`: view=revenue, source=all: ws_share × price + (1-ws_share) × price = price view=revenue, source=vp: ws_share × price (wholesale share) view=revenue, source=mp: (1 - ws_share) × price view=ruc: same, but rate = retail_ruc / wholesale_ruc vp_ontop = SUM(on_top_vp × wholesale rate) mp_ontop = SUM(on_top_mp × retail rate) Past actuals use v_sales_weekly_full × erp_prices, with the same channel split. RUC past actuals come from erp_transactions.ruc_eur. Gross-up: applied to forecast baseline only when include_nonplanned is true. Ratio is channel-specific. """ from datetime import datetime as _dt is_ruc = view == "ruc" has_ruc = self.repo.get_revenue_has_ruc_data() if is_ruc and not has_ruc: view = "revenue" is_ruc = False latest = self.repo.get_latest_forecast_run() run_id = (latest or {}).get("run_id") fc_rows = self.repo.get_revenue_forecast_rows( run_id=run_id, category=category, ) # Load enough history to FULLY cover the selected month(s). With a fixed # 8-week window, viewing e.g. April from June dropped its earliest ISO # week entirely (a whole week of RUC/revenue missing → month understated # vs management's calendar-month figure). When a month is filtered, reach # back to its start (+ a 2-week buffer for boundary weeks); else keep the # recent 8-week default. n_past = 8 if months_filter: try: earliest = min(_dt.strptime(m, "%b %Y") for m in months_filter) days_back = (_dt.now() - earliest).days n_past = max(8, days_back // 7 + 2) except ValueError: n_past = 30 past_rows = self.repo.get_revenue_past_weeks( n_past_weeks=n_past, category=category, include_nonplanned=include_nonplanned, ) # Categories list (sorted) — derived from forecast rows + past rows cat_set: set[str] = set() for r in fc_rows: cat_set.add(r["category"]) for r in past_rows: cat_set.add(r["category"]) categories_sorted = sorted(cat_set) # Gross-up ratios can_grossup = category is None grossup_data = ( self.repo.get_revenue_grossup_ratios() if (include_nonplanned and can_grossup) else None ) if source == "vp": gu_info = (grossup_data or {}).get("ws", {"ratio": 1.0, "share_pct": 0.0}) elif source == "mp": gu_info = (grossup_data or {}).get("retail", {"ratio": 1.0, "share_pct": 0.0}) else: gu_info = (grossup_data or {}).get("all", {"ratio": 1.0, "share_pct": 0.0}) effective_gu = float(gu_info["ratio"]) effective_share = float(gu_info["share_pct"]) # ---------- Forecast aggregation by (year, week, category) ---------- # baseline_value = SUM(units_baseline × channel_rate) # vp_value = SUM(units_vp × wholesale_rate) # mp_value = SUM(units_mp × retail_rate) # rates depend on view (revenue vs ruc) and source filter. def _ret_rate(row: dict) -> float: if is_ruc: r = row.get("retail_ruc_rate") return float(r) if r is not None else 0.0 # Revenue: realized retail net price; fallback to avg_sell_price # for SKUs with no recent sales to derive a rate from. r = row.get("retail_price_rate") return float(r) if r is not None else float(row.get("price") or 0) def _ws_rate(row: dict) -> float: if is_ruc: r = row.get("wholesale_ruc_rate") return float(r) if r is not None else 0.0 # Revenue: realized WHOLESALE net price (much lower than retail) — # not the retail avg_sell_price. r = row.get("wholesale_price_rate") return float(r) if r is not None else float(row.get("price") or 0) # per (year,week,cat) → {baseline, vp, mp, _b_ws, _b_b2c} # _b_ws / _b_b2c track WS-only and B2C-only baseline contributions # separately so per-month retail target calibration (see below) # can scale just the B2C portion. The summed `baseline` is what the # downstream chart/category code reads; calibration updates both # the per-channel breakdown and the summed field consistently. wk_cat: dict[tuple[int, int, str], dict[str, float]] = {} for r in fc_rows: yw_key = (int(r["year"]), int(r["week"]), str(r["category"])) bucket = wk_cat.setdefault(yw_key, { "baseline": 0.0, "vp": 0.0, "mp": 0.0, "_b_ws": 0.0, "_b_b2c": 0.0, }) units_b = float(r["units_baseline"] or 0) ws_share = float(r["ws_share"] or 0) ret_rate = _ret_rate(r) ws_rate = _ws_rate(r) b_ws_part = units_b * ws_share * ws_rate b_b2c_part = units_b * (1 - ws_share) * ret_rate if source == "vp": bucket["baseline"] += b_ws_part bucket["_b_ws"] += b_ws_part elif source == "mp": bucket["baseline"] += b_b2c_part bucket["_b_b2c"] += b_b2c_part else: # all → blended (ws portion at ws rate, retail portion at retail rate) bucket["baseline"] += b_ws_part + b_b2c_part bucket["_b_ws"] += b_ws_part bucket["_b_b2c"] += b_b2c_part if source != "mp": bucket["vp"] += float(r["units_vp"] or 0) * ws_rate if source != "vp": bucket["mp"] += float(r["units_mp"] or 0) * ret_rate # Apply gross-up to baseline only (Streamlit: VP/MP on-tops unscaled) if effective_gu != 1.0: for v in wk_cat.values(): v["baseline"] *= effective_gu v["_b_ws"] *= effective_gu v["_b_b2c"] *= effective_gu # ── RETAIL TARGET CALIBRATION (Revenue Forecast view) ──────────── # Mirrors monthly_plan_service.CALIBRATION_BY_MONTH — when a month # is pinned to mgmt_retail × (1 + pct), scale this view's B2C # baseline + B2C on-top so the displayed retail aligns with the # locked plan. The scale factor is derived in *RUC space* (view- # independent volume reduction) and applied uniformly: reducing # retail volume by factor v drops both retail revenue and retail # RUC by v, so the same multiplier works for both views. from backend.services.monthly_plan_service import CALIBRATION_BY_MONTH # Pre-compute the volume scale factor per calibrated month, using # RUC values from fc_rows (independent of which view is active). retail_scale_by_month: dict[str, float] = {} for month_key, cal in CALIBRATION_BY_MONTH.items(): retail_pct = cal.get("retail_pct_above_mgmt") if retail_pct is None: continue cal_year = month_key // 100 cal_month = month_key % 100 cal_label = _dt(cal_year, cal_month, 1).strftime("%b %Y") if months_filter and cal_label not in months_filter: continue mgmt_retail_ruc = float(self.repo.db.execute(text(""" SELECT COALESCE(SUM(ruc_plan), 0)::float FROM management_plan_lines WHERE month_key = :mk AND channel IN ('Retail domestic', 'Retail international') """), {"mk": month_key}).scalar() or 0) if mgmt_retail_ruc <= 0: continue target_retail_ruc = mgmt_retail_ruc * (1.0 + float(retail_pct)) # Re-aggregate fc_rows to get current retail RUC for this month current_retail_ruc = 0.0 for r in fc_rows: y, w = int(r["year"]), int(r["week"]) try: wk_month = _dt.strptime(f"{y}-W{w:02d}-4", "%G-W%V-%u").strftime("%b %Y") except ValueError: continue if wk_month != cal_label: continue units_b = float(r["units_baseline"] or 0) units_mp = float(r["units_mp"] or 0) ws_share = float(r["ws_share"] or 0) ruc_rate = r.get("retail_ruc_rate") ruc_rate = float(ruc_rate) if ruc_rate is not None else 0.0 current_retail_ruc += (units_b * (1 - ws_share) + units_mp) * ruc_rate # Gross-up affects baseline only (not on-tops). Approximate by # applying ratio to whole sum — the on-top share is small. current_retail_ruc *= effective_gu if effective_gu > 0 else 1.0 if current_retail_ruc <= 0: continue scale = target_retail_ruc / current_retail_ruc if scale < 1.0: # only reduce retail_scale_by_month[cal_label] = scale # Apply scale to wk_cat B2C portions if retail_scale_by_month: for (y, w, _c), buc in wk_cat.items(): try: lab = _dt.strptime(f"{y}-W{w:02d}-4", "%G-W%V-%u").strftime("%b %Y") except ValueError: continue scale = retail_scale_by_month.get(lab) if not scale: continue delta_b2c = buc["_b_b2c"] * (1.0 - scale) delta_mp = buc.get("mp", 0.0) * (1.0 - scale) buc["_b_b2c"] -= delta_b2c buc["baseline"] -= delta_b2c buc["mp"] -= delta_mp # ---------- Pro-rate missing RUC weeks ---------- # erp_transactions can have gaps for weeks that were never uploaded # (e.g. CW16/2026 — a missed weekly upload). v_sales_weekly_full # still shows revenue for those weeks (from sales_clean_import), # but the RUC join returns 0. To keep the RUC chart smooth we # impute the missing RUC = revenue × margin-ratio, where the # ratio is the qty-weighted mean across the OTHER past weeks # for the SAME category (with a global fallback). cat_mar: dict[str, dict[str, float]] = {} # cat → accumulators for r in past_rows: cat = str(r["category"]) ruc_r = float(r["ruc_retail_total"] or 0) ruc_ws = float(r["ruc_wholesale_total"] or 0) rev_r = float(r["rev_retail"] or 0) + float(r["rev_webshop"] or 0) rev_ws = float(r["rev_wholesale"] or 0) acc = cat_mar.setdefault(cat, { "ruc_r": 0.0, "rev_r": 0.0, "ruc_ws": 0.0, "rev_ws": 0.0, }) # Only count weeks with non-zero RUC — the gap weeks should # not bias the margin estimate they're trying to fill. if ruc_r > 0: acc["ruc_r"] += ruc_r; acc["rev_r"] += rev_r if ruc_ws > 0: acc["ruc_ws"] += ruc_ws; acc["rev_ws"] += rev_ws global_mar = { "retail": ( sum(c["ruc_r"] for c in cat_mar.values()) / max(sum(c["rev_r"] for c in cat_mar.values()), 1e-9) ), "wholesale": ( sum(c["ruc_ws"] for c in cat_mar.values()) / max(sum(c["rev_ws"] for c in cat_mar.values()), 1e-9) ), } def _margin(cat: str, channel: str) -> float: agg = cat_mar.get(cat) if agg: rev_key = "rev_r" if channel == "retail" else "rev_ws" ruc_key = "ruc_r" if channel == "retail" else "ruc_ws" if agg[rev_key] > 0: return agg[ruc_key] / agg[rev_key] return global_mar[channel] imputed_weeks: set[tuple[int, int]] = set() imputed_eur = 0.0 # ---------- Past actuals aggregation by (year, week, category) ---------- past_wk: dict[tuple[int, int, str], dict[str, float]] = {} for r in past_rows: y, w = int(r["year"]), int(r["week"]) cat = str(r["category"]) key = (y, w, cat) bucket = past_wk.setdefault(key, {"actual": 0.0}) rev_r = float(r["rev_retail"] or 0) + float(r["rev_webshop"] or 0) rev_ws = float(r["rev_wholesale"] or 0) ruc_r = float(r["ruc_retail_total"] or 0) ruc_ws = float(r["ruc_wholesale_total"] or 0) # Pro-rate missing RUC weeks from category-level margin ratio. if ruc_r == 0 and rev_r > 0: est = rev_r * _margin(cat, "retail") if est > 0: ruc_r = est imputed_weeks.add((y, w)) imputed_eur += est if ruc_ws == 0 and rev_ws > 0: est = rev_ws * _margin(cat, "wholesale") if est > 0: ruc_ws = est imputed_weeks.add((y, w)) imputed_eur += est if is_ruc: if source == "vp": bucket["actual"] += ruc_ws elif source == "mp": bucket["actual"] += ruc_r else: bucket["actual"] += ruc_r + ruc_ws else: if source == "vp": bucket["actual"] += rev_ws elif source == "mp": bucket["actual"] += rev_r else: bucket["actual"] += rev_r + rev_ws # ---------- Build chart per (year, week) ---------- # First collect ordered (year, week) list and compute month labels. all_weeks: set[tuple[int, int]] = set() for (y, w, _c) in wk_cat: all_weeks.add((y, w)) for (y, w, _c) in past_wk: all_weeks.add((y, w)) ordered = sorted(all_weeks) # Month label of a week (Thursday rule — used only for the month # dropdown labels in the UI; the data attribution under filter uses # day-fraction below). def _month(y: int, w: int) -> str: try: return _dt.strptime(f"{y}-W{w:02d}-4", "%G-W%V-%u").strftime("%b %Y") except ValueError: return f"W{w} {y}" # Day-fraction in the selected months. When no months_filter is # active every week contributes 100%; when months_filter is set, # weeks that straddle the boundary contribute pro-rata by day count. # This makes the filtered total agree with the monthly snapshot # (which uses day-fraction too) instead of clipping by Thursday rule. from datetime import timedelta def _frac_in_months(y: int, w: int, month_set: set[str]) -> float: try: mon = _dt.strptime(f"{y}-W{w:02d}-1", "%G-W%V-%u").date() except ValueError: return 0.0 days = 0 for off in range(7): d = mon + timedelta(days=off) if d.strftime("%b %Y") in month_set: days += 1 return days / 7.0 if months_filter: month_set = set(months_filter) week_frac: dict[tuple[int, int], float] = { (y, w): _frac_in_months(y, w, month_set) for (y, w) in ordered } ordered = [(y, w) for (y, w) in ordered if week_frac[(y, w)] > 0] else: week_frac = {(y, w): 1.0 for (y, w) in ordered} cur_iso = _dt.now().isocalendar() cur_yw = int(cur_iso[0]) * 100 + int(cur_iso[1]) # --- Plan distribution ---------------------------------------------- # The annual plan workbook gives monthly totals per group. We divide # each monthly total by the number of ISO weeks IN THAT CALENDAR # MONTH (4 or 5, counted by Mondays) so every week of the year gets a # fair share of the monthly plan. Aggregating those weekly slices # back to a calendar month therefore reconstructs the monthly target # only when the whole month is visible; for a partial month visible # in the chart, the comparison is partial-actual vs partial-plan # rather than partial-actual vs full-month-plan. # # Skip when the filter is VP/MP only (the workbook isn't split by # channel — showing the all-channel target against a VP/MP-only line # would be misleading). # Plan source changed (May 2026) — the legacy plan_loader / plan # revenu_ruc.xlsx is removed. The demand-planning cycle now runs # entirely through the app: KAM/CM enter on-tops via UI → # `on_top_inputs` table → Run Forecast → engine writes `forecasts` # where `forecasts.total = baseline × planner_factor + # on_top_wholesale + on_top_retail + promo_uplift`. The "plan" IS # the forecast — no separate Plan line on this page. plan_enabled = False chart: list[dict] = [] for (y, w) in ordered: yw = y * 100 + w is_past = yw < cur_yw frac = week_frac.get((y, w), 1.0) baseline = sum(wk_cat.get((y, w, c), {}).get("baseline", 0.0) for c in categories_sorted) * frac vp_on = sum(wk_cat.get((y, w, c), {}).get("vp", 0.0) for c in categories_sorted) * frac mp_on = sum(wk_cat.get((y, w, c), {}).get("mp", 0.0) for c in categories_sorted) * frac actual = sum(past_wk.get((y, w, c), {}).get("actual", 0.0) for c in categories_sorted) * frac forecast_val: Optional[float] if source == "vp": forecast_val = baseline + vp_on elif source == "mp": forecast_val = baseline + mp_on else: forecast_val = baseline + vp_on + mp_on # Don't show forecast for past weeks (just actuals) forecast_for_point = None if is_past else forecast_val actual_for_point = actual if is_past else None plan_for_point: Optional[float] = None # legacy field, always None now try: monday = _dt.strptime(f"{y}-W{w:02d}-1", "%G-W%V-%u").date().isoformat() except ValueError: monday = "" chart.append({ "year": y, "week": w, "year_week": yw, "cw_label": f"CW{w:02d}", "month_label": _month(y, w), "date_from": monday, "is_past": is_past, "actual": actual_for_point, "forecast": forecast_for_point, "vp_ontop": vp_on if not is_past and source in ("all", "vp") else None, "mp_ontop": mp_on if not is_past and source in ("all", "mp") else None, "plan": plan_for_point, "day_fraction": frac, }) # ---------- Category breakdown ---------- cat_break: list[dict] = [] for cat in categories_sorted: base = sum( wk_cat.get((y, w, cat), {}).get("baseline", 0.0) * week_frac.get((y, w), 1.0) for (y, w) in ordered ) vp = sum( wk_cat.get((y, w, cat), {}).get("vp", 0.0) * week_frac.get((y, w), 1.0) for (y, w) in ordered ) mp = sum( wk_cat.get((y, w, cat), {}).get("mp", 0.0) * week_frac.get((y, w), 1.0) for (y, w) in ordered ) if source == "vp": shown = base + vp elif source == "mp": shown = base + mp else: shown = base + vp + mp if shown == 0 and base == 0 and vp == 0 and mp == 0: continue cat_break.append({ "category": cat, "baseline": base, "vp_ontop": vp, "mp_ontop": mp, "shown_total": shown, }) cat_break.sort(key=lambda r: r["shown_total"], reverse=True) # ---------- Totals ---------- total_value = sum((p["actual"] or 0) + (p["forecast"] or 0) for p in chart) n_weeks = max(len(chart), 1) avg_weekly = total_value / n_weeks plan_total: Optional[float] = None # legacy field, always None — see comment above # Margin %: only meaningful in RUC view with a non-zero revenue denominator. margin_pct: Optional[float] = None if is_ruc and total_value > 0: # Recompute equivalent revenue total using prices instead of RUC rates. # Simple approach: estimate from cat_break.shown_total ratio between RUC and revenue. # We skip this for simplicity in v1 — frontend can hide if null. pass # ---------- Month list for the filter ---------- # Build "all months that the data spans" all_months: list[str] = [] seen_m: set[str] = set() for (y, w) in sorted(all_weeks): m = _month(y, w) if m not in seen_m: seen_m.add(m) all_months.append(m) months: list[dict] = [] for m in all_months: cws = [ f"CW{w:02d}" for (y, w) in sorted(all_weeks) if _month(y, w) == m ] months.append({"label": m, "cws": cws}) note: Optional[str] = None if not fc_rows: note = "No forecast rows for the selected filter — run the forecast pipeline or widen the filter." elif is_ruc and imputed_weeks: wks = sorted(imputed_weeks) label = ", ".join(f"CW{w:02d}" for (_, w) in wks) note = ( f"RUC imputed for {len(wks)} week(s) with missing ERP uploads: " f"{label}. Revenue is from sales history; RUC estimated as " f"revenue × category-average margin ratio from neighbouring " f"weeks (≈ €{imputed_eur:,.0f} total imputed)." ) return { "view": view, "source": source, "unit": "RUC" if is_ruc else "Revenue", "unit_symbol": "€", "has_ruc_data": has_ruc, "can_grossup": can_grossup, "categories": categories_sorted, "months": months, "chart": chart, "category_breakdown": cat_break, "total_value": total_value, "avg_weekly": avg_weekly, "margin_pct": margin_pct, "gross_up": { "applied": effective_gu != 1.0, "ratio": effective_gu, "nonplan_share_pct": effective_share, }, "plan_total": plan_total, "note": note, } @staticmethod def _planning_note(has_run: bool) -> str: if not has_run: return ( "No forecast runs yet. Demand Planning needs the forecast " "engine to populate the `forecasts` table — wire it up in " "prompt 4A. Filters and the on-top summary will work as " "soon as data lands." ) return ( "Latest forecast run produced no rows for the selected filters. " "Try widening the filter (e.g. include all tiers) or re-running " "the forecast pipeline." ) # ------------------------------------------------------------------ # Consensus snapshots — list / detail / diff # ------------------------------------------------------------------ def get_consensus_history(self) -> dict: raw = self.repo.get_consensus_snapshots() snapshots = [self._serialize_summary(r) for r in raw] note = None if snapshots else ( "No consensus snapshots yet. They're created when a planning " "cycle is closed and the consensus plan is frozen." ) return {"snapshots": snapshots, "note": note} def get_consensus_snapshot(self, snapshot_id: int) -> Optional[dict]: row = self.repo.get_consensus_detail(snapshot_id) if not row: return None # JSONB fields may come back as list, dict, or None depending on what # was written. Normalise to list[dict] for the frontend. def _as_list(v): if v is None: return [] if isinstance(v, list): return [r for r in v if isinstance(r, dict)] if isinstance(v, dict): return [v] return [] return { **self._serialize_summary(row), "rows": _as_list(row.get("rows")), "wholesale_inputs": _as_list(row.get("wholesale_inputs")), "retail_inputs": _as_list(row.get("retail_inputs")), } def compare_snapshots(self, a_id: int, b_id: int) -> Optional[dict]: """Diff two snapshots by SKU.total. Returns None if either snapshot doesn't exist (router → 404).""" a = self.repo.get_consensus_detail(a_id) b = self.repo.get_consensus_detail(b_id) if not a or not b: return None def _index(rows) -> dict[str, dict]: if not isinstance(rows, list): return {} out: dict[str, dict] = {} for r in rows: if isinstance(r, dict) and r.get("sku"): out[str(r["sku"])] = r return out a_idx = _index(a.get("rows")) b_idx = _index(b.get("rows")) all_skus = sorted(set(a_idx) | set(b_idx)) def _total(r: Optional[dict]) -> Optional[float]: if not r: return None for k in ("total", "qty_total", "grand_total"): if k in r and r[k] is not None: try: return float(r[k]) except Exception: pass return None diff_rows = [] n_added = n_removed = n_changed = n_unchanged = 0 for sku in all_skus: ar = a_idx.get(sku) br = b_idx.get(sku) at = _total(ar) bt = _total(br) name = (br or ar or {}).get("name") if ar is None: change = "added" n_added += 1 delta = bt delta_pct = None elif br is None: change = "removed" n_removed += 1 delta = -(at or 0) if at is not None else None delta_pct = None else: if at == bt: change = "unchanged" n_unchanged += 1 else: change = "changed" n_changed += 1 delta = ((bt or 0) - (at or 0)) if (at is not None or bt is not None) else None delta_pct = ( (delta / at * 100) if (delta is not None and at not in (None, 0)) else None ) diff_rows.append({ "sku": sku, "name": name, "a_total": at, "b_total": bt, "delta": delta, "delta_pct": delta_pct, "change": change, }) return { "a": self._serialize_summary(a), "b": self._serialize_summary(b), "a_total_rev": float(a.get("total_rev") or 0), "b_total_rev": float(b.get("total_rev") or 0), "delta_rev": float((b.get("total_rev") or 0) - (a.get("total_rev") or 0)), "n_added": n_added, "n_removed": n_removed, "n_changed": n_changed, "n_unchanged": n_unchanged, "rows": diff_rows, } @staticmethod def _serialize_summary(row: dict) -> dict: return { "id": int(row["id"]), "label": row.get("label"), "cycle_id": row.get("cycle_id"), "cycle_year_week": row.get("cycle_year_week"), "n_skus": row.get("n_skus"), "total_rev": float(row["total_rev"]) if row.get("total_rev") is not None else None, "created_at": row["created_at"].isoformat() if row.get("created_at") else "", } # ------------------------------------------------------------------ # S&OP meeting — agenda + exceptions + KPIs assembly # ------------------------------------------------------------------ def get_sop_meeting_data(self) -> dict: latest_yw = self.repo.get_latest_data_week() # Exceptions low_cov_raw = self.repo.get_low_coverage_skus(limit=20) low_cov = [{ "sku": r["sku"], "name": r.get("name"), "tier": r.get("tier"), "category": r.get("category"), "metric": float(r.get("coverage_weeks") or 0), "metric_label": f"{float(r['coverage_weeks']):.1f} weeks" if r.get("coverage_weeks") is not None else "—", "detail": f"avg {float(r.get('last_4w_avg') or 0):.0f}/wk · " f"stock {float(r.get('stock_qty') or 0):.0f}", } for r in low_cov_raw] fa_misses_raw = self.repo.get_biggest_fa_misses(limit=10) fa_misses = [{ "sku": r["sku"], "name": r.get("name"), "tier": r.get("tier"), "year": int(r["year"]), "week": int(r["week"]), "forecast": float(r.get("forecast") or 0), "actual": float(r.get("actual") or 0), "abs_error": float(r.get("abs_error") or 0), "fa": float(r.get("fa") or 0), "bias": float(r.get("bias") or 0), } for r in fa_misses_raw] on_top_anomalies_raw = self.repo.get_on_top_anomalies(ratio_threshold=3.0) on_top_anomalies = [{ "sku": r["sku"], "name": r.get("name"), "person": r.get("person"), "channel": r.get("channel"), "year": int(r["year"]), "week": int(r["week"]), "qty": float(r.get("qty") or 0), "avg_qty_for_sku": float(r.get("avg_qty") or 0), "ratio": float(r.get("ratio") or 0), } for r in on_top_anomalies_raw] promo_overlaps_raw = self.repo.get_promo_overlaps(min_skus=5, weeks_ahead=8) promo_overlaps = [{ "year": int(r["year"]), "week": int(r["week"]), "year_week": int(r["year"]) * 100 + int(r["week"]), "cw_label": f"CW{int(r['week']):02d}", "n_skus": int(r["n_skus"]), "skus": list(r.get("skus") or [])[:10], } for r in promo_overlaps_raw] # KPIs fa_by_tier_raw = self.repo.get_fa_by_tier_latest_week() fa_by_tier = [] sum_a = 0.0 sum_f = 0.0 sum_n = 0 sum_hits = 0 for r in fa_by_tier_raw: a = float(r.get("sum_a") or 0) f = float(r.get("sum_f") or 0) n = int(r.get("n") or 0) hits = int(r.get("hits") or 0) sum_a += a; sum_f += f; sum_n += n; sum_hits += hits if a <= 0: continue fa_by_tier.append({ "tier": r["tier"], "fa": max(0.0, 1 - abs(f - a) / a) * 100, "fa_signed": (f / a) * 100, "bias": ((f - a) / a) * 100, "hit_rate": (hits / n * 100) if n else 0.0, "n_sku_weeks": n, }) overall_fa = max(0.0, 1 - abs(sum_f - sum_a) / sum_a) * 100 if sum_a > 0 else 0.0 overall_hit_rate = (sum_hits / sum_n * 100) if sum_n else 0.0 ot_status_raw = self.repo.get_on_top_status() ot_status = { "n_submitters": len(ot_status_raw), "by_person": [{ "person": r["person"], "role": r.get("role"), "qty_total": float(r.get("qty_total") or 0), "n_sku_weeks": int(r.get("n_sku_weeks") or 0), "last_submitted_at": r["last_submitted_at"].isoformat() if r.get("last_submitted_at") else None, } for r in ot_status_raw], } # Upcoming promos upcoming_raw = self.repo.get_upcoming_promos_with_volume(limit=20) upcoming = [{ "sku": r["sku"], "name": r.get("name"), "tier": r.get("tier"), "year": int(r["year"]), "week": int(r["week"]), "cw_label": f"CW{int(r['week']):02d}", "promo_types": r.get("promo_types"), "expected_volume": float(r.get("expected_volume") or 0), } for r in upcoming_raw] # Cycle info — reuse the latest forecast run's cycle as proxy. latest_run = self.repo.get_latest_forecast_run() cycle_info = None if latest_run: cycle_info = { "cycle_id": latest_run.get("cycle_id"), "cycle_year_week": latest_run.get("cycle_year_week"), "cycle_status": latest_run.get("cycle_status"), "run_id": latest_run.get("run_id"), "run_started_at": latest_run.get("started_at").isoformat() if latest_run.get("started_at") else None, "run_by": latest_run.get("run_by"), "n_skus": latest_run.get("n_skus"), } # Suggested actions — derived from what we found. suggested_actions: list[str] = [] if low_cov: suggested_actions.append( f"Review reorder for {len(low_cov)} SKU(s) with coverage < 2 weeks" ) if fa_misses: top = fa_misses[0] suggested_actions.append( f"Investigate forecast for {top['sku']} " f"(CW{top['week']:02d}: forecast {top['forecast']:.0f} vs actual {top['actual']:.0f})" ) if on_top_anomalies: suggested_actions.append( f"Verify {len(on_top_anomalies)} unusually large on-top commitments" ) if promo_overlaps: big = promo_overlaps[0] suggested_actions.append( f"Coordinate cannibalisation risk for {big['n_skus']} SKUs promoing CW{big['week']:02d}" ) if not suggested_actions: suggested_actions.append("No exceptions surfaced — cycle looks clean.") notes: list[str] = [] if cycle_info is None: notes.append( "No active S&OP cycle. Cycle metadata appears once a cycle is " "opened via the planning workflow." ) if not ot_status["by_person"]: notes.append( "On-top inputs not yet submitted for this cycle — KPI column " "will populate as VP/MP planners upload their commitments." ) return { "cycle_info": cycle_info, "latest_data_week": latest_yw, "exceptions": { "low_coverage": low_cov, "biggest_fa_misses": fa_misses, "on_top_anomalies": on_top_anomalies, "promo_overlaps": promo_overlaps, }, "kpis": { "fa_by_tier": fa_by_tier, "on_top_status": ot_status, "overall_fa": overall_fa, "overall_hit_rate": overall_hit_rate, "n_sku_weeks_scoring": sum_n, }, "upcoming_promos": upcoming, "suggested_actions": suggested_actions, "notes": notes, } # ------------------------------------------------------------------ # SKU list — unchanged # ------------------------------------------------------------------ def get_sku_list(self) -> list[dict]: return self.repo.get_sku_list() # ------------------------------------------------------------------ # SKU detail — assemble full profile from multiple sources # ------------------------------------------------------------------ def get_sku_profile(self, sku: str) -> Optional[dict]: """Returns None when the SKU doesn't exist — the router converts that to a 404. Empty downstream tables (forecasts, promo_weeks) produce empty lists, not errors.""" header = self.repo.get_sku_header(sku) if not header: return None pid = int(header["product_id"]) pricing = self.repo.get_sku_pricing(pid) stock = self.repo.get_sku_stock(pid) sales = self.repo.get_sku_sales_26w(pid) forecast = self.repo.get_sku_forecast_13w(pid) promo = self.repo.get_sku_promo_weeks(pid) # Decorate per-week rows with year_week + cw_label so the frontend # doesn't need to recompute these in 3 different components. sales_pts = [self._decorate_yw(r) | { "qty_retail": float(r.get("qty_retail") or 0), "qty_webshop": float(r.get("qty_webshop") or 0), "qty_wholesale": float(r.get("qty_wholesale") or 0), "qty_total": float(r.get("qty_total") or 0), "is_promo": bool(r.get("is_promo")), } for r in sales] forecast_pts = [self._decorate_yw(r) | { "baseline": float(r.get("baseline") or 0), "on_top_vp": float(r.get("on_top_vp") or 0), "on_top_mp": float(r.get("on_top_mp") or 0), "promo_uplift": float(r.get("promo_uplift") or 0), "planner_factor": float(r.get("planner_factor") or 1.0), "total": float(r.get("total") or 0), } for r in forecast] # 4-week average from the last 4 entries in sales_26w (or fewer if # the SKU is new). Coverage = stock_qty / weekly_avg, None when # there's no recent demand to divide by. recent = sales_pts[-4:] if sales_pts else [] avg_4w = ( sum(p["qty_total"] for p in recent) / len(recent) if recent else 0.0 ) coverage = ( float(stock.get("stock_qty") or 0) / avg_4w if avg_4w > 0 else None ) return { "sku": header["sku"], "name": header.get("name"), "category": header.get("category"), "subcategory": header.get("subcategory"), "brand": header.get("brand"), "flavor_color": header.get("flavor_color"), "size": header.get("size"), "supplier": header.get("supplier"), "family": header.get("family"), "active": bool(header.get("active") if header.get("active") is not None else True), "tier": header.get("tier"), "xyz": header.get("xyz"), "total_cv": header.get("total_cv"), "ws_xyz": header.get("ws_xyz"), "ws_cv": header.get("ws_cv"), "ws_share_26w": header.get("ws_share_26w"), "pricing": { "avg_sell_price": pricing.get("avg_sell_price"), "normal_retail_ppp": pricing.get("normal_retail_ppp"), "normal_webshop_ppp": pricing.get("normal_webshop_ppp"), "vpc": pricing.get("vpc") or header.get("vpc"), "cost_price": pricing.get("cost_price"), "cost_source": pricing.get("cost_source"), "ruc": pricing.get("ruc"), "ruc_retail": pricing.get("ruc_retail"), "ruc_wholesale": pricing.get("ruc_wholesale"), "ruc_units": pricing.get("ruc_units"), "ruc_source": pricing.get("ruc_source"), "ruc_window": pricing.get("ruc_window"), "cost_valid_from": pricing["cost_valid_from"].isoformat() if pricing.get("cost_valid_from") else None, }, "stock": { "stock_qty": float(stock.get("stock_qty") or 0), "purchase_value": float(stock.get("purchase_value") or 0), "retail_value": float(stock.get("retail_value") or 0), "minimum_total": float(stock["minimum_total"]) if stock.get("minimum_total") is not None else None, "optimum_total": float(stock["optimum_total"]) if stock.get("optimum_total") is not None else None, "n_stores": int(stock.get("n_stores") or 0), "updated_at": stock["updated_at"].isoformat() if stock.get("updated_at") else None, }, "coverage_weeks": coverage, "avg_weekly_qty_4w": avg_4w, "sales_26w": sales_pts, "forecast_13w": forecast_pts, "promo_periods": self._collapse_promo_periods(promo), } @staticmethod def _decorate_yw(r: dict) -> dict: y = int(r["year"]) w = int(r["week"]) return { "year": y, "week": w, "year_week": y * 100 + w, "cw_label": f"CW{w:02d}", } @staticmethod def _collapse_promo_periods(rows: list[dict]) -> list[dict]: """Merges contiguous ISO weeks into periods. Weeks are contiguous if the next (year, week) is the next-week step from the current one (handles year boundaries by checking week == 52 → 1 across years without trying to count ISO max weeks for a year — close enough for display, and rare edge case).""" if not rows: return [] periods: list[dict] = [] cur = None prev_yw = None for r in rows: y, w = int(r["year"]), int(r["week"]) yw = y * 100 + w is_next = ( prev_yw is not None and ( (yw - prev_yw == 1) # same year, next week or (prev_yw % 100 in (52, 53) and y - prev_yw // 100 == 1 and w == 1) ) ) if cur is None or not is_next: if cur is not None: periods.append(cur) cur = { "year_week_from": yw, "year_week_to": yw, "cw_label_from": f"CW{w:02d}", "cw_label_to": f"CW{w:02d}", "n_weeks": 1, "_types": set(), } else: cur["year_week_to"] = yw cur["cw_label_to"] = f"CW{w:02d}" cur["n_weeks"] += 1 if r.get("promo_types"): cur["_types"].add(r["promo_types"]) prev_yw = yw if cur is not None: periods.append(cur) for p in periods: types = sorted(p.pop("_types")) p["promo_types"] = ", ".join(types) if types else None return periods # ------------------------------------------------------------------ # Watchlist # ------------------------------------------------------------------ _WATCHLIST_SORTS = {"volume", "fa", "coverage", "revenue"} def get_watchlist_data( self, *, n: int = 30, sort_by: str = "volume", ) -> dict: sort_by = sort_by if sort_by in self._WATCHLIST_SORTS else "volume" rows = self.repo.get_watchlist(n=n, sort_by=sort_by) n_total = self.repo.count_watchlist_candidates() out: list[dict] = [] for i, r in enumerate(rows, start=1): stock = float(r.get("stock_qty") or 0) cov = r.get("coverage_weeks") avg4 = float(r.get("last_4w_avg") or 0) fa = r.get("fa_last_4w") low_coverage = (cov is not None and cov < 2.0 and avg4 > 0) low_fa = (fa is not None and fa < 50.0) stockout = (stock <= 0 and avg4 > 0) out.append({ "rank": i, "sku": r["sku"], "name": r.get("name"), "category": r.get("category"), "tier": r.get("tier"), "xyz": r.get("xyz"), "last_4w_avg": avg4, "forecast_next_4w": float(r.get("forecast_next_4w") or 0), "fa_last_4w": float(fa) if fa is not None else None, "stock_qty": stock, "coverage_weeks": float(cov) if cov is not None else None, "revenue_last_4w": float(r.get("revenue_last_4w") or 0), "low_coverage": low_coverage, "low_fa": low_fa, "stockout_risk": stockout, }) note: Optional[str] = None # Forecasts being empty + FA being NULL is normal in dev — surface it. if all(r["forecast_next_4w"] == 0 for r in out) and out: note = ( "`forecast_next_4w` is 0 across all rows — the `forecasts` " "table is empty. Volume / FA / coverage rankings work without " "it; the column lights up once the forecast pipeline writes " "rows (prompt 4A)." ) return { "sort_by": sort_by, "rows": out, "n_total_candidates": n_total, "note": note, } # ── KAM / CM input ──────────────────────────────────────────────────── @staticmethod def _horizon_yws(n: int = 13) -> tuple[list[tuple[int, int]], list[str], list[int]]: """Compute the forecast horizon starting from next week. Returns ([(year, week), ...], ['CW21', ...], [202621, ...]). Handles year-boundary wrap when week > 52. """ from datetime import datetime iso = datetime.now().isocalendar() cur_year, cur_week = int(iso[0]), int(iso[1]) yws, labels, yw_ints = [], [], [] for i in range(1, n + 1): w = cur_week + i y = cur_year if w > 52: w -= 52 y += 1 yws.append((y, w)) labels.append(f"CW{w}") yw_ints.append(y * 100 + w) return yws, labels, yw_ints def get_kam_template(self, user_id: int) -> dict: import json as _json user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") role = (user["role"] or "").upper() channel = user["channel"] or "" categories = user["categories"] # list[str] or "ALL" or None buyers = user["buyers"] # list[str] or None # Horizon horizon_yws, horizon_labels, horizon_yw_ints = self._horizon_yws(13) # Baselines from latest forecast run baselines = self.repo.get_latest_baselines(horizon_yws) # Current active cycle cycle = self.repo.get_active_cycle() cycle_id = cycle["id"] if cycle else None # Previous submissions for this user prev = {} if cycle_id: prev = self.repo.get_previous_inputs(cycle_id, user_id) # SKU list — VP sees all; CM sees category-filtered subset if role == "VP" or categories == "ALL" or categories is None and role == "VP": cat_filter = None else: cat_filter = categories if isinstance(categories, list) else None skus = self.repo.get_planning_skus(cat_filter) # Build buyer groups if role == "VP" and isinstance(buyers, list) and buyers: buyer_list = buyers else: buyer_list = [None] groups = [] for buyer in buyer_list: buyer_key = buyer or "" rows_out = [] for s in skus: pid = s["id"] prev_for_buyer = prev.get(pid, {}).get(buyer_key, {}) base = baselines.get(pid, {}) rows_out.append({ "sku": s["sku"], "product_id": pid, "name": s["name"], "tier": s["tier"], "category": s["cat"], "buyer": buyer, "baseline": {lbl: base.get(lbl, 0.0) for lbl in horizon_labels}, "weeks": {lbl: prev_for_buyer.get(lbl, 0.0) for lbl in horizon_labels}, }) groups.append({"buyer": buyer, "rows": rows_out}) # Deadline from kam_cm_config.json deadline = "Monday 17:00" try: import json, pathlib cfg_path = pathlib.Path("data/kam_cm_config.json") if cfg_path.exists(): cfg = json.loads(cfg_path.read_text(encoding="utf-8")) sc = cfg.get("slack_config", {}) deadline = f"{sc.get('deadline_day','Monday')} {sc.get('deadline_hour',17):02d}:00" except Exception: pass return { "user_id": user_id, "user_name": user["display_name"], "role": role, "channel": channel, "cycle_id": cycle_id, "horizon": horizon_labels, "horizon_year_weeks": horizon_yw_ints, "groups": groups, "deadline": deadline, } def save_kam_inputs(self, user_id: int, inputs: list[dict]) -> dict: user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") channel = user["channel"] or "wholesale" role = (user["role"] or "").upper() # Validate: CM users can only submit for their categories warnings = [] if role == "MP" and isinstance(user["categories"], list): allowed_cats = set(user["categories"]) all_skus = {s["id"]: s for s in self.repo.get_planning_skus(None)} filtered = [] for item in inputs: sku_info = all_skus.get(item["product_id"]) if sku_info and sku_info["cat"] not in allowed_cats: warnings.append( f"SKU {sku_info['sku']} not in assigned categories — skipped" ) continue filtered.append(item) inputs = filtered # Get or use existing cycle cycle = self.repo.get_active_cycle() if not cycle: raise ValueError("No active SOP cycle found — ask the demand planner to open one") cycle_id = cycle["id"] result = self.repo.save_on_top_inputs(user_id, cycle_id, channel, inputs) result["warnings"] = warnings return result # ────────────────────────────────────────────────────────────────── # New audit-aware save (replaces the old full-replace path for the # 4-step wizard). # ────────────────────────────────────────────────────────────────── def save_kam_inputs_v2( self, *, user_id: int, buyer: Optional[str], inputs: list[dict], # [{product_id, year_week, qty}] acting_as_id: Optional[int] = None, ) -> dict: """Diff-and-apply save with audit + lock enforcement. `user_id` is the identity the rows belong to (whoever's on-top this is). `acting_as_id`, when set and != user_id, marks an admin acting on behalf of someone else — that admin's id is what gets stamped in on_top_changes.changed_by_id, with the represented user landing in acting_as_id.""" user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") channel = user["channel"] or "wholesale" role = (user["role"] or "").upper() warnings: list[str] = [] # CM users: filter to their assigned categories (mirrors old flow) if role == "MP" and isinstance(user.get("categories"), list): allowed_cats = set(user["categories"]) sku_meta = {s["id"]: s for s in self.repo.get_planning_skus(None)} filtered = [] for item in inputs: meta = sku_meta.get(item["product_id"]) if meta and meta["cat"] not in allowed_cats: warnings.append( f"SKU {meta['sku']} not in assigned categories — skipped" ) continue filtered.append(item) inputs = filtered cycle = self.repo.get_active_cycle() if not cycle: raise ValueError("No active SOP cycle — ask the demand planner to open one") cycle_id = cycle["id"] # Lock cutoff = current_yw + 3 (first 4 weeks: cur, cur+1, cur+2, cur+3) cur_y, cur_w = self.repo.get_current_iso_week() cur_yw = cur_y * 100 + cur_w # Walk 3 weeks forward to find the inclusive cutoff yw lock_cutoff_yw = cur_yw for _ in range(3): cur_w += 1 if cur_w > 52: cur_w = 1 cur_y += 1 lock_cutoff_yw = cur_y * 100 + cur_w result = self.repo.save_on_top_inputs_audited( user_id=user_id, cycle_id=cycle_id, channel=channel, buyer=buyer, inputs=inputs, acting_as_id=acting_as_id, lock_cutoff_yw=lock_cutoff_yw, ) # Merge warnings (category filter + lock-window rejects) result["warnings"] = warnings + result.get("warnings", []) result["lock_cutoff_yw"] = lock_cutoff_yw return result # ────────────────────────────────────────────────────────────────── def list_kam_cm_users(self, *, requester_id: int, requester_is_admin: bool) -> list[dict]: """Step-1 identity dropdown. Admin sees all VP/MP users; non-admin sees only themselves. """ if requester_is_admin: return self.repo.list_input_users() u = self.repo.get_user(requester_id) # Normalize the Croatian role names to the internal VP/MP codes the # frontend keys off — otherwise a 'Veleprodaja'/'Maloprodaja' user gets # an empty identity dropdown and can't select themselves. norm = {"VELEPRODAJA": "VP", "MALOPRODAJA": "MP", "VP": "VP", "MP": "MP"}.get((u.get("role") or "").upper()) if u else None if not u or norm not in ("VP", "MP"): return [] return [{ "user_id": u["id"], "display_name": u["display_name"], "role": norm, "channel": u.get("channel"), }] def list_user_buyers(self, user_id: int) -> list[str]: """Step-2 buyer dropdown. Returns DISTINCT buyer names from this user's past on_top_inputs (the ones they themselves inserted).""" return self.repo.list_user_buyers(user_id) # ────────────────────────────────────────────────────────────────── # Wholesale buyer-listings input — KAM enters reg-increase / on-top for # exactly the SKUs LISTED for the selected buyer; Save recomputes the # wholesale forecast immediately. # ────────────────────────────────────────────────────────────────── def list_wholesale_input_buyers(self, user_id: int) -> list[str]: """Buyers with a listed assortment under this KAM (from wholesale_listings, keyed on the KAM's display_name).""" user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") return self.repo.list_kam_listing_buyers(user["display_name"]) def get_wholesale_input_template(self, user_id: int, buyer: str) -> dict: """Grid for (KAM, buyer): one row per SKU listed for that buyer, with the existing regular-increase and on-top portions pre-filled per week so the frontend selector can switch between them without a refetch.""" user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") kam_name = user["display_name"] horizon_yws, horizon_labels, horizon_yw_ints = self._horizon_yws(13) baselines = self.repo.get_latest_baselines(horizon_yws) cycle = self.repo.get_active_cycle() cycle_id = cycle["id"] if cycle else None listing_rows = self.repo.get_wholesale_listing_rows(kam_name, buyer) # Cycle-agnostic: shows the live plan even when it sits in a prior cycle. splits = self.repo.get_wholesale_existing_splits(user_id=user_id, buyer=buyer) rows = [] for L in listing_rows: pid = L["product_id"] base = baselines.get(pid, {}) reg: dict = {} on_top: dict = {} for lbl, yw in zip(horizon_labels, horizon_yw_ints): s = splits.get((pid, yw)) if s: reg[lbl] = round(s["reg"], 2) on_top[lbl] = round(max(0.0, s["quantity"] - s["reg"]), 2) else: reg[lbl] = 0.0 on_top[lbl] = 0.0 rows.append({ "sku": L["sku"], "product_id": pid, "name": L["name"], "tier": L["tier"], "rank": L["rank"], "category": L["category"] or "", "baseline": {lbl: base.get(lbl, 0.0) for lbl in horizon_labels}, "reg": reg, "on_top": on_top, }) deadline = "Monday 17:00" try: import json, pathlib cfg_path = pathlib.Path("data/kam_cm_config.json") if cfg_path.exists(): cfg = json.loads(cfg_path.read_text(encoding="utf-8")) sc = cfg.get("slack_config", {}) deadline = f"{sc.get('deadline_day','Monday')} {sc.get('deadline_hour',17):02d}:00" except Exception: pass return { "user_id": user_id, "user_name": user["display_name"], "kam": kam_name, "buyer": buyer, "cycle_id": cycle_id, "horizon": horizon_labels, "horizon_year_weeks": horizon_yw_ints, "rows": rows, "deadline": deadline, } def upload_buyer_listings( self, *, kam: str, buyer: str, file_bytes: bytes, filename: str, dry_run: bool = False, ) -> dict: """Parse an uploaded Excel of a buyer's listed assortment and (unless dry_run) replace that (kam, buyer)'s rows in wholesale_listings. Format-tolerant: scans every sheet/column, picks the column with the most SKU-shaped codes (e.g. POL04229) as the SKU column and any column whose values look like gold/silver/bronze as the rank. Returns a preview summary (parsed / matched / unmatched / ranked).""" import io, re import pandas as pd kam = (kam or "").strip() buyer = (buyer or "").strip() if not kam or not buyer: raise ValueError("Both KAM and buyer are required.") SKU_RE = re.compile(r"^[A-Za-z]{2,5}\d{3,}$") try: xls = pd.ExcelFile(io.BytesIO(file_bytes)) except Exception as e: raise ValueError(f"Could not read the Excel file: {e}") best_rows: list[dict] = [] for sheet in xls.sheet_names: try: df = pd.read_excel(xls, sheet_name=sheet, header=None, dtype=str) except Exception: continue if df.empty: continue # SKU column = the one with the most SKU-shaped values. sku_col, best_cnt = None, 0 for col in df.columns: vals = df[col].dropna().astype(str).str.strip() cnt = int(sum(1 for v in vals if SKU_RE.match(v))) if cnt > best_cnt: best_cnt, sku_col = cnt, col if sku_col is None or best_cnt == 0: continue # Rank column = values that look like gold/silver/bronze. rank_col = None for col in df.columns: if col == sku_col: continue vals = df[col].dropna().astype(str).str.strip().str.lower() hits = int(sum(1 for v in vals if v in ("gold", "silver", "bronze"))) if hits >= max(2, int(best_cnt * 0.25)): rank_col = col break rows: list[dict] = [] for _, rr in df.iterrows(): sv = rr[sku_col] sku = str(sv).strip() if pd.notna(sv) else "" if not SKU_RE.match(sku): continue rank = None if rank_col is not None and pd.notna(rr[rank_col]): rv = str(rr[rank_col]).strip().lower() rank = rv if rv in ("gold", "silver", "bronze") else None rows.append({"sku": sku.upper(), "rank": rank}) if len(rows) > len(best_rows): best_rows = rows # Dedupe by SKU (first occurrence wins). seen: dict[str, dict] = {} for r in best_rows: seen.setdefault(r["sku"], r) parsed = list(seen.values()) if not parsed: raise ValueError("No SKU-shaped codes found (expected codes like POL04229, CEL11615).") pid_map = self.repo.match_skus_to_products([r["sku"] for r in parsed]) for r in parsed: r["product_id"] = pid_map.get(r["sku"]) unmatched = [r["sku"] for r in parsed if r["product_id"] is None] n_ranked = sum(1 for r in parsed if r["rank"]) committed = False if not dry_run: self.repo.replace_buyer_listings( kam=kam, buyer=buyer, rows=parsed, source_file=filename or "upload.xlsx", ) committed = True return { "kam": kam, "buyer": buyer, "n_parsed": len(parsed), "n_matched": len(parsed) - len(unmatched), "n_unmatched": len(unmatched), "unmatched_skus": unmatched[:50], "n_ranked": n_ranked, "committed": committed, } # ────────────────────────────────────────────────────────────────── # Weekly KAM/CM input upload — full 13-week snapshot per buyer sheet. # ────────────────────────────────────────────────────────────────── def parse_kam_upload(self, *, kam_user_id: int, file_bytes: bytes) -> dict: """Parse a weekly KAM/CM template and return the detected sheets with a suggested canonical buyer for each (for the mapping screen). No writes.""" user = self.repo.get_user(kam_user_id) if not user: raise ValueError(f"User {kam_user_id} not found") parsed = _parse_kam_workbook(file_bytes) file_type = parsed["type"] or ((user.get("role") or "").upper() if (user.get("role") or "").upper() in ("VP", "MP") else None) existing = self.repo.list_kam_listing_buyers(user["display_name"]) if file_type == "VP" else [] existing_lc = {b.lower().strip(): b for b in existing} def _suggest(label: str) -> Optional[str]: lc = (label or "").strip().lower() if lc in existing_lc: return existing_lc[lc] # startswith / contains either direction for k, v in existing_lc.items(): if k.startswith(lc) or lc.startswith(k) or lc in k or k in lc: return v return None sheets = [] for s in parsed["sheets"]: sug = _suggest(s["buyer_label"]) if file_type == "VP" else None sheets.append({ "sheet": s["sheet"], "buyer_label": s["buyer_label"], "n_skus": len(s["skus"]), "suggested_buyer": sug, "is_new": file_type == "VP" and sug is None, }) return { "kam_name": parsed["kam_name"], "file_type": file_type, "selected_kam": user["display_name"], "kam_match": (parsed["kam_name"] or "").strip().lower() == user["display_name"].strip().lower() if parsed["kam_name"] else True, "cw_labels": parsed["cw_labels"], "sheets": sheets, "existing_buyers": existing, } def apply_kam_upload( self, *, kam_user_id: int, file_bytes: bytes, mapping: dict[str, str], seed_listings: Optional[list[str]] = None, acting_as_id: Optional[int] = None, dry_run: bool = False, ) -> dict: """Apply a weekly template: per buyer sheet, validate → diff vs current app state → write deltas under the 4-week lock → recompute the forecast. `mapping` = {sheet_name: canonical_buyer}; `seed_listings` = sheet names whose listing should be seeded from the sheet's SKUs (new buyers).""" from backend.services.wholesale_review_service import ( _recompute_ws_forecast, _recompute_retail_forecast, ) user = self.repo.get_user(kam_user_id) if not user: raise ValueError(f"User {kam_user_id} not found") parsed = _parse_kam_workbook(file_bytes) role = (user.get("role") or "").upper() file_type = parsed["type"] or (role if role in ("VP", "MP") else "VP") channel = "wholesale" if file_type == "VP" else "retail" seed_set = set(seed_listings or []) cycle = self.repo.get_active_cycle() if not cycle: raise ValueError("No active SOP cycle — ask the demand planner to open one") cycle_id = cycle["id"] cur_y, cur_w = self.repo.get_current_iso_week() lock_yws = _lock_year_weeks(cur_y, cur_w) # next 4 weeks (cur+1..cur+4) cw_map = _cw_to_year_week_map([c for c in parsed["cw_labels"]], cur_y, cur_w) # SKU→category map for MP validation. cat_map: dict[str, str] = {} allowed_cats = None if file_type == "MP": cat_map = {s["sku"]: (s["cat"] or "") for s in self.repo.get_planning_skus(None)} cats = user.get("categories") allowed_cats = set(cats) if isinstance(cats, list) else None # None ⇒ all results = [] affected_ws: set[tuple[int, int]] = set() affected_mp: set[tuple[int, int]] = set() for sheet in parsed["sheets"]: sname = sheet["sheet"] if file_type == "VP": buyer = (mapping.get(sname) or sheet["buyer_label"] or "").strip() if not buyer: continue # Seed a brand-new buyer's listing from this sheet if asked. if sname in seed_set and not dry_run: seed_skus = [r["sku"] for r in sheet["skus"]] pidm = self.repo.match_skus_to_products(seed_skus) self.repo.replace_buyer_listings( kam=user["display_name"], buyer=buyer, rows=[{"sku": s, "rank": None, "product_id": pidm.get(s)} for s in seed_skus], source_file=f"seed:{sname}", ) listed = {r["sku"] for r in self.repo.get_wholesale_listing_rows(user["display_name"], buyer)} else: buyer = None listed = None # Build file_state for valid SKUs; collect unmatched. skus = [r["sku"] for r in sheet["skus"]] pid_map = self.repo.match_skus_to_products(skus) file_state: dict[tuple[int, int], dict] = {} unmatched: list[str] = [] horizon = sorted(set(cw_map.values())) for r in sheet["skus"]: sku = r["sku"] # Collect this SKU's non-zero cells. A blank row carries no # signal — ignore it entirely (don't flag it as unmatched). cells: dict[int, dict] = {} for cw, yw in cw_map.items(): on_top = float(r["on_top"].get(cw, 0) or 0) reg = float(r["reg"].get(cw, 0) or 0) total = on_top + reg if total > 0: cells[yw] = {"total": total, "reg": reg} if not cells: continue # Validation (E): VP must be listed; MP must be in allowed cats. pid = pid_map.get(sku) if file_type == "VP": ok = pid is not None and sku in listed else: ok = pid is not None and (allowed_cats is None or cat_map.get(sku, "") in allowed_cats) if not ok: unmatched.append(sku) continue for yw, val in cells.items(): file_state[(pid, yw)] = val res = self.repo.apply_upload_snapshot( user_id=kam_user_id, cycle_id=cycle_id, buyer=buyer, channel=channel, horizon_yws=horizon, file_state=file_state, lock_yws=lock_yws, acting_as_id=acting_as_id, dry_run=dry_run, ) (affected_ws if channel == "wholesale" else affected_mp).update( {tuple(x) for x in res["affected"]} ) results.append({ "sheet": sname, "buyer": buyer or "(retail / no buyer)", "added": res["added"], "removed": res["removed"], "changed": res["changed"], "unchanged": res["unchanged"], "skipped_locked": res["skipped_locked"], "unmatched_skus": unmatched[:50], "n_unmatched": len(unmatched), }) if not dry_run: for (pid, yw) in affected_ws: _recompute_ws_forecast(self.repo.db, int(pid), int(yw)) for (pid, yw) in affected_mp: _recompute_retail_forecast(self.repo.db, int(pid), int(yw)) self.repo.db.commit() tot = lambda k: sum(r[k] for r in results) # noqa: E731 return { "file_type": file_type, "kam": user["display_name"], "dry_run": dry_run, "sheets": results, "totals": { "added": tot("added"), "removed": tot("removed"), "changed": tot("changed"), "unchanged": tot("unchanged"), "skipped_locked": sum(len(r["skipped_locked"]) for r in results), "unmatched": sum(r["n_unmatched"] for r in results), }, "recomputed_cells": len(affected_ws) + len(affected_mp), } def save_wholesale_input( self, *, user_id: int, buyer: str, portion: str, inputs: list[dict], acting_as_id: Optional[int] = None, ) -> dict: """Save one portion (regular-increase or on-top) of the buyer-listings grid, preserving the other portion, then recompute the affected wholesale forecast cells so the change lands in the forecast at once.""" if portion not in ("reg", "on_top"): raise ValueError("portion must be 'reg' or 'on_top'") user = self.repo.get_user(user_id) if not user: raise ValueError(f"User {user_id} not found") cycle = self.repo.get_active_cycle() if not cycle: raise ValueError("No active SOP cycle — ask the demand planner to open one") cycle_id = cycle["id"] result = self.repo.save_wholesale_portion_inputs( user_id=user_id, cycle_id=cycle_id, buyer=buyer, portion=portion, inputs=inputs, acting_as_id=acting_as_id, ) # Fold the edits straight into the latest forecast run (on_top_wholesale # + total) for each touched (product, week), mirroring the wholesale # cell-editor. Done before commit so save + recompute are atomic. from backend.services.wholesale_review_service import _recompute_ws_forecast for (pid, yw) in result.get("affected", []): _recompute_ws_forecast(self.repo.db, int(pid), int(yw)) self.repo.db.commit() result["recomputed_cells"] = len(result.get("affected", [])) return result def get_on_top_changes( self, *, user_id: Optional[int] = None, from_yw: Optional[int] = None, to_yw: Optional[int] = None, weeks_back: Optional[int] = None, limit: int = 500, ) -> list[dict]: """Change-control feed. Filter by submitter, on-top week range, and/or recency (weeks_back). Delete+insert pairs from the same save are collapsed into a single 'move' event so a week move reads as one line.""" rows = self.repo.get_on_top_changes( user_id=user_id, from_yw=from_yw, to_yw=to_yw, weeks_back=weeks_back, limit=limit, ) return self._collapse_moves(rows) @staticmethod def _collapse_moves(rows: list[dict]) -> list[dict]: """Pair a delete + insert written in the SAME save (identical changed_at, submitter, product, buyer, channel) into one 'move' row: from_week→to_week with the qty that moved. Anything more complex than a clean 1-delete/1-insert pair is left as separate rows. Postgres now() is the transaction timestamp, so every audit row from a single save_on_top_inputs_audited() call shares an exact changed_at — that's what lets us group a move's two halves reliably.""" from collections import defaultdict groups: dict[tuple, list[dict]] = defaultdict(list) for r in rows: key = ( r.get("changed_at"), r.get("changed_by_id"), r.get("product_id"), (r.get("buyer") or ""), (r.get("channel") or ""), ) groups[key].append(r) rep_to_move: dict[int, dict] = {} # insert row id -> synthetic move row drop_ids: set[int] = set() # delete row ids folded into a move for grp in groups.values(): dels = [r for r in grp if r["change_type"] == "delete"] inss = [r for r in grp if r["change_type"] == "insert"] others = [r for r in grp if r["change_type"] not in ("delete", "insert")] if (len(dels) == 1 and len(inss) == 1 and not others and dels[0]["year_week"] != inss[0]["year_week"]): d, i = dels[0], inss[0] move = dict(i) move["change_type"] = "move" move["from_week"] = d["year_week"] move["to_week"] = i["year_week"] move["year_week"] = i["year_week"] # destination is the headline week move["old_qty"] = d["old_qty"] # qty that left the old week move["new_qty"] = i["new_qty"] # qty now sitting in the new week rep_to_move[i["id"]] = move drop_ids.add(d["id"]) out: list[dict] = [] for r in rows: if r["id"] in rep_to_move: out.append(rep_to_move[r["id"]]) elif r["id"] in drop_ids: continue else: r.setdefault("from_week", None) r.setdefault("to_week", None) out.append(r) return out def get_kam_buyer_fa( self, *, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, category: Optional[list[str]] = None, date_from: Optional[int] = None, date_to: Optional[int] = None, ) -> dict: """Buyer-level FA for VP/KAM wholesale on-top commitments. Compares on_top_inputs (wholesale channel, per buyer) against ERP actuals (erp_transactions). Only covers wholesale — CM retail has no buyer-level tracking. Rows with unmatched buyer names (no ERP partner found via ILIKE) are excluded from FA but surfaced in `unmatched_buyers`. """ raw = self.repo.get_buyer_fa_raw( tier=tier, xyz=xyz, category=category, date_from=date_from, date_to=date_to, ) if not raw: return { "headline": self._empty_headline(), "by_person": [], "unmatched_buyers": [], "note": ( "No wholesale on-top inputs found. VP/KAM users need to " "submit on-top quantities for the S&OP cycle before buyer " "FA can be computed." ), } # Split into matched (partner resolved) vs unmatched. matched_raw: list[dict] = [] unmatched_buyers: set[str] = set() for r in raw: if r.get("partner_name") is None: unmatched_buyers.add(r.get("buyer") or "?") else: matched_raw.append(r) # _enrich_one drops rows where actual <= 0 (unmatched → actual None). enriched: list[dict] = [] for r in matched_raw: e = self._enrich_one(r) if e is None: continue e["person"] = r.get("person") or "(unassigned)" e["role"] = r.get("role") e["buyer"] = r.get("buyer") or "?" e["partner_name"] = r.get("partner_name") enriched.append(e) if not enriched: return { "headline": self._empty_headline(), "by_person": [], "unmatched_buyers": sorted(unmatched_buyers), "note": ( "No comparable ERP actuals found for the submitted weeks. " "Buyer FA compares on-top inputs against erp_transactions " "for the same (product, year, week) — check that ERP data " "covers the on-top horizon." ), } # Headline across all matched enriched rows. h = self._per_week_avg_metrics(enriched) hit_rate = sum(1 for r in enriched if r["hit"]) / len(enriched) * 100 unique_weeks = {(r["year"], r["week"]) for r in enriched} unique_skus = {r["sku"] for r in enriched} # Build: {person → {buyer → [rows]}} by_pb: dict[str, dict[str, list[dict]]] = defaultdict( lambda: defaultdict(list) ) person_roles: dict[str, Optional[str]] = {} partner_labels: dict[str, str] = {} # buyer → partner_name for r in enriched: by_pb[r["person"]][r["buyer"]].append(r) person_roles.setdefault(r["person"], r.get("role")) partner_labels[r["buyer"]] = r.get("partner_name") or r["buyer"] by_person: list[dict] = [] for person, buyers_map in by_pb.items(): all_rows = [r for rows in buyers_map.values() for r in rows] pm = self._per_week_avg_metrics(all_rows) p_hit = sum(1 for r in all_rows if r["hit"]) / len(all_rows) * 100 by_buyer: list[dict] = [] for buyer, b_rows in buyers_map.items(): bm = self._per_week_avg_metrics(b_rows) b_hit = sum(1 for r in b_rows if r["hit"]) / len(b_rows) * 100 # Per-SKU rollup inside each buyer for drill-down. by_sku_map: dict[str, list[dict]] = defaultdict(list) for r in b_rows: by_sku_map[r["sku"]].append(r) by_sku: list[dict] = [] for sku, s_rows in by_sku_map.items(): sm = self._per_week_avg_metrics(s_rows) s_hit = sum(1 for r in s_rows if r["hit"]) / len(s_rows) * 100 by_sku.append({ "sku": sku, "name": s_rows[0].get("name"), "n_weeks": len({(r["year"], r["week"]) for r in s_rows}), "forecast": sum(r["forecast"] for r in s_rows), "actual": sum(r["actual"] for r in s_rows), "fa": sm["fa"], "fa_signed": sm["fa_signed"], "bias": sm["bias"], "hit_rate": s_hit, }) by_sku.sort(key=lambda d: d["fa"]) # worst first by_buyer.append({ "buyer": buyer, "partner_name": partner_labels.get(buyer), "n_sku_weeks": len(b_rows), "forecast": sum(r["forecast"] for r in b_rows), "actual": sum(r["actual"] for r in b_rows), "fa": bm["fa"], "fa_signed": bm["fa_signed"], "bias": bm["bias"], "hit_rate": b_hit, "by_sku": by_sku, }) by_buyer.sort(key=lambda d: d["fa"], reverse=True) by_person.append({ "person": person, "role": person_roles.get(person), "n_sku_weeks": len(all_rows), "forecast": sum(r["forecast"] for r in all_rows), "actual": sum(r["actual"] for r in all_rows), "fa": pm["fa"], "fa_signed": pm["fa_signed"], "bias": pm["bias"], "hit_rate": p_hit, "by_buyer": by_buyer, }) by_person.sort(key=lambda d: d["fa"], reverse=True) return { "headline": { "fa": h["fa"], "fa_signed": h["fa_signed"], "bias": h["bias"], "hit_rate": hit_rate, "n_sku_weeks": len(enriched), "n_skus": len(unique_skus), "n_weeks": len(unique_weeks), }, "by_person": by_person, "unmatched_buyers": sorted(unmatched_buyers), "note": None, } def get_submission_overview(self, cycle_id: Optional[int] = None) -> dict: if cycle_id is None: cycle = self.repo.get_active_cycle() cycle_id = cycle["id"] if cycle else None rows = self.repo.get_submission_status(cycle_id) deadline = "Monday 17:00" try: import json, pathlib cfg_path = pathlib.Path("data/kam_cm_config.json") if cfg_path.exists(): cfg = json.loads(cfg_path.read_text(encoding="utf-8")) sc = cfg.get("slack_config", {}) deadline = f"{sc.get('deadline_day','Monday')} {sc.get('deadline_hour',17):02d}:00" except Exception: pass cycle_week = None if cycle_id: cycle = self.repo.get_active_cycle() if cycle: yw = cycle["year_week"] cycle_week = int(str(yw)[-2:]) if yw else None return { "cycle_id": cycle_id, "cycle_week": cycle_week, "deadline": deadline, "users": rows, } # ------------------------------------------------------------------ # Per-SKU weekly sales — last 13 ISO weeks, channel split + promo flag # + wholesale-by-buyer breakdown (for the rewound Sales Weekly page). # ------------------------------------------------------------------ def get_sku_weekly_sales(self, sku: str, n_weeks: int = 13) -> dict: from datetime import date, timedelta db = self.repo.db prod = db.execute(text( "SELECT id, COALESCE(name, sku) AS name FROM dim_products WHERE sku = :s" ), {"s": sku}).mappings().first() if not prod: return {"sku": sku, "name": None, "found": False, "weeks": []} pid = int(prod["id"]) # Build the last n_weeks ISO weeks ending at the current week. today = date.today() monday = today - timedelta(days=today.weekday()) week_keys: list[tuple[int, int]] = [] cur = monday for _ in range(n_weeks): iso = cur.isocalendar() week_keys.append((int(iso[0]), int(iso[1]))) cur -= timedelta(days=7) week_keys.reverse() # oldest → newest yws = [y * 100 + w for (y, w) in week_keys] start = (monday - timedelta(weeks=n_weeks - 1)).isoformat() # Channel qty per ISO week chan = db.execute(text(""" SELECT EXTRACT(ISOYEAR FROM et.transaction_date)::int AS y, EXTRACT(WEEK FROM et.transaction_date)::int AS w, cm.channel AS channel, SUM(et.quantity)::float AS qty FROM erp_transactions et LEFT JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE et.product_id = :pid AND et.transaction_date >= DATE :start GROUP BY y, w, cm.channel """), {"pid": pid, "start": start}).mappings().all() chan_map: dict[tuple[int, int], dict[str, float]] = {} for r in chan: key = (int(r["y"]), int(r["w"])) slot = chan_map.setdefault(key, {"retail": 0.0, "webshop": 0.0, "wholesale": 0.0, "other": 0.0}) ch = r["channel"] if r["channel"] in ("retail", "webshop", "wholesale") else "other" slot[ch] += float(r["qty"] or 0) # Promo flag + campaign label per week promo = db.execute(text(""" SELECT year, week, bool_or(is_erp_promo) AS on_promo, STRING_AGG(DISTINCT promo_types, ', ') AS types FROM erp_promo_weeks WHERE product_id = :pid AND (year * 100 + week) = ANY(:yws) GROUP BY year, week """), {"pid": pid, "yws": yws}).mappings().all() promo_map = {(int(r["year"]), int(r["week"])): (bool(r["on_promo"]), r["types"]) for r in promo} # Wholesale qty by buyer per week (for the hover breakdown) wb = db.execute(text(""" SELECT EXTRACT(ISOYEAR FROM et.transaction_date)::int AS y, EXTRACT(WEEK FROM et.transaction_date)::int AS w, COALESCE(dp.name, '(unknown partner)') AS buyer, SUM(et.quantity)::float AS qty FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id AND cm.channel = 'wholesale' LEFT JOIN dim_partners dp ON dp.id = et.partner_id WHERE et.product_id = :pid AND et.transaction_date >= DATE :start GROUP BY y, w, dp.name """), {"pid": pid, "start": start}).mappings().all() wb_map: dict[tuple[int, int], list[dict]] = {} for r in wb: key = (int(r["y"]), int(r["w"])) wb_map.setdefault(key, []).append( {"buyer": r["buyer"], "qty": round(float(r["qty"] or 0), 1)}) for k in wb_map: wb_map[k].sort(key=lambda x: x["qty"], reverse=True) weeks = [] for (y, w) in week_keys: c = chan_map.get((y, w), {"retail": 0.0, "webshop": 0.0, "wholesale": 0.0, "other": 0.0}) on_promo, types = promo_map.get((y, w), (False, None)) total = c["retail"] + c["webshop"] + c["wholesale"] + c["other"] weeks.append({ "year": y, "week": w, "cw_label": f"CW{w:02d}", "qty_retail": round(c["retail"], 1), "qty_webshop": round(c["webshop"], 1), "qty_wholesale": round(c["wholesale"], 1), "qty_other": round(c["other"], 1), "qty_total": round(total, 1), "on_promo": on_promo, "promo_label": types, "wholesale_buyers": wb_map.get((y, w), []), }) return {"sku": sku, "name": prod["name"], "found": True, "weeks": weeks} # ───────────────────────────────────────────────────────────────────── # Weekly KAM/CM template parsing helpers (module-level) # ───────────────────────────────────────────────────────────────────── def _parse_qty(v) -> float: """Tolerant number parse for unit-count cells (commas = thousands).""" if v is None: return 0.0 s = str(v).strip().replace(" ", "").replace(" ", "") if not s or s.lower() == "nan": return 0.0 s = s.replace(",", "") try: return float(s) except ValueError: return 0.0 def _lock_year_weeks(cur_y: int, cur_w: int) -> set: """The next 4 ISO weeks AFTER the current one (cur+1 … cur+4) as YYYYWW.""" from datetime import date, timedelta try: mon = date.fromisocalendar(cur_y, cur_w, 1) except (ValueError, AttributeError): mon = date.today() - timedelta(days=date.today().weekday()) out = set() for i in range(1, 5): iso = (mon + timedelta(weeks=i)).isocalendar() out.add(int(iso[0]) * 100 + int(iso[1])) return out def _cw_to_year_week_map(cw_labels: list, cur_y: int, cur_w: int) -> dict: """Map 'CWnn' labels to real YYYYWW, anchored to the current week so a label like CW02 resolves to next year when appropriate. First forward occurrence wins; the window starts one week before current to allow a current-week column.""" from datetime import date, timedelta import re as _re try: start = date.fromisocalendar(cur_y, cur_w, 1) - timedelta(weeks=1) except (ValueError, AttributeError): start = date.today() - timedelta(days=date.today().weekday() + 7) wk_to_yw: dict[int, int] = {} for i in range(20): iso = (start + timedelta(weeks=i)).isocalendar() wk_to_yw.setdefault(int(iso[1]), int(iso[0]) * 100 + int(iso[1])) out: dict[str, int] = {} for lbl in cw_labels: m = _re.match(r"^CW(\d+)$", str(lbl).strip()) if not m: continue wnum = int(m.group(1)) if wnum in wk_to_yw: out[lbl] = wk_to_yw[wnum] return out def _parse_kam_workbook(file_bytes: bytes) -> dict: """Parse a DEMAND INPUT template (VP per-buyer sheets / MP single sheet). Returns {kam_name, type, cw_labels, sheets:[{sheet, buyer_label, skus:[ {sku, on_top:{cw:qty}, reg:{cw:qty}}]}]}. Two rows per SKU (Type column = 'on-top demand' / 'regular increase').""" import io, re as _re import pandas as pd SKU_RE = _re.compile(r"^[A-Za-z]{2,5}\d{3,}$") try: xls = pd.ExcelFile(io.BytesIO(file_bytes)) except Exception as e: raise ValueError(f"Could not read the Excel file: {e}") kam_name = None ftype = None if "_meta" in xls.sheet_names: m = pd.read_excel(xls, "_meta", header=None, dtype=str) md: dict[str, str] = {} for _, row in m.iterrows(): k = str(row[0]).strip().lower() if pd.notna(row[0]) else "" v = str(row[1]).strip() if len(row) > 1 and pd.notna(row[1]) else "" if k: md[k] = v kam_name = md.get("kam_name") or None t = (md.get("type") or "").upper() ftype = t if t in ("VP", "MP") else None cw_labels: list[str] = [] sheets: list[dict] = [] for sheet in xls.sheet_names: if sheet == "_meta": continue df = pd.read_excel(xls, sheet_name=sheet, header=None, dtype=str) if df.empty: continue # Locate the header row (has both 'SKU' and 'Type'). hdr = None for i in range(min(12, len(df))): vals = [str(x).strip() if pd.notna(x) else "" for x in df.iloc[i].tolist()] if "SKU" in vals and "Type" in vals: hdr = i break if ftype is None: joined = " ".join(vals).upper() if "VP (WHOLESALE)" in joined: ftype = "VP" elif "MP (" in joined or "MARKETING/RETAIL" in joined: ftype = "MP" if hdr is None: continue header = [str(x).strip() if pd.notna(x) else "" for x in df.iloc[hdr].tolist()] try: sku_col = header.index("SKU") type_col = header.index("Type") except ValueError: continue cw_cols: dict[int, str] = {} for ci, h in enumerate(header): if _re.match(r"^CW\d+$", h): cw_cols[ci] = h if h not in cw_labels: cw_labels.append(h) # Buyer label from a 'Buyer:' cell above the header, else sheet name. buyer_label = sheet for i in range(hdr): for x in df.iloc[i].tolist(): if pd.notna(x) and "Buyer:" in str(x): buyer_label = str(x).split("Buyer:")[-1].strip().lstrip(":").strip() sku_rows: dict[str, dict] = {} for i in range(hdr + 1, len(df)): row = df.iloc[i].tolist() sku = str(row[sku_col]).strip() if sku_col < len(row) and pd.notna(row[sku_col]) else "" if not SKU_RE.match(sku): continue typ = (str(row[type_col]).strip().lower() if type_col < len(row) and pd.notna(row[type_col]) else "") if "regular" in typ: target = "reg" elif "on-top" in typ or "on top" in typ or "demand" in typ: target = "on_top" else: continue slot = sku_rows.setdefault(sku.upper(), {"sku": sku.upper(), "on_top": {}, "reg": {}}) for ci, cw in cw_cols.items(): if ci < len(row): q = _parse_qty(row[ci]) if q > 0: slot[target][cw] = q sheets.append({"sheet": sheet, "buyer_label": buyer_label or sheet, "skus": list(sku_rows.values())}) return {"kam_name": kam_name, "type": ftype, "cw_labels": cw_labels, "sheets": sheets}