File size: 2,167 Bytes
cf86c3a dd8c606 c6d403e 3ec40df c6d403e dd8c606 c6d403e dd8c606 cf86c3a c6d403e 235d85c c6d403e 235d85c c6d403e 235d85c c6d403e 235d85c c6d403e 235d85c c6d403e 3ec40df c6d403e 3ec40df c6d403e 5d337dd c6d403e 5d337dd 3ec40df c6d403e 5d337dd c6d403e dd8c606 c6d403e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
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()
|