app/model.py

from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

import joblib
import numpy as np
import shap


@dataclass(frozen=True)
class ScoreResult:
    probability: float
    decision: str
    top_factors: List[Dict[str, Any]]


class FraudModel:
    def __init__(self, artifacts_path: Path):
        payload = joblib.load(artifacts_path)
        self.model = payload["model"]
        self.imputer = payload["imputer"]
        self.feature_names = list(payload["feature_names"])
        self.freq_mappings = payload["freq_mappings"]
        self.categorical_cols = list(payload.get("categorical_cols") or [])

        self.explainer = shap.TreeExplainer(self.model)

    def _decision(self, p: float) -> str:
        return "ALLOW" if p < 0.2 else "REVIEW" if p < 0.6 else "BLOCK"

    def _freq_encode_one(self, row: Dict[str, Any]) -> np.ndarray:
        vals: List[float] = []
        for col in self.feature_names:
            v = row.get(col, None)
            if col in self.categorical_cols:
                mapping = self.freq_mappings.get(col, {})
                key = "NA" if v is None else str(v)
                vals.append(float(mapping.get(key, 0)))
            else:
                vals.append(np.nan if v is None else float(v))

        X = np.array(vals, dtype=float).reshape(1, -1)
        X = self.imputer.transform(X)
        return X

    def score(self, row: Dict[str, Any], top_k: int = 8) -> ScoreResult:
        X = self._freq_encode_one(row)
        p = float(self.model.predict_proba(X)[:, 1][0])

        shap_vals = self.explainer.shap_values(X)
        if isinstance(shap_vals, list):
            shap_vals = shap_vals[0]
        shap_vals = np.array(shap_vals).reshape(-1)

        order = np.argsort(np.abs(shap_vals))[::-1][:top_k]
        top = []
        for idx in order:
            top.append(
                {
                    "feature": self.feature_names[int(idx)],
                    "contribution": float(shap_vals[int(idx)]),
                    "abs_contribution": float(abs(shap_vals[int(idx)])),
                }
            )

        return ScoreResult(probability=p, decision=self._decision(p), top_factors=top)