"""Forecast-engine orchestration (v3.7-db). Engine refactor: subprocess removed. The engine now exposes `run()` which reads from Postgres directly (with CSV fallback), writes results to the `forecasts` + `factor_history` tables, and returns a dict. We invoke it in-process via a single-worker ThreadPoolExecutor so the FastAPI event loop stays responsive. Pipeline: 1. Pre-flight (forecast_engine module + DB connectivity) 2. Submit run() to ThreadPoolExecutor — non-blocking 3. Wait on the future (HTTP request holds open for the run) 4. Engine itself writes to forecasts/factor_history — wrapper just reports the run_id back This eliminates: - subprocess overhead - forecast_log.csv → forecasts ingestion (engine writes DB directly) - factor_history.csv → factor_history ingestion (same) - cwd management (engine uses CSV paths only as fallback) """ from __future__ import annotations import logging import os import sys import time from concurrent.futures import ThreadPoolExecutor from datetime import datetime from pathlib import Path from typing import Optional from sqlalchemy import text from sqlalchemy.orm import Session from backend.repositories.base import BaseRepository logger = logging.getLogger(__name__) PROJECT_ROOT = Path(__file__).resolve().parents[2] DATA_DIR = PROJECT_ROOT / "data" # Required inputs — engine still reads these as CSV-first fallback. When DB # is the source of truth they're optional; for current state we still want # them present (full historical sales etc.) PLANNING_BOOK = DATA_DIR / "Polleo_Demand_Planning_Book.xlsx" LAST_PLAN = DATA_DIR / "Polleo_Demand_Plan.xlsx" SALES_CLEAN = DATA_DIR / "sales_clean.csv" # Single-worker pool so only one forecast can run at a time (heavy memory) _executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="fc-engine") def _get_iso_year_week(d: Optional[datetime] = None) -> tuple[int, int]: iso = (d or datetime.now()).isocalendar() return int(iso[0]), int(iso[1]) def _check_prerequisites() -> tuple[bool, list[str]]: """forecast_v4 prereqs. The v4 engine is DB-driven, so the checks are: (1) the forecast_v4 package imports, (2) sku_planning has tiered SKUs to forecast, (3) there is sales history in the DB. No CSV/workbook needed.""" missing: list[str] = [] try: from forecast_v4.run import run_v4 # noqa: F401 except Exception as exc: missing.append(f"forecast_v4 import failed: {exc}") try: from sqlalchemy import text as _t from backend.models.database import engine as _eng with _eng.connect() as c: n_plan = c.execute(_t( "SELECT count(*) FROM sku_planning WHERE tier IS NOT NULL")).scalar() or 0 n_sales = c.execute(_t("SELECT count(*) FROM erp_transactions")).scalar() or 0 if n_plan == 0: missing.append("sku_planning has no tiered SKUs — populate the planning list first") if n_sales == 0: missing.append("no sales history in erp_transactions — run Upload sales first") except Exception as exc: missing.append(f"DB check failed: {exc}") return (len(missing) == 0), missing class ForecastService(BaseRepository): # ----- Public: pre-flight check --------------------------------------- @staticmethod def check_prereqs() -> dict: """v4 pre-flight. The response booleans are repurposed for the v4 engine (the frontend labels them accordingly): engine_exists -> forecast_v4 package imports planning_book_exists -> sku_planning has tiered SKUs sales_clean_exists -> sales history present in DB.""" ok, missing = _check_prerequisites() engine_ok = not any("forecast_v4 import" in m for m in missing) plan_ok = not any("sku_planning" in m for m in missing) sales_ok = not any("erp_transactions" in m or "sales history" in m for m in missing) return { "ok": ok, "missing": missing, "planning_book_exists": plan_ok, "last_plan_exists": plan_ok, "sales_clean_exists": sales_ok, "engine_exists": engine_ok, "data_dir": str(DATA_DIR), "db_available": engine_ok and (plan_ok or sales_ok), } # ----- Public: run forecast in-process -------------------------------- def run_forecast( self, *, run_type: str = "baseline", run_by_id: Optional[int] = None, generate_xlsx: bool = False, ) -> dict: """Submit engine.run() to the worker pool and block on the result. generate_xlsx=False (default) skips the engine's Excel build — the on-demand /api/demand/download-plan endpoint generates a DB-backed xlsx instead. Set True to also write data/Polleo_Demand_Plan.xlsx (legacy Monika workflow). """ # forecast_v4 bottom-up engine (per-buyer archetype routing + KAM/CM # on-top folding). Runs in the single-worker pool so concurrent runs # serialize and the FastAPI event loop isn't blocked. run_v4 reads from # and writes to the DB directly and allocates forecasts_detail itself. t0 = time.time() try: future = _executor.submit(_invoke_v4, run_by_id, run_type) result = future.result(timeout=25 * 60) # v4 ~3-5 min for ~500 SKUs except Exception as exc: return { "success": False, "run_id": None, "cycle_id": None, "sku_count": 0, "rows_inserted": 0, "factor_history_inserted": 0, "duration_seconds": int(time.time() - t0), "warnings": [], "error": f"forecast_v4 raised: {type(exc).__name__}: {exc}", "stdout_tail": None, } return { "success": True, "run_id": result.get('run_id'), "cycle_id": None, # run_v4 creates the cycle internally "sku_count": int(result.get('n_skus') or 0), "rows_inserted": int(result.get('forecast_rows') or 0), "factor_history_inserted": int(result.get('detail_rows') or 0), "duration_seconds": int(time.time() - t0), "warnings": [], "error": None, "stdout_tail": None, } # ----- Recent runs (unchanged) --------------------------------------- def get_recent_runs(self, *, limit: int = 20) -> list[dict]: rows = self.db.execute(text(""" SELECT fr.id, fr.run_type, fr.n_skus, fr.started_at, sc.year_week, COALESCE(NULLIF(u.display_name, ''), u.username) AS run_by FROM forecast_runs fr LEFT JOIN sop_cycles sc ON sc.id = fr.cycle_id LEFT JOIN users u ON u.id = fr.run_by_id ORDER BY fr.started_at DESC NULLS LAST, fr.id DESC LIMIT :limit """), {"limit": limit}).mappings().all() out: list[dict] = [] for r in rows: t = r.get("started_at") t_str = t.isoformat() if t is not None and hasattr(t, "isoformat") else (str(t) if t else None) out.append({ "run_id": int(r["id"]), "run_type": r.get("run_type"), "n_skus": int(r.get("n_skus") or 0), "year_week": int(r.get("year_week") or 0), "started_at": t_str, "run_by": r.get("run_by"), }) return out # ----- Plan download/upload ------------------------------------------ @staticmethod def generate_plan_xlsx(run_id: Optional[int] = None) -> tuple[bytes, int]: """Return (xlsx bytes, resolved_run_id) — the rich v4 Demand Plan workbook (Demand Planning / Forecast Detail / Demand Output Total / On-Top), built from forecasts + on_top_inputs + actual sales.""" from forecast_v4.plan_workbook import build_plan_workbook return build_plan_workbook(run_id=run_id) @staticmethod def apply_plan_upload(file_bytes: bytes) -> dict: """Parse the edited Demand Planning sheet (per-week Planner Factor) and apply it to the run, recomputing total = baseline*factor + on-tops.""" from forecast_v4.plan_workbook import parse_plan_workbook, apply_plan_factors parsed = parse_plan_workbook(file_bytes) warnings = parsed.get("warnings", []) updates = parsed.get("updates", []) if not updates: return { "success": False, "error": "No planner-factor changes parsed from the Demand Planning sheet", "run_id": parsed.get("run_id"), "warnings": warnings, "n_skus_updated": 0, "n_rows_updated": 0, "factors_changed": [], } res = apply_plan_factors(parsed.get("run_id"), updates) return { "success": True, "run_id": res["run_id"], "n_skus_updated": res["n_skus_updated"], "n_rows_updated": res["n_rows_updated"], "factors_changed": res["factors_changed"], "warnings": warnings, } # --------------------------------------------------------------------------- # Module-level worker function (so it pickles cleanly in the thread pool) # --------------------------------------------------------------------------- def _invoke_v4(run_by_id: Optional[int], run_type: str = "with_ontop") -> dict: """Run the forecast_v4 bottom-up engine in the worker thread. 'baseline' skips KAM/CM on-top folding; 'with_ontop' folds them in. Returns {run_id, n_skus, forecast_rows, detail_rows}.""" from forecast_v4.run import run_v4 keep = run_type != "baseline" # baseline = clean machine forecast return run_v4(run_by_id=run_by_id, fold_ontops=keep, carry_factors=keep) def _invoke_engine(run_type: str, run_by_id: Optional[int], generate_xlsx: bool) -> dict: """Imported here to avoid circular issues during module load.""" # sys.path already configured by caller import forecast_engine return forecast_engine.run( input_file=None, generate_xlsx=generate_xlsx, run_type=run_type, run_by_id=run_by_id, )