""" SHAP Explainer Module - Uses XGBoost's native pred_contribs (no shap library dependency at runtime) - Cached DMatrix construction - top_n parameter """ import numpy as np import xgboost as xgb from src.value_coercion import coerce_float FEATURE_DESCRIPTIONS = { 'tx_count_total': 'Total number of outgoing transactions', 'tx_count_7d': 'Transaction count in the last 7 days', 'amount_sent_total': 'Total amount of funds sent', 'amount_sent_7d': 'Amount of funds sent in the last 7 days', 'amount_received_total': 'Total amount of funds received', 'amount_received_7d': 'Amount of funds received in the last 7 days', 'forward_ratio': 'Percentage of received funds immediately forwarded', 'avg_tx_amount': 'Average transaction amount sent', 'amount_std': 'Consistency of transaction amounts', 'in_out_ratio': 'Ratio of received to sent funds', 'pagerank_score': 'Network influence of this account', 'betweenness_score': 'Bridge importance — how often this account lies on shortest paths', 'in_degree': 'Number of accounts sending funds to this account', 'out_degree': 'Number of accounts receiving funds from this account', 'fan_in_ratio': 'Concentration of incoming vs outgoing connections', 'community_encoded': 'Fraud rate of the network community this account belongs to', 'cycle_length': 'Length of circular transaction loop this account is part of (0 if none)', 'cycle_max_amount': 'Peak amount transacted in the detected circular loop', 'account_age_days': 'Age of the account based on transaction history', 'days_since_last_tx': 'Days elapsed since the most recent transaction', 'currency_diversity': 'Number of distinct currencies used', 'channel_diversity': 'Number of distinct payment channels used', 'bank_diversity': 'Number of distinct destination banks used', 'velocity_ratio_7d': 'Proportion of all-time activity concentrated in last 7 days', 'gnn_fraud_score': 'Graph Neural Network fraud probability from neighbourhood analysis', 'hybrid_score': 'Topological fraud risk based on directional graph flow analysis.', } from functools import lru_cache @lru_cache(maxsize=100) def explain_prediction( account_id: str, feature_df_hash: int, # hash so cache clears when features change top_n: int = 5, ) -> list[dict]: from src.state import AppState fba = AppState.features_by_account bundle = AppState.xgb_bundle if not bundle or not fba: return [] feat = fba.get(account_id) if not feat: return [] feature_cols = [c for c in bundle['feature_cols'] if c in feat] # Build a guaranteed float64 numpy array and wrap in DMatrix directly. # This bypasses both pandas dtype inference AND the shap library's # TreeExplainer internals (which on some shap/xgboost version combos # tries to float()-cast model metadata stored as '[8.754906E-1]'). vals = np.array( [coerce_float(feat.get(col, 0.0)) for col in feature_cols], dtype=np.float64, ) dmat = xgb.DMatrix(vals.reshape(1, -1), feature_names=feature_cols) # XGBoost's native SHAP contributions: shape (1, n_features + 1). # Last column is the bias term (base score), not a feature contribution. contribs = bundle['model'].get_booster().predict(dmat, pred_contribs=True) sv_row = contribs[0, :-1] # drop bias column results = [] for i, col in enumerate(feature_cols): sv = float(sv_row[i]) results.append({ 'feature_name': col, 'shap_value': sv, 'feature_value': float(vals[i]), 'direction': 'increases risk' if sv > 0 else 'decreases risk', 'description': FEATURE_DESCRIPTIONS.get(col, col), }) results.sort(key=lambda x: abs(x['shap_value']), reverse=True) return results[:top_n]