Spaces:
Sleeping
Sleeping
File size: 4,871 Bytes
c9875bd a124d29 892b85b a124d29 c9875bd | 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 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """
Data models for the anomaly detection system
"""
from dataclasses import dataclass, field, asdict
from datetime import datetime
from typing import Dict, Any, List, Optional
@dataclass
class TelecomConfig:
"""Configuration for telecom anomaly detection"""
# Call duration thresholds (seconds)
short_call_threshold: float = 3.0
long_call_threshold: float = 5400.0 # 90 minutes
extreme_call_threshold: float = 10800.0 # 3 hours
# Call frequency thresholds
max_calls_per_hour: int = 60
max_calls_per_day: int = 300
max_calls_per_minute: int = 10
# SMS thresholds
max_sms_per_hour: int = 100
max_sms_per_minute: int = 20
ignore_service_sms: bool = True
# Geographic thresholds
max_location_change_speed: float = 1000.0 # km/h
location_cluster_radius: float = 10.0 # km
location_cluster_km: float = 1.0 # Margin for 'same spot' clustering
# Behavioral thresholds
call_ratio_outgoing_incoming: float = 20.0
unique_contacts_per_day: int = 100
# Service numbers to ignore
service_numbers: List[str] = field(default_factory=list)
# Statistical thresholds
z_score_threshold: float = 4.0
iqr_multiplier: float = 2.0
# Time-based patterns
night_hours_start: int = 23
night_hours_end: int = 5
# Minimum samples needed
min_samples_for_pattern: int = 20
# Performance
batch_size: int = 10000
# Paths
output_dir: Optional[str] = None
save_pdf_dir: Optional[str] = None
def validate(self):
"""Validate configuration"""
assert self.short_call_threshold >= 0, "Short call threshold must be non-negative"
assert self.long_call_threshold > self.short_call_threshold, \
"Long call threshold must be greater than short call threshold"
assert self.extreme_call_threshold > self.long_call_threshold, \
"Extreme call threshold must be greater than long call threshold"
return self
@classmethod
def from_dict(cls, config_dict: Dict[str, Any]) -> 'TelecomConfig':
"""Create config from dictionary"""
return cls(**{k: v for k, v in config_dict.items() if k in cls.__dataclass_fields__})
@dataclass
class AnomalyResult:
"""Anomaly detection result"""
row_index: int
column_name: str
value: Any
detection_method: str
score: float
threshold: float
confidence: float
timestamp: datetime = field(default_factory=datetime.now)
metadata: Dict[str, Any] = field(default_factory=dict)
msisdn: str = ""
def to_dict(self) -> Dict:
"""Convert to dictionary"""
result = asdict(self)
result['timestamp'] = self.timestamp.isoformat()
return result
@dataclass
class DetectionStatistics:
"""Statistics from anomaly detection"""
total_rows: int
rows_with_anomalies: int
total_anomalies_detected: int
anomaly_rate: float
unique_subscribers: int
anomaly_types: Dict[str, int]
# Per-feature counts of "this feature was the #1 driver" across all
# IsolationForest flags. Empty when use_model=False (or no model
# anomalies fired). Lets the frontend chart "what features the AI
# model is finding anomalies in" alongside the rule-detector
# breakdown.
model_feature_drivers: Dict[str, int] = field(default_factory=dict)
# Consolidated IsolationForest summary — everything the frontend
# needs to render iforest-specific charts in one place. Empty when
# use_model=False or the model produced zero flags. Keys:
# total_flags — int, count of iforest anomalies
# subscribers_scored — int, msisdns that met min_samples
# subscribers_skipped — int, msisdns below min_samples
# category_counts — {category: count}, AI: … only
# feature_driver_counts — {feature: count}, same shape as the
# top-level model_feature_drivers map
# feature_z_means — {feature: mean_abs_z}, signal strength
# per feature across all flags
# confidence_buckets — {"0.60-0.65": n, …}, 7 buckets
# flags_by_hour — list[int] of length 24, hourly counts
# of AI flags only (so the frontend can
# draw a "when does the AI fire" chart
# distinct from the all-data hourly
# distribution it already has)
# top_subscribers — list[{msisdn, flag_count}], top 5 by
# AI flag count
model_summary: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict:
"""Convert to dictionary"""
return asdict(self) |