"""Safety stock / reorder point math. The quantile forecasts feed inventory decisions here: safety_stock = z * demand_std * sqrt(lead_time_days) reorder_point = avg_daily_demand * lead_time_days + safety_stock suggested_order_qty = max(0, reorder_point - current_inventory) where, per DECISIONS.md: - avg_daily_demand = mean of the P50 forecast over the lead-time horizon - demand_std = mean of (p90 - p50) / 1.2816 over the horizon (1.2816 = z-value of the 90th percentile: if daily demand is ~normal, the P90-P50 spread recovers the daily std) - z = normal inverse CDF of the target service level """ import math from dataclasses import dataclass from typing import Sequence # z-value of the 90th percentile of the standard normal; converts the P90-P50 # forecast spread back into a daily demand standard deviation. Z_90 = 1.2816 @dataclass(frozen=True) class Recommendation: avg_daily_demand: float demand_std: float z: float safety_stock: float reorder_point: float suggested_order_qty: float service_level: float lead_time_days: int current_inventory: float def z_from_service_level(service_level: float) -> float: """Normal inverse CDF (probit). 0.95 -> 1.6449, 0.90 -> 1.2816, 0.50 -> 0.0. Uses Acklam's rational approximation (|error| < 1.15e-9) to avoid a scipy dependency in the serving image. """ if not 0.0 < service_level < 1.0: raise ValueError(f"service_level must be in (0, 1), got {service_level}") p = service_level # coefficients for Acklam's inverse normal CDF approximation a = (-3.969683028665376e01, 2.209460984245205e02, -2.759285104469687e02, 1.383577518672690e02, -3.066479806614716e01, 2.506628277459239e00) b = (-5.447609879822406e01, 1.615858368580409e02, -1.556989798598866e02, 6.680131188771972e01, -1.328068155288572e01) c = (-7.784894002430293e-03, -3.223964580411365e-01, -2.400758277161838e00, -2.549732539343734e00, 4.374664141464968e00, 2.938163982698783e00) d = (7.784695709041462e-03, 3.224671290700398e-01, 2.445134137142996e00, 3.754408661907416e00) p_low, p_high = 0.02425, 1 - 0.02425 if p < p_low: q = math.sqrt(-2 * math.log(p)) return (((((c[0] * q + c[1]) * q + c[2]) * q + c[3]) * q + c[4]) * q + c[5]) / ( (((d[0] * q + d[1]) * q + d[2]) * q + d[3]) * q + 1 ) if p > p_high: q = math.sqrt(-2 * math.log(1 - p)) return -(((((c[0] * q + c[1]) * q + c[2]) * q + c[3]) * q + c[4]) * q + c[5]) / ( (((d[0] * q + d[1]) * q + d[2]) * q + d[3]) * q + 1 ) q = p - 0.5 r = q * q return (((((a[0] * r + a[1]) * r + a[2]) * r + a[3]) * r + a[4]) * r + a[5]) * q / ( ((((b[0] * r + b[1]) * r + b[2]) * r + b[3]) * r + b[4]) * r + 1 ) def demand_std_from_quantiles(p50: Sequence[float], p90: Sequence[float]) -> float: """Daily demand std from the forecast quantile spread, averaged over the horizon.""" if len(p50) != len(p90) or len(p50) == 0: raise ValueError("p50 and p90 must be non-empty and the same length") spreads = [(hi - mid) / Z_90 for mid, hi in zip(p50, p90)] return max(0.0, sum(spreads) / len(spreads)) def recommend( p50: Sequence[float], p90: Sequence[float], current_inventory: float = 0.0, service_level: float = 0.95, lead_time_days: int = 7, ) -> Recommendation: """Reorder recommendation from quantile forecasts over the lead-time horizon. p50/p90 should cover at least `lead_time_days` days; only the first `lead_time_days` entries are used. """ if lead_time_days < 1: raise ValueError(f"lead_time_days must be >= 1, got {lead_time_days}") if len(p50) < lead_time_days: raise ValueError( f"need at least {lead_time_days} forecast days, got {len(p50)}" ) p50_h = list(p50[:lead_time_days]) p90_h = list(p90[:lead_time_days]) avg_daily_demand = sum(p50_h) / lead_time_days demand_std = demand_std_from_quantiles(p50_h, p90_h) z = z_from_service_level(service_level) safety_stock = z * demand_std * math.sqrt(lead_time_days) reorder_point = avg_daily_demand * lead_time_days + safety_stock suggested_order_qty = max(0.0, reorder_point - current_inventory) return Recommendation( avg_daily_demand=avg_daily_demand, demand_std=demand_std, z=z, safety_stock=safety_stock, reorder_point=reorder_point, suggested_order_qty=suggested_order_qty, service_level=service_level, lead_time_days=lead_time_days, current_inventory=current_inventory, )