app.py

import gradio as gr
import pandas as pd
import numpy as np
import joblib
import datetime
import os

# Load trained model
model = joblib.load("anomaly_detector_rf_model.pkl")

def detect_anomalies(df):
    # Feature Engineering (must match training phase)
    df['datetime'] = pd.to_datetime(df['timestamp'])
    df['hour'] = df['datetime'].dt.hour
    df['day_of_week'] = df['datetime'].dt.dayofweek
    df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
    df['log_amount'] = np.log1p(df['amount'])
    df['amount_zscore'] = (df['amount'] - df['amount'].mean()) / df['amount'].std()

    expected_features = ['amount', 'log_amount', 'amount_zscore', 'hour', 'day_of_week', 'is_weekend']

    df['is_anomalous'] = model.predict(df[expected_features])

    anomalies = df[df['is_anomalous'] == 1]
    anomalies_display = anomalies[['transaction_id', 'merchant', 'location', 'amount']].reset_index(drop=True)

    # Save CSV for download
    csv_path = "/tmp/anomalies.csv"
    anomalies_display.to_csv(csv_path, index=False)

    return anomalies_display, csv_path, {
        "Total Transactions": len(df),
        "Anomalies Detected": len(anomalies),
        "Anomaly %": f"{(len(anomalies)/len(df)*100):.2f}%"
    }

def app_interface(file):
    try:
        df = pd.read_csv(file.name)
        anomalies, csv_path, summary = detect_anomalies(df)
        return anomalies, csv_path, summary
    except Exception as e:
        return pd.DataFrame(), None, {"Error": str(e)}

# Gradio UI
with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown("# 🧠 Financial Anomaly Detector\nUpload transaction data to detect anomalies using ML.")

    with gr.Row():
        file_input = gr.File(label="📤 Upload CSV", file_types=[".csv"])
        download_button = gr.File(label="📥 Download Anomalies CSV")

    with gr.Row():
        output_table = gr.Dataframe(label="🚨 Detected Anomalies", wrap=True, height=300)
        summary_box = gr.JSON(label="📊 Summary")

    file_input.change(fn=app_interface, inputs=file_input, outputs=[output_table, download_button, summary_box])

# If running locally
if __name__ == "__main__":
    demo.launch()