Spaces:
Running
Running
File size: 5,361 Bytes
44cd54e 42029e4 44cd54e 42029e4 44cd54e 42029e4 | 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 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Scoring service - wraps the trained model + builder + explainer."""
from __future__ import annotations
import time
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
import numpy as np
import pandas as pd
from ..utils.io import load_joblib
from ..utils.logging import get_logger
from .schemas import LoanApplication, ReasonCode, ScoreResponse
log = get_logger(__name__)
class ScoringService:
"""Singleton-style scoring service used by the API endpoints."""
def __init__(
self,
artifacts_dir: str | Path = "artifacts",
review_threshold: float = 0.30,
decline_threshold: float = 0.70,
model_version: str = "0.1.0",
):
self.artifacts_dir = Path(artifacts_dir)
self.review_threshold = review_threshold
self.decline_threshold = decline_threshold
self.model_version = model_version
self.builder: Any = None
self.model: Any = None
self.explainer: Any = None
self.loaded_at: float | None = None
# ------------------------------------------------------------------ #
def load(self) -> None:
if not self.artifacts_dir.exists():
log.warning(
f"Artifacts dir {self.artifacts_dir} not found. Service starts empty - "
"call /reload after training."
)
return
builder_path = self.artifacts_dir / "feature_builder.joblib"
if builder_path.exists():
self.builder = load_joblib(builder_path)
ensemble_path = self.artifacts_dir / "model_ensemble.joblib"
xgb_path = self.artifacts_dir / "model_xgboost.joblib"
if ensemble_path.exists():
self.model = load_joblib(ensemble_path)
elif xgb_path.exists():
self.model = load_joblib(xgb_path)
# Build explainer lazily once we have something to explain
if self.model is not None and hasattr(self.model, "base_models"):
# Use XGB / LGBM from the ensemble for SHAP
for m in self.model.base_models: # type: ignore
if m.name in ("xgboost", "lightgbm"):
try:
from ..evaluation.explainability import ShapExplainer
self.explainer = ShapExplainer(m).fit()
log.info(f"SHAP explainer built from base model: {m.name}")
break
except Exception as exc: # pragma: no cover
log.warning(f"Could not init SHAP explainer: {exc}")
elif self.model is not None:
try:
from ..evaluation.explainability import ShapExplainer
self.explainer = ShapExplainer(self.model).fit()
except Exception as exc: # pragma: no cover
log.warning(f"Could not init SHAP explainer: {exc}")
self.loaded_at = time.time()
log.info(f"Scoring service ready. version={self.model_version}")
@property
def is_ready(self) -> bool:
return self.builder is not None and self.model is not None
# ------------------------------------------------------------------ #
def score_one(self, app: LoanApplication) -> ScoreResponse:
if not self.is_ready:
raise RuntimeError("ScoringService is not loaded")
df = pd.DataFrame([app.model_dump()])
# Coerce dates
for c in ("issue_d", "earliest_cr_line"):
if c in df.columns:
df[c] = pd.to_datetime(df[c], errors="coerce")
if "issue_d" not in df.columns or df["issue_d"].isna().all():
# tz-naive "now" - issue_d only drives seasonality/velocity features
df["issue_d"] = pd.Timestamp.now()
X = self.builder.transform(df)
proba = float(self.model.predict_proba(X)[0])
if proba >= self.decline_threshold:
decision = "DECLINE"
elif proba >= self.review_threshold:
decision = "REVIEW"
else:
decision = "APPROVE"
# Reason codes
reasons: list[ReasonCode] = []
if self.explainer is not None:
try:
for exp in self.explainer.explain_one(X.iloc[[0]], top_k=5):
reasons.append(
ReasonCode(
feature=exp.feature,
value=float(exp.value),
contribution=float(exp.contribution),
direction=exp.direction,
)
)
except Exception as exc: # pragma: no cover
log.warning(f"Could not produce reason codes: {exc}")
return ScoreResponse(
application_id=app.id,
fraud_score=round(proba, 6),
decision=decision, # type: ignore[arg-type]
threshold_review=self.review_threshold,
threshold_decline=self.decline_threshold,
reason_codes=reasons,
model_version=self.model_version,
scored_at=datetime.now(timezone.utc),
)
def score_many(self, apps: list[LoanApplication]) -> list[ScoreResponse]:
if not self.is_ready:
raise RuntimeError("ScoringService is not loaded")
return [self.score_one(a) for a in apps]
|