""" Monitoring module — Evidently-based drift detection and retrain triggers. Usage ----- from src.monitor import DriftMonitor monitor = DriftMonitor(reference_df, current_df) report_path = monitor.generate_report() if monitor.should_retrain(): # trigger retrain pipeline """ from __future__ import annotations import json import logging from datetime import datetime from pathlib import Path from typing import Dict, List, Optional, Tuple import pandas as pd from evidently import ColumnMapping from evidently.metric_preset import DataDriftPreset, TargetDriftPreset from evidently.report import Report from src.config import ( CATEGORICAL_FEATURES, LEAK_COLUMNS, MONITORING_DIR, NUMERIC_FEATURES, TARGET_COL, ) from src.features import engineer_features, _resolve_columns, _ENGINEERED_NUMERIC logger = logging.getLogger(__name__) # A retrain is suggested when this fraction of numeric features have drifted. RETRAIN_DRIFT_FRACTION = 0.30 class DriftMonitor: """Compare a reference (training) dataset to a current (production) batch.""" def __init__( self, reference_df: pd.DataFrame, current_df: pd.DataFrame, target_col: str = TARGET_COL, ) -> None: self.reference = engineer_features(reference_df.copy()) self.current = engineer_features(current_df.copy()) self.target_col = target_col num_cols, cat_cols = _resolve_columns(self.reference) # Remove leak columns that won't be present at inference time self.num_cols = [c for c in num_cols if c not in LEAK_COLUMNS] self.cat_cols = cat_cols self.column_mapping = ColumnMapping( target=self.target_col if self.target_col in self.reference.columns else None, numerical_features=self.num_cols, categorical_features=self.cat_cols, ) self._report: Optional[Report] = None self._drift_summary: Optional[Dict] = None def generate_report(self, output_dir: Path = MONITORING_DIR) -> Path: """Run Evidently DataDriftPreset and save HTML + JSON reports.""" metrics = [DataDriftPreset()] if self.target_col in self.reference.columns and self.target_col in self.current.columns: metrics.append(TargetDriftPreset()) report = Report(metrics=metrics) report.run( reference_data=self.reference, current_data=self.current, column_mapping=self.column_mapping, ) self._report = report timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S") html_path = output_dir / f"drift_report_{timestamp}.html" json_path = output_dir / f"drift_report_{timestamp}.json" report.save_html(str(html_path)) report.save_json(str(json_path)) # Parse the JSON for downstream decisions with open(json_path) as f: self._drift_summary = json.load(f) logger.info("Drift report saved → %s", html_path) return html_path def get_drifted_features(self) -> List[str]: """Return list of feature names flagged as drifted by Evidently.""" if self._drift_summary is None: self.generate_report() drifted: List[str] = [] try: metrics = self._drift_summary.get("metrics", []) for metric in metrics: result = metric.get("result", {}) drift_by_columns = result.get("drift_by_columns", {}) for col_name, col_info in drift_by_columns.items(): if col_info.get("drift_detected", False): drifted.append(col_name) except (KeyError, TypeError): logger.warning("Could not parse drift summary — returning empty list.") return drifted def should_retrain(self) -> bool: """ Simple retrain trigger: if more than RETRAIN_DRIFT_FRACTION of tracked numeric features have drifted, recommend retraining. """ drifted = self.get_drifted_features() tracked = set(self.num_cols + self.cat_cols) drifted_tracked = [f for f in drifted if f in tracked] fraction = len(drifted_tracked) / max(len(tracked), 1) should = fraction >= RETRAIN_DRIFT_FRACTION logger.info( "Drift check — %d/%d features drifted (%.0f%%) → retrain=%s", len(drifted_tracked), len(tracked), fraction * 100, should, ) return should def retrain_trigger( reference_df: pd.DataFrame, current_df: pd.DataFrame, ) -> Tuple[bool, Path]: """ Convenience function: run full monitoring pipeline and return (should_retrain, report_path). In production, this would be called by a scheduler (Airflow, cron, etc.) and the boolean would gate a retraining DAG. """ monitor = DriftMonitor(reference_df, current_df) report_path = monitor.generate_report() return monitor.should_retrain(), report_path