""" Fraud detection hook example. This hook identifies suspicious transaction patterns and flags or rejects them. Uses simple heuristics like large amounts, high frequency, and unusual patterns. """ from datetime import datetime, timedelta from decimal import Decimal from typing import Dict, List, Tuple from wallet_utils.hooks import HookContext from wallet_utils.exceptions import HookRejectionError class FraudDetector: """ Simple fraud detection system for wallet transactions. In production, replace with machine learning models and sophisticated detection. """ def __init__(self): # Track recent transactions: {user_id: [(timestamp, amount, operation), ...]} self._recent_transactions: Dict[int, List[Tuple[datetime, Decimal, str]]] = {} # Thresholds self.large_amount_threshold = Decimal("10000.00") self.high_frequency_threshold = 10 # transactions per hour self.rapid_fire_seconds = 5 # seconds between transactions self.velocity_check_minutes = 60 # check last N minutes # Suspicious patterns self.suspicious_amounts = [ Decimal("9999.00"), # Just below limits Decimal("9999.99"), ] def record_transaction(self, user_id: int, amount: Decimal, operation: str) -> None: """Record a transaction for fraud tracking.""" if user_id not in self._recent_transactions: self._recent_transactions[user_id] = [] self._recent_transactions[user_id].append((datetime.now(), amount, operation)) # Keep only last 24 hours cutoff = datetime.now() - timedelta(hours=24) self._recent_transactions[user_id] = [ tx for tx in self._recent_transactions[user_id] if tx[0] > cutoff ] def check_large_amount(self, amount: Decimal) -> Tuple[bool, str]: """Check if amount is suspiciously large.""" if amount > self.large_amount_threshold: return True, f"Amount ${amount} exceeds threshold ${self.large_amount_threshold}" return False, "" def check_suspicious_amount(self, amount: Decimal) -> Tuple[bool, str]: """Check if amount matches suspicious patterns.""" for suspicious in self.suspicious_amounts: if amount == suspicious: return True, f"Amount ${amount} matches suspicious pattern" return False, "" def check_high_frequency(self, user_id: int) -> Tuple[bool, str]: """Check if user has too many recent transactions.""" if user_id not in self._recent_transactions: return False, "" cutoff = datetime.now() - timedelta(minutes=self.velocity_check_minutes) recent = [tx for tx in self._recent_transactions[user_id] if tx[0] > cutoff] if len(recent) >= self.high_frequency_threshold: return True, f"{len(recent)} transactions in last {self.velocity_check_minutes} minutes" return False, "" def check_rapid_fire(self, user_id: int) -> Tuple[bool, str]: """Check if transactions are happening too quickly.""" if user_id not in self._recent_transactions: return False, "" transactions = self._recent_transactions[user_id] if len(transactions) < 2: return False, "" # Check last transaction last_tx_time = transactions[-1][0] time_since = (datetime.now() - last_tx_time).total_seconds() if time_since < self.rapid_fire_seconds: return True, f"Transaction within {time_since:.1f} seconds of previous" return False, "" def check_unusual_pattern(self, user_id: int, amount: Decimal) -> Tuple[bool, str]: """Check for unusual transaction patterns.""" if user_id not in self._recent_transactions: return False, "" recent = self._recent_transactions[user_id][-10:] # Last 10 transactions if len(recent) < 3: return False, "" # Check if all recent transactions are nearly identical (automation?) amounts = [tx[1] for tx in recent] if len(set(amounts)) == 1 and amount == amounts[0]: return True, "Identical repeated transactions detected" return False, "" # Global detector instance _detector = FraudDetector() def check_fraud_patterns(context: HookContext) -> bool: """ PRE hook to detect fraudulent transaction patterns. Checks for: - Unusually large amounts - High transaction frequency - Rapid-fire transactions - Suspicious amount patterns - Unusual repetitive patterns Args: context: Hook context with transaction details Returns: True if no fraud detected, raises HookRejectionError if suspicious Raises: HookRejectionError: When fraudulent pattern is detected Example: >>> from wallet_utils.hooks import register_hook, HookType >>> register_hook( ... name="fraud_detector", ... hook_type=HookType.PRE, ... callback=check_fraud_patterns, ... operation="*", # Check all operations ... priority=10, # Highest priority security check ... ) """ # Only check deductions and transfers (adds are usually safe) if context.operation == "add": return True user_id = context.user_id amount = context.amount operation = context.operation fraud_reasons = [] # Run all fraud checks is_large, reason = _detector.check_large_amount(amount) if is_large: fraud_reasons.append(("large_amount", reason)) is_suspicious, reason = _detector.check_suspicious_amount(amount) if is_suspicious: fraud_reasons.append(("suspicious_amount", reason)) is_high_freq, reason = _detector.check_high_frequency(user_id) if is_high_freq: fraud_reasons.append(("high_frequency", reason)) is_rapid, reason = _detector.check_rapid_fire(user_id) if is_rapid: fraud_reasons.append(("rapid_fire", reason)) is_unusual, reason = _detector.check_unusual_pattern(user_id, amount) if is_unusual: fraud_reasons.append(("unusual_pattern", reason)) # If multiple fraud indicators, reject if len(fraud_reasons) >= 2: raise HookRejectionError( error_code="FRAUD_DETECTED", message="Transaction flagged as potentially fraudulent", details={ "user_id": user_id, "amount": float(amount), "operation": operation, "fraud_indicators": [ {"type": ftype, "reason": reason} for ftype, reason in fraud_reasons ], "severity": "high" if len(fraud_reasons) >= 3 else "medium", } ) # If single indicator and large amount, also reject if len(fraud_reasons) == 1 and fraud_reasons[0][0] == "large_amount": raise HookRejectionError( error_code="AMOUNT_REQUIRES_APPROVAL", message="Large transaction requires manual approval", details={ "user_id": user_id, "amount": float(amount), "operation": operation, "reason": fraud_reasons[0][1], } ) # Store fraud check result in metadata context.metadata["fraud_check_passed"] = True context.metadata["fraud_score"] = len(fraud_reasons) return True def record_transaction_for_fraud_detection(context: HookContext) -> None: """ POST hook to record transaction for fraud tracking. Should be registered alongside check_fraud_patterns to track patterns. Args: context: Hook context with transaction details Example: >>> from wallet_utils.hooks import register_hook, HookType >>> register_hook( ... name="fraud_tracker", ... hook_type=HookType.POST, ... callback=record_transaction_for_fraud_detection, ... operation="*", ... priority=100, ... ) """ _detector.record_transaction( context.user_id, context.amount, context.operation ) def get_detector() -> FraudDetector: """Get the global fraud detector instance.""" return _detector def configure_fraud_thresholds( large_amount: Decimal | None = None, high_frequency: int | None = None, rapid_fire_seconds: int | None = None, ) -> None: """ Configure fraud detection thresholds. Args: large_amount: Threshold for large transactions high_frequency: Max transactions per hour rapid_fire_seconds: Min seconds between transactions """ if large_amount is not None: _detector.large_amount_threshold = large_amount if high_frequency is not None: _detector.high_frequency_threshold = high_frequency if rapid_fire_seconds is not None: _detector.rapid_fire_seconds = rapid_fire_seconds