Deploy FastAPI Telecom Anomaly Detection to Hugging Face

.dockerignore

@@ -0,0 +1,19 @@
+.git
+.gitignore
+.venv
+.python-version
+__pycache__
+*.pyc
+*.pyo
+*.pyd
+.pytest_cache
+data/
+logs/
+uv.lock
+pyproject.toml
+README.md
+Task.txt
+Task2.txt
+Task3.txt
+error.txt
+tests/

.gitignore

@@ -0,0 +1,10 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv

.python-version

@@ -0,0 +1 @@
+3.14

Dockerfile

@@ -0,0 +1,30 @@
+# Use official Python slim image
+FROM python:3.9-slim
+
+# Install Java Runtime Environment (REQUIRED for tabula-py to process PDFs)
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends default-jre && \
+    apt-get clean && \
+    rm -rf /var/lib/apt/lists/*
+
+# Set working directory
+WORKDIR /app
+
+# Set PYTHONPATH to ensure imports work correctly
+ENV PYTHONPATH=/app
+
+# Copy requirements and install
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application
+COPY . .
+
+# Create necessary directories (though the app is configured not to save, logic uses them)
+RUN mkdir -p data/raw_csv_temp data/output_process_data data/strictly_cleaned_data data/output data/save_pdf logs
+
+# Expose port 7860 (Hugging Face default)
+EXPOSE 7860
+
+# Command to run the API
+CMD ["uvicorn", "telecom_anomaly.api:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,10 +1,168 @@
+# 📱 Telecom Anomaly Detection System
+
+A production-ready anomaly detection system for telecommunications Call Data Records (CDR). This system identifies unusual patterns in subscriber behavior, ranging from potential fraud to technical malfunctions.
+
+## 🚀 How it Works
+
+The system operates as an automated pipeline that transforms raw CDR data into actionable security insights:
+
+1.  **Data Ingestion**: Reads CSV files from the `data/input` directory or direct CLI arguments.
+2.  **Validation & Cleaning**: Filters out invalid records, parses dates/times, and categorizes calls (e.g., voice vs. SMS, service vs. subscriber).
+3.  **Multi-Layered Detection**: Parallel analysis of each record across multiple specialized detector modules.
+4.  **Anomaly Aggregation**: Combines findings per subscriber and per record, calculating confidence scores and eliminating duplicates.
+5.  **Reporting**: Generates multiple output formats for different stakeholders (analysts, systems, executives).
+
 ---
-title: Teleprocessinggh
-emoji: 🏢
-colorFrom: pink
-colorTo: yellow
-sdk: docker
-pinned: false
+
+## 🔍 Core Detection Modules
+
+The system uses 6 primary detection strategies:
+
+| Detector | Logic | Use Case |
+| :--- | :--- | :--- |
+| **Duration** | Monitors for "infinite calls" or "flash calls" | Fraud, billing errors |
+| **Frequency** | Tracks calls per minute/hour/day | Automated dialing, SMS flooding |
+| **SMS** | Specifically analyzes SMS volume and rate | Spam, bulk-SMS marketing |
+| **Location** | Calculates velocity between consecutive activities | Account sharing, travel anomalies |
+| **IMEI** | Tracks hardware-to-subscriber changes | SIM swapping, device theft |
+| **Time Pattern** | Identifies activity during unusual (night) hours | Targeted attacks, outlier behavior |
+
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+## 🛠 Architecture
+
+The project follows a modular orchestrator-based architecture:
+
+-   **`TelecomAnomalyDetector` (Core)**: The central brain that coordinates the reader, validator, and all detector modules.
+-   **`TelecomDataReader`**: Handles high-performance batch reading of large CSV files.
+-   **`DataValidator`**: Ensures MSISDN formats, geographical coordinates, and timestamps are valid.
+-   **`ResultExporter`**: Formats findings into JSON (for APIs), CSV (for Excel), and Markdown (for reports).
+
+---
+
+## 📊 Data Requirements
+
+The system expects CSV files with the following core columns:
+
+-   `msisdn`: Subscriber identification (e.g., `233XXXXXXXXX`)
+-   `usage_type`: `VOICE` or `SMS`
+-   `duration`: Call length in seconds
+-   `latitude` / `longitude`: Geographical coordinates of the activity
+-   `date` / `time`: Activity timestamps (`YYYY-MM-DD` and `HH:MM:SS`)
+-   `imei`: Device hardware identifier
+
+---
+
+## ⚙️ Configuration
+
+Detection sensitivity can be tuned in `config/settings.py` or via `TelecomConfig` overrides:
+
+```python
+DEFAULT_THRESHOLDS = {
+    'extreme_call_threshold': 10800.0,  # 3 hours
+    'max_calls_per_hour': 60,           # >1 call/minute
+    'max_location_change_speed': 1000.0,# km/h (impossible travel)
+    'night_hours_start': 23,            # Start of "unusual" hours
+}
+```
+
+---
+
+## 💻 Installation & Usage
+
+### Setup with uv
+The project uses [uv](https://github.com/astral-sh/uv) for high-performance dependency management.
+
+```bash
+# Clone and enter the repository
+git clone https://github.com/yourusername/telecom-anomaly-detection.git
+cd telecom-anomaly-detection
+
+# Sync dependencies and create virtual environment
+uv sync
+```
+
+### Running the Analysis
+Place your CSV files in `data/input/` and run using `uv run`:
+```bash
+uv run python -m telecom_anomaly.main
+```
+
+Or specify files directly:
+```bash
+uv run python -m telecom_anomaly.main data/input/records_2023_10.csv
+```
+
+### Development
+```bash
+# Run tests
+uv run pytest
+
+# Format code
+uv run black .
+
+# Lint code
+uv run flake8 .
+```
+
+## 🌐 API (FastAPI)
+
+The project includes a FastAPI-based REST API for programmatic access to the anomaly detection engine.
+
+### Running the API
+```bash
+uv run python -m uvicorn telecom_anomaly.api:app --reload
+```
+The API will be available at `http://127.0.0.1:8000`.
+
+### API Documentation
+Interactive documentation is available at:
+- Swagger UI: `http://127.0.0.1:8000/docs`
+- ReDoc: `http://127.0.0.1:8000/redoc`
+
+### Key Endpoints
+
+#### `POST /analyze`
+Upload a CDR CSV file for real-time analysis.
+
+**Example using curl:**
+```bash
+curl -X 'POST' \
+  'http://127.0.0.1:8000/analyze' \
+  -H 'accept: application/json' \
+  -H 'Content-Type: multipart/form-data' \
+  -F 'file=@data/input/sample_cdr.csv'
+```
+
+**Response Example:**
+```json
+{
+  "filename": "sample_cdr.csv",
+  "statistics": {
+    "total_rows": 1000,
+    "rows_with_anomalies": 15,
+    "total_anomalies_detected": 22,
+    "anomaly_rate": 1.5,
+    "unique_subscribers": 12,
+    "anomaly_types": {
+      "short_duration": 10,
+      "impossible_speed": 5
+    }
+  },
+  "anomalies": [
+    {
+      "row_index": 42,
+      "column_name": "duration",
+      "value": 1.2,
+      "detection_method": "short_duration",
+      "score": 2.5,
+      "threshold": 3.0,
+      "confidence": 0.75,
+      "msisdn": "233123456789"
+    }
+  ]
+}
+```
+
+#### `GET /health`
+Returns the health status of the API.

READ_ME.md

@@ -0,0 +1,10 @@
+---
+title: Teleprocessinggh
+emoji: 🏢
+colorFrom: pink
+colorTo: yellow
+sdk: docker
+pinned: false
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

RunAPI.md

@@ -0,0 +1,78 @@
+# Deployment & API Guide
+
+This guide explains how to push your code to GitHub and deploy the API so you can access the Swagger UI.
+
+## 1. Push to GitHub
+
+1. **Initialize Git** (if not already done):
+   ```bash
+   git init
+   git add .
+   git commit -m "Initial commit: Joined PDF/CSV API with Docker support"
+   ```
+
+2. **Create a Repository** on GitHub (e.g., `telecom-anomaly-detection`).
+
+3. **Link and Push**:
+   ```bash
+   git remote add origin https://github.com/YOUR_USERNAME/YOUR_REPO_NAME.git
+   git branch -M main
+   git push -u origin main
+   ```
+
+---
+
+## 2. Deploying to Vercel (Serverless)
+
+Vercel is primarily for frontend, but you can run the FastAPI backend using their Python runtime.
+
+1. **Install Vercel CLI**: `npm i -g vercel`
+2. **Login**: `vercel login`
+3. **Deploy**: Run `vercel` in the project root.
+4. **Access Swagger**: Once deployed, go to `https://your-project-name.vercel.app/docs`.
+
+**⚠️ Note:** If the PDF processing fails on Vercel due to "Java not found", you MUST use the Docker method below.
+
+---
+
+## 3. Recommended: Deploying with Docker (Railway / Render)
+
+Since your project requires **Java** (for `tabula-py`), using a platform that supports your `Dockerfile` is much more reliable.
+
+### Option A: Railway.app (Easiest)
+1. Go to [Railway.app](https://railway.app/) and login with GitHub.
+2. Click **New Project** -> **Deploy from GitHub repo**.
+3. Select your repository.
+4. Railway will see your `Dockerfile` and deploy it automatically.
+5. Your Swagger UI will be at: `https://your-railway-url.up.railway.app/docs`.
+
+### Option B: Render.com
+1. Go to [Render.com](https://render.com/).
+2. Click **New** -> **Web Service**.
+3. Connect your GitHub and select the repo.
+4. Select **Environment: Docker**.
+5. Click **Deploy**.
+6. Your Swagger UI will be at: `https://your-subdomain.onrender.com/docs`.
+
+---
+
+## 4. Accessing Swagger UI Locally
+
+To test everything locally before pushing:
+
+1. **Run with Uvicorn**:
+   ```bash
+   uvicorn telecom_anomaly.api:app --reload
+   ```
+2. **Open Browser**:
+   - **Interactive Documentation (Swagger)**: [http://127.0.0.1:8000/docs](http://127.0.0.1:8000/docs)
+   - **Alternative Docs (Redoc)**: [http://127.0.0.1:8000/redoc](http://127.0.0.1:8000/redoc)
+
+## 5. Using the Unified Endpoint
+
+- **Endpoint**: `POST /analyze`
+- **Function**: Upload a PDF (or CSV).
+- **Parameters**: 
+    - `save_results`: Set to `true` if you want to save files on the server (Docker only).
+    - `filter_movements`: Set to `true` to clean up geographic data.
+    - All other settings match your original logic.

config/__init__.py

@@ -0,0 +1 @@
+# config package

config/settings.py

@@ -0,0 +1,59 @@
+"""
+Configuration settings for the telecom anomaly detection system
+"""
+
+import os
+from pathlib import Path
+
+# Base paths
+BASE_DIR = Path(__file__).parent.parent
+DATA_DIR = BASE_DIR / 'data'
+INPUT_DIR = DATA_DIR / 'input'
+OUTPUT_DIR = DATA_DIR / 'output'
+SAVE_PDF_DIR = DATA_DIR / 'save_pdf'
+LOG_DIR = BASE_DIR / 'logs'
+
+# Create directories if they don't exist
+# On environments like Vercel, this may fail due to read-only filesystem
+for dir_path in [DATA_DIR, INPUT_DIR, OUTPUT_DIR, SAVE_PDF_DIR, LOG_DIR]:
+    try:
+        dir_path.mkdir(parents=True, exist_ok=True)
+    except Exception as e:
+        print(f"Warning: Could not create directory {dir_path}: {e}")
+
+# Logging configuration
+LOGGING_CONFIG = {
+    'level': 'INFO',
+    'format': '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    'log_file': LOG_DIR / 'telecom_anomaly_detection.log'
+}
+
+# Default detection thresholds (can be overridden by config file)
+DEFAULT_THRESHOLDS = {
+    'short_call_threshold': 3.0,
+    'long_call_threshold': 5400.0,
+    'extreme_call_threshold': 10800.0,
+    'max_calls_per_hour': 60,
+    'max_calls_per_day': 300,
+    'max_calls_per_minute': 10,
+    'max_sms_per_hour': 100,
+    'max_sms_per_minute': 20,
+    'max_location_change_speed': 1000.0,
+    'location_cluster_radius': 10.0,
+    'call_ratio_outgoing_incoming': 20.0,
+    'unique_contacts_per_day': 100,
+    'z_score_threshold': 4.0,
+    'iqr_multiplier': 2.0,
+    'night_hours_start': 23,
+    'night_hours_end': 5,
+    'min_samples_for_pattern': 20,
+    'batch_size': 10000
+}
+
+# Service numbers to ignore
+SERVICE_NUMBERS = [
+    'MTN', 'MOBILEMONEY', 'MTN-DATA', 'MTNXTRATIME', 'MTN-JUST4U',
+    'EVD', '5040', '1506', '431', '1326', '1349', '438840',
+    '54211', '435942', '5381060', '783273271', '141', '402',
+    'TWO SURE LP', 'DRMBAWUMIA', 'MYMTN 2.0', 'EOCO', 'DREAMTRAVEL'
+]

config/telecom_anomaly/core/models.py

@@ -0,0 +1,102 @@
+"""
+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
+    
+    # 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
+    
+    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]
+    
+    def to_dict(self) -> Dict:
+        """Convert to dictionary"""
+        return asdict(self)

main.py

@@ -0,0 +1,5 @@
+from telecom_anomaly.api import app
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

pyproject.toml

@@ -0,0 +1,29 @@
+[project]
+name = "telecom-anomaly-detection"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.9"
+dependencies = [
+    "fastapi>=0.135.2",
+    "fpdf2>=2.8.7",
+    "jpype1>=1.6.0",
+    "matplotlib>=3.4.0",
+    "numpy>=1.21.0",
+    "pandas>=1.3.0",
+    "python-multipart>=0.0.22",
+    "scikit-learn>=1.0.0",
+    "seaborn>=0.11.0",
+    "tabula-py>=2.10.0",
+    "uvicorn>=0.42.0",
+]
+
+[project.scripts]
+app = "telecom_anomaly.api:app"
+
+[dependency-groups]
+dev = [
+    "black>=21.0.0",
+    "flake8>=3.9.0",
+    "pytest>=6.2.0",
+]

requirements.txt

@@ -0,0 +1,23 @@
+# Core Analysis
+pandas>=1.3.0
+numpy>=1.21.0
+scikit-learn>=1.0.0
+tabula-py>=2.4.0
+
+# API
+fastapi>=0.95.0
+uvicorn>=0.21.0
+python-multipart>=0.0.6
+pydantic>=1.10.0
+
+# Export & PDF
+fpdf2>=2.7.0
+
+# Visualization (if needed for exports)
+matplotlib>=3.4.0
+seaborn>=0.11.0
+
+# Development
+pytest>=6.2.0
+black>=21.0.0
+flake8>=3.9.0

setup.py

@@ -0,0 +1,41 @@
+"""
+Setup configuration for telecom anomaly detection package
+"""
+
+from setuptools import setup, find_packages
+
+with open("README.md", "r", encoding="utf-8") as fh:
+    long_description = fh.read()
+
+setup(
+    name="telecom-anomaly-detection",
+    version="1.0.0",
+    author="Your Name",
+    author_email="your.email@example.com",
+    description="Production-ready anomaly detection system for telecommunications call data records",
+    long_description=long_description,
+    long_description_content_type="text/markdown",
+    url="https://github.com/yourusername/telecom-anomaly-detection",
+    packages=find_packages(),
+    classifiers=[
+        "Development Status :: 5 - Production/Stable",
+        "Intended Audience :: Telecommunications Industry",
+        "License :: OSI Approved :: MIT License",
+        "Operating System :: OS Independent",
+        "Programming Language :: Python :: 3",
+        "Programming Language :: Python :: 3.8",
+        "Programming Language :: Python :: 3.9",
+        "Programming Language :: Python :: 3.10",
+    ],
+    python_requires=">=3.8",
+    install_requires=[
+        "pandas>=1.3.0",
+        "numpy>=1.21.0",
+        "scikit-learn>=1.0.0",
+    ],
+    entry_points={
+        "console_scripts": [
+            "telecom-anomaly=telecom_anomaly.main:main",
+        ],
+    },
+)

telecom_anomaly/__init__.py

@@ -0,0 +1,9 @@
+"""
+Telecom Anomaly Detection System
+"""
+
+from telecom_anomaly.core.detector import TelecomAnomalyDetector
+from telecom_anomaly.core.models import TelecomConfig, AnomalyResult
+
+__version__ = '1.0.0'
+__all__ = ['TelecomAnomalyDetector', 'TelecomConfig', 'AnomalyResult']

telecom_anomaly/api.py

@@ -0,0 +1,167 @@
+import os
+import shutil
+import tempfile
+import logging
+import pandas as pd
+import io
+from pathlib import Path
+from typing import List, Dict, Any, Optional
+from fastapi import FastAPI, UploadFile, File, HTTPException
+
+from telecom_anomaly.core.detector import TelecomAnomalyDetector
+from telecom_anomaly.core.models import TelecomConfig
+from config.settings import SERVICE_NUMBERS, OUTPUT_DIR, SAVE_PDF_DIR
+from telecom_anomaly.utils.pdf_processor import (
+    extract_pdf_to_lines, 
+    process_telecom_data_df, 
+    strict_clean_df
+)
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+    title="Telecom Anomaly Detection API",
+    description="API for detecting anomalies in telecom Call Data Records (CDR)",
+    version="1.0.0"
+)
+
+@app.get("/health")
+def health_check():
+    """Health check endpoint"""
+    return {"status": "healthy", "version": "1.0.0"}
+
+@app.post("/analyze")
+async def analyze_cdr(
+    file: UploadFile = File(...), 
+    save_results: bool = False, 
+    save_pdf: bool = False, 
+    filter_movements: bool = False,
+    run_anomaly_detection: bool = True,
+    cluster_km: float = 1.0
+):
+    """
+    Analyze a PDF or CSV file for telecom anomalies.
+    Settings are matched to the original CSV analysis endpoint.
+    """
+    filename = file.filename.lower()
+    is_pdf = filename.endswith('.pdf')
+    is_csv = filename.endswith('.csv')
+    
+    if not (is_pdf or is_csv):
+        raise HTTPException(status_code=400, detail="Only PDF and CSV files are supported")
+    
+    # Create a temporary file
+    with tempfile.NamedTemporaryFile(delete=False, suffix=Path(file.filename).suffix) as tmp:
+        shutil.copyfileobj(file.file, tmp)
+        tmp_path = tmp.name
+    
+    try:
+        # Step 1: Initialize detector with original settings
+        config = TelecomConfig()
+        config.service_numbers = SERVICE_NUMBERS
+        config.output_dir = str(OUTPUT_DIR) if save_results else None
+        config.save_pdf_dir = str(SAVE_PDF_DIR) if save_pdf else None
+        config.location_cluster_km = cluster_km
+        
+        detector = TelecomAnomalyDetector(config)
+        
+        # Step 2: Load Data
+        if is_pdf:
+            logger.info(f"Processing PDF: {file.filename}")
+            base_name = os.path.splitext(file.filename)[0]
+            
+            # Use original logic via lines -> df -> strict_clean
+            raw_lines = extract_pdf_to_lines(tmp_path)
+            if not raw_lines:
+                raise HTTPException(status_code=500, detail="Failed to extract tables from PDF")
+                
+            df_processed = process_telecom_data_df(raw_lines, base_name)
+            if df_processed is None:
+                raise HTTPException(status_code=500, detail="Failed to process extracted telecom data")
+                
+            df_final = strict_clean_df(df_processed)
+            if df_final is None or df_final.empty:
+                raise HTTPException(status_code=500, detail="No valid records found after cleaning")
+            
+            # Load into detector
+            detector.load_dataframe(df_final, filter_movements=filter_movements)
+        else:
+            logger.info(f"Processing CSV: {file.filename}")
+            # Use original CSV loading logic
+            detector.load_data([tmp_path], filter_movements=filter_movements)
+        
+        # Step 3: Run Analysis
+        anomalies = []
+        if run_anomaly_detection:
+            anomalies = detector.detect_anomalies()
+        
+        # Get statistics
+        stats = detector.get_statistics()
+        
+        # Step 4: Prepare Results (Exactly like original)
+        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}
+            
+            # Format dates/times for JSON parity
+            if hasattr(record['event_date'], 'isoformat'):
+                record['event_date'] = record['event_date'].isoformat()
+            if hasattr(record['event_time'], 'isoformat'):
+                record['event_time'] = record['event_time'].isoformat()
+
+            record.update({
+                "has_anomaly": "YES" if methods else "NO",
+                "anomaly_count": len(methods),
+                "anomaly_methods": "; ".join(methods)
+            })
+            final_records_to_return.append(record)
+        
+        saved_files = []
+        if save_results and detector.exporter:
+            outputs = detector.export_results(only_anomalies=run_anomaly_detection)
+            saved_files.extend([Path(p).name for p in outputs.values()])
+            
+        if save_pdf and detector.pdf_exporter:
+            pdf_path = detector.export_pdf(final_records_to_return, file.filename)
+            saved_files.append(Path(pdf_path).name)
+        
+        return {
+            "filename": file.filename,
+            "statistics": stats.to_dict() if run_anomaly_detection else {"total_rows": len(detector.data)},
+            "results": final_records_to_return,
+            "saved_files": saved_files
+        }
+        
+    except Exception as e:
+        logger.error(f"Analysis failed: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=str(e))
+    
+    finally:
+        if os.path.exists(tmp_path):
+            try:
+                os.remove(tmp_path)
+            except:
+                pass
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

telecom_anomaly/core/detector.py

@@ -0,0 +1,312 @@
+"""
+Main anomaly detector orchestrator
+"""
+
+import logging
+import numpy as np
+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
+
+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
+        
+        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()
+        
+        # Initialize detectors
+        self.detectors = [
+            DurationDetector(self.config),
+            FrequencyDetector(self.config),
+            SMSDetector(self.config),
+            LocationDetector(self.config),
+            # IMEIDetector(self.config),
+            TimePatternDetector(self.config)
+        ]
+        
+        # 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)
+        self.subscriber_anomalies: Dict[str, Dict] = defaultdict(
+            lambda: {'total': 0, 'methods': set()}
+        )
+        
+        logger.info(f"Initialized TelecomAnomalyDetector with {len(self.detectors)} detectors")
+    
+    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) -> List[AnomalyResult]:
+        """Run all anomaly detectors"""
+        start_time = datetime.now()
+        logger.info("Starting anomaly detection")
+        
+        all_anomalies = []
+        
+        # Run each detector
+        for detector in self.detectors:
+            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)
+            
+            # 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
+        
+        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)
+        )
+    
+    def export_results(self, format: str = 'json', only_anomalies: bool = True) -> Dict[str, str]:
+        """Export all results"""
+        stats = self.get_statistics()
+        anomaly_dicts = [a.to_dict() for a in self.anomalies]
+        
+        outputs = {}
+        
+        # Export trackable data
+        outputs['trackable'] = self.exporter.export_trackable_data(
+            self.original_data, self.headers, self.row_anomaly_map,
+            only_anomalies=only_anomalies
+        )
+        
+        # Export JSON results
+        outputs['json'] = self.exporter.export_results(
+            anomaly_dicts, stats.to_dict(), 'json'
+        )
+        
+        # Export CSV results
+        outputs['csv'] = self.exporter.export_results(
+            anomaly_dicts, stats.to_dict(), 'csv'
+        )
+        
+        # Export report
+        outputs['report'] = self.exporter.export_report(stats.to_dict())
+        
+        # Export suspicious subscribers
+        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"""
+        stats = self.get_statistics().to_dict()
+        return self.pdf_exporter.generate_report(stats, results, filename)

telecom_anomaly/core/models.py

@@ -0,0 +1,107 @@
+"""
+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]
+    
+    def to_dict(self) -> Dict:
+        """Convert to dictionary"""
+        return asdict(self)

telecom_anomaly/detection/base.py

@@ -0,0 +1,65 @@
+"""
+Base classes for anomaly detectors
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any, Optional
+from collections import defaultdict
+
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class BaseDetector(ABC):
+    """Base class for all anomaly detectors"""
+    
+    def __init__(self, config: TelecomConfig):
+        self.config = config
+        self.name = self.__class__.__name__
+    
+    @abstractmethod
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect anomalies in the data"""
+        pass
+    
+    def _create_anomaly(self, 
+                        row_index: int,
+                        column_name: str,
+                        value: Any,
+                        method: str,
+                        score: float,
+                        threshold: float,
+                        confidence: float,
+                        msisdn: str = "",
+                        metadata: Optional[Dict] = None) -> AnomalyResult:
+        """Helper method to create an anomaly result"""
+        return AnomalyResult(
+            row_index=row_index,
+            column_name=column_name,
+            value=value,
+            detection_method=method,
+            score=score,
+            threshold=threshold,
+            confidence=confidence,
+            msisdn=msisdn,
+            metadata=metadata or {}
+        )
+
+
+class SubscriberBasedDetector(BaseDetector):
+    """Base class for detectors that operate on subscriber data"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+        self.subscriber_data: Dict[str, List[Dict]] = defaultdict(list)
+    
+    def prepare_data(self, data: List[Dict]):
+        """Prepare data by grouping by subscriber"""
+        self.subscriber_data.clear()
+        for row in data:
+            msisdn = row.get('msisdn')
+            if msisdn:
+                self.subscriber_data[msisdn].append(row)
+    
+    @abstractmethod
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        pass

telecom_anomaly/detection/duration.py

@@ -0,0 +1,70 @@
+"""
+Call duration anomaly detector
+"""
+
+from typing import List, Dict, Any
+
+from telecom_anomaly.detection.base import BaseDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class DurationDetector(BaseDetector):
+    """Detect anomalous call durations"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+    
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect duration anomalies"""
+        anomalies = []
+        
+        for idx, row in enumerate(data):
+            # Skip non-voice, service calls, or zero duration
+            if row.get('usage_type') != 'VOICE' or \
+               row.get('is_service') or \
+               row.get('duration', 0) == 0:
+                continue
+            
+            duration = row['duration']
+            msisdn = row.get('msisdn', '')
+            
+            # Extreme duration
+            if duration > self.config.extreme_call_threshold:
+                anomalies.append(self._create_anomaly(
+                    row_index=idx,
+                    column_name='duration',
+                    value=duration,
+                    method='extreme_duration',
+                    score=min(duration / self.config.extreme_call_threshold, 10.0),
+                    threshold=self.config.extreme_call_threshold,
+                    confidence=0.95,
+                    msisdn=msisdn
+                ))
+            
+            # Long duration
+            elif duration > self.config.long_call_threshold:
+                anomalies.append(self._create_anomaly(
+                    row_index=idx,
+                    column_name='duration',
+                    value=duration,
+                    method='long_duration',
+                    score=min(duration / self.config.long_call_threshold, 5.0),
+                    threshold=self.config.long_call_threshold,
+                    confidence=0.80,
+                    msisdn=msisdn
+                ))
+            
+            # Short duration
+            elif 0 < duration < self.config.short_call_threshold:
+                anomalies.append(self._create_anomaly(
+                    row_index=idx,
+                    column_name='duration',
+                    value=duration,
+                    method='short_duration',
+                    score=self.config.short_call_threshold / max(duration, 0.1),
+                    threshold=self.config.short_call_threshold,
+                    confidence=0.75,
+                    msisdn=msisdn
+                ))
+        
+        return anomalies

telecom_anomaly/detection/frequency.py

@@ -0,0 +1,59 @@
+"""
+Call frequency anomaly detector
+"""
+
+from typing import List, Dict, Any
+from collections import defaultdict
+
+from telecom_anomaly.detection.base import SubscriberBasedDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class FrequencyDetector(SubscriberBasedDetector):
+    """Detect abnormal call frequency"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+    
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect frequency anomalies"""
+        self.prepare_data(data)
+        anomalies = []
+        
+        for msisdn, calls in self.subscriber_data.items():
+            # Filter for voice calls with valid datetime
+            voice_calls = [
+                (idx, call) for idx, call in enumerate(data)
+                if call.get('msisdn') == msisdn and
+                call.get('usage_type') == 'VOICE' and
+                not call.get('is_service') and
+                call.get('datetime')
+            ]
+            
+            if len(voice_calls) < 10:
+                continue
+            
+            # Sort by datetime
+            voice_calls.sort(key=lambda x: x[1]['datetime'])
+            
+            # Check for call bursts (calls per minute)
+            for i in range(len(voice_calls) - self.config.max_calls_per_minute):
+                time_window = (
+                    voice_calls[i + self.config.max_calls_per_minute - 1][1]['datetime'] - 
+                    voice_calls[i][1]['datetime']
+                )
+                
+                if time_window and time_window.total_seconds() < 60:
+                    anomalies.append(self._create_anomaly(
+                        row_index=voice_calls[i][0],
+                        column_name='frequency',
+                        value=self.config.max_calls_per_minute,
+                        method='call_burst',
+                        score=1.0,
+                        threshold=self.config.max_calls_per_minute,
+                        confidence=0.85,
+                        msisdn=msisdn
+                    ))
+                    break  # One burst per subscriber is enough
+        
+        return anomalies

telecom_anomaly/detection/location.py

@@ -0,0 +1,78 @@
+"""
+Location jump anomaly detector
+"""
+
+from typing import List, Dict, Any, Optional
+from datetime import datetime
+
+from telecom_anomaly.detection.base import SubscriberBasedDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+from telecom_anomaly.utils.geo import haversine_distance, calculate_speed
+
+
+class LocationDetector(SubscriberBasedDetector):
+    """Detect impossible location jumps"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+    
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect location anomalies"""
+        self.prepare_data(data)
+        anomalies = []
+        
+        for msisdn, calls in self.subscriber_data.items():
+            # Get calls with valid location and datetime
+            valid_calls = []
+            for call in calls:
+                idx = data.index(call) if call in data else -1
+                if (call.get('latitude') and call.get('longitude') and 
+                    not call.get('is_service') and 
+                    call.get('datetime') and idx >= 0):
+                    valid_calls.append((idx, call))
+            
+            if len(valid_calls) < 3:
+                continue
+            
+            # Sort by datetime
+            valid_calls.sort(key=lambda x: x[1]['datetime'])
+            
+            # Check consecutive calls
+            for i in range(len(valid_calls) - 1):
+                idx1, call1 = valid_calls[i]
+                idx2, call2 = valid_calls[i + 1]
+                
+                dt1: Optional[datetime] = call1.get('datetime')
+                dt2: Optional[datetime] = call2.get('datetime')
+                
+                if dt1 and dt2:
+                    time_diff = (dt2 - dt1).total_seconds() / 3600  # hours
+                    
+                    # Check calls within 24 hours
+                    if 0 < time_diff < 24:
+                        distance = haversine_distance(
+                            call1['latitude'], call1['longitude'],
+                            call2['latitude'], call2['longitude']
+                        )
+                        
+                        if distance > 100:  # More than 100km
+                            speed = calculate_speed(distance, time_diff)
+                            
+                            if speed > self.config.max_location_change_speed:
+                                anomalies.append(self._create_anomaly(
+                                    row_index=idx2,
+                                    column_name='location',
+                                    value=distance,
+                                    method='impossible_speed',
+                                    score=min(speed / self.config.max_location_change_speed, 10.0),
+                                    threshold=self.config.max_location_change_speed,
+                                    confidence=0.90,
+                                    msisdn=msisdn,
+                                    metadata={
+                                        'speed': speed,
+                                        'distance': distance,
+                                        'time_diff_hours': time_diff
+                                    }
+                                ))
+        
+        return anomalies

telecom_anomaly/detection/sms.py

@@ -0,0 +1,55 @@
+"""
+SMS flooding detector
+"""
+
+from typing import List, Dict, Any
+
+from telecom_anomaly.detection.base import SubscriberBasedDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class SMSDetector(SubscriberBasedDetector):
+    """Detect SMS flooding"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+    
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect SMS anomalies"""
+        self.prepare_data(data)
+        anomalies = []
+        
+        for msisdn, sms_list in self.subscriber_data.items():
+            # Filter for SMS with valid datetime
+            sms_events = [
+                (idx, sms) for idx, sms in enumerate(data)
+                if sms.get('msisdn') == msisdn and
+                sms.get('usage_type') == 'SMS' and
+                not sms.get('is_service') and
+                sms.get('datetime')
+            ]
+            
+            if len(sms_events) < 10:
+                continue
+            
+            # Sort by datetime
+            sms_events.sort(key=lambda x: x[1]['datetime'])
+            
+            # Check for SMS bursts (5 in 30 seconds)
+            for i in range(len(sms_events) - 5):
+                time_window = sms_events[i + 4][1]['datetime'] - sms_events[i][1]['datetime']
+                
+                if time_window and time_window.total_seconds() < 30:
+                    anomalies.append(self._create_anomaly(
+                        row_index=sms_events[i][0],
+                        column_name='sms',
+                        value=5,
+                        method='sms_burst',
+                        score=1.0,
+                        threshold=5,
+                        confidence=0.90,
+                        msisdn=msisdn
+                    ))
+                    break  # One burst per subscriber is enough
+        
+        return anomalies

telecom_anomaly/detection/time_patterns.py

@@ -0,0 +1,37 @@
+"""
+Time-based pattern anomaly detector
+"""
+
+from typing import List, Dict, Any
+
+from telecom_anomaly.detection.base import BaseDetector
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class TimePatternDetector(BaseDetector):
+    """Detect unusual time patterns"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+    
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect time-based anomalies"""
+        anomalies = []
+        
+        for idx, row in enumerate(data):
+            # Check for night calls (excluding service)
+            if row.get('is_night') and not row.get('is_service'):
+                # Simple heuristic: flag all night calls
+                # This can be enhanced with statistical models
+                anomalies.append(self._create_anomaly(
+                    row_index=idx,
+                    column_name='time',
+                    value=row.get('event_time', ''),
+                    method='night_call',
+                    score=1.0,
+                    threshold=0.5,
+                    confidence=0.60,
+                    msisdn=row.get('msisdn', '')
+                ))
+        
+        return anomalies

telecom_anomaly/io/exporter.py

@@ -0,0 +1,183 @@
+"""
+Export utilities for anomaly detection results
+"""
+
+import csv
+import json
+import logging
+from datetime import datetime
+from typing import List, Dict, Any, Optional
+from pathlib import Path
+
+logger = logging.getLogger(__name__)
+
+
+class ResultExporter:
+    """Export anomaly detection results in various formats"""
+    
+    def __init__(self, output_dir: Optional[str] = None):
+        self.output_dir = Path(output_dir) if output_dir else Path.cwd()
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+    
+    def export_trackable_data(self, 
+                              original_data: List[Dict[str, Any]],
+                              headers: List[str],
+                              row_anomaly_map: Dict[int, List[str]],
+                              output_path: Optional[str] = None,
+                              only_anomalies: bool = True) -> str:
+        """Export original data with anomaly flags"""
+        
+        if not output_path:
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            output_path = self.output_dir / f"telecom_data_results_{timestamp}.csv"
+        
+        logger.info(f"Generating trackable data: {output_path}")
+        
+        # Prepare headers
+        trackable_headers = headers.copy()
+        trackable_headers.extend(['has_anomaly', 'anomaly_count', 'anomaly_methods'])
+        
+        with open(output_path, 'w', newline='', encoding='utf-8') as f:
+            writer = csv.writer(f)
+            writer.writerow(trackable_headers)
+            
+            exported_rows = 0
+            
+            for original_idx, original_row in enumerate(original_data):
+                anomaly_methods = row_anomaly_map.get(original_idx, [])
+                
+                # If only_anomalies is True, only export rows that HAVE anomalies
+                if only_anomalies and not anomaly_methods:
+                    continue
+                
+                exported_rows += 1
+                has_anomaly = 'YES' if anomaly_methods else 'NO'
+                
+                # Get original data in header order
+                row_data = [original_row.get(h, '') for h in headers]
+                
+                # Add anomaly columns
+                row_data.extend([
+                    has_anomaly,
+                    len(anomaly_methods),
+                    '; '.join(anomaly_methods[:3])
+                ])
+                
+                writer.writerow(row_data)
+            
+            logger.info(f"Exported {exported_rows} records to CSV")
+        
+        return str(output_path)
+        
+        return str(output_path)
+    
+    def export_results(self,
+                       anomalies: List[Dict],
+                       statistics: Dict[str, Any],
+                       format: str = 'json',
+                       output_path: Optional[str] = None) -> str:
+        """Export results in specified format"""
+        
+        if not output_path:
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            output_path = self.output_dir / f"telecom_anomaly_results_{timestamp}.{format}"
+        
+        if format == 'json':
+            self._export_json(anomalies, statistics, output_path)
+        elif format == 'csv':
+            self._export_csv(anomalies, output_path)
+        else:
+            raise ValueError(f"Unsupported format: {format}")
+        
+        logger.info(f"Results exported to {output_path}")
+        return str(output_path)
+    
+    def _export_json(self, anomalies: List[Dict], statistics: Dict[str, Any], output_path: Path):
+        """Export as JSON"""
+        data = {
+            'metadata': {
+                'timestamp': datetime.now().isoformat(),
+                'statistics': statistics
+            },
+            'anomalies': anomalies
+        }
+        
+        with open(output_path, 'w') as f:
+            json.dump(data, f, indent=2)
+    
+    def _export_csv(self, anomalies: List[Dict], output_path: Path):
+        """Export as CSV"""
+        with open(output_path, 'w', newline='') as f:
+            writer = csv.writer(f)
+            writer.writerow(['row_index', 'msisdn', 'method', 'value', 'confidence', 'column'])
+            
+            for a in anomalies:
+                if a.get('row_index', -1) >= 0:
+                    writer.writerow([
+                        a.get('row_index', ''),
+                        a.get('msisdn', ''),
+                        a.get('detection_method', ''),
+                        f"{a.get('value', 0):.2f}" if isinstance(a.get('value'), (int, float)) else a.get('value', ''),
+                        f"{a.get('confidence', 0):.2f}",
+                        a.get('column_name', '')
+                    ])
+    
+    def export_report(self,
+                      statistics: Dict[str, Any],
+                      output_path: Optional[str] = None) -> str:
+        """Generate human-readable report"""
+        
+        if not output_path:
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            output_path = self.output_dir / f"telecom_anomaly_report_{timestamp}.txt"
+        
+        with open(output_path, 'w') as f:
+            f.write("=" * 60 + "\n")
+            f.write("TELECOM ANOMALY DETECTION REPORT\n")
+            f.write("=" * 60 + "\n\n")
+            
+            f.write(f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
+            
+            f.write("SUMMARY\n")
+            f.write("-" * 40 + "\n")
+            f.write(f"Total Records: {statistics['total_rows']:,}\n")
+            f.write(f"Rows with Anomalies: {statistics['rows_with_anomalies']:,}\n")
+            f.write(f"Total Anomalies: {statistics['total_anomalies_detected']:,}\n")
+            f.write(f"Anomaly Rate: {statistics['anomaly_rate']:.2f}%\n")
+            f.write(f"Unique Subscribers: {statistics['unique_subscribers']:,}\n\n")
+            
+            f.write("ANOMALIES BY TYPE\n")
+            f.write("-" * 40 + "\n")
+            for method, count in sorted(statistics['anomaly_types'].items(), 
+                                       key=lambda x: x[1], reverse=True):
+                f.write(f"  {method}: {count}\n")
+            
+            f.write("\n" + "=" * 60 + "\n")
+        
+        logger.info(f"Report generated: {output_path}")
+        return str(output_path)
+    
+    def export_suspicious_subscribers(self,
+                                     subscriber_anomalies: Dict[str, Dict],
+                                     output_path: Optional[str] = None) -> str:
+        """Export suspicious subscribers"""
+        
+        if not output_path:
+            timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+            output_path = self.output_dir / f"suspicious_subscribers_{timestamp}.csv"
+        
+        with open(output_path, 'w', newline='') as f:
+            writer = csv.writer(f)
+            writer.writerow(['msisdn', 'anomaly_count', 'unique_methods', 'methods'])
+            
+            for msisdn, data in sorted(subscriber_anomalies.items(), 
+                                      key=lambda x: x[1]['total'], reverse=True):
+                writer.writerow([
+                    msisdn,
+                    data['total'],
+                    len(data['methods']),
+                    ', '.join(data['methods'])
+                ])
+        
+        logger.info(f"Suspicious subscribers exported to {output_path}")
+        return str(output_path)

telecom_anomaly/io/pdf_exporter.py

@@ -0,0 +1,107 @@
+from fpdf import FPDF
+from datetime import datetime
+from pathlib import Path
+from typing import Dict, Any, List
+
+class PDFExporter:
+    """Generates professional Landscape PDF reports for telecom anomaly detection"""
+    
+    def __init__(self, output_dir: str):
+        self.output_dir = Path(output_dir)
+        self.output_dir.mkdir(parents=True, exist_ok=True)
+
+    def generate_report(self, stats: Dict[str, Any], results: List[Dict[str, Any]], filename: str) -> str:
+        # Use Landscape orientation to fit 20 columns
+        pdf = FPDF(orientation="L", unit="mm", format="A4")
+        pdf.add_page()
+        
+        # Title
+        pdf.set_font("helvetica", "B", 18)
+        pdf.set_text_color(0, 51, 102)
+        pdf.cell(0, 12, "Telecom Anomaly Detection - Detailed Report", ln=True, align="C")
+        
+        # Metadata
+        pdf.set_font("helvetica", "I", 9)
+        pdf.set_text_color(100, 100, 100)
+        pdf.cell(0, 8, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} | Source: {filename}", ln=True, align="C")
+        pdf.ln(5)
+        
+        # Summary Section
+        pdf.set_font("helvetica", "B", 12)
+        pdf.set_text_color(0, 0, 0)
+        pdf.set_fill_color(240, 240, 240)
+        pdf.cell(0, 8, " 1. Detection Summary", ln=True, fill=True)
+        pdf.ln(3)
+        
+        pdf.set_font("helvetica", "", 10)
+        summary_text = (f"Total Records: {stats['total_rows']:,} | "
+                       f"Anomalies: {stats['rows_with_anomalies']:,} | "
+                       f"Subscribers: {stats['unique_subscribers']:,} | "
+                       f"Rate: {stats['anomaly_rate']:.2f}%")
+        pdf.cell(0, 8, summary_text, ln=True)
+        pdf.ln(5)
+        
+        # Detailed Table
+        pdf.set_font("helvetica", "B", 12)
+        pdf.cell(0, 8, " 2. Detailed Anomalous Records (All 20 Fields)", ln=True, fill=True)
+        pdf.ln(3)
+        
+        # Define 20 columns and their widths (Landscape A4 width is ~277mm inside margins)
+        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),
+            ("Long", "longitude", 14),
+            ("Lat", "latitude", 14),
+            ("Azi", "azimuth", 8),
+            ("Anom", "has_anomaly", 8),
+            ("Cnt", "anomaly_count", 6),
+            ("Methods", "anomaly_methods", 0) # 0 means fill remaining width
+        ]
+        
+        # Calculate width of last column
+        used_width = sum(f[2] for f in fields[:-1])
+        fields[-1] = (fields[-1][0], fields[-1][1], 277 - used_width)
+
+        # Header
+        pdf.set_font("helvetica", "B", 5.5) # Small font for many columns
+        pdf.set_fill_color(200, 220, 255)
+        for label, _, width in fields:
+            pdf.cell(width, 7, label, border=1, fill=True, align="C")
+        pdf.ln()
+        
+        # Data Rows
+        pdf.set_font("helvetica", "", 5.5)
+        for record in results[:100]: # Limit to top 100 in PDF for performance
+            # Draw row
+            for _, key, width in fields:
+                val = str(record.get(key, ""))
+                # Truncate if too long for cell
+                if pdf.get_string_width(val) > width - 1:
+                    while pdf.get_string_width(val + "..") > width - 1 and len(val) > 0:
+                        val = val[:-1]
+                    val += ".."
+                
+                pdf.cell(width, 6, val, border=1)
+            pdf.ln()
+
+        if len(results) > 100:
+            pdf.set_font("helvetica", "I", 7)
+            pdf.cell(0, 10, f"Showing top 100 of {len(results)} anomalies. Please refer to CSV/JSON for full dataset.", ln=True, align="R")
+
+        # Save PDF
+        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+        output_path = self.output_dir / f"Anomaly_Report_Detailed_{timestamp}.pdf"
+        pdf.output(str(output_path))
+        return str(output_path)

telecom_anomaly/io/reader.py

@@ -0,0 +1,181 @@
+"""
+Data reader for telecom CSV files
+"""
+
+import csv
+import os
+import logging
+import pandas as pd
+from typing import List, Dict, Any, Optional, Union
+from datetime import datetime
+
+from telecom_anomaly.validation.validator import DataValidator
+
+logger = logging.getLogger(__name__)
+
+
+class TelecomDataReader:
+    """Read and validate telecom data from CSV files"""
+    
+    def __init__(self, config, batch_size: int = 10000):
+        self.config = config
+        self.batch_size = batch_size
+        self.headers: List[str] = []
+        self.validator = DataValidator()
+    
+    def read_files(self, file_paths: Union[str, List[str]], 
+                   encoding: str = 'utf-8-sig') -> List[Dict[str, Any]]:
+        """
+        Read and validate data from CSV files
+        
+        Returns list of cleaned records with original indices
+        """
+        if isinstance(file_paths, str):
+            file_paths = [file_paths]
+        
+        all_records = []
+        original_data = []
+        
+        for file_path in file_paths:
+            if not os.path.exists(file_path):
+                logger.warning(f"File not found: {file_path}")
+                continue
+            
+            logger.info(f"Reading data from {file_path}")
+            
+            try:
+                file_records, file_original = self._read_single_file(
+                    file_path, encoding, len(original_data)
+                )
+                all_records.extend(file_records)
+                original_data.extend(file_original)
+                
+            except Exception as e:
+                logger.error(f"Failed to read {file_path}: {e}")
+        
+        logger.info(f"Read {len(all_records)} valid records from {len(original_data)} total rows")
+        return all_records, original_data
+
+    def read_dataframe(self, df: 'pd.DataFrame') -> List[Dict[str, Any]]:
+        """
+        Read and validate data from a pandas DataFrame
+        
+        Returns list of cleaned records with original indices
+        """
+        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]
+        
+        # Convert DataFrame to records (list of dicts)
+        # We need to handle it row by row to match the original index-based cleaning
+        for idx, row in df.iterrows():
+            # Create row dictionary
+            row_dict = row.to_dict()
+            # Ensure all values are strings or expected types for cleaning
+            row_dict = {k: str(v) if pd.notna(v) else "" for k, v in row_dict.items()}
+            
+            # Store original
+            original_copy = row_dict.copy()
+            original_copy['_file_source'] = "dataframe_input"
+            original_copy['_row_number'] = idx + 1
+            original_data.append(original_copy)
+            
+            # Clean record
+            cleaned = self._clean_record(row_dict, len(original_data) - 1)
+            if cleaned:
+                records.append(cleaned)
+        
+        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, 
+                          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:
+                self.headers = next(reader)
+                self.headers = [h.strip() for h in self.headers]
+            else:
+                next(reader)  # Skip header if already read
+            
+            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)]
+                
+                # Create row dictionary
+                row_dict = dict(zip(self.headers, row))
+                
+                # Store original
+                original_copy = row_dict.copy()
+                original_copy['_file_source'] = os.path.basename(file_path)
+                original_copy['_row_number'] = row_num
+                original_data.append(original_copy)
+                
+                # Clean record
+                cleaned = self._clean_record(row_dict, base_idx + len(original_data) - 1)
+                if cleaned:
+                    records.append(cleaned)
+        
+        return records, original_data
+    
+    def _clean_record(self, row: Dict[str, Any], original_idx: int) -> Optional[Dict[str, Any]]:
+        """Clean and validate a single record"""
+        cleaned = {}
+        
+        # Strip string values
+        for key, value in row.items():
+            cleaned[key] = value.strip() if isinstance(value, str) else value
+        
+        # Add metadata
+        cleaned['_original_idx'] = original_idx
+        
+        # Validate MSISDN
+        msisdn = cleaned.get('msisdn', '')
+        if not self.validator.validate_msisdn(msisdn):
+            return None
+        
+        # Validate duration
+        cleaned['duration'] = self.validator.validate_duration(cleaned.get('duration'))
+        
+        # Validate coordinates
+        lat, lon = self.validator.validate_coordinates(
+            cleaned.get('latitude'), cleaned.get('longitude')
+        )
+        cleaned['latitude'] = lat
+        cleaned['longitude'] = lon
+        
+        # Parse datetime
+        cleaned['datetime'] = self.validator.parse_datetime(
+            cleaned.get('event_date', ''),
+            cleaned.get('event_time', '')
+        )
+        
+        # Check night call
+        cleaned['is_night'] = self.validator.is_night_call(
+            cleaned.get('event_time', ''),
+            self.config.night_hours_start,
+            self.config.night_hours_end
+        )
+        
+        # Check service number
+        calling = cleaned.get('calling_no', '')
+        cleaned['is_service'] = self.validator.is_service_number(
+            calling, self.config.service_numbers
+        )
+        
+        return cleaned

telecom_anomaly/main.py

@@ -0,0 +1,136 @@
+"""
+Main entry point for telecom anomaly detection system
+"""
+
+import sys
+import logging
+from pathlib import Path
+
+from telecom_anomaly import TelecomAnomalyDetector, TelecomConfig
+from config.settings import DEFAULT_THRESHOLDS, SERVICE_NUMBERS, INPUT_DIR, OUTPUT_DIR, SAVE_PDF_DIR
+
+# Configure logging
+from config.settings import LOGGING_CONFIG
+logging.basicConfig(
+    level=LOGGING_CONFIG['level'],
+    format=LOGGING_CONFIG['format'],
+    handlers=[
+        logging.FileHandler(LOGGING_CONFIG['log_file']),
+        logging.StreamHandler(sys.stdout)
+    ]
+)
+logger = logging.getLogger(__name__)
+
+
+def main():
+    """Main function"""
+    print("\n" + "=" * 60)
+    print("📱 TELECOM ANOMALY DETECTION SYSTEM")
+    print("=" * 60)
+    
+    # Create config with defaults
+    config = TelecomConfig()
+    config.service_numbers = SERVICE_NUMBERS
+    config.output_dir = str(OUTPUT_DIR)
+    config.save_pdf_dir = str(SAVE_PDF_DIR)
+    
+    # Get files
+    if len(sys.argv) > 1:
+        file_paths = sys.argv[1:]
+    else:
+        print(f"\nEnter CSV file paths (or place files in {INPUT_DIR}):")
+        print("Press Enter to scan input directory, or type paths:")
+        
+        user_input = input().strip()
+        
+        if not user_input:
+            # Scan input directory
+            file_paths = [str(p) for p in INPUT_DIR.glob('*.csv')]
+            if file_paths:
+                print(f"Found {len(file_paths)} CSV files in input directory")
+            else:
+                print("No CSV files found in input directory")
+                file_paths = []
+        else:
+            file_paths = [user_input]
+            
+            # Get additional files
+            while True:
+                path = input().strip()
+                if not path:
+                    break
+                file_paths.append(path)
+    
+    # Validate files
+    valid_files = []
+    for path in file_paths:
+        if Path(path).exists():
+            valid_files.append(path)
+        else:
+            print(f"❌ File not found: {path}")
+    
+    if not valid_files:
+        print("\n❌ No valid input files. Exiting.")
+        return
+    
+    print(f"\n📊 Processing {len(valid_files)} file(s)...")
+    
+    # Run detection
+    try:
+        detector = TelecomAnomalyDetector(config)
+        detector.load_data(valid_files)
+        anomalies = detector.detect_anomalies()
+        
+        # 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", 
+            "longitude", "latitude", "azimuth"
+        ]
+        for idx, row in enumerate(detector.original_data):
+            methods = detector.row_anomaly_map.get(idx, [])
+            if methods:
+                record = {field: row.get(field, "") for field in requested_fields}
+                record.update({
+                    "has_anomaly": "YES",
+                    "anomaly_count": len(methods),
+                    "anomaly_methods": "; ".join(methods)
+                })
+                pdf_results.append(record)
+        
+        # Generate PDF report
+        pdf_source_name = Path(valid_files[0]).name if valid_files else "Batch_Analysis"
+        pdf_path = detector.export_pdf(pdf_results, pdf_source_name)
+        outputs['pdf_report'] = pdf_path
+        
+        # Show results
+        stats = detector.get_statistics()
+        
+        print("\n" + "=" * 60)
+        print("✅ ANALYSIS COMPLETE")
+        print("=" * 60)
+        print(f"📁 Total records: {stats.total_rows:,}")
+        print(f"👤 Subscribers with anomalies: {stats.unique_subscribers:,}")
+        print(f"🔍 Rows with anomalies: {stats.rows_with_anomalies:,}")
+        print(f"⚠️  Total anomalies: {stats.total_anomalies_detected:,}")
+        print(f"📊 Anomaly rate: {stats.anomaly_rate:.2f}%")
+        print("\n📄 Files created:")
+        for name, path in outputs.items():
+            print(f"  - {name}: {Path(path).name}")
+        print("\n" + "=" * 60)
+        
+    except Exception as e:
+        logger.error(f"Detection failed: {e}", exc_info=True)
+        print(f"\n❌ Error: {e}")
+        return 1
+    
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

telecom_anomaly/telecom_anomaly/detection/base.py

@@ -0,0 +1,65 @@
+"""
+Base classes for anomaly detectors
+"""
+
+from abc import ABC, abstractmethod
+from typing import List, Dict, Any, Optional
+from collections import defaultdict
+
+from telecom_anomaly.core.models import AnomalyResult, TelecomConfig
+
+
+class BaseDetector(ABC):
+    """Base class for all anomaly detectors"""
+    
+    def __init__(self, config: TelecomConfig):
+        self.config = config
+        self.name = self.__class__.__name__
+    
+    @abstractmethod
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        """Detect anomalies in the data"""
+        pass
+    
+    def _create_anomaly(self, 
+                        row_index: int,
+                        column_name: str,
+                        value: Any,
+                        method: str,
+                        score: float,
+                        threshold: float,
+                        confidence: float,
+                        msisdn: str = "",
+                        metadata: Optional[Dict] = None) -> AnomalyResult:
+        """Helper method to create an anomaly result"""
+        return AnomalyResult(
+            row_index=row_index,
+            column_name=column_name,
+            value=value,
+            detection_method=method,
+            score=score,
+            threshold=threshold,
+            confidence=confidence,
+            msisdn=msisdn,
+            metadata=metadata or {}
+        )
+
+
+class SubscriberBasedDetector(BaseDetector):
+    """Base class for detectors that operate on subscriber data"""
+    
+    def __init__(self, config: TelecomConfig):
+        super().__init__(config)
+        self.subscriber_data: Dict[str, List[Dict]] = defaultdict(list)
+    
+    def prepare_data(self, data: List[Dict]):
+        """Prepare data by grouping by subscriber"""
+        self.subscriber_data.clear()
+        for row in data:
+            msisdn = row.get('msisdn')
+            if msisdn:
+                self.subscriber_data[msisdn].append(row)
+    
+    @abstractmethod
+    def detect(self, data: List[Dict], **kwargs) -> List[AnomalyResult]:
+        pass

telecom_anomaly/utils/geo.py

@@ -0,0 +1,56 @@
+"""
+Geographic utilities for location-based anomaly detection
+"""
+
+import math
+from typing import Optional, Tuple
+
+
+def haversine_distance(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """
+    Calculate the great circle distance between two points
+    on the Earth (specified in decimal degrees)
+    
+    Returns distance in kilometers
+    """
+    if None in (lat1, lon1, lat2, lon2):
+        return 0.0
+    
+    # Convert decimal degrees to radians
+    lat1, lon1, lat2, lon2 = map(math.radians, [lat1, lon1, lat2, lon2])
+    
+    # Haversine formula
+    dlat = lat2 - lat1
+    dlon = lon2 - lon1
+    a = math.sin(dlat/2)**2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlon/2)**2
+    c = 2 * math.asin(min(math.sqrt(a), 1.0))
+    
+    # Radius of Earth in kilometers
+    r = 6371
+    return c * r
+
+
+def calculate_speed(distance_km: float, time_hours: float) -> float:
+    """Calculate speed in km/h"""
+    if time_hours <= 0:
+        return 0.0
+    return distance_km / time_hours
+
+
+def is_plausible_location_change(
+    lat1: float, lon1: float, 
+    lat2: float, lon2: float,
+    time_diff_hours: float,
+    max_speed_kmh: float = 1000.0
+) -> Tuple[bool, float, float]:
+    """
+    Check if a location change is plausible
+    
+    Returns: (is_plausible, distance_km, speed_kmh)
+    """
+    distance = haversine_distance(lat1, lon1, lat2, lon2)
+    speed = calculate_speed(distance, time_diff_hours)
+    
+    is_plausible = speed <= max_speed_kmh
+    
+    return is_plausible, distance, speed

telecom_anomaly/utils/helpers.py

@@ -0,0 +1,63 @@
+"""
+Helper utilities
+"""
+
+import hashlib
+from typing import Any, Dict, List, Set
+from collections import defaultdict
+
+
+def generate_row_hash(row: Dict[str, Any]) -> str:
+    """Generate a unique hash for a row"""
+    row_str = ''.join(str(v) for v in row.values() if v is not None)
+    return hashlib.md5(row_str.encode()).hexdigest()
+
+
+def safe_divide(numerator: float, denominator: float, default: float = 0.0) -> float:
+    """Safe division with default value"""
+    try:
+        if denominator == 0:
+            return default
+        return numerator / denominator
+    except (ZeroDivisionError, TypeError, ValueError):
+        return default
+
+
+def chunk_list(lst: List, chunk_size: int):
+    """Split list into chunks"""
+    for i in range(0, len(lst), chunk_size):
+        yield lst[i:i + chunk_size]
+
+
+class SubscriberTracker:
+    """Track subscriber data efficiently"""
+    
+    def __init__(self):
+        self.subscriber_data: Dict[str, List[Dict]] = defaultdict(list)
+        self.subscriber_indices: Dict[str, Set[int]] = defaultdict(set)
+    
+    def add_record(self, msisdn: str, record: Dict, index: int):
+        """Add a record for a subscriber"""
+        self.subscriber_data[msisdn].append(record)
+        self.subscriber_indices[msisdn].add(index)
+    
+    def get_records(self, msisdn: str) -> List[Dict]:
+        """Get all records for a subscriber"""
+        return self.subscriber_data.get(msisdn, [])
+    
+    def get_indices(self, msisdn: str) -> Set[int]:
+        """Get all indices for a subscriber"""
+        return self.subscriber_indices.get(msisdn, set())
+    
+    def get_all_subscribers(self) -> List[str]:
+        """Get all subscriber MSISDNs"""
+        return list(self.subscriber_data.keys())
+    
+    def get_subscriber_count(self) -> int:
+        """Get number of unique subscribers"""
+        return len(self.subscriber_data)
+    
+    def clear(self):
+        """Clear all data"""
+        self.subscriber_data.clear()
+        self.subscriber_indices.clear()

telecom_anomaly/utils/pdf_processor.py

@@ -0,0 +1,159 @@
+import tabula
+import pandas as pd
+import io
+import re
+import os
+import tempfile
+from pathlib import Path
+
+# --- CONFIGURATION (Restored) ---
+DATA_BASE = Path("data")
+RAW_CSV_DIR = DATA_BASE / "raw_csv_temp"
+PROCESS_OUTPUT_DIR = DATA_BASE / "output_process_data"
+STRICT_CLEAN_DIR = DATA_BASE / "strictly_cleaned_data"
+
+def setup_folders():
+    """Ensure all necessary output directories exist."""
+    for folder in [RAW_CSV_DIR, PROCESS_OUTPUT_DIR, STRICT_CLEAN_DIR]:
+        try:
+            if not os.path.exists(folder):
+                os.makedirs(folder)
+        except Exception as e:
+            print(f"Warning: Could not create folder {folder}: {e}")
+
+# --- ORIGINAL FUNCTIONS (Restored) ---
+
+def extract_pdf_to_csv(pdf_path, base_name):
+    """Stage 1: Convert PDF tables to a raw CSV file."""
+    setup_folders()
+    csv_path = os.path.join(RAW_CSV_DIR, f"{base_name}_raw.csv")
+    try:
+        tabula.convert_into(pdf_path, csv_path, output_format="csv", pages="all", lattice=True)
+        return csv_path
+    except Exception as e:
+        print(f"Error extracting {pdf_path}: {e}")
+        return None
+
+def process_telecom_data_complete(input_path, base_name):
+    """Stage 2: Parse raw CSV and repair fragmented rows."""
+    try:
+        with open(input_path, 'r', encoding='utf-8') as f:
+            lines = f.readlines()
+        return _process_lines_logic(lines, base_name, save_files=True)
+    except Exception as e:
+        print(f"Error reading {input_path}: {e}")
+        return None
+
+def strict_clean_data(file_path, base_name):
+    """Stage 3: Final strict cleaning."""
+    df = pd.read_csv(file_path)
+    df_cleaned = _strict_clean_logic(df)
+    
+    setup_folders()
+    final_path = os.path.join(STRICT_CLEAN_DIR, f'{base_name}_strictly_cleaned.csv')
+    df_cleaned.to_csv(final_path, index=False)
+    return final_path
+
+# --- IN-MEMORY VERSIONS (Exact same logic) ---
+
+def extract_pdf_to_lines(pdf_file_path):
+    """Stage 1 (In-Memory): Returns raw CSV lines."""
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".csv") as tmp:
+        tmp_path = tmp.name
+    try:
+        tabula.convert_into(pdf_file_path, tmp_path, output_format="csv", pages="all", lattice=True)
+        with open(tmp_path, 'r', encoding='utf-8') as f:
+            lines = f.readlines()
+        return lines
+    finally:
+        if os.path.exists(tmp_path):
+            os.remove(tmp_path)
+
+def process_telecom_data_df(lines, base_name):
+    """Stage 2 (In-Memory): Returns DataFrame."""
+    return _process_lines_logic(lines, base_name, save_files=False)
+
+def strict_clean_df(df):
+    """Stage 3 (In-Memory): Returns Cleaned DataFrame."""
+    return _strict_clean_logic(df)
+
+# --- SHARED LOGIC (To ensure 100% parity) ---
+
+def _process_lines_logic(lines, base_name, save_files=False):
+    header_idx = -1
+    for i, line in enumerate(lines):
+        if "event_date_time" in line:
+            header_idx = i
+            break
+    
+    if header_idx == -1:
+        return None
+
+    if save_files:
+        setup_folders()
+        customer_text = "".join(lines[:header_idx])
+        df_customer = pd.read_csv(io.StringIO(customer_text))
+        df_customer.columns = [re.sub(r'[\r\n\s]+', ' ', c).strip() for c in df_customer.columns]
+        cust_path = os.path.join(PROCESS_OUTPUT_DIR, f'{base_name}_customer_info.csv')
+        df_customer.to_csv(cust_path, index=False)
+
+    call_text = "".join(lines[header_idx:])
+    df_raw = pd.read_csv(io.StringIO(call_text))
+    df_raw.columns = [c.strip() for c in df_raw.columns]
+    main_col = df_raw.columns[0] 
+    df_raw = df_raw[df_raw[main_col].astype(str).str.contains("event_date_time") == False]
+
+    fixed_rows = []
+    current_row = None
+    for _, row in df_raw.iterrows():
+        val = str(row[main_col]).strip()
+        if re.match(r'^\d{4}-\d{2}-\d{2}', val):
+            if current_row is not None:
+                fixed_rows.append(current_row)
+            current_row = row.copy()
+        elif current_row is not None:
+            current_row[main_col] = (str(current_row[main_col]) + " " + val).replace('\r', '').replace('\n', '')
+            for col in df_raw.columns:
+                if col != main_col and pd.api.types.is_string_dtype(df_raw[col]):
+                    fragment = str(row[col]).strip()
+                    if fragment.lower() != 'nan':
+                        current_row[col] = (str(current_row[col]) + " " + fragment).replace('\r', '').replace('\n', ' ')
+
+    if current_row is not None:
+        fixed_rows.append(current_row)
+        
+    df_calls = pd.DataFrame(fixed_rows)
+    df_calls[main_col] = df_calls[main_col].str.replace(r'[\r\n]+', '', regex=True).str.replace(r'\s+', ' ', regex=True).str.strip()
+    dt_series = pd.to_datetime(df_calls[main_col], errors='coerce')
+    df_calls['event_date'] = dt_series.dt.date
+    df_calls['event_time'] = dt_series.dt.time
+    df_calls.drop(columns=[main_col], inplace=True)
+
+    uppercase_cols = ['district', 'city', 'usage_type', 'usage_sub_type', 'calling_no', 'called_no', 'imei', 'location_id', 'msisdn']
+    for col in df_calls.columns:
+        c_low = col.lower()
+        if df_calls[col].dtype == 'object':
+            df_calls[col] = df_calls[col].astype(str).str.replace(r'[\r\n\s]+', ' ', regex=True).str.strip()
+            if c_low in uppercase_cols:
+                df_calls[col] = df_calls[col].str.upper()
+            elif c_low == 'region':
+                df_calls[col] = df_calls[col].str.title()
+            elif c_low == 'call_direction':
+                df_calls[col] = df_calls[col].str.title().replace({'Incomingmt': 'Incoming (MT)', '\(\(': '(', '\)\)': ')'}, regex=True)
+
+    ordered_cols = ['event_date', 'event_time'] + [c for c in df_calls.columns if c not in ['event_date', 'event_time']]
+    df_calls = df_calls[ordered_cols]
+    
+    if save_files:
+        interim_path = os.path.join(PROCESS_OUTPUT_DIR, f'{base_name}_call_records_final.csv')
+        df_calls.to_csv(interim_path, index=False)
+        return interim_path
+    return df_calls
+
+def _strict_clean_logic(df):
+    if 'duration' in df.columns and df['duration'].dtype == 'object':
+        df['duration'] = df['duration'].str.replace(',', '.').astype(float)
+    df['event_date_copy'] = pd.to_datetime(df['event_date'], errors='coerce')
+    df['event_time_check'] = pd.to_datetime(df['event_time'].astype(str), format='%H:%M:%S', errors='coerce')
+    df_cleaned = df.dropna().copy()
+    return df_cleaned.drop(columns=['event_date_copy', 'event_time_check'])

telecom_anomaly/validation/validator.py

@@ -0,0 +1,78 @@
+"""
+Data validation utilities for telecom data
+"""
+
+from datetime import datetime
+from typing import Optional, Tuple, Any
+import logging
+
+logger = logging.getLogger(__name__)
+
+
+class DataValidator:
+    """Validate and clean telecom data"""
+    
+    @staticmethod
+    def validate_msisdn(msisdn: str) -> bool:
+        """Validate MSISDN format"""
+        if not msisdn:
+            return False
+        msisdn_str = str(msisdn).strip()
+        # Allow service names
+        if msisdn_str in ['MTN', 'MOBILEMONEY', 'MTN-DATA', 'MTNXTRATIME', 'EVD']:
+            return True
+        # Ghana format: starts with 233 and 12 digits
+        return msisdn_str.startswith('233') and len(msisdn_str) == 12
+    
+    @staticmethod
+    def validate_duration(duration: Any) -> float:
+        """Validate and clean duration"""
+        if duration is None or duration == '':
+            return 0.0
+        try:
+            duration_float = float(duration)
+            return max(0.0, duration_float)
+        except (ValueError, TypeError):
+            return 0.0
+    
+    @staticmethod
+    def validate_coordinates(lat: Any, lon: Any) -> Tuple[Optional[float], Optional[float]]:
+        """Validate latitude and longitude"""
+        try:
+            lat_float = float(lat) if lat not in (None, '', '0.0') else None
+            lon_float = float(lon) if lon not in (None, '', '0.0') else None
+            
+            if lat_float is not None and not (-90 <= lat_float <= 90):
+                lat_float = None
+            if lon_float is not None and not (-180 <= lon_float <= 180):
+                lon_float = None
+                
+            return lat_float, lon_float
+        except (ValueError, TypeError):
+            return None, None
+    
+    @staticmethod
+    def parse_datetime(date_str: str, time_str: str) -> Optional[datetime]:
+        """Parse datetime from date and time strings"""
+        try:
+            if date_str and time_str and date_str != 'NULL' and time_str != 'NULL':
+                return datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M:%S")
+        except (ValueError, TypeError) as e:
+            logger.debug(f"Failed to parse datetime: {date_str} {time_str} - {e}")
+        return None
+    
+    @staticmethod
+    def is_night_call(time_str: str, night_start: int = 23, night_end: int = 5) -> bool:
+        """Check if call occurred at night"""
+        try:
+            if not time_str or time_str == 'NULL':
+                return False
+            hour = int(time_str.split(':')[0])
+            return hour >= night_start or hour < night_end
+        except (ValueError, IndexError):
+            return False
+    
+    @staticmethod
+    def is_service_number(calling_no: str, service_numbers: list) -> bool:
+        """Check if number is a service number"""
+        return calling_no in service_numbers

tests/test_validator.py

@@ -0,0 +1,61 @@
+"""
+Tests for data validator
+"""
+
+import unittest
+from datetime import datetime
+from telecom_anomaly.validation.validator import DataValidator
+
+
+class TestDataValidator(unittest.TestCase):
+    
+    def setUp(self):
+        self.validator = DataValidator()
+    
+    def test_validate_msisdn_valid(self):
+        self.assertTrue(self.validator.validate_msisdn('233123456789'))
+        self.assertTrue(self.validator.validate_msisdn('MTN'))
+    
+    def test_validate_msisdn_invalid(self):
+        self.assertFalse(self.validator.validate_msisdn(''))
+        self.assertFalse(self.validator.validate_msisdn('12345'))
+        self.assertFalse(self.validator.validate_msisdn('23312345'))  # Too short
+    
+    def test_validate_duration(self):
+        self.assertEqual(self.validator.validate_duration('120'), 120.0)
+        self.assertEqual(self.validator.validate_duration('-10'), 0.0)
+        self.assertEqual(self.validator.validate_duration(''), 0.0)
+        self.assertEqual(self.validator.validate_duration(None), 0.0)
+    
+    def test_validate_coordinates(self):
+        lat, lon = self.validator.validate_coordinates('5.5', '-0.2')
+        self.assertEqual(lat, 5.5)
+        self.assertEqual(lon, -0.2)
+        
+        lat, lon = self.validator.validate_coordinates('', '')
+        self.assertIsNone(lat)
+        self.assertIsNone(lon)
+        
+        lat, lon = self.validator.validate_coordinates('100', '200')  # Invalid
+        self.assertIsNone(lat)
+        self.assertIsNone(lon)
+    
+    def test_parse_datetime(self):
+        dt = self.validator.parse_datetime('2024-01-01', '12:30:00')
+        self.assertIsNotNone(dt)
+        self.assertEqual(dt.year, 2024)
+        self.assertEqual(dt.month, 1)
+        self.assertEqual(dt.day, 1)
+        
+        dt = self.validator.parse_datetime('NULL', 'NULL')
+        self.assertIsNone(dt)
+    
+    def test_is_night_call(self):
+        self.assertTrue(self.validator.is_night_call('23:30:00', 23, 5))
+        self.assertTrue(self.validator.is_night_call('02:00:00', 23, 5))
+        self.assertFalse(self.validator.is_night_call('14:00:00', 23, 5))
+        self.assertFalse(self.validator.is_night_call('', 23, 5))
+
+
+if __name__ == '__main__':
+    unittest.main()