File size: 5,697 Bytes
221080b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | """Single-process cache for the gold / audit artifacts the API serves.
The pipeline emits parquet + CSV files under data/gold and outputs/audit.
This module loads them once at process startup and exposes typed accessors.
The FastAPI app holds the cache as application state so endpoints share
the same in-memory frames.
"""
from __future__ import annotations
import os
from dataclasses import dataclass, field
from functools import lru_cache
from pathlib import Path
from typing import Optional
import pandas as pd
def _repo_root() -> Path:
"""Locate the repository root from REPO_ROOT env var or by walking up."""
env_root = os.environ.get("REPO_ROOT")
if env_root:
return Path(env_root).resolve()
here = Path(__file__).resolve()
for parent in here.parents:
if (parent / "data" / "gold").is_dir():
return parent
raise RuntimeError("Could not locate repo root containing data/gold/")
@dataclass
class DataCache:
"""In-memory snapshot of all artifacts the API needs."""
repo_root: Path
features: pd.DataFrame = field(default_factory=pd.DataFrame)
predictions: pd.DataFrame = field(default_factory=pd.DataFrame)
counterfactuals: pd.DataFrame = field(default_factory=pd.DataFrame)
shap_drivers: pd.DataFrame = field(default_factory=pd.DataFrame)
shap_global: pd.DataFrame = field(default_factory=pd.DataFrame)
budget: pd.DataFrame = field(default_factory=pd.DataFrame)
budget_channels: pd.DataFrame = field(default_factory=pd.DataFrame)
budget_by_distributor: pd.DataFrame = field(default_factory=pd.DataFrame)
cooler_roi_full: pd.DataFrame = field(default_factory=pd.DataFrame)
cooler_roi_top100: pd.DataFrame = field(default_factory=pd.DataFrame)
dormancy: pd.DataFrame = field(default_factory=pd.DataFrame)
dormancy_top: pd.DataFrame = field(default_factory=pd.DataFrame)
scorecard: pd.DataFrame = field(default_factory=pd.DataFrame)
territories: pd.DataFrame = field(default_factory=pd.DataFrame)
cluster_membership: pd.DataFrame = field(default_factory=pd.DataFrame)
outlet_actions: pd.DataFrame = field(default_factory=pd.DataFrame)
forensics: pd.DataFrame = field(default_factory=pd.DataFrame)
bronze_master: pd.DataFrame = field(default_factory=pd.DataFrame)
rejected_coords: pd.DataFrame = field(default_factory=pd.DataFrame)
@property
def outlets_table(self) -> pd.DataFrame:
"""Returns the canonical outlet table the listing pages render."""
if self._outlets_table is None:
self._outlets_table = self._build_outlets_table()
return self._outlets_table
_outlets_table: Optional[pd.DataFrame] = None
def _build_outlets_table(self) -> pd.DataFrame:
cols = [
"Outlet_ID", "Outlet_Type", "Outlet_Size", "Distributor_ID", "Province",
"Cooler_Count", "Latitude", "Longitude",
"active_months", "monthly_volume_mean", "monthly_volume_q90",
"competitors_1km", "hhi_1500m", "spatial_demand_score",
"replenishment_friction",
]
df = self.features[[c for c in cols if c in self.features.columns]].copy()
df = df.merge(self.predictions, on="Outlet_ID", how="left")
if not self.dormancy.empty:
df = df.merge(self.dormancy[["Outlet_ID", "dormancy_risk_score", "risk_band"]],
on="Outlet_ID", how="left")
if not self.cluster_membership.empty:
df = df.merge(self.cluster_membership, on="Outlet_ID", how="left")
if not self.budget.empty:
df = df.merge(self.budget, on="Outlet_ID", how="left")
return df
def _read_parquet(p: Path) -> pd.DataFrame:
return pd.read_parquet(p) if p.exists() else pd.DataFrame()
def _read_csv(p: Path) -> pd.DataFrame:
return pd.read_csv(p) if p.exists() else pd.DataFrame()
@lru_cache(maxsize=1)
def get_cache() -> DataCache:
root = _repo_root()
gold = root / "data" / "gold"
out = root / "outputs"
audit = out / "audit"
cache = DataCache(repo_root=root)
cache.features = _read_parquet(gold / "outlet_features.parquet")
cache.predictions = _read_csv(out / "DataX_predictions.csv")
cache.counterfactuals = _read_parquet(gold / "counterfactuals.parquet")
cache.shap_drivers = _read_csv(audit / "shap_top_drivers_per_outlet.csv")
cache.shap_global = _read_csv(audit / "shap_global_importance.csv")
cache.budget = _read_csv(out / "DataX_budget_allocations.csv")
cache.budget_channels = _read_csv(audit / "budget_allocation_by_channel.csv")
cache.budget_by_distributor = _read_csv(audit / "budget_allocation_by_distributor.csv")
cache.cooler_roi_full = _read_csv(audit / "cooler_roi_full.csv")
cache.cooler_roi_top100 = _read_csv(audit / "cooler_roi_top100.csv")
cache.dormancy = _read_parquet(gold / "dormancy_risk.parquet")
cache.dormancy_top = _read_csv(audit / "dormancy_top200_at_risk.csv")
cache.scorecard = _read_csv(audit / "distributor_scorecard.csv")
cache.territories = _read_csv(audit / "territory_clusters_summary.csv")
cache.cluster_membership = _read_parquet(gold / "outlet_clusters.parquet")
cache.outlet_actions = _read_parquet(gold / "outlet_actions.parquet")
cache.forensics = _read_csv(audit / "forensics_findings.csv")
bronze = root / "data" / "bronze"
quarantine = root / "data" / "silver" / "quarantine"
cache.bronze_master = _read_csv(bronze / "outlet_master.csv")
cache.rejected_coords = _read_parquet(quarantine / "outlet_coordinates_rejected.parquet")
return cache
|