adding model

telecom_anomaly/api.py

@@ -1,8 +1,11 @@
+import asyncio
 import os
 import shutil
 import tempfile
 import logging
 from pathlib import Path
+from typing import Any, Dict, List
+
 from fastapi import FastAPI, UploadFile, File, HTTPException
 from fastapi.responses import RedirectResponse
 
@@ -19,7 +22,6 @@ from telecom_anomaly.utils.pdf_processor import (
     STRICT_CLEAN_DIR,
 )
 
-# Configure logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
@@ -32,6 +34,17 @@ app = FastAPI(
     version="1.0.0"
 )
 
+# Canonical 14-column output schema shared across CSV and PDF inputs.
+# msisdn is derived internally when missing; it intentionally is not
+# emitted in the response because call_direction + calling_no + called_no
+# already identify the subscriber unambiguously.
+REQUESTED_FIELDS: List[str] = [
+    "event_date", "event_time", "call_direction",
+    "calling_no", "called_no", "usage_type", "duration", "imei",
+    "location_id", "region", "district", "city",
+    "longitude", "latitude", "azimuth",
+]
+
 
 @app.get("/")
 async def root():
@@ -50,13 +63,38 @@ async def analyze_cdr(
     file: UploadFile = File(...),
     filter_movements: bool = False,
     run_anomaly_detection: bool = True,
+    use_model: bool = False,
     cluster_km: float = 1.0,
+    max_results: int = 100,
 ):
     """
     Analyze a PDF or CSV file for telecom anomalies. PDFs go through the
     3-stage disk-based pipeline (extract -> repair -> strict-clean) so the
-    detectors see exactly the same normalized inputs as telecom_anomaly_detection.
-    All intermediate CSVs are removed before the response returns.
+    detectors see exactly the same normalized inputs as
+    telecom_anomaly_detection. All intermediate CSVs are removed before
+    the response returns.
+
+    Args:
+        file: PDF or CSV file to analyze.
+        filter_movements: If true, clusters coordinates within cluster_km
+            and keeps only distinct locations.
+        run_anomaly_detection: If true, performs anomaly detection. When
+            false the endpoint just returns the cleaned/clustered rows.
+        use_model: If true (and run_anomaly_detection is true), also runs
+            the Isolation Forest model-based detector alongside the rule
+            detectors. Default false preserves the original rule-only
+            pipeline.
+        cluster_km: Distance margin in kilometres for geographic
+            clustering (default 1.0).
+        max_results: Maximum number of records to include inline in the
+            JSON response (default 100). Records are ranked by
+            anomaly_count descending so the most interesting rows
+            survive truncation. Set to 0 for no cap. This cap exists
+            so Swagger UI / the browser does not freeze rendering huge
+            payloads when use_model=true on large inputs.
+
+    Returns:
+        Detection statistics and a list of results (capped per max_results).
     """
     filename = file.filename.lower()
     is_pdf = filename.endswith('.pdf')
@@ -70,76 +108,25 @@ async def analyze_cdr(
         tmp_path = tmp.name
 
     base_name = os.path.splitext(os.path.basename(file.filename))[0]
-    raw_csv = None
-    interim_csv = None
-    final_csv = None
+    # The threaded helper appends intermediate-CSV paths to this list so
+    # the finally block can clean them up even when an exception fires.
+    paths_to_cleanup: List[str] = []
 
     try:
-        config = TelecomConfig()
-        config.service_numbers = SERVICE_NUMBERS
-        config.output_dir = None
-        config.save_pdf_dir = None
-        config.location_cluster_km = cluster_km
-
-        detector = TelecomAnomalyDetector(config)
-
-        if is_pdf:
-            logger.info(f"Processing PDF: {file.filename}")
-
-            raw_csv = extract_pdf_to_csv(tmp_path, base_name)
-            if not raw_csv:
-                raise HTTPException(status_code=500, detail="Failed to extract tables from PDF")
-
-            interim_csv = process_telecom_data_complete(raw_csv, base_name)
-            if not interim_csv:
-                raise HTTPException(status_code=500, detail="Failed to repair fragmented rows in CSV")
-
-            final_csv = strict_clean_data(interim_csv, base_name)
-            if not final_csv or not os.path.exists(final_csv):
-                raise HTTPException(status_code=500, detail="Strict-clean stage produced no output")
-
-            detector.load_data([final_csv], filter_movements=filter_movements)
-        else:
-            logger.info(f"Processing CSV: {file.filename}")
-            detector.load_data([tmp_path], filter_movements=filter_movements)
-
-        if run_anomaly_detection:
-            detector.detect_anomalies()
-
-        stats = detector.get_statistics()
-
-        requested_fields = [
-            "event_date", "event_time", "call_direction", "usage_sub_type",
-            "calling_no", "called_no", "usage_type", "duration", "imei",
-            "location_id", "region", "district", "city", "msisdn",
-            "longitude", "latitude", "azimuth",
-        ]
-
-        final_records_to_return = []
-        kept_original_indices = set(detector.data_indices.values())
-
-        for idx, row in enumerate(detector.original_data):
-            if idx not in kept_original_indices:
-                continue
-
-            methods = detector.row_anomaly_map.get(idx, [])
-
-            if run_anomaly_detection and not methods:
-                continue
-
-            record = {field: row.get(field, "") for field in requested_fields}
-            record.update({
-                "has_anomaly": "YES" if methods else "NO",
-                "anomaly_count": len(methods),
-                "anomaly_methods": "; ".join(methods),
-            })
-            final_records_to_return.append(record)
-
-        return {
-            "filename": file.filename,
-            "statistics": stats.to_dict() if run_anomaly_detection else {"total_rows": len(detector.data)},
-            "results": final_records_to_return,
-        }
+        payload = await asyncio.to_thread(
+            _run_analysis,
+            tmp_path=tmp_path,
+            original_filename=file.filename,
+            base_name=base_name,
+            is_pdf=is_pdf,
+            cluster_km=cluster_km,
+            filter_movements=filter_movements,
+            run_anomaly_detection=run_anomaly_detection,
+            use_model=use_model,
+            max_results=max_results,
+            paths_to_cleanup=paths_to_cleanup,
+        )
+        return payload
 
     except HTTPException:
         raise
@@ -155,13 +142,12 @@ async def analyze_cdr(
             except OSError:
                 pass
 
-        # Remove intermediate CSVs from each pipeline stage so the deploy stays stateless
-        for path in (
-            raw_csv,
-            interim_csv,
-            final_csv,
-            os.path.join(PROCESS_OUTPUT_DIR, f"{base_name}_customer_info.csv"),
-        ):
+        # Remove intermediate CSVs from each pipeline stage so the deploy
+        # stays stateless (matches the original behaviour).
+        paths_to_cleanup.append(
+            os.path.join(PROCESS_OUTPUT_DIR, f"{base_name}_customer_info.csv")
+        )
+        for path in paths_to_cleanup:
             if path and os.path.exists(path):
                 try:
                     os.remove(path)
@@ -169,6 +155,108 @@ async def analyze_cdr(
                     pass
 
 
+def _run_analysis(
+    tmp_path: str,
+    original_filename: str,
+    base_name: str,
+    is_pdf: bool,
+    cluster_km: float,
+    filter_movements: bool,
+    run_anomaly_detection: bool,
+    use_model: bool,
+    max_results: int,
+    paths_to_cleanup: List[str],
+) -> Dict[str, Any]:
+    """
+    Synchronous heavy work for /analyze.
+
+    Runs in a worker thread (via asyncio.to_thread) so the FastAPI event
+    loop stays responsive — without this, the browser sees the page as
+    frozen during long Isolation Forest fits and the user is prompted to
+    abandon the tab.
+
+    Mutates `paths_to_cleanup` with each intermediate CSV created so the
+    caller's finally block can delete them stateless-style.
+    """
+    config = TelecomConfig()
+    config.service_numbers = SERVICE_NUMBERS
+    config.output_dir = None
+    config.save_pdf_dir = None
+    config.location_cluster_km = cluster_km
+
+    detector = TelecomAnomalyDetector(config)
+
+    if is_pdf:
+        logger.info(f"Processing PDF: {original_filename}")
+
+        raw_csv = extract_pdf_to_csv(tmp_path, base_name)
+        if not raw_csv:
+            raise HTTPException(status_code=500, detail="Failed to extract tables from PDF")
+        paths_to_cleanup.append(raw_csv)
+
+        interim_csv = process_telecom_data_complete(raw_csv, base_name)
+        if not interim_csv:
+            raise HTTPException(status_code=500, detail="Failed to repair fragmented rows in CSV")
+        paths_to_cleanup.append(interim_csv)
+
+        final_csv = strict_clean_data(interim_csv, base_name)
+        if not final_csv or not os.path.exists(final_csv):
+            raise HTTPException(status_code=500, detail="Strict-clean stage produced no output")
+        paths_to_cleanup.append(final_csv)
+
+        detector.load_data([final_csv], filter_movements=filter_movements)
+    else:
+        logger.info(f"Processing CSV: {original_filename}")
+        detector.load_data([tmp_path], filter_movements=filter_movements)
+
+    if run_anomaly_detection:
+        detector.detect_anomalies(use_model=use_model)
+
+    stats = detector.get_statistics()
+    kept_original_indices = set(detector.data_indices.values())
+
+    full_records: List[Dict[str, Any]] = []
+    for idx, row in enumerate(detector.original_data):
+        if idx not in kept_original_indices:
+            continue
+        methods = detector.row_anomaly_map.get(idx, [])
+        if run_anomaly_detection and not methods:
+            continue
+        record = {field: row.get(field, "") for field in REQUESTED_FIELDS}
+        record.update({
+            "has_anomaly": "YES" if methods else "NO",
+            "anomaly_count": len(methods),
+            "anomaly_methods": "; ".join(methods),
+        })
+        full_records.append(record)
+
+    # Inline response cap. Rank by anomaly_count desc so the most
+    # interesting rows survive truncation. max_results == 0 disables.
+    if max_results and max_results > 0 and len(full_records) > max_results:
+        response_records = sorted(
+            full_records,
+            key=lambda r: r.get("anomaly_count", 0),
+            reverse=True,
+        )[:max_results]
+        truncated = True
+    else:
+        response_records = full_records
+        truncated = False
+
+    return {
+        "filename": original_filename,
+        "statistics": (
+            stats.to_dict()
+            if run_anomaly_detection
+            else {"total_rows": len(detector.data)}
+        ),
+        "results": response_records,
+        "total_results": len(full_records),
+        "results_truncated": truncated,
+        "results_cap": max_results,
+    }
+
+
 if __name__ == "__main__":
     import uvicorn
     uvicorn.run(app, host="0.0.0.0", port=8000)

telecom_anomaly/core/detector.py

@@ -23,6 +23,7 @@ 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__)
 
@@ -54,15 +55,20 @@ class TelecomAnomalyDetector:
 
         self.validator = DataValidator()
         
-        # Initialize detectors
-        self.detectors = [
+        # 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)
+            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]] = []
@@ -78,7 +84,11 @@ class TelecomAnomalyDetector:
             lambda: {'total': 0, 'methods': set()}
         )
         
-        logger.info(f"Initialized TelecomAnomalyDetector with {len(self.detectors)} detectors")
+        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"""
@@ -176,15 +186,26 @@ class TelecomAnomalyDetector:
         
         logger.info(f"Geographic filtering complete: {original_count} -> {len(self.data)} records")
     
-    def detect_anomalies(self) -> List[AnomalyResult]:
-        """Run all anomaly detectors"""
+    def detect_anomalies(self, use_model: bool = False) -> List[AnomalyResult]:
+        """
+        Run anomaly detectors over the loaded data.
+
+        Args:
+            use_model: When True, also runs model-based detectors
+                (Isolation Forest). Default False keeps behaviour
+                identical to the original rule-only pipeline.
+        """
         start_time = datetime.now()
-        logger.info("Starting anomaly detection")
-        
+        logger.info(
+            f"Starting anomaly detection (use_model={use_model})"
+        )
+
+        detectors_to_run = list(self.rule_detectors)
+        if use_model:
+            detectors_to_run.extend(self.model_detectors)
+
         all_anomalies = []
-        
-        # Run each detector
-        for detector in self.detectors:
+        for detector in detectors_to_run:
             try:
                 detector_anomalies = detector.detect(self.data)
                 all_anomalies.extend(detector_anomalies)

telecom_anomaly/detection/isolation_forest.py

@@ -0,0 +1,171 @@
+"""
+Isolation Forest anomaly detector.
+
+Acts as a 6th detector alongside the rule-based ones. The rule detectors
+each check a single column against a hard threshold; this one looks for
+records that are unusual in their *combination* of features (e.g. a short
+call at an out-of-region tower at 03:00 from a rarely-seen IMEI, where no
+single rule fires).
+
+Trained per-msisdn so each subscriber's "normal" is defined relative to
+their own history. Subscribers below `min_samples_for_model` rows are
+skipped to avoid overfitting noise.
+"""
+
+from collections import Counter, defaultdict
+from typing import List, Dict, Any
+
+import numpy as np
+from sklearn.ensemble import IsolationForest
+
+from telecom_anomaly.detection.base import BaseDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class IsolationForestDetector(BaseDetector):
+    """Multivariate unsupervised anomaly detector."""
+
+    FEATURE_COLUMNS = [
+        'duration',
+        'hour_of_day',
+        'is_night',
+        'is_sms',
+        'direction_outgoing',
+        'latitude',
+        'longitude',
+        'region_freq',
+        'district_freq',
+    ]
+
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+        self.contamination = getattr(config, 'iforest_contamination', 'auto')
+        self.min_samples = getattr(config, 'iforest_min_samples', 50)
+        self.random_state = getattr(config, 'iforest_random_state', 42)
+        self.n_estimators = getattr(config, 'iforest_n_estimators', 100)
+
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        if not data:
+            return []
+
+        per_subscriber: Dict[str, List[int]] = defaultdict(list)
+        for idx, row in enumerate(data):
+            msisdn = row.get('msisdn', '')
+            if msisdn:
+                per_subscriber[msisdn].append(idx)
+
+        anomalies: List[AnomalyResult] = []
+        for msisdn, indices in per_subscriber.items():
+            if len(indices) < self.min_samples:
+                continue
+            anomalies.extend(self._detect_for_subscriber(msisdn, indices, data))
+
+        return anomalies
+
+    def _detect_for_subscriber(self, msisdn: str, indices: List[int],
+                               data: List[Dict]) -> List[AnomalyResult]:
+        rows = [data[i] for i in indices]
+        features = self._build_features(rows)
+
+        if features.shape[0] < self.min_samples:
+            return []
+
+        model = IsolationForest(
+            n_estimators=self.n_estimators,
+            contamination=self.contamination,
+            random_state=self.random_state,
+            n_jobs=1,
+        )
+        model.fit(features)
+
+        predictions = model.predict(features)          # -1 = anomaly, 1 = normal
+        scores = model.score_samples(features)         # lower = more anomalous
+
+        # Map raw scores to a 0..1 confidence using min-max within this
+        # subscriber. Scores are negative; the minimum score gets the
+        # highest confidence. This keeps the confidence field comparable
+        # to the rule detectors' 0.75-0.95 range.
+        s_min, s_max = float(scores.min()), float(scores.max())
+        denom = (s_max - s_min) or 1.0
+
+        results: List[AnomalyResult] = []
+        for local_idx, global_idx in enumerate(indices):
+            if predictions[local_idx] != -1:
+                continue
+            raw_score = float(scores[local_idx])
+            normalized = (s_max - raw_score) / denom            # 0..1
+            confidence = round(0.6 + 0.35 * normalized, 4)      # 0.60..0.95
+            results.append(self._create_anomaly(
+                row_index=global_idx,
+                column_name='multivariate',
+                value=raw_score,
+                method='iforest',
+                score=raw_score,
+                threshold=float(model.offset_),
+                confidence=confidence,
+                msisdn=msisdn,
+                metadata={
+                    'features': self.FEATURE_COLUMNS,
+                    'subscriber_samples': len(indices),
+                },
+            ))
+        return results
+
+    def _build_features(self, rows: List[Dict[str, Any]]) -> np.ndarray:
+        region_freq = self._frequency_map(rows, 'region')
+        district_freq = self._frequency_map(rows, 'district')
+
+        lat_values = [r.get('latitude') for r in rows if r.get('latitude') is not None]
+        lon_values = [r.get('longitude') for r in rows if r.get('longitude') is not None]
+        lat_default = float(np.median(lat_values)) if lat_values else 0.0
+        lon_default = float(np.median(lon_values)) if lon_values else 0.0
+
+        features = []
+        for row in rows:
+            duration = float(row.get('duration') or 0.0)
+            hour = self._hour_of_day(row.get('event_time', ''))
+            is_night = 1 if row.get('is_night') else 0
+            is_sms = 1 if (str(row.get('usage_type') or '')).upper() == 'SMS' else 0
+            direction = (str(row.get('call_direction') or '')).upper()
+            direction_outgoing = 1 if ('OUT' in direction or 'MO' in direction) else 0
+            lat = row.get('latitude')
+            lon = row.get('longitude')
+            lat = float(lat) if lat is not None else lat_default
+            lon = float(lon) if lon is not None else lon_default
+            region = str(row.get('region') or '')
+            district = str(row.get('district') or '')
+
+            features.append([
+                duration,
+                hour,
+                is_night,
+                is_sms,
+                direction_outgoing,
+                lat,
+                lon,
+                region_freq.get(region, 0.0),
+                district_freq.get(district, 0.0),
+            ])
+
+        return np.asarray(features, dtype=np.float64)
+
+    @staticmethod
+    def _frequency_map(rows: List[Dict], column: str) -> Dict[str, float]:
+        """
+        Frequency-encode a categorical column. Rare values get small
+        numbers; common values get large numbers. Rarity itself is the
+        signal Isolation Forest can exploit.
+        """
+        counts = Counter(str(row.get(column) or '') for row in rows)
+        total = sum(counts.values()) or 1
+        return {value: count / total for value, count in counts.items()}
+
+    @staticmethod
+    def _hour_of_day(time_str) -> int:
+        try:
+            from datetime import time as _time
+            if isinstance(time_str, _time):
+                return int(time_str.hour)
+            return int(str(time_str).split(':')[0])
+        except (ValueError, IndexError, AttributeError):
+            return 0

telecom_anomaly/io/pdf_exporter.py

@@ -43,30 +43,30 @@ class PDFExporter:
         
         # Detailed Table
         pdf.set_font("helvetica", "B", 12)
-        pdf.cell(0, 8, " 2. Detailed Anomalous Records (All 20 Fields)", ln=True, fill=True)
+        pdf.cell(0, 8, " 2. Detailed Anomalous Records", ln=True, fill=True)
         pdf.ln(3)
-        
-        # Define 20 columns and their widths (Landscape A4 width is ~277mm inside margins)
+
+        # Columns mirror the 14 common-input fields plus the 3 anomaly
+        # columns. Widths sum to <277mm (Landscape A4 minus margins);
+        # the last column is auto-sized to fill remaining width.
         fields = [
-            ("Date", "event_date", 16),
-            ("Time", "event_time", 14),
-            ("Dir", "call_direction", 8),
-            ("Sub-Type", "usage_sub_type", 16),
-            ("Calling No", "calling_no", 22),
-            ("Called No", "called_no", 22),
-            ("Type", "usage_type", 11),
-            ("Dur", "duration", 8),
-            ("IMEI", "imei", 24),
-            ("Loc-ID", "location_id", 14),
-            ("Region", "region", 15),
-            ("District", "district", 15),
-            ("City", "city", 15),
-            ("MSISDN", "msisdn", 22),
+            ("Date", "event_date", 18),
+            ("Time", "event_time", 16),
+            ("Dir", "call_direction", 14),
+            ("Calling No", "calling_no", 24),
+            ("Called No", "called_no", 24),
+            ("Type", "usage_type", 12),
+            ("Dur", "duration", 10),
+            ("IMEI", "imei", 26),
+            ("Loc-ID", "location_id", 18),
+            ("Region", "region", 16),
+            ("District", "district", 16),
+            ("City", "city", 16),
             ("Long", "longitude", 14),
             ("Lat", "latitude", 14),
-            ("Azi", "azimuth", 8),
-            ("Anom", "has_anomaly", 8),
-            ("Cnt", "anomaly_count", 6),
+            ("Azi", "azimuth", 10),
+            ("Anom", "has_anomaly", 10),
+            ("Cnt", "anomaly_count", 8),
             ("Methods", "anomaly_methods", 0) # 0 means fill remaining width
         ]
         

telecom_anomaly/io/reader.py

@@ -65,9 +65,12 @@ class TelecomDataReader:
         records = []
         original_data = []
         
-        # Set headers from dataframe columns if not already set
-        if not self.headers:
-            self.headers = [str(c).strip() for c in df.columns]
+        # Always extend the known-headers set with this DataFrame's
+        # columns so multi-call usage doesn't silently misalign columns.
+        for col in df.columns:
+            name = str(col).strip()
+            if name not in self.headers:
+                self.headers.append(name)
         
         # Convert DataFrame to records (list of dicts)
         from datetime import date, time
@@ -96,34 +99,36 @@ class TelecomDataReader:
         logger.info(f"Processed {len(records)} valid records from {len(original_data)} DataFrame rows")
         return records, original_data
     
-    def _read_single_file(self, file_path: str, encoding: str, 
+    def _read_single_file(self, file_path: str, encoding: str,
                           base_idx: int) -> tuple:
         """Read and validate a single CSV file"""
         records = []
         original_data = []
-        
+
         with open(file_path, 'r', encoding=encoding) as f:
             reader = csv.reader(f)
-            
-            # Read headers
-            if not self.headers:
-                header_row = next(reader)
-                self.headers = [h.strip() for h in header_row]
-            else:
-                next(reader)  # Skip header if already read
-            
+
+            # Headers must be read fresh for every file: different sources
+            # ship slightly different column sets (e.g. some have msisdn,
+            # others have first_site_name/tbl_dt instead). The reader keeps
+            # the union of headers seen so downstream code can still iterate.
+            file_headers = [h.strip() for h in next(reader)]
+            for h in file_headers:
+                if h not in self.headers:
+                    self.headers.append(h)
+
             for row_num, row in enumerate(reader, start=1):
                 if not row or all(not cell for cell in row):
                     continue
-                
+
                 # Ensure row length matches headers
-                if len(row) < len(self.headers):
-                    row.extend([''] * (len(self.headers) - len(row)))
-                elif len(row) > len(self.headers):
-                    row = row[:len(self.headers)]
-                
+                if len(row) < len(file_headers):
+                    row.extend([''] * (len(file_headers) - len(row)))
+                elif len(row) > len(file_headers):
+                    row = row[:len(file_headers)]
+
                 # Create row dictionary
-                row_dict = dict(zip(self.headers, row))
+                row_dict = dict(zip(file_headers, row))
                 
                 # Store original
                 original_copy = row_dict.copy()
@@ -148,11 +153,21 @@ class TelecomDataReader:
         
         # Add metadata
         cleaned['_original_idx'] = original_idx
-        
-        # Validate MSISDN
+
+        # MSISDN: prefer an explicit column, otherwise derive from the leg
+        # numbers using call_direction. Files (or DataFrames) where msisdn
+        # is absent must still produce a subscriber id so the per-subscriber
+        # detectors and exporters keep working.
         msisdn = cleaned.get('msisdn', '')
         if not self.validator.validate_msisdn(msisdn):
-            return None
+            msisdn = self.validator.derive_msisdn(
+                cleaned.get('call_direction', ''),
+                cleaned.get('calling_no', ''),
+                cleaned.get('called_no', ''),
+            )
+            if not msisdn:
+                return None
+            cleaned['msisdn'] = msisdn
         
         # Validate duration
         cleaned['duration'] = self.validator.validate_duration(cleaned.get('duration'))

telecom_anomaly/main.py

@@ -34,9 +34,20 @@ def main():
     config.output_dir = str(OUTPUT_DIR)
     config.save_pdf_dir = str(SAVE_PDF_DIR)
     
+    # Parse the optional --model / --no-model flag out of argv. Anything
+    # else on the command line is treated as a CSV path (existing behaviour).
+    cli_args = sys.argv[1:]
+    use_model = False
+    if '--model' in cli_args:
+        cli_args.remove('--model')
+        use_model = True
+    if '--no-model' in cli_args:
+        cli_args.remove('--no-model')
+        use_model = False
+
     # Get files
-    if len(sys.argv) > 1:
-        file_paths = sys.argv[1:]
+    if cli_args:
+        file_paths = cli_args
     else:
         print(f"\nEnter CSV file paths (or place files in {INPUT_DIR}):")
         print("Press Enter to scan input directory, or type paths:")
@@ -74,22 +85,23 @@ def main():
         return
     
     print(f"\n📊 Processing {len(valid_files)} file(s)...")
-    
+    print(f"   Model-based detection: {'ON (Isolation Forest)' if use_model else 'OFF'}")
+
     # Run detection
     try:
         detector = TelecomAnomalyDetector(config)
         detector.load_data(valid_files)
-        anomalies = detector.detect_anomalies()
-        
+        anomalies = detector.detect_anomalies(use_model=use_model)
+
         # Export results
         outputs = detector.export_results()
-        
+
         # Prepare results for PDF (enriched records)
         pdf_results = []
         requested_fields = [
-            "event_date", "event_time", "call_direction", "usage_sub_type", 
-            "calling_no", "called_no", "usage_type", "duration", "imei", 
-            "location_id", "region", "district", "city", "msisdn", 
+            "event_date", "event_time", "call_direction",
+            "calling_no", "called_no", "usage_type", "duration", "imei",
+            "location_id", "region", "district", "city",
             "longitude", "latitude", "azimuth"
         ]
         for idx, row in enumerate(detector.original_data):

telecom_anomaly/validation/validator.py

@@ -23,6 +23,32 @@ class DataValidator:
             return True
         # Ghana format: starts with 233 and 12 digits
         return msisdn_str.startswith('233') and len(msisdn_str) == 12
+
+    @staticmethod
+    def derive_msisdn(call_direction: str, calling_no: str, called_no: str) -> str:
+        """
+        Derive the subscriber's MSISDN from the leg numbers when the CSV
+        does not provide an explicit msisdn column.
+
+        Outgoing legs -> subscriber is the caller; Incoming legs -> the callee.
+        Falls back to whichever side is a valid Ghana MSISDN.
+        """
+        direction = (call_direction or '').upper()
+        calling = (str(calling_no).strip() if calling_no is not None else '')
+        called = (str(called_no).strip() if called_no is not None else '')
+
+        if 'OUT' in direction or 'MO' in direction:
+            primary, secondary = calling, called
+        elif 'IN' in direction or 'MT' in direction:
+            primary, secondary = called, calling
+        else:
+            primary, secondary = calling, called
+
+        if DataValidator.validate_msisdn(primary):
+            return primary
+        if DataValidator.validate_msisdn(secondary):
+            return secondary
+        return ''
     
     @staticmethod
     def validate_duration(duration: Any) -> float: