"""RestockIQ FastAPI app: serves the precomputed forecasts, reorder recommendations, and the offline backtest comparison. Nothing is trained at request time — LightGBM outputs are precomputed into data/forecasts.parquet by the pipeline. """ import json import math from contextlib import asynccontextmanager import pandas as pd from fastapi import FastAPI, HTTPException, Query from fastapi.responses import FileResponse from fastapi.staticfiles import StaticFiles from app import config as cfg from app.inventory_math import Z_90, recommend, z_from_service_level @asynccontextmanager async def lifespan(_app: FastAPI): load_data() yield app = FastAPI(title="RestockIQ", version="1.1", lifespan=lifespan) # ---- data loaded once at startup (small, precomputed) ---------------------------- _forecasts: pd.DataFrame | None = None _history: pd.DataFrame | None = None _backtest_summary: dict | None = None _per_sku: pd.DataFrame | None = None # memoized action-list responses (underlying data is immutable per process) _action_cache: dict = {} ACTION_CACHE_MAX = 256 def load_data() -> None: global _forecasts, _history, _backtest_summary, _per_sku _forecasts = pd.read_parquet(cfg.FORECASTS_PARQUET) _forecasts = _forecasts.set_index(["store_id", "item_id"]).sort_index() history_path = cfg.DATA_DIR / "history_tail.parquet" if history_path.exists(): _history = pd.read_parquet(history_path) _history = _history.set_index(["store_id", "item_id"]).sort_index() if cfg.BACKTEST_SUMMARY_JSON.exists(): _backtest_summary = json.loads(cfg.BACKTEST_SUMMARY_JSON.read_text()) if cfg.BACKTEST_PER_SKU_PARQUET.exists(): _per_sku = pd.read_parquet(cfg.BACKTEST_PER_SKU_PARQUET) def _series_or_404(store: str, item: str) -> pd.DataFrame: try: rows = _forecasts.loc[(store, item)] except KeyError: raise HTTPException(status_code=404, detail=f"unknown series {store}/{item}") return rows.sort_values("d") # ---- API --------------------------------------------------------------------------- @app.get("/health") def health() -> dict: return {"status": "ok"} @app.get("/api/skus") def skus() -> dict: """Available (store, item) pairs. M5 is a full cross product, so the dropdowns are served as separate store and item lists; every combination is valid.""" idx = _forecasts.index return { "stores": sorted(idx.get_level_values(0).unique().tolist()), "items": sorted(idx.get_level_values(1).unique().tolist()), } @app.get("/api/forecast/{store}/{item}") def forecast(store: str, item: str) -> dict: rows = _series_or_404(store, item) out = { "dates": rows["date"].dt.strftime("%Y-%m-%d").tolist(), "p10": [round(float(v), 3) for v in rows["p10"]], "p50": [round(float(v), 3) for v in rows["p50"]], "p90": [round(float(v), 3) for v in rows["p90"]], "actual": [int(v) for v in rows["actual"]], } if _history is not None: try: h = _history.loc[(store, item)].sort_values("d") out["history_dates"] = h["date"].dt.strftime("%Y-%m-%d").tolist() out["history_actual"] = [int(v) for v in h["actual"]] except KeyError: pass return out @app.get("/api/recommendation/{store}/{item}") def recommendation( store: str, item: str, current_inventory: float = Query( cfg.DEFAULT_CURRENT_INVENTORY, ge=0, description="Illustrative on-hand inventory (dataset has no real inventory column). Default 0.", ), service_level: float = Query( cfg.DEFAULT_SERVICE_LEVEL, gt=0, lt=1, description="Target cycle service level. Default 0.95.", ), lead_time_days: int = Query( cfg.DEFAULT_LEAD_TIME_DAYS, ge=1, le=cfg.HOLDOUT_DAYS, description="Assumed replenishment lead time in days. Default 7.", ), ) -> dict: rows = _series_or_404(store, item) rec = recommend( p50=rows["p50"].tolist(), p90=rows["p90"].tolist(), current_inventory=current_inventory, service_level=service_level, lead_time_days=lead_time_days, ) return { "reorder_point": round(rec.reorder_point, 2), "safety_stock": round(rec.safety_stock, 2), "suggested_order_qty": round(rec.suggested_order_qty, 2), "service_level": rec.service_level, "avg_daily_demand": round(rec.avg_daily_demand, 3), "demand_std": round(rec.demand_std, 3), "lead_time_days": rec.lead_time_days, "current_inventory": rec.current_inventory, } @app.get("/api/action_list/{store}") def action_list( store: str, current_inventory: float = Query( cfg.DEFAULT_CURRENT_INVENTORY, ge=0, description="Illustrative on-hand units assumed for EVERY product. Default 0.", ), service_level: float = Query(cfg.DEFAULT_SERVICE_LEVEL, gt=0, lt=1), lead_time_days: int = Query(cfg.DEFAULT_LEAD_TIME_DAYS, ge=1, le=cfg.HOLDOUT_DAYS), limit: int = Query(25, ge=1, le=200), ) -> dict: """Store-wide 'what needs ordering now' list: every product ranked by suggested order size, with a traffic-light urgency status. Results are memoized per parameter set: the forecast table is static for the life of the process, so identical requests (the common case — every page load asks for the default assumptions) skip the 3,049-item groupby entirely. """ cache_key = (store, current_inventory, service_level, lead_time_days, limit) if cache_key in _action_cache: return _action_cache[cache_key] df = _forecasts.loc[store] if df.empty: raise HTTPException(status_code=404, detail=f"unknown store {store}") z = z_from_service_level(service_level) lead = lead_time_days win = df[df["d"] <= int(df["d"].min()) + lead - 1] g = win.groupby(level="item_id", observed=True) avg_daily = g["p50"].mean() demand_std = ((g["p90"].mean() - g["p50"].mean()) / Z_90).clip(lower=0) safety = z * demand_std * math.sqrt(lead) rop = avg_daily * lead + safety qty = (rop - current_inventory).clip(lower=0) # urgency: red = stock won't survive the lead time; amber = below reorder point lead_demand = avg_daily * lead status = pd.Series("ok", index=rop.index) status[current_inventory <= rop] = "order_soon" status[current_inventory <= lead_demand] = "order_now" status[qty <= 0] = "ok" out = pd.DataFrame( { "item_id": rop.index, "suggested_order_qty": qty.round(1), "expected_daily_sales": avg_daily.round(2), "safety_cushion": safety.round(1), "status": status, } ).sort_values("suggested_order_qty", ascending=False) result = { "store": store, "assumptions": { "current_inventory_per_item": current_inventory, "service_level": service_level, "lead_time_days": lead, }, "counts": status.value_counts().to_dict(), "items": out.head(limit).to_dict(orient="records"), "total_items": len(out), } if len(_action_cache) >= ACTION_CACHE_MAX: _action_cache.pop(next(iter(_action_cache))) # evict oldest (FIFO) _action_cache[cache_key] = result return result @app.get("/api/backtest_summary") def backtest_summary() -> dict: if _backtest_summary is None: raise HTTPException(status_code=503, detail="backtest not computed yet") out = dict(_backtest_summary) if _per_sku is not None: top = _per_sku.sort_values("total_demand", ascending=False).head(20) out["top_skus"] = top.to_dict(orient="records") return out # ---- static frontend (mounted last so /api and /health win) ----------------------- @app.get("/") def index() -> FileResponse: return FileResponse(cfg.ROOT / "app" / "static" / "index.html") app.mount("/", StaticFiles(directory=cfg.ROOT / "app" / "static"), name="static")