Spaces:
Running on Zero
Running on Zero
| """ | |
| decision_engine.py | |
| ------------------ | |
| RazorShield Risk Decision Engine core orchestrator. | |
| Loads pre-trained calibrated transaction models and Phase 4 deployable spike models. | |
| Executes real-time transaction risk evaluation with structured evidence output. | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from pathlib import Path | |
| from typing import Any | |
| import joblib | |
| import numpy as np | |
| import pandas as pd | |
| from src.risk_engine.campaign import CampaignManager | |
| from src.risk_engine.merchant_state import MerchantStateManager | |
| from src.risk_engine.policies import PolicyEngine | |
| from src.risk_engine.schemas import CampaignRegistration, RiskDecision, TransactionInput | |
| ROOT = Path(__file__).resolve().parents[2] | |
| MODELS_DIR = ROOT / "models" | |
| LOGGER = logging.getLogger("risk-decision-engine") | |
| NUMERIC_FEATURES_TX = [ | |
| "amount", "amount_log1p", "hour", "day_of_week", "is_weekend", | |
| "customer_txn_count_past", "customer_amount_mean_past", "customer_amount_std_past", | |
| "device_txn_count_past", "customer_amount_dev", "identity_available", | |
| "missing_p_email", "missing_r_email", "missing_addr1", "missing_device_info" | |
| ] | |
| CATEGORICAL_FEATURES_TX = [ | |
| "ProductCD", "card1", "card2", "card3", "card4", "card5", "card6", | |
| "addr1", "addr2", "P_emaildomain", "R_emaildomain", "DeviceType", "DeviceInfo" | |
| ] | |
| PHASE4_SPIKE_FEATURES = [ | |
| "rolling_txn_15m", | |
| "baseline_txn_15m", | |
| "velocity_ratio", | |
| "estimated_fraud_rate_15m", | |
| "baseline_fraud_rate", | |
| "estimated_fraud_rate_deviation", | |
| "amount_deviation", | |
| "fraud_signal_ratio", | |
| "estimated_fraud_count_15m", | |
| "expected_fraud_count_15m", | |
| "fraud_excess_ratio", | |
| "volume_deviation", | |
| "fraud_excess_minus_velocity", | |
| "amount_shift_indicator", | |
| ] | |
| class RiskDecisionEngine: | |
| """Core deterministic risk decision engine.""" | |
| def __init__( | |
| self, | |
| policy_mode: str = "BALANCED", | |
| models_dir: Path | None = None, | |
| ): | |
| if models_dir is None: | |
| models_dir = MODELS_DIR | |
| self.models_dir = models_dir | |
| self.state_manager = MerchantStateManager() | |
| self.campaign_manager = CampaignManager() | |
| self.policy_engine = PolicyEngine(mode=policy_mode) | |
| self._load_models() | |
| def _load_models(self): | |
| """Loads trained transaction, calibration, and spike model artifacts.""" | |
| tx_path = self.models_dir / "transaction_model" / "xgboost_model.joblib" | |
| enc_path = self.models_dir / "transaction_model" / "encoder.joblib" | |
| cal_path = self.models_dir / "transaction_model" / "calibrated_model.joblib" | |
| spike_path = self.models_dir / "spike_model" / "xgboost_spike_model_v2.joblib" | |
| if not tx_path.exists() or not spike_path.exists(): | |
| raise FileNotFoundError("Required model artifacts missing from models directory.") | |
| self.tx_model = joblib.load(tx_path) | |
| self.tx_encoder = joblib.load(enc_path) | |
| self.cal_model = joblib.load(cal_path) if cal_path.exists() else None | |
| self.spike_model = joblib.load(spike_path) | |
| def _predict_calibrated_fraud_prob(self, tx: TransactionInput) -> float: | |
| """Predicts calibrated transaction-level fraud probability P(fraud | transaction).""" | |
| hour = tx.event_time.hour | |
| day_of_week = tx.event_time.weekday() | |
| is_weekend = 1 if day_of_week >= 5 else 0 | |
| amount_log1p = float(np.log1p(max(0.0, tx.amount))) | |
| # Historical proxies from merchant state | |
| m_state = self.state_manager.get_state(tx.merchant_id) | |
| cust_past_cnt = max(0, m_state.transaction_count - 1) | |
| cust_mean_past = m_state.baseline_amount | |
| cust_std_past = 0.0 | |
| dev_past_cnt = max(0, m_state.transaction_count - 1) | |
| cust_dev = float(tx.amount / max(1.0, cust_mean_past)) | |
| num_vals = [ | |
| tx.amount, amount_log1p, hour, day_of_week, is_weekend, | |
| cust_past_cnt, cust_mean_past, cust_std_past, | |
| dev_past_cnt, cust_dev, 1, 0, 0, 0, 0 | |
| ] | |
| cat_vals = [["unknown"] * len(CATEGORICAL_FEATURES_TX)] | |
| cat_encoded = self.tx_encoder.transform(cat_vals) | |
| X_tx = np.hstack([np.array(num_vals, dtype=np.float32).reshape(1, -1), cat_encoded.astype(np.float32)]) | |
| raw_prob = float(self.tx_model.predict_proba(X_tx)[0, 1]) | |
| if self.cal_model is not None: | |
| if hasattr(self.cal_model, "transform"): | |
| cal_prob = float(self.cal_model.transform(np.array([raw_prob]))[0]) | |
| else: | |
| cal_prob = float(self.cal_model.predict_proba(np.array([[raw_prob]]))[0, 1]) | |
| else: | |
| cal_prob = raw_prob | |
| return float(min(1.0, max(0.0, cal_prob))) | |
| def register_campaign(self, campaign: CampaignRegistration): | |
| """Registers a merchant promotional campaign.""" | |
| self.campaign_manager.register_campaign(campaign) | |
| def process_transaction( | |
| self, | |
| tx: TransactionInput, | |
| calibrated_fraud_prob: float | None = None, | |
| ) -> RiskDecision: | |
| """ | |
| Processes a single transaction through the risk decision pipeline. | |
| Returns a RiskDecision object with structured evidence. | |
| """ | |
| # 1. Calibrated transaction model | |
| if calibrated_fraud_prob is None: | |
| cal_prob = self._predict_calibrated_fraud_prob(tx) | |
| else: | |
| cal_prob = float(min(1.0, max(0.0, calibrated_fraud_prob))) | |
| # 2. Chronological merchant state update | |
| feature_dict = self.state_manager.update_merchant( | |
| merchant_id=tx.merchant_id, | |
| event_time=tx.event_time, | |
| amount=tx.amount, | |
| calibrated_fraud_prob=cal_prob, | |
| ) | |
| # 3. Campaign check | |
| is_active, vol_mult = self.campaign_manager.is_campaign_active(tx.merchant_id, tx.event_time) | |
| adj_features = self.campaign_manager.adjust_features_for_campaign( | |
| feature_dict, is_active, vol_mult | |
| ) | |
| # 4. Spike model prediction using deployable features ONLY | |
| X_spike = np.array( | |
| [[adj_features[f] for f in PHASE4_SPIKE_FEATURES]], dtype=np.float32 | |
| ) | |
| spike_prob = float(self.spike_model.predict_proba(X_spike)[0, 1]) | |
| spike_prob = float(min(1.0, max(0.0, spike_prob))) | |
| # 5. Policy evaluation & structured evidence generation | |
| decision = self.policy_engine.evaluate_decision( | |
| tx=tx, | |
| calibrated_fraud_prob=cal_prob, | |
| spike_prob=spike_prob, | |
| feature_dict=adj_features, | |
| campaign_active=is_active, | |
| ) | |
| return decision | |
| def reset_state(self): | |
| """Resets merchant states and campaigns.""" | |
| self.state_manager.reset() | |
| self.campaign_manager.clear() | |