"""Finance / CFO module — live compute of the 5 S&OP analyses. Refactor of scripts/cfo_audit.py into an importable service. Every method returns a pandas DataFrame (sorted by the relevant € desc) plus the summary rollups the UI cards need. The router wraps these for JSON responses + Excel downloads. Each call hits the DB once for the master context, then runs the report-specific math in-memory. Cached results are intentionally NOT implemented — the data updates often (uploads, on-top edits) and the compute is sub-second once the context is loaded. """ from __future__ import annotations import io import math from datetime import date, datetime from pathlib import Path from typing import Any, Optional import pandas as pd from sqlalchemy import text from sqlalchemy.orm import Session from backend.services.coverage_classifier import classify_coverage # Constants — same as the audit script HORIZON_WEEKS = 13 CONTEST_START_CW = 27 CONTEST_END_CW = 31 # We only started running a real demand forecast in April 2026 — before # that the "plan" is a reconstructed trailing-avg proxy that doesn't # reflect any actual planning decision. Bridge results before this # cutoff are misleading, so the monthly bridge filters them out. FORECAST_START_YEAR = 2026 FORECAST_START_MONTH = 4 # Top-N by Sifrarnik DataLink that we treat as "newly listed" — these # SKUs are excluded from DEAD_STOCK / SLOW_MOVER / OVERSTOCK # classification and tagged NEW_LISTING instead, because lack of sales # in a new SKU is expected, not problematic. NEW_LISTING_TOP_N = 10_000 # Per-SKU lead time comes from supply_master.lead_time_weeks. Tier-based # fallbacks were removed (2026-05) — supply_master must be populated for every # planned SKU. When missing, _lead_time() returns None so callers can flag the # data quality issue rather than silently substituting a guess. SAFETY_WEEKS = {"01 GOLD": 2.0, "02 SILVER": 1.5, "03 BRONZE": 1.0} DEFAULT_SAFETY = 1.0 REVIEW_PERIOD_WEEKS = 1.0 GADGET_CATEGORIES = {"GADGETI", "GADGETS"} NON_FOOD_CATEGORIES = { "GADGETI", "DRINKWARE I HOME", "BORILAČKA OPREMA", "BORILACKA OPREMA", "ODJEĆA", "ODJECA", "OBUĆA", "OBUCA", "REKVIZITI", "TEKSTIL", "OPREMA", } # Locked Cash — non-food categories use a different overstock rule (lead-time # data is missing/unreliable for these supplier types, and demand is too lumpy # for a multiplier-of-LT to be meaningful). See report_locked_cash() for the # two-track logic (active overstock vs slow stock). LOCKED_CASH_NON_FOOD_CATS = { "ODJEĆA I OBUĆA", "FITNESS OPREMA", "DRINKWARE I HOME", "BORILAČKA OPREMA", "BORILACKA OPREMA", "GADGETI", "GADGETS", "REKVIZITI", "TEKSTIL", "OPREMA", } # Per-category "did it actually sell?" gate over the last 26 weeks. # Below the gate → SKU goes into the "slow stock" track (any meaningful stock # is considered locked). At/above the gate → "active overstock" track where # target = forward 26-week forecasted demand. NON_FOOD_SELL_THROUGH_X = { "ODJEĆA I OBUĆA": 10, "FITNESS OPREMA": 3, "DRINKWARE I HOME": 20, "BORILAČKA OPREMA": 3, "BORILACKA OPREMA": 3, "GADGETI": 5, "GADGETS": 5, "REKVIZITI": 5, "TEKSTIL": 10, "OPREMA": 3, } DEFAULT_NON_FOOD_X = 5 NON_FOOD_FORWARD_HORIZON_WEEKS = 26 # Slow-stock floor — anything above this much WH stock is treated as locked # when sell-through gate fails. Set low: even 2 units of an apparel SKU that # hasn't sold in 6 months is cash that won't be recovered organically. NON_FOOD_SLOW_STOCK_FLOOR = 1.0 # Minimum excess units to flag as locked cash. Anything below this is noise # — not worth a buyer's attention even if the math technically shows excess. LOCKED_CASH_MIN_EXCESS_UNITS = 5.0 # Pseudo-SKUs that aren't real products (coupons, marketing line items, etc.) # Excluded from the lost-sales top-5 so the CFO list stays actionable. PSEUDO_SKU_PREFIXES = ("MKT", "OST", "USL", "POPUST", "PROMO_") def _is_pseudo(sku: str) -> bool: s = (sku or "").upper() return any(s.startswith(p) for p in PSEUDO_SKU_PREFIXES) def _week_series(start_y: int, start_w: int, n: int) -> list[tuple[int, int]]: out, y, w = [], start_y, start_w for _ in range(n): out.append((y, w)) w += 1 if w > 52: y, w = y + 1, 1 return out # ───────────────────────────────────────────────────────────────────────── # Context loader — single round-trip to DB # ───────────────────────────────────────────────────────────────────────── class FinanceContext: """Bundles every dataset the finance reports read.""" def __init__(self, db: Session): cur = db.execute(text( "SELECT EXTRACT(ISOYEAR FROM now())::int AS y, " " EXTRACT(WEEK FROM now())::int AS w" )).mappings().first() self.cur_year = int(cur["y"]) self.cur_week = int(cur["w"]) self.horizon = _week_series(self.cur_year, self.cur_week, HORIZON_WEEKS) self.products = pd.read_sql(text(""" SELECT p.id AS product_id, p.sku, COALESCE(p.name, '') AS name, p.datalink::float AS datalink, COALESCE(c.name, 'OTHER') AS category, COALESCE(sp.tier, '') AS tier, COALESCE(sm.lead_time_weeks::float, NULL) AS lead_time_weeks_raw, COALESCE(ds.name, '(unknown)') AS supplier, ec.cost_price::float AS erp_cost, ep.avg_sell_price::float AS avg_sell_price, ep.normal_retail_ppp::float AS retail_ppp, ep.normal_webshop_ppp::float AS webshop_ppp, sp.vpc::float AS vpc, sp.ws_share_26w::float AS ws_share FROM dim_products p LEFT JOIN dim_categories c ON c.id = p.category_id LEFT JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN supply_master sm ON sm.product_id = p.id LEFT JOIN dim_suppliers ds ON ds.id = sm.supplier_id LEFT JOIN erp_costs ec ON ec.product_id = p.id LEFT JOIN erp_prices ep ON ep.product_id = p.id """), db.bind) self.products["product_id"] = self.products["product_id"].astype(int) # Compute the "newly listed" threshold — top N by datalink. SKUs at # or above this threshold get the NEW_LISTING classification when # they have no sales (instead of being misclassified as DEAD_STOCK). with_dl = self.products["datalink"].dropna().sort_values(ascending=False) if len(with_dl) > NEW_LISTING_TOP_N: self.new_listing_threshold = float(with_dl.iloc[NEW_LISTING_TOP_N - 1]) else: self.new_listing_threshold = float(with_dl.min()) if len(with_dl) else 0.0 stock_rows = pd.read_sql(text(""" SELECT p.id AS product_id, p.sku, CASE WHEN ds.is_warehouse THEN 'wh' ELSE 'store' END AS bucket, SUM(esc.stock_qty)::float AS qty FROM erp_stock_current esc JOIN dim_products p ON p.id = esc.product_id JOIN dim_stores ds ON ds.id = esc.store_id GROUP BY p.id, p.sku, bucket """), db.bind) self.wh_map = {int(r.product_id): float(r.qty) for r in stock_rows.itertuples() if r.bucket == "wh"} self.store_map = {int(r.product_id): float(r.qty) for r in stock_rows.itertuples() if r.bucket == "store"} fc_df = pd.read_sql(text(""" SELECT product_id, year, week, COALESCE(total, 0)::float AS demand FROM forecasts WHERE run_id = (SELECT MAX(run_id) FROM forecasts) """), db.bind) self.fc_total: dict[tuple[int, int, int], float] = { (int(r.product_id), int(r.year), int(r.week)): float(r.demand) for r in fc_df.itertuples() } # Demand fallback: avg of last 13 ISO weeks EXCLUDING promo weeks. # If a SKU has all 13 weeks flagged as promo, we fall back to the # unfiltered 13-week avg so it doesn't collapse to zero. Window # length aligned with Supply (get_run_rates) and Executive — every # module uses the same fallback definition for SKUs without forecast. if self.cur_week > 13: avg_cutoff_y, avg_cutoff_w = self.cur_year, self.cur_week - 13 else: avg_cutoff_y, avg_cutoff_w = self.cur_year - 1, 52 + self.cur_week - 13 avg_cutoff_yw = avg_cutoff_y * 100 + avg_cutoff_w max_yw = db.execute(text( "SELECT MAX(year*100+week) FROM v_sales_weekly_full" )).scalar() or 0 avg_df = pd.read_sql(text(""" WITH win AS ( SELECT v.product_id, v.year, v.week, v.qty_total, COALESCE(epw.is_erp_promo, FALSE) AS is_promo FROM v_sales_weekly_full v LEFT JOIN erp_promo_weeks epw ON epw.product_id = v.product_id AND epw.year = v.year AND epw.week = v.week WHERE v.year*100+v.week BETWEEN :cutoff AND :max_yw ), non_promo AS ( SELECT product_id, AVG(qty_total)::float AS avg_np, COUNT(*)::int AS n_np FROM win WHERE NOT is_promo GROUP BY product_id ), all_weeks AS ( SELECT product_id, AVG(qty_total)::float AS avg_all FROM win GROUP BY product_id ) SELECT a.product_id, COALESCE(np.avg_np, a.avg_all)::float AS avg_qty, COALESCE(np.n_np, 0)::int AS n_weeks_used FROM all_weeks a LEFT JOIN non_promo np ON np.product_id = a.product_id """), db.bind, params={"cutoff": avg_cutoff_yw, "max_yw": max_yw}) self.avg_map = {int(r.product_id): float(r.avg_qty or 0.0) for r in avg_df.itertuples()} # Stats for transparency / UI self.avg_window_weeks = 12 self.avg_skus_with_data = int((avg_df["avg_qty"] > 0).sum()) self.avg_skus_all_promo = int((avg_df["n_weeks_used"] == 0).sum()) # Set of pids that had any qty_total > 0 in the last 26 ISO weeks. # Used together with stock + datalink to identify "dormant" SKUs # — no stock, not newly-listed, no sales in 6 months — which the # reports skip entirely (otherwise a forecasted SKU with no real # demand still appears as "lost sales" against zero stock). if self.cur_week > 26: cutoff_y, cutoff_w = self.cur_year, self.cur_week - 26 else: cutoff_y, cutoff_w = self.cur_year - 1, 52 + self.cur_week - 26 cutoff_yw = cutoff_y * 100 + cutoff_w sold_df = pd.read_sql(text(""" SELECT v.product_id, SUM(v.qty_total)::float AS qty_26w FROM v_sales_weekly_full v WHERE v.year * 100 + v.week >= :cutoff GROUP BY v.product_id """), db.bind, params={"cutoff": cutoff_yw}) self.sold_26w: set[int] = {int(r.product_id) for r in sold_df.itertuples() if (r.qty_26w or 0) > 0} # Per-SKU 26-week sold quantity — used by Locked Cash to gate the # non-food "active overstock" vs "slow stock" tracks. self.sold_26w_qty_map: dict[int, float] = { int(r.product_id): float(r.qty_26w or 0.0) for r in sold_df.itertuples() } realized_df = pd.read_sql(text(""" SELECT product_id, SUM(quantity)::float AS units, SUM(purchase_value)::float AS pv, SUM(ruc_eur)::float AS ruc, SUM(total_value)::float AS rev FROM erp_transactions WHERE transaction_date >= (CURRENT_DATE - INTERVAL '13 weeks') GROUP BY product_id """), db.bind) self.realized: dict[int, dict] = {} for r in realized_df.itertuples(): u = float(r.units or 0) if u <= 0: continue self.realized[int(r.product_id)] = { "units": u, "cost_per_unit": (r.pv / u) if r.pv else None, "rev_per_unit": (r.rev / u) if r.rev else None, } npd_df = pd.read_sql(text( "SELECT sku, cost_price::float AS npd_cost, msrp_hr::float AS msrp_hr " "FROM npd_products" ), db.bind) self.npd_cost = {r.sku: float(r.npd_cost) for r in npd_df.itertuples() if r.npd_cost is not None and not math.isnan(r.npd_cost)} self.npd_price = {r.sku: float(r.msrp_hr) for r in npd_df.itertuples() if r.msrp_hr is not None and not math.isnan(r.msrp_hr)} inc_df = pd.read_sql(text(""" SELECT product_id, year, week, SUM(quantity)::float AS qty FROM incoming_supply WHERE COALESCE(LOWER(status), '') <> 'cancelled' GROUP BY product_id, year, week """), db.bind) self.inc_map: dict[tuple[int, int, int], float] = { (int(r.product_id), int(r.year), int(r.week)): float(r.qty) for r in inc_df.itertuples() } # ── Historical receipt cost from data/NabavneCijene.xlsx ────── # Per-SKU per-month avg unit cost from goods-in receipts. Used by # the monthly bridge to compute real COGS effect — plan_cost = # prior-3-month avg, actual_cost = current month avg. Falls back # to current Sifrarnik NabCj (erp_costs.cost_price) for SKUs # without receipt history in a given window. nab_path = Path(__file__).resolve().parents[2] / "data" / "NabavneCijene.xlsx" self.cost_history: pd.DataFrame = pd.DataFrame( columns=["sku", "yr", "mo", "unit_cost", "qty"], ) if nab_path.exists(): try: nab = pd.read_excel(nab_path) nab = nab.rename(columns={ "Šifra": "sku", "Datum": "date", "Količina": "qty", "Nabavna vrijednost €": "cost_eur", }) nab["sku"] = nab["sku"].astype(str).str.strip() nab["date"] = pd.to_datetime(nab["date"], errors="coerce") nab["qty"] = pd.to_numeric(nab["qty"], errors="coerce") nab["cost_eur"] = pd.to_numeric(nab["cost_eur"], errors="coerce") nab = nab.dropna(subset=["sku", "date", "qty", "cost_eur"]) nab = nab[nab["qty"] > 0] nab["yr"] = nab["date"].dt.year nab["mo"] = nab["date"].dt.month # Weighted avg unit cost per (sku, year, month) using # quantity as the weight — large receipts dominate the avg g = nab.groupby(["sku", "yr", "mo"], as_index=False).agg( sum_cost=("cost_eur", "sum"), sum_qty =("qty", "sum"), ) g["unit_cost"] = g["sum_cost"] / g["sum_qty"].replace(0, pd.NA) self.cost_history = g[["sku", "yr", "mo", "unit_cost", "sum_qty"] ].rename(columns={"sum_qty": "qty"}) except Exception: pass # Promo uplift map from data/sku_uplift.csv self.uplift_map: dict[str, float] = {} uplift_path = Path(__file__).resolve().parents[2] / "data" / "sku_uplift.csv" if uplift_path.exists(): up = pd.read_csv(uplift_path) for r in up.itertuples(): u = float(getattr(r, "promo_uplift", 1.0) or 1.0) if u > 1.0: self.uplift_map[r.sku] = u # ── Dormant SKUs — excluded from every Finance report ───────── # A SKU is "dormant" when ALL three are true: # • no stock anywhere (WH + stores = 0) # • NOT newly listed (datalink below the top-10k threshold, # or no datalink at all) # • no sales in last 26 weeks # These SKUs distort the lost-sales walk (a stale forecast # against zero stock = phantom lost sales) and have no business # in the locked-cash / slow-mover frames either. self.dormant_pids: set[int] = set() for prod in self.products.itertuples(): pid = int(prod.product_id) if (self.wh_map.get(pid, 0.0) + self.store_map.get(pid, 0.0)) > 0: continue if pid in self.sold_26w: continue dl = prod.datalink is_new = (dl is not None and not (isinstance(dl, float) and math.isnan(dl)) and dl >= self.new_listing_threshold) if is_new: continue self.dormant_pids.add(pid) # ───────────────────────────────────────────────────────────────────────── # Per-SKU helpers # ───────────────────────────────────────────────────────────────────────── def _cost(pid: int, sku: str, prod, ctx: FinanceContext) -> float: v = prod.erp_cost if v is not None and not (isinstance(v, float) and math.isnan(v)): return float(v) r = ctx.realized.get(pid) if r and r.get("cost_per_unit"): return float(r["cost_per_unit"]) return float(ctx.npd_cost.get(sku, 0.0) or 0.0) def _price(pid: int, sku: str, prod, ctx: FinanceContext) -> float: """Realized blended selling price per unit (the user's "pondered" price from Q4). Cascade: 1. **Realized**: SUM(total_value) / SUM(quantity) from `erp_transactions` over the last 13 weeks — already inherently weighted by the SKU's channel mix (87% wholesale at €13 → blended price reflects that). This is the "Vrijednost" column number, reconciles with Executive Revenue Pulse exactly. 2. **List fallback**: `erp_prices.avg_sell_price` — used only for SKUs that haven't sold in 13 weeks (no realized signal). 3. **NPD list**: `npd_products.msrp_hr` — for brand-new launches. Order matters: realized > list. The old order (list first) inflated Finance numbers 30%+ on wholesale-heavy SKUs because avg_sell_price excludes the wholesale channel entirely.""" r = ctx.realized.get(pid) if r and r.get("rev_per_unit"): return float(r["rev_per_unit"]) v = prod.avg_sell_price if v is not None and not (isinstance(v, float) and math.isnan(v)) and v > 0: return float(v) return float(ctx.npd_price.get(sku, 0.0) or 0.0) def _lead_time(prod) -> Optional[float]: """Lead time in weeks from supply_master. Returns None when missing — callers should treat that as a data-quality issue (no silent default).""" v = prod.lead_time_weeks_raw if v is not None and not (isinstance(v, float) and math.isnan(v)) and v > 0: return float(v) return None def _incoming_in_lt(pid: int, lt: Optional[float], ctx: "FinanceContext") -> float: """Sum incoming_supply qty for `pid` within the first `lt` weeks of the 13-week horizon. Partial trailing week prorated. Returns 0.0 if `lt` is None or <=0 (i.e. no supply_master row).""" if lt is None or lt <= 0 or not ctx.horizon: return 0.0 whole = int(lt) frac = lt - whole total = 0.0 for i, (y, w) in enumerate(ctx.horizon): if i < whole: total += float(ctx.inc_map.get((pid, y, w), 0.0)) elif i == whole and frac > 0: total += frac * float(ctx.inc_map.get((pid, y, w), 0.0)) else: break return total def _weekly_demand(pid: int, y: int, w: int, ctx: FinanceContext) -> float: v = ctx.fc_total.get((pid, y, w)) if v is not None: return float(v) return float(ctx.avg_map.get(pid, 0.0)) def _is_new_listing(prod, ctx: FinanceContext) -> bool: """True if SKU is in the top-N most recently listed products (by Sifrarnik DataLink). New listings have legitimate reasons for large stock + low/no sales so they're excluded from Locked Cash, Lost Sales and Contest Risk where they'd otherwise be flagged.""" dl = getattr(prod, "datalink", None) if dl is None: return False if isinstance(dl, float) and math.isnan(dl): return False return dl >= ctx.new_listing_threshold # ───────────────────────────────────────────────────────────────────────── # Report 1 — Lost sales # ───────────────────────────────────────────────────────────────────────── def report_lost_sales(ctx: FinanceContext) -> dict: rows: list[dict] = [] n_new_excluded = 0 for prod in ctx.products.itertuples(): pid, sku = int(prod.product_id), prod.sku if _is_pseudo(sku): continue if pid in ctx.dormant_pids: continue # no stock + not new + no sales 26w → skip if _is_new_listing(prod, ctx): n_new_excluded += 1 continue # newly listed SKUs have no demand baseline yet wh = ctx.wh_map.get(pid, 0.0) store = ctx.store_map.get(pid, 0.0) avg = ctx.avg_map.get(pid, 0.0) has_fc = any((pid, y, w) in ctx.fc_total for (y, w) in ctx.horizon) if wh + store < 1 and avg < 0.1 and not has_fc: continue cost = _cost(pid, sku, prod, ctx) price = _price(pid, sku, prod, ctx) or (cost * 2 if cost else 0.0) stock_now = wh + store s = stock_now total_demand = total_incoming = lost_units = 0.0 stockout_yw: Optional[int] = None weeks_out = 0 for (y, w) in ctx.horizon: d = _weekly_demand(pid, y, w, ctx) i = ctx.inc_map.get((pid, y, w), 0.0) total_demand += d total_incoming += i closing = s - d + i if closing < 0: lost_units += -closing weeks_out += 1 if stockout_yw is None: stockout_yw = y * 100 + w s = 0.0 else: s = closing rows.append({ "sku": sku, "name": prod.name, "tier": prod.tier, "category": prod.category, "supplier": prod.supplier, "stock_now": round(stock_now, 1), "wh_stock": round(wh, 1), "store_stock": round(store, 1), "total_incoming_h": round(total_incoming, 1), "total_demand_h": round(total_demand, 1), "stockout_week": (f"CW{stockout_yw % 100} {stockout_yw // 100}" if stockout_yw else "SAFE"), "weeks_out_of_stock": weeks_out, "lost_demand_units": round(lost_units, 1), "cost_price": round(cost, 4), "selling_price": round(price, 4), "lost_sales_cost_eur": round(lost_units * cost, 2), "lost_sales_revenue_eur": round(lost_units * price, 2), }) df = pd.DataFrame(rows).sort_values("lost_sales_revenue_eur", ascending=False) at_risk = df[df["stockout_week"] != "SAFE"] by_cat = at_risk.groupby("category").agg( skus_at_risk=("sku", "count"), lost_units=("lost_demand_units", "sum"), lost_cost_eur=("lost_sales_cost_eur", "sum"), lost_revenue_eur=("lost_sales_revenue_eur", "sum"), ).reset_index().sort_values("lost_revenue_eur", ascending=False) by_tier = at_risk.groupby("tier").agg( skus_at_risk=("sku", "count"), lost_units=("lost_demand_units", "sum"), lost_cost_eur=("lost_sales_cost_eur", "sum"), lost_revenue_eur=("lost_sales_revenue_eur", "sum"), ).reset_index().sort_values("lost_revenue_eur", ascending=False) return { "df": df, "summary": { "total_cost_eur": float(at_risk["lost_sales_cost_eur"].sum()), "total_revenue_eur": float(at_risk["lost_sales_revenue_eur"].sum()), "total_units": float(at_risk["lost_demand_units"].sum()), "n_total_skus": int(len(df)), "n_at_risk": int(len(at_risk)), "n_safe": int(len(df) - len(at_risk)), "n_new_excluded": n_new_excluded, }, "by_category": by_cat, "by_tier": by_tier, "top5": at_risk.head(5).to_dict("records"), } # ───────────────────────────────────────────────────────────────────────── # Report 1b — Lost sales, MONTHLY (past-looking, RUC-focused) # ───────────────────────────────────────────────────────────────────────── def report_lost_sales_monthly( db: Session, year: int, month: int, tiers: Optional[list[str]] = None, ) -> dict: """Past-month lost RUC due to apparent OOS, restricted to tiered SKUs. Methodology — per (tiered SKU, ISO week landing in target month by Thursday rule): expected_qty = backtest/forecast prediction for that week, else trailing-13w avg ending BEFORE the target month actual_qty = v_sales_weekly_full quantity for that week shortfall = expected_qty − actual_qty (only counted when actual_qty ≤ 30% × expected_qty AND expected_qty ≥ 5 — filters small forecast errors; isolates the apparent-OOS signal of sales collapsing well below a non-trivial baseline) lost_ruc = shortfall × ruc_rate_per_unit (ruc_rate = trailing 8w retail+webshop+wholesale blended RUC per unit from erp_transactions) Tier filter: tiers parameter defaults to Gold/Silver/Bronze. Non-tiered SKUs are excluded — the user only cares about ranked range. """ if tiers is None: tiers = ["01 GOLD", "02 SILVER", "03 BRONZE"] # Heuristic thresholds for the OOS-vs-forecast-miss filter. The 30% # actual / 5-unit baseline gate is permissive enough to catch SKUs # that sold a few residual units before stocking out, while excluding # noisy long-tail SKUs whose baseline was tiny to begin with. MIN_EXPECTED = 5.0 ACTUAL_RATIO = 0.30 # ── 1) Resolve ISO weeks belonging to this calendar month (Thursday rule). # Exclude the current ISO week — partial-week actuals always read as # "actual << expected", which the OOS gate would falsely flag for # nearly every SKU. Buyers want the realised lost RUC for COMPLETED # weeks; the current week is still in progress and gets included # next time the page loads after Sunday rolls over. from datetime import date as _date, timedelta as _td today = _date.today() cur_iso = today.isocalendar() cur_yw_int = int(cur_iso[0]) * 100 + int(cur_iso[1]) first = _date(year, month, 1) if month == 12: last = _date(year + 1, 1, 1) - _td(days=1) else: last = _date(year, month + 1, 1) - _td(days=1) yws: set[tuple[int, int]] = set() cur = first while cur <= last: iso = cur.isocalendar() thu = cur + _td(days=(3 - cur.weekday()) % 7) # Only include the week if its Thursday lands in this month — # mirrors monthly_plan_service's "thursday_rule" for clean # one-week-belongs-to-one-month bucketing. if thu.year == year and thu.month == month: iy, iw = int(iso[0]), int(iso[1]) # Skip the current (incomplete) ISO week and anything in the # future — only completed weeks have a meaningful actual vs # expected comparison. if iy * 100 + iw < cur_yw_int: yws.add((iy, iw)) cur += _td(days=1) if not yws: return _empty_lost_sales_monthly(year, month, tiers) yws_sorted = sorted(yws) yw_keys = [y * 100 + w for (y, w) in yws_sorted] # ── 2) Trailing-13w avg ending BEFORE the target month (anchored to # 1 day before month start). Used as fallback expected_qty when a # SKU has no forecast row for the target week. anchor = first - _td(days=1) anchor_yw = anchor.isocalendar() anchor_yw_int = int(anchor_yw[0]) * 100 + int(anchor_yw[1]) # ── 3) Build per-SKU dataset in a single CTE query: tier-filtered # products + per-week expected + per-week actual + RUC rate. sql = text(""" WITH eligible AS ( SELECT p.id AS product_id, p.sku, COALESCE(p.name, '') AS name, COALESCE(c.name, 'OTHER') AS category, sp.tier FROM dim_products p JOIN sku_planning sp ON sp.product_id = p.id LEFT JOIN dim_categories c ON c.id = p.category_id WHERE sp.tier = ANY(:tiers) ), target_weeks AS ( SELECT UNNEST(CAST(:yw_keys AS INTEGER[])) AS yw ), ruc_rate AS ( -- Blended trailing-8w RUC/qty per SKU from erp_transactions, -- all channels (retail + webshop + wholesale combined). The -- "lost" units could have sold on any channel; we use the -- realised mix for valuation. SELECT et.product_id, NULLIF(SUM(et.quantity), 0)::float AS qty, SUM(et.ruc_eur)::float AS ruc FROM erp_transactions et WHERE et.transaction_date >= (DATE :first_day - INTERVAL '8 weeks') AND et.transaction_date < DATE :first_day GROUP BY et.product_id ), actual AS ( SELECT v.product_id, v.year, v.week, SUM(COALESCE(v.qty_total, 0))::float AS qty FROM v_sales_weekly_full v JOIN target_weeks tw ON tw.yw = v.year * 100 + v.week GROUP BY v.product_id, v.year, v.week ), baseline_avg AS ( -- Trailing-13w avg per SKU ending BEFORE target month. SELECT v.product_id, AVG(COALESCE(v.qty_total, 0))::float AS avg_qty FROM v_sales_weekly_full v WHERE (v.year*100 + v.week) BETWEEN (SELECT MIN(year*100+week) FROM ( SELECT DISTINCT year, week FROM v_sales_weekly_full WHERE (year*100+week) <= :anchor_yw ORDER BY year DESC, week DESC LIMIT 13 ) sub) AND :anchor_yw GROUP BY v.product_id ), latest_fc AS ( SELECT DISTINCT ON (product_id, year, week) product_id, year, week, COALESCE(baseline, 0)::float * COALESCE(planner_factor, 1)::float AS fc_qty FROM forecasts WHERE (year*100+week) = ANY(:yw_keys) ORDER BY product_id, year, week, run_id DESC ), backtest AS ( SELECT product_id, year, week, forecast::float AS fc_qty FROM backtest_results WHERE (year*100+week) = ANY(:yw_keys) AND forecast IS NOT NULL ), expected AS ( -- Backtest preferred for past weeks (already validated against -- actuals); forecasts table used for current/future weeks the -- backtest hasn't covered. baseline_avg is the SKU-level -- fallback when neither source has a row. SELECT e.product_id, tw.yw / 100 AS year, tw.yw % 100 AS week, COALESCE(b.fc_qty, f.fc_qty, ba.avg_qty, 0)::float AS expected_qty FROM eligible e CROSS JOIN target_weeks tw LEFT JOIN backtest b ON b.product_id = e.product_id AND b.year * 100 + b.week = tw.yw LEFT JOIN latest_fc f ON f.product_id = e.product_id AND f.year * 100 + f.week = tw.yw LEFT JOIN baseline_avg ba ON ba.product_id = e.product_id ) SELECT e.product_id, e.sku, e.name, e.category, e.tier, exp.year, exp.week, exp.expected_qty, COALESCE(a.qty, 0)::float AS actual_qty, CASE WHEN rr.qty IS NOT NULL AND rr.qty > 0 THEN (rr.ruc / rr.qty)::float ELSE 0.0 END AS ruc_rate FROM eligible e JOIN expected exp ON exp.product_id = e.product_id LEFT JOIN actual a ON a.product_id = e.product_id AND a.year = exp.year AND a.week = exp.week LEFT JOIN ruc_rate rr ON rr.product_id = e.product_id ORDER BY e.tier, e.sku, exp.year, exp.week """) rows = db.execute(sql, { "tiers": tiers, "yw_keys": yw_keys, "first_day": first.isoformat(), "anchor_yw": anchor_yw_int, }).mappings().all() if not rows: return _empty_lost_sales_monthly(year, month, tiers) df = pd.DataFrame([dict(r) for r in rows]) # ── 4) Apply the OOS gate per (sku, week) and aggregate per SKU df["expected_qty"] = df["expected_qty"].astype(float).clip(lower=0) df["actual_qty"] = df["actual_qty"].astype(float).clip(lower=0) df["ruc_rate"] = df["ruc_rate"].astype(float).clip(lower=0) gate = (df["expected_qty"] >= MIN_EXPECTED) & \ (df["actual_qty"] <= df["expected_qty"] * ACTUAL_RATIO) df["lost_qty_wk"] = 0.0 df.loc[gate, "lost_qty_wk"] = (df.loc[gate, "expected_qty"] - df.loc[gate, "actual_qty"]).clip(lower=0) df["lost_ruc_wk"] = df["lost_qty_wk"] * df["ruc_rate"] df["oos_week"] = gate.astype(int) per_sku = df.groupby( ["product_id", "sku", "name", "category", "tier"], as_index=False ).agg( expected_qty=("expected_qty", "sum"), actual_qty=("actual_qty", "sum"), lost_qty=("lost_qty_wk", "sum"), lost_ruc_eur=("lost_ruc_wk", "sum"), ruc_rate=("ruc_rate", "mean"), n_oos_weeks=("oos_week", "sum"), ) per_sku = per_sku.sort_values("lost_ruc_eur", ascending=False) # OOS-weeks badge: pick the WEEK(s) where this SKU stocked out. UI # shows the first one so the row tells the buyer where to look. oos_lookup: dict[int, str] = {} for pid, grp in df[df["oos_week"] == 1].groupby("product_id"): weeks = [f"CW{int(w):02d}" for w in sorted(grp["week"].unique())] oos_lookup[int(pid)] = ", ".join(weeks[:3]) + ("…" if len(weeks) > 3 else "") per_sku["oos_weeks"] = per_sku["product_id"].map(oos_lookup).fillna("") # Filter out SKUs with no lost RUC — they don't belong in the report at_risk = per_sku[per_sku["lost_ruc_eur"] > 0].copy() # ── 5) Aggregations by_tier = at_risk.groupby("tier", as_index=False).agg( n_skus=("sku", "count"), lost_qty=("lost_qty", "sum"), lost_ruc_eur=("lost_ruc_eur", "sum"), ).sort_values("lost_ruc_eur", ascending=False) by_category = at_risk.groupby("category", as_index=False).agg( n_skus=("sku", "count"), lost_qty=("lost_qty", "sum"), lost_ruc_eur=("lost_ruc_eur", "sum"), ).sort_values("lost_ruc_eur", ascending=False) summary = { "year": year, "month": month, "label": _date(year, month, 1).strftime("%b %Y"), "weeks_covered": ", ".join(f"CW{w:02d}" for (_y, w) in yws_sorted), "tiers": tiers, "total_lost_ruc_eur": float(at_risk["lost_ruc_eur"].sum()), "total_lost_qty": float(at_risk["lost_qty"].sum()), "n_skus_lost": int(len(at_risk)), "n_skus_evaluated": int(len(per_sku)), } return { "summary": summary, "rows": at_risk.round(2).to_dict("records"), "by_tier": by_tier.round(2).to_dict("records"), "by_category":by_category.round(2).to_dict("records"), } def _empty_lost_sales_monthly(year: int, month: int, tiers: list[str]) -> dict: from datetime import date as _date return { "summary": { "year": year, "month": month, "label": _date(year, month, 1).strftime("%b %Y"), "weeks_covered": "", "tiers": tiers, "total_lost_ruc_eur": 0.0, "total_lost_qty": 0.0, "n_skus_lost": 0, "n_skus_evaluated": 0, }, "rows": [], "by_tier": [], "by_category": [], } # ───────────────────────────────────────────────────────────────────────── # Report 2 — Locked cash (overstock) # ───────────────────────────────────────────────────────────────────────── def report_locked_cash(ctx: FinanceContext) -> dict: """Locked cash — two-track flagging rule (May 2026). **Food / supplements** (`category` NOT in LOCKED_CASH_NON_FOOD_CATS): - WH stock only - target = SUM of forecasted demand over next 1.5×LT weeks (fill weeks beyond the 13-week forecast horizon with trailing-13w avg) - flagged when wh_stock > target; excess = wh_stock − target **Non-food** (apparel, fitness equipment, drinkware, gadgets, etc.): No usable lead times for these suppliers and demand is lumpy/seasonal, so the LT-multiplier rule breaks down. Two-track logic instead: active overstock — sold_26w >= category gate (X) target = sum of forecast over NEXT 26 weeks slow stock — sold_26w < X target = floor (≈1 unit); everything above is locked Incoming POs are NOT netted in either track — reviewed weekly. Same row schema for both tracks so the UI table renders uniformly.""" rows: list[dict] = [] n_new_excluded = 0 n_below_threshold = 0 n_missing_lt = 0 THRESHOLD_LT_MULT = 1.5 import math as _math for prod in ctx.products.itertuples(): pid, sku = int(prod.product_id), prod.sku if _is_pseudo(sku): continue wh, store = ctx.wh_map.get(pid, 0.0), ctx.store_map.get(pid, 0.0) stock_now = wh # WH-only — store stock excluded from locked-cash math if stock_now <= 0: continue if _is_new_listing(prod, ctx): # New listings just arrived in WH and haven't established # sales yet — their stock isn't "locked cash", it's planned # inventory build. Excluded from this report. n_new_excluded += 1 continue category = prod.category or "" is_non_food = category in LOCKED_CASH_NON_FOOD_CATS cost = _cost(pid, sku, prod, ctx) price = _price(pid, sku, prod, ctx) or (cost * 2 if cost else 0.0) trailing_avg = ctx.avg_map.get(pid, 0.0) sku_in_forecast = any((pid, y, w) in ctx.fc_total for (y, w) in ctx.horizon) def _forecast_sum(n_weeks_needed: int, frac_last: float = 0.0) -> tuple[float, int, int]: """Sum forecast over the requested horizon, filling weeks beyond the 13-week forecast run (or whole horizon if SKU absent) with trailing-13w avg. `frac_last` lets the caller prorate the (n_weeks_needed + 1)-th week. Returns (total, n_fc_weeks_used, n_fill_weeks_used).""" per_week: list[float] = [] n_fc_used = n_fill_used = 0 for i in range(n_weeks_needed + (1 if frac_last > 0 else 0)): if sku_in_forecast and i < len(ctx.horizon): (y, w) = ctx.horizon[i] per_week.append(ctx.fc_total.get((pid, y, w), 0.0)) n_fc_used += 1 else: per_week.append(trailing_avg) n_fill_used += 1 total = sum(per_week[:n_weeks_needed]) if frac_last > 0 and n_weeks_needed < len(per_week): total += per_week[n_weeks_needed] * frac_last return total, n_fc_used, n_fill_used if is_non_food: # ── Non-food two-track rule ───────────────────────────────── sold_26w = ctx.sold_26w_qty_map.get(pid, 0.0) x_gate = NON_FOOD_SELL_THROUGH_X.get(category, DEFAULT_NON_FOOD_X) if sold_26w >= x_gate: # Active overstock: compare WH to forward 26-week demand target_units, n_fc, n_fill = _forecast_sum(NON_FOOD_FORWARD_HORIZON_WEEKS) target_weeks = float(NON_FOOD_FORWARD_HORIZON_WEEKS) track = "active_overstock" else: # Slow stock: anything materially above the floor is locked target_units = NON_FOOD_SLOW_STOCK_FLOOR target_weeks = 0.0 n_fc = n_fill = 0 track = "slow_stock" lt = _lead_time(prod) # informational only; may be None if stock_now <= target_units: n_below_threshold += 1 continue excess = stock_now - target_units if excess < LOCKED_CASH_MIN_EXCESS_UNITS: n_below_threshold += 1 continue avg = (target_units / target_weeks) if target_weeks > 0 else (sold_26w / 26.0) weeks_cover = (stock_now / avg) if avg > 0 else None if track == "active_overstock": if n_fc == 0: demand_src = f"nonfood_active_trailing26w" else: demand_src = f"nonfood_active_fc{n_fc}w+fill{n_fill}w" else: demand_src = f"nonfood_slow_sold26w={int(sold_26w)} 0 else 0.0 weeks_cover = (stock_now / avg) if avg > 0 else None if not sku_in_forecast: demand_src = "trailing_13w_actuals" elif n_fill == 0: demand_src = f"forecast_{n_fc}w" else: demand_src = f"forecast_{n_fc}w+fill_{n_fill}w" rows.append({ "sku": sku, "name": prod.name, "tier": prod.tier, "category": prod.category, "supplier": prod.supplier, "stock_now": round(stock_now, 1), "wh_stock": round(wh, 1), "store_stock": round(store, 1), "weekly_demand": round(avg, 2), "weeks_cover": round(weeks_cover, 1) if weeks_cover is not None else None, "effective_weeks_cover": round(weeks_cover, 1) if weeks_cover is not None else None, "incoming_in_lt": 0.0, "status": "OVERSTOCK", "lead_time_weeks": round(lt, 1) if lt is not None else None, "target_stock_weeks": round(target_weeks, 1), "target_stock_units": round(target_units, 1), "excess_units": round(excess, 1), "cost_price": round(cost, 4), "selling_price": round(price, 4), "stock_value_eur": round(stock_now * cost, 2), "locked_cash_cost_eur": round(excess * cost, 2), "locked_cash_revenue_eur": round(excess * price, 2), "demand_source": demand_src, }) df = pd.DataFrame(rows).sort_values("locked_cash_cost_eur", ascending=False) overstocked = df[df["excess_units"] > 0] by_cat = (df.groupby("category") .agg(sku_count=("sku", "count"), total_stock_eur=("stock_value_eur", "sum"), total_excess_eur=("locked_cash_cost_eur", "sum")) .reset_index()) by_cat["excess_pct"] = (by_cat["total_excess_eur"] / by_cat["total_stock_eur"].replace(0, 1)).round(3) by_cat = by_cat.sort_values("total_excess_eur", ascending=False) by_sup = (df.groupby("supplier") .agg(sku_count=("sku", "count"), total_excess_eur=("locked_cash_cost_eur", "sum")) .reset_index().sort_values("total_excess_eur", ascending=False)) return { "df": df, "summary": { "total_stock_eur": float(df["stock_value_eur"].sum()), "total_excess_cost_eur": float(df["locked_cash_cost_eur"].sum()), "total_excess_rev_eur": float(df["locked_cash_revenue_eur"].sum()), "n_skus": int(len(df)), "n_overstocked": int(len(overstocked)), "n_new_excluded": n_new_excluded, "n_below_threshold": n_below_threshold, "n_missing_lt": n_missing_lt, }, "by_category": by_cat, "by_supplier": by_sup, } # ───────────────────────────────────────────────────────────────────────── # Report 3 — Contest July risk # ───────────────────────────────────────────────────────────────────────── def list_contest_months(ctx: FinanceContext) -> list[dict]: """Months available in contest_monthly_plan that are still upcoming — powers the month selector on the Contest Risk page. A month is 'upcoming' while its calendar month is the current month or later (so the running May contest stays visible all of May, then drops off in June).""" from sqlalchemy import text as _text today = pd.Timestamp.today() cur_yyyymm = today.year * 100 + today.month rows = pd.read_sql(_text(""" SELECT month_key FROM contest_monthly_plan WHERE month_key >= :cur_yyyymm GROUP BY month_key ORDER BY month_key """), ctx_db_bind(ctx), params={"cur_yyyymm": cur_yyyymm}) out: list[dict] = [] for mk in rows["month_key"].astype(int).tolist(): y, m = mk // 100, mk % 100 out.append({ "month_key": mk, "label": f"{pd.Timestamp(year=y, month=m, day=1):%B %Y}", }) return out def report_contest_risk( ctx: FinanceContext, uplift_factor: float = 1.7, month_key: Optional[int] = None, ) -> dict: """Contest risk — scoped to the curated `contest_monthly_plan` for the target month (May 2026 rewrite). Previously this swept every Gold/Silver SKU against a hardcoded July window. Now it only evaluates the handful of SKUs that are actually in the month's Glavni/Sporedni Contest sections (per akcije mjesecne.xlsx, mapped to promo_policy_items via the contest_monthly_plan table). Contest weeks come from the policy's valid_from/valid_to (so July, June, August all work without code changes). `month_key` (YYYYMM): pin the report to a specific upcoming contest. When None, auto-pick the next contest (smallest month_key >= current YYYYMM). """ from sqlalchemy import text as _text # 1) Resolve the active contest plan. # If the caller pinned a month_key (selector), use it; otherwise pick # the next upcoming one, falling back to the most recent if nothing # upcoming exists. today = pd.Timestamp.today() cur_yyyymm = today.year * 100 + today.month if month_key: target_month_key = int(month_key) else: plan = pd.read_sql(_text(""" WITH future_plans AS ( SELECT month_key FROM contest_monthly_plan WHERE month_key >= :cur_yyyymm GROUP BY month_key ORDER BY month_key LIMIT 1 ), any_plan AS ( SELECT month_key FROM contest_monthly_plan GROUP BY month_key ORDER BY month_key DESC LIMIT 1 ) SELECT COALESCE( (SELECT month_key FROM future_plans), (SELECT month_key FROM any_plan) ) AS target_month_key """), ctx_db_bind(ctx), params={"cur_yyyymm": cur_yyyymm}) target_month_key = int(plan.iloc[0]["target_month_key"] or 0) rows: list[dict] = [] contest_weeks: list[tuple[int, int]] = [] if target_month_key: cdf = pd.read_sql(_text(""" SELECT cmp.month_key, cmp.section, cmp.parent_opis, pp.policy_name, pp.valid_from, pp.valid_to, ppi.product_id, p.sku, p.name AS prod_name FROM contest_monthly_plan cmp JOIN promo_policies pp ON pp.policy_name = cmp.policy_name JOIN promo_policy_items ppi ON ppi.policy_id = pp.id AND ppi.description = cmp.parent_opis AND ppi.product_id IS NOT NULL JOIN dim_products p ON p.id = ppi.product_id WHERE cmp.month_key = :mk ORDER BY cmp.section, p.sku """), ctx_db_bind(ctx), params={"mk": target_month_key}) if not cdf.empty: # Derive contest ISO weeks from the policy window (use min from # → max to across all rows in this month, they should agree). vf = pd.to_datetime(cdf["valid_from"]).min() vt = pd.to_datetime(cdf["valid_to"]).max() d = vf.normalize() seen: set[tuple[int, int]] = set() from datetime import timedelta as _td while d <= vt: iso = d.isocalendar() seen.add((int(iso[0]), int(iso[1]))) d = d + _td(days=1) contest_weeks = sorted(seen) # Iterate ONLY contest SKUs — not all products. prod_lookup = {int(p.product_id): p for p in ctx.products.itertuples()} for _, r in cdf.iterrows(): pid = int(r["product_id"]) sku = r["sku"] prod = prod_lookup.get(pid) if prod is None: continue if _is_pseudo(sku): continue wh = ctx.wh_map.get(pid, 0.0) store = ctx.store_map.get(pid, 0.0) stock_now = wh + store cost = _cost(pid, sku, prod, ctx) price = _price(pid, sku, prod, ctx) or (cost * 2 if cost else 0.0) # ── Week-by-week running balance ────────────────────────── # Carry stock forward week by week, flooring at 0 (real stock # can't go negative). Each week: incoming arrives first, then # demand is served. A week that can't be fully served is a # stockout; its shortfall is lost units. This respects WHICH # week each PO lands, so a late delivery no longer papers over # an earlier gap (the lump-sum model's flaw). # # Demand comes straight from the forecast — the engine already # bakes in promo signals via the ERP calendar + per-SKU uplift, # so no extra multiplier (would double-count). _weekly_demand # falls back to trailing-13w avg when a week has no forecast. first_cw = contest_weeks[0] if contest_weeks else None pre_weeks = [(y, w) for (y, w) in ctx.horizon if first_cw is None or (y, w) < first_cw] pre_inc = sum(ctx.inc_map.get((pid, y, w), 0.0) for (y, w) in pre_weeks) # Run forward through the pre-contest weeks to get the stock the # SKU actually enters the contest with (floored at 0). running = stock_now for (y, w) in pre_weeks: running = max(0.0, running + ctx.inc_map.get((pid, y, w), 0.0) - _weekly_demand(pid, y, w, ctx)) available = running # projected on-hand entering the contest # Walk the contest weeks, accumulating any weekly shortfall. contest_demand = 0.0 inc_during = 0.0 gap = 0.0 # total unmet units across contest first_stockout_cw: Optional[str] = None for (y, w) in contest_weeks: dem = _weekly_demand(pid, y, w, ctx) inc = ctx.inc_map.get((pid, y, w), 0.0) contest_demand += dem inc_during += inc running += inc if running < dem: # can't fully serve this week gap += dem - running if first_stockout_cw is None: first_stockout_cw = f"CW{w:02d}" running = 0.0 else: running -= dem contest_label = "AT RISK" if gap > 0.5 else "SAFE" rows.append({ "sku": sku, "name": prod.name, "tier": prod.tier, "category": prod.category, "supplier": prod.supplier, "stock_now": round(stock_now, 1), "wh_stock": round(wh, 1), "store_stock": round(store, 1), "incoming_before_contest": round(pre_inc, 1), "available_at_contest": round(available, 1), "incoming_during_contest": round(inc_during, 1), "baseline_demand_july": round(contest_demand, 1), # kept name for FE compat "promo_uplift_factor": 1.0, # no extra uplift — forecast already promo-aware "contest_demand_july": round(contest_demand, 1), # kept name for FE compat "stockout_baseline": contest_label, # identical now "stockout_contest": contest_label, "gap_units": round(gap, 1), "gap_cost_eur": round(gap * cost, 2), "gap_revenue_eur": round(gap * price, 2), "first_stockout_cw": first_stockout_cw, "contest_month_key": target_month_key, "contest_section": r["section"], "contest_parent": r["parent_opis"], }) df = pd.DataFrame(rows).sort_values("gap_revenue_eur", ascending=False) \ if rows else pd.DataFrame() at_risk = df[df["stockout_contest"] == "AT RISK"] if not df.empty else df summary = { "uplift_factor": uplift_factor, "n_skus_evaluated": int(len(df)), "n_at_risk_baseline": int((df["stockout_baseline"] == "AT RISK").sum()) if not df.empty else 0, "n_at_risk_contest": int(len(at_risk)), "gap_revenue_eur_baseline": float(df[df["stockout_baseline"] == "AT RISK"]["gap_revenue_eur"].sum()) if not df.empty else 0.0, "gap_revenue_eur_contest": float(at_risk["gap_revenue_eur"].sum()) if not at_risk.empty else 0.0, "gap_cost_eur_contest": float(at_risk["gap_cost_eur"].sum()) if not at_risk.empty else 0.0, "n_new_excluded": 0, "contest_month_key": target_month_key, "contest_weeks": [f"CW{w:02d}" for (_, w) in contest_weeks], } return { "df": df, "summary": summary, "top10_at_risk": at_risk.head(10).to_dict("records") if not at_risk.empty else [], } def ctx_db_bind(ctx: FinanceContext): """Helper to expose a SQLAlchemy connectable that report_contest_risk's pd.read_sql can use. The ctx itself doesn't carry the engine, but every DataFrame on it was loaded from the same session — re-grab one via its product_id query was overkill, so we just expose a fresh engine via SQLALCHEMY_DATABASE_URL on first call. Kept tiny so it's easy to swap if ctx ever grows an explicit `.bind` attribute.""" if not hasattr(ctx, "_bound_engine"): from sqlalchemy import create_engine import os url = os.environ.get("DATABASE_URL") or \ "postgresql://polleo:polleo_dev@localhost:5432/polleo_demand" ctx._bound_engine = create_engine(url, pool_pre_ping=True) return ctx._bound_engine # ───────────────────────────────────────────────────────────────────────── # Report 4 — Slow movers # ───────────────────────────────────────────────────────────────────────── def report_slow_movers(ctx: FinanceContext) -> dict: rows: list[dict] = [] for prod in ctx.products.itertuples(): pid, sku = int(prod.product_id), prod.sku if _is_pseudo(sku): continue stock = ctx.wh_map.get(pid, 0.0) + ctx.store_map.get(pid, 0.0) if stock <= 0: continue cost = _cost(pid, sku, prod, ctx) price = _price(pid, sku, prod, ctx) or (cost * 2 if cost else 0.0) avg = ctx.avg_map.get(pid, 0.0) fc_13w = sum(ctx.fc_total.get((pid, y, w), 0.0) for (y, w) in ctx.horizon) lt = _lead_time(prod) dl = prod.datalink is_new_listing = ( dl is not None and not (isinstance(dl, float) and math.isnan(dl)) and dl >= ctx.new_listing_threshold ) no_sales = (avg == 0 and fc_13w == 0) # DEAD_STOCK takes precedence over MISSING_LT: a SKU with stock # sitting in WH+stores that hasn't sold in 13+ weeks is dead # regardless of whether supply_master has a lead time for it. # Long-tail / non-tiered (no Gold/Silver/Bronze) often lacks an # LT entry — without this gate they used to silently drop into # MISSING_LT and stayed invisible. weeks_cover = (stock / avg) if avg > 0 else None if is_new_listing and no_sales: klass = "NEW_LISTING" target_weeks = None excess = 0.0 elif no_sales: klass = "DEAD_STOCK" target_weeks = None excess = stock # value the whole stock as locked elif lt is None: # Has demand but no LT → can't classify cover-based buckets. klass = "MISSING_LT" target_weeks = None excess = 0.0 else: safety = SAFETY_WEEKS.get(prod.tier or "", DEFAULT_SAFETY) target_weeks = lt + REVIEW_PERIOD_WEEKS + safety # Overstock band tightened (2026-05-26 per Lovro): cover # above 1.5×LT is overstock instead of the previous 2×LT. # SLOW_MOVER threshold moved down in lockstep (was 3×LT, # now 2.5×LT) so the two bands stay an LT apart. if weeks_cover is None: klass = "DEAD_STOCK" elif weeks_cover > lt * 2.5: klass = "SLOW_MOVER" elif weeks_cover > lt * 1.5: klass = "OVERSTOCK" elif weeks_cover < lt: klass = "UNDERSTOCK" else: klass = "HEALTHY" excess = max(0.0, stock - target_weeks * avg) if avg > 0 else stock rows.append({ "sku": sku, "name": prod.name, "tier": prod.tier, "category": prod.category, "supplier": prod.supplier, "stock_units": round(stock, 1), "stock_cost_eur": round(stock * cost, 2), "stock_revenue_eur": round(stock * price, 2), "weekly_demand": round(avg, 2), "weeks_cover": round(weeks_cover, 1) if weeks_cover is not None else None, "lead_time_weeks": round(lt, 1) if lt is not None else None, "classification": klass, "excess_units": round(excess, 1), "excess_cost_eur": round(excess * cost, 2), "excess_revenue_eur": round(excess * price, 2), }) df = pd.DataFrame(rows).sort_values("stock_cost_eur", ascending=False) all_classes = ("DEAD_STOCK", "SLOW_MOVER", "OVERSTOCK", "HEALTHY", "UNDERSTOCK", "NEW_LISTING") def _by(group_col: str) -> pd.DataFrame: if df.empty: return pd.DataFrame() agg = (df.groupby([group_col, "classification"]) .agg(stock_eur=("stock_cost_eur", "sum")) .unstack(fill_value=0)) agg.columns = [c[1] for c in agg.columns] for k in all_classes: if k not in agg.columns: agg[k] = 0.0 agg["total_stock_eur"] = sum(agg[k] for k in all_classes) # New listings are NOT counted as unhealthy agg["unhealthy_eur"] = agg["DEAD_STOCK"] + agg["SLOW_MOVER"] + agg["OVERSTOCK"] agg["unhealthy_pct"] = (agg["unhealthy_eur"] / agg["total_stock_eur"].replace(0, 1)).round(3) return agg.reset_index().sort_values("unhealthy_eur", ascending=False) by_cat = _by("category") by_sup = _by("supplier") def _sum_class(k): return float(df[df["classification"] == k]["stock_cost_eur"].sum()) return { "df": df, "summary": { "dead_stock_eur": _sum_class("DEAD_STOCK"), "slow_mover_eur": _sum_class("SLOW_MOVER"), "overstock_eur": _sum_class("OVERSTOCK"), "healthy_eur": _sum_class("HEALTHY"), "understock_eur": _sum_class("UNDERSTOCK"), "new_listing_eur": _sum_class("NEW_LISTING"), "gadgets_eur": float(df[df["category"].isin(GADGET_CATEGORIES)]["stock_cost_eur"].sum()), "non_food_eur": float(df[df["category"].isin(NON_FOOD_CATEGORIES)]["stock_cost_eur"].sum()), "n_skus": int(len(df)), "by_class_count": df["classification"].value_counts().to_dict(), }, "by_category": by_cat, "by_supplier": by_sup, } # ───────────────────────────────────────────────────────────────────────── # Bridge — monthly Volume / Price / Mix / COGS variance decomposition # ───────────────────────────────────────────────────────────────────────── from backend.services.time_utils import ( days_in_calendar_month, split_iso_week_across_months, ) _MONTH_LABEL = ["", "Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] def _explode_weekly_to_months(df: pd.DataFrame, value_cols: list[str]) -> pd.DataFrame: """Split weekly rows into calendar-month rows by day-count fraction. Input: df with columns [..., 'year', 'week', value_cols...] Output: same columns but extra ('cy', 'cm') plus value_cols scaled by the day-fraction of each week that fell in that calendar month. A week wholly inside one month produces one output row; a straddle-week produces two rows. The sum of value_cols across the output rows for a given input week equals the original weekly value (mass-preserving). """ if df.empty: out = df.copy() out["cy"] = pd.Series(dtype="int") out["cm"] = pd.Series(dtype="int") return out out_rows = [] for r in df.itertuples(index=False): d = r._asdict() for cy, cm, frac in split_iso_week_across_months(int(r.year), int(r.week)): row = dict(d) row["cy"] = cy row["cm"] = cm for col in value_cols: row[col] = float(d[col] or 0) * frac out_rows.append(row) out = pd.DataFrame(out_rows) return out def report_bridge_monthly(ctx: FinanceContext, db: Session) -> dict: """Build a monthly bridge for every calendar month that has both actuals + a plan baseline. Plan baseline cascade per month: - **Tier 3 — full**: backtest_fa.forecast for SKUs that have it - **Tier 1 — volume-only fallback**: SKU's trailing 12w non-promo avg × number of weeks in the month → used when backtest is missing for that SKU-week. Actuals always come from: - units: v_sales_weekly_full.qty_total - revenue (net of VAT): erp_transactions.tax_base summed - cost: erp_transactions.purchase_value / quantity Months without erp_transactions detail → Tier 1 (volume effect only). Months with erp_transactions → Tier 3 (full V/P/M/COGS). """ from datetime import date # Load v_sales_weekly_full for the last 18 months to give us a wide # historical window. Aggregate to (sku, year, week, qty). actuals_w = pd.read_sql(text(""" SELECT p.sku, p.id AS product_id, v.year, v.week, v.qty_total::float AS qty FROM v_sales_weekly_full v JOIN dim_products p ON p.id = v.product_id WHERE v.qty_total > 0 """), db.bind) if actuals_w.empty: return {"available": False, "type": "no_actuals", "months": []} # Split each weekly row across calendar months by day-count fraction. # A week straddling a month boundary contributes pro-rata to each # month — so CW18 2026 (Apr 27-May 3) contributes 4/7 of its value # to April and 3/7 to May. This is the mass-preserving alternative # to "snap to Thursday". actuals_w = _explode_weekly_to_months(actuals_w, ["qty"]) actuals_w["month_key"] = actuals_w["cy"] * 100 + actuals_w["cm"] # Backtest plan (where available) — day-fraction splitting. # Loaded from DB (backtest_results) now that the migration added the # per-channel forecast_retail / forecast_wholesale columns (from # data/backtest_fa.csv). "retail" in the engine = retail + webshop # combined (= "B2C"), "wholesale" stands alone. The bridge uses these # to split plan_qty per channel for the wholesale + B2C bridges. bt_df = pd.read_sql(text(""" SELECT p.sku, b.year, b.week, b.forecast::float AS forecast, COALESCE(b.forecast_wholesale, 0)::float AS forecast_ws, COALESCE(b.forecast_retail, 0)::float AS forecast_b2c FROM backtest_results b JOIN dim_products p ON p.id = b.product_id WHERE b.forecast IS NOT NULL AND b.forecast > 0 """), db.bind) bt: Optional[pd.DataFrame] = None if not bt_df.empty: bt = _explode_weekly_to_months( bt_df, ["forecast", "forecast_ws", "forecast_b2c"] ) bt["month_key"] = bt["cy"] * 100 + bt["cm"] # ERP transactions aggregated per (sku, year, month) for price+cost. # We split by channel (wholesale vs B2C = retail+webshop+RAC) so the # bridge can decompose price effects per channel rather than blending # wholesale's lower VPC into a single number that looks like a pricing # disaster. txn = pd.read_sql(text(""" SELECT p.sku, EXTRACT(YEAR FROM et.transaction_date)::int AS cy, EXTRACT(MONTH FROM et.transaction_date)::int AS cm, -- All-channel totals (back-compat with existing bridge logic) SUM(et.quantity)::float AS qty, SUM(et.total_value)::float AS rev_gross, SUM(COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80))::float AS rev_net, SUM(et.purchase_value)::float AS pv, -- Wholesale-only (cm.channel = 'wholesale'; RAC is wholesale per -- user clarification — Polleo handles B2B-ish web orders through -- KAM channel even though they're booked via web) SUM(CASE WHEN cm.channel = 'wholesale' THEN et.quantity ELSE 0 END)::float AS qty_ws, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.total_value ELSE 0 END)::float AS rev_gross_ws, SUM(CASE WHEN cm.channel = 'wholesale' THEN COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80) ELSE 0 END)::float AS rev_net_ws, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.purchase_value ELSE 0 END)::float AS pv_ws, -- B2C-only (retail + webshop) SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.quantity ELSE 0 END)::float AS qty_b2c, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.total_value ELSE 0 END)::float AS rev_gross_b2c, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80) ELSE 0 END)::float AS rev_net_b2c, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.purchase_value ELSE 0 END)::float AS pv_b2c, -- Retail-only (used to pick the right list price for B2C) SUM(CASE WHEN cm.channel = 'retail' THEN et.quantity ELSE 0 END)::float AS qty_retail, SUM(CASE WHEN cm.channel = 'webshop' THEN et.quantity ELSE 0 END)::float AS qty_webshop FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id JOIN lookup_channel_map cm ON cm.id = et.channel_map_id GROUP BY p.sku, cy, cm """), db.bind) txn["month_key"] = txn["cy"] * 100 + txn["cm"] txn["actual_unit_price"] = txn["rev_net"] / txn["qty"].replace(0, pd.NA) txn["actual_unit_cost"] = txn["pv"] / txn["qty"].replace(0, pd.NA) # Channel-specific actual unit prices (net of VAT) txn["actual_unit_price_ws"] = txn["rev_net_ws"] / txn["qty_ws"].replace(0, pd.NA) txn["actual_unit_price_b2c"] = txn["rev_net_b2c"] / txn["qty_b2c"].replace(0, pd.NA) # Channel-specific actual unit cost (booked COGS, channel-split) txn["actual_unit_cost_ws"] = txn["pv_ws"] / txn["qty_ws"].replace(0, pd.NA) txn["actual_unit_cost_b2c"] = txn["pv_b2c"] / txn["qty_b2c"].replace(0, pd.NA) # ── Per-SKU per-channel RUC rate (May 2026 v2 refactor) ───────────── # Pulled from the last 4 ISO weeks of erp_transactions — same source # Revenue Forecast page uses for its RUC view rates. Bridge plan_margin # is computed as forecast_qty × this rate per channel, so Bridge and # Revenue Forecast agree on plan (= forecast × per-unit RUC) for any # overlapping period. ruc_rates = pd.read_sql(text(""" WITH recent_weeks AS ( SELECT DISTINCT year, week FROM v_sales_weekly_full ORDER BY year DESC, week DESC LIMIT 4 ), recent_yws AS ( SELECT year * 100 + week AS yw FROM recent_weeks ) SELECT p.sku, NULLIF(SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.quantity ELSE 0 END), 0)::float AS qty_r, SUM(CASE WHEN cm.channel IN ('retail','webshop') THEN et.ruc_eur ELSE 0 END)::float AS ruc_r, NULLIF(SUM(CASE WHEN cm.channel = 'wholesale' THEN et.quantity ELSE 0 END), 0)::float AS qty_w, SUM(CASE WHEN cm.channel = 'wholesale' THEN et.ruc_eur ELSE 0 END)::float AS ruc_w FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE (EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100 + EXTRACT(WEEK FROM et.transaction_date)::int) IN (SELECT yw FROM recent_yws) GROUP BY p.sku """), db.bind) ruc_rates["retail_ruc_rate"] = ruc_rates["ruc_r"] / ruc_rates["qty_r"] ruc_rates["wholesale_ruc_rate"] = ruc_rates["ruc_w"] / ruc_rates["qty_w"] retail_ruc_map = dict(zip(ruc_rates["sku"], ruc_rates["retail_ruc_rate"])) wholesale_ruc_map = dict(zip(ruc_rates["sku"], ruc_rates["wholesale_ruc_rate"])) # ── ROI overlays ──────────────────────────────────────────────────── # Promo, loyalty, and new-listing volumes per (sku, year, month) drive # the three new lines on the bridge: Promo ROI, Loyalty impact, and # New-listing ROI. Each is a slice of the underlying transactions # tagged via promo_policy_items, has_loyalty, or 52w lookback. # Promo overlay — B2C transactions only (per Polleo bridge spec: # wholesale has no promo campaigns). Channel-appropriate list price # (normal_retail_ppp / normal_webshop_ppp) is applied at compute time # rather than blended avg_sell_price, so promo investment reflects # the depth of discount against the catalog price not the trailing- # 12w realized price (which may include past promos). # NET-of-VAT revenue throughout (tax_base = porezna osnovica from # Rekapitulacija — VAT excluded). Fallback to total_value * 0.80 only # when tax_base is missing (legacy/incomplete rows). promo_overlay = pd.read_sql(text(""" SELECT p.sku, EXTRACT(YEAR FROM et.transaction_date)::int AS cy, EXTRACT(MONTH FROM et.transaction_date)::int AS cm, SUM(et.quantity)::float AS promo_qty, SUM(COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80))::float AS promo_rev, SUM(CASE WHEN cm.channel = 'retail' THEN et.quantity ELSE 0 END)::float AS promo_qty_retail, SUM(CASE WHEN cm.channel = 'webshop' THEN et.quantity ELSE 0 END)::float AS promo_qty_webshop FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id JOIN lookup_channel_map cm ON cm.id = et.channel_map_id JOIN promo_policy_items ppi ON ppi.product_id = et.product_id AND et.transaction_date BETWEEN ppi.valid_from AND ppi.valid_to WHERE cm.channel IN ('retail', 'webshop') -- B2C only; wholesale gets no promo treatment GROUP BY p.sku, cy, cm """), db.bind) loyalty_overlay = pd.read_sql(text(""" SELECT p.sku, EXTRACT(YEAR FROM et.transaction_date)::int AS cy, EXTRACT(MONTH FROM et.transaction_date)::int AS cm, SUM(et.quantity)::float AS loyalty_qty, SUM(COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80))::float AS loyalty_rev FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id WHERE et.has_loyalty = TRUE GROUP BY p.sku, cy, cm """), db.bind) # New-listing detection: (partner_id, product_id) pair whose first ever # transaction (in the available history) lands in the bridge period. # We use the earliest known transaction date as the "first appearance". new_listing_overlay = pd.read_sql(text(""" WITH first_seen AS ( SELECT product_id, partner_id, MIN(transaction_date) AS first_date FROM erp_transactions et JOIN lookup_channel_map cm ON cm.id = et.channel_map_id WHERE cm.channel = 'wholesale' AND et.partner_id IS NOT NULL GROUP BY product_id, partner_id ) SELECT p.sku, EXTRACT(YEAR FROM et.transaction_date)::int AS cy, EXTRACT(MONTH FROM et.transaction_date)::int AS cm, SUM(et.quantity)::float AS nl_qty, SUM(COALESCE(NULLIF(et.tax_base, 0), et.total_value * 0.80))::float AS nl_rev FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id JOIN lookup_channel_map cm ON cm.id = et.channel_map_id JOIN first_seen fs ON fs.product_id = et.product_id AND fs.partner_id = et.partner_id WHERE cm.channel = 'wholesale' AND fs.first_date = et.transaction_date -- the first-ever day of this pair GROUP BY p.sku, cy, cm """), db.bind) # Current prices/costs (used as plan if no historical snapshot exists) cost_map = {r.sku: float(r.erp_cost or 0) for r in ctx.products.itertuples() if r.erp_cost is not None and not (isinstance(r.erp_cost, float) and math.isnan(r.erp_cost))} price_map = {r.sku: float(r.avg_sell_price or 0) for r in ctx.products.itertuples() if r.avg_sell_price is not None and not (isinstance(r.avg_sell_price, float) and math.isnan(r.avg_sell_price))} # Channel-specific list prices (catalog lists, not blended realized): # - wholesale → sku_planning.vpc (manual KAM list) # - retail → erp_prices.normal_retail_ppp (mode-based list) # - webshop → erp_prices.normal_webshop_ppp # `_b2c` is a per-SKU weighted blend of retail+webshop list weighted by # the SKU's actual B2C qty mix in the period. retail_ppp_map = {r.sku: float(r.retail_ppp or 0) for r in ctx.products.itertuples() if hasattr(r, "retail_ppp") and r.retail_ppp is not None and not (isinstance(r.retail_ppp, float) and math.isnan(r.retail_ppp))} webshop_ppp_map = {r.sku: float(r.webshop_ppp or 0) for r in ctx.products.itertuples() if hasattr(r, "webshop_ppp") and r.webshop_ppp is not None and not (isinstance(r.webshop_ppp, float) and math.isnan(r.webshop_ppp))} vpc_map = {r.sku: float(r.vpc or 0) for r in ctx.products.itertuples() if hasattr(r, "vpc") and r.vpc is not None and not (isinstance(r.vpc, float) and math.isnan(r.vpc))} prod_meta = ctx.products[["sku", "tier", "category"]].set_index("sku") # Pre-build a sku → product_id map once (avoids quadratic indexing inside the loop) sku_to_pid = dict(zip(ctx.products["sku"], ctx.products["product_id"])) sku_avg_weekly = {sku: ctx.avg_map.get(int(pid), 0.0) for sku, pid in sku_to_pid.items()} # Pre-pivot cost history for fast (sku, year, month) lookups ch = ctx.cost_history cost_lookup: dict[tuple[str, int, int], float] = {} if not ch.empty: for r in ch.itertuples(index=False): cost_lookup[(r.sku, int(r.yr), int(r.mo))] = float(r.unit_cost) # Trailing-3-month avg of BOOKED COGS per SKU — used as plan_cost (the # supplier-cost expectation at planning time). Aligned with actual_cost # source (purchase_value/qty), so the COGS effect measures real period- # over-period cost movement instead of the systematic ~43% gap between # NabavneCijene receipts and ERP-booked COGS. booked_cost_monthly = {(r.sku, int(r.cy), int(r.cm)): (float(r.pv) / float(r.qty)) if r.qty and r.qty > 0 else None for r in txn.itertuples() if r.pv is not None and r.qty is not None and r.qty > 0} def _prior_booked_cost(sku: str, year: int, month: int) -> Optional[float]: """Avg of last up-to-3 months of BOOKED purchase_value/qty cost BEFORE (year, month). The new plan_cost source — same as actual_cost source, just shifted in time. Eliminates the NabavneCijene vs purchase_value methodology gap.""" vals: list[float] = [] y, m = year, month for _ in range(3): m -= 1 if m == 0: m = 12; y -= 1 v = booked_cost_monthly.get((sku, y, m)) if v is not None and v > 0: vals.append(v) return sum(vals) / len(vals) if vals else None def _prior_avg_cost(sku: str, year: int, month: int) -> Optional[float]: """Avg of last up-to-3 months of receipt cost BEFORE (year, month).""" vals: list[float] = [] y, m = year, month for _ in range(3): m -= 1 if m == 0: m = 12; y -= 1 v = cost_lookup.get((sku, y, m)) if v is not None and v > 0: vals.append(v) return (sum(vals) / len(vals)) if vals else None # Universe of months that have ANY actuals — clipped to start at # the month we actually started forecasting from (Apr 2026). Earlier # months have no real plan baseline so the bridge variance would be # noise against a reconstructed proxy. cutoff_key = FORECAST_START_YEAR * 100 + FORECAST_START_MONTH month_keys = sorted(k for k in actuals_w["month_key"].unique() if k >= cutoff_key) months_out: list[dict] = [] per_sku_rows: list[dict] = [] top_movers_rows: list[dict] = [] for mk in month_keys: mk = int(mk) cy, cm = mk // 100, mk % 100 label = f"{_MONTH_LABEL[cm]} {cy}" # Skip very-thin months m_actual = actuals_w[actuals_w["month_key"] == mk] if m_actual.empty: continue # Build per-SKU actual quantity for this month — sourced from # erp_transactions (m_txn) NOT v_sales_weekly_full. Two reasons: # (1) erp_transactions has raw signed qty (returns subtract); v_sales # clips negatives via GREATEST() → over-counts sales for SKUs with # returns. (2) Using erp qty makes actual_qty × (price-cost) reconcile # exactly to SUM(net_rev) − SUM(cogs) per SKU, so the headline # actual_margin equals raw ERP RUC instead of being inflated by ~30%. # SKUs in v_sales but not in m_txn (pre-2026 legacy CSV without # erp_transactions rows) drop out — fine for current months. m_txn_for_qty = txn[txn["month_key"] == mk] actual_qty = (m_txn_for_qty[["sku", "qty"]] .groupby("sku")["qty"].sum() .rename("actual_qty").reset_index()) if actual_qty.empty: # Fall back to v_sales source for very-early months where # erp_transactions has nothing (historical-only data). actual_qty = (m_actual.groupby("sku")["qty"].sum() .rename("actual_qty").reset_index()) if bt is not None: m_plan = bt[bt["month_key"] == mk] plan_qty = (m_plan.groupby("sku") .agg(plan_qty_bt=("forecast", "sum"), plan_qty_bt_ws=("forecast_ws", "sum"), plan_qty_bt_b2c=("forecast_b2c", "sum")) .reset_index()) else: plan_qty = pd.DataFrame(columns=["sku", "plan_qty_bt", "plan_qty_bt_ws", "plan_qty_bt_b2c"]) # Trailing baseline = 12w non-promo avg × observed-week-fraction # in this calendar month. Use the sum of day-fractions actually # present in the actuals (some end-of-period months are partial). # For full closed months this equals exactly days_in_month / 7. obs_days = float(m_actual["qty"].notna().sum()) # rows already exploded → 1 row per (sku, week-fragment) # Sum of fractions across distinct (year, week) → "weeks present" # in this calendar month wk_fracs = (m_actual.assign(yw=lambda d: d["year"]*100 + d["week"]) .groupby("yw")["qty"] .apply(lambda s: 1.0)) # one row per week-fragment present # actually compute properly: distinct (year, week) where this month had ≥1 day weeks_observed = m_actual[["year", "week"]].drop_duplicates().shape[0] # convert to day-fraction: count how many days of this calendar month # are covered by the observed weeks (sum of split fractions for this # month across those weeks) observed_days_in_month = 0.0 for r in m_actual[["year", "week"]].drop_duplicates().itertuples(index=False): for cy2, cm2, frac in split_iso_week_across_months(int(r.year), int(r.week)): if cy2 == cy and cm2 == cm: observed_days_in_month += frac * 7 observed_weeks_equiv = float(observed_days_in_month) / 7.0 days_total = int(days_in_calendar_month(cy, cm)) coverage_pct = round(float(observed_days_in_month) / days_total, 4) if days_total else 0.0 observed_days_in_month = float(observed_days_in_month) baseline_per_sku = { sku: sku_avg_weekly.get(sku, 0.0) * observed_weeks_equiv for sku in actual_qty["sku"] } merged = actual_qty.merge(plan_qty, on="sku", how="left") merged["baseline_qty"] = merged["sku"].map(baseline_per_sku).fillna(0.0) # Plan qty cascade: backtest → baseline (12w-non-promo × weeks) merged["plan_qty"] = merged["plan_qty_bt"].fillna(merged["baseline_qty"]) # Price + booked COGS from erp_transactions. `actual_unit_cost_booked` # is `SUM(purchase_value)/SUM(quantity)` — the ERP's actual cost-of- # sale per line ("Nabavna vrijednost €" from Rekapitulacija). This is # the authoritative COGS that hits the P&L, irrespective of when we # received the goods. NabavneCijene receipt-month avg becomes a # fallback for SKUs without transaction-level cost (rare). m_txn = txn[txn["month_key"] == mk][ ["sku", "actual_unit_price", "actual_unit_cost"] ].rename(columns={"actual_unit_cost": "actual_unit_cost_booked"}) merged = merged.merge(m_txn, on="sku", how="left") has_price = not m_txn.empty # Plan COGS — primary source is "trailing prior 3-month avg of # BOOKED COGS" (purchase_value/qty). Same source as actual_cost, # just shifted in time. Falls back to NabavneCijene receipt-month # avg, then to current Sifrarnik cost. plan_cost_booked_map = { sku: _prior_booked_cost(sku, cy, cm) for sku in merged["sku"] } plan_cost_receipt_map = { sku: _prior_avg_cost(sku, cy, cm) for sku in merged["sku"] } merged["plan_cost_booked"] = merged["sku"].map(plan_cost_booked_map) merged["plan_cost_receipt"] = merged["sku"].map(plan_cost_receipt_map) cur_cost_map = merged["sku"].map(cost_map).fillna(0.0) # Plan cost cascade: trailing booked → NabavneCijene receipts → Sifrarnik merged["plan_cost"] = ( merged["plan_cost_booked"] .fillna(merged["plan_cost_receipt"]) .fillna(cur_cost_map) ) # Actual cost cascade: booked (this month) → NabavneCijene receipts # (this month) → plan_cost (so COGS effect = 0 for SKUs without # transaction-level cost in this month). actual_cost_receipt_map = { sku: cost_lookup.get((sku, cy, cm)) for sku in merged["sku"] } merged["actual_cost_receipt"] = merged["sku"].map(actual_cost_receipt_map) merged["actual_unit_cost"] = ( merged["actual_unit_cost_booked"] .fillna(merged["actual_cost_receipt"]) .fillna(merged["plan_cost"]) ) has_cost = bool( merged["actual_unit_cost_booked"].notna().any() or merged["actual_cost_receipt"].notna().any() ) merged["plan_price"] = merged["sku"].map(price_map).fillna(0.0) merged["actual_unit_price"] = merged["actual_unit_price"].fillna(merged["plan_price"]) has_price_cost = has_price and has_cost # Volume effect = (actual_qty - plan_qty) × (plan_price - plan_cost) merged["plan_margin_per_unit"] = merged["plan_price"] - merged["plan_cost"] merged["volume_effect"] = ((merged["actual_qty"] - merged["plan_qty"]) * merged["plan_margin_per_unit"]) # Price effect = (actual_price - plan_price) × actual_qty merged["price_effect"] = ((merged["actual_unit_price"] - merged["plan_price"]) * merged["actual_qty"]) # COGS effect = -(actual_cost - plan_cost) × actual_qty merged["cogs_effect"] = -((merged["actual_unit_cost"] - merged["plan_cost"]) * merged["actual_qty"]) # Plan margin total + per-SKU frozen plan prices/cost — sourced from # monthly_plan_snapshots when locked, else computed live. The snapshot # is the canonical "plan" once user locks it for the S&OP cycle. snapshot_row = db.execute(text( "SELECT total_ruc_eur, total_ruc_ws, total_ruc_b2c, " "total_revenue_eur, total_qty_ws, total_qty_b2c, " "plan_wholesale_rebate_eur, per_sku_data " "FROM monthly_plan_snapshots WHERE month_key = :mk" ), {"mk": int(mk)}).mappings().first() plan_source = "snapshot" if snapshot_row else "live" # Per-SKU frozen plan prices/cost lookup (from snapshot.per_sku_data). # Bridge uses these instead of live erp_prices / sku_planning so the # decomposition stays anchored to what was locked at S&OP time, not # whatever ERP has at the moment bridge runs. snap_plan_price_ws_map: dict[str, float] = {} snap_plan_price_b2c_map: dict[str, float] = {} snap_plan_cost_map: dict[str, float] = {} snap_vpc_map: dict[str, float] = {} if snapshot_row and snapshot_row.get("per_sku_data"): for entry in snapshot_row["per_sku_data"]: sku = entry.get("sku") if not sku: continue pp_ws = entry.get("plan_price_ws") pp_b2c = entry.get("plan_price_b2c") pc = entry.get("plan_cost") vpc = entry.get("vpc") if pp_ws and pp_ws > 0: snap_plan_price_ws_map[sku] = float(pp_ws) if pp_b2c and pp_b2c > 0: snap_plan_price_b2c_map[sku] = float(pp_b2c) if pc and pc > 0: snap_plan_cost_map[sku] = float(pc) if vpc and vpc > 0: snap_vpc_map[sku] = float(vpc) merged["retail_ruc_rate"] = merged["sku"].map(retail_ruc_map) merged["ws_ruc_rate"] = merged["sku"].map(wholesale_ruc_map) # Per-SKU plan margin contribution from each channel plan_qty_ws_per_sku = merged["plan_qty_bt_ws"].fillna(0) plan_qty_b2c_per_sku = merged["plan_qty_bt_b2c"].fillna(0) # Use RUC rate per channel where available; fall back to per-SKU # (avg_sell_price - plan_cost) for SKUs without recent transactions # in that channel (so we don't drop them from plan_margin entirely). fallback_margin = merged["plan_margin_per_unit"] ws_plan_per_sku = plan_qty_ws_per_sku * merged["ws_ruc_rate"].fillna(fallback_margin) b2c_plan_per_sku = plan_qty_b2c_per_sku * merged["retail_ruc_rate"].fillna(fallback_margin) plan_margin_total_live = float(ws_plan_per_sku.sum() + b2c_plan_per_sku.sum()) # If a locked snapshot exists for this month, use it as the authoritative # plan number. The live computation above is what would have been locked # at the moment of the snapshot — but the snapshot is immutable, so it's # the canonical "plan" once the user locks it. Falls back to live when # no snapshot exists yet (months you haven't locked). if snapshot_row: plan_margin_total = float(snapshot_row["total_ruc_eur"]) else: plan_margin_total = plan_margin_total_live actual_margin_total = float((merged["actual_qty"] * (merged["actual_unit_price"] - merged["actual_unit_cost"])).sum()) total_variance = actual_margin_total - plan_margin_total pri = float(merged["price_effect"].sum()) cog = float(merged["cogs_effect"].sum()) # AGGREGATE-LEVEL volume + mix decomposition. # Volume effect at constant mix: (Σactual_qty − Σplan_qty) × plan_avg_margin # Mix effect: how SKU mix shifted at plan prices/costs — the residual after V+P+C. # This makes mix meaningful (≠ 0). Per-SKU V/P/C still sum to total variance # and remain useful for top-movers ranking, but the headline waterfall uses # the aggregate decomposition. total_plan_qty = float(merged["plan_qty"].sum()) total_actual_qty = float(merged["actual_qty"].sum()) plan_avg_margin = (plan_margin_total / total_plan_qty) if total_plan_qty else 0.0 vol = (total_actual_qty - total_plan_qty) * plan_avg_margin mix = total_variance - vol - pri - cog # ── CHANNEL-SPLIT bridge (Polleo v2, May 2026) ────────────────── # Wholesale uses VPC as plan price; B2C uses normal_*_ppp blended # by SKU's actual retail-vs-webshop qty mix. plan_qty per channel # comes from backtest_results.forecast_wholesale / forecast_retail # (engine already produced this split). channel_mix = residual # after wholesale + B2C decomposition. m_txn_full = txn[txn["month_key"] == mk] # Per-SKU wholesale actuals + plan ws = merged[["sku", "plan_qty", "plan_qty_bt_ws"]].copy() b2c = merged[["sku", "plan_qty", "plan_qty_bt_b2c"]].copy() # WS plan qty: prefer the engine's forecast_wholesale; fall back # to total plan_qty × historical WS share (computed from txn for # this month, falls to 0 if no WS sales). ws_txn = m_txn_full[["sku", "qty_ws", "rev_net_ws", "pv_ws", "actual_unit_price_ws", "actual_unit_cost_ws"]] b2c_txn = m_txn_full[["sku", "qty_b2c", "rev_net_b2c", "pv_b2c", "actual_unit_price_b2c", "actual_unit_cost_b2c", "qty_retail", "qty_webshop"]] ws = ws.merge(ws_txn, on="sku", how="left") b2c = b2c.merge(b2c_txn, on="sku", how="left") # Use backtest's per-channel split when available, else fall back # to total_plan × current_channel_share within this month's txn. ws["plan_qty_ws"] = ws["plan_qty_bt_ws"].fillna(0) b2c["plan_qty_b2c"] = b2c["plan_qty_bt_b2c"].fillna(0) # Channel-specific plan prices — frozen from snapshot when locked, # else fall back to realised-rate trailing-4w (matches snapshot logic # so live months reconcile to what a snapshot would have captured). # # plan_price_ws = realised wholesale rate (post-rebate, NET). # This is the "expected price KAMs will achieve" — # rebate already absorbed in the baseline. # plan_price_b2c = realised retail+webshop NET blended rate. # We no longer use VPC list / retail_ppp catalog directly as plan # baseline — that comparison conflates contractual rebate (a known # recurring effect) with anomalous price slippage. VPC is kept as # frozen reference for the Wholesale Rebate display line only. # Channel-specific plan prices: snapshot first; for SKUs missing # from snapshot (non-planned that nevertheless transacted), fall back # to catalog × 0.80 (B2C, NET of HR 25% VAT) or VPC (WS, already NET # for B2B). Filling with 0 would inflate price_effect by attributing # full actual revenue to "price slippage" for these SKUs. catalog_ws_fallback = ws["sku"].map(vpc_map).fillna(0.0) # B2C catalog: retail_ppp if present, else webshop_ppp, then NET via 0.80 b2c_retail_ppp = b2c["sku"].map(retail_ppp_map).fillna(0.0) b2c_webshop_ppp = b2c["sku"].map(webshop_ppp_map).fillna(0.0) catalog_b2c_gross = b2c_retail_ppp.where(b2c_retail_ppp > 0, b2c_webshop_ppp) catalog_b2c_fallback = catalog_b2c_gross * 0.80 if snap_plan_price_ws_map: ws["plan_price_ws"] = ws["sku"].map(snap_plan_price_ws_map).fillna( catalog_ws_fallback) else: ws["plan_price_ws"] = catalog_ws_fallback if snap_plan_price_b2c_map: b2c["plan_price_b2c"] = b2c["sku"].map(snap_plan_price_b2c_map).fillna( catalog_b2c_fallback) else: b2c["plan_price_b2c"] = catalog_b2c_fallback # Plan cost per SKU — frozen from snapshot when available, else live # trailing 3-month avg. if snap_plan_cost_map: ws["plan_cost"] = ws["sku"].map(snap_plan_cost_map).fillna( pd.Series(merged.set_index("sku")["plan_cost"].reindex(ws["sku"]).values, index=ws.index)) b2c["plan_cost"] = b2c["sku"].map(snap_plan_cost_map).fillna( pd.Series(merged.set_index("sku")["plan_cost"].reindex(b2c["sku"]).values, index=b2c.index)) else: all_chan_plan_cost = merged.set_index("sku")["plan_cost"] ws["plan_cost"] = pd.Series(all_chan_plan_cost.reindex(ws["sku"]).values, index=ws.index) b2c["plan_cost"] = pd.Series(all_chan_plan_cost.reindex(b2c["sku"]).values, index=b2c.index) # Channel-specific actual cost: use channel-segregated booked # purchase_value; fall back to all-channel booked cost. all_chan_cost = merged.set_index("sku")["actual_unit_cost"] ws_fallback_cost = pd.Series(all_chan_cost.reindex(ws["sku"]).values, index=ws.index) b2c_fallback_cost = pd.Series(all_chan_cost.reindex(b2c["sku"]).values, index=b2c.index) ws["actual_unit_cost_ws"] = ws["actual_unit_cost_ws"].fillna(ws_fallback_cost) b2c["actual_unit_cost_b2c"] = b2c["actual_unit_cost_b2c"].fillna(b2c_fallback_cost) # Per-channel margins — May 2026 refactor: align with Revenue Forecast. # Plan margin per unit = realized RUC rate from last 4 weeks of # erp_transactions (channel-specific). Falls back to (list price − # plan cost) when no recent transactions for that channel. ws["ws_ruc_rate"] = ws["sku"].map(wholesale_ruc_map) b2c["retail_ruc_rate"] = b2c["sku"].map(retail_ruc_map) ws["plan_margin_per_u"] = ws["ws_ruc_rate"].fillna( ws["plan_price_ws"] - ws["plan_cost"] ) b2c["plan_margin_per_u"] = b2c["retail_ruc_rate"].fillna( b2c["plan_price_b2c"] - b2c["plan_cost"] ) ws["actual_qty_ws"] = ws["qty_ws"].fillna(0) b2c["actual_qty_b2c"] = b2c["qty_b2c"].fillna(0) ws["actual_unit_price_ws"] = ws["actual_unit_price_ws"].fillna(ws["plan_price_ws"]) b2c["actual_unit_price_b2c"] = b2c["actual_unit_price_b2c"].fillna(b2c["plan_price_b2c"]) # Per-SKU per-channel effect components ws["volume_eff_ws"] = (ws["actual_qty_ws"] - ws["plan_qty_ws"]) * ws["plan_margin_per_u"] ws["price_eff_ws"] = (ws["actual_unit_price_ws"] - ws["plan_price_ws"]) * ws["actual_qty_ws"] ws["cogs_eff_ws"] = -((ws["actual_unit_cost_ws"] - ws["plan_cost"]) * ws["actual_qty_ws"]) b2c["volume_eff_b2c"] = (b2c["actual_qty_b2c"] - b2c["plan_qty_b2c"]) * b2c["plan_margin_per_u"] b2c["price_eff_b2c"] = (b2c["actual_unit_price_b2c"] - b2c["plan_price_b2c"]) * b2c["actual_qty_b2c"] b2c["cogs_eff_b2c"] = -((b2c["actual_unit_cost_b2c"] - b2c["plan_cost"]) * b2c["actual_qty_b2c"]) # Aggregate per-channel margins plan_margin_ws_live = float((ws["plan_qty_ws"] * ws["plan_margin_per_u"]).sum()) plan_margin_b2c_live = float((b2c["plan_qty_b2c"] * b2c["plan_margin_per_u"]).sum()) # Prefer snapshot's per-channel plan margin if available — keeps the # channel-split decomposition consistent with the locked snapshot. if snapshot_row: plan_margin_ws = float(snapshot_row["total_ruc_ws"] or plan_margin_ws_live) plan_margin_b2c = float(snapshot_row["total_ruc_b2c"] or plan_margin_b2c_live) else: plan_margin_ws = plan_margin_ws_live plan_margin_b2c = plan_margin_b2c_live actual_margin_ws = float((ws["actual_qty_ws"] * (ws["actual_unit_price_ws"] - ws["actual_unit_cost_ws"])).sum()) actual_margin_b2c = float((b2c["actual_qty_b2c"] * (b2c["actual_unit_price_b2c"] - b2c["actual_unit_cost_b2c"])).sum()) total_variance_ws = actual_margin_ws - plan_margin_ws total_variance_b2c = actual_margin_b2c - plan_margin_b2c # Per-channel aggregate V/P/M/COGS using "constant-mix volume" approach ws_pri = float(ws["price_eff_ws"].sum()) ws_cog = float(ws["cogs_eff_ws"].sum()) ws_total_plan_qty = float(ws["plan_qty_ws"].sum()) ws_total_actual_qty = float(ws["actual_qty_ws"].sum()) ws_plan_avg_margin = (plan_margin_ws / ws_total_plan_qty) if ws_total_plan_qty else 0.0 ws_vol = (ws_total_actual_qty - ws_total_plan_qty) * ws_plan_avg_margin ws_mix = total_variance_ws - ws_vol - ws_pri - ws_cog b2c_pri = float(b2c["price_eff_b2c"].sum()) b2c_cog = float(b2c["cogs_eff_b2c"].sum()) b2c_total_plan_qty = float(b2c["plan_qty_b2c"].sum()) b2c_total_actual_qty = float(b2c["actual_qty_b2c"].sum()) b2c_plan_avg_margin = (plan_margin_b2c / b2c_total_plan_qty) if b2c_total_plan_qty else 0.0 b2c_vol = (b2c_total_actual_qty - b2c_total_plan_qty) * b2c_plan_avg_margin b2c_mix = total_variance_b2c - b2c_vol - b2c_pri - b2c_cog # Channel mix = residual after channel decomposition. Captures the # share-shift between wholesale and B2C and rolling-up effects of # SKUs that appear in one channel only. channel_mix = total_variance - total_variance_ws - total_variance_b2c # ── HEADLINE WATERFALL — channel-first rebuild ───────────────────── # Replace the all-channel headline (which compared blended actual # against `avg_sell_price` and mixed mixed-channel-mix into price) # with a clean sum of per-channel decomposition + channel_mix. # Conservation: plan + vol + pri + cog + mix = actual. vol = ws_vol + b2c_vol pri = ws_pri + b2c_pri cog = ws_cog + b2c_cog mix = ws_mix + b2c_mix + channel_mix # Override per-SKU price_effect with channel-aware sum so category # rollups and top-movers reconcile with the new headline. Volume/ # COGS at SKU level are intentionally left on the all-channel path # for now — the previous methodology was sound there. ws_pri_per_sku = ws.set_index("sku")["price_eff_ws"].to_dict() b2c_pri_per_sku = b2c.set_index("sku")["price_eff_b2c"].to_dict() merged["price_effect"] = ( merged["sku"].map(ws_pri_per_sku).fillna(0.0) + merged["sku"].map(b2c_pri_per_sku).fillna(0.0) ) # ── WHOLESALE REBATE — display line (not in math reconciliation) ── # actual rebate = (VPC − actual_price_ws) × actual_qty_ws for SKUs # with VPC>0 and actual_price < VPC. plan rebate is frozen in # snapshot. Deviation is already absorbed in price_eff_ws because # plan_price_ws is post-expected-rebate. ws_with_vpc = ws.copy() # Use snapshot's frozen VPC if available, else live vpc_map if snap_vpc_map: ws_with_vpc["vpc"] = ws_with_vpc["sku"].map(snap_vpc_map).fillna( ws_with_vpc["sku"].map(vpc_map).fillna(0.0)) else: ws_with_vpc["vpc"] = ws_with_vpc["sku"].map(vpc_map).fillna(0.0) ws_with_vpc["rebate_per_u"] = (ws_with_vpc["vpc"] - ws_with_vpc["actual_unit_price_ws"]).clip(lower=0) # Only count for SKUs that have a VPC reference + actual WS sales ws_with_vpc["rebate_eur"] = (ws_with_vpc["rebate_per_u"] * ws_with_vpc["actual_qty_ws"]).fillna(0) ws_with_vpc.loc[ws_with_vpc["vpc"] <= 0, "rebate_eur"] = 0 wholesale_rebate_actual = float(ws_with_vpc["rebate_eur"].sum()) wholesale_rebate_plan = float(snapshot_row["plan_wholesale_rebate_eur"] or 0) \ if snapshot_row and snapshot_row.get("plan_wholesale_rebate_eur") is not None else 0.0 # Tier reflects which effects are based on real period-specific data: # FULL — price (erp_transactions) + cost (NabavneCijene) # COST_ONLY — cost only → V + M + COGS real, Price=0 by construction # VOLUME_ONLY — neither; only Volume effect is meaningful if has_price and has_cost: tier = "FULL" elif has_cost: tier = "COST_ONLY" else: tier = "VOLUME_ONLY" # Attach product meta merged = merged.join(prod_meta, on="sku") # Top 10 positive + top 10 negative per effect for effect in ("volume", "price", "mix", "cogs"): col = f"{effect}_effect" if effect != "mix" else "volume_effect" # mix is residual, no per-sku if effect == "mix": continue # skip — mix is residual at category level only sorted_pos = merged.nlargest(10, col) sorted_neg = merged.nsmallest(10, col) for direction, sub in (("positive", sorted_pos), ("negative", sorted_neg)): cum = 0.0 tot_abs = float(abs(sub[col]).sum()) or 1.0 for i, (_, r) in enumerate(sub.iterrows(), start=1): cum += float(abs(r[col])) top_movers_rows.append({ "month": label, "effect_type": effect, "direction": direction, "rank": i, "sku": r["sku"], "tier": (r.get("tier") or "") if pd.notna(r.get("tier", None)) else "", "category": (r.get("category") or "") if pd.notna(r.get("category", None)) else "", "effect_eur": round(float(r[col]), 2), "cumulative_pct": round(cum / tot_abs, 4), }) # Per-SKU rows for this month for _, r in merged.iterrows(): per_sku_rows.append({ "month": label, "month_key": int(mk), "sku": r["sku"], "tier": (r.get("tier") or "") if pd.notna(r.get("tier", None)) else "", "category": (r.get("category") or "OTHER") if pd.notna(r.get("category", None)) else "OTHER", "plan_qty": round(float(r["plan_qty"]), 2), "actual_qty": round(float(r["actual_qty"]), 2), "qty_variance": round(float(r["actual_qty"] - r["plan_qty"]), 2), "plan_price": round(float(r["plan_price"]), 4), "actual_price":round(float(r["actual_unit_price"]), 4), "plan_cost": round(float(r["plan_cost"]), 4), "actual_cost": round(float(r["actual_unit_cost"]), 4), "plan_margin_sku": round(float(r["plan_qty"] * r["plan_margin_per_unit"]), 2), "actual_margin_sku": round(float(r["actual_qty"] * (r["actual_unit_price"] - r["actual_unit_cost"])), 2), "volume_effect": round(float(r["volume_effect"]), 2), "price_effect": round(float(r["price_effect"]), 2), "cogs_effect": round(float(r["cogs_effect"]), 2), }) # By-category rollup by_cat = (merged.groupby("category").agg( volume_effect=("volume_effect", "sum"), price_effect= ("price_effect", "sum"), cogs_effect= ("cogs_effect", "sum"), plan_margin = ("plan_margin_per_unit", lambda s: float((merged.loc[s.index, "plan_qty"] * merged.loc[s.index, "plan_margin_per_unit"]).sum())), actual_margin=("actual_unit_price", lambda s: float((merged.loc[s.index, "actual_qty"] * (merged.loc[s.index, "actual_unit_price"] - merged.loc[s.index, "actual_unit_cost"])).sum())), ).reset_index()) by_cat["total_variance"] = by_cat["actual_margin"] - by_cat["plan_margin"] by_cat["mix_effect"] = (by_cat["total_variance"] - by_cat["volume_effect"] - by_cat["price_effect"] - by_cat["cogs_effect"]) # Top contributing SKUs (positive + negative) for the month top_pos = merged.nlargest(5, "volume_effect")[["sku", "category", "volume_effect"]].to_dict("records") top_neg = merged.nsmallest(5, "volume_effect")[["sku", "category", "volume_effect"]].to_dict("records") # ── Per-month ROI overlays (Promo / Loyalty / New-listing) ────── # These slice the actual transactions in this month and value the # qty at the same plan_avg_margin used for the main bridge — so # the numbers compose with the existing volume/price/mix lines. m_promo = promo_overlay[ (promo_overlay["cy"] == cy) & (promo_overlay["cm"] == cm) ] m_loyalty = loyalty_overlay[ (loyalty_overlay["cy"] == cy) & (loyalty_overlay["cm"] == cm) ] m_newlist = new_listing_overlay[ (new_listing_overlay["cy"] == cy) & (new_listing_overlay["cm"] == cm) ] def _slice_eur(slice_df: pd.DataFrame, qty_col: str, rev_col: str, list_per_unit_eur: Optional[pd.Series] = None) -> dict: """Returns volume_effect, investment_effect, and ROI for a slice. `list_per_unit_eur` (if provided) is the channel-appropriate catalog list price to use as the "would-have-charged" baseline for the investment line. Falls back to avg_sell_price for backwards-compat with loyalty / new-listing overlays. """ if slice_df.empty: return {"qty": 0.0, "revenue_eur": 0.0, "volume_effect": 0.0, "investment_effect": 0.0, "roi": 0.0} slice_with_margin = slice_df.merge( merged[["sku", "plan_margin_per_unit"]], on="sku", how="left" ) if list_per_unit_eur is not None: slice_with_margin["list_price"] = list_per_unit_eur.values else: # Fallback for loyalty / new-listing (uses blended avg) fallback_map = { r.sku: float(r.avg_sell_price or 0) for r in ctx.products.itertuples() if r.avg_sell_price is not None } slice_with_margin["list_price"] = ( slice_with_margin["sku"].map(fallback_map).fillna(0.0) ) qty = float(slice_with_margin[qty_col].sum()) rev = float(slice_with_margin[rev_col].sum()) vol_eff = float( (slice_with_margin[qty_col] * slice_with_margin["plan_margin_per_unit"].fillna(0)).sum() ) list_val = float( (slice_with_margin[qty_col] * slice_with_margin["list_price"]).sum() ) investment = -(list_val - rev) return { "qty": qty, "revenue_eur": rev, "volume_effect": vol_eff, "investment_effect": investment, "roi": vol_eff + investment, } # Promo slice uses channel-appropriate catalog list price: # weighted blend of normal_retail_ppp × promo_qty_retail and # normal_webshop_ppp × promo_qty_webshop per SKU. if not m_promo.empty: mp = m_promo.copy() mp["retail_ppp"] = mp["sku"].map(retail_ppp_map).fillna(0.0) mp["webshop_ppp"] = mp["sku"].map(webshop_ppp_map).fillna(0.0) mp["promo_qty_retail"] = mp["promo_qty_retail"].fillna(0) mp["promo_qty_webshop"] = mp["promo_qty_webshop"].fillna(0) mp_total_qty = (mp["promo_qty_retail"] + mp["promo_qty_webshop"]) mp["list_per_u"] = ( (mp["promo_qty_retail"] * mp["retail_ppp"] + mp["promo_qty_webshop"] * mp["webshop_ppp"]) / mp_total_qty.replace(0, pd.NA) ).fillna(mp["retail_ppp"]) promo_metrics = _slice_eur(mp, "promo_qty", "promo_rev", list_per_unit_eur=mp["list_per_u"]) else: promo_metrics = _slice_eur(m_promo, "promo_qty", "promo_rev") loyalty_metrics = _slice_eur(m_loyalty, "loyalty_qty", "loyalty_rev") newlist_metrics = _slice_eur(m_newlist, "nl_qty", "nl_rev") # Loyalty doesn't have a "volume_effect" line — it's pure discount cost. loyalty_impact = loyalty_metrics["investment_effect"] months_out.append({ "month": label, "month_key": int(mk), "tier": tier, "n_skus": int(len(merged)), "days_observed": round(observed_days_in_month, 1), "days_in_month": days_total, "coverage_pct": coverage_pct, "plan_margin_total": round(plan_margin_total, 0), "plan_source": plan_source, # 'snapshot' | 'live' "actual_margin_total": round(actual_margin_total, 0), "total_variance": round(total_variance, 0), "volume_effect": round(vol, 0), "price_effect": round(pri, 0), "mix_effect": round(mix, 0), "cogs_effect": round(cog, 0), # ── Channel-split bridge (Polleo v2) ──────────────────────── "wholesale": { "plan_qty": round(ws_total_plan_qty, 0), "actual_qty": round(ws_total_actual_qty, 0), "plan_margin": round(plan_margin_ws, 0), "actual_margin": round(actual_margin_ws, 0), "total_variance": round(total_variance_ws, 0), "volume_effect": round(ws_vol, 0), "price_effect": round(ws_pri, 0), "mix_effect": round(ws_mix, 0), "cogs_effect": round(ws_cog, 0), }, "b2c": { "plan_qty": round(b2c_total_plan_qty, 0), "actual_qty": round(b2c_total_actual_qty, 0), "plan_margin": round(plan_margin_b2c, 0), "actual_margin": round(actual_margin_b2c, 0), "total_variance": round(total_variance_b2c, 0), "volume_effect": round(b2c_vol, 0), "price_effect": round(b2c_pri, 0), "mix_effect": round(b2c_mix, 0), "cogs_effect": round(b2c_cog, 0), }, "channel_mix_effect": round(channel_mix, 0), # ── Wholesale Rebate (Polleo v3, May 2026) ────────────────── # Display-only: actual rebate KAMs gave this month vs plan # rebate frozen in snapshot. Deviation from plan is already # absorbed in price_effect — this surface is for CFO visibility. "wholesale_rebate_actual": round(wholesale_rebate_actual, 0), "wholesale_rebate_plan": round(wholesale_rebate_plan, 0), "wholesale_rebate_delta": round(wholesale_rebate_actual - wholesale_rebate_plan, 0), # ── ROI overlays (new May 2026) ───────────────────────────── "promo_qty": round(promo_metrics["qty"], 0), "promo_revenue_eur": round(promo_metrics["revenue_eur"], 0), "promo_uplift": round(promo_metrics["volume_effect"], 0), "promo_investment": round(promo_metrics["investment_effect"], 0), "promo_roi": round(promo_metrics["roi"], 0), "loyalty_qty": round(loyalty_metrics["qty"], 0), "loyalty_revenue_eur": round(loyalty_metrics["revenue_eur"], 0), "loyalty_impact": round(loyalty_impact, 0), "new_listing_qty": round(newlist_metrics["qty"], 0), "new_listing_revenue_eur": round(newlist_metrics["revenue_eur"], 0), "new_listing_volume": round(newlist_metrics["volume_effect"], 0), "new_listing_investment": round(newlist_metrics["investment_effect"], 0), "new_listing_roi": round(newlist_metrics["roi"], 0), "by_category": [ {"category": r["category"], "volume_effect": round(float(r["volume_effect"]), 0), "price_effect": round(float(r["price_effect"]), 0), "mix_effect": round(float(r["mix_effect"]), 0), "cogs_effect": round(float(r["cogs_effect"]), 0), "total_variance":round(float(r["total_variance"]),0)} for _, r in by_cat.iterrows() ], "top_positive_skus": top_pos, "top_negative_skus": top_neg, }) # YTD rollup if months_out: ytd_year = max(m["month_key"] // 100 for m in months_out) ytd_months = [m for m in months_out if m["month_key"] // 100 == ytd_year] ytd = { "year": ytd_year, "n_months": len(ytd_months), "plan_margin_total": sum(m["plan_margin_total"] for m in ytd_months), "actual_margin_total": sum(m["actual_margin_total"] for m in ytd_months), "volume_effect": sum(m["volume_effect"] for m in ytd_months), "price_effect": sum(m["price_effect"] for m in ytd_months), "mix_effect": sum(m["mix_effect"] for m in ytd_months), "cogs_effect": sum(m["cogs_effect"] for m in ytd_months), "trend": [ {"month": m["month"], "volume": m["volume_effect"], "price": m["price_effect"], "mix": m["mix_effect"], "cogs": m["cogs_effect"]} for m in ytd_months ], } else: ytd = None return { "available": bool(months_out), "months": months_out, "ytd": ytd, "per_sku_rows": per_sku_rows, "top_movers_rows": top_movers_rows, } # Keep the old single-window bridge for now — it's still useful for a # quick "last 8 weeks" snapshot. The monthly bridge above is the main one. def report_bridge(ctx: FinanceContext, db: Session) -> dict: bt_path = Path(__file__).resolve().parents[2] / "data" / "backtest_fa.csv" if not bt_path.exists(): return {"available": False, "type": "missing", "df": pd.DataFrame(), "summary": {}, "waterfall": [], "by_category": pd.DataFrame()} bt = pd.read_csv(bt_path) recent_yws = (bt[(bt["forecast"].notna()) & (bt["actual"].notna())] .assign(yw=lambda d: d["year"]*100 + d["week"]) .groupby("yw").size().sort_index().tail(8).index.tolist()) if not recent_yws: return {"available": False, "type": "no_overlap", "df": pd.DataFrame(), "summary": {}, "waterfall": [], "by_category": pd.DataFrame()} bt = bt[(bt["year"] * 100 + bt["week"]).isin(recent_yws)] min_yw, max_yw = min(recent_yws), max(recent_yws) txn = pd.read_sql(text(""" SELECT p.sku, EXTRACT(ISOYEAR FROM et.transaction_date)::int AS year, EXTRACT(WEEK FROM et.transaction_date)::int AS week, SUM(et.quantity)::float AS qty, SUM(et.total_value)::float AS rev, SUM(et.purchase_value)::float AS pv, SUM(et.tax_base)::float AS tax_base FROM erp_transactions et JOIN dim_products p ON p.id = et.product_id WHERE EXTRACT(ISOYEAR FROM et.transaction_date)::int * 100 + EXTRACT(WEEK FROM et.transaction_date)::int BETWEEN :min_yw AND :max_yw GROUP BY p.sku, year, week """), db.bind, params={"min_yw": min_yw, "max_yw": max_yw}) txn = txn.assign(yw=lambda d: d["year"]*100 + d["week"]) txn = txn[txn["yw"].isin(recent_yws)] cost_map = {r.sku: float(r.erp_cost or 0) for r in ctx.products.itertuples() if r.erp_cost is not None and not (isinstance(r.erp_cost, float) and math.isnan(r.erp_cost))} price_map = {r.sku: float(r.avg_sell_price or 0) for r in ctx.products.itertuples() if r.avg_sell_price is not None and not (isinstance(r.avg_sell_price, float) and math.isnan(r.avg_sell_price))} merged = bt.merge(txn[["sku","year","week","qty","rev","pv","tax_base"]], on=["sku","year","week"], how="left") merged["actual_qty"] = merged["qty"].fillna(merged["actual"]) merged["plan_qty"] = merged["forecast"] merged["actual_revenue"] = merged["tax_base"] merged["actual_unit_price"] = merged["actual_revenue"] / merged["actual_qty"] merged["actual_unit_cost"] = merged["pv"] / merged["actual_qty"] merged["plan_price"] = merged["sku"].map(price_map).fillna(0.0) merged["plan_cost"] = merged["sku"].map(cost_map).fillna(0.0) merged["plan_revenue"] = merged["plan_qty"] * merged["plan_price"] merged["plan_cogs"] = merged["plan_qty"] * merged["plan_cost"] merged["actual_cogs"] = merged["actual_qty"] * merged["actual_unit_cost"].fillna(merged["plan_cost"]) merged["volume_effect_eur"] = (merged["actual_qty"] - merged["plan_qty"]) * merged["plan_price"] merged["price_effect_eur"] = (merged["actual_unit_price"].fillna(merged["plan_price"]) - merged["plan_price"]) * merged["actual_qty"] merged["mix_effect_eur"] = (merged["actual_revenue"].fillna(0) - merged["plan_revenue"] - merged["volume_effect_eur"] - merged["price_effect_eur"]) merged["cogs_effect_eur"] = -(merged["actual_cogs"] - merged["plan_cogs"]) merged["margin_effect_total_eur"] = (merged["volume_effect_eur"] + merged["price_effect_eur"] + merged["mix_effect_eur"] + merged["cogs_effect_eur"]) prod_idx = ctx.products[["sku","tier","category"]].set_index("sku") merged = merged.join(prod_idx, on="sku") merged["period"] = [f"{int(y)}W{int(w):02d}" for y, w in zip(merged["year"], merged["week"])] out_cols = ["sku","tier","category","period", "plan_qty","actual_qty", "plan_price","actual_unit_price", "plan_revenue","actual_revenue", "volume_effect_eur","price_effect_eur","mix_effect_eur", "plan_cogs","actual_cogs","cogs_effect_eur", "margin_effect_total_eur"] df = merged[out_cols].copy() df = df.round({c: 2 for c in out_cols if c not in ("sku","tier","category","period")}) by_cat = (merged.groupby("category") .agg(volume_effect=("volume_effect_eur","sum"), price_effect=("price_effect_eur","sum"), mix_effect=("mix_effect_eur","sum"), cogs_effect=("cogs_effect_eur","sum"), total_margin_effect=("margin_effect_total_eur","sum")) .reset_index().round(0) .sort_values("total_margin_effect")) plan_margin = float(merged["plan_revenue"].sum() - merged["plan_cogs"].sum()) actual_margin = float(merged["actual_revenue"].fillna(0).sum() - merged["actual_cogs"].sum()) vol = float(merged["volume_effect_eur"].sum()) pri = float(merged["price_effect_eur"].sum()) mix = float(merged["mix_effect_eur"].sum()) cog = float(merged["cogs_effect_eur"].sum()) waterfall = [ {"label": "Plan margin", "value": round(plan_margin, 0), "type": "base"}, {"label": "Volume", "value": round(vol, 0), "type": "increase" if vol >= 0 else "decrease"}, {"label": "Price", "value": round(pri, 0), "type": "increase" if pri >= 0 else "decrease"}, {"label": "Mix", "value": round(mix, 0), "type": "increase" if mix >= 0 else "decrease"}, {"label": "COGS", "value": round(cog, 0), "type": "increase" if cog >= 0 else "decrease"}, {"label": "Actual margin", "value": round(actual_margin, 0), "type": "total"}, ] return { "available": True, "type": "full", "df": df, "by_category": by_cat, "waterfall": waterfall, "summary": { "window_weeks": [int(yw) for yw in recent_yws], "n_skus": int(merged["sku"].nunique()), "plan_margin": plan_margin, "actual_margin": actual_margin, "volume_effect": vol, "price_effect": pri, "mix_effect": mix, "cogs_effect": cog, "total_variance": vol + pri + mix + cog, }, } # ───────────────────────────────────────────────────────────────────────── # Dashboard rollup # ───────────────────────────────────────────────────────────────────────── def dashboard(ctx: FinanceContext, db: Session) -> dict: ls = report_lost_sales(ctx) lc = report_locked_cash(ctx) cr = report_contest_risk(ctx) sm = report_slow_movers(ctx) br = report_bridge(ctx, db) return { "anchor_year_week": f"{ctx.cur_year}W{ctx.cur_week:02d}", "horizon_weeks": HORIZON_WEEKS, "n_products_in_scope": int(len(ctx.products)), "n_dormant_excluded": int(len(ctx.dormant_pids)), "lost_sales": ls["summary"], "locked_cash": lc["summary"], "contest_risk": cr["summary"], "slow_movers": sm["summary"], "bridge": br["summary"] if br["available"] else {"available": False}, "top5_lost_sales": ls["top5"], "top10_contest": cr["top10_at_risk"], } # ───────────────────────────────────────────────────────────────────────── # Excel export — one workbook per report # ───────────────────────────────────────────────────────────────────────── def _bytes_from_excel(sheets: dict[str, pd.DataFrame], title: str) -> bytes: buf = io.BytesIO() with pd.ExcelWriter(buf, engine="openpyxl") as xl: for name, df in sheets.items(): if df is None or df.empty: continue df.to_excel(xl, sheet_name=name[:31], index=False) ws = xl.sheets[name[:31]] # Freeze header + autosize columns ws.freeze_panes = "A2" for col_idx, col_name in enumerate(df.columns, start=1): width = max(len(str(col_name)), int(df[col_name].astype(str).str.len().clip(upper=40).max() or 10)) ws.column_dimensions[ws.cell(1, col_idx).column_letter].width = min(width + 2, 42) return buf.getvalue() def export_lost_sales_xlsx(result: dict) -> bytes: return _bytes_from_excel({ "Per SKU": result["df"], "By Category": result["by_category"], "By Tier": result["by_tier"], }, "Lost sales") def export_locked_cash_xlsx(result: dict) -> bytes: return _bytes_from_excel({ "Per SKU": result["df"], "By Category": result["by_category"], "By Supplier": result["by_supplier"], }, "Locked cash") def export_contest_risk_xlsx(result: dict) -> bytes: return _bytes_from_excel({"Per SKU": result["df"]}, "Contest July risk") def export_slow_movers_xlsx(result: dict) -> bytes: return _bytes_from_excel({ "Per SKU": result["df"], "By Category": result["by_category"], "By Supplier": result["by_supplier"], }, "Slow movers") def export_bridge_xlsx(result: dict) -> bytes: if not result.get("available"): return _bytes_from_excel({"Note": pd.DataFrame([{"status": "bridge unavailable — backtest_fa.csv missing"}])}, "Bridge") return _bytes_from_excel({ "Per SKU-week": result["df"], "By Category": result["by_category"], "Waterfall": pd.DataFrame(result["waterfall"]), }, "Bridge") def export_bridge_monthly_xlsx(result: dict) -> bytes: if not result.get("available"): return _bytes_from_excel({"Note": pd.DataFrame([ {"status": "monthly bridge unavailable — no actuals"} ])}, "Bridge") months = result["months"] summary_rows = [{ "month": m["month"], "tier": m["tier"], "n_skus": m["n_skus"], "plan_margin": m["plan_margin_total"], "actual_margin": m["actual_margin_total"], "total_variance": m["total_variance"], "volume_effect": m["volume_effect"], "price_effect": m["price_effect"], "mix_effect": m["mix_effect"], "cogs_effect": m["cogs_effect"], } for m in months] by_cat_rows = [] for m in months: for c in m["by_category"]: by_cat_rows.append({"month": m["month"], **c}) return _bytes_from_excel({ "Per SKU-month": pd.DataFrame(result["per_sku_rows"]), "Summary by Month": pd.DataFrame(summary_rows), "By Category": pd.DataFrame(by_cat_rows), "Top Movers": pd.DataFrame(result["top_movers_rows"]), }, "Bridge monthly")