""" DTOs para casos de uso de Detección de Anomalías. """ from dataclasses import dataclass from typing import List, Optional, Dict @dataclass class AnomalyPointDTO: """DTO para un punto de anomalía.""" index: int value: float expected: float lower_bound: float upper_bound: float is_anomaly: bool z_score: float = 0.0 severity: str = "normal" # normal, low, medium, high def to_dict(self) -> Dict: """Convierte a diccionario.""" return { "index": self.index, "value": self.value, "expected": self.expected, "lower_bound": self.lower_bound, "upper_bound": self.upper_bound, "is_anomaly": self.is_anomaly, "z_score": round(self.z_score, 2), "severity": self.severity } @dataclass class AnomalyDetectionInputDTO: """DTO de entrada para detección de anomalías.""" context_values: List[float] recent_values: List[float] quantile_low: float = 0.05 quantile_high: float = 0.95 context_timestamps: Optional[List[str]] = None freq: str = "D" def validate(self) -> None: """Valida los datos de entrada.""" if not self.context_values: raise ValueError("context_values no puede estar vacío") if not self.recent_values: raise ValueError("recent_values no puede estar vacío") if len(self.context_values) < 3: raise ValueError("context_values debe tener al menos 3 puntos") if not (0 < self.quantile_low < 0.5): raise ValueError("quantile_low debe estar en (0, 0.5)") if not (0.5 < self.quantile_high < 1): raise ValueError("quantile_high debe estar en (0.5, 1)") if self.context_timestamps and len(self.context_timestamps) != len(self.context_values): raise ValueError("context_timestamps y context_values deben tener la misma longitud") @dataclass class AnomalyDetectionOutputDTO: """DTO de salida para detección de anomalías.""" anomalies: List[AnomalyPointDTO] total_points: int anomaly_count: int anomaly_rate: float summary: Dict def to_dict(self) -> Dict: """Convierte a diccionario.""" return { "anomalies": [a.to_dict() for a in self.anomalies], "total_points": self.total_points, "anomaly_count": self.anomaly_count, "anomaly_rate": round(self.anomaly_rate, 3), "summary": self.summary }