app/scoring.py

"""Fraud scoring service: model + SHAP explainability (EXPL-01..04).

`ModelBundle` is loaded exactly once (at FastAPI startup, via `load_model_bundle`)
and reused for every request -- never re-loaded per-request (EXPL-04). It wraps:

- the trained `imblearn.pipeline.Pipeline` saved by `training/train.py`
  (its resampling step is a no-op at predict time -- imblearn pipelines
  only resample during `.fit()`)
- a `shap.TreeExplainer` built directly against the pipeline's underlying
  XGBoost estimator
- the training-time metadata (`model_v1.meta.json`): feature column order,
  model/imbalance-strategy identifiers, and the threshold chosen at
  MODEL-05/06

`score_transaction` is the single entry point used by every route that
scores a transaction (`/predict`, `/transactions/batch`, `/test/transactions`)
so real and simulated traffic always go through identical logic (EXPL-03) --
there is no hard-coded bypass for any transaction `type`.
"""

from __future__ import annotations

import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Sequence

import joblib
import numpy as np
import pandas as pd
import shap
from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from app.features import FEATURE_COLUMNS, engineer_features

TOP_N_FEATURES = 5

RAW_PAYSIM_COLUMNS = [
    "step",
    "type",
    "amount",
    "nameOrig",
    "oldbalanceOrg",
    "newbalanceOrig",
    "nameDest",
    "oldbalanceDest",
    "newbalanceDest",
]


@dataclass(frozen=True)
class ModelBundle:
    pipeline: Any
    explainer: shap.TreeExplainer
    feature_columns: list[str]
    model_version: str
    model_name: str


def load_model_bundle(model_path: str) -> ModelBundle:
    """Load the joblib artifact + its metadata sidecar exactly once.

    Raises `FileNotFoundError`/`ValueError` with a clear message if the
    artifact or its `.meta.json` sidecar is missing -- fail loud at startup
    rather than serve requests against a model that never loaded.
    """
    artifact_path = Path(model_path)
    if not artifact_path.exists():
        raise FileNotFoundError(
            f"MODEL_PATH '{model_path}' does not exist. Run "
            "`python -m training.train finalize ...` to produce it."
        )
    meta_path = artifact_path.with_suffix("").with_suffix(".meta.json")
    if not meta_path.exists():
        raise FileNotFoundError(
            f"Model metadata sidecar '{meta_path}' not found next to "
            f"'{model_path}' -- both are required (training-time versions "
            "and feature column order must be known before serving)."
        )

    pipeline = joblib.load(artifact_path)
    meta = json.loads(meta_path.read_text())
    feature_columns = meta["feature_columns"]
    clf = pipeline.named_steps["clf"]
    explainer = shap.TreeExplainer(clf)

    return ModelBundle(
        pipeline=pipeline,
        explainer=explainer,
        feature_columns=feature_columns,
        model_version=meta_path.stem.replace(".meta", ""),
        model_name=meta["model_name"],
    )


def risk_tier(probability: float) -> str:
    """Fixed low/medium/high display bands, independent of the alert
    threshold: MODEL-04's candidate thresholds (0.3/0.5/0.7) form a natural
    three-way split so this exactly matches metrics already reviewed with
    the user. Whether an alert fires is a *separate* decision governed by
    `FRAUD_ALERT_THRESHOLD` -- risk tier is purely informational.
    """
    if probability >= 0.7:
        return "high"
    if probability >= 0.3:
        return "medium"
    return "low"


def _history_rows_to_frame(rows: Sequence[Any]) -> pd.DataFrame:
    return pd.DataFrame(
        [
            {
                "step": r.step,
                "type": r.type,
                "amount": float(r.amount),
                "nameOrig": r.name_orig,
                "oldbalanceOrg": float(r.oldbalance_org),
                "newbalanceOrig": float(r.newbalance_orig),
                "nameDest": r.name_dest,
                "oldbalanceDest": float(r.oldbalance_dest),
                "newbalanceDest": float(r.newbalance_dest),
            }
            for r in rows
        ],
        columns=RAW_PAYSIM_COLUMNS,
    )


def fetch_account_history(
    db: Session, history_model: type, name_orig: str, name_dest: str
) -> pd.DataFrame:
    """Prior transactions touching either account, oldest first.

    `app.features.engineer_features`'s velocity features are prior-only
    aggregates, so this history (everything *before* the new transaction)
    is exactly what's needed to compute them correctly for a single
    real-time transaction -- the same shared feature module used by
    training, never a re-implementation of its formulas (FEAT-01).
    """
    stmt = (
        select(history_model)
        .where(
            or_(
                history_model.name_orig == name_orig,
                history_model.name_dest == name_dest,
            )
        )
        .order_by(history_model.step)
    )
    rows = db.execute(stmt).scalars().all()
    return _history_rows_to_frame(rows)


_NUMERIC_COLUMNS = [
    "step",
    "amount",
    "oldbalanceOrg",
    "newbalanceOrig",
    "oldbalanceDest",
    "newbalanceDest",
]


def build_feature_row(history: pd.DataFrame, raw_transaction: dict) -> pd.DataFrame:
    """Append the new transaction to its account history and engineer
    features, returning only the new transaction's feature row.
    """
    new_row = pd.DataFrame([raw_transaction], columns=RAW_PAYSIM_COLUMNS)
    combined = pd.concat([history, new_row], ignore_index=True)
    # An empty (no-history) frame defaults numeric columns to object dtype,
    # which breaks the groupby cumsum/cumcount inside engineer_features
    # once concatenated -- coerce explicitly rather than rely on pandas to
    # infer dtypes across an empty-plus-one-row concat.
    combined[_NUMERIC_COLUMNS] = combined[_NUMERIC_COLUMNS].apply(
        pd.to_numeric, errors="raise"
    )
    engineered = engineer_features(combined)
    return engineered.iloc[[-1]]


def score_transaction(
    bundle: ModelBundle, feature_row: pd.DataFrame
) -> dict[str, Any]:
    """Score a single already-feature-engineered transaction row.

    Returns probability, risk tier, and the top-N SHAP features (by
    descending absolute SHAP value) as a list of
    `{"feature": str, "shap_value": float}` dicts -- the exact shape stored
    in `fraud_predictions.top_features` / `test_predictions.top_features`.
    """
    X = feature_row[bundle.feature_columns].astype("float32")
    probability = float(bundle.pipeline.predict_proba(X)[0, 1])

    shap_values = bundle.explainer.shap_values(X)
    shap_row = np.asarray(shap_values)[0]
    order = np.argsort(-np.abs(shap_row))[:TOP_N_FEATURES]
    top_features = [
        {"feature": bundle.feature_columns[i], "shap_value": float(shap_row[i])}
        for i in order
    ]

    return {
        "probability": probability,
        "risk_tier": risk_tier(probability),
        "top_features": top_features,
        "model_version": bundle.model_version,
    }