File size: 2,104 Bytes
8bab071 cf86c3a dd8c606 c6d403e 8b0c3b4 dd8c606 8b0c3b4 dd8c606 cf86c3a 8bab071 c41e114 8bab071 235d85c 8bab071 235d85c 8bab071 c41e114 8bab071 8b0c3b4 235d85c 8bab071 8b0c3b4 8bab071 c6d403e 8bab071 5d337dd 8bab071 5d337dd 8bab071 5d337dd 8bab071 dd8c606 8bab071 |
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 |
# app.py
import gradio as gr
import pandas as pd
import numpy as np
import pickle
from datetime import datetime
# Load model
with open("anomaly_detector_rf_model.pkl", "rb") as f:
model = pickle.load(f)
def detect_anomalies(df):
df["log_amount"] = np.log1p(df["amount"])
df["amount_zscore"] = (df["amount"] - df["amount"].mean()) / df["amount"].std()
df["transaction_date"] = pd.to_datetime(df["transaction_date"])
df["day_of_week"] = df["transaction_date"].dt.dayofweek
df["hour"] = df["transaction_date"].dt.hour
df["is_weekend"] = df["day_of_week"].isin([5, 6]).astype(int)
features = ["amount", "log_amount", "amount_zscore", "day_of_week", "hour", "is_weekend"]
df["is_anomalous"] = model.predict(df[features])
anomalies = df[df["is_anomalous"] == 1][["transaction_id", "amount", "merchant", "location", "transaction_date"]]
# Save anomalies to CSV
csv_path = "/tmp/anomalies.csv"
anomalies.to_csv(csv_path, index=False)
return anomalies, csv_path
# Summary function
def generate_summary(df):
total = len(df)
anomalous = df["is_anomalous"].sum()
return f"Total Transactions: {total} | 🚨 Anomalies: {anomalous} | ⚠️ Rate: {anomalous/total:.2%}"
# Gradio UI
with gr.Blocks() as app:
gr.Markdown("## 💸 Financial Transaction Anomaly Detector")
gr.Markdown("Upload your CSV with columns: `transaction_id`, `amount`, `merchant`, `location`, `transaction_date`")
with gr.Row():
file_input = gr.File(file_types=[".csv"], label="📁 Upload Transaction File")
download_button = gr.File(label="⬇️ Download Anomalies CSV")
summary_text = gr.Textbox(label="📊 Summary", interactive=False)
output_table = gr.Dataframe(label="🚨 Detected Anomalies")
def app_interface(file):
df = pd.read_csv(file.name)
anomalies, csv_path = detect_anomalies(df)
summary = generate_summary(df)
return summary, anomalies, csv_path
file_input.change(fn=app_interface, inputs=[file_input], outputs=[summary_text, output_table, download_button])
app.launch()
|