new types of explainability added

api/app.py

@@ -1,99 +1,84 @@
 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
-import xgboost as xgb
 from fastapi import FastAPI, HTTPException
-from huggingface_hub import hf_hub_download
 from pydantic import BaseModel, Field
+import xgboost as xgb
+
+import os
+from huggingface_hub import hf_hub_download
 
 # 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")
-
 
-def download_pipeline_artifact() -> Path:
-    """
-    Download the .pkl artifact from Hugging Face Hub (cached locally).
-    """
-    try:
-        p = hf_hub_download(
+try:
+    MODEL_PATH = Path(
+        hf_hub_download(
             repo_id=HF_MODEL_REPO,
-            filename=HF_MODEL_FILENAME,
+            filename="xgboost_pipeline.pkl",
             revision=HF_MODEL_REVISION,
         )
-        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
-
-
-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())
-
+    )
+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.")
 
-# -----------------------------
-# Feature groups (unchanged)
-# -----------------------------
+_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: [])())
 RAW_FEATURE_SET = set((META.get("raw_num_cols") or []) + (META.get("raw_cat_cols") or []))
-
 FEATURE_GROUPS = {
-    "Borrowing History & Maturity": [
-        "account_age_days",
-        "avg_past_amount",
-        "avg_past_daily_burden",
-        "avg_time_bw_loans",
-        "borrower_history_strength",
-        "days_since_last_loan",
-        "loan_frequency_per_year",
-        "num_previous_loans",
-        "std_past_amount",
-        "std_past_daily_burden",
-        "trend_in_amount",
-        "trend_in_burden",
-    ],
-    "Repayment Speed & Delinquency": [
+    "Repayment Activity": [
         "num_previous_defaults",
         "past_default_rate",
         "repayment_consistency",
+        "repayment_intensity",
     ],
-    "Current Loan Size, Pricing & Burden": [
+    "Loan Amount & Burden": [
         "Total_Amount",
         "Total_Amount_to_Repay",
         "amount_bucket",
@@ -102,84 +87,43 @@ FEATURE_GROUPS = {
         "duration",
         "duration_bucket",
         "interest_rate",
-    ],
-    "Affordability & Risk Ratios": [
         "amount_ratio",
         "burden_ratio",
-        "repayment_intensity",
+        "lender_exposure_ratio",
+    ],
+    "Borrowing History": [
+        "account_age_days",
+        "avg_past_amount",
+        "avg_past_daily_burden",
+        "avg_time_bw_loans",
+        "borrower_history_strength",
+        "days_since_last_loan",
+        "loan_frequency_per_year",
+        "num_previous_loans",
+        "std_past_amount",
+        "std_past_daily_burden",
+        "trend_in_amount",
+        "trend_in_burden",
+        "lender_id",
+        "lender_risk_profile",
     ],
-    "Seasonality & Timing": [
+    "Spending & Transactions": [
+        "latest_amount_ma3",
         "days_to_local_festival",
         "days_to_salary_day",
         "month",
         "quarter",
         "week_of_year",
     ],
-    "Operational, Referral & Lender Signals": [
-        "lender_exposure_ratio",
-        "lender_id",
-        "lender_risk_profile",
-    ],
-    "Time-based Trends & Volatility": [
-        "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")
 
@@ -209,6 +153,9 @@ class FeatureContribution(BaseModel):
 class GroupContribution(BaseModel):
     group: str
     total_shap_value: float
+    percentage: float
+    direction: str
+    label: str
     features: List[FeatureContribution]
 
 
@@ -233,32 +180,18 @@ 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
 
 
@@ -293,11 +226,57 @@ def _base_feature_name(name: str) -> str:
     return base
 
 
-def get_booster() -> xgb.Booster:
+def _label_for_percentage(pct: float) -> str:
+    if pct >= 30:
+        return "Exceptional"
+    if pct >= 20:
+        return "Very Good"
+    if pct >= 10:
+        return "Good"
+    if pct >= 5:
+        return "Bad"
+    return "Very Bad"
+
+
+def _direction_for_value(val: float) -> str:
+    if val > 0:
+        return "raises risk"
+    if val < 0:
+        return "reduces risk"
+    return "neutral"
+
+
+def _build_group_contribs(
+    group_totals: Dict[str, float],
+    group_details: Dict[str, List[FeatureContribution]],
+    top_k: Optional[int],
+) -> List[GroupContribution]:
+    denom = sum(abs(v) for v in group_totals.values())
+    if denom == 0:
+        denom = 1e-12  # avoid division by zero; all percentages become ~0
+    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 top_k:
+            feats = feats[:top_k]
+        pct = abs(total) / denom * 100
+        group_contribs.append(
+            GroupContribution(
+                group=grp,
+                total_shap_value=total,
+                percentage=pct,
+                direction=_direction_for_value(total),
+                label=_label_for_percentage(pct),
+                features=feats,
+            )
+        )
+    return group_contribs
+
+
+def get_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:
@@ -310,36 +289,22 @@ def get_booster() -> xgb.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:
@@ -351,51 +316,33 @@ 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]
-            group_contribs.append(GroupContribution(group=grp, total_shap_value=total, features=feats))
-
+        group_contribs = _build_group_contribs(group_totals, group_details, req.top_k)
         explanations.append(
             ExplainItem(
                 probability=float(probas[i]),
@@ -403,59 +350,39 @@ 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]
-            group_contribs.append(GroupContribution(group=grp, total_shap_value=total, features=feats))
-
+        group_contribs = _build_group_contribs(group_totals, group_details, req.top_k)
         score_val = int(round(float(pd_to_score(np.array([probas[i]]))[0])))
-
         items.append(
             PredictExplainItem(
                 probability=float(probas[i]),
@@ -464,7 +391,4 @@ def predict_explain(req: ExplainRequest) -> PredictExplainResponse:
                 group_contributions=group_contribs,
             )
         )
-
     return PredictExplainResponse(results=items)
-
-# DEPLOY_MARKER: 2026-01-24_1