"""LT-relative coverage classifier for "is this SKU at risk / healthy / overstocked". SCOPE — read before assuming this is used everywhere. As of 2026-05, the only callers are: - `executive_service` (Executive Stockout Risk) - `finance_service` (Locked Cash + Lost Sales) The Supply module deliberately does NOT use this classifier: - `supply_service._classify` uses ABSOLUTE weeks (<2 Order Now / <4 Order Soon / >13 Pull in) for the dashboard, stock projection, and coverage grid — those are the numbers long-time users know, kept stable on purpose. - `supply_service.get_alerts` / `_compute_suggestions` use a third, lead-time-aware variant for reorder alerts. So the same SKU can carry different status labels on Supply vs Executive/Finance pages. That divergence is known and accepted (the team chose stability of the Supply numbers over cross-page uniformity); it is documented here rather than silently "fixed". If you ever do want one classifier everywhere, migrate Supply's callers onto this function — the golden tests in backend/tests/ make that safe. The rule this function implements: effective_cover = (stock_now + incoming_in_LT) / weekly_demand ┌──────────────────┬──────────────────────────────────────────┐ │ effective_cover │ status │ ├──────────────────┼──────────────────────────────────────────┤ │ ≤ LT │ CRITICAL ("Order Now") │ │ LT < x ≤ 1.5×LT │ ORDER_SOON │ │ 1.5×LT < x ≤ 2×LT│ HEALTHY │ │ > 2×LT │ OVERSTOCK │ └──────────────────┴──────────────────────────────────────────┘ - `weekly_demand` is the per-week forecast number from `forecasts.total`; when a SKU isn't in the live run we fall back to its 13-week non-promo run-rate (same window everywhere in the codebase — see Q7). - `incoming_in_LT` is the sum of `incoming_supply.quantity` for the SKU inside the lead-time window — POs landing later don't help the cover-now question. - `LT` comes from `supply_master.lead_time_weeks`. When missing the SKU can't be classified at all and is bucketed as `MISSING_LT` so the data gap is visible rather than silently defaulted (see Q5). """ from __future__ import annotations from dataclasses import dataclass from typing import Literal, Optional CoverageStatus = Literal[ "CRITICAL", # ≤ LT — order now "ORDER_SOON", # LT < x ≤ 1.5×LT "HEALTHY", # 1.5×LT < x ≤ 2×LT "OVERSTOCK", # > 2×LT "MISSING_LT", # supply_master.lead_time_weeks unknown "NO_DEMAND", # demand ≤ 0 — can't compute weeks of cover ] @dataclass(frozen=True) class CoverageResult: status: CoverageStatus weeks_cover: Optional[float] # bare: stock_now / weekly_demand effective_weeks_cover: Optional[float] # (stock + incoming_in_LT) / weekly_demand lead_time_weeks: Optional[float] soon_threshold: Optional[float] # 1.5 × LT overstock_threshold: Optional[float] # 2 × LT incoming_in_lt: float # qty arriving within the LT window is_at_risk: bool # CRITICAL or ORDER_SOON is_overstock: bool # OVERSTOCK is_healthy: bool # HEALTHY def classify_coverage( *, stock_now: float, weekly_demand: float, lead_time_weeks: Optional[float], incoming_in_lt: float = 0.0, ) -> CoverageResult: """Classify a SKU's coverage position. See module docstring for the rule. All four inputs are expected to already be normalized: - stock_now: WH + store units (or just WH for warehouse-only views) - weekly_demand: per-week qty (forecast or 13w non-promo fallback) - lead_time_weeks: from supply_master, or None - incoming_in_lt: POs arriving in the next `lead_time_weeks` weeks The function is pure — no DB, no SQL. Callers prep the inputs. """ if lead_time_weeks is None or lead_time_weeks <= 0: return CoverageResult( status="MISSING_LT", weeks_cover=None, effective_weeks_cover=None, lead_time_weeks=None, soon_threshold=None, overstock_threshold=None, incoming_in_lt=float(incoming_in_lt), is_at_risk=False, is_overstock=False, is_healthy=False, ) if weekly_demand <= 0: return CoverageResult( status="NO_DEMAND", weeks_cover=None, effective_weeks_cover=None, lead_time_weeks=float(lead_time_weeks), soon_threshold=1.5 * lead_time_weeks, overstock_threshold=2.0 * lead_time_weeks, incoming_in_lt=float(incoming_in_lt), is_at_risk=False, is_overstock=False, is_healthy=False, ) lt = float(lead_time_weeks) bare = float(stock_now) / weekly_demand eff = (float(stock_now) + float(incoming_in_lt)) / weekly_demand soon = 1.5 * lt over = 2.0 * lt if eff <= lt: status: CoverageStatus = "CRITICAL" elif eff <= soon: status = "ORDER_SOON" elif eff <= over: status = "HEALTHY" else: status = "OVERSTOCK" return CoverageResult( status=status, weeks_cover=bare, effective_weeks_cover=eff, lead_time_weeks=lt, soon_threshold=soon, overstock_threshold=over, incoming_in_lt=float(incoming_in_lt), is_at_risk=status in ("CRITICAL", "ORDER_SOON"), is_overstock=(status == "OVERSTOCK"), is_healthy=(status == "HEALTHY"), ) def incoming_within_lt( horizon: list[tuple[int, int]], incoming_map: dict[tuple[int, int, int], float], pid: int, lead_time_weeks: float, ) -> float: """Sum incoming_supply for `pid` over the first `lead_time_weeks` of the horizon. Partial trailing week is prorated. Mirrors `_demand_over()` in supply_service so safety-stock math and the coverage classifier consume incoming the same way. """ if lead_time_weeks <= 0 or not horizon: return 0.0 whole = int(lead_time_weeks) frac = lead_time_weeks - whole total = 0.0 for i, (y, w) in enumerate(horizon): if i < whole: total += float(incoming_map.get((pid, y, w), 0.0)) elif i == whole and frac > 0: total += frac * float(incoming_map.get((pid, y, w), 0.0)) else: break return total