File size: 2,608 Bytes
c40c447
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
"""
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
        }