Deploy MzansiScore API Space

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+ENV MZANSISCORE_MODEL_SOURCE=hub
+ENV MZANSISCORE_MODEL_REPO=Pieter182/mzansiscore-credit-risk
+
+COPY requirements.txt requirements.txt
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+EXPOSE 7860
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,54 @@
----
-title: Mzansiscore Api
-emoji: 👁
-colorFrom: purple
-colorTo: blue
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: MzansiScore Credit Risk API
+emoji: 📉
+colorFrom: blue
+colorTo: indigo
+sdk: docker
+app_port: 7860
+---
+
+# MzansiScore Credit Risk API
+
+FastAPI backend for scoring South African credit applications using the trained MzansiScore model artifacts.
+
+## Endpoints
+
+- `GET /v1/health`
+- `GET /v1/model/info`
+- `POST /v1/score`
+
+## Local run
+
+```powershell
+pip install -r api/requirements.txt
+uvicorn api.main:app --reload --port 7860
+```
+
+If you are inside the `api/` folder itself, run:
+
+```powershell
+pip install -r requirements.txt
+uvicorn main:app --reload --port 7860
+```
+
+## Model artifact source
+
+The API supports both local artifacts and HuggingFace Hub downloads.
+
+- Default HF model repo: `Pieter182/mzansiscore-credit-risk`
+- Optional env var: `MZANSISCORE_MODEL_REPO`
+- Optional env var: `MZANSISCORE_MODEL_SOURCE=hub` to force Hub downloads
+- Optional env var: `HF_TOKEN` for private repos or authenticated access
+
+The Docker image is configured to prefer Hub artifacts by default:
+
+- `MZANSISCORE_MODEL_SOURCE=hub`
+- `MZANSISCORE_MODEL_REPO=Pieter182/mzansiscore-credit-risk`
+
+## HuggingFace Docker Space
+
+This folder is now self-contained for deployment as its own HuggingFace Docker Space repo.
+
+- push the contents of `api/` as the Space repository root
+- `Dockerfile` now builds from that root directly
+- runtime feature engineering is bundled via `features_runtime.py`

features_runtime.py

@@ -0,0 +1,80 @@
+import numpy as np
+import pandas as pd
+
+
+def build_features(df):
+    df = df.copy()
+
+    df['living_to_income'] = df['living_expenses'] / df['net_income'].replace(0, np.nan)
+    df['discretionary_ratio'] = df['discretionary_income'] / df['net_income'].replace(0, np.nan)
+    df['gross_to_net_ratio'] = df['net_income'] / df['gross_income'].replace(0, np.nan)
+
+    df['score_norm'] = df['credit_score'] / 850
+    df['score_risk'] = 1 - df['score_norm']
+    df['combined_risk'] = df['score_risk'] + df['debt_service_ratio']
+
+    df['food_ratio'] = df['food_expense'] / df['net_income'].replace(0, np.nan)
+    df['housing_ratio'] = df['accommodation_expense'] / df['net_income'].replace(0, np.nan)
+    df['transport_ratio'] = df['transport_expense'] / df['net_income'].replace(0, np.nan)
+
+    if 'credit_bureau_monthly_debt' in df.columns:
+        df['bureau_debt_to_income'] = df['credit_bureau_monthly_debt'] / df['net_income'].replace(0, np.nan)
+
+    if 'affordability_surplus' in df.columns:
+        df['affordability_surplus_ratio'] = df['affordability_surplus'] / df['net_income'].replace(0, np.nan)
+
+    if 'affordability_expense_basis' in df.columns:
+        df['affordability_basis_ratio'] = df['affordability_expense_basis'] / df['net_income'].replace(0, np.nan)
+
+    if {'affordability_expense_basis', 'living_expenses'}.issubset(df.columns):
+        df['expense_gap_ratio'] = (
+            (df['affordability_expense_basis'] - df['living_expenses']) /
+            df['net_income'].replace(0, np.nan)
+        )
+
+    if {'minimum_living_expense', 'living_expenses'}.issubset(df.columns):
+        df['norm_to_declared_ratio'] = df['minimum_living_expense'] / df['living_expenses'].replace(0, np.nan)
+
+    if 'predicted_monthly_income' in df.columns:
+        df['income_prediction_gap'] = (
+            (df['predicted_monthly_income'] - df['gross_income']) /
+            df['gross_income'].replace(0, np.nan)
+        )
+
+    if 'household_gross_income' in df.columns:
+        df['household_to_income_ratio'] = df['household_gross_income'] / df['gross_income'].replace(0, np.nan)
+
+    if {'cpa_commitments', 'net_income'}.issubset(df.columns):
+        df['cpa_to_income_ratio'] = df['cpa_commitments'] / df['net_income'].replace(0, np.nan)
+
+    if {'nlr_commitments', 'net_income'}.issubset(df.columns):
+        df['nlr_to_income_ratio'] = df['nlr_commitments'] / df['net_income'].replace(0, np.nan)
+
+    if {'contactability_index', 'bureau_utilisation'}.issubset(df.columns):
+        df['contactability_risk_gap'] = (100 - df['contactability_index']) / 100 + df['bureau_utilisation']
+
+    if {'monthly_instalment', 'net_income'}.issubset(df.columns):
+        df['instalment_to_income'] = df['monthly_instalment'] / df['net_income'].replace(0, np.nan)
+
+    if {'total_debt_outstanding_zar', 'gross_income'}.issubset(df.columns):
+        df['debt_to_income_multiple'] = df['total_debt_outstanding_zar'] / df['gross_income'].replace(0, np.nan)
+
+    if {'asset_total_zar', 'gross_income'}.issubset(df.columns):
+        df['assets_to_income_multiple'] = df['asset_total_zar'] / df['gross_income'].replace(0, np.nan)
+
+    if {'avg_bank_balance_30d_zar', 'net_income'}.issubset(df.columns):
+        df['avg_balance_to_income'] = df['avg_bank_balance_30d_zar'] / df['net_income'].replace(0, np.nan)
+
+    if {'min_bank_balance_30d_zar', 'net_income'}.issubset(df.columns):
+        df['min_balance_to_income'] = df['min_bank_balance_30d_zar'] / df['net_income'].replace(0, np.nan)
+
+    if {'credit_history_months'}.issubset(df.columns):
+        df['credit_history_years'] = df['credit_history_months'] / 12
+
+    df['income_band'] = pd.cut(
+        df['gross_income'],
+        bins=[0, 8000, 15000, 25000, 45000, 80000, np.inf],
+        labels=['very_low', 'low', 'lower_mid', 'middle', 'upper_mid', 'high'],
+    )
+
+    return df

main.py

@@ -0,0 +1,80 @@
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, HTTPException
+from fastapi.middleware.cors import CORSMiddleware
+
+try:
+    from api.models import HealthResponse, ModelInfoResponse, ScoreRequest, ScoreResponse
+    from api.scorer import ModelScorer
+except ImportError:
+    from models import HealthResponse, ModelInfoResponse, ScoreRequest, ScoreResponse
+    from scorer import ModelScorer
+
+scorer: ModelScorer | None = None
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    global scorer
+    scorer = ModelScorer()
+    yield
+
+
+app = FastAPI(
+    title='MzansiScore Credit Risk API',
+    version='0.1.0',
+    description='FastAPI scorer for the MzansiScore SA credit risk model.',
+    lifespan=lifespan,
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=['*'],
+    allow_credentials=True,
+    allow_methods=['*'],
+    allow_headers=['*'],
+)
+
+
+@app.get('/')
+def root() -> dict[str, str | bool | None]:
+    return {
+        'service': 'MzansiScore Credit Risk API',
+        'status': 'ok',
+        'model_loaded': scorer is not None,
+        'best_model': None if scorer is None else scorer.meta.get('best_model'),
+        'artifact_source': None if scorer is None else scorer.artifact_source,
+        'model_repo': None if scorer is None else scorer.repo_id,
+        'docs': '/docs',
+        'health': '/v1/health',
+    }
+
+
+@app.get('/v1/health', response_model=HealthResponse)
+def health() -> HealthResponse:
+    return HealthResponse(
+        status='ok',
+        model_loaded=scorer is not None,
+        best_model=None if scorer is None else scorer.meta.get('best_model'),
+        artifact_source=None if scorer is None else scorer.artifact_source,
+        model_repo=None if scorer is None else scorer.repo_id,
+    )
+
+
+@app.get('/v1/model/info', response_model=ModelInfoResponse)
+def model_info() -> ModelInfoResponse:
+    if scorer is None:
+        raise HTTPException(status_code=503, detail='Model not loaded')
+    return ModelInfoResponse(
+        **scorer.model_info(),
+        artifact_source=scorer.artifact_source,
+        model_repo=scorer.repo_id,
+    )
+
+
+@app.post('/v1/score', response_model=ScoreResponse)
+def score_application(request: ScoreRequest) -> ScoreResponse:
+    if scorer is None:
+        raise HTTPException(status_code=503, detail='Model not loaded')
+    result = scorer.score(request.to_payload())
+    return ScoreResponse(**result)

models.py

@@ -0,0 +1,68 @@
+from typing import Any
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class ScoreRequest(BaseModel):
+    model_config = ConfigDict(extra='allow')
+
+    gross_income: float = Field(..., gt=0)
+    net_income: float = Field(..., gt=0)
+    living_expenses: float = Field(..., ge=0)
+    credit_score: int = Field(..., ge=250, le=850)
+    age: int = Field(..., ge=18, le=100)
+
+    discretionary_income: float | None = None
+    debt_service_ratio: float | None = Field(default=None, ge=0)
+    credit_bureau_monthly_debt: float | None = Field(default=None, ge=0)
+    monthly_instalment: float | None = Field(default=None, ge=0)
+    loan_amount_requested: float | None = Field(default=None, ge=0)
+    loan_term_months: int | None = Field(default=None, ge=1)
+
+    province: str | None = None
+    gender: str | None = None
+    marital_status: str | None = None
+    employment_type: str | None = None
+    education_level: str | None = None
+    dependants: int | None = Field(default=None, ge=0)
+    residence_type: str | None = None
+    loan_product_type: str | None = None
+    salary_frequency: str | None = None
+
+    affordability_expense_basis: float | None = None
+    affordability_surplus: float | None = None
+    affordability_margin: float | None = None
+    nca_affordability_pass: int | None = Field(default=None, ge=0, le=1)
+
+    def to_payload(self) -> dict[str, Any]:
+        return self.model_dump(exclude_none=True)
+
+
+class ScoreResponse(BaseModel):
+    decision_id: str
+    risk_score: int
+    risk_band: str
+    recommendation: str
+    default_probability: float
+    nca_pass: int
+    policy_reason: str
+    shap_top3: list[dict[str, float | str]]
+    processing_time_ms: float
+
+
+class HealthResponse(BaseModel):
+    status: str
+    model_loaded: bool
+    best_model: str | None = None
+    artifact_source: str | None = None
+    model_repo: str | None = None
+
+
+class ModelInfoResponse(BaseModel):
+    best_model: str
+    holdout_auc: float
+    holdout_brier: float
+    runtime_human: str | None = None
+    top_features: list[dict[str, float | str]]
+    artifact_source: str | None = None
+    model_repo: str | None = None

requirements.txt

@@ -0,0 +1,13 @@
+fastapi>=0.115
+uvicorn[standard]>=0.30
+pydantic>=2.8
+pandas>=2.2
+numpy>=1.26
+joblib>=1.4
+scikit-learn>=1.5
+xgboost>=2.0
+lightgbm>=4.3
+shap>=0.46
+huggingface_hub>=0.30
+pytest>=8.3
+httpx>=0.27

scorer.py

@@ -0,0 +1,287 @@
+import json
+import os
+import sys
+import time
+import uuid
+from pathlib import Path
+from typing import Any
+
+import joblib
+import numpy as np
+import pandas as pd
+import shap
+from huggingface_hub import hf_hub_download
+
+ROOT_DIR = Path(__file__).resolve().parents[1]
+ML_SRC_DIR = ROOT_DIR / 'ml' / 'src'
+ML_DATA_DIR = ROOT_DIR / 'ml' / 'data'
+HF_MODEL_REPO = os.getenv('MZANSISCORE_MODEL_REPO', 'Pieter182/mzansiscore-credit-risk')
+HF_TOKEN = os.getenv('HF_TOKEN') or os.getenv('HUGGINGFACE_HUB_TOKEN')
+ARTIFACT_FILES = [
+    'best_tree_model.joblib',
+    'feature_cols.joblib',
+    'label_encoders.joblib',
+    'model_meta.json',
+]
+STATIC_DEFAULTS: dict[str, Any] = {
+    'province': 'Gauteng',
+    'gender': 'Male',
+    'marital_status': 'Single',
+    'employment_type': 'permanent',
+    'education_level': 'matric',
+    'residence_type': 'rented_formal',
+    'loan_product_type': 'personal_loan',
+    'salary_frequency': 'Monthly',
+    'dependants': 0,
+    'gross_income': 25000.0,
+    'net_income': 19000.0,
+    'living_expenses': 7000.0,
+    'discretionary_income': 6500.0,
+    'debt_service_ratio': 0.28,
+    'credit_bureau_monthly_debt': 2500.0,
+    'monthly_instalment': 2800.0,
+    'loan_amount_requested': 50000.0,
+    'loan_term_months': 24,
+    'age': 34,
+    'credit_score': 640,
+    'food_expense': 1800.0,
+    'transport_expense': 1200.0,
+    'accommodation_expense': 2800.0,
+    'utilities_expense': 900.0,
+    'medical_expense': 500.0,
+    'minimum_living_expense': 7200.0,
+    'affordability_expense_basis': 7200.0,
+    'affordability_surplus': 6500.0,
+    'affordability_margin': 0.34,
+    'predicted_monthly_income': 25500.0,
+    'household_gross_income': 32000.0,
+    'cpa_commitments': 700.0,
+    'nlr_commitments': 600.0,
+    'contactability_index': 82.0,
+    'bureau_utilisation': 0.33,
+    'total_debt_outstanding_zar': 42000.0,
+    'asset_total_zar': 110000.0,
+    'avg_bank_balance_30d_zar': 8400.0,
+    'min_bank_balance_30d_zar': 2200.0,
+    'credit_history_months': 48,
+    'nca_affordability_pass': 1,
+}
+
+if str(ML_SRC_DIR) not in sys.path:
+    sys.path.append(str(ML_SRC_DIR))
+
+try:
+    from api.features_runtime import build_features  # type: ignore
+except ImportError:
+    try:
+        from features_runtime import build_features  # type: ignore
+    except ImportError:
+        from features import build_features  # type: ignore  # noqa: E402
+
+
+class ModelScorer:
+    def __init__(self, repo_id: str | None = None, token: str | None = None) -> None:
+        self.repo_id = repo_id or HF_MODEL_REPO
+        self.token = token or HF_TOKEN
+        self.artifact_source = 'unknown'
+        self.artifact_dir = self._resolve_artifact_dir()
+
+        self.best_model = joblib.load(self.artifact_dir / 'best_tree_model.joblib')
+        self.feature_cols = joblib.load(self.artifact_dir / 'feature_cols.joblib')
+        self.label_encoders = joblib.load(self.artifact_dir / 'label_encoders.joblib')
+        with open(self.artifact_dir / 'model_meta.json', 'r', encoding='utf-8') as handle:
+            self.meta = json.load(handle)
+
+        self.defaults = self._build_defaults()
+        self.explainer = shap.TreeExplainer(self.best_model)
+
+    def _resolve_artifact_dir(self) -> Path:
+        local_ready = all((ML_DATA_DIR / name).exists() for name in ARTIFACT_FILES)
+        prefer_hub = os.getenv('MZANSISCORE_MODEL_SOURCE', '').lower() == 'hub' or not local_ready
+
+        if prefer_hub and self.repo_id:
+            try:
+                self.artifact_source = 'hub'
+                return self._download_artifacts_from_hub()
+            except Exception:
+                if not local_ready:
+                    raise
+
+        if local_ready:
+            self.artifact_source = 'local'
+            return ML_DATA_DIR
+
+        if self.repo_id:
+            self.artifact_source = 'hub'
+            return self._download_artifacts_from_hub()
+
+        raise FileNotFoundError('No local model artifacts found and no HuggingFace repo configured.')
+
+    def _download_artifacts_from_hub(self) -> Path:
+        downloaded_paths = [
+            Path(
+                hf_hub_download(
+                    repo_id=self.repo_id,
+                    filename=filename,
+                    repo_type='model',
+                    token=self.token,
+                ),
+            )
+            for filename in ARTIFACT_FILES
+        ]
+        return downloaded_paths[0].parent
+
+    def _build_defaults(self) -> dict[str, Any]:
+        dataset_path = ML_DATA_DIR / 'sa_credit_dataset.csv'
+        if not dataset_path.exists():
+            return dict(STATIC_DEFAULTS)
+
+        df = pd.read_csv(dataset_path)
+        defaults: dict[str, Any] = dict(STATIC_DEFAULTS)
+
+        for col in df.columns:
+            series = df[col]
+            if pd.api.types.is_numeric_dtype(series):
+                defaults[col] = float(series.median()) if not pd.isna(series.median()) else 0.0
+            else:
+                mode = series.mode(dropna=True)
+                defaults[col] = str(mode.iloc[0]) if not mode.empty else ''
+
+        defaults.setdefault('province', 'Gauteng')
+        defaults.setdefault('gender', 'Male')
+        defaults.setdefault('marital_status', 'Single')
+        defaults.setdefault('employment_type', 'permanent')
+        defaults.setdefault('education_level', 'matric')
+        defaults.setdefault('residence_type', 'rented_formal')
+        defaults.setdefault('loan_product_type', 'personal_loan')
+        defaults.setdefault('salary_frequency', 'Monthly')
+        defaults.setdefault('dependants', 0)
+        return defaults
+
+    def _merge_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
+        row = dict(self.defaults)
+        row.update(payload)
+
+        if row.get('discretionary_income') is None:
+            row['discretionary_income'] = (
+                float(row.get('net_income', 0))
+                - float(row.get('living_expenses', 0))
+                - float(row.get('credit_bureau_monthly_debt', 0))
+                - float(row.get('monthly_instalment', 0))
+            )
+
+        if row.get('affordability_expense_basis') is None:
+            row['affordability_expense_basis'] = max(
+                float(row.get('living_expenses', 0)),
+                float(row.get('minimum_living_expense', row.get('living_expenses', 0))),
+            )
+
+        if row.get('affordability_surplus') is None:
+            row['affordability_surplus'] = (
+                float(row.get('net_income', 0))
+                - float(row.get('affordability_expense_basis', 0))
+                - float(row.get('credit_bureau_monthly_debt', 0))
+                - float(row.get('monthly_instalment', 0))
+            )
+
+        if row.get('affordability_margin') is None:
+            row['affordability_margin'] = float(row['affordability_surplus']) / max(float(row.get('net_income', 1)), 1.0)
+
+        if row.get('debt_service_ratio') is None:
+            row['debt_service_ratio'] = (
+                float(row.get('credit_bureau_monthly_debt', 0)) + float(row.get('monthly_instalment', 0))
+            ) / max(float(row.get('net_income', 1)), 1.0)
+
+        if row.get('nca_affordability_pass') is None:
+            row['nca_affordability_pass'] = int(
+                float(row.get('affordability_surplus', 0)) > 800
+                and float(row.get('affordability_margin', 0)) > 0.02
+            )
+
+        return row
+
+    def _encode_for_tree(self, df: pd.DataFrame) -> pd.DataFrame:
+        model_df = df.copy()
+        for col in self.feature_cols:
+            if col not in model_df.columns:
+                model_df[col] = np.nan
+
+        model_df = model_df[self.feature_cols].copy()
+
+        for col, encoder in self.label_encoders.items():
+            if col not in model_df.columns:
+                continue
+            values = model_df[col].astype(str).fillna('MISSING')
+            known = set(encoder.classes_)
+            values = values.where(values.isin(known), 'UNKNOWN')
+            model_df[col] = encoder.transform(values)
+
+        return model_df.fillna(-999)
+
+    def _build_shap_top3(self, encoded_df: pd.DataFrame) -> list[dict[str, Any]]:
+        shap_values = self.explainer.shap_values(encoded_df)
+        if isinstance(shap_values, list):
+            shap_values = shap_values[0]
+        row_vals = np.asarray(shap_values)[0]
+        top_idx = np.argsort(np.abs(row_vals))[-3:][::-1]
+        return [
+            {
+                'feature': self.feature_cols[int(idx)],
+                'shap_value': round(float(row_vals[int(idx)]), 4),
+            }
+            for idx in top_idx
+        ]
+
+    def score(self, payload: dict[str, Any]) -> dict[str, Any]:
+        start = time.perf_counter()
+        merged = self._merge_payload(payload)
+        raw_df = pd.DataFrame([merged])
+        feature_df = build_features(raw_df)
+        encoded_df = self._encode_for_tree(feature_df)
+
+        default_probability = float(self.best_model.predict_proba(encoded_df)[0, 1])
+        nca_pass = int(merged['nca_affordability_pass'])
+
+        if not nca_pass:
+            risk_band = 'DECLINE'
+            recommendation = 'DECLINE'
+            policy_reason = 'Affordability rule failed'
+        elif default_probability < 0.10:
+            risk_band = 'LOW'
+            recommendation = 'APPROVE'
+            policy_reason = 'Affordability pass'
+        elif default_probability < 0.25:
+            risk_band = 'MEDIUM'
+            recommendation = 'REVIEW'
+            policy_reason = 'Affordability pass'
+        elif default_probability < 0.50:
+            risk_band = 'HIGH'
+            recommendation = 'REVIEW'
+            policy_reason = 'Affordability pass'
+        else:
+            risk_band = 'DECLINE'
+            recommendation = 'DECLINE'
+            policy_reason = 'High predicted default probability'
+
+        elapsed_ms = round((time.perf_counter() - start) * 1000, 2)
+        return {
+            'decision_id': f"MZS-{uuid.uuid4().hex[:8].upper()}",
+            'risk_score': int(round((1 - default_probability) * 999)),
+            'risk_band': risk_band,
+            'recommendation': recommendation,
+            'default_probability': round(default_probability, 4),
+            'nca_pass': nca_pass,
+            'policy_reason': policy_reason,
+            'shap_top3': self._build_shap_top3(encoded_df),
+            'processing_time_ms': elapsed_ms,
+        }
+
+    def model_info(self) -> dict[str, Any]:
+        calibrated = self.meta['results'].get('XGBoost_cal', self.meta['results']['XGBoost'])
+        return {
+            'best_model': self.meta['best_model'],
+            'holdout_auc': calibrated['holdout_auc'],
+            'holdout_brier': calibrated['holdout_brier'],
+            'runtime_human': self.meta.get('runtime_human'),
+            'top_features': self.meta.get('top_features', [])[:5],
+        }

tests/test_main.py

@@ -0,0 +1,57 @@
+from fastapi.testclient import TestClient
+
+from api.main import app
+
+
+SAMPLE_PAYLOAD = {
+    'gross_income': 25000,
+    'net_income': 19000,
+    'living_expenses': 7000,
+    'credit_score': 640,
+    'age': 34,
+    'monthly_instalment': 2800,
+    'credit_bureau_monthly_debt': 2500,
+    'loan_amount_requested': 50000,
+    'loan_term_months': 24,
+}
+
+
+def test_health_endpoint():
+    with TestClient(app) as client:
+        response = client.get('/v1/health')
+        assert response.status_code == 200
+        data = response.json()
+        assert data['status'] == 'ok'
+        assert data['model_loaded'] is True
+        assert data['artifact_source'] in {'local', 'hub'}
+
+
+def test_root_endpoint():
+    with TestClient(app) as client:
+        response = client.get('/')
+        assert response.status_code == 200
+        data = response.json()
+        assert data['service'] == 'MzansiScore Credit Risk API'
+        assert data['health'] == '/v1/health'
+
+
+def test_model_info_endpoint():
+    with TestClient(app) as client:
+        response = client.get('/v1/model/info')
+        assert response.status_code == 200
+        data = response.json()
+        assert 'best_model' in data
+        assert 'holdout_auc' in data
+        assert 'top_features' in data
+        assert data['artifact_source'] in {'local', 'hub'}
+
+
+def test_score_endpoint():
+    with TestClient(app) as client:
+        response = client.post('/v1/score', json=SAMPLE_PAYLOAD)
+        assert response.status_code == 200
+        data = response.json()
+        assert data['decision_id'].startswith('MZS-')
+        assert data['risk_band'] in {'LOW', 'MEDIUM', 'HIGH', 'DECLINE'}
+        assert data['recommendation'] in {'APPROVE', 'REVIEW', 'DECLINE'}
+        assert len(data['shap_top3']) == 3

tests/test_scorer.py

@@ -0,0 +1,44 @@
+from api.scorer import ModelScorer
+
+
+SAMPLE_PAYLOAD = {
+    'gross_income': 25000,
+    'net_income': 19000,
+    'living_expenses': 7000,
+    'credit_score': 640,
+    'age': 34,
+    'monthly_instalment': 2800,
+    'credit_bureau_monthly_debt': 2500,
+    'loan_amount_requested': 50000,
+    'loan_term_months': 24,
+    'province': 'Gauteng',
+    'employment_type': 'permanent',
+    'education_level': 'diploma',
+    'dependants': 1,
+    'residence_type': 'rented_formal',
+    'loan_product_type': 'personal_loan',
+    'salary_frequency': 'Monthly',
+}
+
+
+def test_model_scorer_returns_expected_payload_shape():
+    scorer = ModelScorer()
+    result = scorer.score(SAMPLE_PAYLOAD)
+
+    assert result['decision_id'].startswith('MZS-')
+    assert 0 <= result['default_probability'] <= 1
+    assert 0 <= result['risk_score'] <= 999
+    assert result['risk_band'] in {'LOW', 'MEDIUM', 'HIGH', 'DECLINE'}
+    assert result['recommendation'] in {'APPROVE', 'REVIEW', 'DECLINE'}
+    assert result['nca_pass'] in {0, 1}
+    assert len(result['shap_top3']) == 3
+
+
+def test_model_info_contains_core_metrics():
+    scorer = ModelScorer()
+    info = scorer.model_info()
+
+    assert info['best_model']
+    assert info['holdout_auc'] > 0.5
+    assert info['holdout_brier'] >= 0
+    assert isinstance(info['top_features'], list)