Upload folder using huggingface_hub

.env.template

@@ -0,0 +1,46 @@
+# ModelX Intelligence Platform - Production Configuration
+# Copy this to .env and fill in your values
+
+# =============================================================================
+# LLM CONFIGURATION
+# =============================================================================
+# Get your free API key from https://console.groq.com
+GROQ_API_KEY=your_groq_api_key_here
+
+# =============================================================================
+# STORAGE CONFIGURATION
+# =============================================================================
+
+# SQLite Cache
+SQLITE_DB_PATH=data/cache/feeds.db
+SQLITE_RETENTION_HOURS=24
+
+# ChromaDB (Semantic Search)
+CHROMADB_PATH=data/chromadb
+CHROMADB_COLLECTION=modelx_feeds
+CHROMADB_SIMILARITY_THRESHOLD=0.85
+CHROMADB_EMBEDDING_MODEL=sentence-transformers/all-MiniLM-L6-v2
+
+# Neo4j Knowledge Graph
+NEO4J_ENABLED=false  # Set to 'true' to enable (requires Docker)
+NEO4J_URI=bolt://localhost:7687
+NEO4J_USER=neo4j
+NEO4J_PASSWORD=modelx2024
+
+# CSV Export
+CSV_EXPORT_DIR=data/feeds
+
+# Deduplication
+EXACT_MATCH_CHARS=120
+
+# =============================================================================
+# API CONFIGURATION
+# =============================================================================
+API_HOST=0.0.0.0
+API_PORT=8000
+API_WORKERS=1
+
+# =============================================================================
+# FRONTEND CONFIGURATION
+# =============================================================================
+NEXT_PUBLIC_API_URL=http://localhost:8000

.gitignore

@@ -9,15 +9,59 @@ wheels/
 # Virtual environments
 .venv
 
-.env 
-.env.template
+# Environment files
+.env
 
+# LangGraph
+.langgraph_api
 
-#Data & Models (Un-ignored for Hackathon Demo persistence)
-models/
+# =============================================================================
+# ML MODEL ARTIFACTS (Large files - don't push to Git)
+# =============================================================================
+
+# Trained model weights (large binary files)
+*.h5
+*.hdf5
+*.joblib
+*.pkl
+*.pickle
+*.pt
+*.pth
+*.onnx
+*.pb
+
+# Model output directories
+models/*/artifacts/models/
+models/*/output/
+models/*/models_cache/
+models/*/checkpoints/
+
+# Airflow local state
+models/*/.astro/
+
+# MLflow artifacts (tracked separately)
+mlruns/
+mlartifacts/
+
+# =============================================================================
+# DATA FILES (Can be large)
+# =============================================================================
 data/
 datasets/
-data 
-datasets
-models
-.langgraph_api
+
+# Database files
+*.db
+*.sqlite
+*.sqlite3
+
+# ChromaDB persistence (can be large)
+chroma_db/
+
+# =============================================================================
+# KEEP THESE (source code, configs)
+# =============================================================================
+# The models/ folders themselves ARE tracked for:
+# - main.py, src/, dags/ (pipeline code)
+# - requirements.txt, setup.py (dependencies)
+# - data_schema/ (validation configs)
+# - README.md (documentation)

models/anomaly-detection/.dockerignore

@@ -0,0 +1,8 @@
+astro
+.git
+.env
+airflow_settings.yaml
+logs/
+.venv
+airflow.db
+airflow.cfg

models/anomaly-detection/.gitignore

@@ -0,0 +1,11 @@
+.git
+.env
+.DS_Store
+airflow_settings.yaml
+__pycache__/
+astro
+.venv
+airflow-webserver.pid
+webserver_config.py
+airflow.cfg
+airflow.db

models/anomaly-detection/Dockerfile

@@ -0,0 +1 @@
+FROM astrocrpublic.azurecr.io/runtime:3.1-7

models/anomaly-detection/README.md

@@ -0,0 +1,45 @@
+Overview
+========
+
+Welcome to Astronomer! This project was generated after you ran 'astro dev init' using the Astronomer CLI. This readme describes the contents of the project, as well as how to run Apache Airflow on your local machine.
+
+Project Contents
+================
+
+Your Astro project contains the following files and folders:
+
+- dags: This folder contains the Python files for your Airflow DAGs. By default, this directory includes one example DAG:
+    - `example_astronauts`: This DAG shows a simple ETL pipeline example that queries the list of astronauts currently in space from the Open Notify API and prints a statement for each astronaut. The DAG uses the TaskFlow API to define tasks in Python, and dynamic task mapping to dynamically print a statement for each astronaut. For more on how this DAG works, see our [Getting started tutorial](https://www.astronomer.io/docs/learn/get-started-with-airflow).
+- Dockerfile: This file contains a versioned Astro Runtime Docker image that provides a differentiated Airflow experience. If you want to execute other commands or overrides at runtime, specify them here.
+- include: This folder contains any additional files that you want to include as part of your project. It is empty by default.
+- packages.txt: Install OS-level packages needed for your project by adding them to this file. It is empty by default.
+- requirements.txt: Install Python packages needed for your project by adding them to this file. It is empty by default.
+- plugins: Add custom or community plugins for your project to this file. It is empty by default.
+- airflow_settings.yaml: Use this local-only file to specify Airflow Connections, Variables, and Pools instead of entering them in the Airflow UI as you develop DAGs in this project.
+
+Deploy Your Project Locally
+===========================
+
+Start Airflow on your local machine by running 'astro dev start'.
+
+This command will spin up five Docker containers on your machine, each for a different Airflow component:
+
+- Postgres: Airflow's Metadata Database
+- Scheduler: The Airflow component responsible for monitoring and triggering tasks
+- DAG Processor: The Airflow component responsible for parsing DAGs
+- API Server: The Airflow component responsible for serving the Airflow UI and API
+- Triggerer: The Airflow component responsible for triggering deferred tasks
+
+When all five containers are ready the command will open the browser to the Airflow UI at http://localhost:8080/. You should also be able to access your Postgres Database at 'localhost:5432/postgres' with username 'postgres' and password 'postgres'.
+
+Note: If you already have either of the above ports allocated, you can either [stop your existing Docker containers or change the port](https://www.astronomer.io/docs/astro/cli/troubleshoot-locally#ports-are-not-available-for-my-local-airflow-webserver).
+
+Deploy Your Project to Astronomer
+=================================
+
+If you have an Astronomer account, pushing code to a Deployment on Astronomer is simple. For deploying instructions, refer to Astronomer documentation: https://www.astronomer.io/docs/astro/deploy-code/
+
+Contact
+=======
+
+The Astronomer CLI is maintained with love by the Astronomer team. To report a bug or suggest a change, reach out to our support.

models/anomaly-detection/artifacts/data_ingestion/ingested_data_20251208_152330.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3273d845dda217a76cae1c63e9e181fad07c212ed7bbbce008ceaa380012e586
+size 104390

models/anomaly-detection/artifacts/data_transformation/embeddings_20251208_152444.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:f476c6f7aee3c0d1acb74e054dd4a3b5c47170b8f476f5f273a9a87c106d4d14
+size 586880

models/anomaly-detection/artifacts/data_transformation/features_20251208_152444.npy

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:fad8dcf39385d403aca32cae883fa2030055bef9d0e5ee589bd86799adad69e3
+size 1185856

models/anomaly-detection/artifacts/data_transformation/transformed_data_20251208_152444.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:42202d32a4c0a02477aa9b90c902a2dd0ce057dc66562a1629198636d5431f80
+size 112761

models/anomaly-detection/artifacts/data_validation/validated_data_20251208_152332.parquet

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:3273d845dda217a76cae1c63e9e181fad07c212ed7bbbce008ceaa380012e586
+size 104390

models/anomaly-detection/artifacts/data_validation/validation_report_20251208_152332.yaml

@@ -0,0 +1,7 @@
+errors: []
+input_path: C:\Users\LENOVO\Desktop\ModelX-Ultimate\models\anomaly-detection\artifacts\data_ingestion\ingested_data_20251208_152330.parquet
+invalid_records: 0
+total_records: 191
+valid_records: 191
+validation_status: true
+validation_timestamp: '20251208_152332'

models/anomaly-detection/dags/exampledag.py

@@ -0,0 +1,100 @@
+"""
+## Astronaut ETL example DAG
+
+This DAG queries the list of astronauts currently in space from the
+Open Notify API and prints each astronaut's name and flying craft.
+
+There are two tasks, one to get the data from the API and save the results,
+and another to print the results. Both tasks are written in Python using
+Airflow's TaskFlow API, which allows you to easily turn Python functions into
+Airflow tasks, and automatically infer dependencies and pass data.
+
+The second task uses dynamic task mapping to create a copy of the task for
+each Astronaut in the list retrieved from the API. This list will change
+depending on how many Astronauts are in space, and the DAG will adjust
+accordingly each time it runs.
+
+For more explanation and getting started instructions, see our Write your
+first DAG tutorial: https://www.astronomer.io/docs/learn/get-started-with-airflow
+
+![Picture of the ISS](https://www.esa.int/var/esa/storage/images/esa_multimedia/images/2010/02/space_station_over_earth/10293696-3-eng-GB/Space_Station_over_Earth_card_full.jpg)
+"""
+
+from airflow.sdk.definitions.asset import Asset
+from airflow.decorators import dag, task
+from pendulum import datetime
+import requests
+
+
+# Define the basic parameters of the DAG, like schedule and start_date
+@dag(
+    start_date=datetime(2024, 1, 1),
+    schedule="@daily",
+    catchup=False,
+    doc_md=__doc__,
+    default_args={"owner": "Astro", "retries": 3},
+    tags=["example"],
+)
+def example_astronauts():
+    # Define tasks
+    @task(
+        # Define a dataset outlet for the task. This can be used to schedule downstream DAGs when this task has run.
+        outlets=[Asset("current_astronauts")]
+    )  # Define that this task updates the `current_astronauts` Dataset
+    def get_astronauts(**context) -> list[dict]:
+        """
+        This task uses the requests library to retrieve a list of Astronauts
+        currently in space. The results are pushed to XCom with a specific key
+        so they can be used in a downstream pipeline. The task returns a list
+        of Astronauts to be used in the next task.
+        """
+        try:
+            r = requests.get("http://api.open-notify.org/astros.json")
+            r.raise_for_status()
+            number_of_people_in_space = r.json()["number"]
+            list_of_people_in_space = r.json()["people"]
+        except Exception:
+            print("API currently not available, using hardcoded data instead.")
+            number_of_people_in_space = 12
+            list_of_people_in_space = [
+                {"craft": "ISS", "name": "Oleg Kononenko"},
+                {"craft": "ISS", "name": "Nikolai Chub"},
+                {"craft": "ISS", "name": "Tracy Caldwell Dyson"},
+                {"craft": "ISS", "name": "Matthew Dominick"},
+                {"craft": "ISS", "name": "Michael Barratt"},
+                {"craft": "ISS", "name": "Jeanette Epps"},
+                {"craft": "ISS", "name": "Alexander Grebenkin"},
+                {"craft": "ISS", "name": "Butch Wilmore"},
+                {"craft": "ISS", "name": "Sunita Williams"},
+                {"craft": "Tiangong", "name": "Li Guangsu"},
+                {"craft": "Tiangong", "name": "Li Cong"},
+                {"craft": "Tiangong", "name": "Ye Guangfu"},
+            ]
+
+        context["ti"].xcom_push(
+            key="number_of_people_in_space", value=number_of_people_in_space
+        )
+        return list_of_people_in_space
+
+    @task
+    def print_astronaut_craft(greeting: str, person_in_space: dict) -> None:
+        """
+        This task creates a print statement with the name of an
+        Astronaut in space and the craft they are flying on from
+        the API request results of the previous task, along with a
+        greeting which is hard-coded in this example.
+        """
+        craft = person_in_space["craft"]
+        name = person_in_space["name"]
+
+        print(f"{name} is currently in space flying on the {craft}! {greeting}")
+
+    # Use dynamic task mapping to run the print_astronaut_craft task for each
+    # Astronaut in space
+    print_astronaut_craft.partial(greeting="Hello! :)").expand(
+        person_in_space=get_astronauts()  # Define dependencies using TaskFlow API syntax
+    )
+
+
+# Instantiate the DAG
+example_astronauts()

models/anomaly-detection/dags/train_anomaly_model.py

@@ -0,0 +1,241 @@
+"""
+models/anomaly-detection/dags/train_anomaly_model.py
+Apache Airflow DAG for scheduled anomaly detection model training
+Uses Astronomer (Astro) for deployment
+"""
+from datetime import datetime, timedelta
+from airflow import DAG
+from airflow.operators.python import PythonOperator, BranchPythonOperator
+from airflow.operators.empty import EmptyOperator
+from airflow.sensors.python import PythonSensor
+import os
+import sys
+import logging
+
+# Add project to path
+PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, PROJECT_ROOT)
+
+# Load .env from root ModelX directory for MLflow credentials
+try:
+    from dotenv import load_dotenv
+    root_env = os.path.join(PROJECT_ROOT, '..', '..', '.env')
+    if os.path.exists(root_env):
+        load_dotenv(root_env)
+    else:
+        load_dotenv()  # Try default locations
+except ImportError:
+    pass
+
+logger = logging.getLogger(__name__)
+
+# Configuration
+BATCH_THRESHOLD = int(os.getenv("BATCH_THRESHOLD", "1000"))
+SQLITE_DB_PATH = os.getenv("SQLITE_DB_PATH", "")
+
+# Default DAG arguments
+default_args = {
+    'owner': 'modelx',
+    'depends_on_past': False,
+    'email_on_failure': False,
+    'email_on_retry': False,
+    'retries': 2,
+    'retry_delay': timedelta(minutes=5),
+}
+
+
+def check_new_records(**context) -> bool:
+    """
+    Sensor function to check if enough new records exist.
+    Returns True if batch threshold is met or daily run is due.
+    """
+    import sqlite3
+    from datetime import datetime, timedelta
+    
+    try:
+        # Get last training timestamp from XCom or default to 24h ago
+        last_training = context['ti'].xcom_pull(key='last_training_timestamp')
+        if not last_training:
+            last_training = (datetime.utcnow() - timedelta(hours=24)).isoformat()
+        
+        # Check SQLite for new records
+        if SQLITE_DB_PATH and os.path.exists(SQLITE_DB_PATH):
+            conn = sqlite3.connect(SQLITE_DB_PATH)
+            cursor = conn.execute(
+                'SELECT COUNT(*) FROM seen_hashes WHERE last_seen > ?',
+                (last_training,)
+            )
+            new_records = cursor.fetchone()[0]
+            conn.close()
+            
+            logger.info(f"[AnomalyDAG] New records since {last_training}: {new_records}")
+            
+            if new_records >= BATCH_THRESHOLD:
+                logger.info(f"[AnomalyDAG] Batch threshold met ({new_records} >= {BATCH_THRESHOLD})")
+                return True
+        
+        # Check if 24 hours have passed (daily fallback)
+        if last_training:
+            last_dt = datetime.fromisoformat(last_training)
+            hours_since = (datetime.utcnow() - last_dt).total_seconds() / 3600
+            if hours_since >= 24:
+                logger.info(f"[AnomalyDAG] Daily run triggered ({hours_since:.1f}h since last run)")
+                return True
+        
+        logger.info(f"[AnomalyDAG] Waiting for more records...")
+        return False
+        
+    except Exception as e:
+        logger.error(f"[AnomalyDAG] Error checking records: {e}")
+        # Trigger anyway on error
+        return True
+
+
+def run_data_ingestion(**context):
+    """Run data ingestion step"""
+    from src.components import DataIngestion
+    from src.entity import DataIngestionConfig
+    
+    config = DataIngestionConfig()
+    ingestion = DataIngestion(config)
+    artifact = ingestion.ingest()
+    
+    # Store artifact path in XCom
+    context['ti'].xcom_push(key='ingestion_artifact', value={
+        'raw_data_path': artifact.raw_data_path,
+        'total_records': artifact.total_records,
+        'is_data_available': artifact.is_data_available
+    })
+    
+    if not artifact.is_data_available:
+        raise ValueError("No data available for training")
+    
+    return artifact.raw_data_path
+
+
+def run_data_validation(**context):
+    """Run data validation step"""
+    from src.components import DataValidation
+    from src.entity import DataValidationConfig
+    
+    # Get ingestion output from XCom
+    ingestion = context['ti'].xcom_pull(key='ingestion_artifact', task_ids='data_ingestion')
+    raw_data_path = ingestion['raw_data_path']
+    
+    config = DataValidationConfig()
+    validation = DataValidation(config)
+    artifact = validation.validate(raw_data_path)
+    
+    # Store artifact in XCom
+    context['ti'].xcom_push(key='validation_artifact', value={
+        'validated_data_path': artifact.validated_data_path,
+        'validation_status': artifact.validation_status,
+        'valid_records': artifact.valid_records
+    })
+    
+    return artifact.validated_data_path
+
+
+def run_data_transformation(**context):
+    """Run data transformation step"""
+    from src.components import DataTransformation
+    from src.entity import DataTransformationConfig
+    
+    # Get validation output from XCom
+    validation = context['ti'].xcom_pull(key='validation_artifact', task_ids='data_validation')
+    validated_data_path = validation['validated_data_path']
+    
+    config = DataTransformationConfig()
+    transformation = DataTransformation(config)
+    artifact = transformation.transform(validated_data_path)
+    
+    # Store artifact in XCom
+    context['ti'].xcom_push(key='transformation_artifact', value={
+        'feature_store_path': artifact.feature_store_path,
+        'language_distribution': artifact.language_distribution,
+        'total_records': artifact.total_records
+    })
+    
+    return artifact.feature_store_path
+
+
+def run_model_training(**context):
+    """Run model training with Optuna and MLflow"""
+    from src.components import ModelTrainer
+    from src.entity import ModelTrainerConfig
+    from datetime import datetime
+    
+    # Get transformation output from XCom
+    transformation = context['ti'].xcom_pull(key='transformation_artifact', task_ids='data_transformation')
+    feature_path = transformation['feature_store_path']
+    
+    config = ModelTrainerConfig()
+    trainer = ModelTrainer(config)
+    artifact = trainer.train(feature_path)
+    
+    # Store training timestamp for next run
+    context['ti'].xcom_push(key='last_training_timestamp', value=datetime.utcnow().isoformat())
+    
+    # Store artifact in XCom
+    context['ti'].xcom_push(key='training_artifact', value={
+        'best_model_name': artifact.best_model_name,
+        'best_model_path': artifact.best_model_path,
+        'mlflow_run_id': artifact.mlflow_run_id,
+        'n_anomalies': artifact.n_anomalies
+    })
+    
+    return artifact.best_model_path
+
+
+# Create DAG
+with DAG(
+    'anomaly_detection_training',
+    default_args=default_args,
+    description='Train anomaly detection models on feed data',
+    schedule_interval=timedelta(hours=4),  # Check every 4 hours
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+    tags=['ml', 'anomaly', 'modelx'],
+) as dag:
+    
+    # Start
+    start = EmptyOperator(task_id='start')
+    
+    # Sensor: Check for new records
+    check_records = PythonSensor(
+        task_id='check_new_records',
+        python_callable=check_new_records,
+        timeout=3600,
+        poke_interval=300,  # Check every 5 minutes
+        mode='poke',
+    )
+    
+    # Data Ingestion
+    data_ingestion = PythonOperator(
+        task_id='data_ingestion',
+        python_callable=run_data_ingestion,
+    )
+    
+    # Data Validation
+    data_validation = PythonOperator(
+        task_id='data_validation',
+        python_callable=run_data_validation,
+    )
+    
+    # Data Transformation
+    data_transformation = PythonOperator(
+        task_id='data_transformation',
+        python_callable=run_data_transformation,
+    )
+    
+    # Model Training
+    model_training = PythonOperator(
+        task_id='model_training',
+        python_callable=run_model_training,
+    )
+    
+    # End
+    end = EmptyOperator(task_id='end')
+    
+    # Pipeline flow
+    start >> check_records >> data_ingestion >> data_validation >> data_transformation >> model_training >> end

models/anomaly-detection/data_schema/schema.yaml

@@ -0,0 +1,133 @@
+# Data Schema for Anomaly Detection Pipeline
+# Based on the feeds dataset created by the combined graph
+
+feed_columns:
+  # Required columns
+  post_id:
+    dtype: str
+    required: true
+    description: "Unique identifier for the post"
+  
+  timestamp:
+    dtype: str  # ISO format or Unix timestamp
+    required: true
+    description: "Post timestamp"
+  
+  platform:
+    dtype: str
+    required: true
+    allowed_values: ["reddit", "facebook", "twitter", "linkedin", "instagram", "web"]
+    description: "Source platform"
+  
+  category:
+    dtype: str
+    required: true
+    description: "Post category (competitor_mention, profile_monitoring, etc.)"
+  
+  text:
+    dtype: str
+    required: true
+    min_length: 10
+    max_length: 5000
+    description: "Main text content"
+  
+  content_hash:
+    dtype: str
+    required: true
+    description: "MD5/SHA256 hash for deduplication"
+  
+  # Optional columns
+  entity:
+    dtype: str
+    required: false
+    description: "Referenced entity (Dialog, SLT, etc.)"
+  
+  poster:
+    dtype: str
+    required: false
+    description: "Author/poster username"
+  
+  post_url:
+    dtype: str
+    required: false
+    description: "URL to original post"
+  
+  title:
+    dtype: str
+    required: false
+    description: "Post title if available"
+  
+  engagement_score:
+    dtype: float
+    required: false
+    min_value: 0
+    description: "Overall engagement score"
+  
+  engagement_likes:
+    dtype: int
+    required: false
+    min_value: 0
+    description: "Number of likes"
+  
+  engagement_shares:
+    dtype: int
+    required: false
+    min_value: 0
+    description: "Number of shares"
+  
+  engagement_comments:
+    dtype: int
+    required: false
+    min_value: 0
+    description: "Number of comments"
+  
+  source_tool:
+    dtype: str
+    required: false
+    description: "Tool used for scraping (scrape_reddit, scrape_facebook_profile, etc.)"
+
+# SQLite schema for feed cache
+sqlite_schema:
+  table: seen_hashes
+  columns:
+    - content_hash: TEXT PRIMARY KEY
+    - first_seen: TIMESTAMP
+    - last_seen: TIMESTAMP
+    - event_id: TEXT
+    - summary_preview: TEXT
+
+# Feature engineering configuration
+features:
+  temporal:
+    - hour_of_day
+    - day_of_week
+    - is_weekend
+    - is_business_hours
+  
+  engagement:
+    - normalized_score
+    - log_engagement
+    - engagement_ratio
+  
+  text:
+    - language  # en, si, ta
+    - vector_embedding  # 768-dim for BERT models
+    - text_length
+    - word_count
+
+# Language detection configuration
+languages:
+  supported:
+    - code: en
+      name: English
+      model: distilbert-base-uncased
+    - code: si  
+      name: Sinhala
+      model: keshan/SinhalaBERTo
+    - code: ta
+      name: Tamil
+      model: l3cube-pune/tamil-bert
+  
+  detection:
+    method: fasttext  # or lingua-py
+    min_confidence: 0.8

models/anomaly-detection/download_models.py

@@ -0,0 +1,86 @@
+"""
+models/anomaly-detection/download_models.py
+Script to pre-download all required models for the pipeline.
+"""
+import os
+import sys
+import requests
+import logging
+from pathlib import Path
+from tqdm import tqdm
+
+# Add src to path
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+# Configure logging
+logging.basicConfig(level=logging.INFO, format='%(message)s')
+logger = logging.getLogger("downloader")
+
+# Constants
+CACHE_DIR = Path(__file__).parent / "models_cache"
+FASTTEXT_URL = "https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.bin"
+FASTTEXT_PATH = CACHE_DIR / "lid.176.bin"
+
+def download_file(url, destination):
+    """Download file with progress bar"""
+    response = requests.get(url, stream=True)
+    total_size = int(response.headers.get('content-length', 0))
+    
+    with open(destination, 'wb') as file, tqdm(
+        desc=destination.name,
+        total=total_size,
+        unit='iB',
+        unit_scale=True,
+        unit_divisor=1024,
+    ) as bar:
+        for data in response.iter_content(chunk_size=1024):
+            size = file.write(data)
+            bar.update(size)
+
+def main():
+    logger.info("=" * 50)
+    logger.info("⬇️  MODEL DOWNLOADER")
+    logger.info("=" * 50)
+    
+    # Ensure cache directory exists
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    logger.info(f"📂 Cache Directory: {CACHE_DIR}")
+    
+    # 1. Download FastText Model
+    logger.info("\n[1/2] Checking FastText Model (Language Detection)...")
+    if not FASTTEXT_PATH.exists():
+        logger.info(f"   Downloading lid.176.bin...")
+        try:
+            download_file(FASTTEXT_URL, FASTTEXT_PATH)
+            logger.info("   ✅ Download complete")
+        except Exception as e:
+            logger.error(f"   ❌ Failed to download FastText: {e}")
+    else:
+        logger.info("   ✅ FastText model already exists")
+
+    # 2. Download HuggingFace Models
+    logger.info("\n[2/2] Checking HuggingFace BERT Models (Vectorization)...")
+    try:
+        from src.utils.vectorizer import get_vectorizer
+        
+        # Initialize vectorizer which handles HF downloads
+        logger.info("   Initializing vectorizer to trigger downloads...")
+        vectorizer = get_vectorizer(models_cache_dir=str(CACHE_DIR))
+        
+        # Trigger downloads for all languages
+        vectorizer.download_all_models()
+        
+        logger.info("   ✅ All BERT models ready")
+        
+    except ImportError:
+        logger.error("   ❌ Could not import vectorizer. Install requirements first:")
+        logger.error("      pip install -r requirements.txt")
+    except Exception as e:
+        logger.error(f"   ❌ Error downloading BERT models: {e}")
+
+    logger.info("\n" + "=" * 50)
+    logger.info("✨ SETUP COMPLETE")
+    logger.info("=" * 50)
+
+if __name__ == "__main__":
+    main()

models/anomaly-detection/main.py

@@ -0,0 +1,85 @@
+"""
+models/anomaly-detection/main.py
+Entry point for the anomaly detection training pipeline
+"""
+import os
+import sys
+import logging
+from pathlib import Path
+
+# Add src to path
+sys.path.insert(0, str(Path(__file__).parent / "src"))
+
+from src.pipeline import run_training_pipeline
+from src.entity import PipelineConfig
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
+    handlers=[
+        logging.StreamHandler(),
+        logging.FileHandler("training.log")
+    ]
+)
+
+logger = logging.getLogger("main")
+
+
+def main():
+    """Run the anomaly detection training pipeline"""
+    logger.info("=" * 60)
+    logger.info("ANOMALY DETECTION PIPELINE")
+    logger.info("=" * 60)
+    
+    # Load environment variables
+    from dotenv import load_dotenv
+    load_dotenv()
+    
+    # Create configuration
+    config = PipelineConfig()
+    
+    # Run pipeline
+    try:
+        artifact = run_training_pipeline(config)
+        
+        logger.info("\n" + "=" * 60)
+        logger.info("PIPELINE RESULTS")
+        logger.info("=" * 60)
+        logger.info(f"Status: {artifact.pipeline_status}")
+        logger.info(f"Run ID: {artifact.pipeline_run_id}")
+        logger.info(f"Duration: {artifact.pipeline_start_time} to {artifact.pipeline_end_time}")
+        
+        logger.info("\n--- Data Ingestion ---")
+        logger.info(f"Total records: {artifact.data_ingestion.total_records}")
+        logger.info(f"From SQLite: {artifact.data_ingestion.records_from_sqlite}")
+        logger.info(f"From CSV: {artifact.data_ingestion.records_from_csv}")
+        
+        logger.info("\n--- Data Validation ---")
+        logger.info(f"Valid records: {artifact.data_validation.valid_records}")
+        logger.info(f"Validation status: {artifact.data_validation.validation_status}")
+        
+        logger.info("\n--- Data Transformation ---")
+        logger.info(f"Language distribution: {artifact.data_transformation.language_distribution}")
+        
+        logger.info("\n--- Model Training ---")
+        logger.info(f"Best model: {artifact.model_trainer.best_model_name}")
+        logger.info(f"Best metrics: {artifact.model_trainer.best_model_metrics}")
+        logger.info(f"MLflow run: {artifact.model_trainer.mlflow_run_id}")
+        
+        if artifact.model_trainer.n_anomalies:
+            logger.info(f"Anomalies detected: {artifact.model_trainer.n_anomalies}")
+        
+        logger.info("\n" + "=" * 60)
+        logger.info("PIPELINE COMPLETE")
+        logger.info("=" * 60)
+        
+        return artifact
+        
+    except Exception as e:
+        logger.error(f"Pipeline failed: {e}")
+        raise
+
+
+if __name__ == "__main__":
+    main()

models/anomaly-detection/requirements.txt

@@ -0,0 +1,37 @@
+# Anomaly Detection Pipeline - Requirements
+
+# ML & Clustering
+optuna>=3.0
+scikit-learn>=1.0
+hdbscan>=0.8.29
+umap-learn>=0.5
+
+# Language Detection
+fasttext-wheel>=0.9.2
+lingua-language-detector>=2.0
+
+# NLP & Transformers
+transformers>=4.30
+sentence-transformers>=2.0
+torch>=2.0
+
+# MLflow & Tracking
+mlflow>=2.0
+dagshub>=0.3
+
+# Database
+pymongo>=4.0
+
+# Data Processing
+pandas>=2.0
+numpy>=1.24
+pyyaml>=6.0
+
+# Airflow (managed by Astro)
+# apache-airflow>=2.7
+
+# Utilities
+joblib>=1.3
+tqdm>=4.65
+
+astro-run-dag # This package is needed for the astro run command. It will be removed before a deploy

models/anomaly-detection/src/__init__.py

@@ -0,0 +1,18 @@
+"""
+models/anomaly-detection/src/__init__.py
+Anomaly Detection Pipeline Package
+"""
+
+from .components.data_ingestion import DataIngestion
+from .components.data_validation import DataValidation
+from .components.data_transformation import DataTransformation
+from .components.model_trainer import ModelTrainer
+
+__all__ = [
+    "DataIngestion",
+    "DataValidation",
+    "DataTransformation",
+    "ModelTrainer"
+]
+
+__version__ = "1.0.0"

models/anomaly-detection/src/components/__init__.py

@@ -0,0 +1,29 @@
+"""
+models/anomaly-detection/src/components/__init__.py
+
+Sets up paths for integration with main project before importing components.
+"""
+import sys
+from pathlib import Path
+
+# Add main project root to path for vectorization agent graph access
+# Path: models/anomaly-detection/src/components/__init__.py -> go up 4 levels to ModelX-Ultimate
+# Note: This is secondary to anomaly-detection path. Direct graph import won't work
+# due to 'src' namespace collision. Use VectorizationAPI HTTP calls instead.
+_main_project_root = Path(__file__).parent.parent.parent.parent.parent
+_main_path = str(_main_project_root)
+if _main_path not in sys.path:
+    sys.path.append(_main_path)  # Append, don't insert at 0
+
+from .data_ingestion import DataIngestion
+from .data_validation import DataValidation
+from .data_transformation import DataTransformation
+from .model_trainer import ModelTrainer
+
+__all__ = [
+    "DataIngestion",
+    "DataValidation",
+    "DataTransformation",
+    "ModelTrainer"
+]
+

models/anomaly-detection/src/components/data_ingestion.py

@@ -0,0 +1,247 @@
+"""
+models/anomaly-detection/src/components/data_ingestion.py
+Data ingestion from SQLite feed cache and CSV files
+"""
+import os
+import sqlite3
+import pandas as pd
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+from ..entity import DataIngestionConfig, DataIngestionArtifact
+
+logger = logging.getLogger("data_ingestion")
+
+
+class DataIngestion:
+    """
+    Data ingestion component that fetches feed data from:
+    1. SQLite database (feed_cache.db) - production deduped feeds
+    2. CSV files in datasets/political_feeds/ - historical data
+    """
+    
+    def __init__(self, config: Optional[DataIngestionConfig] = None):
+        """
+        Initialize data ingestion component.
+        
+        Args:
+            config: Optional configuration, uses defaults if None
+        """
+        self.config = config or DataIngestionConfig()
+        
+        # Ensure output directory exists
+        Path(self.config.output_directory).mkdir(parents=True, exist_ok=True)
+        
+        logger.info(f"[DataIngestion] Initialized")
+        logger.info(f"  SQLite: {self.config.sqlite_db_path}")
+        logger.info(f"  CSV Dir: {self.config.csv_directory}")
+        logger.info(f"  Output: {self.config.output_directory}")
+    
+    def _fetch_from_sqlite(self) -> pd.DataFrame:
+        """
+        Fetch feed data from SQLite cache database.
+        
+        Returns:
+            DataFrame with feed records
+        """
+        db_path = self.config.sqlite_db_path
+        
+        if not os.path.exists(db_path):
+            logger.warning(f"[DataIngestion] SQLite DB not found: {db_path}")
+            return pd.DataFrame()
+        
+        try:
+            conn = sqlite3.connect(db_path)
+            
+            # Query the seen_hashes table
+            query = """
+                SELECT 
+                    content_hash as post_id,
+                    first_seen as timestamp,
+                    event_id,
+                    summary_preview as text
+                FROM seen_hashes
+                ORDER BY last_seen DESC
+            """
+            df = pd.read_sql_query(query, conn)
+            conn.close()
+            
+            # Add default columns for compatibility
+            if not df.empty:
+                df["platform"] = "mixed"
+                df["category"] = "feed"
+                df["content_hash"] = df["post_id"]
+                df["source"] = "sqlite"
+            
+            logger.info(f"[DataIngestion] Fetched {len(df)} records from SQLite")
+            return df
+            
+        except Exception as e:
+            logger.error(f"[DataIngestion] SQLite error: {e}")
+            return pd.DataFrame()
+    
+    def _fetch_from_csv(self) -> pd.DataFrame:
+        """
+        Fetch feed data from CSV files in datasets directory.
+        
+        Returns:
+            Combined DataFrame from all CSV files
+        """
+        csv_dir = Path(self.config.csv_directory)
+        
+        if not csv_dir.exists():
+            logger.warning(f"[DataIngestion] CSV directory not found: {csv_dir}")
+            return pd.DataFrame()
+        
+        all_dfs = []
+        csv_files = list(csv_dir.glob("*.csv"))
+        
+        for csv_file in csv_files:
+            try:
+                df = pd.read_csv(csv_file)
+                df["source_file"] = csv_file.name
+                df["source"] = "csv"
+                all_dfs.append(df)
+                logger.info(f"[DataIngestion] Loaded {len(df)} records from {csv_file.name}")
+            except Exception as e:
+                logger.warning(f"[DataIngestion] Failed to load {csv_file}: {e}")
+        
+        if not all_dfs:
+            return pd.DataFrame()
+        
+        combined = pd.concat(all_dfs, ignore_index=True)
+        logger.info(f"[DataIngestion] Total {len(combined)} records from {len(csv_files)} CSV files")
+        return combined
+    
+    def _deduplicate(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Remove duplicate records based on content_hash.
+        
+        Args:
+            df: Input DataFrame
+            
+        Returns:
+            Deduplicated DataFrame
+        """
+        if df.empty:
+            return df
+        
+        initial_count = len(df)
+        
+        # Use content_hash for deduplication, fallback to post_id
+        if "content_hash" in df.columns:
+            df = df.drop_duplicates(subset=["content_hash"], keep="first")
+        elif "post_id" in df.columns:
+            df = df.drop_duplicates(subset=["post_id"], keep="first")
+        
+        deduped_count = len(df)
+        removed = initial_count - deduped_count
+        
+        if removed > 0:
+            logger.info(f"[DataIngestion] Deduplicated: removed {removed} duplicates")
+        
+        return df
+    
+    def _filter_valid_records(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Filter records with sufficient text content.
+        
+        Args:
+            df: Input DataFrame
+            
+        Returns:
+            Filtered DataFrame
+        """
+        if df.empty:
+            return df
+        
+        initial_count = len(df)
+        
+        # Ensure text column exists
+        if "text" not in df.columns:
+            # Try alternative column names
+            text_cols = ["summary_preview", "title", "content"]
+            for col in text_cols:
+                if col in df.columns:
+                    df["text"] = df[col]
+                    break
+        
+        if "text" not in df.columns:
+            logger.warning("[DataIngestion] No text column found")
+            df["text"] = ""
+        
+        # Filter by minimum text length
+        df = df[df["text"].str.len() >= self.config.min_text_length]
+        
+        filtered_count = len(df)
+        removed = initial_count - filtered_count
+        
+        if removed > 0:
+            logger.info(f"[DataIngestion] Filtered: removed {removed} short texts")
+        
+        return df
+    
+    def ingest(self) -> DataIngestionArtifact:
+        """
+        Execute data ingestion pipeline.
+        
+        Returns:
+            DataIngestionArtifact with paths and statistics
+        """
+        logger.info("[DataIngestion] Starting data ingestion...")
+        
+        # Fetch from both sources
+        sqlite_df = self._fetch_from_sqlite()
+        csv_df = self._fetch_from_csv()
+        
+        records_from_sqlite = len(sqlite_df)
+        records_from_csv = len(csv_df)
+        
+        # Combine sources
+        if not sqlite_df.empty and not csv_df.empty:
+            # Ensure compatible columns
+            common_cols = list(set(sqlite_df.columns) & set(csv_df.columns))
+            combined_df = pd.concat([
+                sqlite_df[common_cols], 
+                csv_df[common_cols]
+            ], ignore_index=True)
+        elif not sqlite_df.empty:
+            combined_df = sqlite_df
+        elif not csv_df.empty:
+            combined_df = csv_df
+        else:
+            combined_df = pd.DataFrame()
+        
+        # Deduplicate
+        combined_df = self._deduplicate(combined_df)
+        
+        # Filter valid records
+        combined_df = self._filter_valid_records(combined_df)
+        
+        total_records = len(combined_df)
+        is_data_available = total_records > 0
+        
+        # Save to output
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        output_path = Path(self.config.output_directory) / f"ingested_data_{timestamp}.parquet"
+        
+        if is_data_available:
+            combined_df.to_parquet(output_path, index=False)
+            logger.info(f"[DataIngestion] Saved {total_records} records to {output_path}")
+        else:
+            output_path = str(output_path)
+            logger.warning("[DataIngestion] No data available to save")
+        
+        artifact = DataIngestionArtifact(
+            raw_data_path=str(output_path),
+            total_records=total_records,
+            records_from_sqlite=records_from_sqlite,
+            records_from_csv=records_from_csv,
+            ingestion_timestamp=timestamp,
+            is_data_available=is_data_available
+        )
+        
+        logger.info(f"[DataIngestion] ✓ Complete: {total_records} records")
+        return artifact

models/anomaly-detection/src/components/data_transformation.py

@@ -0,0 +1,458 @@
+"""
+models/anomaly-detection/src/components/data_transformation.py
+Data transformation with language detection and text vectorization
+Integrates with Vectorization Agent Graph for LLM-enhanced processing
+"""
+import os
+import pandas as pd
+import numpy as np
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, Dict, Any, List
+from tqdm import tqdm
+
+from ..entity import DataTransformationConfig, DataTransformationArtifact
+from ..utils import detect_language, get_vectorizer
+
+logger = logging.getLogger("data_transformation")
+
+
+class DataTransformation:
+    """
+    Data transformation component that:
+    1. Detects language (Sinhala/Tamil/English)
+    2. Extracts text embeddings using language-specific BERT models
+    3. Engineers temporal and engagement features
+    4. Optionally integrates with Vectorizer Agent Graph for LLM insights
+    """
+    
+    def __init__(self, config: Optional[DataTransformationConfig] = None, use_agent_graph: bool = True):
+        """
+        Initialize data transformation component.
+        
+        Args:
+            config: Optional configuration, uses defaults if None
+            use_agent_graph: If True, use vectorizer agent graph for processing
+        """
+        self.config = config or DataTransformationConfig()
+        self.use_agent_graph = use_agent_graph
+        
+        # Ensure output directory exists
+        Path(self.config.output_directory).mkdir(parents=True, exist_ok=True)
+        
+        # Get vectorizer (lazy loaded)
+        self.vectorizer = get_vectorizer(self.config.models_cache_dir)
+        
+        # Vectorization API integration
+        # Note: Direct import of vectorizationAgentGraph fails due to 'src' namespace collision
+        # between this project (models/anomaly-detection/src) and main project (src).
+        # Instead, we call the Vectorization API via HTTP when available.
+        self.vectorizer_graph = None  # Not used - we use HTTP API instead
+        self.vectorization_api_url = os.getenv("VECTORIZATION_API_URL", "http://localhost:8001")
+        self.vectorization_api_available = False
+        
+        if self.use_agent_graph:
+            # Check if vectorization API is available
+            try:
+                import requests
+                response = requests.get(f"{self.vectorization_api_url}/health", timeout=10)
+                if response.status_code == 200:
+                    self.vectorization_api_available = True
+                    logger.info(f"[DataTransformation] [OK] Vectorization API available at {self.vectorization_api_url}")
+                else:
+                    logger.warning(f"[DataTransformation] Vectorization API returned status {response.status_code}")
+            except Exception as e:
+                logger.warning(f"[DataTransformation] Vectorization API not available: {e}")
+                logger.info("[DataTransformation] Using local vectorization (no LLM insights)")
+        
+        logger.info(f"[DataTransformation] Initialized")
+        logger.info(f"  Models cache: {self.config.models_cache_dir}")
+        logger.info(f"  Vectorization API: {'enabled' if self.vectorization_api_available else 'disabled (using local)'}")
+    
+    def _process_with_agent_graph(self, texts: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Process texts through the Vectorization API.
+        
+        Uses HTTP calls to the vectorization API server which runs the
+        Vectorizer Agent Graph. This avoids the 'src' namespace collision.
+        
+        This provides:
+        - Language detection
+        - Vector embeddings
+        - LLM expert summary
+        - Opportunity/threat analysis
+        
+        Args:
+            texts: List of {text, post_id, metadata} dicts
+            
+        Returns:
+            Dict with language_detection_results, vector_embeddings, expert_summary, etc.
+        """
+        if not self.vectorization_api_available:
+            logger.warning("[DataTransformation] Vectorization API not available, using fallback")
+            return None
+        
+        try:
+            import requests
+            
+            batch_id = datetime.now().strftime("%Y%m%d_%H%M%S")
+            
+            # Prepare request payload
+            payload = {
+                "texts": [
+                    {
+                        "text": item.get("text", ""),
+                        "post_id": item.get("post_id", f"text_{i}"),
+                        "metadata": item.get("metadata", {})
+                    }
+                    for i, item in enumerate(texts)
+                ],
+                "batch_id": batch_id,
+                "include_vectors": True,
+                "include_expert_summary": True
+            }
+            
+            # Call vectorization API
+            response = requests.post(
+                f"{self.vectorization_api_url}/vectorize",
+                json=payload,
+                timeout=120  # 2 minutes for large batches
+            )
+            
+            if response.status_code == 200:
+                result = response.json()
+                logger.info(f"[DataTransformation] Vectorization API processed {len(texts)} texts")
+                
+                # Convert API response to expected format
+                return {
+                    "language_detection_results": result.get("vectors", []),
+                    "vector_embeddings": result.get("vectors", []),
+                    "expert_summary": result.get("expert_summary", ""),
+                    "opportunities": [],  # Extracted from domain_insights
+                    "threats": [],  # Extracted from domain_insights
+                    "domain_insights": result.get("domain_insights", []),
+                    "processing_stats": {
+                        "language_distribution": result.get("language_distribution", {}),
+                        "processing_time": result.get("processing_time_seconds", 0)
+                    }
+                }
+            else:
+                logger.error(f"[DataTransformation] Vectorization API error: {response.status_code}")
+                return None
+            
+        except Exception as e:
+            logger.error(f"[DataTransformation] Vectorization API call failed: {e}")
+            return None
+    
+    def _detect_languages(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Detect language for each text entry.
+        
+        Args:
+            df: Input DataFrame with 'text' column
+            
+        Returns:
+            DataFrame with 'language' and 'language_confidence' columns
+        """
+        logger.info("[DataTransformation] Detecting languages...")
+        
+        languages = []
+        confidences = []
+        
+        for text in tqdm(df["text"].fillna(""), desc="Language Detection"):
+            lang, conf = detect_language(text)
+            languages.append(lang)
+            confidences.append(conf)
+        
+        df["language"] = languages
+        df["language_confidence"] = confidences
+        
+        # Log distribution
+        lang_counts = df["language"].value_counts()
+        logger.info(f"[DataTransformation] Language distribution:")
+        for lang, count in lang_counts.items():
+            logger.info(f"  {lang}: {count} ({100*count/len(df):.1f}%)")
+        
+        return df
+    
+    def _extract_temporal_features(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Extract temporal features from timestamp.
+        
+        Args:
+            df: Input DataFrame with 'timestamp' column
+            
+        Returns:
+            DataFrame with temporal feature columns
+        """
+        logger.info("[DataTransformation] Extracting temporal features...")
+        
+        if "timestamp" not in df.columns:
+            logger.warning("[DataTransformation] No timestamp column found")
+            return df
+        
+        # Convert to datetime
+        try:
+            df["datetime"] = pd.to_datetime(df["timestamp"], errors='coerce')
+        except Exception as e:
+            logger.warning(f"[DataTransformation] Timestamp conversion error: {e}")
+            return df
+        
+        # Extract features
+        df["hour_of_day"] = df["datetime"].dt.hour.fillna(0).astype(int)
+        df["day_of_week"] = df["datetime"].dt.dayofweek.fillna(0).astype(int)
+        df["is_weekend"] = (df["day_of_week"] >= 5).astype(int)
+        df["is_business_hours"] = ((df["hour_of_day"] >= 9) & (df["hour_of_day"] <= 17)).astype(int)
+        
+        # Drop intermediate column
+        df = df.drop(columns=["datetime"], errors='ignore')
+        
+        return df
+    
+    def _extract_engagement_features(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Extract and normalize engagement features.
+        
+        Args:
+            df: Input DataFrame
+            
+        Returns:
+            DataFrame with engagement feature columns
+        """
+        logger.info("[DataTransformation] Extracting engagement features...")
+        
+        # Check for engagement columns
+        engagement_cols = ["engagement_score", "engagement_likes", "engagement_shares", "engagement_comments"]
+        
+        for col in engagement_cols:
+            if col not in df.columns:
+                df[col] = 0
+        
+        # Combined engagement score
+        df["total_engagement"] = (
+            df["engagement_likes"].fillna(0) +
+            df["engagement_shares"].fillna(0) * 2 +  # Shares weighted more
+            df["engagement_comments"].fillna(0)
+        )
+        
+        # Log transform for better distribution
+        df["log_engagement"] = np.log1p(df["total_engagement"])
+        
+        # Normalize to 0-1 range
+        max_engagement = df["total_engagement"].max()
+        if max_engagement > 0:
+            df["normalized_engagement"] = df["total_engagement"] / max_engagement
+        else:
+            df["normalized_engagement"] = 0
+        
+        return df
+    
+    def _extract_text_features(self, df: pd.DataFrame) -> pd.DataFrame:
+        """
+        Extract basic text features.
+        
+        Args:
+            df: Input DataFrame with 'text' column
+            
+        Returns:
+            DataFrame with text feature columns
+        """
+        logger.info("[DataTransformation] Extracting text features...")
+        
+        df["text_length"] = df["text"].fillna("").str.len()
+        df["word_count"] = df["text"].fillna("").str.split().str.len().fillna(0).astype(int)
+        
+        return df
+    
+    def _vectorize_texts(self, df: pd.DataFrame) -> np.ndarray:
+        """
+        Vectorize texts using language-specific BERT models.
+        
+        Args:
+            df: Input DataFrame with 'text' and 'language' columns
+            
+        Returns:
+            numpy array of shape (n_samples, 768)
+        """
+        logger.info("[DataTransformation] Vectorizing texts with BERT models...")
+        
+        embeddings = []
+        
+        for idx, row in tqdm(df.iterrows(), total=len(df), desc="Text Vectorization"):
+            text = row.get("text", "")
+            language = row.get("language", "english")
+            
+            try:
+                embedding = self.vectorizer.vectorize(text, language)
+                embeddings.append(embedding)
+            except Exception as e:
+                logger.debug(f"Vectorization error at {idx}: {e}")
+                embeddings.append(np.zeros(self.config.vector_dim))
+        
+        return np.array(embeddings)
+    
+    def _build_feature_matrix(self, df: pd.DataFrame, embeddings: np.ndarray) -> np.ndarray:
+        """
+        Combine all features into a single feature matrix.
+        
+        Args:
+            df: DataFrame with engineered features
+            embeddings: Text embeddings array
+            
+        Returns:
+            Combined feature matrix
+        """
+        logger.info("[DataTransformation] Building feature matrix...")
+        
+        # Numeric features to include
+        numeric_cols = [
+            "hour_of_day", "day_of_week", "is_weekend", "is_business_hours",
+            "log_engagement", "normalized_engagement",
+            "text_length", "word_count"
+        ]
+        
+        # Filter to available columns
+        available_cols = [col for col in numeric_cols if col in df.columns]
+        
+        if available_cols:
+            numeric_features = df[available_cols].fillna(0).values
+            # Normalize numeric features
+            from sklearn.preprocessing import StandardScaler
+            scaler = StandardScaler()
+            numeric_features = scaler.fit_transform(numeric_features)
+        else:
+            numeric_features = np.zeros((len(df), 1))
+        
+        # Combine with embeddings
+        feature_matrix = np.hstack([embeddings, numeric_features])
+        
+        logger.info(f"[DataTransformation] Feature matrix shape: {feature_matrix.shape}")
+        return feature_matrix
+    
+    def transform(self, data_path: str) -> DataTransformationArtifact:
+        """
+        Execute data transformation pipeline.
+        Integrates with Vectorizer Agent Graph for LLM-enhanced processing.
+        
+        Args:
+            data_path: Path to validated data
+            
+        Returns:
+            DataTransformationArtifact with paths and statistics
+        """
+        import json
+        
+        logger.info(f"[DataTransformation] Starting transformation: {data_path}")
+        
+        # Load data
+        df = pd.read_parquet(data_path)
+        total_records = len(df)
+        logger.info(f"[DataTransformation] Loaded {total_records} records")
+        
+        # Initialize agent graph results
+        agent_result = None
+        expert_summary = None
+        
+        # Try to process with vectorizer agent graph first
+        if self.vectorizer_graph and self.use_agent_graph:
+            logger.info("[DataTransformation] Using Vectorizer Agent Graph...")
+            
+            # Prepare texts for agent graph
+            texts_for_agent = []
+            for idx, row in df.iterrows():
+                texts_for_agent.append({
+                    "post_id": str(row.get("id", idx)),
+                    "text": str(row.get("text", "")),
+                    "metadata": {
+                        "source": row.get("source", "unknown"),
+                        "timestamp": str(row.get("timestamp", ""))
+                    }
+                })
+            
+            # Process through agent graph
+            agent_result = self._process_with_agent_graph(texts_for_agent)
+            
+            if agent_result:
+                expert_summary = agent_result.get("expert_summary", "")
+                logger.info(f"[DataTransformation] Agent graph completed with expert summary")
+        
+        # Run standard transformations (fallback or additional)
+        df = self._detect_languages(df)
+        df = self._extract_temporal_features(df)
+        df = self._extract_engagement_features(df)
+        df = self._extract_text_features(df)
+        
+        # Vectorize texts (use agent result if available, otherwise fallback)
+        if agent_result and agent_result.get("vector_embeddings"):
+            # Extract vectors from agent graph result
+            agent_embeddings = agent_result.get("vector_embeddings", [])
+            embeddings = np.array([
+                item.get("vector", [0.0] * 768) for item in agent_embeddings
+            ])
+            logger.info(f"[DataTransformation] Using agent graph vectors: {embeddings.shape}")
+        else:
+            # Fallback to direct vectorization
+            embeddings = self._vectorize_texts(df)
+        
+        # Build combined feature matrix
+        feature_matrix = self._build_feature_matrix(df, embeddings)
+        
+        # Save outputs
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        
+        # Save transformed dataframe
+        transformed_path = Path(self.config.output_directory) / f"transformed_data_{timestamp}.parquet"
+        df.to_parquet(transformed_path, index=False)
+        
+        # Save embeddings
+        embeddings_path = Path(self.config.output_directory) / f"embeddings_{timestamp}.npy"
+        np.save(embeddings_path, embeddings)
+        
+        # Save feature matrix
+        features_path = Path(self.config.output_directory) / f"features_{timestamp}.npy"
+        np.save(features_path, feature_matrix)
+        
+        # Save agent graph insights if available
+        insights_path = None
+        if agent_result:
+            insights_path = Path(self.config.output_directory) / f"llm_insights_{timestamp}.json"
+            insights_data = {
+                "expert_summary": agent_result.get("expert_summary", ""),
+                "opportunities": agent_result.get("opportunities", []),
+                "threats": agent_result.get("threats", []),
+                "domain_insights": agent_result.get("domain_insights", []),
+                "processing_stats": agent_result.get("processing_stats", {})
+            }
+            with open(insights_path, "w", encoding="utf-8") as f:
+                json.dump(insights_data, f, indent=2, ensure_ascii=False)
+            logger.info(f"[DataTransformation] Saved LLM insights to {insights_path}")
+        
+        # Language distribution
+        lang_dist = df["language"].value_counts().to_dict()
+        
+        # Build report
+        report = {
+            "timestamp": timestamp,
+            "total_records": total_records,
+            "embedding_dim": embeddings.shape[1] if len(embeddings.shape) > 1 else 0,
+            "feature_dim": feature_matrix.shape[1],
+            "language_distribution": lang_dist,
+            "used_agent_graph": agent_result is not None,
+            "expert_summary_available": expert_summary is not None
+        }
+        
+        artifact = DataTransformationArtifact(
+            transformed_data_path=str(transformed_path),
+            vector_embeddings_path=str(embeddings_path),
+            feature_store_path=str(features_path),
+            total_records=total_records,
+            language_distribution=lang_dist,
+            transformation_report=report
+        )
+        
+        logger.info(f"[DataTransformation] ✓ Complete: {feature_matrix.shape}")
+        if agent_result:
+            logger.info(f"[DataTransformation] ✓ LLM Expert Summary: {len(expert_summary or '')} chars")
+        return artifact
+

models/anomaly-detection/src/components/data_validation.py

@@ -0,0 +1,261 @@
+"""
+models/anomaly-detection/src/components/data_validation.py
+Data validation component based on schema.yaml
+"""
+import os
+import yaml
+import pandas as pd
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, List, Dict, Any
+
+from ..entity import DataValidationConfig, DataValidationArtifact
+
+logger = logging.getLogger("data_validation")
+
+
+class DataValidation:
+    """
+    Data validation component that validates feed data against schema.
+    Checks column types, required fields, and value constraints.
+    """
+    
+    def __init__(self, config: Optional[DataValidationConfig] = None):
+        """
+        Initialize data validation component.
+        
+        Args:
+            config: Optional configuration, uses defaults if None
+        """
+        self.config = config or DataValidationConfig()
+        
+        # Ensure output directory exists
+        Path(self.config.output_directory).mkdir(parents=True, exist_ok=True)
+        
+        # Load schema
+        self.schema = self._load_schema()
+        
+        logger.info(f"[DataValidation] Initialized with schema: {self.config.schema_file}")
+    
+    def _load_schema(self) -> Dict[str, Any]:
+        """Load schema from YAML file"""
+        if not os.path.exists(self.config.schema_file):
+            logger.warning(f"[DataValidation] Schema file not found: {self.config.schema_file}")
+            return {}
+        
+        try:
+            with open(self.config.schema_file, 'r', encoding='utf-8') as f:
+                return yaml.safe_load(f)
+        except Exception as e:
+            logger.error(f"[DataValidation] Failed to load schema: {e}")
+            return {}
+    
+    def _validate_required_columns(self, df: pd.DataFrame) -> List[Dict[str, Any]]:
+        """
+        Check that all required columns are present.
+        
+        Returns:
+            List of validation errors
+        """
+        errors = []
+        
+        for col in self.config.required_columns:
+            if col not in df.columns:
+                errors.append({
+                    "type": "missing_column",
+                    "column": col,
+                    "message": f"Required column '{col}' is missing"
+                })
+        
+        return errors
+    
+    def _validate_column_types(self, df: pd.DataFrame) -> List[Dict[str, Any]]:
+        """
+        Validate column data types based on schema.
+        
+        Returns:
+            List of validation errors
+        """
+        errors = []
+        
+        if "feed_columns" not in self.schema:
+            return errors
+        
+        for col_name, col_spec in self.schema["feed_columns"].items():
+            if col_name not in df.columns:
+                continue
+            
+            expected_dtype = col_spec.get("dtype", "str")
+            
+            # Check for null values in required columns
+            if col_spec.get("required", False):
+                null_count = df[col_name].isna().sum()
+                if null_count > 0:
+                    errors.append({
+                        "type": "null_values",
+                        "column": col_name,
+                        "count": int(null_count),
+                        "message": f"Column '{col_name}' has {null_count} null values"
+                    })
+            
+            # Check min/max length for strings
+            if expected_dtype == "str" and col_name in df.columns:
+                min_len = col_spec.get("min_length", 0)
+                max_len = col_spec.get("max_length", float('inf'))
+                
+                if min_len > 0:
+                    short_count = (df[col_name].fillna("").str.len() < min_len).sum()
+                    if short_count > 0:
+                        errors.append({
+                            "type": "min_length_violation",
+                            "column": col_name,
+                            "count": int(short_count),
+                            "message": f"Column '{col_name}' has {short_count} values shorter than {min_len}"
+                        })
+            
+            # Check allowed values
+            allowed = col_spec.get("allowed_values")
+            if allowed and col_name in df.columns:
+                invalid_mask = ~df[col_name].isin(allowed) & df[col_name].notna()
+                invalid_count = invalid_mask.sum()
+                if invalid_count > 0:
+                    errors.append({
+                        "type": "invalid_value",
+                        "column": col_name,
+                        "count": int(invalid_count),
+                        "allowed": allowed,
+                        "message": f"Column '{col_name}' has {invalid_count} values not in allowed list"
+                    })
+        
+        return errors
+    
+    def _validate_numeric_ranges(self, df: pd.DataFrame) -> List[Dict[str, Any]]:
+        """
+        Validate numeric column ranges.
+        
+        Returns:
+            List of validation errors
+        """
+        errors = []
+        
+        if "feed_columns" not in self.schema:
+            return errors
+        
+        for col_name, col_spec in self.schema["feed_columns"].items():
+            if col_name not in df.columns:
+                continue
+            
+            expected_dtype = col_spec.get("dtype")
+            
+            if expected_dtype in ["int", "float"]:
+                min_val = col_spec.get("min_value")
+                max_val = col_spec.get("max_value")
+                
+                if min_val is not None:
+                    try:
+                        below_count = (pd.to_numeric(df[col_name], errors='coerce') < min_val).sum()
+                        if below_count > 0:
+                            errors.append({
+                                "type": "below_minimum",
+                                "column": col_name,
+                                "count": int(below_count),
+                                "min_value": min_val,
+                                "message": f"Column '{col_name}' has {below_count} values below {min_val}"
+                            })
+                    except Exception:
+                        pass
+                
+                if max_val is not None:
+                    try:
+                        above_count = (pd.to_numeric(df[col_name], errors='coerce') > max_val).sum()
+                        if above_count > 0:
+                            errors.append({
+                                "type": "above_maximum",
+                                "column": col_name,
+                                "count": int(above_count),
+                                "max_value": max_val,
+                                "message": f"Column '{col_name}' has {above_count} values above {max_val}"
+                            })
+                    except Exception:
+                        pass
+        
+        return errors
+    
+    def validate(self, data_path: str) -> DataValidationArtifact:
+        """
+        Execute data validation pipeline.
+        
+        Args:
+            data_path: Path to input data (parquet or csv)
+            
+        Returns:
+            DataValidationArtifact with validation results
+        """
+        logger.info(f"[DataValidation] Validating: {data_path}")
+        
+        # Load data
+        if data_path.endswith(".parquet"):
+            df = pd.read_parquet(data_path)
+        elif data_path.endswith(".csv"):
+            df = pd.read_csv(data_path)
+        else:
+            raise ValueError(f"Unsupported file format: {data_path}")
+        
+        total_records = len(df)
+        logger.info(f"[DataValidation] Loaded {total_records} records")
+        
+        # Run validations
+        all_errors = []
+        all_errors.extend(self._validate_required_columns(df))
+        all_errors.extend(self._validate_column_types(df))
+        all_errors.extend(self._validate_numeric_ranges(df))
+        
+        # Calculate valid/invalid records
+        invalid_records = 0
+        for error in all_errors:
+            if "count" in error:
+                invalid_records = max(invalid_records, error["count"])
+        
+        valid_records = total_records - invalid_records
+        validation_status = len(all_errors) == 0
+        
+        # Log validation results
+        if validation_status:
+            logger.info("[DataValidation] ✓ All validations passed")
+        else:
+            logger.warning(f"[DataValidation] ⚠ Found {len(all_errors)} validation issues")
+            for error in all_errors[:5]:  # Log first 5
+                logger.warning(f"  - {error['message']}")
+        
+        # Save validated data (even with warnings, we continue)
+        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
+        validated_path = Path(self.config.output_directory) / f"validated_data_{timestamp}.parquet"
+        df.to_parquet(validated_path, index=False)
+        
+        # Save validation report
+        report_path = Path(self.config.output_directory) / f"validation_report_{timestamp}.yaml"
+        report = {
+            "validation_timestamp": timestamp,
+            "input_path": data_path,
+            "total_records": total_records,
+            "valid_records": valid_records,
+            "invalid_records": invalid_records,
+            "validation_status": validation_status,
+            "errors": all_errors
+        }
+        with open(report_path, 'w') as f:
+            yaml.dump(report, f, default_flow_style=False)
+        
+        artifact = DataValidationArtifact(
+            validated_data_path=str(validated_path),
+            validation_report_path=str(report_path),
+            total_records=total_records,
+            valid_records=valid_records,
+            invalid_records=invalid_records,
+            validation_status=validation_status,
+            validation_errors=all_errors
+        )
+        
+        logger.info(f"[DataValidation] ✓ Complete: {valid_records}/{total_records} valid records")
+        return artifact

models/anomaly-detection/src/components/model_trainer.py

@@ -0,0 +1,478 @@
+"""
+models/anomaly-detection/src/components/model_trainer.py
+Model training with Optuna hyperparameter tuning for clustering/anomaly detection
+"""
+import os
+import logging
+import joblib
+from datetime import datetime
+from pathlib import Path
+from typing import Optional, Dict, Any, List
+import numpy as np
+
+from ..entity import ModelTrainerConfig, ModelTrainerArtifact
+from ..utils import calculate_clustering_metrics, calculate_optuna_objective, format_metrics_report
+
+logger = logging.getLogger("model_trainer")
+
+# MLflow
+try:
+    import mlflow
+    import mlflow.sklearn
+    MLFLOW_AVAILABLE = True
+except ImportError:
+    MLFLOW_AVAILABLE = False
+    logger.warning("MLflow not available. Install with: pip install mlflow")
+
+# Optuna
+try:
+    import optuna
+    from optuna.samplers import TPESampler
+    OPTUNA_AVAILABLE = True
+except ImportError:
+    OPTUNA_AVAILABLE = False
+    logger.warning("Optuna not available. Install with: pip install optuna")
+
+# Clustering algorithms
+try:
+    from sklearn.cluster import DBSCAN, KMeans
+    from sklearn.ensemble import IsolationForest
+    from sklearn.neighbors import LocalOutlierFactor
+    SKLEARN_AVAILABLE = True
+except ImportError:
+    SKLEARN_AVAILABLE = False
+
+try:
+    import hdbscan
+    HDBSCAN_AVAILABLE = True
+except ImportError:
+    HDBSCAN_AVAILABLE = False
+    logger.warning("HDBSCAN not available. Install with: pip install hdbscan")
+
+
+class ModelTrainer:
+    """
+    Model training component with:
+    1. Optuna hyperparameter optimization
+    2. Multiple clustering algorithms (DBSCAN, KMeans, HDBSCAN)
+    3. Anomaly detection (Isolation Forest, LOF)
+    4. MLflow experiment tracking
+    """
+    
+    def __init__(self, config: Optional[ModelTrainerConfig] = None):
+        """
+        Initialize model trainer.
+        
+        Args:
+            config: Optional configuration
+        """
+        self.config = config or ModelTrainerConfig()
+        
+        # Ensure output directory exists
+        Path(self.config.output_directory).mkdir(parents=True, exist_ok=True)
+        
+        # Setup MLflow
+        self._setup_mlflow()
+        
+        logger.info(f"[ModelTrainer] Initialized")
+        logger.info(f"  Models to train: {self.config.models_to_train}")
+        logger.info(f"  Optuna trials: {self.config.n_optuna_trials}")
+    
+    def _setup_mlflow(self):
+        """Configure MLflow tracking"""
+        if not MLFLOW_AVAILABLE:
+            logger.warning("[ModelTrainer] MLflow not available")
+            return
+        
+        try:
+            # Set tracking URI
+            mlflow.set_tracking_uri(self.config.mlflow_tracking_uri)
+            
+            # Set credentials for DagsHub
+            if self.config.mlflow_username and self.config.mlflow_password:
+                os.environ["MLFLOW_TRACKING_USERNAME"] = self.config.mlflow_username
+                os.environ["MLFLOW_TRACKING_PASSWORD"] = self.config.mlflow_password
+            
+            # Create or get experiment
+            try:
+                mlflow.create_experiment(self.config.experiment_name)
+            except Exception:
+                pass
+            mlflow.set_experiment(self.config.experiment_name)
+            
+            logger.info(f"[ModelTrainer] MLflow configured: {self.config.mlflow_tracking_uri}")
+            
+        except Exception as e:
+            logger.warning(f"[ModelTrainer] MLflow setup error: {e}")
+    
+    def _train_dbscan(self, X: np.ndarray, trial: Optional['optuna.Trial'] = None) -> Dict[str, Any]:
+        """
+        Train DBSCAN with optional Optuna tuning.
+        """
+        if not SKLEARN_AVAILABLE:
+            return {"error": "sklearn not available"}
+        
+        # Hyperparameters
+        if trial:
+            eps = trial.suggest_float("eps", 0.1, 2.0)
+            min_samples = trial.suggest_int("min_samples", 2, 20)
+        else:
+            eps = 0.5
+            min_samples = 5
+        
+        model = DBSCAN(eps=eps, min_samples=min_samples, n_jobs=-1)
+        labels = model.fit_predict(X)
+        
+        metrics = calculate_clustering_metrics(X, labels)
+        metrics["eps"] = eps
+        metrics["min_samples"] = min_samples
+        
+        return {
+            "model": model,
+            "labels": labels,
+            "metrics": metrics,
+            "params": {"eps": eps, "min_samples": min_samples}
+        }
+    
+    def _train_kmeans(self, X: np.ndarray, trial: Optional['optuna.Trial'] = None) -> Dict[str, Any]:
+        """
+        Train KMeans with optional Optuna tuning.
+        """
+        if not SKLEARN_AVAILABLE:
+            return {"error": "sklearn not available"}
+        
+        # Hyperparameters
+        if trial:
+            n_clusters = trial.suggest_int("n_clusters", 2, 20)
+            n_init = trial.suggest_int("n_init", 5, 20)
+        else:
+            n_clusters = 5
+            n_init = 10
+        
+        model = KMeans(n_clusters=n_clusters, n_init=n_init, random_state=42)
+        labels = model.fit_predict(X)
+        
+        metrics = calculate_clustering_metrics(X, labels)
+        metrics["n_clusters"] = n_clusters
+        
+        return {
+            "model": model,
+            "labels": labels,
+            "metrics": metrics,
+            "params": {"n_clusters": n_clusters, "n_init": n_init}
+        }
+    
+    def _train_hdbscan(self, X: np.ndarray, trial: Optional['optuna.Trial'] = None) -> Dict[str, Any]:
+        """
+        Train HDBSCAN with optional Optuna tuning.
+        """
+        if not HDBSCAN_AVAILABLE:
+            return {"error": "hdbscan not available"}
+        
+        # Hyperparameters
+        if trial:
+            min_cluster_size = trial.suggest_int("min_cluster_size", 5, 50)
+            min_samples = trial.suggest_int("min_samples", 1, 20)
+        else:
+            min_cluster_size = 15
+            min_samples = 5
+        
+        model = hdbscan.HDBSCAN(
+            min_cluster_size=min_cluster_size,
+            min_samples=min_samples,
+            core_dist_n_jobs=-1
+        )
+        labels = model.fit_predict(X)
+        
+        metrics = calculate_clustering_metrics(X, labels)
+        
+        return {
+            "model": model,
+            "labels": labels,
+            "metrics": metrics,
+            "params": {"min_cluster_size": min_cluster_size, "min_samples": min_samples}
+        }
+    
+    def _train_isolation_forest(self, X: np.ndarray, trial: Optional['optuna.Trial'] = None) -> Dict[str, Any]:
+        """
+        Train Isolation Forest for anomaly detection.
+        """
+        if not SKLEARN_AVAILABLE:
+            return {"error": "sklearn not available"}
+        
+        # Hyperparameters
+        if trial:
+            contamination = trial.suggest_float("contamination", 0.01, 0.3)
+            n_estimators = trial.suggest_int("n_estimators", 50, 200)
+        else:
+            contamination = 0.1
+            n_estimators = 100
+        
+        model = IsolationForest(
+            contamination=contamination,
+            n_estimators=n_estimators,
+            random_state=42,
+            n_jobs=-1
+        )
+        predictions = model.fit_predict(X)
+        labels = (predictions == -1).astype(int)  # -1 = anomaly
+        
+        n_anomalies = int(np.sum(labels))
+        
+        return {
+            "model": model,
+            "labels": labels,
+            "metrics": {
+                "n_anomalies": n_anomalies,
+                "anomaly_rate": n_anomalies / len(X),
+                "contamination": contamination,
+                "n_estimators": n_estimators
+            },
+            "params": {"contamination": contamination, "n_estimators": n_estimators},
+            "anomaly_indices": np.where(labels == 1)[0].tolist()
+        }
+    
+    def _train_lof(self, X: np.ndarray, trial: Optional['optuna.Trial'] = None) -> Dict[str, Any]:
+        """
+        Train Local Outlier Factor for anomaly detection.
+        """
+        if not SKLEARN_AVAILABLE:
+            return {"error": "sklearn not available"}
+        
+        # Hyperparameters
+        if trial:
+            n_neighbors = trial.suggest_int("n_neighbors", 5, 50)
+            contamination = trial.suggest_float("contamination", 0.01, 0.3)
+        else:
+            n_neighbors = 20
+            contamination = 0.1
+        
+        model = LocalOutlierFactor(
+            n_neighbors=n_neighbors,
+            contamination=contamination,
+            n_jobs=-1,
+            novelty=True  # For prediction on new data
+        )
+        model.fit(X)
+        predictions = model.predict(X)
+        labels = (predictions == -1).astype(int)  # -1 = anomaly
+        
+        n_anomalies = int(np.sum(labels))
+        
+        return {
+            "model": model,
+            "labels": labels,
+            "metrics": {
+                "n_anomalies": n_anomalies,
+                "anomaly_rate": n_anomalies / len(X),
+                "n_neighbors": n_neighbors,
+                "contamination": contamination
+            },
+            "params": {"n_neighbors": n_neighbors, "contamination": contamination},
+            "anomaly_indices": np.where(labels == 1)[0].tolist()
+        }
+    
+    def _optimize_model(self, model_name: str, X: np.ndarray) -> Dict[str, Any]:
+        """
+        Use Optuna to find best hyperparameters for a model.
+        """
+        if not OPTUNA_AVAILABLE:
+            logger.warning("[ModelTrainer] Optuna not available, using defaults")
+            return self._train_model(model_name, X, None)
+        
+        train_func = {
+            "dbscan": self._train_dbscan,
+            "kmeans": self._train_kmeans,
+            "hdbscan": self._train_hdbscan,
+            "isolation_forest": self._train_isolation_forest,
+            "lof": self._train_lof
+        }.get(model_name)
+        
+        if not train_func:
+            return {"error": f"Unknown model: {model_name}"}
+        
+        def objective(trial):
+            try:
+                result = train_func(X, trial)
+                if "error" in result:
+                    return -1.0
+                
+                metrics = result.get("metrics", {})
+                
+                # For clustering: use silhouette
+                if model_name in ["dbscan", "kmeans", "hdbscan"]:
+                    score = metrics.get("silhouette_score", -1)
+                    return score if score is not None else -1
+                
+                # For anomaly detection: balance anomaly rate
+                else:
+                    # Target anomaly rate around 5-15%
+                    rate = metrics.get("anomaly_rate", 0)
+                    target = 0.1
+                    return -abs(rate - target)  # Closer to target is better
+                    
+            except Exception as e:
+                logger.debug(f"Trial failed: {e}")
+                return -1.0
+        
+        # Create and run study
+        study = optuna.create_study(
+            direction="maximize",
+            sampler=TPESampler(seed=42)
+        )
+        
+        study.optimize(
+            objective,
+            n_trials=self.config.n_optuna_trials,
+            timeout=self.config.optuna_timeout_seconds,
+            show_progress_bar=True
+        )
+        
+        logger.info(f"[ModelTrainer] {model_name} best params: {study.best_params}")
+        logger.info(f"[ModelTrainer] {model_name} best score: {study.best_value:.4f}")
+        
+        # Train with best params
+        best_result = train_func(X, None)  # Use defaults as base
+        # Override with best params
+        if study.best_params:
+            # Re-train with best params would require custom logic
+            # For now, we just log the best params
+            best_result["best_params"] = study.best_params
+            best_result["best_score"] = study.best_value
+            best_result["study_name"] = study.study_name
+        
+        return best_result
+    
+    def _train_model(self, model_name: str, X: np.ndarray, trial=None) -> Dict[str, Any]:
+        """Train a single model"""
+        train_funcs = {
+            "dbscan": self._train_dbscan,
+            "kmeans": self._train_kmeans,
+            "hdbscan": self._train_hdbscan,
+            "isolation_forest": self._train_isolation_forest,
+            "lof": self._train_lof
+        }
+        
+        func = train_funcs.get(model_name)
+        if func:
+            return func(X, trial)
+        return {"error": f"Unknown model: {model_name}"}
+    
+    def train(self, feature_path: str) -> ModelTrainerArtifact:
+        """
+        Execute model training pipeline.
+        
+        Args:
+            feature_path: Path to feature matrix (.npy)
+            
+        Returns:
+            ModelTrainerArtifact with results
+        """
+        logger.info(f"[ModelTrainer] Starting training: {feature_path}")
+        start_time = datetime.now()
+        
+        # Load features
+        X = np.load(feature_path)
+        logger.info(f"[ModelTrainer] Loaded features: {X.shape}")
+        
+        # Start MLflow run
+        mlflow_run_id = ""
+        mlflow_experiment_id = ""
+        
+        if MLFLOW_AVAILABLE:
+            try:
+                run = mlflow.start_run()
+                mlflow_run_id = run.info.run_id
+                mlflow_experiment_id = run.info.experiment_id
+                
+                mlflow.log_param("n_samples", X.shape[0])
+                mlflow.log_param("n_features", X.shape[1])
+                mlflow.log_param("models", self.config.models_to_train)
+            except Exception as e:
+                logger.warning(f"[ModelTrainer] MLflow run start error: {e}")
+        
+        # Train all models
+        trained_models = []
+        best_model = None
+        best_score = -float('inf')
+        
+        for model_name in self.config.models_to_train:
+            logger.info(f"[ModelTrainer] Training {model_name}...")
+            
+            try:
+                result = self._optimize_model(model_name, X)
+                
+                if "error" in result:
+                    logger.warning(f"[ModelTrainer] {model_name} error: {result['error']}")
+                    continue
+                
+                # Save model
+                model_path = Path(self.config.output_directory) / f"{model_name}_model.joblib"
+                joblib.dump(result["model"], model_path)
+                
+                # Log to MLflow
+                if MLFLOW_AVAILABLE:
+                    try:
+                        mlflow.log_params({f"{model_name}_{k}": v for k, v in result.get("params", {}).items()})
+                        mlflow.log_metrics({f"{model_name}_{k}": v for k, v in result.get("metrics", {}).items() if isinstance(v, (int, float))})
+                        mlflow.sklearn.log_model(result["model"], model_name)
+                    except Exception as e:
+                        logger.debug(f"MLflow log error: {e}")
+                
+                # Track results
+                model_info = {
+                    "name": model_name,
+                    "path": str(model_path),
+                    "params": result.get("params", {}),
+                    "metrics": result.get("metrics", {})
+                }
+                trained_models.append(model_info)
+                
+                # Check if best (for clustering models)
+                score = result.get("metrics", {}).get("silhouette_score", -1)
+                if score and score > best_score:
+                    best_score = score
+                    best_model = model_info
+                
+                logger.info(f"[ModelTrainer] ✓ {model_name} complete")
+                
+            except Exception as e:
+                logger.error(f"[ModelTrainer] {model_name} failed: {e}")
+        
+        # End MLflow run
+        if MLFLOW_AVAILABLE:
+            try:
+                mlflow.end_run()
+            except Exception:
+                pass
+        
+        # Calculate duration
+        duration = (datetime.now() - start_time).total_seconds()
+        
+        # Get anomaly info from best anomaly detector
+        n_anomalies = None
+        anomaly_indices = None
+        for model_info in trained_models:
+            if model_info["name"] in ["isolation_forest", "lof"]:
+                n_anomalies = model_info["metrics"].get("n_anomalies")
+                break
+        
+        # Build artifact
+        artifact = ModelTrainerArtifact(
+            best_model_name=best_model["name"] if best_model else "",
+            best_model_path=best_model["path"] if best_model else "",
+            best_model_metrics=best_model["metrics"] if best_model else {},
+            trained_models=trained_models,
+            mlflow_run_id=mlflow_run_id,
+            mlflow_experiment_id=mlflow_experiment_id,
+            n_clusters=best_model["metrics"].get("n_clusters") if best_model else None,
+            n_anomalies=n_anomalies,
+            anomaly_indices=anomaly_indices,
+            training_duration_seconds=duration,
+            optuna_study_name=None
+        )
+        
+        logger.info(f"[ModelTrainer] ✓ Training complete in {duration:.1f}s")
+        logger.info(f"[ModelTrainer] Best model: {best_model['name'] if best_model else 'N/A'}")
+        
+        return artifact

models/anomaly-detection/src/entity/__init__.py

@@ -0,0 +1,30 @@
+"""
+models/anomaly-detection/src/entity/__init__.py
+"""
+from .config_entity import (
+    DataIngestionConfig,
+    DataValidationConfig,
+    DataTransformationConfig,
+    ModelTrainerConfig,
+    PipelineConfig
+)
+from .artifact_entity import (
+    DataIngestionArtifact,
+    DataValidationArtifact,
+    DataTransformationArtifact,
+    ModelTrainerArtifact,
+    PipelineArtifact
+)
+
+__all__ = [
+    "DataIngestionConfig",
+    "DataValidationConfig", 
+    "DataTransformationConfig",
+    "ModelTrainerConfig",
+    "PipelineConfig",
+    "DataIngestionArtifact",
+    "DataValidationArtifact",
+    "DataTransformationArtifact",
+    "ModelTrainerArtifact",
+    "PipelineArtifact"
+]

models/anomaly-detection/src/entity/artifact_entity.py

@@ -0,0 +1,79 @@
+"""
+models/anomaly-detection/src/entity/artifact_entity.py
+Artifact entities for pipeline outputs
+"""
+from dataclasses import dataclass
+from typing import List, Dict, Any, Optional
+from pathlib import Path
+
+
+@dataclass
+class DataIngestionArtifact:
+    """Artifact from data ingestion step"""
+    raw_data_path: str
+    total_records: int
+    records_from_sqlite: int
+    records_from_csv: int
+    ingestion_timestamp: str
+    is_data_available: bool
+
+
+@dataclass
+class DataValidationArtifact:
+    """Artifact from data validation step"""
+    validated_data_path: str
+    validation_report_path: str
+    total_records: int
+    valid_records: int
+    invalid_records: int
+    validation_status: bool
+    validation_errors: List[Dict[str, Any]]
+
+
+@dataclass
+class DataTransformationArtifact:
+    """Artifact from data transformation step"""
+    transformed_data_path: str
+    vector_embeddings_path: str
+    feature_store_path: str
+    total_records: int
+    language_distribution: Dict[str, int]
+    transformation_report: Dict[str, Any]
+
+
+@dataclass
+class ModelTrainerArtifact:
+    """Artifact from model training step"""
+    # Best model info
+    best_model_name: str
+    best_model_path: str
+    best_model_metrics: Dict[str, float]
+    
+    # All trained models
+    trained_models: List[Dict[str, Any]]
+    
+    # MLflow tracking
+    mlflow_run_id: str
+    mlflow_experiment_id: str
+    
+    # Cluster/anomaly results
+    n_clusters: Optional[int]
+    n_anomalies: Optional[int]
+    anomaly_indices: Optional[List[int]]
+    
+    # Training info
+    training_duration_seconds: float
+    optuna_study_name: Optional[str]
+
+
+@dataclass
+class PipelineArtifact:
+    """Complete pipeline artifact"""
+    data_ingestion: DataIngestionArtifact
+    data_validation: DataValidationArtifact
+    data_transformation: DataTransformationArtifact
+    model_trainer: ModelTrainerArtifact
+    pipeline_run_id: str
+    pipeline_start_time: str
+    pipeline_end_time: str
+    pipeline_status: str

models/anomaly-detection/src/entity/config_entity.py

@@ -0,0 +1,109 @@
+"""
+models/anomaly-detection/src/entity/config_entity.py
+Configuration entities for the anomaly detection pipeline
+"""
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import List, Optional
+import os
+
+
+@dataclass
+class DataIngestionConfig:
+    """Configuration for data ingestion component"""
+    sqlite_db_path: str = field(default_factory=lambda: os.getenv(
+        "SQLITE_DB_PATH",
+        str(Path(__file__).parent.parent.parent.parent.parent / "data" / "feeds" / "feed_cache.db")
+    ))
+    csv_directory: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent.parent.parent / "datasets" / "political_feeds"
+    ))
+    output_directory: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "artifacts" / "data_ingestion"
+    ))
+    batch_size: int = 1000
+    min_text_length: int = 10
+
+
+@dataclass
+class DataValidationConfig:
+    """Configuration for data validation component"""
+    schema_file: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "data_schema" / "schema.yaml"
+    ))
+    required_columns: List[str] = field(default_factory=lambda: [
+        "post_id", "timestamp", "platform", "category", "text", "content_hash"
+    ])
+    output_directory: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "artifacts" / "data_validation"
+    ))
+
+
+@dataclass
+class DataTransformationConfig:
+    """Configuration for data transformation/vectorization component"""
+    # Huggingface models - will be downloaded locally
+    models_cache_dir: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "models_cache"
+    ))
+    
+    # Language-specific BERT models
+    english_model: str = "distilbert-base-uncased"
+    sinhala_model: str = "keshan/SinhalaBERTo"
+    tamil_model: str = "l3cube-pune/tamil-bert"
+    
+    # Language detection
+    fasttext_model_path: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "models_cache" / "lid.176.bin"  # FastText language ID model
+    ))
+    
+    # Vector dimensions
+    vector_dim: int = 768  # Standard BERT dimension
+    
+    # Output
+    output_directory: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "artifacts" / "data_transformation"
+    ))
+
+
+@dataclass
+class ModelTrainerConfig:
+    """Configuration for model training component"""
+    # MLflow configuration
+    mlflow_tracking_uri: str = field(default_factory=lambda: os.getenv(
+        "MLFLOW_TRACKING_URI", "https://dagshub.com/sliitguy/SecurityNetwork.mlflow"
+    ))
+    mlflow_username: str = field(default_factory=lambda: os.getenv(
+        "MLFLOW_TRACKING_USERNAME", ""
+    ))
+    mlflow_password: str = field(default_factory=lambda: os.getenv(
+        "MLFLOW_TRACKING_PASSWORD", ""
+    ))
+    experiment_name: str = "anomaly_detection_feeds"
+    
+    # Model configurations
+    models_to_train: List[str] = field(default_factory=lambda: [
+        "dbscan", "kmeans", "hdbscan", "isolation_forest", "lof"
+    ])
+    
+    # Optuna hyperparameter tuning
+    n_optuna_trials: int = 50
+    optuna_timeout_seconds: int = 3600  # 1 hour
+    
+    # Model output
+    output_directory: str = field(default_factory=lambda: str(
+        Path(__file__).parent.parent.parent / "artifacts" / "model_trainer"
+    ))
+
+
+@dataclass
+class PipelineConfig:
+    """Master configuration for the entire pipeline"""
+    data_ingestion: DataIngestionConfig = field(default_factory=DataIngestionConfig)
+    data_validation: DataValidationConfig = field(default_factory=DataValidationConfig)
+    data_transformation: DataTransformationConfig = field(default_factory=DataTransformationConfig)
+    model_trainer: ModelTrainerConfig = field(default_factory=ModelTrainerConfig)
+    
+    # Pipeline settings
+    batch_threshold: int = 1000  # Trigger training after this many new records
+    run_interval_hours: int = 24  # Fallback daily run

models/anomaly-detection/src/pipeline/__init__.py

@@ -0,0 +1,6 @@
+"""
+models/anomaly-detection/src/pipeline/__init__.py
+"""
+from .training_pipeline import TrainingPipeline, run_training_pipeline
+
+__all__ = ["TrainingPipeline", "run_training_pipeline"]

models/anomaly-detection/src/pipeline/training_pipeline.py

@@ -0,0 +1,162 @@
+"""
+models/anomaly-detection/src/pipeline/training_pipeline.py
+End-to-end training pipeline orchestrator
+"""
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Optional
+
+from ..entity import (
+    PipelineConfig,
+    PipelineArtifact,
+    DataIngestionArtifact,
+    DataValidationArtifact,
+    DataTransformationArtifact,
+    ModelTrainerArtifact
+)
+from ..components import (
+    DataIngestion,
+    DataValidation,
+    DataTransformation,
+    ModelTrainer
+)
+
+logger = logging.getLogger("training_pipeline")
+
+
+class TrainingPipeline:
+    """
+    End-to-end training pipeline that orchestrates:
+    1. Data Ingestion (SQLite + CSV)
+    2. Data Validation (schema checking)
+    3. Data Transformation (language detection + vectorization)
+    4. Model Training (clustering + anomaly detection)
+    """
+    
+    def __init__(self, config: Optional[PipelineConfig] = None):
+        """
+        Initialize training pipeline.
+        
+        Args:
+            config: Optional pipeline configuration
+        """
+        self.config = config or PipelineConfig()
+        self.run_id = datetime.now().strftime("%Y%m%d_%H%M%S")
+        
+        logger.info(f"[TrainingPipeline] Initialized (run_id: {self.run_id})")
+    
+    def run_data_ingestion(self) -> DataIngestionArtifact:
+        """Execute data ingestion step"""
+        logger.info("=" * 50)
+        logger.info("[TrainingPipeline] STEP 1: Data Ingestion")
+        logger.info("=" * 50)
+        
+        ingestion = DataIngestion(self.config.data_ingestion)
+        artifact = ingestion.ingest()
+        
+        if not artifact.is_data_available:
+            raise ValueError("No data available for training")
+        
+        return artifact
+    
+    def run_data_validation(self, ingestion_artifact: DataIngestionArtifact) -> DataValidationArtifact:
+        """Execute data validation step"""
+        logger.info("=" * 50)
+        logger.info("[TrainingPipeline] STEP 2: Data Validation")
+        logger.info("=" * 50)
+        
+        validation = DataValidation(self.config.data_validation)
+        artifact = validation.validate(ingestion_artifact.raw_data_path)
+        
+        return artifact
+    
+    def run_data_transformation(self, validation_artifact: DataValidationArtifact) -> DataTransformationArtifact:
+        """Execute data transformation step"""
+        logger.info("=" * 50)
+        logger.info("[TrainingPipeline] STEP 3: Data Transformation")
+        logger.info("=" * 50)
+        
+        transformation = DataTransformation(self.config.data_transformation)
+        artifact = transformation.transform(validation_artifact.validated_data_path)
+        
+        return artifact
+    
+    def run_model_training(self, transformation_artifact: DataTransformationArtifact) -> ModelTrainerArtifact:
+        """Execute model training step"""
+        logger.info("=" * 50)
+        logger.info("[TrainingPipeline] STEP 4: Model Training")
+        logger.info("=" * 50)
+        
+        trainer = ModelTrainer(self.config.model_trainer)
+        artifact = trainer.train(transformation_artifact.feature_store_path)
+        
+        return artifact
+    
+    def run(self) -> PipelineArtifact:
+        """
+        Execute the complete training pipeline.
+        
+        Returns:
+            PipelineArtifact with all step results
+        """
+        start_time = datetime.now()
+        logger.info("=" * 60)
+        logger.info("[TrainingPipeline] STARTING TRAINING PIPELINE")
+        logger.info("=" * 60)
+        
+        try:
+            # Step 1: Data Ingestion
+            ingestion_artifact = self.run_data_ingestion()
+            
+            # Step 2: Data Validation
+            validation_artifact = self.run_data_validation(ingestion_artifact)
+            
+            # Step 3: Data Transformation
+            transformation_artifact = self.run_data_transformation(validation_artifact)
+            
+            # Step 4: Model Training
+            training_artifact = self.run_model_training(transformation_artifact)
+            
+            pipeline_status = "SUCCESS"
+            
+        except Exception as e:
+            logger.error(f"[TrainingPipeline] Pipeline failed: {e}")
+            pipeline_status = f"FAILED: {str(e)}"
+            raise
+        
+        finally:
+            end_time = datetime.now()
+            duration = (end_time - start_time).total_seconds()
+            logger.info("=" * 60)
+            logger.info(f"[TrainingPipeline] PIPELINE {pipeline_status}")
+            logger.info(f"[TrainingPipeline] Duration: {duration:.1f}s")
+            logger.info("=" * 60)
+        
+        # Build final artifact
+        artifact = PipelineArtifact(
+            data_ingestion=ingestion_artifact,
+            data_validation=validation_artifact,
+            data_transformation=transformation_artifact,
+            model_trainer=training_artifact,
+            pipeline_run_id=self.run_id,
+            pipeline_start_time=start_time.isoformat(),
+            pipeline_end_time=end_time.isoformat(),
+            pipeline_status=pipeline_status
+        )
+        
+        return artifact
+
+
+def run_training_pipeline(config: Optional[PipelineConfig] = None) -> PipelineArtifact:
+    """
+    Convenience function to run the training pipeline.
+    
+    Args:
+        config: Optional pipeline configuration
+        
+    Returns:
+        PipelineArtifact with results
+    """
+    pipeline = TrainingPipeline(config)
+    return pipeline.run()

models/anomaly-detection/src/utils/__init__.py

@@ -0,0 +1,24 @@
+"""
+models/anomaly-detection/src/utils/__init__.py
+"""
+from .language_detector import LanguageDetector, detect_language, get_detector
+from .vectorizer import MultilingualVectorizer, vectorize_text, get_vectorizer
+from .metrics import (
+    calculate_clustering_metrics,
+    calculate_anomaly_metrics,
+    calculate_optuna_objective,
+    format_metrics_report
+)
+
+__all__ = [
+    "LanguageDetector",
+    "detect_language",
+    "get_detector",
+    "MultilingualVectorizer",
+    "vectorize_text",
+    "get_vectorizer",
+    "calculate_clustering_metrics",
+    "calculate_anomaly_metrics",
+    "calculate_optuna_objective",
+    "format_metrics_report"
+]

models/anomaly-detection/src/utils/language_detector.py

@@ -0,0 +1,209 @@
+"""
+models/anomaly-detection/src/utils/language_detector.py
+Language detection using FastText or lingua-py for Sinhala/Tamil/English
+"""
+import os
+import logging
+from typing import Tuple, Optional
+from pathlib import Path
+import re
+
+logger = logging.getLogger("language_detector")
+
+# Try FastText first, fallback to lingua
+try:
+    import fasttext
+    fasttext.FastText.eprint = lambda x: None  # Suppress warnings
+    FASTTEXT_AVAILABLE = True
+except ImportError:
+    FASTTEXT_AVAILABLE = False
+    logger.warning("FastText not available. Install with: pip install fasttext")
+
+try:
+    from lingua import Language, LanguageDetectorBuilder
+    LINGUA_AVAILABLE = True
+except ImportError:
+    LINGUA_AVAILABLE = False
+    logger.warning("Lingua not available. Install with: pip install lingua-language-detector")
+
+
+class LanguageDetector:
+    """
+    Multilingual language detector supporting Sinhala, Tamil, and English.
+    Uses FastText as primary detector with lingua fallback.
+    """
+    
+    # Language code mapping
+    LANG_MAP = {
+        "en": "english",
+        "si": "sinhala",
+        "ta": "tamil",
+        "__label__en": "english",
+        "__label__si": "sinhala", 
+        "__label__ta": "tamil",
+        "ENGLISH": "english",
+        "SINHALA": "sinhala",
+        "TAMIL": "tamil"
+    }
+    
+    # Unicode ranges for script detection
+    SINHALA_RANGE = (0x0D80, 0x0DFF)
+    TAMIL_RANGE = (0x0B80, 0x0BFF)
+    
+    def __init__(self, models_cache_dir: Optional[str] = None):
+        """
+        Initialize language detector.
+        
+        Args:
+            models_cache_dir: Directory for cached FastText models
+        """
+        self.models_cache_dir = models_cache_dir or str(
+            Path(__file__).parent.parent.parent / "models_cache"
+        )
+        Path(self.models_cache_dir).mkdir(parents=True, exist_ok=True)
+        
+        self.fasttext_model = None
+        self.lingua_detector = None
+        
+        self._init_detectors()
+    
+    def _init_detectors(self):
+        """Initialize detection models"""
+        # Try FastText
+        if FASTTEXT_AVAILABLE:
+            model_path = Path(self.models_cache_dir) / "lid.176.bin"
+            if model_path.exists():
+                try:
+                    self.fasttext_model = fasttext.load_model(str(model_path))
+                    logger.info(f"[LanguageDetector] Loaded FastText model from {model_path}")
+                except Exception as e:
+                    logger.warning(f"[LanguageDetector] Failed to load FastText: {e}")
+            else:
+                logger.warning(f"[LanguageDetector] FastText model not found at {model_path}")
+                logger.info("Download from: https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.bin")
+        
+        # Initialize lingua as fallback
+        if LINGUA_AVAILABLE:
+            try:
+                self.lingua_detector = LanguageDetectorBuilder.from_languages(
+                    Language.ENGLISH,
+                    Language.TAMIL,
+                    # Note: Lingua may not have Sinhala, we'll use script detection
+                ).build()
+                logger.info("[LanguageDetector] Initialized Lingua detector")
+            except Exception as e:
+                logger.warning(f"[LanguageDetector] Failed to init Lingua: {e}")
+    
+    def _detect_by_script(self, text: str) -> Optional[str]:
+        """
+        Detect language by Unicode script analysis.
+        More reliable for Sinhala/Tamil which have distinct scripts.
+        """
+        sinhala_count = 0
+        tamil_count = 0
+        latin_count = 0
+        
+        for char in text:
+            code = ord(char)
+            if self.SINHALA_RANGE[0] <= code <= self.SINHALA_RANGE[1]:
+                sinhala_count += 1
+            elif self.TAMIL_RANGE[0] <= code <= self.TAMIL_RANGE[1]:
+                tamil_count += 1
+            elif char.isalpha() and code < 128:
+                latin_count += 1
+        
+        total_alpha = sinhala_count + tamil_count + latin_count
+        if total_alpha == 0:
+            return None
+        
+        # Threshold-based detection
+        if sinhala_count / total_alpha > 0.3:
+            return "sinhala"
+        if tamil_count / total_alpha > 0.3:
+            return "tamil"
+        if latin_count / total_alpha > 0.5:
+            return "english"
+        
+        return None
+    
+    def detect(self, text: str) -> Tuple[str, float]:
+        """
+        Detect language of text.
+        
+        Args:
+            text: Input text
+            
+        Returns:
+            Tuple of (language_code, confidence)
+            language_code: 'english', 'sinhala', 'tamil', or 'unknown'
+        """
+        if not text or len(text.strip()) < 3:
+            return "unknown", 0.0
+        
+        # Clean text
+        clean_text = re.sub(r'http\S+|@\w+|#\w+', '', text)
+        clean_text = clean_text.strip()
+        
+        if not clean_text:
+            return "unknown", 0.0
+        
+        # 1. First try script detection (most reliable for Sinhala/Tamil)
+        script_lang = self._detect_by_script(clean_text)
+        if script_lang in ["sinhala", "tamil"]:
+            return script_lang, 0.95
+        
+        # 2. Try FastText
+        if self.fasttext_model:
+            try:
+                predictions = self.fasttext_model.predict(clean_text.replace("\n", " "))
+                label = predictions[0][0]
+                confidence = predictions[1][0]
+                
+                lang = self.LANG_MAP.get(label, "unknown")
+                if lang != "unknown" and confidence > 0.5:
+                    return lang, float(confidence)
+            except Exception as e:
+                logger.debug(f"FastText error: {e}")
+        
+        # 3. Try Lingua
+        if self.lingua_detector:
+            try:
+                detected = self.lingua_detector.detect_language_of(clean_text)
+                if detected:
+                    lang = self.LANG_MAP.get(detected.name, "unknown")
+                    # Lingua doesn't return confidence, estimate based on text
+                    confidence = 0.8 if len(clean_text) > 20 else 0.6
+                    return lang, confidence
+            except Exception as e:
+                logger.debug(f"Lingua error: {e}")
+        
+        # 4. Fallback to script detection result or default
+        if script_lang == "english":
+            return "english", 0.7
+        
+        return "english", 0.5  # Default to English
+
+
+# Singleton instance
+_detector: Optional[LanguageDetector] = None
+
+
+def get_detector(models_cache_dir: Optional[str] = None) -> LanguageDetector:
+    """Get or create singleton detector instance"""
+    global _detector
+    if _detector is None:
+        _detector = LanguageDetector(models_cache_dir)
+    return _detector
+
+
+def detect_language(text: str) -> Tuple[str, float]:
+    """
+    Convenience function for language detection.
+    
+    Args:
+        text: Input text
+        
+    Returns:
+        Tuple of (language: str, confidence: float)
+    """
+    return get_detector().detect(text)

models/anomaly-detection/src/utils/metrics.py

@@ -0,0 +1,256 @@
+"""
+models/anomaly-detection/src/utils/metrics.py
+Clustering and anomaly detection metrics for model evaluation
+"""
+import numpy as np
+from typing import Dict, Any, Optional, List
+import logging
+
+logger = logging.getLogger("metrics")
+
+# Scikit-learn metrics
+try:
+    from sklearn.metrics import (
+        silhouette_score,
+        calinski_harabasz_score,
+        davies_bouldin_score,
+        adjusted_rand_score,
+        normalized_mutual_info_score
+    )
+    SKLEARN_AVAILABLE = True
+except ImportError:
+    SKLEARN_AVAILABLE = False
+    logger.warning("scikit-learn not available for metrics")
+
+
+def calculate_clustering_metrics(
+    X: np.ndarray,
+    labels: np.ndarray,
+    true_labels: Optional[np.ndarray] = None
+) -> Dict[str, float]:
+    """
+    Calculate comprehensive clustering quality metrics.
+    
+    Args:
+        X: Feature matrix (n_samples, n_features)
+        labels: Predicted cluster labels
+        true_labels: Optional ground truth labels for supervised metrics
+        
+    Returns:
+        Dict of metric_name -> metric_value
+    """
+    if not SKLEARN_AVAILABLE:
+        logger.warning("sklearn not available, returning empty metrics")
+        return {}
+    
+    metrics = {}
+    
+    # Filter out noise points (label=-1) for some metrics
+    valid_mask = labels >= 0
+    n_clusters = len(set(labels[valid_mask]))
+    
+    # Need at least 2 clusters and >1 samples for metrics
+    if n_clusters < 2 or np.sum(valid_mask) < 2:
+        metrics["n_clusters"] = n_clusters
+        metrics["n_noise_points"] = np.sum(labels == -1)
+        metrics["error"] = "insufficient_clusters"
+        return metrics
+    
+    # Internal metrics (don't need ground truth)
+    try:
+        # Silhouette Score: -1 (bad) to 1 (good)
+        # Measures how similar objects are to their own cluster vs other clusters
+        metrics["silhouette_score"] = float(silhouette_score(
+            X[valid_mask], labels[valid_mask]
+        ))
+    except Exception as e:
+        logger.debug(f"Silhouette score failed: {e}")
+        metrics["silhouette_score"] = None
+    
+    try:
+        # Calinski-Harabasz Index: Higher is better
+        # Ratio of between-cluster dispersion to within-cluster dispersion
+        metrics["calinski_harabasz_score"] = float(calinski_harabasz_score(
+            X[valid_mask], labels[valid_mask]
+        ))
+    except Exception as e:
+        logger.debug(f"Calinski-Harabasz failed: {e}")
+        metrics["calinski_harabasz_score"] = None
+    
+    try:
+        # Davies-Bouldin Index: Lower is better
+        # Average similarity between clusters
+        metrics["davies_bouldin_score"] = float(davies_bouldin_score(
+            X[valid_mask], labels[valid_mask]
+        ))
+    except Exception as e:
+        logger.debug(f"Davies-Bouldin failed: {e}")
+        metrics["davies_bouldin_score"] = None
+    
+    # Cluster statistics
+    metrics["n_clusters"] = n_clusters
+    metrics["n_samples"] = len(labels)
+    metrics["n_noise_points"] = int(np.sum(labels == -1))
+    metrics["noise_ratio"] = float(np.sum(labels == -1) / len(labels))
+    
+    # Cluster size statistics
+    cluster_sizes = [np.sum(labels == i) for i in range(n_clusters)]
+    metrics["min_cluster_size"] = int(min(cluster_sizes)) if cluster_sizes else 0
+    metrics["max_cluster_size"] = int(max(cluster_sizes)) if cluster_sizes else 0
+    metrics["mean_cluster_size"] = float(np.mean(cluster_sizes)) if cluster_sizes else 0
+    
+    # External metrics (if ground truth provided)
+    if true_labels is not None:
+        try:
+            # Adjusted Rand Index: -1 to 1, 1=perfect, 0=random
+            metrics["adjusted_rand_score"] = float(adjusted_rand_score(
+                true_labels, labels
+            ))
+        except Exception as e:
+            logger.debug(f"ARI failed: {e}")
+        
+        try:
+            # Normalized Mutual Information: 0 to 1, 1=perfect agreement
+            metrics["normalized_mutual_info"] = float(normalized_mutual_info_score(
+                true_labels, labels
+            ))
+        except Exception as e:
+            logger.debug(f"NMI failed: {e}")
+    
+    return metrics
+
+
+def calculate_anomaly_metrics(
+    labels: np.ndarray,
+    predicted_anomalies: np.ndarray,
+    true_anomalies: Optional[np.ndarray] = None
+) -> Dict[str, float]:
+    """
+    Calculate anomaly detection metrics.
+    
+    Args:
+        labels: Cluster labels or -1 for anomalies
+        predicted_anomalies: Boolean array of predicted anomaly flags
+        true_anomalies: Optional ground truth anomaly flags
+        
+    Returns:
+        Dict of metric_name -> metric_value
+    """
+    metrics = {}
+    
+    n_samples = len(labels)
+    n_predicted_anomalies = int(np.sum(predicted_anomalies))
+    
+    metrics["n_samples"] = n_samples
+    metrics["n_predicted_anomalies"] = n_predicted_anomalies
+    metrics["anomaly_rate"] = float(n_predicted_anomalies / n_samples) if n_samples > 0 else 0
+    
+    # If ground truth available, calculate precision/recall
+    if true_anomalies is not None:
+        n_true_anomalies = int(np.sum(true_anomalies))
+        
+        # True positives: predicted AND actual anomalies
+        tp = int(np.sum(predicted_anomalies & true_anomalies))
+        # False positives: predicted anomaly but not actual
+        fp = int(np.sum(predicted_anomalies & ~true_anomalies))
+        # False negatives: not predicted but actual anomaly
+        fn = int(np.sum(~predicted_anomalies & true_anomalies))
+        # True negatives
+        tn = int(np.sum(~predicted_anomalies & ~true_anomalies))
+        
+        metrics["true_positives"] = tp
+        metrics["false_positives"] = fp
+        metrics["false_negatives"] = fn
+        metrics["true_negatives"] = tn
+        
+        # Precision: TP / (TP + FP)
+        metrics["precision"] = float(tp / (tp + fp)) if (tp + fp) > 0 else 0
+        
+        # Recall: TP / (TP + FN)
+        metrics["recall"] = float(tp / (tp + fn)) if (tp + fn) > 0 else 0
+        
+        # F1 Score
+        if metrics["precision"] + metrics["recall"] > 0:
+            metrics["f1_score"] = float(
+                2 * metrics["precision"] * metrics["recall"] / 
+                (metrics["precision"] + metrics["recall"])
+            )
+        else:
+            metrics["f1_score"] = 0
+    
+    return metrics
+
+
+def calculate_optuna_objective(
+    X: np.ndarray,
+    labels: np.ndarray,
+    objective_type: str = "silhouette"
+) -> float:
+    """
+    Calculate objective value for Optuna optimization.
+    
+    Args:
+        X: Feature matrix
+        labels: Predicted labels
+        objective_type: 'silhouette', 'calinski', or 'combined'
+        
+    Returns:
+        Objective value (higher is better)
+    """
+    metrics = calculate_clustering_metrics(X, labels)
+    
+    # Check for errors
+    if "error" in metrics:
+        return -1.0  # Return bad score for failed clustering
+    
+    if objective_type == "silhouette":
+        score = metrics.get("silhouette_score")
+        return score if score is not None else -1.0
+    
+    elif objective_type == "calinski":
+        score = metrics.get("calinski_harabasz_score")
+        # Normalize to 0-1 range (approximate)
+        return min(score / 1000, 1.0) if score is not None else -1.0
+    
+    elif objective_type == "combined":
+        # Weighted combination of metrics
+        silhouette = metrics.get("silhouette_score", -1)
+        calinski = min(metrics.get("calinski_harabasz_score", 0) / 1000, 1)
+        davies = metrics.get("davies_bouldin_score", 10)
+        
+        # Davies-Bouldin is lower=better, invert it
+        davies_inv = 1 / (1 + davies) if davies is not None else 0
+        
+        # Weighted combination
+        combined = (0.4 * silhouette + 0.3 * calinski + 0.3 * davies_inv)
+        return float(combined)
+    
+    return -1.0
+
+
+def format_metrics_report(metrics: Dict[str, Any]) -> str:
+    """
+    Format metrics dictionary as a readable report.
+    
+    Args:
+        metrics: Dictionary of metric values
+        
+    Returns:
+        Formatted string report
+    """
+    lines = ["=" * 50]
+    lines.append("CLUSTERING METRICS REPORT")
+    lines.append("=" * 50)
+    
+    for key, value in metrics.items():
+        if value is None:
+            value_str = "N/A"
+        elif isinstance(value, float):
+            value_str = f"{value:.4f}"
+        else:
+            value_str = str(value)
+        
+        lines.append(f"{key:30s}: {value_str}")
+    
+    lines.append("=" * 50)
+    return "\n".join(lines)

models/anomaly-detection/src/utils/vectorizer.py

@@ -0,0 +1,243 @@
+"""
+models/anomaly-detection/src/utils/vectorizer.py
+Text vectorization using language-specific BERT models (downloaded locally)
+"""
+import os
+import logging
+from typing import Dict, List, Optional, Tuple
+from pathlib import Path
+import numpy as np
+
+logger = logging.getLogger("vectorizer")
+
+# Transformers
+try:
+    from transformers import AutoTokenizer, AutoModel
+    import torch
+    TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    TRANSFORMERS_AVAILABLE = False
+    logger.warning("Transformers not available. Install with: pip install transformers torch")
+
+# Sentence Transformers for fallback
+try:
+    from sentence_transformers import SentenceTransformer
+    SENTENCE_TRANSFORMERS_AVAILABLE = True
+except ImportError:
+    SENTENCE_TRANSFORMERS_AVAILABLE = False
+
+
+class MultilingualVectorizer:
+    """
+    Vectorizer using language-specific BERT models.
+    Downloads and caches models locally from HuggingFace.
+    
+    Models:
+    - English: distilbert-base-uncased (fast, accurate)
+    - Sinhala: keshan/SinhalaBERTo (specialized)
+    - Tamil: l3cube-pune/tamil-bert (specialized)
+    """
+    
+    MODEL_MAP = {
+        "english": "distilbert-base-uncased",
+        "sinhala": "keshan/SinhalaBERTo",
+        "tamil": "l3cube-pune/tamil-bert"
+    }
+    
+    def __init__(self, models_cache_dir: Optional[str] = None, device: Optional[str] = None):
+        """
+        Initialize the multilingual vectorizer.
+        
+        Args:
+            models_cache_dir: Directory to cache downloaded models
+            device: 'cuda' or 'cpu' (auto-detected if None)
+        """
+        self.models_cache_dir = models_cache_dir or str(
+            Path(__file__).parent.parent.parent / "models_cache"
+        )
+        Path(self.models_cache_dir).mkdir(parents=True, exist_ok=True)
+        
+        # Set cache dir for HuggingFace
+        os.environ["TRANSFORMERS_CACHE"] = self.models_cache_dir
+        os.environ["HF_HOME"] = self.models_cache_dir
+        
+        # Auto-detect device
+        if device is None:
+            if TRANSFORMERS_AVAILABLE and torch.cuda.is_available():
+                self.device = "cuda"
+            else:
+                self.device = "cpu"
+        else:
+            self.device = device
+        
+        logger.info(f"[Vectorizer] Using device: {self.device}")
+        
+        # Lazy load models
+        self.models: Dict[str, Tuple] = {}  # {lang: (tokenizer, model)}
+        self.fallback_model = None
+        
+    def _load_model(self, language: str) -> Tuple:
+        """
+        Load language-specific model from cache or download.
+        
+        Returns:
+            Tuple of (tokenizer, model)
+        """
+        if language in self.models:
+            return self.models[language]
+        
+        model_name = self.MODEL_MAP.get(language, self.MODEL_MAP["english"])
+        
+        if not TRANSFORMERS_AVAILABLE:
+            raise RuntimeError("Transformers library not available")
+        
+        logger.info(f"[Vectorizer] Loading model: {model_name}")
+        
+        try:
+            tokenizer = AutoTokenizer.from_pretrained(
+                model_name,
+                cache_dir=self.models_cache_dir
+            )
+            model = AutoModel.from_pretrained(
+                model_name,
+                cache_dir=self.models_cache_dir
+            ).to(self.device)
+            model.eval()
+            
+            self.models[language] = (tokenizer, model)
+            logger.info(f"[Vectorizer] ✓ Loaded {model_name} ({language})")
+            return tokenizer, model
+            
+        except Exception as e:
+            logger.error(f"[Vectorizer] Failed to load {model_name}: {e}")
+            # Fallback to English model
+            if language != "english":
+                logger.info("[Vectorizer] Falling back to English model")
+                return self._load_model("english")
+            raise
+    
+    def _get_embedding(self, text: str, tokenizer, model) -> np.ndarray:
+        """
+        Get embedding vector using mean pooling.
+        
+        Args:
+            text: Input text
+            tokenizer: HuggingFace tokenizer
+            model: HuggingFace model
+            
+        Returns:
+            768-dim numpy array
+        """
+        if not TRANSFORMERS_AVAILABLE:
+            raise RuntimeError("Transformers not available")
+        
+        # Tokenize
+        inputs = tokenizer(
+            text,
+            return_tensors="pt",
+            truncation=True,
+            max_length=512,
+            padding=True
+        ).to(self.device)
+        
+        # Get embeddings
+        with torch.no_grad():
+            outputs = model(**inputs)
+        
+        # Mean pooling over sequence length
+        attention_mask = inputs["attention_mask"]
+        hidden_states = outputs.last_hidden_state
+        
+        # Mask and average
+        mask_expanded = attention_mask.unsqueeze(-1).expand(hidden_states.size()).float()
+        sum_embeddings = torch.sum(hidden_states * mask_expanded, 1)
+        sum_mask = torch.clamp(mask_expanded.sum(1), min=1e-9)
+        mean_embedding = sum_embeddings / sum_mask
+        
+        return mean_embedding.cpu().numpy().flatten()
+    
+    def vectorize(self, text: str, language: str = "english") -> np.ndarray:
+        """
+        Convert text to vector embedding.
+        
+        Args:
+            text: Input text
+            language: 'english', 'sinhala', 'tamil', or 'unknown'
+            
+        Returns:
+            768-dim numpy array
+        """
+        if not text or not text.strip():
+            return np.zeros(768)
+        
+        # Map unknown to english
+        if language == "unknown":
+            language = "english"
+        
+        try:
+            tokenizer, model = self._load_model(language)
+            return self._get_embedding(text, tokenizer, model)
+        except Exception as e:
+            logger.error(f"[Vectorizer] Error vectorizing: {e}")
+            # Return zeros as fallback
+            return np.zeros(768)
+    
+    def vectorize_batch(
+        self, 
+        texts: List[str], 
+        languages: Optional[List[str]] = None
+    ) -> np.ndarray:
+        """
+        Batch vectorization for multiple texts.
+        
+        Args:
+            texts: List of text strings
+            languages: Optional list of language codes (same length as texts)
+            
+        Returns:
+            numpy array of shape (n_texts, 768)
+        """
+        if languages is None:
+            languages = ["english"] * len(texts)
+        
+        embeddings = []
+        for text, lang in zip(texts, languages):
+            emb = self.vectorize(text, lang)
+            embeddings.append(emb)
+        
+        return np.array(embeddings)
+    
+    def download_all_models(self):
+        """Pre-download all language models"""
+        for language in self.MODEL_MAP.keys():
+            try:
+                logger.info(f"[Vectorizer] Pre-downloading {language} model...")
+                self._load_model(language)
+            except Exception as e:
+                logger.warning(f"[Vectorizer] Failed to download {language}: {e}")
+
+
+# Singleton instance
+_vectorizer: Optional[MultilingualVectorizer] = None
+
+
+def get_vectorizer(models_cache_dir: Optional[str] = None) -> MultilingualVectorizer:
+    """Get or create singleton vectorizer instance"""
+    global _vectorizer
+    if _vectorizer is None:
+        _vectorizer = MultilingualVectorizer(models_cache_dir)
+    return _vectorizer
+
+
+def vectorize_text(text: str, language: str = "english") -> np.ndarray:
+    """
+    Convenience function for text vectorization.
+    
+    Args:
+        text: Input text
+        language: Language code
+        
+    Returns:
+        768-dim numpy array
+    """
+    return get_vectorizer().vectorize(text, language)

models/anomaly-detection/tests/dags/test_dag_example.py

@@ -0,0 +1,83 @@
+"""Example DAGs test. This test ensures that all Dags have tags, retries set to two, and no import errors. This is an example pytest and may not be fit the context of your DAGs. Feel free to add and remove tests."""
+
+import os
+import logging
+from contextlib import contextmanager
+import pytest
+from airflow.models import DagBag
+
+
+@contextmanager
+def suppress_logging(namespace):
+    logger = logging.getLogger(namespace)
+    old_value = logger.disabled
+    logger.disabled = True
+    try:
+        yield
+    finally:
+        logger.disabled = old_value
+
+
+def get_import_errors():
+    """
+    Generate a tuple for import errors in the dag bag
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+        def strip_path_prefix(path):
+            return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+        # prepend "(None,None)" to ensure that a test object is always created even if it's a no op.
+        return [(None, None)] + [
+            (strip_path_prefix(k), v.strip()) for k, v in dag_bag.import_errors.items()
+        ]
+
+
+def get_dags():
+    """
+    Generate a tuple of dag_id, <DAG objects> in the DagBag
+    """
+    with suppress_logging("airflow"):
+        dag_bag = DagBag(include_examples=False)
+
+    def strip_path_prefix(path):
+        return os.path.relpath(path, os.environ.get("AIRFLOW_HOME"))
+
+    return [(k, v, strip_path_prefix(v.fileloc)) for k, v in dag_bag.dags.items()]
+
+
+@pytest.mark.parametrize(
+    "rel_path,rv", get_import_errors(), ids=[x[0] for x in get_import_errors()]
+)
+def test_file_imports(rel_path, rv):
+    """Test for import errors on a file"""
+    if rel_path and rv:
+        raise Exception(f"{rel_path} failed to import with message \n {rv}")
+
+
+APPROVED_TAGS = {}
+
+
+@pytest.mark.parametrize(
+    "dag_id,dag,fileloc", get_dags(), ids=[x[2] for x in get_dags()]
+)
+def test_dag_tags(dag_id, dag, fileloc):
+    """
+    test if a DAG is tagged and if those TAGs are in the approved list
+    """
+    assert dag.tags, f"{dag_id} in {fileloc} has no tags"
+    if APPROVED_TAGS:
+        assert not set(dag.tags) - APPROVED_TAGS
+
+
+@pytest.mark.parametrize(
+    "dag_id,dag, fileloc", get_dags(), ids=[x[2] for x in get_dags()]
+)
+def test_dag_retries(dag_id, dag, fileloc):
+    """
+    test if a DAG has retries set
+    """
+    assert (
+        dag.default_args.get("retries", None) >= 2
+    ), f"{dag_id} in {fileloc} must have task retries >= 2."

models/currency-volatility-prediction/.gitignore

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

models/currency-volatility-prediction/.python-version

@@ -0,0 +1 @@
+3.11

models/currency-volatility-prediction/Dockerfile

@@ -0,0 +1,8 @@
+FROM python:3.10-slim-bookworm
+WORKDIR /app
+COPY . /app
+
+RUN apt update -y && apt install awscli -y
+
+RUN apt-get update && pip install -r requirements.txt
+CMD ["python3", "app.py"]

models/currency-volatility-prediction/dags/currency_prediction_dag.py

@@ -0,0 +1,212 @@
+"""
+models/currency-volatility-prediction/dags/currency_prediction_dag.py
+Airflow DAG for daily USD/LKR currency prediction
+Runs at 4:00 AM IST daily
+"""
+import os
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+# Add paths for imports
+PIPELINE_ROOT = Path(__file__).parent.parent
+sys.path.insert(0, str(PIPELINE_ROOT / "src"))
+
+from airflow import DAG
+from airflow.operators.python import PythonOperator
+
+# Load environment variables from root .env
+try:
+    from dotenv import load_dotenv
+    # Path: dags/ -> currency-volatility-prediction/ -> models/ -> root/
+    env_path = Path(__file__).parent.parent.parent.parent / ".env"
+    if env_path.exists():
+        load_dotenv(env_path)
+        print(f"[MLflow] ✓ Loaded env from {env_path}")
+except ImportError:
+    pass
+
+
+# Default arguments
+default_args = {
+    "owner": "modelx",
+    "depends_on_past": False,
+    "email_on_failure": False,
+    "email_on_retry": False,
+    "retries": 2,
+    "retry_delay": timedelta(minutes=5),
+}
+
+
+def ingest_data(**context):
+    """Task: Ingest currency data from yfinance."""
+    from components.data_ingestion import CurrencyDataIngestion
+    from entity.config_entity import DataIngestionConfig
+    
+    print("[CURRENCY DAG] Starting data ingestion...")
+    
+    config = DataIngestionConfig(history_period="2y")
+    ingestion = CurrencyDataIngestion(config)
+    
+    # Check if we have recent data
+    try:
+        df = ingestion.load_existing()
+        latest_date = df["date"].max()
+        if isinstance(latest_date, str):
+            latest_date = datetime.strptime(latest_date, "%Y-%m-%d")
+        
+        days_old = (datetime.now() - latest_date).days
+        
+        if days_old < 1:
+            print(f"[CURRENCY DAG] Data is current ({days_old} days old)")
+            context["ti"].xcom_push(key="data_path", value=str(ingestion.config.raw_data_dir))
+            return str(ingestion.config.raw_data_dir)
+    except FileNotFoundError:
+        pass
+    
+    # Full ingestion
+    data_path = ingestion.ingest_all()
+    context["ti"].xcom_push(key="data_path", value=data_path)
+    
+    print(f"[CURRENCY DAG] ✓ Data saved to {data_path}")
+    return data_path
+
+
+def train_model(**context):
+    """Task: Train GRU model."""
+    from components.model_trainer import CurrencyGRUTrainer
+    from components.data_ingestion import CurrencyDataIngestion
+    from entity.config_entity import ModelTrainerConfig
+    
+    print("[CURRENCY DAG] Starting model training...")
+    
+    # Load data
+    ingestion = CurrencyDataIngestion()
+    df = ingestion.load_existing()
+    
+    print(f"[CURRENCY DAG] Loaded {len(df)} records")
+    
+    # Train
+    config = ModelTrainerConfig(
+        epochs=100,
+        batch_size=16,
+        early_stopping_patience=15
+    )
+    trainer = CurrencyGRUTrainer(config)
+    
+    results = trainer.train(df=df, use_mlflow=True)
+    
+    print(f"[CURRENCY DAG] ✓ Training complete!")
+    print(f"  MAE: {results['test_mae']:.4f} LKR")
+    print(f"  Direction Accuracy: {results['direction_accuracy']*100:.1f}%")
+    
+    context["ti"].xcom_push(key="model_path", value=results["model_path"])
+    return results
+
+
+def generate_prediction(**context):
+    """Task: Generate next-day prediction."""
+    from components.predictor import CurrencyPredictor
+    from components.data_ingestion import CurrencyDataIngestion
+    
+    print("[CURRENCY DAG] Generating prediction...")
+    
+    predictor = CurrencyPredictor()
+    
+    try:
+        # Load latest data
+        ingestion = CurrencyDataIngestion()
+        df = ingestion.load_existing()
+        
+        # Generate prediction
+        prediction = predictor.predict(df)
+        
+    except FileNotFoundError:
+        # Model not trained, use fallback
+        print("[CURRENCY DAG] Model not trained, using fallback")
+        prediction = predictor.generate_fallback_prediction()
+    except Exception as e:
+        print(f"[CURRENCY DAG] Error predicting: {e}")
+        prediction = predictor.generate_fallback_prediction()
+    
+    # Save prediction
+    output_path = predictor.save_prediction(prediction)
+    
+    print(f"[CURRENCY DAG] ✓ Prediction generated!")
+    print(f"  Current: {prediction['current_rate']} LKR/USD")
+    print(f"  Predicted: {prediction['predicted_rate']} LKR/USD")
+    print(f"  Change: {prediction['expected_change_pct']:+.2f}%")
+    print(f"  Direction: {prediction['direction']}")
+    
+    context["ti"].xcom_push(key="prediction_path", value=output_path)
+    return prediction
+
+
+def publish_prediction(**context):
+    """Task: Log prediction summary."""
+    prediction = context["ti"].xcom_pull(task_ids="generate_prediction")
+    
+    if prediction:
+        print("\n" + "="*50)
+        print(f"USD/LKR CURRENCY PREDICTION")
+        print("="*50)
+        print(f"Prediction for: {prediction.get('prediction_date')}")
+        print(f"Current Rate: {prediction.get('current_rate')} LKR/USD")
+        print(f"Predicted Rate: {prediction.get('predicted_rate')} LKR/USD")
+        print(f"Expected Change: {prediction.get('expected_change_pct'):+.3f}%")
+        print(f"Direction: {prediction.get('direction_emoji')} {prediction.get('direction')}")
+        print(f"Volatility: {prediction.get('volatility_class')}")
+        if prediction.get('is_fallback'):
+            print("⚠️  Using fallback model")
+        print("="*50 + "\n")
+    
+    return True
+
+
+# Define DAG
+with DAG(
+    dag_id="currency_prediction_daily",
+    default_args=default_args,
+    description="Daily USD/LKR currency prediction using GRU neural network",
+    schedule_interval="0 4 * * *",  # 4:00 AM daily (IST is UTC+5:30)
+    start_date=datetime(2024, 12, 1),
+    catchup=False,
+    tags=["currency", "ml", "prediction", "gru", "forex"],
+) as dag:
+    
+    # Task 1: Ingest Data
+    task_ingest = PythonOperator(
+        task_id="ingest_data",
+        python_callable=ingest_data,
+        provide_context=True,
+    )
+    
+    # Task 2: Train Model
+    task_train = PythonOperator(
+        task_id="train_model",
+        python_callable=train_model,
+        provide_context=True,
+    )
+    
+    # Task 3: Generate Prediction
+    task_predict = PythonOperator(
+        task_id="generate_prediction",
+        python_callable=generate_prediction,
+        provide_context=True,
+    )
+    
+    # Task 4: Publish Prediction
+    task_publish = PythonOperator(
+        task_id="publish_prediction",
+        python_callable=publish_prediction,
+        provide_context=True,
+    )
+    
+    # Dependencies
+    task_ingest >> task_train >> task_predict >> task_publish
+
+
+if __name__ == "__main__":
+    print("Currency Prediction DAG loaded successfully")
+    print(f"Schedule: Daily at 4:00 AM")
+    print(f"Tasks: {[t.task_id for t in dag.tasks]}")

models/currency-volatility-prediction/main.py

@@ -0,0 +1,179 @@
+"""
+models/currency-volatility-prediction/main.py
+Entry point for Currency Prediction Pipeline
+Can run data collection, training, or prediction independently
+"""
+import os
+import sys
+import logging
+import argparse
+from pathlib import Path
+from datetime import datetime
+
+# Setup paths
+PIPELINE_ROOT = Path(__file__).parent
+sys.path.insert(0, str(PIPELINE_ROOT / "src"))
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger("currency_prediction")
+
+
+def run_data_ingestion(period: str = "2y"):
+    """Run data ingestion from yfinance."""
+    from components.data_ingestion import CurrencyDataIngestion
+    from entity.config_entity import DataIngestionConfig
+    
+    logger.info(f"Starting data ingestion ({period})...")
+    
+    config = DataIngestionConfig(history_period=period)
+    ingestion = CurrencyDataIngestion(config)
+    
+    data_path = ingestion.ingest_all()
+    
+    df = ingestion.load_existing(data_path)
+    
+    logger.info("Data Ingestion Complete!")
+    logger.info(f"Total records: {len(df)}")
+    logger.info(f"Features: {len(df.columns)}")
+    logger.info(f"Date range: {df['date'].min()} to {df['date'].max()}")
+    logger.info(f"Latest rate: {df['close'].iloc[-1]:.2f} LKR/USD")
+    
+    return data_path
+
+
+def run_training(epochs: int = 100):
+    """Run GRU model training."""
+    from components.data_ingestion import CurrencyDataIngestion
+    from components.model_trainer import CurrencyGRUTrainer
+    from entity.config_entity import ModelTrainerConfig
+    
+    logger.info("Starting model training...")
+    
+    # Load data
+    ingestion = CurrencyDataIngestion()
+    df = ingestion.load_existing()
+    
+    logger.info(f"Loaded {len(df)} records with {len(df.columns)} features")
+    
+    # Train
+    config = ModelTrainerConfig(epochs=epochs)
+    trainer = CurrencyGRUTrainer(config)
+    
+    results = trainer.train(df=df, use_mlflow=True)
+    
+    logger.info(f"\nTraining Results:")
+    logger.info(f"  MAE: {results['test_mae']:.4f} LKR")
+    logger.info(f"  RMSE: {results['rmse']:.4f} LKR")
+    logger.info(f"  Direction Accuracy: {results['direction_accuracy']*100:.1f}%")
+    logger.info(f"  Epochs: {results['epochs_trained']}")
+    logger.info(f"  Model saved: {results['model_path']}")
+    
+    return results
+
+
+def run_prediction():
+    """Run prediction for next day."""
+    from components.data_ingestion import CurrencyDataIngestion
+    from components.predictor import CurrencyPredictor
+    
+    logger.info("Generating prediction...")
+    
+    predictor = CurrencyPredictor()
+    
+    try:
+        ingestion = CurrencyDataIngestion()
+        df = ingestion.load_existing()
+        prediction = predictor.predict(df)
+    except FileNotFoundError:
+        logger.warning("Model not trained, using fallback")
+        prediction = predictor.generate_fallback_prediction()
+    except Exception as e:
+        logger.error(f"Error: {e}")
+        prediction = predictor.generate_fallback_prediction()
+    
+    output_path = predictor.save_prediction(prediction)
+    
+    # Display
+    logger.info(f"\n{'='*50}")
+    logger.info(f"USD/LKR PREDICTION FOR {prediction['prediction_date']}")
+    logger.info(f"{'='*50}")
+    logger.info(f"Current Rate:   {prediction['current_rate']:.2f} LKR/USD")
+    logger.info(f"Predicted Rate: {prediction['predicted_rate']:.2f} LKR/USD")
+    logger.info(f"Expected Change: {prediction['expected_change_pct']:+.3f}%")
+    logger.info(f"Direction: {prediction['direction_emoji']} LKR {prediction['direction']}")
+    logger.info(f"Volatility: {prediction['volatility_class']}")
+    
+    if prediction.get('weekly_trend'):
+        logger.info(f"Weekly Trend: {prediction['weekly_trend']:+.2f}%")
+    if prediction.get('monthly_trend'):
+        logger.info(f"Monthly Trend: {prediction['monthly_trend']:+.2f}%")
+    
+    logger.info(f"{'='*50}")
+    logger.info(f"Saved to: {output_path}")
+    
+    return prediction
+
+
+def run_full_pipeline():
+    """Run the complete pipeline: ingest → train → predict."""
+    logger.info("=" * 60)
+    logger.info("CURRENCY PREDICTION PIPELINE - FULL RUN")
+    logger.info("=" * 60)
+    
+    # Step 1: Data Ingestion
+    try:
+        run_data_ingestion(period="2y")
+    except Exception as e:
+        logger.error(f"Data ingestion failed: {e}")
+        return None
+    
+    # Step 2: Training
+    try:
+        run_training(epochs=100)
+    except Exception as e:
+        logger.error(f"Training failed: {e}")
+    
+    # Step 3: Prediction
+    prediction = run_prediction()
+    
+    logger.info("=" * 60)
+    logger.info("PIPELINE COMPLETE!")
+    logger.info("=" * 60)
+    
+    return prediction
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Currency Prediction Pipeline")
+    parser.add_argument(
+        "--mode",
+        choices=["ingest", "train", "predict", "full"],
+        default="predict",
+        help="Pipeline mode to run"
+    )
+    parser.add_argument(
+        "--period",
+        type=str,
+        default="2y",
+        help="Data period (1y, 2y, 5y)"
+    )
+    parser.add_argument(
+        "--epochs",
+        type=int,
+        default=100,
+        help="Training epochs"
+    )
+    
+    args = parser.parse_args()
+    
+    if args.mode == "ingest":
+        run_data_ingestion(period=args.period)
+    elif args.mode == "train":
+        run_training(epochs=args.epochs)
+    elif args.mode == "predict":
+        run_prediction()
+    elif args.mode == "full":
+        run_full_pipeline()

models/currency-volatility-prediction/pyproject.toml

@@ -0,0 +1,20 @@
+[project]
+name = "modelx-final-models"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "certifi>=2025.11.12",
+    "dagshub>=0.6.3",
+    "fastapi>=0.122.0",
+    "mlflow>=3.6.0",
+    "numpy>=2.3.5",
+    "pandas>=2.3.3",
+    "pyaml>=25.7.0",
+    "pymongo[srv]>=4.15.4",
+    "python-dotenv>=1.2.1",
+    "python-multipart>=0.0.20",
+    "scikit-learn>=1.7.2",
+    "uvicorn>=0.38.0",
+]

models/currency-volatility-prediction/requirements.txt

@@ -0,0 +1,15 @@
+python-dotenv
+pandas
+numpy 
+pymongo
+certifi
+pymongo[srv]
+scikit-learn
+mlflow
+pyaml
+dagshub
+fastapi
+uvicorn
+python-multipart
+
+#-e .

models/currency-volatility-prediction/setup.py

@@ -0,0 +1,47 @@
+
+'''
+The setup.py file is an essential part of packaging and 
+distributing Python projects. It is used by setuptools
+(or for distutils in older Python versions) to define the configuration
+of your project, such as its metadata, dependencies, and more
+'''
+
+from setuptools import find_packages, setup 
+# this scans through all the folders and gets the folders that has the __init__ file
+# setup is reponsible of providing all the information about the project
+
+from typing import List
+
+def get_requirements()->List[str]:
+    """
+    This function will return a list of requirements
+    """
+    requirement_lst:List[str]=[]
+    try:
+        with open('requirements.txt', 'r') as file:
+            # Read lines from the file
+            lines=file.readlines()
+            ## Process each line
+            for line in lines:
+                requirement=line.strip()
+                ## Ignore empty lines and -e .
+                
+                if requirement and requirement != '-e .':
+                    requirement_lst.append(requirement)
+
+
+
+    except FileNotFoundError:
+        print("requirements.txt file not found")
+
+    return requirement_lst
+
+setup(
+    name="NetworkSecurity",
+    version="0.0.1", # This can be changed later
+    author="Nivakaran S.",
+    author_email="nivakaran@hotmail.com",
+    packages=find_packages(),
+    install_requires=get_requirements()
+)
+

models/currency-volatility-prediction/src/__init__.py

@@ -0,0 +1,21 @@
+import logging
+import os 
+from datetime import datetime
+
+LOG_FILE=f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
+
+logs_path=os.path.join(os.getcwd(), "logs", LOG_FILE)
+
+os.makedirs(logs_path, exist_ok=True) 
+# Create the file only if it is not created
+
+LOG_FILE_PATH=os.path.join(logs_path, LOG_FILE)
+
+logging.basicConfig(
+    filename=LOG_FILE_PATH,
+    format="[ %(asctime)s ] %(lineno)d %(name)s - %(levelname)s - %(message)s",
+    level=logging.INFO 
+)
+
+
+