Fraud Detection Systems: Machine Learning Models for Payment Security

Introduction

Payment fraud costs the global economy over $28 billion annually. Modern fraud detection requires machine learning models that can identify suspicious patterns in real-time while minimizing false positives. This guide covers building production fraud detection systems with practical ML techniques, feature engineering, and deployment strategies.

Key Statistics:

1 in 33 payment transactions is fraudulent
Average fraud detection latency: 2-3 seconds
False positive rate target: <0.5%
ML-based systems reduce fraud by 40-60%

Core Concepts & Terminology

1. Fraud Detection

Identifying unauthorized or suspicious transactions using patterns and anomalies.

2. Feature Engineering

Creating meaningful features from raw transaction data to improve model performance.

3. Anomaly Detection

Identifying transactions that deviate significantly from normal patterns.

4. Real-Time Detection

Processing transactions and making fraud decisions within milliseconds.

5. Class Imbalance

Problem where fraudulent transactions are rare (0.1-1%) compared to legitimate ones.

6. False Positive

Legitimate transaction incorrectly flagged as fraudulent.

7. False Negative

Fraudulent transaction incorrectly classified as legitimate.

8. Precision & Recall

Precision: % of flagged transactions that are actually fraudulent Recall: % of actual fraudulent transactions that are detected

9. Model Drift

When model performance degrades over time due to changing fraud patterns.

10. Ensemble Methods

Combining multiple models to improve detection accuracy.

Fraud Detection Architecture

┌─────────────────────────────────────────────────────────────┐
│                    Transaction Flow                          │
└─────────────────────────────────────────────────────────────┘
                            │
┌───────────────────────────▼───────────────────────────────┐
│              Real-Time Feature Extraction                  │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐    │
│  │ Transaction  │  │ User History │  │ Merchant     │    │
│  │ Features     │  │ Features     │  │ Features     │    │
│  └──────────────┘  └──────────────┘  └──────────────┘    │
└───────────────────────────▼───────────────────────────────┘
                            │
┌───────────────────────────▼───────────────────────────────┐
│              Model Scoring Layer                           │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐    │
│  │ Gradient     │  │ Random       │  │ Neural       │    │
│  │ Boosting     │  │ Forest       │  │ Network      │    │
│  └──────────────┘  └──────────────┘  └──────────────┘    │
└───────────────────────────▼───────────────────────────────┘
                            │
┌───────────────────────────▼───────────────────────────────┐
│              Decision Engine                               │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐    │
│  │ Risk Score   │  │ Rule Engine  │  │ Velocity     │    │
│  │ Calculation  │  │ (Thresholds) │  │ Checks       │    │
│  └──────────────┘  └──────────────┘  └──────────────┘    │
└───────────────────────────▼───────────────────────────────┘
                            │
                ┌───────────┴───────────┐
                │                       │
                ▼                       ▼
            APPROVE              CHALLENGE/BLOCK

Feature Engineering for Fraud Detection

Transaction Features

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

class TransactionFeatureExtractor:
    """Extract features from transaction data"""
    
    def __init__(self, user_history_window_days: int = 90):
        self.window_days = user_history_window_days
    
    def extract_transaction_features(self, transaction: dict) -> dict:
        """Extract basic transaction features"""
        features = {
            # Amount features
            'amount': transaction['amount'],
            'amount_log': np.log1p(transaction['amount']),
            'amount_normalized': transaction['amount'] / 1000,  # Normalize to thousands
            
            # Time features
            'hour_of_day': datetime.fromisoformat(transaction['timestamp']).hour,
            'day_of_week': datetime.fromisoformat(transaction['timestamp']).weekday(),
            'is_weekend': datetime.fromisoformat(transaction['timestamp']).weekday() >= 5,
            'is_night': datetime.fromisoformat(transaction['timestamp']).hour in [0, 1, 2, 3, 4, 5],
            
            # Merchant features
            'merchant_category': transaction.get('merchant_category', 'unknown'),
            'is_high_risk_merchant': transaction.get('merchant_category') in ['gambling', 'adult'],
            
            # Device features
            'device_type': transaction.get('device_type', 'unknown'),
            'is_mobile': transaction.get('device_type') == 'mobile',
        }
        
        return features
    
    def extract_user_history_features(self, user_id: str, 
                                     transaction: dict,
                                     user_transactions: list[dict]) -> dict:
        """Extract features based on user history"""
        
        # Filter transactions within window
        cutoff_date = datetime.now() - timedelta(days=self.window_days)
        recent_txns = [
            t for t in user_transactions 
            if datetime.fromisoformat(t['timestamp']) > cutoff_date
        ]
        
        if not recent_txns:
            return {
                'user_transaction_count': 0,
                'user_avg_amount': 0,
                'user_std_amount': 0,
                'user_max_amount': 0,
                'user_merchant_diversity': 0,
                'amount_deviation': 0,
            }
        
        amounts = [t['amount'] for t in recent_txns]
        
        features = {
            'user_transaction_count': len(recent_txns),
            'user_avg_amount': np.mean(amounts),
            'user_std_amount': np.std(amounts),
            'user_max_amount': np.max(amounts),
            'user_min_amount': np.min(amounts),
            
            # Merchant diversity
            'user_merchant_diversity': len(set(t.get('merchant_id') for t in recent_txns)),
            
            # Amount deviation from user average
            'amount_deviation': (transaction['amount'] - np.mean(amounts)) / (np.std(amounts) + 1),
            
            # Frequency features
            'transactions_today': len([t for t in recent_txns 
                                      if datetime.fromisoformat(t['timestamp']).date() == datetime.now().date()]),
            'transactions_this_hour': len([t for t in recent_txns 
                                          if (datetime.now() - datetime.fromisoformat(t['timestamp'])).total_seconds() < 3600]),
        }
        
        return features
    
    def extract_velocity_features(self, user_id: str,
                                 transaction: dict,
                                 user_transactions: list[dict]) -> dict:
        """Extract velocity-based features"""
        
        recent_txns = [
            t for t in user_transactions 
            if (datetime.now() - datetime.fromisoformat(t['timestamp'])).total_seconds() < 3600
        ]
        
        features = {
            'velocity_count_1h': len(recent_txns),
            'velocity_amount_1h': sum(t['amount'] for t in recent_txns),
            'velocity_avg_amount_1h': np.mean([t['amount'] for t in recent_txns]) if recent_txns else 0,
            'is_velocity_spike': len(recent_txns) > 5,  # More than 5 txns in 1 hour
        }
        
        return features
    
    def extract_location_features(self, transaction: dict,
                                 user_transactions: list[dict]) -> dict:
        """Extract location-based features"""
        
        current_location = (transaction.get('latitude'), transaction.get('longitude'))
        
        # Get previous transaction location
        if user_transactions:
            prev_location = (
                user_transactions[-1].get('latitude'),
                user_transactions[-1].get('longitude')
            )
            
            # Calculate distance (simplified)
            distance = np.sqrt(
                (current_location[0] - prev_location[0])**2 + 
                (current_location[1] - prev_location[1])**2
            )
        else:
            distance = 0
        
        features = {
            'location_distance_from_previous': distance,
            'is_impossible_travel': distance > 1000,  # More than 1000 km
            'location_changed': distance > 0.1,
        }
        
        return features

# Usage
extractor = TransactionFeatureExtractor()

transaction = {
    'amount': 150.00,
    'timestamp': datetime.now().isoformat(),
    'merchant_category': 'retail',
    'device_type': 'mobile',
    'latitude': 40.7128,
    'longitude': -74.0060
}

user_transactions = [
    {'amount': 50, 'timestamp': (datetime.now() - timedelta(hours=2)).isoformat()},
    {'amount': 75, 'timestamp': (datetime.now() - timedelta(hours=5)).isoformat()},
]

features = {}
features.update(extractor.extract_transaction_features(transaction))
features.update(extractor.extract_user_history_features('user123', transaction, user_transactions))
features.update(extractor.extract_velocity_features('user123', transaction, user_transactions))
features.update(extractor.extract_location_features(transaction, user_transactions))

print("Extracted features:")
for key, value in features.items():
    print(f"  {key}: {value}")

Machine Learning Models

Gradient Boosting Model

import xgboost as xgb
from sklearn.model_selection import train_test_split
from sklearn.metrics import precision_recall_curve, roc_auc_score
from sklearn.preprocessing import StandardScaler

class FraudDetectionModel:
    """Production fraud detection model"""
    
    def __init__(self):
        self.model = None
        self.scaler = StandardScaler()
        self.feature_names = None
    
    def train(self, X_train: pd.DataFrame, y_train: pd.Series):
        """Train fraud detection model"""
        
        # Handle class imbalance
        scale_pos_weight = len(y_train[y_train == 0]) / len(y_train[y_train == 1])
        
        self.model = xgb.XGBClassifier(
            n_estimators=100,
            max_depth=7,
            learning_rate=0.1,
            subsample=0.8,
            colsample_bytree=0.8,
            scale_pos_weight=scale_pos_weight,  # Handle imbalance
            random_state=42,
            n_jobs=-1
        )
        
        # Scale features
        X_scaled = self.scaler.fit_transform(X_train)
        
        # Train model
        self.model.fit(
            X_scaled, y_train,
            eval_set=[(X_scaled, y_train)],
            verbose=False
        )
        
        self.feature_names = X_train.columns.tolist()
    
    def predict_proba(self, X: pd.DataFrame) -> np.ndarray:
        """Get fraud probability"""
        X_scaled = self.scaler.transform(X)
        return self.model.predict_proba(X_scaled)[:, 1]
    
    def predict(self, X: pd.DataFrame, threshold: float = 0.5) -> np.ndarray:
        """Predict fraud (binary)"""
        proba = self.predict_proba(X)
        return (proba >= threshold).astype(int)
    
    def get_feature_importance(self) -> pd.DataFrame:
        """Get feature importance"""
        importance = self.model.feature_importances_
        return pd.DataFrame({
            'feature': self.feature_names,
            'importance': importance
        }).sort_values('importance', ascending=False)
    
    def evaluate(self, X_test: pd.DataFrame, y_test: pd.Series) -> dict:
        """Evaluate model performance"""
        proba = self.predict_proba(X_test)
        
        # Calculate metrics
        auc_score = roc_auc_score(y_test, proba)
        
        # Find optimal threshold
        precision, recall, thresholds = precision_recall_curve(y_test, proba)
        f1_scores = 2 * (precision * recall) / (precision + recall + 1e-10)
        optimal_idx = np.argmax(f1_scores)
        optimal_threshold = thresholds[optimal_idx]
        
        predictions = (proba >= optimal_threshold).astype(int)
        
        from sklearn.metrics import confusion_matrix, classification_report
        tn, fp, fn, tp = confusion_matrix(y_test, predictions).ravel()
        
        return {
            'auc_score': auc_score,
            'optimal_threshold': optimal_threshold,
            'precision': tp / (tp + fp),
            'recall': tp / (tp + fn),
            'false_positive_rate': fp / (fp + tn),
            'true_positive_rate': tp / (tp + fn),
            'confusion_matrix': {'tn': tn, 'fp': fp, 'fn': fn, 'tp': tp}
        }

# Usage
model = FraudDetectionModel()

# Prepare data
X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

# Train model
model.train(X_train, y_train)

# Evaluate
metrics = model.evaluate(X_test, y_test)
print(f"AUC Score: {metrics['auc_score']:.4f}")
print(f"Precision: {metrics['precision']:.4f}")
print(f"Recall: {metrics['recall']:.4f}")
print(f"False Positive Rate: {metrics['false_positive_rate']:.4f}")

# Feature importance
importance = model.get_feature_importance()
print("\nTop 10 Features:")
print(importance.head(10))

Real-Time Detection Pipeline

Production Scoring

import redis
import json
from typing import Optional

class RealTimeFraudDetector:
    """Real-time fraud detection pipeline"""
    
    def __init__(self, model: FraudDetectionModel, 
                 redis_client: redis.Redis,
                 fraud_threshold: float = 0.7):
        self.model = model
        self.redis = redis_client
        self.fraud_threshold = fraud_threshold
        self.feature_extractor = TransactionFeatureExtractor()
    
    def score_transaction(self, transaction: dict, 
                         user_id: str) -> dict:
        """Score transaction for fraud in real-time"""
        
        try:
            # Extract features
            features = {}
            features.update(self.feature_extractor.extract_transaction_features(transaction))
            
            # Get user history from cache
            user_history_key = f"user_history:{user_id}"
            user_history_json = self.redis.get(user_history_key)
            user_history = json.loads(user_history_json) if user_history_json else []
            
            features.update(self.feature_extractor.extract_user_history_features(
                user_id, transaction, user_history
            ))
            features.update(self.feature_extractor.extract_velocity_features(
                user_id, transaction, user_history
            ))
            features.update(self.feature_extractor.extract_location_features(
                transaction, user_history
            ))
            
            # Convert to DataFrame
            X = pd.DataFrame([features])
            
            # Get fraud probability
            fraud_probability = self.model.predict_proba(X)[0]
            
            # Make decision
            is_fraud = fraud_probability >= self.fraud_threshold
            
            # Log transaction
            self.log_transaction(user_id, transaction, fraud_probability, is_fraud)
            
            # Update user history
            self.update_user_history(user_id, transaction)
            
            return {
                'transaction_id': transaction.get('id'),
                'fraud_probability': float(fraud_probability),
                'is_fraud': bool(is_fraud),
                'decision': 'BLOCK' if is_fraud else 'APPROVE',
                'features': features
            }
        
        except Exception as e:
            # Fail open on error
            return {
                'transaction_id': transaction.get('id'),
                'fraud_probability': 0.0,
                'is_fraud': False,
                'decision': 'APPROVE',
                'error': str(e)
            }
    
    def log_transaction(self, user_id: str, transaction: dict,
                       fraud_probability: float, is_fraud: bool):
        """Log transaction for monitoring"""
        log_entry = {
            'user_id': user_id,
            'transaction_id': transaction.get('id'),
            'amount': transaction.get('amount'),
            'fraud_probability': fraud_probability,
            'is_fraud': is_fraud,
            'timestamp': datetime.now().isoformat()
        }
        
        # Store in Redis for real-time monitoring
        self.redis.lpush(f"fraud_logs:{user_id}", json.dumps(log_entry))
        self.redis.ltrim(f"fraud_logs:{user_id}", 0, 999)  # Keep last 1000
    
    def update_user_history(self, user_id: str, transaction: dict):
        """Update user transaction history"""
        user_history_key = f"user_history:{user_id}"
        
        # Get current history
        user_history_json = self.redis.get(user_history_key)
        user_history = json.loads(user_history_json) if user_history_json else []
        
        # Add new transaction
        user_history.append(transaction)
        
        # Keep last 100 transactions
        user_history = user_history[-100:]
        
        # Store back
        self.redis.setex(
            user_history_key,
            86400 * 30,  # 30 days
            json.dumps(user_history)
        )

# Usage
redis_client = redis.Redis(host='localhost', port=6379)
detector = RealTimeFraudDetector(model, redis_client, fraud_threshold=0.7)

transaction = {
    'id': 'txn_123',
    'amount': 500.00,
    'timestamp': datetime.now().isoformat(),
    'merchant_category': 'retail',
    'device_type': 'mobile'
}

result = detector.score_transaction(transaction, 'user_456')
print(f"Decision: {result['decision']}")
print(f"Fraud Probability: {result['fraud_probability']:.4f}")

Best Practices

Handle Class Imbalance: Use techniques like SMOTE, class weights, or threshold adjustment
Monitor Model Drift: Track performance metrics and retrain when needed
Minimize False Positives: Balance precision and recall based on business needs
Real-Time Processing: Use caching and efficient feature extraction
Explainability: Provide reasons for fraud decisions
Continuous Learning: Retrain models with new fraud patterns
A/B Testing: Test new models before production deployment
Fallback Mechanisms: Have rules-based backup for model failures
Privacy Protection: Encrypt sensitive data and comply with regulations
Team Collaboration: Work with fraud analysts to improve features

Common Pitfalls

Ignoring Class Imbalance: Using accuracy as metric for imbalanced data
Data Leakage: Using future information in features
No Monitoring: Deploying without tracking performance
Overfitting: Creating models that don’t generalize
Ignoring False Positives: Blocking legitimate customers
Static Models: Not updating models with new fraud patterns
Poor Feature Engineering: Using raw features without domain knowledge
No Explainability: Customers don’t understand why they’re blocked
Ignoring Latency: Models that are too slow for real-time use
No Fallback: System fails completely when model fails

Model Comparison Table

Model	Accuracy	Speed	Interpretability	Scalability
Logistic Regression	85%	Very Fast	High	Excellent
Random Forest	92%	Fast	Medium	Good
Gradient Boosting	94%	Medium	Medium	Good
Neural Network	95%	Slow	Low	Excellent
Ensemble	96%	Medium	Low	Good

External Resources

Conclusion

Building effective fraud detection systems requires careful feature engineering, appropriate ML models, and robust real-time infrastructure. By implementing the techniques in this guide, you can detect 90%+ of fraud while keeping false positives below 0.5%. The key is continuous monitoring and improvement as fraud patterns evolve.

Next Steps:

Collect and prepare training data
Engineer features based on domain knowledge
Train and evaluate models
Deploy real-time detection pipeline
Monitor and iterate