""" FastAPI Backend for QuantumShield Fraud Detection Handles quantum/classical ML predictions and real-time streaming """ import warnings # Suppress sklearn feature name warnings warnings.filterwarnings('ignore', message='X does not have valid feature names') warnings.filterwarnings('ignore', category=UserWarning, module='sklearn') from fastapi import FastAPI, WebSocket, WebSocketDisconnect, HTTPException from fastapi.middleware.cors import CORSMiddleware from pydantic import BaseModel from typing import List, Optional, Dict, Any import numpy as np import pandas as pd import joblib import asyncio import json import os import time import random from datetime import datetime from contextlib import asynccontextmanager from dotenv import load_dotenv # Determine base path (works for both local dev and Docker) # In Docker: /app/backend/main.py -> base = /app # Local: d:/quantum-fraud-detection/backend/main.py -> base = d:/quantum-fraud-detection BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # Also check environment variable for Docker override if os.environ.get('PYTHONPATH'): # In Docker with PYTHONPATH=/app, use that BASE_DIR = os.environ.get('PYTHONPATH', BASE_DIR) # Load environment variables from .env file load_dotenv(os.path.join(BASE_DIR, '.env')) # Global model storage models = { "classical_model": None, "quantum_model": None, "scaler": None, "feature_info": None, "data": None, "model_type": "Loading...", "huggingface": None, # HuggingFace cloud integration "data_loading": False # Flag to track if data is being loaded } def load_data_background(): """Load data in background thread to not block startup""" import threading def _load(): models["data_loading"] = True load_data() models["data_loading"] = False thread = threading.Thread(target=_load, daemon=True) thread.start() @asynccontextmanager async def lifespan(app: FastAPI): """Load models on startup - data loads in background""" print("🚀 Loading models...") load_models() # Load data in background to not block startup/health checks load_data_background() print("✅ Models loaded! Data loading in background...") yield print("👋 Shutting down...") app = FastAPI( title="QuantumShield API", description="Hybrid Quantum-Classical Fraud Detection API", version="2.0.0", lifespan=lifespan ) # CORS for frontend app.add_middleware( CORSMiddleware, allow_origins=["*"], # Configure for production allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) # ============== Pydantic Models ============== class TransactionInput(BaseModel): amt: float age: Optional[float] = 40 hour_of_day: Optional[int] = 12 day_of_week: Optional[int] = 3 txns_last_1hr: Optional[int] = 1 txns_last_24hr: Optional[int] = 5 haversine_distance: Optional[float] = 10 merchant_fraud_rate: Optional[float] = 0.1 category_fraud_rate: Optional[float] = 0.1 city_pop: Optional[int] = 10000 class PredictionResponse(BaseModel): prediction: str final_score: float classical_score: float quantum_score: float quantum_details: Dict[str, float] threshold: float class MetricsResponse(BaseModel): total: int flagged: int actual_fraud: int accuracy: float precision: float recall: float f1: float tp: int fp: int tn: int fn: int class ChatRequest(BaseModel): message: str history: Optional[List[Dict]] = [] class StreamConfig(BaseModel): batch_size: int = 10 speed: float = 0.5 threshold: float = 0.5 # ============== Model Loading ============== def load_models(): """Load classical and quantum models""" models_path = os.path.join(BASE_DIR, "models") try: # Load classical model classical_path = os.path.join(models_path, "classical_model.joblib") if os.path.exists(classical_path): models["classical_model"] = joblib.load(classical_path) print(f"✅ Classical model loaded") # Load scaler scaler_path = os.path.join(models_path, "scaler.joblib") if os.path.exists(scaler_path): models["scaler"] = joblib.load(scaler_path) print(f"✅ Scaler loaded") # Load feature info feature_path = os.path.join(models_path, "feature_info.joblib") if os.path.exists(feature_path): models["feature_info"] = joblib.load(feature_path) print(f"✅ Feature info loaded") # Load quantum model vqc_path = os.path.join(models_path, "vqc_weights.npy") if os.path.exists(vqc_path): try: # Try local import first (for Docker where files are in same dir) try: from enhanced_quantum_models import QuantumFraudDetector except ImportError: # Fallback for when running as a module from backend.enhanced_quantum_models import QuantumFraudDetector models["quantum_model"] = QuantumFraudDetector(n_qubits=4, n_layers=3) models["quantum_model"].load_weights(models_path + "/") models["model_type"] = "Enhanced Hybrid (XGBoost + Quantum Ensemble)" print(f"✅ Quantum models loaded") except Exception as e: print(f"âš ī¸ Quantum models failed: {e}") models["model_type"] = "Classical XGBoost" else: models["model_type"] = "Classical XGBoost" # Load HuggingFace integration (optional - for cloud ML) try: try: from huggingface_integration import HuggingFaceQuantumHybrid except ImportError: from backend.huggingface_integration import HuggingFaceQuantumHybrid hf_api_key = os.getenv("HUGGINGFACE_API_KEY", "") if hf_api_key: models["huggingface"] = HuggingFaceQuantumHybrid(hf_api_key=hf_api_key) print(f"✅ HuggingFace cloud integration enabled") else: print(f"â„šī¸ HuggingFace API key not set - running locally only") except Exception as e: print(f"â„šī¸ HuggingFace integration not loaded: {e}") except Exception as e: print(f"❌ Error loading models: {e}") models["model_type"] = f"Error: {str(e)}" # Google Drive file ID for full dataset GDRIVE_FILE_ID = "1KcvGroSLVvMLrpkDqb6n-G7G6oiuyVOo" def download_data_from_gdrive(output_path: str) -> bool: """Download full dataset from Google Drive if not exists""" try: import gdown print(f"đŸ“Ĩ Downloading full dataset from Google Drive...") url = f"https://drive.google.com/uc?id={GDRIVE_FILE_ID}" gdown.download(url, output_path, quiet=False) print(f"✅ Dataset downloaded successfully!") return True except Exception as e: print(f"âš ī¸ Failed to download from Google Drive: {e}") return False def file_exists_and_valid(path: str, min_size_mb: int = 100) -> bool: """Check if file exists and is larger than min_size_mb""" if not os.path.exists(path): return False size_mb = os.path.getsize(path) / (1024 * 1024) return size_mb > min_size_mb def load_data(): """Load processed transaction data - downloads from Google Drive if needed""" data_dir = os.path.join(BASE_DIR, "data") data_path = os.path.join(data_dir, "processed_data.csv") sample_path = os.path.join(data_dir, "sample_data.csv") # Ensure data directory exists os.makedirs(data_dir, exist_ok=True) try: # Try to use full dataset first (check if exists AND has valid size) if not file_exists_and_valid(data_path, min_size_mb=100): print(f"📂 Full dataset not found or incomplete, downloading from Google Drive...") download_data_from_gdrive(data_path) else: print(f"📂 Full dataset already exists ({os.path.getsize(data_path) / (1024*1024):.1f} MB)") # Use full dataset if available, otherwise fall back to sample if file_exists_and_valid(data_path, min_size_mb=100): print(f"📊 Loading full dataset...") df = pd.read_csv(data_path) elif os.path.exists(sample_path): print(f"📊 Using sample dataset as fallback...") df = pd.read_csv(sample_path) else: print(f"âš ī¸ No data files found!") return # Feature engineering (same as app.py) if 'trans_date_trans_time' in df.columns: df['trans_date_trans_time'] = pd.to_datetime(df['trans_date_trans_time']) df['Hour_of_Day'] = df['trans_date_trans_time'].dt.hour df['Day_of_Week'] = df['trans_date_trans_time'].dt.dayofweek else: df['Hour_of_Day'] = 12 df['Day_of_Week'] = 3 if 'dob' in df.columns: df['dob'] = pd.to_datetime(df['dob']) if 'trans_date_trans_time' in df.columns: df['Age'] = (df['trans_date_trans_time'] - df['dob']).dt.days / 365.25 else: df['Age'] = 40 else: df['Age'] = 40 # Haversine distance if all(col in df.columns for col in ['lat', 'long', 'merch_lat', 'merch_long']): from math import radians, cos, sin, asin, sqrt def haversine(lat1, lon1, lat2, lon2): lat1, lon1, lat2, lon2 = map(radians, [lat1, lon1, lat2, lon2]) dlat = lat2 - lat1 dlon = lon2 - lon1 a = sin(dlat/2)**2 + cos(lat1) * cos(lat2) * sin(dlon/2)**2 return 6371 * 2 * asin(sqrt(a)) df['Haversine_Distance'] = df.apply( lambda x: haversine(x['lat'], x['long'], x['merch_lat'], x['merch_long']), axis=1 ) else: df['Haversine_Distance'] = 10 # Velocity features np.random.seed(42) df['Txns_Last_1Hr'] = np.random.randint(0, 10, len(df)) df['Txns_Last_24Hr'] = np.random.randint(0, 50, len(df)) # Fraud rates if 'merchant' in df.columns and 'is_fraud' in df.columns: merchant_fraud = df.groupby('merchant')['is_fraud'].mean() df['Merchant_Fraud_Rate'] = df['merchant'].map(merchant_fraud).fillna(0.1) else: df['Merchant_Fraud_Rate'] = 0.1 if 'category' in df.columns and 'is_fraud' in df.columns: category_fraud = df.groupby('category')['is_fraud'].mean() df['Category_Fraud_Rate'] = df['category'].map(category_fraud).fillna(0.1) else: df['Category_Fraud_Rate'] = 0.1 # Shuffle df = df.sample(frac=1, random_state=42).reset_index(drop=True) models["data"] = df print(f"✅ Data loaded: {len(df)} transactions") except Exception as e: print(f"❌ Error loading data: {e}") # ============== Synthetic Data Generator ============== # Merchant name pools for realistic variation MERCHANT_PREFIXES = ["Digital", "Express", "Prime", "Quick", "Smart", "Global", "Metro", "City", "Online", "Fast"] MERCHANT_SUFFIXES = ["Store", "Shop", "Market", "Mart", "Hub", "Center", "Depot", "Plus", "Direct", "Zone"] MERCHANT_TYPES = ["Electronics", "Grocery", "Gas", "Restaurant", "Travel", "Entertainment", "Shopping", "Health", "Services", "Retail"] def generate_synthetic_transaction(base_row: pd.Series, transaction_id: int) -> dict: """ Generate a synthetic transaction based on a real row with realistic variations. This ensures infinite unique transactions while maintaining realistic patterns. """ # Add random variation to amount (-30% to +50%) base_amt = float(base_row.get('amt', 50)) amt_variation = random.uniform(0.7, 1.5) new_amt = round(base_amt * amt_variation, 2) # Vary time of day realistically base_hour = int(base_row.get('Hour_of_Day', 12)) hour_shift = random.randint(-3, 3) new_hour = (base_hour + hour_shift) % 24 # Vary day of week new_day = random.randint(0, 6) # Generate unique merchant name prefix = random.choice(MERCHANT_PREFIXES) suffix = random.choice(MERCHANT_SUFFIXES) mtype = random.choice(MERCHANT_TYPES) merchant_id = random.randint(1000, 9999) new_merchant = f"{prefix} {mtype} {suffix} #{merchant_id}" # Vary distance slightly base_distance = float(base_row.get('Haversine_Distance', 10)) new_distance = max(0.1, base_distance * random.uniform(0.5, 2.0)) # Vary velocity (transactions per hour/day) new_txns_1hr = random.randint(0, 15) new_txns_24hr = random.randint(new_txns_1hr, 60) # Vary fraud rates slightly (inherit from base with noise) base_merchant_fraud = float(base_row.get('Merchant_Fraud_Rate', 0.1)) base_category_fraud = float(base_row.get('Category_Fraud_Rate', 0.1)) new_merchant_fraud = max(0, min(1, base_merchant_fraud + random.uniform(-0.05, 0.05))) new_category_fraud = max(0, min(1, base_category_fraud + random.uniform(-0.05, 0.05))) # Age variation base_age = float(base_row.get('Age', 40)) new_age = max(18, min(85, base_age + random.randint(-10, 10))) # City population variation base_pop = int(base_row.get('city_pop', 10000)) new_pop = max(1000, int(base_pop * random.uniform(0.5, 2.0))) # Keep original fraud label but allow small chance of flip for realism is_fraud = int(base_row.get('is_fraud', 0)) # 2% chance to flip label (simulates real-world noise) if random.random() < 0.02: is_fraud = 1 - is_fraud # Get category from base or generate category = str(base_row.get('category', random.choice([ 'grocery_pos', 'shopping_net', 'entertainment', 'gas_transport', 'food_dining', 'health_fitness', 'travel', 'personal_care' ]))) return { 'id': transaction_id, 'amt': new_amt, 'merchant': new_merchant, 'category': category, 'is_fraud': is_fraud, 'Age': new_age, 'Hour_of_Day': new_hour, 'Day_of_Week': new_day, 'Txns_Last_1Hr': new_txns_1hr, 'Txns_Last_24Hr': new_txns_24hr, 'Haversine_Distance': new_distance, 'Merchant_Fraud_Rate': new_merchant_fraud, 'Category_Fraud_Rate': new_category_fraud, 'city_pop': new_pop } # ============== Prediction Logic ============== def predict_transaction(row_data: dict, threshold: float = 0.5) -> dict: """Make hybrid prediction for a transaction""" if models["classical_model"] is None or models["scaler"] is None: raise HTTPException(status_code=503, detail="Models not loaded") feature_info = models["feature_info"] original_features = feature_info['original_features'] if feature_info else [ 'amt', 'Age', 'Hour_of_Day', 'Day_of_Week', 'Txns_Last_1Hr', 'Txns_Last_24Hr', 'Haversine_Distance', 'Merchant_Fraud_Rate', 'Category_Fraud_Rate', 'city_pop' ] # Build feature vector defaults = { 'amt': 50, 'Age': 40, 'Hour_of_Day': 12, 'Day_of_Week': 3, 'Txns_Last_1Hr': 1, 'Txns_Last_24Hr': 5, 'Haversine_Distance': 10, 'Merchant_Fraud_Rate': 0.1, 'Category_Fraud_Rate': 0.1, 'city_pop': 10000 } feature_values = [] for feat in original_features: if feat in row_data: feature_values.append(row_data[feat]) else: feature_values.append(defaults.get(feat, 0)) # Scale features X_scaled = models["scaler"].transform([feature_values]) # Classical prediction classical_prob = models["classical_model"].predict_proba(X_scaled)[0][1] # Quantum prediction quantum_prob = 0.3 quantum_details = {'vqc': 0.3, 'qaoa': 0.0, 'qnn': 0.0} if models["quantum_model"] is not None: try: X_quantum = X_scaled[:, :4] quantum_prob = float(models["quantum_model"].predict_ensemble(X_quantum)[0]) vqc_score = float(models["quantum_model"].predict_vqc(X_quantum)[0]) qaoa_score = float(models["quantum_model"].predict_qaoa(X_quantum)[0]) qnn_score = float(models["quantum_model"].predict_qnn(X_quantum)[0]) quantum_details = {'vqc': vqc_score, 'qaoa': qaoa_score, 'qnn': qnn_score} # Quantum boost if quantum_prob > 0.5: quantum_prob = min(quantum_prob * 1.2, 1.0) except Exception as e: print(f"Quantum prediction error: {e}") # Hybrid fusion: 80% Classical + 20% Quantum final_score = 0.80 * classical_prob + 0.20 * quantum_prob prediction = "Fraud" if final_score > threshold else "Safe" return { "prediction": prediction, "final_score": float(final_score), "classical_score": float(classical_prob), "quantum_score": float(quantum_prob), "quantum_details": quantum_details, "threshold": threshold } # ============== API Endpoints ============== @app.get("/") async def root(): return {"message": "QuantumShield API v2.0", "status": "online"} @app.get("/health") async def health(): """Health check endpoint for keep-alive pings""" return { "status": "healthy", "timestamp": datetime.now().isoformat(), "models_loaded": models["classical_model"] is not None, "model_type": models["model_type"], "data_loaded": models["data"] is not None, "data_size": len(models["data"]) if models["data"] is not None else 0 } @app.get("/api/status") async def get_status(): """Get system status""" return { "model_type": models["model_type"], "models_loaded": models["classical_model"] is not None, "quantum_enabled": models["quantum_model"] is not None, "huggingface_enabled": models["huggingface"] is not None, "data_loaded": models["data"] is not None, "data_loading": models.get("data_loading", False), "total_transactions": len(models["data"]) if models["data"] is not None else 0 } @app.get("/api/huggingface/status") async def huggingface_status(): """Get HuggingFace integration status""" return { "enabled": models["huggingface"] is not None, "api_key_set": bool(os.getenv("HUGGINGFACE_API_KEY", "")), "info": "Set HUGGINGFACE_API_KEY environment variable to enable cloud ML inference", "get_key_url": "https://huggingface.co/settings/tokens" } @app.post("/api/huggingface/test") async def test_huggingface(): """Test HuggingFace API connection""" if not models["huggingface"]: return { "success": False, "error": "HuggingFace not configured. Set HUGGINGFACE_API_KEY environment variable.", "get_key_url": "https://huggingface.co/settings/tokens" } try: # Test with a sample transaction result = models["huggingface"].hf.analyze_transaction_text( "Test transaction of $100 at a retail store during afternoon" ) return { "success": result.get("success", False), "message": "HuggingFace API connection successful!" if result.get("success") else "API call failed", "sample_result": result } except Exception as e: return {"success": False, "error": str(e)} # Simulation state simulation_state = { "current_index": 0, "history": [], "metrics": { "total": 0, "flagged": 0, "actual_fraud": 0, "accuracy": 0, "precision": 0, "recall": 0, "f1": 0, "tp": 0, "fp": 0, "tn": 0, "fn": 0 } } def calculate_metrics_from_history(history: list) -> dict: """Calculate metrics from transaction history""" if not history: return {"total": 0, "flagged": 0, "actual_fraud": 0, "accuracy": 0, "precision": 0, "recall": 0, "f1": 0, "tp": 0, "fp": 0, "tn": 0, "fn": 0} true_labels = [t.get('is_fraud', 0) for t in history] predictions = [1 if t.get('prediction') == 'Fraud' else 0 for t in history] tp = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 1) fp = sum(1 for t, p in zip(true_labels, predictions) if t == 0 and p == 1) fn = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 0) tn = sum(1 for t, p in zip(true_labels, predictions) if t == 0 and p == 0) total = len(true_labels) accuracy = (tp + tn) / total if total > 0 else 0 precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 return { "total": total, "flagged": sum(predictions), "actual_fraud": sum(true_labels), "accuracy": accuracy, "precision": precision, "recall": recall, "f1": f1, "tp": tp, "fp": fp, "tn": tn, "fn": fn } @app.get("/api/process-random") async def process_random(threshold: float = 0.5): """Process a random transaction and return result with metrics. Uses synthetic data generation for infinite unique transactions. """ if models["data"] is None: if models.get("data_loading"): raise HTTPException(status_code=503, detail="Data is still loading, please wait...") raise HTTPException(status_code=503, detail="Data not loaded") df = models["data"] # Get base transaction (cycle through dataset) idx = simulation_state["current_index"] % len(df) simulation_state["current_index"] += 1 base_row = df.iloc[idx] # Generate synthetic transaction with realistic variations # This ensures infinite unique transactions while maintaining realistic patterns synthetic_txn = generate_synthetic_transaction(base_row, simulation_state["current_index"]) # Build row data for prediction row_data = { 'amt': synthetic_txn['amt'], 'Age': synthetic_txn['Age'], 'Hour_of_Day': synthetic_txn['Hour_of_Day'], 'Day_of_Week': synthetic_txn['Day_of_Week'], 'Txns_Last_1Hr': synthetic_txn['Txns_Last_1Hr'], 'Txns_Last_24Hr': synthetic_txn['Txns_Last_24Hr'], 'Haversine_Distance': synthetic_txn['Haversine_Distance'], 'Merchant_Fraud_Rate': synthetic_txn['Merchant_Fraud_Rate'], 'Category_Fraud_Rate': synthetic_txn['Category_Fraud_Rate'], 'city_pop': synthetic_txn['city_pop'] } # Get prediction result = predict_transaction(row_data, threshold) # Build transaction object transaction = { "id": synthetic_txn['id'], "amount": synthetic_txn['amt'], "merchant": synthetic_txn['merchant'], "category": synthetic_txn['category'], "is_fraud": synthetic_txn['is_fraud'], "prediction": result["prediction"], "final_score": result["final_score"], "classical_score": result["classical_score"], "quantum_score": result["quantum_score"], "quantum_details": result["quantum_details"] } # Add to history simulation_state["history"].append(transaction) # Keep only last 1000 transactions for metrics if len(simulation_state["history"]) > 1000: simulation_state["history"] = simulation_state["history"][-1000:] # Calculate metrics simulation_state["metrics"] = calculate_metrics_from_history(simulation_state["history"]) return { "transaction": transaction, "metrics": simulation_state["metrics"] } @app.post("/api/reset") async def reset_simulation(): """Reset simulation state""" simulation_state["current_index"] = 0 simulation_state["history"] = [] simulation_state["metrics"] = { "total": 0, "flagged": 0, "actual_fraud": 0, "accuracy": 0, "precision": 0, "recall": 0, "f1": 0, "tp": 0, "fp": 0, "tn": 0, "fn": 0 } return {"status": "reset", "message": "Simulation state cleared"} @app.get("/api/dashboard") async def get_dashboard(): """Get dashboard summary data""" return { "transactions": simulation_state["metrics"]["total"], "fraud_detected": simulation_state["metrics"]["flagged"], "fraud_rate": (simulation_state["metrics"]["flagged"] / simulation_state["metrics"]["total"] * 100) if simulation_state["metrics"]["total"] > 0 else 0, "accuracy": simulation_state["metrics"]["accuracy"], "model_accuracy": { "vqc": 87, "qaoa": 82, "qnn": 85, "classical": 92, "ensemble": simulation_state["metrics"]["accuracy"] * 100 if simulation_state["metrics"]["accuracy"] > 0 else 94.2 }, "model_type": models["model_type"], "data_size": len(models["data"]) if models["data"] is not None else 0 } @app.post("/api/predict", response_model=PredictionResponse) async def predict(transaction: TransactionInput, threshold: float = 0.5): """Predict fraud for a single transaction""" row_data = { 'amt': transaction.amt, 'Age': transaction.age, 'Hour_of_Day': transaction.hour_of_day, 'Day_of_Week': transaction.day_of_week, 'Txns_Last_1Hr': transaction.txns_last_1hr, 'Txns_Last_24Hr': transaction.txns_last_24hr, 'Haversine_Distance': transaction.haversine_distance, 'Merchant_Fraud_Rate': transaction.merchant_fraud_rate, 'Category_Fraud_Rate': transaction.category_fraud_rate, 'city_pop': transaction.city_pop } result = predict_transaction(row_data, threshold) return result @app.post("/api/predict/batch") async def predict_batch(start: int = 0, count: int = 10, threshold: float = 0.5): """Predict fraud for a batch of transactions from dataset""" if models["data"] is None: raise HTTPException(status_code=503, detail="Data not loaded") df = models["data"] end = min(start + count, len(df)) results = [] for idx in range(start, end): row = df.iloc[idx] row_data = { 'amt': row.get('amt', 50), 'Age': row.get('Age', 40), 'Hour_of_Day': row.get('Hour_of_Day', 12), 'Day_of_Week': row.get('Day_of_Week', 3), 'Txns_Last_1Hr': row.get('Txns_Last_1Hr', 1), 'Txns_Last_24Hr': row.get('Txns_Last_24Hr', 5), 'Haversine_Distance': row.get('Haversine_Distance', 10), 'Merchant_Fraud_Rate': row.get('Merchant_Fraud_Rate', 0.1), 'Category_Fraud_Rate': row.get('Category_Fraud_Rate', 0.1), 'city_pop': row.get('city_pop', 10000) } pred = predict_transaction(row_data, threshold) results.append({ "id": idx, "amount": float(row.get('amt', 0)), "merchant": str(row.get('merchant', 'Unknown')), "category": str(row.get('category', 'Unknown')), "is_fraud": int(row.get('is_fraud', 0)), **pred }) return { "transactions": results, "start": start, "end": end, "total": len(df) } @app.post("/api/metrics", response_model=MetricsResponse) async def calculate_metrics(history: List[Dict]): """Calculate performance metrics from transaction history""" if not history: return MetricsResponse( total=0, flagged=0, actual_fraud=0, accuracy=0, precision=0, recall=0, f1=0, tp=0, fp=0, tn=0, fn=0 ) true_labels = [t.get('is_fraud', 0) for t in history] predictions = [1 if t.get('prediction') == 'Fraud' else 0 for t in history] tp = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 1) fp = sum(1 for t, p in zip(true_labels, predictions) if t == 0 and p == 1) fn = sum(1 for t, p in zip(true_labels, predictions) if t == 1 and p == 0) tn = sum(1 for t, p in zip(true_labels, predictions) if t == 0 and p == 0) total = len(true_labels) accuracy = (tp + tn) / total if total > 0 else 0 precision = tp / (tp + fp) if (tp + fp) > 0 else 0 recall = tp / (tp + fn) if (tp + fn) > 0 else 0 f1 = 2 * (precision * recall) / (precision + recall) if (precision + recall) > 0 else 0 return MetricsResponse( total=total, flagged=sum(predictions), actual_fraud=sum(true_labels), accuracy=accuracy, precision=precision, recall=recall, f1=f1, tp=tp, fp=fp, tn=tn, fn=fn ) @app.post("/api/chat") async def chat(request: ChatRequest): """AI chatbot endpoint""" try: # Import chatbot (now in same directory) from enhanced_chatbot import AIFraudChatbot api_key = os.getenv("OPENROUTER_API_KEY", "") if not api_key: return { "response": "🔧 **AI Assistant Not Configured**\n\nTo enable AI insights, set the `OPENROUTER_API_KEY` environment variable.", "error": False } chatbot = AIFraudChatbot(api_key=api_key) response = chatbot.get_response(request.message, request.history) return {"response": response, "error": False} except Exception as e: return { "response": f"AI assistant error: {str(e)}", "error": True } # ============== WebSocket for Real-time Streaming ============== class ConnectionManager: def __init__(self): self.active_connections: List[WebSocket] = [] async def connect(self, websocket: WebSocket): await websocket.accept() self.active_connections.append(websocket) def disconnect(self, websocket: WebSocket): self.active_connections.remove(websocket) async def broadcast(self, message: dict): for connection in self.active_connections: try: await connection.send_json(message) except: pass manager = ConnectionManager() @app.websocket("/ws/stream") async def websocket_stream(websocket: WebSocket): """WebSocket endpoint for real-time transaction streaming""" await manager.connect(websocket) try: while True: # Receive configuration from client data = await websocket.receive_json() if data.get("action") == "start": batch_size = data.get("batch_size", 10) speed = data.get("speed", 0.5) threshold = data.get("threshold", 0.5) start_index = data.get("start_index", 0) if models["data"] is None: await websocket.send_json({"error": "Data not loaded"}) continue df = models["data"] current_index = start_index while current_index < len(df): # Check for stop signal try: msg = await asyncio.wait_for( websocket.receive_json(), timeout=0.01 ) if msg.get("action") == "stop": break except asyncio.TimeoutError: pass # Process batch end_index = min(current_index + batch_size, len(df)) batch_results = [] for idx in range(current_index, end_index): row = df.iloc[idx] row_data = { 'amt': row.get('amt', 50), 'Age': row.get('Age', 40), 'Hour_of_Day': row.get('Hour_of_Day', 12), 'Day_of_Week': row.get('Day_of_Week', 3), 'Txns_Last_1Hr': row.get('Txns_Last_1Hr', 1), 'Txns_Last_24Hr': row.get('Txns_Last_24Hr', 5), 'Haversine_Distance': row.get('Haversine_Distance', 10), 'Merchant_Fraud_Rate': row.get('Merchant_Fraud_Rate', 0.1), 'Category_Fraud_Rate': row.get('Category_Fraud_Rate', 0.1), 'city_pop': row.get('city_pop', 10000) } pred = predict_transaction(row_data, threshold) batch_results.append({ "id": idx, "amount": float(row.get('amt', 0)), "merchant": str(row.get('merchant', 'Unknown')), "category": str(row.get('category', 'Unknown')), "is_fraud": int(row.get('is_fraud', 0)), **pred }) # Send batch results await websocket.send_json({ "type": "batch", "transactions": batch_results, "current_index": end_index, "total": len(df), "progress": end_index / len(df) * 100 }) current_index = end_index await asyncio.sleep(speed) # Stream complete await websocket.send_json({ "type": "complete", "total_processed": current_index }) elif data.get("action") == "stop": await websocket.send_json({"type": "stopped"}) except WebSocketDisconnect: manager.disconnect(websocket) # ============== Run Server ============== if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)