src/utils/monitoring.py

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]