File size: 2,756 Bytes
cf86c3a dd8c606 c6d403e 8b0c3b4 c6d403e 8b0c3b4 dd8c606 8b0c3b4 dd8c606 8b0c3b4 cf86c3a 8b0c3b4 c6d403e 235d85c 8b0c3b4 235d85c 8b0c3b4 c6d403e 235d85c 8b0c3b4 c6d403e 235d85c 8b0c3b4 235d85c 8b0c3b4 3ec40df 8b0c3b4 3ec40df c6d403e 8b0c3b4 c6d403e 8b0c3b4 c6d403e 3ec40df 8b0c3b4 5d337dd 8b0c3b4 5d337dd 3ec40df 8b0c3b4 c6d403e 5d337dd 8b0c3b4 dd8c606 8b0c3b4 |
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 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 |
import gradio as gr
import pandas as pd
import numpy as np
import pickle
import tempfile
import os
from datetime import datetime
# Load model
with open("anomaly_detector_rf_model.pkl", "rb") as f:
model = pickle.load(f)
# Expected features for prediction
expected_features = [
"amount", "log_amount", "amount_zscore",
"day_of_week", "hour", "is_weekend",
]
# Anomaly Detection Logic
def detect_anomalies(df):
# Feature Engineering
df['log_amount'] = np.log1p(df['amount'])
df['amount_zscore'] = (df['amount'] - df['amount'].mean()) / df['amount'].std()
if 'timestamp' in df.columns:
df['timestamp'] = pd.to_datetime(df['timestamp'])
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['hour'] = df['timestamp'].dt.hour
df['is_weekend'] = df['day_of_week'].apply(lambda x: 1 if x >= 5 else 0)
else:
# If timestamp missing, fill with default values
df['day_of_week'] = 0
df['hour'] = 0
df['is_weekend'] = 0
# Ensure all expected features exist
for col in expected_features:
if col not in df.columns:
df[col] = 0
# Predict
df['is_anomalous'] = model.predict(df[expected_features])
# Extract anomaly rows
anomalies = df[df['is_anomalous'] == 1]
# Columns to return
display_cols = ['amount', 'transaction_id', 'merchant', 'location']
# CSV path
temp_dir = tempfile.mkdtemp()
csv_path = os.path.join(temp_dir, "anomalies.csv")
anomalies.to_csv(csv_path, index=False)
return anomalies[display_cols], csv_path
# Gradio interface function
def app_interface(file):
try:
df = pd.read_csv(file.name)
anomalies_df, csv_path = detect_anomalies(df)
summary = {
"Total Transactions": len(df),
"Anomalies Detected": len(anomalies_df),
"Anomaly %": round(100 * len(anomalies_df) / len(df), 2)
}
return anomalies_df, 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")
gr.Markdown("Upload your transaction CSV file to detect possible financial abuse in the elderly.")
with gr.Row():
file_input = gr.File(label="📤 Upload Transaction CSV", file_types=[".csv"])
download_button = gr.File(label="📥 Download Anomalies (CSV)")
with gr.Row():
output_table = gr.Dataframe(label="🚨 Detected Anomalies")
summary_box = gr.JSON(label="📊 Summary")
file_input.change(
fn=app_interface,
inputs=file_input,
outputs=[output_table, download_button, summary_box]
)
demo.launch()
|