""" Main anomaly detector orchestrator """ import logging import numpy as np from pathlib import Path from typing import List, Dict, Any, Optional, Set from collections import defaultdict from datetime import datetime from sklearn.cluster import DBSCAN from telecom_anomaly.core.models import TelecomConfig, AnomalyResult, DetectionStatistics from telecom_anomaly.io.reader import TelecomDataReader from telecom_anomaly.io.exporter import ResultExporter from telecom_anomaly.io.pdf_exporter import PDFExporter from telecom_anomaly.validation.validator import DataValidator # Import detectors from telecom_anomaly.detection.duration import DurationDetector from telecom_anomaly.detection.frequency import FrequencyDetector from telecom_anomaly.detection.sms import SMSDetector from telecom_anomaly.detection.location import LocationDetector # from telecom_anomaly.detection.imei import IMEIDetector from telecom_anomaly.detection.time_patterns import TimePatternDetector from telecom_anomaly.detection.isolation_forest import IsolationForestDetector logger = logging.getLogger(__name__) class TelecomAnomalyDetector: """ Main orchestrator for telecom anomaly detection """ def __init__(self, config: Optional[TelecomConfig] = None): self.config = config or TelecomConfig() self.config.validate() # Initialize components self.reader = TelecomDataReader(self.config, self.config.batch_size) self.exporter = None self.pdf_exporter = None # Exporters are only constructed when an output directory is provided. # The FastAPI service leaves these as None so no files are written. if self.config.output_dir: self.exporter = ResultExporter(self.config.output_dir) if self.config.save_pdf_dir: self.pdf_exporter = PDFExporter(self.config.save_pdf_dir) elif self.config.output_dir: pdf_dir = str(Path(self.config.output_dir).parent / "save_pdf") self.pdf_exporter = PDFExporter(pdf_dir) self.validator = DataValidator() # Rule-based detectors always run. Model-based detectors are # opt-in via detect_anomalies(use_model=True). self.rule_detectors = [ DurationDetector(self.config), FrequencyDetector(self.config), SMSDetector(self.config), LocationDetector(self.config), # IMEIDetector(self.config), TimePatternDetector(self.config), ] self.model_detectors = [ IsolationForestDetector(self.config), ] self.detectors = self.rule_detectors # back-compat alias # Data storage self.data: List[Dict[str, Any]] = [] self.headers: List[str] = [] self.original_data: List[Dict[str, Any]] = [] self.data_indices: Dict[int, int] = {} # processed_idx -> original_idx self.anomalies: List[AnomalyResult] = [] # Anomaly tracking self.rows_with_anomalies: Set[int] = set() self.row_anomaly_map: Dict[int, List[str]] = defaultdict(list) # Per-row "reason" strings, parallel to row_anomaly_map (index i in # the value list corresponds to method i). Empty string when the # method didn't supply a reason (rule detectors today). The iforest # detector populates this from anomaly.metadata['reason']. self.row_anomaly_reasons: Dict[int, List[str]] = defaultdict(list) # Per-row structured attribution, parallel to row_anomaly_map. # Each entry is a list of {feature, z, value, direction} dicts # (or an empty list for methods that didn't compute attribution — # i.e. the rule detectors today). Lets the frontend draw per-row # bar charts of which features drove a flag. self.row_anomaly_attribution: Dict[int, List[List[Dict[str, Any]]]] = \ defaultdict(list) self.subscriber_anomalies: Dict[str, Dict] = defaultdict( lambda: {'total': 0, 'methods': set()} ) logger.info( f"Initialized TelecomAnomalyDetector with " f"{len(self.rule_detectors)} rule detectors and " f"{len(self.model_detectors)} model detectors (opt-in)" ) def load_data(self, file_paths: List[str], filter_movements: bool = False) -> 'TelecomAnomalyDetector': """Load data from files""" self.data, self.original_data = self.reader.read_files(file_paths) self.headers = self.reader.headers if filter_movements and self.data: self.filter_distinct_movements() # Build index mapping self.data_indices = {} for processed_idx, row in enumerate(self.data): if '_original_idx' in row: self.data_indices[processed_idx] = row['_original_idx'] logger.info(f"Loaded {len(self.data)} valid records") return self def load_dataframe(self, df: 'pd.DataFrame', filter_movements: bool = False) -> 'TelecomAnomalyDetector': """Load data from a pandas DataFrame""" self.data, self.original_data = self.reader.read_dataframe(df) self.headers = self.reader.headers if filter_movements and self.data: self.filter_distinct_movements() # Build index mapping self.data_indices = {} for processed_idx, row in enumerate(self.data): if '_original_idx' in row: self.data_indices[processed_idx] = row['_original_idx'] logger.info(f"Loaded {len(self.data)} valid records from DataFrame") return self def filter_distinct_movements(self): """ Filter data to keep only representative records for distinct geographic locations. Uses DBSCAN with Haversine distance. """ if not self.data: return logger.info("Filtering for distinct geographic movements...") # Extract coordinates for valid location rows valid_indices = [] coords = [] for i, row in enumerate(self.data): lat = row.get('latitude') lon = row.get('longitude') if lat is not None and lon is not None: valid_indices.append(i) coords.append([lat, lon]) if not coords: return coords_np = np.array(coords) # Parameters for DBSCAN distance_km = self.config.location_cluster_km kms_per_radian = 6371.0088 epsilon = distance_km / kms_per_radian # Run DBSCAN db = DBSCAN( eps=epsilon, min_samples=1, algorithm='ball_tree', metric='haversine' ).fit(np.radians(coords_np)) # Identify representative row for each cluster cluster_labels = db.labels_ unique_clusters = set(cluster_labels) kept_indices = set() cluster_to_idx = {} for i, label in enumerate(cluster_labels): if label not in cluster_to_idx: cluster_to_idx[label] = valid_indices[i] kept_indices.add(valid_indices[i]) # Also keep rows without coordinates all_indices = set(range(len(self.data))) coord_indices = set(valid_indices) no_coord_indices = all_indices - coord_indices final_indices = sorted(list(kept_indices | no_coord_indices)) original_count = len(self.data) self.data = [self.data[i] for i in final_indices] logger.info(f"Geographic filtering complete: {original_count} -> {len(self.data)} records") def detect_anomalies( self, use_model: bool = False, run_rules: bool = True, ) -> List[AnomalyResult]: """ Run anomaly detectors over the loaded data. Args: use_model: When True, runs the model-based detectors (Isolation Forest). run_rules: When True, runs the rule-based detectors. Set to False to run the model alone — this is what powers the Flutter "Use AI Model" toggle without "Anomaly Detection". Defaults to True so existing callers (CLI, prior tests) keep their rule-only behaviour. """ start_time = datetime.now() logger.info( f"Starting anomaly detection " f"(run_rules={run_rules}, use_model={use_model})" ) detectors_to_run = [] if run_rules: detectors_to_run.extend(self.rule_detectors) if use_model: detectors_to_run.extend(self.model_detectors) all_anomalies = [] for detector in detectors_to_run: try: detector_anomalies = detector.detect(self.data) all_anomalies.extend(detector_anomalies) logger.info(f"{detector.name}: {len(detector_anomalies)} anomalies") except Exception as e: logger.error(f"Error in {detector.name}: {e}") # Combine anomalies self.anomalies = self._combine_anomalies(all_anomalies) # Update tracking self._update_anomaly_tracking() elapsed = (datetime.now() - start_time).total_seconds() logger.info(f"Detection completed in {elapsed:.2f}s") logger.info(f"Rows with anomalies: {len(self.rows_with_anomalies)}/{len(self.data)}") return self.anomalies def _combine_anomalies(self, anomalies: List[AnomalyResult]) -> List[AnomalyResult]: """Combine anomalies, removing duplicates""" if not anomalies: return [] # Separate row and subscriber anomalies row_anomalies = defaultdict(list) subscriber_anomalies = defaultdict(list) for a in anomalies: if a.row_index >= 0: row_anomalies[a.row_index].append(a) else: subscriber_anomalies[a.msisdn].append(a) combined = [] # Keep unique methods per row, highest confidence for row_idx, row_anoms in row_anomalies.items(): best_per_method = {} for anom in row_anoms: if anom.detection_method not in best_per_method or \ anom.confidence > best_per_method[anom.detection_method].confidence: best_per_method[anom.detection_method] = anom combined.extend(best_per_method.values()) # Keep unique methods per subscriber for msisdn, sub_anoms in subscriber_anomalies.items(): best_per_method = {} for anom in sub_anoms: if anom.detection_method not in best_per_method: best_per_method[anom.detection_method] = anom combined.extend(best_per_method.values()) return combined def _update_anomaly_tracking(self): """Update anomaly tracking maps""" for anomaly in self.anomalies: # Track rows with anomalies if anomaly.row_index >= 0: self.rows_with_anomalies.add(anomaly.row_index) original_idx = self.data_indices.get(anomaly.row_index) if original_idx is not None: self.row_anomaly_map[original_idx].append(anomaly.detection_method) # Mirror reason + structured attribution in the parallel # lists. Rule detectors don't currently set # metadata['reason'] / metadata['attribution'] so they # fall back to '' / [] to keep all three lists # index-aligned. reason = '' attribution: List[Dict[str, Any]] = [] if anomaly.metadata: reason = str(anomaly.metadata.get('reason', '') or '') attribution = list( anomaly.metadata.get('attribution', []) or [] ) self.row_anomaly_reasons[original_idx].append(reason) self.row_anomaly_attribution[original_idx].append(attribution) # Track subscriber anomalies if anomaly.msisdn: self.subscriber_anomalies[anomaly.msisdn]['total'] += 1 self.subscriber_anomalies[anomaly.msisdn]['methods'].add(anomaly.detection_method) def get_statistics(self) -> DetectionStatistics: """Get detection statistics""" rows_with_anomalies = len(self.rows_with_anomalies) # Calculate anomaly rate correctly anomaly_rate = (rows_with_anomalies / len(self.data)) * 100 if self.data else 0 # Count anomaly types anomaly_types = defaultdict(int) for anomaly in self.anomalies: anomaly_types[anomaly.detection_method] += 1 # Aggregate model-flag top-driver counts. Only iforest anomalies # carry a `family='iforest'` metadata marker today, so this naturally # ignores rule detectors. The output is shaped for direct rendering # by a bar chart on the visualization screen. model_feature_drivers: Dict[str, int] = defaultdict(int) for anomaly in self.anomalies: meta = anomaly.metadata or {} if meta.get('family') != 'iforest': continue top = meta.get('top_feature') or '' if not top: # Multivariate-but-no-top-driver flags get bucketed under # an explicit label so the chart still accounts for them. top = '_multivariate' model_feature_drivers[top] += 1 model_summary = self._build_model_summary( dict(model_feature_drivers) ) return DetectionStatistics( total_rows=len(self.data), rows_with_anomalies=rows_with_anomalies, total_anomalies_detected=len(self.anomalies), anomaly_rate=anomaly_rate, unique_subscribers=len(set(a.msisdn for a in self.anomalies if a.msisdn)), anomaly_types=dict(anomaly_types), model_feature_drivers=dict(model_feature_drivers), model_summary=model_summary, ) def _build_model_summary( self, feature_driver_counts: Dict[str, int], ) -> Dict[str, Any]: """ Consolidated IsolationForest report. Empty dict if no model anomalies fired (so the frontend can skip rendering the AI-specific charts cleanly). """ model_anomalies = [ a for a in self.anomalies if (a.metadata or {}).get('family') == 'iforest' ] if not model_anomalies: return {} # Category counts (AI: … only). Lets the frontend draw a bar # chart that's specifically about model output, without the # rule-detector counts mixed in like `anomaly_types` has. category_counts: Dict[str, int] = defaultdict(int) for a in model_anomalies: category_counts[a.detection_method] += 1 # Mean |z| per feature across every model flag. Highest value = # the feature with the strongest average pull on the model — a # natural "feature importance" chart. z_sums: Dict[str, List[float]] = defaultdict(list) for a in model_anomalies: for entry in (a.metadata.get('attribution') or []): z_sums[entry['feature']].append(abs(float(entry['z']))) feature_z_means: Dict[str, float] = { feat: round(sum(zs) / len(zs), 2) for feat, zs in z_sums.items() } # 7 fixed buckets covering the detector's 0.60..0.95 confidence # band. Fixed-width buckets so a histogram chart can render # without the frontend deciding bin edges. confidence_buckets: Dict[str, int] = { '0.60-0.65': 0, '0.65-0.70': 0, '0.70-0.75': 0, '0.75-0.80': 0, '0.80-0.85': 0, '0.85-0.90': 0, '0.90-0.95': 0, } for a in model_anomalies: c = a.confidence if c < 0.65: k = '0.60-0.65' elif c < 0.70: k = '0.65-0.70' elif c < 0.75: k = '0.70-0.75' elif c < 0.80: k = '0.75-0.80' elif c < 0.85: k = '0.80-0.85' elif c < 0.90: k = '0.85-0.90' else: k = '0.90-0.95' confidence_buckets[k] += 1 # Hour-of-day distribution restricted to AI flags only — a # "when does the model fire" view distinct from the frontend's # existing all-data hourly chart. flags_by_hour: List[int] = [0] * 24 for a in model_anomalies: original_idx = self.data_indices.get(a.row_index) if original_idx is None: continue row = self.original_data[original_idx] time_str = row.get('event_time') or '' try: hour = int(str(time_str).split(':')[0]) except (ValueError, IndexError): continue if 0 <= hour < 24: flags_by_hour[hour] += 1 # Top 5 subscribers by AI flag count — useful leaderboard chart # on multi-subscriber files. Meaningless on single-subscriber # files (just shows the one subscriber); the frontend can # gracefully hide the card when length <= 1. sub_counts: Dict[str, int] = defaultdict(int) for a in model_anomalies: if a.msisdn: sub_counts[a.msisdn] += 1 top_subscribers = [ {'msisdn': m, 'flag_count': c} for m, c in sorted( sub_counts.items(), key=lambda x: -x[1] )[:5] ] # Pull subscribers_scored / subscribers_skipped from whichever # model detector ran. Defensive — model_detectors list may be # empty (shouldn't happen at this point but cheap to guard). scored = 0 skipped = 0 for det in self.model_detectors: last = getattr(det, 'last_run_stats', None) or {} scored += int(last.get('subscribers_scored', 0)) skipped += int(last.get('subscribers_skipped', 0)) summary = { 'total_flags': len(model_anomalies), 'subscribers_scored': scored, 'subscribers_skipped': skipped, 'category_counts': dict(category_counts), 'feature_driver_counts': feature_driver_counts, 'feature_z_means': feature_z_means, 'confidence_buckets': confidence_buckets, 'flags_by_hour': flags_by_hour, 'top_subscribers': top_subscribers, } # Single-line server log so anyone tailing the API sees the # model output at the end of every run without grepping the # JSON response. logger.info( "IsolationForest summary: %d flags across %d categor%s; " "%d subscribers scored, %d skipped", summary['total_flags'], len(category_counts), 'y' if len(category_counts) == 1 else 'ies', scored, skipped, ) return summary def export_results(self, format: str = 'json', only_anomalies: bool = True) -> Dict[str, str]: """Export all results. CLI-only — the API never calls this.""" if self.exporter is None: return {} stats = self.get_statistics() anomaly_dicts = [a.to_dict() for a in self.anomalies] outputs = {} outputs['trackable'] = self.exporter.export_trackable_data( self.original_data, self.headers, self.row_anomaly_map, only_anomalies=only_anomalies ) outputs['json'] = self.exporter.export_results(anomaly_dicts, stats.to_dict(), 'json') outputs['csv'] = self.exporter.export_results(anomaly_dicts, stats.to_dict(), 'csv') outputs['report'] = self.exporter.export_report(stats.to_dict()) outputs['subscribers'] = self.exporter.export_suspicious_subscribers( dict(self.subscriber_anomalies) ) return outputs def export_pdf(self, results: List[Dict[str, Any]], filename: str) -> str: """Export results as PDF report. CLI-only — the API never calls this.""" if self.pdf_exporter is None: return "" stats = self.get_statistics().to_dict() return self.pdf_exporter.generate_report(stats, results, filename)