app/inventory_math.py

"""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,
    )