Spaces:

winsweb
/

teleprocessinggh

Sleeping

teleprocessinggh / telecom_anomaly /core /detector.py

fix AI model

892b85b 6 months ago

20.9 kB

"""
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)

	"""
	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)