kafka_consumer.py

import json
import pandas as pd
from kafka import KafkaConsumer
import joblib
from scripts.prepare_labeled_data import parse_meta
from scripts.neo4j_updater import push_anomalies_to_neo4j

# Load trained model and expected feature columns
model, feature_cols = joblib.load("trained_fraud_model.pkl")

# Set of all columns that the model expects
required_columns = feature_cols.copy()

consumer = KafkaConsumer(
    'transactions',
    bootstrap_servers='localhost:9092',
    auto_offset_reset='earliest',
    enable_auto_commit=True,
    group_id='fraud-detector',
    value_deserializer=lambda m: json.loads(m.decode('utf-8'))
)

print("✅ Kafka Consumer listening on topic 'transactions'...")

for message in consumer:
    record = message.value
    print(f"\n📥 Received transaction: {record.get('user_uid')} - {record.get('transaction_amount')}")

    # Convert single transaction to DataFrame
    df = pd.DataFrame([record])

    # Parse meta_data JSON and flatten it
    if "meta_data" in df.columns:
        meta_df = df["meta_data"].apply(parse_meta).apply(pd.Series)
        df = pd.concat([df.drop(columns=["meta_data"]), meta_df], axis=1)

    # Extract time-based features
    df["transaction_dttm"] = pd.to_datetime(df["transaction_dttm"], errors="coerce")
    df["hour"] = df["transaction_dttm"].dt.hour
    df["dayofweek"] = df["transaction_dttm"].dt.dayofweek

    # One-hot encode relevant fields
    categorical_cols = ["transaction_type", "transaction_currency", "terminal_uid"]
    df = pd.get_dummies(df, columns=categorical_cols, prefix=categorical_cols)

    # Add missing one-hot columns (not seen during training)
    missing_cols = [col for col in required_columns if col not in df.columns]
    if missing_cols:
        print(f"[WARN] Adding missing columns: {missing_cols}")
        filler_df = pd.DataFrame(0, index=df.index, columns=missing_cols)
        df = pd.concat([df, filler_df], axis=1)

    # Ensure column order and defragment
    df = df.reindex(columns=required_columns)
    df = df.copy()

    # Run prediction
    X = df[feature_cols]
    df["predicted_label"] = model.predict(X)

    if df["predicted_label"].iloc[0] == 1:
        print("🚨 Fraud detected!")
        print(df[["first_name", "last_name", "transaction_amount", "transaction_dttm"]])
        push_anomalies_to_neo4j(df)
    else:
        print("✅ Transaction is normal.")