Spaces:
Sleeping
Sleeping
| """Monitoring service for model health reporting. | |
| Reports model health status, metrics, and prediction statistics | |
| for all registered models. | |
| """ | |
| import logging | |
| from datetime import datetime, timezone | |
| from app.core.config import settings | |
| from app.models.registry import ModelRegistry, REGISTERED_MODELS | |
| from app.monitoring.prediction_logger import PredictionLogger | |
| from app.schemas.monitoring import ModelHealthItem, MonitoringResponse | |
| logger = logging.getLogger(__name__) | |
| class MonitoringService: | |
| """Reports model health and prediction statistics.""" | |
| def __init__( | |
| self, | |
| registry: ModelRegistry, | |
| pred_logger: PredictionLogger, | |
| ) -> None: | |
| self._registry = registry | |
| self._pred_logger = pred_logger | |
| def get_model_health(self) -> MonitoringResponse: | |
| """Return health status, metrics, and prediction stats for all models. | |
| Algorithm: | |
| 1. Get all model statuses from registry | |
| 2. For each model: get metadata for version, last_trained, key metrics | |
| 3. Query prediction_logger for last 24h stats (count, avg confidence) | |
| 4. Return aggregated health report | |
| """ | |
| all_status = self._registry.get_all_status() | |
| model_items: list[ModelHealthItem] = [] | |
| for model_name in REGISTERED_MODELS: | |
| status_entry = all_status.get(model_name, {}) | |
| status = status_entry.get("status", "not_loaded") | |
| # Get metadata from registry | |
| metadata = self._registry.get_metadata(model_name) | |
| if metadata is not None: | |
| model_version = metadata.get( | |
| "model_version", f"{model_name}_v2_baseline_001" | |
| ) | |
| last_trained = metadata.get("trained_at") | |
| metrics = metadata.get("metrics", {}) | |
| else: | |
| # Defaults when metadata is not available | |
| model_version = f"{model_name}_v2_baseline_001" | |
| last_trained = None | |
| metrics = {} | |
| # Query prediction logger for last 24h stats | |
| recent_stats = self._pred_logger.get_recent_stats(model_name) | |
| predictions_last_24h = recent_stats.get("prediction_count", 0) | |
| avg_confidence_last_24h = recent_stats.get("avg_confidence") | |
| model_items.append( | |
| ModelHealthItem( | |
| model_name=model_name, | |
| model_version=model_version, | |
| status=status, | |
| last_trained=last_trained, | |
| metrics=metrics, | |
| predictions_last_24h=predictions_last_24h, | |
| avg_confidence_last_24h=avg_confidence_last_24h, | |
| ) | |
| ) | |
| return MonitoringResponse( | |
| service_version=settings.ai_service_version, | |
| timestamp=datetime.now(timezone.utc).isoformat(), | |
| models=model_items, | |
| ) | |