"""Business logic for the supply module. Layering: takes a SQLAlchemy Session at construction, delegates SQL to SupplyRepository, never imports FastAPI. Same pattern as DemandService. Core operation: roll-forward closing stock per SKU per week. For each week in the 13-week horizon: opening_stock_w0 = current_stock opening_stock_wN = closing_stock_{wN-1} closing_stock_wN = max(0, opening_stock_wN - demand_wN + incoming_wN) Status thresholds (matches Streamlit's S&OP semantics): coverage_weeks < 2 → "Order Now" 2 <= coverage_weeks < 4 → "Order Soon" 4 <= coverage_weeks <= 13 → "OK" coverage_weeks > 13 → "Pull in" The Alerts page uses a lead-time-aware variant: effective_coverage = (current_stock + incoming_in_LT) / avg_demand coverage < lead_time → "Order Now" lead_time ≤ coverage < 2×LT → "Order Soon" coverage > 13 AND incoming → "Pull in" (defer the inbound) everything else → not in the alert list Order suggestion formula: target_stock = 2 × LT × avg_weekly_demand gap = target_stock - current_stock - incoming_in_LT suggested_qty = ceil(gap / MOQ) × MOQ (when gap > 0, else 0) KAM exclusion: subtract on-top quantities belonging to excluded (submitter, buyer) combinations from demand for each (sku, year, week). When on_top_inputs is empty (current dev state), this is a no-op. """ from __future__ import annotations import math from typing import Optional from sqlalchemy.orm import Session from backend.repositories.supply_repo import SupplyRepository _STATUS_OK = "OK" _STATUS_ORDER_SOON = "Order Soon" _STATUS_ORDER_NOW = "Order Now" _STATUS_PULL_IN = "Pull in" def _safe_float(v) -> Optional[float]: """Coerce to float, returning None for NULL / NaN / uncoercible. supply_master.lead_time_weeks / moq are Postgres NUMERIC, which can hold a literal NaN (most rows do — lead time simply isn't configured). NaN is neither NULL nor `!= itself` in Postgres, so it slips past IS NULL filters and an `is None` check, then detonates int(ceil(NaN)). Funnel both NULL and NaN to None here so callers treat them identically as 'missing'.""" if v is None: return None try: f = float(v) except (TypeError, ValueError): return None return None if math.isnan(f) else f def _classify(coverage: Optional[float]) -> str: if coverage is None: return _STATUS_OK if coverage < 2: return _STATUS_ORDER_NOW if coverage < 4: return _STATUS_ORDER_SOON if coverage > 13: return _STATUS_PULL_IN return _STATUS_OK def _cw_label(week: int) -> str: return f"CW{week:02d}" def _roll_forward( *, pid: int, current_stock: float, avg: float, horizon: list[tuple[int, int]], fc_map: dict, incoming_map: dict, excluded_map: Optional[dict] = None, ) -> list[dict]: """Roll closing stock forward across the horizon for one SKU. This is the core S&OP projection (see module docstring) shared by the supply dashboard, stock projection, and coverage grid: opening_w0 = current_stock demand_w = fc_map[(pid, y, w)] (fallback: avg weekly demand) minus on-top exclusions when excluded_map is given closing_w = max(0, opening_w - demand_w + incoming_w) opening_{w+1} = closing_w Returns one dict per horizon week — {year, week, opening, demand, incoming, closing} — in horizon order. Callers add their own labels, per-cell coverage, and cross-SKU aggregation. The max(0, …) floor is essential: a stocked-out SKU contributes 0 from that week on and stops absorbing further demand. `excluded_map` is opt-in so callers that don't subtract KAM on-top demand keep their exact prior behaviour (no clamp on the raw forecast). """ # Build this SKU's demand + incoming arrays (forecast else avg fallback, # minus on-top exclusions when given), then run the canonical walk-forward # (backend.services.coverage) — one engine shared with the scenario planner. from backend.services.coverage import walk_forward demand_arr: list[float] = [] incoming_arr: list[float] = [] for (y, w) in horizon: key = (pid, y, w) demand = float(fc_map.get(key, avg)) if excluded_map is not None: demand = max(0.0, demand - float(excluded_map.get(key, 0.0))) demand_arr.append(demand) incoming_arr.append(float(incoming_map.get(key, 0.0))) walked = walk_forward(current_stock, demand_arr, incoming_arr) return [ {"year": y, "week": w, **walked[i]} for i, (y, w) in enumerate(horizon) ] def _forward_cover(stock: float, demand: list[float], incoming: list[float], fallback_avg: float) -> Optional[float]: """Weeks until stock-out from this point, walking the ACTUAL forward demand path (not stock ÷ a static average). Incoming is added at the start of its week. If the stock survives the whole forward window, extrapolate the tail with `fallback_avg` (the SKU's own mean forecast demand) so a fully-covered horizon doesn't read as infinite. This keeps 'weeks of cover' consistent with the closing-stock depletion shown in the same grid.""" remaining = float(stock) weeks = 0.0 for j, d in enumerate(demand): remaining += float(incoming[j]) if j < len(incoming) else 0.0 d = float(d) if d <= 0: # no demand this week → free week of cover weeks += 1 continue if remaining >= d: remaining -= d weeks += 1 else: return round(weeks + remaining / d, 1) if fallback_avg > 0: return round(weeks + remaining / fallback_avg, 1) return None class SupplyService: # Mixin order: inventory-health + store-overstock methods come from # _InventoryHealthMixin (defined later in the file). Python's MRO walks # SupplyService → mixin → object, so the mixin methods are available # on every SupplyService instance. def __init__(self, db: Session): self.repo = SupplyRepository(db) # ------------------------------------------------------------------ # Dashboard — stock totals + horizon aggregate # ------------------------------------------------------------------ def get_supply_dashboard(self) -> dict: stock = self.repo.get_stock_summary() horizon = self.repo.get_horizon_weeks(n_weeks=13) # Use ALL SKUs that have stock OR planning as the universe — same # set the projection page operates on (without filters). products = self.repo.get_stock_by_product() pids = [int(p["product_id"]) for p in products] # Demand: forecasts first, fall back to recent avg per-week. fc_map = self.repo.get_forecast_demand_horizon(product_ids=pids, horizon=horizon) avg_map = self.repo.get_avg_weekly_demand(product_ids=pids) incoming_map = self.repo.get_incoming_horizon(product_ids=pids, horizon=horizon) # Per-SKU roll-forward (just for the dashboard aggregates; we don't # need the per-SKU rows here, only the totals + coverage stats). total_demand = 0.0 total_incoming = 0.0 coverages: list[float] = [] n_low = 0 n_over = 0 # Per-week aggregates for the dashboard line chart. week_demand: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} week_incoming: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} opening_for_week: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} closing_for_week: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} # Unplanned demand fallback: non-promo run rate. Mirrors the user # spec — planned SKUs use forecasts when available, unplanned fall # back to "last 13w non-promo avg". run_rate_map = self.repo.get_run_rates(product_ids=pids, weeks=13) for p in products: pid = int(p["product_id"]) is_planned = bool(p.get("is_planned")) current_stock = float(p.get("current_stock") or 0) # Planned SKUs: 4-week trailing avg from v_sales_weekly_full (existing # signal). Unplanned: non-promo run rate (longer window). if is_planned: avg = float(avg_map.get(pid, 0.0)) else: avg = float(run_rate_map.get(pid, {}).get("avg_weekly", 0.0)) cells_raw = _roll_forward(pid=pid, current_stock=current_stock, avg=avg, horizon=horizon, fc_map=fc_map, incoming_map=incoming_map) for cell in cells_raw: yw = (cell["year"], cell["week"]) week_demand[yw] += cell["demand"] week_incoming[yw] += cell["incoming"] opening_for_week[yw] += cell["opening"] closing_for_week[yw] += cell["closing"] total_demand += cell["demand"] total_incoming += cell["incoming"] # Coverage = real forward run-out (incl. incoming POs), not stock ÷ avg. d_arr = [c["demand"] for c in cells_raw] i_arr = [c["incoming"] for c in cells_raw] fwd_avg = (sum(d_arr) / len(d_arr)) if d_arr else 0.0 cov = _forward_cover(current_stock, d_arr, i_arr, fwd_avg) if cov is not None: coverages.append(cov) if cov < 2: n_low += 1 if cov > 13: n_over += 1 avg_coverage = sum(coverages) / len(coverages) if coverages else None horizon_rows = [{ "year": y, "week": w, "year_week": y * 100 + w, "cw_label": _cw_label(w), "demand": week_demand[(y, w)], "incoming": week_incoming[(y, w)], "opening_stock": opening_for_week[(y, w)], "closing_stock": closing_for_week[(y, w)], } for (y, w) in horizon] note = None if not fc_map: note = ( "Demand for the projection is based on a 4-week trailing " "average — the forecasts table is empty in this environment. " "It will switch to forecasts.total automatically once the " "pipeline starts writing." ) return { "stock": { "total_units": float(stock.get("total_units") or 0), "total_purchase_value": float(stock.get("total_purchase_value") or 0), "total_retail_value": float(stock.get("total_retail_value") or 0), "warehouse_units": float(stock.get("warehouse_units") or 0), "store_units": float(stock.get("store_units") or 0), "n_products": int(stock.get("n_products") or 0), "updated_at": stock["updated_at"].isoformat() if stock.get("updated_at") else None, "by_location": [{ "country": r.get("country") or "—", "is_warehouse": bool(r.get("is_warehouse")), "n_stores": int(r.get("n_stores") or 0), "units": float(r.get("units") or 0), "purchase_value": float(r.get("purchase_value") or 0), "retail_value": float(r.get("retail_value") or 0), } for r in stock.get("by_location", [])], }, "total_demand_13w": total_demand, "total_incoming_13w": total_incoming, "avg_coverage_weeks": avg_coverage, "n_skus_low_coverage": n_low, "n_skus_overstocked": n_over, "horizon": horizon_rows, "note": note, } # ------------------------------------------------------------------ # Stock projection — per-SKU 13-week roll-forward # ------------------------------------------------------------------ def get_stock_projection_data( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, sku: Optional[list[str]] = None, excluded_kams: Optional[list[str]] = None, excluded_buyers: Optional[list[str]] = None, ) -> dict: horizon = self.repo.get_horizon_weeks(n_weeks=13) products = self.repo.get_stock_by_product( category=category, tier=tier, xyz=xyz, sku=sku, ) pids = [int(p["product_id"]) for p in products] fc_map = self.repo.get_forecast_demand_horizon(product_ids=pids, horizon=horizon) avg_map = self.repo.get_avg_weekly_demand(product_ids=pids, weeks_back=4) incoming_map = self.repo.get_incoming_horizon(product_ids=pids, horizon=horizon) excluded_map = self.repo.get_on_top_quantities( product_ids=pids, horizon=horizon, excluded_kams=excluded_kams, excluded_buyers=excluded_buyers, ) rows: list[dict] = [] totals_status: dict[str, int] = { _STATUS_OK: 0, _STATUS_ORDER_SOON: 0, _STATUS_ORDER_NOW: 0, _STATUS_PULL_IN: 0, } sum_current = 0.0 sum_warehouse = 0.0 sum_store = 0.0 sum_demand = 0.0 sum_incoming = 0.0 for p in products: pid = int(p["product_id"]) current_stock = float(p.get("current_stock") or 0) avg = float(avg_map.get(pid, 0.0)) weeks: list[dict] = [] demand_13w = 0.0 incoming_13w = 0.0 # excluded_map subtracts on-top demand for the deselected KAM(s)/buyer(s). cells_raw = _roll_forward(pid=pid, current_stock=current_stock, avg=avg, horizon=horizon, fc_map=fc_map, incoming_map=incoming_map, excluded_map=excluded_map) for cell in cells_raw: y, w = cell["year"], cell["week"] weeks.append({ "year": y, "week": w, "year_week": y * 100 + w, "cw_label": _cw_label(w), "opening_stock": cell["opening"], "demand": cell["demand"], "incoming": cell["incoming"], "closing_stock": cell["closing"], }) demand_13w += cell["demand"] incoming_13w += cell["incoming"] # Order status from the real forward run-out (incl. incoming POs and # any KAM/buyer exclusions), not stock ÷ static trailing avg. d_arr = [c["demand"] for c in cells_raw] i_arr = [c["incoming"] for c in cells_raw] fwd_avg = (sum(d_arr) / len(d_arr)) if d_arr else 0.0 coverage = _forward_cover(current_stock, d_arr, i_arr, fwd_avg) status = _classify(coverage) totals_status[status] = totals_status.get(status, 0) + 1 sum_current += current_stock sum_warehouse += float(p.get("warehouse_stock") or 0) sum_store += float(p.get("store_stock") or 0) sum_demand += demand_13w sum_incoming += incoming_13w rows.append({ "sku": p["sku"], "name": p.get("name"), "category": p.get("category"), "tier": p.get("tier"), "xyz": p.get("xyz"), "current_stock": current_stock, "warehouse_stock": float(p.get("warehouse_stock") or 0), "store_stock": float(p.get("store_stock") or 0), "avg_weekly_demand": round(fwd_avg, 1), # forward forecast rate (matches projection), not trailing avg "demand_13w": demand_13w, "incoming_13w": incoming_13w, "coverage_weeks": coverage, "closing_stock_eow_13": weeks[-1]["closing_stock"] if weeks else current_stock, "status": status, "weeks": weeks, }) # Default sort: most urgent first (Order Now → Order Soon → others). status_order = { _STATUS_ORDER_NOW: 0, _STATUS_ORDER_SOON: 1, _STATUS_PULL_IN: 2, _STATUS_OK: 3, } rows.sort(key=lambda r: (status_order.get(r["status"], 9), r["coverage_weeks"] if r["coverage_weeks"] is not None else 1e9)) horizon_labels = [_cw_label(w) for (_, w) in horizon] # KAM submission list for the exclusion UI kams_raw = self.repo.get_on_top_by_kam() kams = [{ "submitted_by_id": r.get("submitted_by_id"), "person": r.get("person") or "(unassigned)", "role": r.get("role"), "channel": r.get("channel") or None, "n_sku_weeks": int(r.get("n_sku_weeks") or 0), "qty_total": float(r.get("qty_total") or 0), "buyers": list(r.get("buyers") or []), } for r in kams_raw] # Build the user-facing note about what was subtracted. note = None notes: list[str] = [] if not fc_map: notes.append( "Demand for the projection is using a 4-week trailing average " "(forecasts table is empty)." ) if (excluded_kams or excluded_buyers) and not excluded_map: notes.append( "Exclusion filter applied but no on-top commitments exist to " "subtract — on_top_inputs is empty in this environment." ) elif excluded_map: subtracted = sum(excluded_map.values()) notes.append( f"Subtracted {subtracted:,.0f} units of on-top demand from the " f"selected KAM(s)/buyer(s) across the horizon." ) if notes: note = " ".join(notes) # ── Streamlit-parity € rollup (current + closing per-week value) ── # Cost prices for every SKU in scope. erp_costs is the source, with # erp_prices.avg_sell_price as fallback. Missing → 0. unit_vals = self.repo.get_unit_values(product_ids=pids) if pids else {} cost_by_sku: dict[str, float] = {} for p in products: uv = unit_vals.get(int(p["product_id"]), {}) cost_by_sku[p["sku"]] = float(uv.get("cost_price") or 0.0) # Planned-SKU set for the forecasted vs long-tail split planned_pids: set[int] = set() try: planned_rows = self.repo.get_stock_by_product(planned_only=True) planned_pids = {int(p["product_id"]) for p in planned_rows} except Exception: pass planned_skus = {p["sku"] for p in products if int(p["product_id"]) in planned_pids} total_current_value = 0.0 total_forecasted_value = 0.0 total_longtail_value = 0.0 # Per-week aggregate accumulators wk_open: dict[tuple[int, int], float] = {} wk_dem: dict[tuple[int, int], float] = {} wk_inc: dict[tuple[int, int], float] = {} wk_close:dict[tuple[int, int], float] = {} wk_val: dict[tuple[int, int], float] = {} for r in rows: sku = r["sku"] cost = cost_by_sku.get(sku, 0.0) cur_val = float(r["current_stock"]) * cost total_current_value += cur_val if sku in planned_skus: total_forecasted_value += cur_val else: total_longtail_value += cur_val for w in r["weeks"]: key = (int(w["year"]), int(w["week"])) wk_open[key] = wk_open.get(key, 0.0) + float(w["opening_stock"]) wk_dem[key] = wk_dem.get(key, 0.0) + float(w["demand"]) wk_inc[key] = wk_inc.get(key, 0.0) + float(w["incoming"]) wk_close[key] = wk_close.get(key, 0.0) + float(w["closing_stock"]) wk_val[key] = wk_val.get(key, 0.0) + float(w["closing_stock"]) * cost # Supplier-plan overlay: per-week €NV of incoming goods sourced from # the buying team's "Template_Plan_ulaza_izlaza_2026" workbook. # incoming_supply has per-SKU PO data up to CW28; this overlay paints # the planned arrivals from CW29 onward on the value chart so the # full 13-week horizon shows a continuous picture without needing # us to fabricate per-SKU rows. from backend.services.supplier_plan_template import planned_incoming_eur weekly_agg = [ { "year": y, "week": w, "cw_label": _cw_label(w), "opening_stock": wk_open.get((y, w), 0.0), "demand": wk_dem.get((y, w), 0.0), "incoming": wk_inc.get((y, w), 0.0), "closing_stock": wk_close.get((y, w), 0.0), "closing_value_eur": wk_val.get((y, w), 0.0), "planned_incoming_eur": planned_incoming_eur(y, w), } for (y, w) in horizon ] return { "rows": rows, "totals": { "n_skus": len(rows), "total_current_stock": sum_current, "total_warehouse_stock": sum_warehouse, "total_store_stock": sum_store, "total_demand_13w": sum_demand, "total_incoming_13w": sum_incoming, "by_status": totals_status, "total_current_value_eur": total_current_value, "total_forecasted_value_eur": total_forecasted_value, "total_longtail_value_eur": total_longtail_value, }, "horizon": horizon_labels, "kams_available": kams, "excluded_kams": excluded_kams or [], "excluded_buyers": excluded_buyers or [], "weekly_aggregate": weekly_agg, "note": note, } # ------------------------------------------------------------------ # Coverage workbook — 5 metric rows × N week columns per SKU # ------------------------------------------------------------------ def get_coverage_data( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, sku: Optional[list[str]] = None, ) -> dict: """Same per-SKU roll-forward as the projection, but the response is shaped for the wide coverage grid: one CoverageSku per SKU carrying the per-week cells, plus a per-week totals row for the summary. Planning tool — restricted to SKUs in sku_planning. Unplanned SKUs lack the demand-variability profile needed for the coverage grid and would just pollute the planner's view.""" horizon = self.repo.get_horizon_weeks(n_weeks=13) # Coverage Workbook is a procurement planning tool — only WH stock # is actionable for new orders. Store stock is already at the # customer-facing endpoint and can't be redirected upstream. products = self.repo.get_stock_by_product( category=category, tier=tier, xyz=xyz, sku=sku, planned_only=True, warehouse_only=True, ) pids = [int(p["product_id"]) for p in products] fc_map = self.repo.get_forecast_demand_horizon(product_ids=pids, horizon=horizon) avg_map = self.repo.get_avg_weekly_demand(product_ids=pids) incoming_map = self.repo.get_incoming_horizon(product_ids=pids, horizon=horizon) master_map = self.repo.get_supply_master(product_ids=pids) rows: list[dict] = [] # Per-week aggregates for the summary row at the top of the grid. wk_opening: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} wk_demand: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} wk_incoming: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} wk_closing: dict[tuple[int, int], float] = {(y, w): 0.0 for (y, w) in horizon} for p in products: pid = int(p["product_id"]) current_stock = float(p.get("current_stock") or 0) avg = float(avg_map.get(pid, 0.0)) sm = master_map.get(pid, {}) cells_raw = _roll_forward(pid=pid, current_stock=current_stock, avg=avg, horizon=horizon, fc_map=fc_map, incoming_map=incoming_map) # Coverage walks the real forward demand path, not closing ÷ static # historical avg (which overstated cover when the forecast — incl. # folded KAM/CM on-tops — runs hotter than the trailing average). demand_arr = [c["demand"] for c in cells_raw] fwd_avg = (sum(demand_arr) / len(demand_arr)) if demand_arr else 0.0 # Run-rate for weeks BEYOND the forecast horizon = the SKU's non-promo # weekly baseline (falls back to the forecast mean). tail = float(avg_map.get(pid, 0.0)) or fwd_avg cells: list[dict] = [] for i, cell in enumerate(cells_raw): y, w = cell["year"], cell["week"] # Cover = forward run-out: from THIS week's stock, consume the real # demand curve (forecast incl on-tops for the weeks we have, then the # baseline run-rate) week by week until exhausted. Future deliveries # are NOT added to the number — an empty week reads 0.0w even if a PO # lands later (the stock itself already reflects POs arrived by now, # so cover jumps the week a PO actually lands). cov = _forward_cover(cell["opening"], demand_arr[i:], [], tail) cells.append({ "year": y, "week": w, "year_week": y * 100 + w, "cw_label": _cw_label(w), "opening_stock": cell["opening"], "demand": cell["demand"], "incoming": cell["incoming"], "closing_stock": cell["closing"], "coverage_weeks": cov, }) wk_opening[(y, w)] += cell["opening"] wk_demand[(y, w)] += cell["demand"] wk_incoming[(y, w)] += cell["incoming"] wk_closing[(y, w)] += cell["closing"] rows.append({ "sku": p["sku"], "name": p.get("name"), "category": p.get("category"), "tier": p.get("tier"), "xyz": p.get("xyz"), "current_stock": current_stock, "avg_weekly_demand": round(fwd_avg, 1), # forward forecast rate (matches depletion), not trailing avg "cover_tail": round(tail, 4), # exact run-rate the cover walk uses past the horizon (for client-side what-if) "lead_time_weeks": sm.get("lead_time_weeks"), "moq": sm.get("moq"), "supplier": sm.get("supplier"), "weeks": cells, }) # Sort: most urgent first by week-1 forward run-out (lowest cover on top). def _urgency_key(r: dict) -> float: if not r["weeks"]: return 1e9 cov = r["weeks"][0].get("coverage_weeks") return cov if cov is not None else 1e9 rows.sort(key=_urgency_key) totals = [{ "year_week": y * 100 + w, "cw_label": _cw_label(w), "opening_stock": wk_opening[(y, w)], "demand": wk_demand[(y, w)], "incoming": wk_incoming[(y, w)], "closing_stock": wk_closing[(y, w)], } for (y, w) in horizon] note = None if not fc_map: note = ( "Demand uses a 4-week trailing average (forecasts table empty). " "Coverage cells will switch to forecasted demand once the " "pipeline writes rows." ) return { "rows": rows, "horizon": [_cw_label(w) for (_, w) in horizon], "totals_by_week": totals, "note": note, } def build_coverage_xlsx( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, sku: Optional[list[str]] = None, ) -> bytes: """Render the coverage grid to an Excel blob, mirroring the on-screen layout: 5 metric rows per SKU (Stock / Demand / Incoming / Closing / Cov) × the 13-week horizon, a Σ totals block on top, and coverage cells colored red/amber/green/blue by weeks-of-cover (same thresholds as the UI: <2 red, <4 amber, >26 blue, else green).""" data = self.get_coverage_data(category=category, tier=tier, xyz=xyz, sku=sku) rows = data["rows"] horizon = data["horizon"] # CW labels totals = data["totals_by_week"] import io from openpyxl import Workbook from openpyxl.styles import Alignment, Font, PatternFill from openpyxl.utils import get_column_letter RED = PatternFill("solid", fgColor="FECDD3") AMBER = PatternFill("solid", fgColor="FDE68A") GREEN = PatternFill("solid", fgColor="D1FAE5") BLUE = PatternFill("solid", fgColor="DBEAFE") HDR = PatternFill("solid", fgColor="1E293B") TOTF = PatternFill("solid", fgColor="E2E8F0") white_bold = Font(bold=True, color="FFFFFF") bold = Font(bold=True) center = Alignment(horizontal="center") def _cov_fill(v): if v is None: return None if v < 2: return RED if v < 4: return AMBER if v > 26: return BLUE return GREEN wb = Workbook() ws = wb.active ws.title = "Coverage" id_cols = ["SKU", "Name", "Category", "Tier", "Supplier", "Lead (w)", "MOQ", "Metric"] n_id = len(id_cols) header = id_cols + list(horizon) ws.append(header) for c in range(1, len(header) + 1): cell = ws.cell(row=1, column=c) cell.font = white_bold cell.fill = HDR cell.alignment = center # Σ totals block (Demand, then +Incoming) across the exported scope. dem = ["Σ TOTALS", "across rows in scope", "", "", "", "", "", "Demand"] + [t["demand"] for t in totals] inc = ["", "", "", "", "", "", "", "Incoming"] + [t["incoming"] for t in totals] ws.append(dem) ws.append(inc) for rr in (2, 3): for c in range(1, len(header) + 1): cell = ws.cell(row=rr, column=c) cell.fill = TOTF if c <= n_id: cell.font = bold else: cell.number_format = "#,##0" metrics = [ ("opening_stock", "Stock"), ("demand", "Demand"), ("incoming", "Incoming"), ("closing_stock", "Closing"), ("coverage_weeks", "Cov (w)"), ] for row in rows: first = True for key, label in metrics: line = [ row["sku"] if first else "", (row.get("name") or "") if first else "", (row.get("category") or "") if first else "", (row.get("tier") or "") if first else "", (row.get("supplier") or "") if first else "", row.get("lead_time_weeks") if first else "", row.get("moq") if first else "", label, ] line += [wk.get(key) for wk in row["weeks"]] ws.append(line) rr = ws.max_row is_cov = key == "coverage_weeks" for ci, wk in enumerate(row["weeks"]): cell = ws.cell(row=rr, column=n_id + 1 + ci) if is_cov: cell.number_format = "0.0" fill = _cov_fill(wk.get("coverage_weeks")) if fill: cell.fill = fill else: cell.number_format = "#,##0" if first: ws.cell(row=rr, column=1).font = bold first = False # Freeze the 3 header rows + the identity/metric columns. ws.freeze_panes = ws.cell(row=4, column=n_id + 1) for i, w in enumerate((14, 34, 18, 9, 22, 8, 8, 10), start=1): ws.column_dimensions[get_column_letter(i)].width = w for i in range(n_id + 1, n_id + 1 + len(horizon)): ws.column_dimensions[get_column_letter(i)].width = 8 buf = io.BytesIO() wb.save(buf) return buf.getvalue() # ------------------------------------------------------------------ # Reorder alerts — lead-time-aware classification # ------------------------------------------------------------------ def get_alerts( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, ) -> dict: """Categorize each SKU into Order Now / Order Soon / Pull in / (none). Pulls supply_master for lead time and MOQ. SKUs without a supply_master row are skipped and surfaced as `n_missing_lt` rather than being silently defaulted.""" sugg_result = self._compute_suggestions(category=category, tier=tier, xyz=xyz) suggestions = sugg_result["rows"] n_missing_lt = sugg_result["n_missing_lt"] order_now: list[dict] = [] order_soon: list[dict] = [] pull_in: list[dict] = [] for s in suggestions: urg = s["urgency"] avg = s["avg_weekly_demand"] row: dict = { "sku": s["sku"], "name": s["name"], "category": s["category"], "tier": s["tier"], "current_stock": s["current_stock"], "avg_weekly_demand": avg, "coverage_weeks": s["coverage_weeks"], "effective_coverage_weeks": s["effective_coverage_weeks"], "incoming_in_lead_time": s["incoming_in_lead_time"], "lead_time_weeks": s["lead_time_weeks"], "reorder_point_units": ( s["lead_time_weeks"] * 2 * avg if s["lead_time_weeks"] is not None else None ), "suggested_action": _format_action(s), "urgency": urg, "supplier": s["supplier"], } if urg == "Order Now": order_now.append(row) elif urg == "Order Soon": order_soon.append(row) elif urg == "Pull in": pull_in.append(row) # Sort each bucket: most urgent first within Order Now/Soon (lowest # effective coverage), most overstocked first within Pull in. order_now.sort(key=lambda r: r["effective_coverage_weeks"] if r["effective_coverage_weeks"] is not None else 0) order_soon.sort(key=lambda r: r["effective_coverage_weeks"] if r["effective_coverage_weeks"] is not None else 0) pull_in.sort(key=lambda r: -(r["coverage_weeks"] or 0)) note = None notes: list[str] = [] n_no_master = sum( 1 for s in suggestions if s["lead_time_weeks"] is None ) if n_missing_lt > 0: notes.append( f"{n_missing_lt} SKU(s) skipped — missing supply_master " "lead time. Add the supplier row to include them." ) if notes: note = " ".join(notes) return { "order_now": order_now, "order_soon": order_soon, "pull_in": pull_in, "n_total": len(order_now) + len(order_soon) + len(pull_in), "n_missing_lt": n_missing_lt, "note": note, } # ------------------------------------------------------------------ # Order suggestions — pre-filled quantities for the entry form # ------------------------------------------------------------------ def get_order_suggestions( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, ) -> dict: """Returns rows where suggested_qty > 0 (i.e. action recommended) sorted by urgency. The frontend pre-fills the editable qty field with `suggested_qty` and lets the user adjust before saving.""" sugg_result = self._compute_suggestions(category=category, tier=tier, xyz=xyz) suggestions = sugg_result["rows"] n_missing_lt = sugg_result["n_missing_lt"] # Keep rows where the system actually suggests an order. rows = [s for s in suggestions if s["suggested_qty"] > 0] urgency_order = {"Order Now": 0, "Order Soon": 1, "Pull in": 2, "—": 3} rows.sort(key=lambda r: ( urgency_order.get(r["urgency"], 4), r["effective_coverage_weeks"] if r["effective_coverage_weeks"] is not None else 1e9, )) note = None if not rows: note = "No reorder suggestions in scope." if n_missing_lt > 0: extra = (f"{n_missing_lt} SKU(s) skipped — missing supply_master " "lead time.") note = f"{note} {extra}" if note else extra return {"rows": rows, "note": note, "n_missing_lt": n_missing_lt} # ------------------------------------------------------------------ # Shared computation for alerts + suggestions # ------------------------------------------------------------------ def _compute_suggestions( self, *, category: Optional[list[str]] = None, tier: Optional[list[str]] = None, xyz: Optional[list[str]] = None, ) -> dict: """One pass over the SKU universe that produces the fields both the Alerts page and Order Entry page need. Heavy lifting lives here so we don't traverse the catalogue twice. Returns `{"rows": [...], "n_missing_lt": int}`. SKUs without a `supply_master.lead_time_weeks` entry are skipped (no flat default) and counted so callers can surface the data-quality gap. Both endpoints are planning tools — planned SKUs only.""" horizon = self.repo.get_horizon_weeks(n_weeks=13) products = self.repo.get_stock_by_product( category=category, tier=tier, xyz=xyz, planned_only=True, ) pids = [int(p["product_id"]) for p in products] avg_map = self.repo.get_avg_weekly_demand(product_ids=pids) fc_map = self.repo.get_forecast_demand_horizon(product_ids=pids, horizon=horizon) incoming_map = self.repo.get_incoming_horizon(product_ids=pids, horizon=horizon) master_map = self.repo.get_supply_master(product_ids=pids) out: list[dict] = [] n_missing_lt = 0 for p in products: pid = int(p["product_id"]) current_stock = float(p.get("current_stock") or 0) # Demand rate = FORWARD forecast (fallback: trailing avg). Using the # trailing non-promo avg here understated demand and shrank both the # target stock and the suggested order qty -> chronic under-ordering. avg_fallback = float(avg_map.get(pid, 0.0)) demand_arr = [float(fc_map.get((pid, y, w), avg_fallback)) for (y, w) in horizon] incoming_arr = [float(incoming_map.get((pid, y, w), 0.0)) for (y, w) in horizon] avg = (sum(demand_arr) / len(demand_arr)) if demand_arr else 0.0 sm = master_map.get(pid, {}) lt_for_calc = _safe_float(sm.get("lead_time_weeks")) if lt_for_calc is None: # supply_master is mandatory; skip SKUs without a usable lead # time (NULL or a numeric NaN) and surface in the data-quality # counter so missing supplier setup is visible. n_missing_lt += 1 continue moq = _safe_float(sm.get("moq")) moq_for_calc = moq if (moq is not None and moq > 0) else 1.0 # Incoming during the next lead_time weeks. in_lt = 0.0 for i, (y, w) in enumerate(horizon): if i >= lt_for_calc: break in_lt += float(incoming_map.get((pid, y, w), 0.0)) # Coverage = real forward run-out along the forecast demand path # (on-hand only), effective coverage also credits incoming POs. coverage = _forward_cover(current_stock, demand_arr, [0.0] * len(demand_arr), avg) eff_coverage = _forward_cover(current_stock, demand_arr, incoming_arr, avg) target = 2 * lt_for_calc * avg gap = target - current_stock - in_lt if gap > 0 and moq_for_calc > 0: suggested = math.ceil(gap / moq_for_calc) * moq_for_calc else: suggested = 0.0 # Urgency classification (lead-time aware). urgency = "—" if avg > 0 and coverage is not None: if eff_coverage is not None and eff_coverage < lt_for_calc: urgency = "Order Now" elif eff_coverage is not None and eff_coverage < 2 * lt_for_calc: urgency = "Order Soon" elif coverage > 13 and in_lt > 0: urgency = "Pull in" # Target delivery week = horizon[ceil(lead_time) - 1], clamped. lt_idx = max(0, min(len(horizon) - 1, int(math.ceil(lt_for_calc)) - 1)) (ty, tw) = horizon[lt_idx] target_yw = ty * 100 + tw out.append({ "sku": p["sku"], "name": p.get("name"), "category": p.get("category"), "tier": p.get("tier"), "xyz": p.get("xyz"), "current_stock": current_stock, "avg_weekly_demand": round(avg, 1), "incoming_in_lead_time": in_lt, "coverage_weeks": coverage, "effective_coverage_weeks": eff_coverage, "lead_time_weeks": lt_for_calc, "moq": moq, "supplier": sm.get("supplier"), "target_stock_units": target, "gap_units": max(0.0, gap), "suggested_qty": float(suggested), "target_year_week": target_yw, "urgency": urgency, }) return {"rows": out, "n_missing_lt": n_missing_lt} # ------------------------------------------------------------------ # Persist order proposals # ------------------------------------------------------------------ _ALLOWED_STATUSES = {"draft", "pending", "approved"} def save_order_proposals(self, inputs: list[dict]) -> dict: """Bulk-save user-edited order quantities. Resolves SKU → product_id client-side (so the API stays sku-friendly), skips unknown SKUs with a descriptive result row, and reports per-row outcomes.""" skus = [i["sku"] for i in inputs if i.get("sku")] sku_to_pid = self.repo.resolve_skus_to_ids(skus) prepared: list[dict] = [] results_for_skipped: list[dict] = [] for i in inputs: sku = i.get("sku") if not sku: results_for_skipped.append({ "sku": "", "year_week": i.get("year_week", 0), "id": None, "status": "skipped", "message": "missing sku", }) continue pid = sku_to_pid.get(sku) if pid is None: results_for_skipped.append({ "sku": sku, "year_week": int(i.get("year_week") or 0), "id": None, "status": "skipped", "message": "unknown sku", }) continue status = (i.get("status") or "draft").lower() if status not in self._ALLOWED_STATUSES: status = "draft" prepared.append({ "sku": sku, "product_id": pid, "year_week": int(i["year_week"]), "proposed_qty": float(i["proposed_qty"]), "status": status, }) results = self.repo.create_order_proposals(proposals=prepared) all_results = results + results_for_skipped n_created = sum(1 for r in all_results if r["status"] == "created") n_skipped = sum(1 for r in all_results if r["status"] == "skipped") n_errors = sum(1 for r in all_results if r["status"] == "error") return { "n_created": n_created, "n_skipped": n_skipped, "n_errors": n_errors, "results": all_results, } # --------------------------------------------------------------------------- # Inventory health — Streamlit page_supply_inventory_health verbatim # --------------------------------------------------------------------------- # Reference: app.py:9053-9372. The math here is a direct port; the only # deviation from Streamlit is the data source — we read forecasts / # v_sales_weekly_full / erp_costs / erp_prices instead of CSVs. Three statuses # only (overstock / stockout_risk / balanced) and the thresholds are: # # safety_stock = Z × √LT × √(σ² + (avg × (1 − FA))²) # safety_weeks = safety_stock / avg_weekly # max_cover_needed = 1.5 × LT + safety_weeks # max_on_hand = demand_over(max_cover_needed, weekly_demand_series) # excess_units = max(0, on_hand − max_on_hand) # shortage_units = max(0, safety_stock − on_hand) # # overstock ← excess_units > 0 # stockout_risk ← shortage_units > 0 (NB: incoming intentionally not # subtracted — matches Streamlit # comment app.py:9219-9222) # balanced ← otherwise # Tier Z-scores: SL 98% / 95% / 92% (Streamlit defaults). # Unplanned: SL 93% — used when a SKU isn't in sku_planning. _TIER_Z = { "01 GOLD": 2.054, # norm.ppf(0.98) "02 SILVER": 1.645, # norm.ppf(0.95) "03 BRONZE": 1.405, # norm.ppf(0.92) } _UNPLANNED_Z = 1.476 # norm.ppf(0.93) _FA_GLOBAL_DEFAULT = 0.70 # matches Streamlit's "global fallback FA" slider default _FA_MIN_WEEKS = 6 # weeks of backtest history needed to trust per-SKU FA _DEFAULT_CV = 0.5 # Lead time comes exclusively from supply_master.lead_time_weeks (per-SKU, # per-supplier). When missing the SKU is skipped from inventory-health bucketing # and counted as a data-quality issue — no silent flat default any more. def _demand_over(weeks_needed: float, series: list[float]) -> float: """Sum demand over the first `weeks_needed` weeks. Partial trailing week is prorated. If the horizon exceeds the series, extrapolates with the series mean. Mirrors app.py:9168-9183.""" if not series or weeks_needed <= 0: return 0.0 whole = int(weeks_needed) frac = weeks_needed - whole available = len(series) if whole >= available: avg = sum(series) / available if available > 0 else 0.0 return float(sum(series) + (weeks_needed - available) * avg) total = float(sum(series[:whole])) if frac > 0: total += frac * float(series[whole]) return total def _tier_z_for(tier: Optional[str]) -> float: if tier: for k, v in _TIER_Z.items(): if k in tier: return v return _UNPLANNED_Z # Attach inventory-health + store-overstock methods to the existing class. # Keeping them as bound methods (not free functions) so they share the repo # session through self. class _InventoryHealthMixin: """Mixin to keep the inventory-health computation visually separated from the older projection/coverage code. Mixed into SupplyService at file-bottom via class augmentation.""" repo: SupplyRepository # type hint for the mixin — SupplyService sets this in __init__ def get_inventory_health( self, *, tier: Optional[list[str]] = None, category: Optional[list[str]] = None, status: Optional[str] = None, planned_only: bool = False, fa_global: float = _FA_GLOBAL_DEFAULT, fa_min_weeks: int = _FA_MIN_WEEKS, horizon_weeks: int = 13, ) -> dict: horizon = self.repo.get_horizon_weeks(n_weeks=horizon_weeks) products = self.repo.get_stock_by_product( category=category, tier=tier, planned_only=planned_only, ) pids = [int(p["product_id"]) for p in products] # Weekly demand source: forecasts (planned) when present, run_rate # otherwise (unplanned + planned with empty forecast). fc_map = self.repo.get_forecast_demand_horizon(product_ids=pids, horizon=horizon) run_rates = self.repo.get_run_rates(product_ids=pids, weeks=13) incoming_h = self.repo.get_incoming_horizon(product_ids=pids, horizon=horizon) master = self.repo.get_supply_master(product_ids=pids) per_sku_fa = self.repo.get_per_sku_backtest_fa(min_weeks=fa_min_weeks) values = self.repo.get_unit_values(product_ids=pids) rows: list[dict] = [] total_excess = 0.0 total_shortage = 0.0 n_measured = 0 valued_with_cost_count = 0 n_missing_lt = 0 by_status: dict[str, int] = {"overstock": 0, "stockout_risk": 0, "balanced": 0} by_status_planned: dict[str, int] = {"overstock": 0, "stockout_risk": 0, "balanced": 0} by_status_unplanned: dict[str, int] = {"overstock": 0, "stockout_risk": 0, "balanced": 0} for p in products: pid = int(p["product_id"]) is_planned = bool(p.get("is_planned")) sku = p["sku"] on_hand = float(p.get("current_stock") or 0) # Build per-week demand series for this SKU. # # Series shape mirrors Streamlit's `fc_series[sku]` — one float # per week of the horizon. Planned SKUs use forecasts.total when # present; otherwise we fall back to a flat run-rate series. The # flat fallback is what app.py does in its "legacy flat format" # branch (app.py:9157-9160). fc_vals = [ float(fc_map.get((pid, y, w), 0.0)) for (y, w) in horizon ] has_fc = any(v > 0 for v in fc_vals) if is_planned and has_fc: series = fc_vals demand_signal = "Forecast" else: run_rate = float(run_rates.get(pid, {}).get("avg_weekly", 0.0)) series = [run_rate] * len(horizon) demand_signal = "Run rate" fc_total = sum(series) n_weeks = len(series) avg_weekly = fc_total / n_weeks if n_weeks else 0.0 sm = master.get(pid, {}) lt_raw = sm.get("lead_time_weeks") if lt_raw is None: # supply_master is mandatory — surface the data gap rather than # silently using a flat default. SKU is skipped from bucketing. n_missing_lt += 1 continue lt = float(lt_raw) supplier = sm.get("supplier") tier_val = p.get("tier") if is_planned else None cv_raw = p.get("total_cv") cv = float(cv_raw) if (is_planned and cv_raw is not None) else _DEFAULT_CV z = _tier_z_for(tier_val) if is_planned else _UNPLANNED_Z # Per-SKU FA when enough history exists, else tier/global default. fa_info = per_sku_fa.get(pid) if fa_info: fa_used = float(fa_info["fa"]) fa_source = "measured" n_measured += 1 else: fa_used = fa_global if is_planned else _FA_GLOBAL_DEFAULT fa_source = "global" if is_planned else "default" # Safety stock — verbatim port of app.py:9211-9214. sigma_weekly = avg_weekly * cv forecast_err_sigma = avg_weekly * (1.0 - fa_used) combined_sigma = math.sqrt(sigma_weekly ** 2 + forecast_err_sigma ** 2) safety_stock = z * combined_sigma * (math.sqrt(lt) if lt > 0 else 0.0) safety_weeks = (safety_stock / avg_weekly) if avg_weekly > 0 else 0.0 max_cover_needed = 1.5 * lt + safety_weeks max_on_hand = _demand_over(max_cover_needed, series) cover_now = (on_hand / avg_weekly) if avg_weekly > 0 else None excess_units = max(0.0, on_hand - max_on_hand) shortage_units = max(0.0, safety_stock - on_hand) # Unit value — prefer cost, fall back to sell. Matches app.py:9233-9237. v = values.get(pid, {}) cost = v.get("cost_price") sell = v.get("avg_sell_price") if cost is not None and cost > 0: unit_value = float(cost) valued_with = "cost" valued_with_cost_count += 1 elif sell is not None and sell > 0: unit_value = float(sell) valued_with = "sell" else: unit_value = 0.0 valued_with = "—" if excess_units > 0: row_status = "overstock" elif shortage_units > 0: row_status = "stockout_risk" else: row_status = "balanced" by_status[row_status] += 1 (by_status_planned if is_planned else by_status_unplanned)[row_status] += 1 total_excess += excess_units * unit_value total_shortage += shortage_units rows.append({ "sku": sku, "name": p.get("name"), "category": p.get("category"), "tier": tier_val, "is_planned": is_planned, "supplier": supplier, "on_hand": on_hand, "incoming": float(sum(incoming_h.get((pid, y, w), 0.0) for (y, w) in horizon)), "avg_weekly": avg_weekly, "demand_signal": demand_signal, "cv": cv, "lead_time_weeks": lt, "fa_used": fa_used, "fa_source": fa_source, "safety_stock": safety_stock, "safety_weeks": safety_weeks, "max_cover_needed": max_cover_needed, "max_on_hand": max_on_hand, "cover_now": cover_now, "excess_units": excess_units, "excess_eur": excess_units * unit_value, "shortage_units": shortage_units, "status": row_status, "unit_value": unit_value, "valued_with": valued_with, }) # Optional status filter (applied last so by_status totals remain # accurate for the headline cards regardless of view). if status and status != "all": rows = [r for r in rows if r["status"] == status] # Sort: biggest excess € first (matches Streamlit default sort). rows.sort(key=lambda r: r["excess_eur"], reverse=True) n_planned = sum(1 for r in rows if r["is_planned"]) n_unplanned = len(rows) - n_planned note = None if not fc_map: note = ( "Demand series uses non-promo run rate everywhere — the " "forecasts table is empty in this environment. Planned SKUs " "will switch to forecast.total once the pipeline writes rows." ) return { "rows": rows, "total_excess_eur": total_excess, "total_shortage_units": total_shortage, "by_status": by_status, "by_status_planned": by_status_planned, "by_status_unplanned": by_status_unplanned, "n_planned": n_planned, "n_unplanned": n_unplanned, "n_measured_fa": n_measured, "fa_global_default": fa_global, "valued_with_cost": valued_with_cost_count, "n_missing_lt": n_missing_lt, "note": note, } # ------------------------------------------------------------------ # Store overstock — Streamlit page_supply_store_overstock verbatim # ------------------------------------------------------------------ def get_store_overstock( self, *, tier: Optional[list[str]] = None, country: str = "HR", ) -> dict: """Per-(SKU, store) overstock for HR retail. Hard-filtered to planned SKUs (Gold/Silver/Bronze) — unplanned SKUs are excluded entirely, matching app.py:9448-9456. Formula: safety_stock = Z × σ × √LT (LT = 1) cycle_stock = LT × avg_weekly = avg_weekly optimal_stock = safety_stock + cycle_stock overstock_units = max(0, on_hand − optimal_stock) σ from std-dev of non-promo weekly sales per (SKU, store). Pairs with < 4 non-promo weeks → overstock_units forced to 0 ('insufficient_history' flag). """ # 1) Build the universe: every (product, store) row that has stock, # restricted to planned SKUs in the requested tiers. stock_rows = self.repo.get_store_stock(country=country) if not stock_rows: return _empty_store_overstock(country=country) # 2) Resolve planned tier per product via the existing helper. all_products = self.repo.get_stock_by_product( planned_only=True, tier=tier, ) planned_index = { int(p["product_id"]): { "sku": p["sku"], "name": p.get("name"), "category": p.get("category"), "tier": p.get("tier"), } for p in all_products if bool(p.get("is_planned")) } # Keep only the (product, store) pairs whose product is planned. store_rows = [r for r in stock_rows if int(r["product_id"]) in planned_index] # 3) Per-(product, store) non-promo stats + per-product unit value. stats = self.repo.get_store_sku_non_promo_stats(country=country) stats_index = { (int(s["product_id"]), int(s["store_id"])): s for s in stats } values = self.repo.get_unit_values( product_ids=list(planned_index.keys()), ) MIN_WEEKS = 4 LT = 1 rows: list[dict] = [] total_units = 0.0 total_cost = 0.0 total_sell = 0.0 n_with_overstock = 0 n_insufficient = 0 for r in store_rows: pid = int(r["product_id"]) sid = int(r["store_id"]) meta = planned_index[pid] tier_val = meta.get("tier") or "" z = _tier_z_for(tier_val) s = stats_index.get((pid, sid), {}) avg_weekly = float(s.get("avg_weekly") or 0) sigma = float(s.get("sigma") or 0) n_weeks = int(s.get("weeks_with_sales") or 0) safety_stock = z * sigma * math.sqrt(LT) cycle_stock = LT * avg_weekly optimal = safety_stock + cycle_stock on_hand = float(r["on_hand"] or 0) insufficient = n_weeks < MIN_WEEKS overstock_units = max(0.0, on_hand - optimal) if not insufficient else 0.0 v = values.get(pid, {}) cost = float(v.get("cost_price") or 0) sell = float(v.get("avg_sell_price") or 0) eur_cost = overstock_units * cost eur_sell = overstock_units * sell if overstock_units > 0: n_with_overstock += 1 total_units += overstock_units total_cost += eur_cost total_sell += eur_sell if insufficient: n_insufficient += 1 rows.append({ "sku": meta["sku"], "name": meta.get("name"), "category": meta.get("category"), "tier": tier_val, "store_code": str(r.get("store_code") or sid), "store_name": str(r.get("store_name") or r.get("store_code") or sid), "on_hand": on_hand, "avg_weekly": avg_weekly, "sigma": sigma, "weeks_with_sales": n_weeks, "safety_stock": safety_stock, "cycle_stock": cycle_stock, "optimal_stock": optimal, "overstock_units": overstock_units, "cost_price": cost if cost > 0 else None, "sell_price": sell if sell > 0 else None, "overstock_eur_cost": eur_cost, "overstock_eur_sell": eur_sell, "insufficient_history": insufficient, }) rows.sort(key=lambda r: r["overstock_eur_cost"], reverse=True) # Rollups — only over pairs with overstock_units > 0 with_over = [r for r in rows if r["overstock_units"] > 0] def _agg(rs: list[dict], key: str) -> list[dict]: buckets: dict[str, dict] = {} for r in rs: k = str(r.get(key) or "—") b = buckets.setdefault(k, {"label": k, "n_pairs": 0, "overstock_units": 0.0, "overstock_eur_cost": 0.0, "overstock_eur_sell": 0.0}) b["n_pairs"] += 1 b["overstock_units"] += r["overstock_units"] b["overstock_eur_cost"] += r["overstock_eur_cost"] b["overstock_eur_sell"] += r["overstock_eur_sell"] return sorted( buckets.values(), key=lambda b: b["overstock_eur_cost"], reverse=True, ) by_tier = _agg(with_over, "tier") by_store = _agg(with_over, "store_name") by_cat = _agg(with_over, "category") note = None if not rows: note = ( "No HR store stock for planned SKUs in this scope. Check " "tier filter or upload store-level stock data." ) elif n_insufficient and n_insufficient == len(rows): note = ( "All pairs flagged 'insufficient_history' (<4 non-promo " "weeks of HR retail sales). The σ source needs more history." ) return { "rows": rows, "n_pairs_with_overstock": n_with_overstock, "n_pairs_total": len(rows), "n_insufficient_history": n_insufficient, "total_overstock_units": total_units, "total_overstock_eur_cost": total_cost, "total_overstock_eur_sell": total_sell, "by_tier": by_tier, "by_store": by_store, "by_category": by_cat, "country": country, "min_weeks_required": MIN_WEEKS, "lead_time_weeks": LT, "note": note, } def _empty_store_overstock(*, country: str) -> dict: return { "rows": [], "n_pairs_with_overstock": 0, "n_pairs_total": 0, "n_insufficient_history": 0, "total_overstock_units": 0.0, "total_overstock_eur_cost": 0.0, "total_overstock_eur_sell": 0.0, "by_tier": [], "by_store": [], "by_category": [], "country": country, "min_weeks_required": 4, "lead_time_weeks": 1, "note": "No store stock rows for this country in erp_stock_current.", } # Mix _InventoryHealthMixin methods into SupplyService at module import time. SupplyService.get_inventory_health = _InventoryHealthMixin.get_inventory_health # type: ignore[attr-defined] SupplyService.get_store_overstock = _InventoryHealthMixin.get_store_overstock # type: ignore[attr-defined] # --------------------------------------------------------------------------- # MOQ Analysis # --------------------------------------------------------------------------- def _get_moq_analysis( self: SupplyService, *, tier: Optional[list[str]] = None, category: Optional[list[str]] = None, ) -> dict: rows = self.repo.get_moq_analysis(tier=tier, category=category) n_with_moq = sum(1 for r in rows if r.get("moq") is not None) n_missing_moq = len(rows) - n_with_moq note = None if not rows: note = "No planned SKUs found for this filter." elif n_missing_moq == len(rows): note = "No SKUs have MOQ data in supply_master. Upload supply_master to enable this view." return { "rows": rows, "n_with_moq": n_with_moq, "n_missing_moq": n_missing_moq, "note": note, } SupplyService.get_moq_analysis = _get_moq_analysis # type: ignore[attr-defined] # --------------------------------------------------------------------------- # Logistics # --------------------------------------------------------------------------- def _get_logistics(self: SupplyService, *, future_only: bool = True) -> dict: # Derive current year_week from the DB horizon horizon = self.repo.get_horizon_weeks(n_weeks=1) if future_only and horizon: min_yw: Optional[int] = horizon[0][0] * 100 + horizon[0][1] else: min_yw = None raw = self.repo.get_logistics(min_year_week=min_yw) if not raw: return { "by_supplier": [], "by_week": [], "total_qty": 0.0, "total_eur": None, "n_weeks": 0, "note": "No upcoming deliveries in incoming_supply.", } # Build by-supplier groups by_sup: dict[str, dict] = {} by_yw: dict[int, dict] = {} total_qty = 0.0 total_eur = 0.0 has_eur = False for r in raw: sup = str(r["supplier"] or "— unknown —") yw = int(r["year_week"]) week_int = int(r["week"]) cw_label = f"CW{week_int:02d}" qty = float(r["quantity"] or 0) eur = float(r["purchase_value"]) if r.get("purchase_value") is not None else None total_qty += qty if eur is not None: total_eur += eur has_eur = True # Supplier bucket sb = by_sup.setdefault(sup, { "supplier": sup, "n_skus": 0, "n_deliveries": 0, "total_qty": 0.0, "total_eur": 0.0, "has_eur": False, "skus_seen": set(), "deliveries": [], }) sb["n_deliveries"] += 1 sb["total_qty"] += qty if eur is not None: sb["total_eur"] += eur sb["has_eur"] = True sb["skus_seen"].add(r["sku"]) sb["deliveries"].append({ "sku": r["sku"], "name": r.get("name"), "tier": r.get("tier"), "year": int(r["year"]), "week": week_int, "year_week": yw, "cw_label": cw_label, "quantity": qty, "status": r.get("status"), "purchase_value": eur, }) # Week bucket wb = by_yw.setdefault(yw, { "year_week": yw, "cw_label": cw_label, "total_qty": 0.0, "total_eur": 0.0, "has_eur": False, "suppliers": set(), }) wb["total_qty"] += qty if eur is not None: wb["total_eur"] += eur wb["has_eur"] = True wb["suppliers"].add(sup) by_supplier_list = [] for sb in sorted(by_sup.values(), key=lambda x: x["total_qty"], reverse=True): by_supplier_list.append({ "supplier": sb["supplier"], "n_skus": len(sb["skus_seen"]), "n_deliveries": sb["n_deliveries"], "total_qty": sb["total_qty"], "total_eur": sb["total_eur"] if sb["has_eur"] else None, "deliveries": sb["deliveries"], }) by_week_list = [ { "year_week": wb["year_week"], "cw_label": wb["cw_label"], "total_qty": wb["total_qty"], "total_eur": wb["total_eur"] if wb["has_eur"] else None, "n_suppliers": len(wb["suppliers"]), } for wb in sorted(by_yw.values(), key=lambda x: x["year_week"]) ] return { "by_supplier": by_supplier_list, "by_week": by_week_list, "total_qty": total_qty, "total_eur": total_eur if has_eur else None, "n_weeks": len(by_yw), "note": None, } SupplyService.get_logistics = _get_logistics # type: ignore[attr-defined] # --------------------------------------------------------------------------- # Costs # --------------------------------------------------------------------------- def _get_costs( self: SupplyService, *, tier: Optional[list[str]] = None, category: Optional[list[str]] = None, planned_only: bool = False, ) -> dict: raw = self.repo.get_costs(tier=tier, category=category, planned_only=planned_only) rows = list(raw) total_cost = sum(float(r["stock_value_cost"] or 0) for r in rows) total_sell = sum(float(r["stock_value_sell"] or 0) for r in rows) n_with_cost = sum(1 for r in rows if r.get("cost_price") is not None) n_with_price = sum(1 for r in rows if r.get("avg_sell_price") is not None) # By-category rollup cat_map: dict[str, dict] = {} for r in rows: cat = str(r.get("category") or "— uncategorised —") cb = cat_map.setdefault(cat, { "category": cat, "n_skus": 0, "stock_value_cost": 0.0, "stock_value_sell": 0.0, "margins": [], }) cb["n_skus"] += 1 cb["stock_value_cost"] += float(r["stock_value_cost"] or 0) cb["stock_value_sell"] += float(r["stock_value_sell"] or 0) if r.get("margin_pct") is not None: cb["margins"].append(float(r["margin_pct"])) by_category = [] for cb in sorted(cat_map.values(), key=lambda x: x["stock_value_cost"], reverse=True): by_category.append({ "category": cb["category"], "n_skus": cb["n_skus"], "stock_value_cost": cb["stock_value_cost"], "stock_value_sell": cb["stock_value_sell"], "avg_margin_pct": (sum(cb["margins"]) / len(cb["margins"])) if cb["margins"] else None, }) note = None if not rows: note = "No cost or price data found. Upload erp_costs / erp_prices to populate." elif n_with_cost == 0: note = "No cost data in erp_costs. Margins unavailable." return { "rows": rows, "by_category": by_category, "total_stock_value_cost": total_cost, "total_stock_value_sell": total_sell, "n_with_cost": n_with_cost, "n_with_price": n_with_price, "note": note, } SupplyService.get_costs = _get_costs # type: ignore[attr-defined] # --------------------------------------------------------------------------- # Settings # --------------------------------------------------------------------------- def _get_settings(self: SupplyService) -> dict: rows = self.repo.get_settings() n_with = sum(1 for r in rows if r.get("supply_master_id") is not None) n_missing = len(rows) - n_with note = None if n_missing > 0: note = ( f"{n_missing} planned SKU(s) have no supply_master row. " "Use the edit table below to add lead time and MOQ." ) return { "rows": rows, "n_with_supply_master": n_with, "n_missing": n_missing, "note": note, } def _update_settings(self: SupplyService, updates: list[dict]) -> dict: return self.repo.upsert_supply_master(updates) SupplyService.get_settings = _get_settings # type: ignore[attr-defined] SupplyService.update_settings = _update_settings # type: ignore[attr-defined] def _format_action(s: dict) -> str: """Build the 'Suggested action' string for the Alerts UI. Includes qty + MOQ context where available so the planner can read the urgency without opening Order Entry.""" urgency = s["urgency"] qty = s["suggested_qty"] moq = s["moq"] if urgency == "Pull in": return f"Defer incoming · stock already covers {s['coverage_weeks']:.1f}w" if qty <= 0: return "No reorder needed" if moq and moq > 0: n_units = int(qty) n_moqs = int(round(qty / float(moq))) return f"Order {n_units:,} units ({n_moqs}× MOQ of {int(moq)})" return f"Order {int(qty):,} units"