import streamlit as st
import pandas as pd
import numpy as np
import plotly.graph_objects as go
import plotly.express as px
from datetime import datetime
import time
import json
import logging
from typing import Dict, Union
from pathlib import Path
from src.utils.model_utils import ModelPredictor
from src.utils.data_preprocessing import validate_input_data
from src.utils import monitoring
# --- Basic Setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
st.set_page_config(
page_title="Advanced Fraud Detection System",
page_icon="🛡️",
layout="wide",
initial_sidebar_state="expanded"
)
# --- Load CSS from external file ---
def load_css(file_path: Path):
with open(file_path) as f:
st.markdown(f"", unsafe_allow_html=True)
# --- Model & State Management ---
@st.cache_resource
def get_predictor():
"""Cached function to load the model predictor once."""
try:
predictor = ModelPredictor()
return predictor
except Exception as e:
st.error(f"🚨 Critical Error: Could not load the model. Details: {e}")
return None
def create_sample_transaction() -> Dict:
"""Generates a random sample transaction for demonstration."""
np.random.seed(int(time.time()))
sample = {'Time': np.random.randint(0, 172800), 'Amount': round(np.random.uniform(1, 1000), 2)}
for i in range(1, 29):
sample[f'V{i}'] = np.random.normal(0, 1)
return sample
# --- UI Page Functions ---
def single_prediction_page(predictor: ModelPredictor):
st.markdown('🔍 Real-Time Transaction Analysis
Analyze individual transactions for fraud patterns.
', unsafe_allow_html=True)
st.subheader("💳 Transaction Input")
input_method = st.radio("Select input method:", ["✍️ Manual Entry", "🎲 Sample Transaction", "📋 JSON Import"], horizontal=True, key="input_method")
# Form for Manual Entry
if input_method == "✍️ Manual Entry":
with st.form("transaction_form"):
# Your form code remains the same...
st.markdown("**⏰ Timing & Amount**")
c1, c2 = st.columns(2)
time_val = c1.number_input("Time (seconds since first transaction)", min_value=0, value=3600, help="Time elapsed in seconds.")
amount_val = c2.number_input("Amount ($)", min_value=0.01, value=100.0, format="%.2f")
st.markdown("**🔢 Anonymized Features (V1-V28)**")
v_features = {}
with st.expander("Enter V1-V28 Features"):
cols = st.columns(4)
for i in range(1, 29):
with cols[(i-1)%4]:
v_features[f'V{i}'] = st.number_input(f"V{i}", value=0.0, format="%.6f", key=f"v{i}")
submitted = st.form_submit_button("🚀 Analyze Transaction", use_container_width=True, type="primary")
if submitted:
st.session_state.transaction_data = {'Time': time_val, 'Amount': amount_val, **v_features}
st.session_state.analysis_triggered = True
# Logic for other input methods
elif input_method == "🎲 Sample Transaction":
if st.button("🎲 Generate & Analyze Sample", use_container_width=True, type="primary"):
st.session_state.transaction_data = create_sample_transaction()
st.session_state.analysis_triggered = True
elif input_method == "📋 JSON Import":
json_input = st.text_area("Paste JSON data here:", height=250, placeholder='{"Time": 3600, "Amount": 100.0, ...}')
if st.button("🔍 Process JSON", use_container_width=True, type="primary"):
if json_input:
try:
st.session_state.transaction_data = json.loads(json_input)
st.session_state.analysis_triggered = True
except json.JSONDecodeError:
st.error("❌ Invalid JSON format.")
st.markdown('
', unsafe_allow_html=True)
with col2:
st.markdown('', unsafe_allow_html=True)
st.subheader("🎯 Analysis Results")
if st.session_state.get('analysis_triggered', False):
transaction_data = st.session_state.transaction_data
if not validate_input_data(transaction_data):
st.error("❌ Invalid input data. Please check all required fields.")
else:
try:
with st.spinner("🧠 AI analyzing transaction..."):
start_time = time.time()
prediction, probability = predictor.predict_single(transaction_data, use_optimal_threshold=True)
processing_time = time.time() - start_time
monitoring.log_prediction(transaction_data, prediction, probability, processing_time)
if prediction == 1:
st.markdown('
🚨 FRAUD DETECTED 🚨
', unsafe_allow_html=True)
else:
st.markdown('
✅ LEGITIMATE TRANSACTION
', unsafe_allow_html=True)
# ENHANCEMENT: Add Risk Level and Gauge Chart
if probability > 0.8: risk_level, risk_color = "High", "red"
elif probability > 0.5: risk_level, risk_color = "Medium", "orange"
else: risk_level, risk_color = "Low", "green"
st.metric("⚠️ Risk Level", risk_level)
fig = go.Figure(go.Indicator(
mode = "gauge+number",
value = probability,
domain = {'x': [0, 1], 'y': [0, 1]},
title = {'text': "Fraud Score", 'font': {'size': 20, 'color': 'white'}},
number = {'suffix': "%", 'valueformat': '.1f', 'font': {'size': 30, 'color': 'white'}},
gauge = {
'axis': {'range': [None, 1], 'tickwidth': 1, 'tickcolor': "white", 'tickformat': '.0%'},
'bar': {'color': risk_color, 'thickness': 0.3},
'bgcolor': "#4A4A4A",
'steps': [
{'range': [0, 0.5], 'color': '#3A683B'},
{'range': [0.5, 0.8], 'color': '#75602A'},
{'range': [0.8, 1], 'color': '#753228'}],
'threshold': {
'line': {'color': "#FFD700", 'width': 3},
'thickness': 0.75,
'value': predictor.optimal_threshold
}}))
fig.update_layout(paper_bgcolor="rgba(0,0,0,0)", font={'color': "white"})
st.plotly_chart(fig, use_container_width=True)
st.caption(f"Model Threshold: {predictor.optimal_threshold:.2%}")
st.success(f"⚡ Processed in {processing_time*1000:.1f}ms")
except Exception as e:
st.error(f"💥 Analysis error: {e}")
# Reset the trigger
st.session_state.analysis_triggered = False
else:
st.info("👆 Enter transaction details and click analyze to see results.")
st.markdown('
📊 Bulk Transaction Processing
Upload a CSV file to process multiple transactions at once.
📈 Real-Time Session Monitoring
Track prediction patterns for your current session.
🛡️ Advanced Fraud Detection System
AI-Powered Transaction Security Analysis