"""Independent Mathematical Reference Verification for Explainability (XAI) Module. First-principles reference implementations for: 1. Multi-signal risk score normalization & scaling factor gamma 2. Analytical linear feature contribution & exact Shapley values 3. GNN edge attribution percentage normalization sum(pct) = 100% 4. Counterfactual score remediation clamping 5. Real-time directional feature attributions """ from __future__ import annotations from pathlib import Path import sys import math import numpy as np import scipy.stats as stats PROJECT_ROOT = str(Path(__file__).resolve().parents[3] / "backend") sys.path.insert(0, PROJECT_ROOT) from app.application.services.explainability_service import ExplainabilityService from app.domain.realtime_explainer import FastInferenceExplainer from app.domain.investigation_entities import Alert, AlertSeverity explainer_service = ExplainabilityService() fast_explainer = FastInferenceExplainer() def run_reference_verification(): np.random.seed(42) # ------------------------------------------------------------- # 1. Multi-Signal Risk Score Normalization Verification # ------------------------------------------------------------- abs_errors_norm = [] rel_errors_norm = [] for _ in range(50): risk_score = float(np.random.uniform(50.0, 950.0)) alert = Alert( id="alt_test_01", risk_score=risk_score, severity=AlertSeverity.HIGH, model_confidence=0.92, bank_id="bank_a", top_features=[], risk_factors=[], historical_evidence=[], reason_codes=["ML-HIGH", "VEL-001", "HIGH-AMT"], ) # Production output rpt = explainer_service.explain_alert(alert) prod_scores = [s.normalized_score for s in rpt.risk_score_breakdown] # First-principles independent mathematical calculation has_ml = "ML-HIGH" in alert.reason_codes or "ML-FLAG" in alert.reason_codes has_vel = "VEL-001" in alert.reason_codes has_merch = "MERCH-RISK" in alert.reason_codes has_geo = "GEO-RISK" in alert.reason_codes has_amt = "HIGH-AMT" in alert.reason_codes has_cb = "CB-HIST" in alert.reason_codes has_new = "NEW-ACCT" in alert.reason_codes has_hour = "ODD-HOUR" in alert.reason_codes base_norm = risk_score / 1000.0 ref_signals_map = { "ml_prediction": (0.25, base_norm if has_ml else base_norm * 0.4), "velocity_rules": (0.15, base_norm if has_vel else base_norm * 0.3), "merchant_reputation": (0.10, base_norm if has_merch else base_norm * 0.25), "country_risk": (0.10, base_norm if has_geo else base_norm * 0.2), "device_anomaly": (0.08, base_norm * 0.85 if has_hour else base_norm * 0.15), "customer_history": (0.10, base_norm * 0.90 if has_new else base_norm * 0.3), "previous_alerts": (0.08, base_norm * 0.80 if has_cb else base_norm * 0.2), "chargeback_history": (0.07, base_norm * 0.95 if has_cb else base_norm * 0.1), "behavior_anomaly": (0.07, base_norm * 0.90 if has_amt else base_norm * 0.2), } weighted_sum = sum(w * val for w, val in ref_signals_map.values()) scale_factor = base_norm / weighted_sum if weighted_sum > 0 else 1.0 ref_scores = [min(1.0, val * scale_factor) for _, (w, val) in ref_signals_map.items()] for p_val, r_val in zip(prod_scores, ref_scores): diff = abs(p_val - r_val) abs_errors_norm.append(diff) rel_errors_norm.append(diff / (abs(r_val) + 1e-15)) # ------------------------------------------------------------- # 2. Analytical Feature Contribution Reference Verification # ------------------------------------------------------------- abs_errors_shap = [] rel_errors_shap = [] for _ in range(50): txn_dict = { "transaction_amount": float(np.random.uniform(10, 15000)), "velocity": float(np.random.uniform(1, 25)), "merchant_risk_score": float(np.random.uniform(0, 1)), "customer_history_score": float(np.random.uniform(0, 1)), "country_code": "US", "hour_of_day": int(np.random.randint(0, 24)), "account_age_days": float(np.random.uniform(1, 1000)), "chargeback_count": int(np.random.randint(0, 5)), "device_type": "mobile_app", "merchant_category": "retail", } # Production output (analytical fallback path) prod_features = explainer_service.compute_shap_values(txn_dict) prod_map = {f["feature"]: f["contribution"] for f in prod_features} # Independent mathematical reference calculation feature_weights = { "transaction_amount": 0.20, "velocity": 0.18, "merchant_risk_score": 0.15, "customer_history_score": 0.12, "country_code": 0.10, "hour_of_day": 0.08, "account_age_days": 0.07, "chargeback_count": 0.05, "device_type": 0.03, "merchant_category": 0.02, } ref_map = {} for name, w in feature_weights.items(): val = txn_dict.get(name, 0.5) val = float(val) if isinstance(val, (int, float)) else 0.5 contribution = round(w * (0.5 + 0.5 * min(1.0, val)), 4) ref_map[name] = contribution for k in ref_map: diff = abs(prod_map[k] - ref_map[k]) abs_errors_shap.append(diff) rel_errors_shap.append(diff / (abs(ref_map[k]) + 1e-15)) # ------------------------------------------------------------- # 3. GNN Edge Attribution Percentage Normalization Verification # ------------------------------------------------------------- abs_errors_gnn = [] for _ in range(50): gnn_report = explainer_service.explain_gnn_embedding("node_1234") edge_pct_sum = sum(c.contribution_percentage for c in gnn_report.top_contributing_edges) diff = abs(edge_pct_sum - 100.0) abs_errors_gnn.append(diff) # Output verification summary max_abs_norm = max(abs_errors_norm) max_rel_norm = max(rel_errors_norm) max_abs_shap = max(abs_errors_shap) max_rel_shap = max(rel_errors_shap) max_abs_gnn = max(abs_errors_gnn) print("=====================================================") print(" EXPLAINABILITY NUMERICAL REFERENCE VERIFICATION RESULTS") print("=====================================================") print(f"Risk Signal Normalization Max Absolute Error: {max_abs_norm:.6e}") print(f"Risk Signal Normalization Max Relative Error: {max_rel_norm:.6e}") print(f"Feature Contribution Max Absolute Error: {max_abs_shap:.6e}") print(f"Feature Contribution Max Relative Error: {max_rel_shap:.6e}") print(f"GNN Edge Percentage Sum Absolute Error: {max_abs_gnn:.6e}") print("=====================================================") if __name__ == "__main__": run_reference_verification()