Spaces:
Running
Running
| """ | |
| AI Fraud Detection System for BankBot | |
| Uses machine learning to detect suspicious transactions | |
| """ | |
| import json | |
| import os | |
| import numpy as np | |
| import pandas as pd | |
| from datetime import datetime, timedelta | |
| from sklearn.ensemble import IsolationForest, RandomForestClassifier | |
| from sklearn.preprocessing import StandardScaler | |
| import pickle | |
| import uuid | |
| FRAUD_ALERTS_FILE = "fraud_alerts.json" | |
| FRAUD_MODEL_FILE = "fraud_model.pkl" | |
| class FraudDetectionEngine: | |
| """Advanced fraud detection using multiple ML algorithms""" | |
| def __init__(self): | |
| self.isolation_forest = None | |
| self.scaler = StandardScaler() | |
| self.load_model() | |
| def load_model(self): | |
| """Load saved model or create new one""" | |
| if os.path.exists(FRAUD_MODEL_FILE): | |
| try: | |
| with open(FRAUD_MODEL_FILE, "rb") as f: | |
| model_data = pickle.load(f) | |
| self.isolation_forest = model_data.get("model") | |
| self.scaler = model_data.get("scaler", StandardScaler()) | |
| except Exception as e: | |
| print(f"Error loading fraud model: {e}") | |
| self._initialize_model() | |
| else: | |
| self._initialize_model() | |
| def _initialize_model(self): | |
| """Initialize Isolation Forest for anomaly detection""" | |
| self.isolation_forest = IsolationForest( | |
| contamination=0.1, # Expect ~10% anomalies | |
| random_state=42, | |
| n_estimators=100 | |
| ) | |
| def save_model(self): | |
| """Save trained model to disk""" | |
| try: | |
| with open(FRAUD_MODEL_FILE, "wb") as f: | |
| pickle.dump({ | |
| "model": self.isolation_forest, | |
| "scaler": self.scaler | |
| }, f) | |
| except Exception as e: | |
| print(f"Error saving fraud model: {e}") | |
| def extract_features(self, transactions): | |
| """ | |
| Extract numerical features from transaction history | |
| Returns: DataFrame with features for ML model | |
| """ | |
| if not transactions or len(transactions) < 2: | |
| return None | |
| df = pd.DataFrame(transactions) | |
| # Convert date strings to datetime | |
| df['date'] = pd.to_datetime(df['date'], errors='coerce') | |
| features = [] | |
| for txn in transactions: | |
| try: | |
| amount = float(txn.get('amount', 0)) | |
| txn_type = 1 if txn.get('type') == 'debit' else 0 | |
| feature_dict = { | |
| 'amount': amount, | |
| 'type': txn_type, | |
| 'hour': datetime.fromisoformat(txn.get('date', '')).hour if txn.get('date') else 12, | |
| 'day_of_week': datetime.fromisoformat(txn.get('date', '')).weekday() if txn.get('date') else 3, | |
| } | |
| features.append(feature_dict) | |
| except Exception as e: | |
| print(f"Error extracting features: {e}") | |
| continue | |
| return pd.DataFrame(features) if features else None | |
| def detect_anomalies(self, transactions): | |
| """ | |
| Detect anomalous transactions using Isolation Forest | |
| Returns: List of anomaly indices and scores | |
| """ | |
| if not transactions or len(transactions) < 2: | |
| return [], [] | |
| features_df = self.extract_features(transactions) | |
| if features_df is None or len(features_df) < 2: | |
| return [], [] | |
| try: | |
| # Normalize features | |
| X = self.scaler.fit_transform(features_df) | |
| # Detect anomalies (-1 = anomaly, 1 = normal) | |
| predictions = self.isolation_forest.predict(X) | |
| anomaly_scores = self.isolation_forest.score_samples(X) | |
| # Find anomalies | |
| anomalies = np.where(predictions == -1)[0].tolist() | |
| return anomalies, anomaly_scores | |
| except Exception as e: | |
| print(f"Error in anomaly detection: {e}") | |
| return [], [] | |
| def calculate_fraud_score(self, transaction, user_history): | |
| """ | |
| Calculate fraud probability for a single transaction (0-100) | |
| Considers: amount, frequency, location, time patterns | |
| """ | |
| score = 0 | |
| reasons = [] | |
| try: | |
| amount = float(transaction.get('amount', 0)) | |
| # Rule 1: Large withdrawal | |
| avg_transaction = np.mean([float(t.get('amount', 0)) | |
| for t in user_history[-20:] if t.get('type') == 'debit']) | |
| if avg_transaction > 0 and amount > avg_transaction * 3: | |
| score += 25 | |
| reasons.append("Unusually large transaction") | |
| # Rule 2: Rapid consecutive transactions (within 5 minutes) | |
| if len(user_history) > 1: | |
| last_txn_time = datetime.fromisoformat(user_history[0].get('date', '')) | |
| current_time = datetime.fromisoformat(transaction.get('date', '')) | |
| if (current_time - last_txn_time).total_seconds() < 300: | |
| score += 20 | |
| reasons.append("Rapid consecutive transactions") | |
| # Rule 3: Late night transaction (11 PM - 4 AM) | |
| try: | |
| hour = datetime.fromisoformat(transaction.get('date', '')).hour | |
| if hour >= 23 or hour < 4: | |
| score += 15 | |
| reasons.append("Unusual time of transaction") | |
| except: | |
| pass | |
| # Rule 4: Weekend large transaction | |
| try: | |
| day = datetime.fromisoformat(transaction.get('date', '')).weekday() | |
| if day >= 5 and amount > avg_transaction * 2: # Saturday/Sunday | |
| score += 10 | |
| reasons.append("Weekend large transaction") | |
| except: | |
| pass | |
| # Rule 5: Debit after multiple recent debits | |
| debit_count = sum(1 for t in user_history[-5:] if t.get('type') == 'debit') | |
| if debit_count >= 4: | |
| score += 15 | |
| reasons.append("Multiple recent debits") | |
| # Cap score at 100 | |
| score = min(score, 100) | |
| except Exception as e: | |
| print(f"Error calculating fraud score: {e}") | |
| return score, reasons | |
| def check_fraud_alerts(username, users_data): | |
| """ | |
| Check for fraud alerts for a user | |
| Returns: List of fraud alerts | |
| """ | |
| user_data = users_data.get(username, {}) | |
| transactions = user_data.get('transactions', []) | |
| if not transactions: | |
| return [] | |
| detector = FraudDetectionEngine() | |
| alerts = [] | |
| try: | |
| # Analyze recent transactions (last 10) | |
| recent_txns = transactions[:10] | |
| anomalies, scores = detector.detect_anomalies(recent_txns) | |
| # Create alerts for anomalies | |
| for idx in anomalies: | |
| if idx < len(recent_txns): | |
| txn = recent_txns[idx] | |
| fraud_score, reasons = detector.calculate_fraud_score(txn, recent_txns) | |
| if fraud_score > 30: # Alert threshold | |
| alert = { | |
| "id": str(uuid.uuid4()), | |
| "transaction_id": txn.get('id'), | |
| "amount": txn.get('amount'), | |
| "fraud_score": fraud_score, | |
| "reasons": reasons, | |
| "timestamp": datetime.now().isoformat(), | |
| "status": "active" | |
| } | |
| alerts.append(alert) | |
| except Exception as e: | |
| print(f"Error checking fraud alerts: {e}") | |
| return alerts | |
| def get_fraud_alerts_summary(username, users_data): | |
| """Get summary of fraud alerts for a user""" | |
| alerts = check_fraud_alerts(username, users_data) | |
| high_risk = sum(1 for a in alerts if a.get('fraud_score', 0) > 70) | |
| medium_risk = sum(1 for a in alerts if 30 < a.get('fraud_score', 0) <= 70) | |
| return { | |
| "total_alerts": len(alerts), | |
| "high_risk": high_risk, | |
| "medium_risk": medium_risk, | |
| "alerts": alerts[:5] # Return latest 5 | |
| } | |
| def generate_fraud_report(username, users_data, days=30): | |
| """Generate a comprehensive fraud analysis report""" | |
| user_data = users_data.get(username, {}) | |
| transactions = user_data.get('transactions', []) | |
| if not transactions: | |
| return None | |
| # Filter transactions from last N days | |
| cutoff_date = datetime.now() - timedelta(days=days) | |
| recent_txns = [t for t in transactions | |
| if datetime.fromisoformat(t.get('date', '')) > cutoff_date] | |
| detector = FraudDetectionEngine() | |
| # Calculate statistics | |
| total_transactions = len(recent_txns) | |
| total_debit = sum(float(t.get('amount', 0)) for t in recent_txns if t.get('type') == 'debit') | |
| avg_transaction = total_debit / len([t for t in recent_txns if t.get('type') == 'debit']) if any(t.get('type') == 'debit' for t in recent_txns) else 0 | |
| # Run anomaly detection | |
| anomalies, _ = detector.detect_anomalies(recent_txns) | |
| report = { | |
| "period_days": days, | |
| "total_transactions": total_transactions, | |
| "total_debit_amount": total_debit, | |
| "average_transaction": round(avg_transaction, 2), | |
| "anomalies_detected": len(anomalies), | |
| "risk_level": "HIGH" if len(anomalies) > total_transactions * 0.15 else "MEDIUM" if len(anomalies) > total_transactions * 0.05 else "LOW", | |
| "alerts": check_fraud_alerts(username, users_data), | |
| "recommendations": generate_fraud_recommendations(username, users_data) | |
| } | |
| return report | |
| def generate_fraud_recommendations(username, users_data): | |
| """Generate recommendations based on fraud analysis""" | |
| alerts = check_fraud_alerts(username, users_data) | |
| recommendations = [] | |
| if not alerts: | |
| recommendations.append("✅ No suspicious activities detected. Your account is secure.") | |
| else: | |
| high_risk_count = sum(1 for a in alerts if a.get('fraud_score', 0) > 70) | |
| if high_risk_count > 0: | |
| recommendations.append(f"⚠️ {high_risk_count} high-risk transactions detected. Please verify them immediately.") | |
| recommendations.append("💡 Enable transaction alerts for amounts above ₹5,000") | |
| recommendations.append("🔐 Review and update your password regularly") | |
| recommendations.append("📱 Use 2FA for additional security") | |
| return recommendations | |