Spaces:
Sleeping
Sleeping
File size: 2,225 Bytes
4ebf33d aa28882 4ebf33d aa28882 4ebf33d aa28882 | 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 pandas as pd
import streamlit as st
from datetime import datetime
from typing import Dict, Any
# This file is now a functional module that uses Streamlit's session state
# for in-memory, per-user logging without writing any files.
def initialize_monitor_state():
"""
Initializes the session state for monitoring if it doesn't already exist.
Streamlit's session_state persists across script reruns for a single user session.
"""
if 'prediction_logs' not in st.session_state:
st.session_state.prediction_logs = []
def log_prediction(transaction_data: Dict[str, Any], prediction: int, probability: float, processing_time: float):
"""
Logs a single prediction to the in-memory list stored in the user's session state.
This does not write to any files on disk.
"""
# Ensure the log list exists in the session state
initialize_monitor_state()
log_entry = {
"timestamp": datetime.now(),
"prediction": prediction,
"is_fraud": "Fraud" if prediction == 1 else "Legitimate",
"fraud_probability": probability,
"processing_time_ms": processing_time * 1000,
"amount": transaction_data.get("Amount", 0) # Get amount from input for context
}
# Append the new log to the list in the session state
st.session_state.prediction_logs.append(log_entry)
def get_logs_as_dataframe() -> pd.DataFrame:
"""
Retrieves all logs from the current session state and returns them as a pandas DataFrame.
This reads from memory, not from a file.
"""
# Ensure the log list exists in the session state
initialize_monitor_state()
if not st.session_state.prediction_logs:
return pd.DataFrame()
df = pd.DataFrame(st.session_state.prediction_logs)
# Ensure columns are in a consistent order for display
display_columns = [
"timestamp",
"is_fraud",
"fraud_probability",
"amount",
"processing_time_ms",
"prediction"
]
# Filter to only include columns that exist, in case of future changes
existing_columns = [col for col in display_columns if col in df.columns]
return df[existing_columns] |