monitoring/model_monitoring/prediction_logger.py

import pandas as pd
import json
from pathlib import Path
from datetime import datetime
from typing import Any, Dict, List
import threading


class PredictionLogger:
    def __init__(self, log_dir: Path):
        self.log_dir = Path(log_dir)
        self.log_dir.mkdir(parents=True, exist_ok=True)
        self.lock = threading.Lock()
    
    def log_prediction(self, 
                      input_data: Dict[str, Any], 
                      prediction: Any, 
                      model_version: str = "v1",
                      metadata: Dict[str, Any] = None):
        """Log a single prediction"""
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "model_version": model_version,
            "input": input_data,
            "prediction": prediction,
            "metadata": metadata or {}
        }
        
        log_file = self.log_dir / f"predictions_{datetime.now().strftime('%Y%m%d')}.jsonl"
        
        with self.lock:
            with open(log_file, 'a') as f:
                f.write(json.dumps(log_entry) + '\n')
    
    def load_predictions(self, date: str = None) -> List[Dict[str, Any]]:
        """Load predictions from log file"""
        if date is None:
            date = datetime.now().strftime('%Y%m%d')
        
        log_file = self.log_dir / f"predictions_{date}.jsonl"
        
        if not log_file.exists():
            return []
        
        predictions = []
        with open(log_file, 'r') as f:
            for line in f:
                predictions.append(json.loads(line))
        
        return predictions
    
    def get_predictions_df(self, date: str = None) -> pd.DataFrame:
        """Get predictions as DataFrame"""
        predictions = self.load_predictions(date)
        return pd.DataFrame(predictions) if predictions else pd.DataFrame()