the model path i added and refined

api/app.py

@@ -1,76 +1,78 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 import joblib
 import numpy as np
 import pandas as pd
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel, Field
 import xgboost as xgb
-
-import os
+from fastapi import FastAPI, HTTPException
 from huggingface_hub import hf_hub_download
+from pydantic import BaseModel, Field
 
 # Compatibility shim for pickles created with newer sklearn that include _RemainderColsList
 import sklearn.compose._column_transformer as _ct  # type: ignore
+
 if not hasattr(_ct, "_RemainderColsList"):
     class _RemainderColsList(list):  # type: ignore
         pass
+
     _ct._RemainderColsList = _RemainderColsList
 
+
+# -----------------------------
+# Paths & configuration
+# -----------------------------
 ROOT = Path(__file__).resolve().parents[1]
 MODEL_DIR = Path(__file__).resolve().parent / "model"
-# MODEL_PATH = MODEL_DIR / "xgboost_pipeline.pkl"
+
 BOOSTER_PATH = MODEL_DIR / "xgboost_booster.json"
 META_PATH = MODEL_DIR / "explain_meta.json"
 
 HF_MODEL_REPO = os.getenv("HF_MODEL_REPO", "Gutema/frankscore-model-artifact")
 HF_MODEL_REVISION = os.getenv("HF_MODEL_REVISION", "main")
+HF_MODEL_FILENAME = os.getenv("HF_MODEL_FILENAME", "xgboost_pipeline.pkl")
+
 
-try:
-    MODEL_PATH = Path(
-        hf_hub_download(
+def download_pipeline_artifact() -> Path:
+    """
+    Download the .pkl artifact from Hugging Face Hub (cached locally).
+    """
+    try:
+        p = hf_hub_download(
             repo_id=HF_MODEL_REPO,
-            filename="xgboost_pipeline.pkl",
+            filename=HF_MODEL_FILENAME,
             revision=HF_MODEL_REVISION,
         )
-    )
-except Exception as e:
-    raise RuntimeError(f"Failed to download model artifact from HF repo={HF_MODEL_REPO}: {e}") from e
-
-if not META_PATH.exists():
-    raise FileNotFoundError(f"Explainability meta missing at {META_PATH}")
-if not BOOSTER_PATH.exists():
-    raise FileNotFoundError(f"Booster file missing at {BOOSTER_PATH}")
-
-if not MODEL_PATH.exists():
-    raise FileNotFoundError(f"Model file missing at {MODEL_PATH}")
-if not META_PATH.exists():
-    raise FileNotFoundError(f"Explainability meta missing at {META_PATH}")
-if not BOOSTER_PATH.exists():
-    raise FileNotFoundError(f"Booster file missing at {BOOSTER_PATH}")
-
-PIPELINE = joblib.load(MODEL_PATH)
-META = json.loads(META_PATH.read_text())
-
-EXPECTED_FEATURES = list(getattr(PIPELINE, "feature_names_in_", []))
-PREPROCESS = PIPELINE.named_steps.get("preprocess") if hasattr(PIPELINE, "named_steps") else None
-if PREPROCESS is None:
-    raise RuntimeError("Pipeline missing 'preprocess' step; cannot infer columns.")
-
-if not EXPECTED_FEATURES:
-    EXPECTED_FEATURES = list(getattr(PREPROCESS, "feature_names_in_", []))
-    if not EXPECTED_FEATURES:
-        raise RuntimeError("Unable to determine expected feature names from the pipeline.")
+        return Path(p)
+    except Exception as e:
+        raise RuntimeError(
+            f"Failed to download model artifact from HF repo={HF_MODEL_REPO} "
+            f"revision={HF_MODEL_REVISION} filename={HF_MODEL_FILENAME}: {e}"
+        ) from e
+
 
-_col_map = {name: cols for name, _, cols in getattr(PREPROCESS, "transformers_", [])}
-NUM_FEATURES = list(_col_map.get("num", []))
-CAT_FEATURES = list(_col_map.get("cat", []))
-PRE_FEATURE_NAMES = META.get("pre_feature_names") or list(getattr(PREPROCESS, "get_feature_names_out", lambda: [])())
+def require_local_file(p: Path, label: str) -> None:
+    if not p.exists():
+        raise FileNotFoundError(f"{label} missing at {p}")
+
+
+# -----------------------------
+# Load meta (local JSON)
+# -----------------------------
+require_local_file(META_PATH, "Explainability meta")
+require_local_file(BOOSTER_PATH, "Booster file")
+META: Dict[str, Any] = json.loads(META_PATH.read_text())
+
+
+# -----------------------------
+# Feature groups (unchanged)
+# -----------------------------
 RAW_FEATURE_SET = set((META.get("raw_num_cols") or []) + (META.get("raw_cat_cols") or []))
+
 FEATURE_GROUPS = {
     "Borrowing History & Maturity": [
         "account_age_days",
@@ -122,14 +124,62 @@ FEATURE_GROUPS = {
         "latest_amount_ma3",
     ],
 }
+
 FEATURE_GROUP_LOOKUP: Dict[str, str] = {}
 for group, variables in FEATURE_GROUPS.items():
     for var in variables:
         FEATURE_GROUP_LOOKUP[var] = group
 
+
+# -----------------------------
+# FastAPI app
+# -----------------------------
 app = FastAPI(title="FrankScore", version="1.0.0")
 
 
+# Globals populated at startup
+PIPELINE = None
+PREPROCESS = None
+EXPECTED_FEATURES: List[str] = []
+NUM_FEATURES: List[str] = []
+CAT_FEATURES: List[str] = []
+PRE_FEATURE_NAMES: List[str] = []
+
+
+@app.on_event("startup")
+def _startup() -> None:
+    """
+    Download + load pipeline on startup (safer than import-time).
+    """
+    global PIPELINE, PREPROCESS, EXPECTED_FEATURES, NUM_FEATURES, CAT_FEATURES, PRE_FEATURE_NAMES
+
+    model_path = download_pipeline_artifact()
+    PIPELINE = joblib.load(model_path)
+
+    EXPECTED_FEATURES = list(getattr(PIPELINE, "feature_names_in_", []))
+
+    PREPROCESS = PIPELINE.named_steps.get("preprocess") if hasattr(PIPELINE, "named_steps") else None
+    if PREPROCESS is None:
+        raise RuntimeError("Pipeline missing 'preprocess' step; cannot infer columns.")
+
+    if not EXPECTED_FEATURES:
+        EXPECTED_FEATURES = list(getattr(PREPROCESS, "feature_names_in_", []))
+        if not EXPECTED_FEATURES:
+            raise RuntimeError("Unable to determine expected feature names from the pipeline.")
+
+    _col_map = {name: cols for name, _, cols in getattr(PREPROCESS, "transformers_", [])}
+    NUM_FEATURES = list(_col_map.get("num", []))
+    CAT_FEATURES = list(_col_map.get("cat", []))
+
+    # From meta if present; fallback to preprocess get_feature_names_out
+    PRE_FEATURE_NAMES = META.get("pre_feature_names") or list(
+        getattr(PREPROCESS, "get_feature_names_out", lambda: [])()
+    )
+
+
+# -----------------------------
+# Schemas
+# -----------------------------
 class PredictionRequest(BaseModel):
     records: List[Dict[str, Any]] = Field(..., description="List of borrower feature dictionaries")
 
@@ -183,18 +233,32 @@ class PredictExplainResponse(BaseModel):
     results: List[PredictExplainItem]
 
 
+# -----------------------------
+# Helpers
+# -----------------------------
+def _require_loaded() -> None:
+    if PIPELINE is None or PREPROCESS is None:
+        raise HTTPException(status_code=503, detail="Model not loaded yet. Please retry.")
+
+
 def prepare_frame(records: List[Dict[str, Any]]) -> pd.DataFrame:
+    _require_loaded()
+
     if not records:
         raise HTTPException(status_code=400, detail="No records provided.")
     df = pd.DataFrame(records)
+
     for col in EXPECTED_FEATURES:
         if col not in df.columns:
             df[col] = np.nan
+
     df = df[EXPECTED_FEATURES]
+
     if NUM_FEATURES:
         df[NUM_FEATURES] = df[NUM_FEATURES].apply(pd.to_numeric, errors="coerce")
     if CAT_FEATURES:
         df[CAT_FEATURES] = df[CAT_FEATURES].astype("object")
+
     return df
 
 
@@ -229,10 +293,11 @@ def _base_feature_name(name: str) -> str:
     return base
 
 
-def get_booster():
+def get_booster() -> xgb.Booster:
     if not hasattr(get_booster, "_booster"):
         booster = xgb.Booster()
         booster.load_model(str(BOOSTER_PATH))
+
         base_score = booster.attr("base_score")
         if base_score:
             try:
@@ -245,22 +310,36 @@ def get_booster():
                     cleaned_val = "0.5"
                 booster.set_param({"base_score": cleaned_val})
                 booster.set_attr(base_score=cleaned_val)
+
         get_booster._booster = booster
     return get_booster._booster
 
 
+# -----------------------------
+# Endpoints
+# -----------------------------
+@app.get("/health")
+def health() -> Dict[str, str]:
+    # Do not crash health if model isn't loaded yet
+    return {
+        "status": "ok",
+        "hf_repo": HF_MODEL_REPO,
+        "hf_revision": HF_MODEL_REVISION,
+        "hf_filename": HF_MODEL_FILENAME,
+        "meta_path": str(META_PATH),
+        "booster_path": str(BOOSTER_PATH),
+        "loaded": str(PIPELINE is not None),
+    }
+
+
 @app.post("/predict", response_model=PredictionResponse)
 def predict(req: PredictionRequest) -> PredictionResponse:
+    _require_loaded()
     frame = prepare_frame(req.records)
     probas = PIPELINE.predict_proba(frame)[:, 1]
     return PredictionResponse(probabilities=probas.tolist())
 
 
-@app.get("/health")
-def health() -> Dict[str, str]:
-    return {"status": "ok", "model_path": str(MODEL_PATH)}
-
-
 @app.post("/score", response_model=ScoreResponse)
 def score(req: ScoreRequest) -> ScoreResponse:
     if not req.probabilities:
@@ -272,38 +351,51 @@ def score(req: ScoreRequest) -> ScoreResponse:
 
 @app.post("/explain", response_model=ExplainResponse)
 def explain(req: ExplainRequest) -> ExplainResponse:
+    _require_loaded()
     if not req.records:
         raise HTTPException(status_code=400, detail="No records provided.")
+
     frame = prepare_frame(req.records)
     probas = PIPELINE.predict_proba(frame)[:, 1]
+
     booster = get_booster()
     X_proc = PREPROCESS.transform(frame)
+
     feat_names = np.array(PRE_FEATURE_NAMES) if PRE_FEATURE_NAMES else np.array([f"f{i}" for i in range(X_proc.shape[1])])
     sanitized_names = [_sanitize_feature_name(n) for n in feat_names]
+
     dmat = xgb.DMatrix(X_proc, feature_names=sanitized_names)
     contribs = booster.predict(dmat, pred_contribs=True)
+
     if contribs.shape[1] != X_proc.shape[1] + 1:
         raise HTTPException(status_code=500, detail="Unexpected contribution shape from booster.")
+
     base_vals = contribs[:, -1]
     feat_contribs = contribs[:, :-1]
+
     explanations: List[ExplainItem] = []
     for i in range(feat_contribs.shape[0]):
         row_vals = feat_contribs[i]
+
         group_totals: Dict[str, float] = {}
         group_details: Dict[str, List[FeatureContribution]] = {}
+
         for name, val in zip(feat_names, row_vals):
             base = _base_feature_name(str(name))
             group = FEATURE_GROUP_LOOKUP.get(base, "Other")
+
             group_totals[group] = group_totals.get(group, 0.0) + float(val)
             group_details.setdefault(group, []).append(
                 FeatureContribution(feature=str(name), shap_value=float(val))
             )
+
         group_contribs: List[GroupContribution] = []
         for grp, total in sorted(group_totals.items(), key=lambda kv: abs(kv[1]), reverse=True):
             feats = sorted(group_details.get(grp, []), key=lambda fc: abs(fc.shap_value), reverse=True)
             if req.top_k:
-                feats = feats[:req.top_k]
+                feats = feats[: req.top_k]
             group_contribs.append(GroupContribution(group=grp, total_shap_value=total, features=feats))
+
         explanations.append(
             ExplainItem(
                 probability=float(probas[i]),
@@ -311,44 +403,59 @@ def explain(req: ExplainRequest) -> ExplainResponse:
                 group_contributions=group_contribs,
             )
         )
+
     return ExplainResponse(explanations=explanations)
 
 
 @app.post("/predict_explain", response_model=PredictExplainResponse)
 def predict_explain(req: ExplainRequest) -> PredictExplainResponse:
+    _require_loaded()
     if not req.records:
         raise HTTPException(status_code=400, detail="No records provided.")
+
     frame = prepare_frame(req.records)
     probas = PIPELINE.predict_proba(frame)[:, 1]
+
     booster = get_booster()
     X_proc = PREPROCESS.transform(frame)
+
     feat_names = np.array(PRE_FEATURE_NAMES) if PRE_FEATURE_NAMES else np.array([f"f{i}" for i in range(X_proc.shape[1])])
     sanitized_names = [_sanitize_feature_name(n) for n in feat_names]
+
     dmat = xgb.DMatrix(X_proc, feature_names=sanitized_names)
     contribs = booster.predict(dmat, pred_contribs=True)
+
     if contribs.shape[1] != X_proc.shape[1] + 1:
         raise HTTPException(status_code=500, detail="Unexpected contribution shape from booster.")
+
     base_vals = contribs[:, -1]
     feat_contribs = contribs[:, :-1]
+
     items: List[PredictExplainItem] = []
     for i in range(feat_contribs.shape[0]):
         row_vals = feat_contribs[i]
+
         group_totals: Dict[str, float] = {}
         group_details: Dict[str, List[FeatureContribution]] = {}
+
         for name, val in zip(feat_names, row_vals):
             base = _base_feature_name(str(name))
             group = FEATURE_GROUP_LOOKUP.get(base, "Other")
+
             group_totals[group] = group_totals.get(group, 0.0) + float(val)
             group_details.setdefault(group, []).append(
                 FeatureContribution(feature=str(name), shap_value=float(val))
             )
+
         group_contribs: List[GroupContribution] = []
         for grp, total in sorted(group_totals.items(), key=lambda kv: abs(kv[1]), reverse=True):
             feats = sorted(group_details.get(grp, []), key=lambda fc: abs(fc.shap_value), reverse=True)
             if req.top_k:
-                feats = feats[:req.top_k]
+                feats = feats[: req.top_k]
             group_contribs.append(GroupContribution(group=grp, total_shap_value=total, features=feats))
+
         score_val = int(round(float(pd_to_score(np.array([probas[i]]))[0])))
+
         items.append(
             PredictExplainItem(
                 probability=float(probas[i]),
@@ -357,4 +464,5 @@ def predict_explain(req: ExplainRequest) -> PredictExplainResponse:
                 group_contributions=group_contribs,
             )
         )
+
     return PredictExplainResponse(results=items)