Real-Time AI Fraud Detection for Fintech

import pandas as pd
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler
import psycopg2
import joblib
import os

# --- Configuration ---
DB_HOST = os.environ.get('DB_HOST', 'localhost')
DB_NAME = os.environ.get('DB_NAME', 'fraud_db')
DB_USER = os.environ.get('DB_USER', 'fraud_user')
DB_PASSWORD = os.environ.get('DB_PASSWORD', 'password')

MODEL_PATH = 'isolation_forest_model.joblib'
SCALER_PATH = 'standard_scaler.joblib'

# --- Database Connection ---
def get_db_connection():
    try:
        conn = psycopg2.connect(host=DB_HOST, database=DB_NAME, user=DB_USER, password=DB_PASSWORD)
        return conn
    except psycopg2.Error as e:
        print(f"Error connecting to PostgreSQL: {e}")
        return None

# --- Data Loading & Feature Engineering ---
def load_transactions_for_scoring(conn):
    query = """
    SELECT 
        transaction_id, user_id, amount, merchant_id, EXTRACT(EPOCH FROM (NOW() - timestamp)) AS time_since_transaction
    FROM transactions 
    WHERE processed_for_scoring IS FALSE; 
    """
    try:
        df = pd.read_sql(query, conn)
        # Basic feature engineering for illustration
        # In a real scenario, this would be more complex and aligned with training features
        if not df.empty:
            # Example: Calculate transaction frequency for the user (requires more complex query or pre-computation)
            # For simplicity, we'll use basic features here.
            # Ensure these features MATCH the ones used during training!
            pass # Placeholder for more complex feature engineering
        return df
    except Exception as e:
        print(f"Error loading transactions: {e}")
        return pd.DataFrame()

# --- Model Loading & Scoring ---
def load_model_and_scaler():
    try:
        model = joblib.load(MODEL_PATH)
        scaler = joblib.load(SCALER_PATH)
        return model, scaler
    except FileNotFoundError:
        print(f"Error: Model or scaler file not found. Please train the model first.")
        return None, None
    except Exception as e:
        print(f"Error loading model/scaler: {e}")
        return None, None

def predict_anomalies(df, scaler, model):
    if df.empty:
        return pd.DataFrame()
        
    # IMPORTANT: Ensure the features used here EXACTLY match the training features and order.
    # This is a simplified example. You'd likely need to join/aggregate more data for real features.
    # For this example, let's assume 'amount' and 'time_since_transaction' were the ONLY features.
    # In reality, you'd need user_transaction_count, avg_user_amount, etc.
    
    # Placeholder for actual feature alignment with training data
    # For demonstration, we'll use dummy features if they don't exist, but this WILL NOT WORK without proper alignment.
    required_features = ['amount', 'time_since_transaction'] # Example features
    for feature in required_features:
        if feature not in df.columns:
            print(f"Warning: Feature '{feature}' not found. Using dummy data. This WILL affect accuracy.")
            df[feature] = 0 # Placeholder

    features_for_prediction = df[required_features]
    
    try:
        scaled_features = scaler.transform(features_for_prediction)
        anomaly_scores = model.decision_function(scaled_features)
        df['anomaly_score'] = anomaly_scores
        # Predict labels: -1 for outliers/anomalies, 1 for inliers
        df['is_anomaly'] = model.predict(scaled_features)
        return df
    except Exception as e:
        print(f"Error during prediction: {e}")
        return pd.DataFrame()

# --- Updating Database ---
def update_scoring_status(conn, transaction_ids):
    if not transaction_ids:
        return
    query = "UPDATE transactions SET processed_for_scoring = TRUE WHERE transaction_id IN (%s);"
    try:
        with conn.cursor() as cursor:
            # Use execute_many for efficiency if available, or loop
            # For simplicity, joining IDs into a string for a single query (can be inefficient for many IDs)
            ids_tuple = tuple(transaction_ids)
            cursor.execute(query.replace('%s', '%s' * len(ids_tuple)), ids_tuple)
            conn.commit()
            print(f"Marked {len(transaction_ids)} transactions as processed.")
    except Exception as e:
        print(f"Error updating scoring status: {e}")
        conn.rollback()

# --- Main Execution Flow ---
def main():
    conn = get_db_connection()
    if not conn:
        return

    transactions_df = load_transactions_for_scoring(conn)
    if transactions_df.empty:
        print("No new transactions to score.")
        conn.close()
        return

    model, scaler = load_model_and_scaler()
    if not model or not scaler:
        conn.close()
        return

    scored_transactions = predict_anomalies(transactions_df, scaler, model)

    if not scored_transactions.empty:
        # Store results back to DB or a separate table
        # For simplicity, we'll just print and mark as processed.
        print("--- Scored Transactions ---")
        print(scored_transactions[['transaction_id', 'amount', 'anomaly_score', 'is_anomaly']].head())
        
        # Update status in DB
        processed_ids = scored_transactions['transaction_id'].tolist()
        update_scoring_status(conn, processed_ids)
    else:
        print("No anomalies detected or an error occurred during scoring.")

    conn.close()

if __name__ == "__main__":
    # IMPORTANT: Ensure you have a trained model (isolation_forest_model.joblib)
    # and a fitted scaler (standard_scaler.joblib) saved in the same directory.
    # You also need a PostgreSQL database running with the 'transactions' table.
    # Example table schema:
    # CREATE TABLE transactions (
    #     transaction_id UUID PRIMARY KEY,
    #     user_id VARCHAR(255),
    #     amount DECIMAL(10, 2),
    #     merchant_id VARCHAR(255),
    #     timestamp TIMESTAMP WITH TIME ZONE,
    #     processed_for_scoring BOOLEAN DEFAULT FALSE
    # );
    
    # Ensure the correct features are used during training and scoring.
    # This script assumes 'amount' and 'time_since_transaction' are the features.
    # You will need to adapt this to your actual feature set.
    main()