risk_explainability.py

"""Phase 10: SHAP explainability helpers."""

from __future__ import annotations

from typing import Any

import numpy as np
import pandas as pd

try:
	import shap
except ImportError as exc:  # pragma: no cover - surfaced at app startup
	shap = None
	_SHAP_IMPORT_ERROR = exc
else:
	_SHAP_IMPORT_ERROR = None


FEATURE_LABELS: dict[str, str] = {
	"amount": "Transaction amount",
	"use_chip": "Chip usage",
	"merchant_city": "Merchant city",
	"merchant_state": "Merchant state",
	"mcc": "Merchant category code",
	"errors": "Input errors",
	"current_age": "Current age",
	"retirement_age": "Retirement age",
	"birth_year": "Birth year",
	"birth_month": "Birth month",
	"credit_score": "Credit score",
	"transaction_velocity": "Transaction velocity",
	"transaction_gap_from_first_day": "Gap from first transaction",
	"user_tx_frequency": "User transaction frequency",
	"user_active_day_index": "User active day index",
	"amount_deviation": "Amount deviation",
	"rolling_mean_amount": "Rolling mean amount",
	"rolling_std_amount": "Rolling std amount",
	"transaction_history_length": "Transaction history length",
	"is_new_user": "New user indicator",
	"card_to_history_ratio": "Card-to-history ratio",
	"high_card_velocity_flag": "High card velocity flag",
	"merchant_fraud_rate": "Merchant fraud rate",
	"merchant_tx_count": "Merchant transaction count",
	"merchant_avg_amount": "Merchant average amount",
	"merchant_std_amount": "Merchant amount dispersion",
	"merchant_risk_score": "Merchant risk score",
	"merchant_outlier_score": "Merchant outlier score",
	"geo_cluster_fraud_rate": "Geo cluster fraud rate",
	"peer_cluster_fraud_rate": "Peer cluster fraud rate",
	"cluster_avg_amount": "Cluster average amount",
	"cluster_std_amount": "Cluster amount dispersion",
	"cluster_outlier_score": "Cluster outlier score",
	"anomaly_score": "Anomaly score",
	"card_on_dark_web": "Card seen on dark web",
	"cvv": "CVV",
	"expires": "Card expiry",
	"card_number": "Card number",
	"has_chip": "Chip availability",
	"num_cards_issued": "Cards issued to user",
	"credit_limit": "Credit limit",
}

CATEGORY_RULES: dict[str, set[str]] = {
	"account_takeover": {
		"high_card_velocity_flag",
		"transaction_velocity",
		"user_tx_frequency",
		"transaction_gap_from_first_day",
		"rolling_mean_amount",
		"rolling_std_amount",
	},
	"synthetic_identity": {
		"is_new_user",
		"transaction_history_length",
		"card_to_history_ratio",
		"num_cards_issued",
		"credit_score",
		"birth_year",
		"birth_month",
		"current_age",
	},
	"merchant_risk": {
		"merchant_risk_score",
		"merchant_fraud_rate",
		"merchant_outlier_score",
		"merchant_tx_count",
		"merchant_avg_amount",
		"merchant_std_amount",
		"mcc",
	},
	"geo_anomaly": {
		"geo_cluster_fraud_rate",
		"peer_cluster_fraud_rate",
		"cluster_outlier_score",
		"merchant_city",
		"merchant_state",
	},
	"card_testing": {
		"card_on_dark_web",
		"cvv",
		"expires",
		"card_number",
		"use_chip",
		"has_chip",
		"amount",
		"errors",
	},
}


def build_explainer(model: Any, feature_columns: list[str] | None = None) -> Any:
	if shap is None:  # pragma: no cover - dependency issue should fail fast at startup
		raise RuntimeError("shap is required for explainability") from _SHAP_IMPORT_ERROR

	if feature_columns is None:
		inferred = getattr(model, "feature_names_in_", None)
		if inferred is None:
			inferred = getattr(model, "feature_names", None)
		feature_columns = list(inferred) if inferred is not None else []

	try:
		if hasattr(model, "set_params"):
			try:
				model.set_params(base_score=0.5)
			except Exception:
				pass
		return shap.TreeExplainer(model)
	except Exception:
		background = pd.DataFrame(
			np.zeros((1, len(feature_columns)), dtype=np.float32),
			columns=feature_columns,
		)
		return shap.Explainer(model.predict_proba, background)


def _feature_label(feature_name: str) -> str:
	return FEATURE_LABELS.get(feature_name, feature_name.replace("_", " ").title())


def _as_float(value: Any) -> float | None:
	if value is None:
		return None
	try:
		if pd.isna(value):
			return None
	except TypeError:
		pass
	try:
		return float(value)
	except (TypeError, ValueError):
		return None


def _extract_shap_values(explanation: Any) -> tuple[np.ndarray, float]:
	values = getattr(explanation, "values", explanation)
	base_values = getattr(explanation, "base_values", 0.0)

	if isinstance(values, list):
		values = values[-1]

	values_array = np.asarray(values)
	if values_array.ndim == 3:
		values_array = values_array[0, :, -1]
	elif values_array.ndim == 2:
		values_array = values_array[0]
	else:
		values_array = values_array.reshape(-1)

	base_array = np.asarray(base_values).reshape(-1)
	base_value = float(base_array[-1] if base_array.size else 0.0)
	return values_array.astype(float), base_value


def _build_signal(feature: str, shap_value: float, value: Any) -> dict[str, Any]:
	impact = "increased risk" if shap_value > 0 else "reduced risk"
	return {
		"feature": feature,
		"label": _feature_label(feature),
		"value": _as_float(value),
		"shap_value": round(float(shap_value), 6),
		"direction": impact,
	}


def _category_scores(signal_map: dict[str, float]) -> dict[str, float]:
	raw_scores: dict[str, float] = {}
	for category, feature_names in CATEGORY_RULES.items():
		raw_scores[category] = float(
			sum(max(signal_map.get(feature_name, 0.0), 0.0) for feature_name in feature_names)
		)

	total = sum(raw_scores.values())
	if total <= 0:
		return {name: 0.0 for name in raw_scores}

	return {name: round(score / total, 6) for name, score in raw_scores.items()}


def _fraud_type(category_scores: dict[str, float], classification: str) -> tuple[str, float]:
	if classification == "legitimate":
		return "legitimate", 1.0

	if not category_scores:
		return "general_fraud", 0.0

	category, score = max(category_scores.items(), key=lambda item: item[1])
	if score <= 0:
		return "general_fraud", 0.0
	return category, round(float(score), 6)


def explain_prediction(
	*,
	explainer: Any,
	features_df: pd.DataFrame,
	feature_columns: list[str],
	aligned_features: dict[str, float | None],
	classification: str,
	fraud_probability: float,
	top_k: int = 5,
) -> dict[str, Any]:
	explanation = explainer(features_df)
	shap_values, base_value = _extract_shap_values(explanation)

	signal_rows = [
		_build_signal(feature, shap_values[index], aligned_features.get(feature))
		for index, feature in enumerate(feature_columns)
	]

	positive_signals = sorted(
		(signal for signal in signal_rows if signal["shap_value"] > 0),
		key=lambda item: abs(item["shap_value"]),
		reverse=True,
	)[:top_k]
	negative_signals = sorted(
		(signal for signal in signal_rows if signal["shap_value"] < 0),
		key=lambda item: abs(item["shap_value"]),
		reverse=True,
	)[:top_k]

	signal_map = {signal["feature"]: signal["shap_value"] for signal in signal_rows}
	category_scores = _category_scores(signal_map)
	fraud_type, fraud_type_confidence = _fraud_type(category_scores, classification)

	if classification == "legitimate":
		reason_source = negative_signals[0] if negative_signals else None
		explanation_summary = (
			f"The model leaned legitimate because {reason_source['label'].lower()} lowered the risk most."
			if reason_source
			else "The model leaned legitimate because no strong risk-driving signals dominated the baseline."
		)
	else:
		reason_source = positive_signals[0] if positive_signals else None
		explanation_summary = (
			f"The model leaned {classification} because {reason_source['label'].lower()} raised the risk most."
			if reason_source
			else f"The model leaned {classification} because the combined signal pattern exceeded the fraud threshold."
		)

	return {
		"base_value": round(float(base_value), 6),
		"fraud_type": fraud_type,
		"fraud_type_confidence": fraud_type_confidence,
		"fraud_probability": round(float(fraud_probability), 6),
		"explanation_summary": explanation_summary,
		"category_scores": category_scores,
		"top_positive_signals": positive_signals,
		"top_negative_signals": negative_signals,
	}