Upload folder using huggingface_hub

.gitattributes

@@ -36,3 +36,36 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 ModelX[[:space:]]Final[[:space:]]Problem.pdf filter=lfs diff=lfs merge=lfs -text
 trending_detection_visualization.png filter=lfs diff=lfs merge=lfs -text
 vectorizer_anomaly_visualization.png filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/BrowserMetrics-spare.pma filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Crashpad/reports/1bb2b465-675d-47f0-b953-a844af38ce6b.dmp filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Crashpad/reports/55792d7f-8397-4730-8518-c50a507a611a.dmp filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Crashpad/reports/880fc1e0-3241-4d76-a26b-0f9d6135dcd6.dmp filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/data_2 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/data_3 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/f_000002 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/f_000003 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/f_000006 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/f_00000b filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Cache/Cache_Data/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Code[[:space:]]Cache/js/3d01be7861bd5850_0 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Code[[:space:]]Cache/js/4f0cb78a57ef4137_0 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Code[[:space:]]Cache/js/aaeed4cfeb9c324a_0 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Code[[:space:]]Cache/js/bc082d8e612dbd10_0 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Code[[:space:]]Cache/js/e3df1293cf5ee96e_0 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/DawnGraphiteCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/DawnGraphiteCache/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/DawnWebGPUCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/DawnWebGPUCache/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/GPUCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/GPUCache/data_2 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/GPUCache/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/History filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/Default/Web[[:space:]]Data filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/GrShaderCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/GrShaderCache/data_3 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/GrShaderCache/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/GraphiteDawnCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/GraphiteDawnCache/index filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/ShaderCache/data_1 filter=lfs diff=lfs merge=lfs -text
+src/utils/.browser_data/linkedin/ShaderCache/index filter=lfs diff=lfs merge=lfs -text

README.md

@@ -11,7 +11,7 @@ pinned: false
 
 **Real-Time Situational Awareness for Sri Lanka**
 
-A multi-agent AI system that aggregates intelligence from 47+ data sources to provide risk analysis and opportunity detection for businesses operating in Sri Lanka.
+A multi-agent AI system that aggregates intelligence from **50+ data sources** to provide risk analysis and opportunity detection for businesses operating in Sri Lanka.
 
 ## 🌐 Live Demo
 
@@ -24,14 +24,14 @@ A multi-agent AI system that aggregates intelligence from 47+ data sources to pr
 
 ## 🎯 Key Features
 
-✅ **8 Domain Agents** running in parallel:
-- Social Media Monitor (Reddit, Twitter, Facebook, Threads, BlueSky)
-- Political Intelligence (Gazette, Parliament, District Social Media)
-- Economic Analysis (CSE Stock Market + Technical Indicators)
-- Meteorological Alerts (DMC Weather + RiverNet + **FloodWatch Integration**)
-- Intelligence Agent (Brand Monitoring + Threat Detection + **User-Configurable Targets**)
-- Data Retrieval Orchestrator (Web Scraping)
-- Vectorization Agent (Multilingual BERT Embeddings + Anomaly Detection)
+✅ **5 Domain Agents + 2 Orchestrators** running in parallel:
+- **Social Agent** - Reddit, Twitter, Facebook, Threads, BlueSky monitoring
+- **Political Agent** - Gazette, Parliament, District Social Media
+- **Economical Agent** - CSE Stock Market + Technical Indicators (SMA, EMA, RSI, MACD)
+- **Meteorological Agent** - DMC Weather + RiverNet + **FloodWatch Integration**
+- **Intelligence Agent** - Brand Monitoring + Threat Detection + **User-Configurable Targets**
+- **Combined Agent (Orchestrator)** - Fan-out/Fan-in coordination, LLM filtering, feed ranking
+- **Data Retrieval Agent** - Web scraping orchestration with anti-bot features
 
 ✅ **Situational Awareness Dashboard**:
 - **CEB Power Status** - Load shedding / power outage monitoring
@@ -109,6 +109,13 @@ A multi-agent AI system that aggregates intelligence from 47+ data sources to pr
 - Supports: Western, Southern, Central, Northern, Eastern, Sabaragamuwa, Uva, North Western, North Central provinces
 - Both frontend (MapView, DistrictInfoPanel) and backend are synchronized
 
+✅ **3-Tier Storage Architecture** with Deduplication:
+- **Tier 1: SQLite** - Fast hash-based exact match (microseconds)
+- **Tier 2: ChromaDB** - Semantic similarity search with sentence transformers (milliseconds)
+- **Tier 3: Neo4j Aura** - Knowledge graph for event relationships and entity tracking
+- Unified `StorageManager` orchestrates all backends
+- Deduplication prevents duplicate feeds across all domain agents
+
 ---
 
 ## 🏗️ System Architecture
@@ -185,6 +192,40 @@ graph TD
 - **Non-Blocking Refresh**: 60-second cycle with interruptible sleep
   - `threading.Event.wait()` instead of blocking `time.sleep()`
 
+### Storage Data Flow
+
+```
+┌─────────────────────────────────────────────────────────────────────────────┐
+│                         DOMAIN AGENTS (Parallel)                            │
+│  ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────────┐      │
+│  │ Social   │ │Political │ │Economic  │ │  Meteo   │ │ Intelligence │      │
+│  │ Agent    │ │ Agent    │ │ Agent    │ │  Agent   │ │    Agent     │      │
+│  └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ └──────┬───────┘      │
+│       └────────────┴────────────┴────────────┴──────────────┘              │
+│                                 │ Fan-In                                    │
+│                    ┌────────────▼─────────────┐                            │
+│                    │   CombinedAgentNode      │                            │
+│                    │   (LLM Filter + Rank)    │                            │
+│                    └────────────┬─────────────┘                            │
+└─────────────────────────────────┼───────────────────────────────────────────┘
+                                  │
+                    ┌─────────────▼──────────────┐
+                    │      StorageManager        │
+                    │   (3-Tier Deduplication)   │
+                    └─────────────┬──────────────┘
+          ┌───────────────────────┼──────────────────────────┐
+          │                       │                          │
+          ▼                       ▼                          ▼
+┌─────────────────┐    ┌──────────────────┐    ┌─────────────────────────┐
+│     SQLite      │    │    ChromaDB      │    │      Neo4j Aura         │
+│   (Fast Cache)  │    │  (Vector Store)  │    │   (Knowledge Graph)     │
+│  ─────────────  │    │  ──────────────  │    │  ───────────────────    │
+│  Hash-based     │    │  Semantic search │    │  Event relationships    │
+│  Exact match    │    │  Similarity 0.85 │    │  Domain nodes           │
+│  ~microseconds  │    │  ~milliseconds   │    │  Entity tracking        │
+└─────────────────┘    └──────────────────┘    └─────────────────────────┘
+```
+
 ---
 
 ### 2. Political Agent Graph (`politicalAgentGraph.py`)
@@ -870,9 +911,20 @@ Roger-Ultimate/
 # LLM
 GROQ_API_KEY=your_groq_key
 
-# Database
-MONGO_DB_URL=mongodb+srv://...
-SQLITE_DB_PATH=./feed_cache.db
+# Neo4j (Knowledge Graph)
+NEO4J_URI=neo4j+s://your-instance.databases.neo4j.io
+NEO4J_USERNAME=neo4j
+NEO4J_PASSWORD=your_password
+NEO4J_ENABLED=true
+NEO4J_DATABASE=neo4j
+
+# ChromaDB (Vector Store)
+CHROMADB_PATH=./data/chromadb
+CHROMADB_COLLECTION=Roger_feeds
+CHROMADB_SIMILARITY_THRESHOLD=0.85
+
+# SQLite (Fast Cache)
+SQLITE_DB_PATH=./data/cache/feeds.db
 
 # MLflow (DagsHub)
 MLFLOW_TRACKING_URI=https://dagshub.com/...

frontend/app/components/dashboard/EconomicIndicators.tsx

@@ -2,7 +2,7 @@
 
 import { Card } from "../ui/card";
 import { Badge } from "../ui/badge";
-import { TrendingUp, TrendingDown, Minus, Landmark, DollarSign, Percent, Building2 } from "lucide-react";
+import { TrendingUp, TrendingDown, Minus, Landmark, DollarSign, Percent, Building2, Radio } from "lucide-react";
 
 interface EconomicIndicatorsProps {
     economyData?: Record<string, unknown> | null;
@@ -15,6 +15,7 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
     const exchangeRate = indicators?.exchange_rate || {};
     const forexReserves = indicators?.forex_reserves || {};
     const dataAsOf = economyData?.data_as_of as string;
+    const scrapeStatus = economyData?.scrape_status as string;
 
     const getTrendIcon = (trend: string) => {
         if (trend === "improving" || trend === "stable") return <TrendingUp className="w-3 h-3 text-success" />;
@@ -22,6 +23,14 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
         return <Minus className="w-3 h-3 text-muted-foreground" />;
     };
 
+    // Get the exchange rate - prefer mid rate, fallback to sell or buy
+    const usdLkr = (exchangeRate.usd_lkr as number) ||
+        (exchangeRate.usd_lkr_sell as number) ||
+        (exchangeRate.usd_lkr_buy as number) || 0;
+
+    // Get policy rate - prefer overnight, fallback to SDFR
+    const policyRate = (policyRates.overnight_rate as number) || (policyRates.sdfr as number) || 0;
+
     return (
         <Card className="p-4 bg-card border-border">
             <div className="flex items-center justify-between mb-3">
@@ -34,9 +43,17 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
                         <p className="text-xs text-muted-foreground">CBSL Indicators</p>
                     </div>
                 </div>
-                <Badge className="bg-muted text-muted-foreground">
-                    {dataAsOf || "Latest"}
-                </Badge>
+                <div className="flex items-center gap-1">
+                    {scrapeStatus === "live" && (
+                        <Badge className="bg-success/20 text-success text-xs flex items-center gap-1">
+                            <Radio className="w-2 h-2 animate-pulse" />
+                            LIVE
+                        </Badge>
+                    )}
+                    <Badge className="bg-muted text-muted-foreground">
+                        {dataAsOf || "Latest"}
+                    </Badge>
+                </div>
             </div>
 
             <div className="grid grid-cols-2 gap-2">
@@ -44,7 +61,7 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
                 <div className="p-2 rounded-lg bg-muted/30 border border-border">
                     <div className="flex items-center gap-1 mb-1">
                         <Percent className="w-3 h-3 text-muted-foreground" />
-                        <span className="text-xs text-muted-foreground">Inflation (YoY)</span>
+                        <span className="text-xs text-muted-foreground">CCPI Inflation</span>
                     </div>
                     <div className="flex items-center gap-1">
                         <span className="text-lg font-bold">{inflation.ccpi_yoy as number || 0}%</span>
@@ -59,18 +76,25 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
                         <span className="text-xs text-muted-foreground">USD/LKR</span>
                     </div>
                     <div className="flex items-center gap-1">
-                        <span className="text-lg font-bold">{exchangeRate.usd_lkr as number || 0}</span>
+                        <span className="text-lg font-bold">{usdLkr.toFixed(2)}</span>
                         {getTrendIcon(exchangeRate.trend as string)}
                     </div>
+                    {/* Show Buy/Sell if available */}
+                    {((exchangeRate.usd_lkr_buy as number | undefined) || (exchangeRate.usd_lkr_sell as number | undefined)) && (
+                        <p className="text-xs text-muted-foreground mt-0.5">
+                            Buy: {((exchangeRate.usd_lkr_buy as number | undefined)?.toFixed(2)) || "—"} |
+                            Sell: {((exchangeRate.usd_lkr_sell as number | undefined)?.toFixed(2)) || "—"}
+                        </p>
+                    )}
                 </div>
 
                 {/* Policy Rate */}
                 <div className="p-2 rounded-lg bg-muted/30 border border-border">
                     <div className="flex items-center gap-1 mb-1">
                         <Landmark className="w-3 h-3 text-muted-foreground" />
-                        <span className="text-xs text-muted-foreground">SDFR Rate</span>
+                        <span className="text-xs text-muted-foreground">Policy Rate</span>
                     </div>
-                    <span className="text-lg font-bold">{policyRates.sdfr as number || 0}%</span>
+                    <span className="text-lg font-bold">{policyRate}%</span>
                 </div>
 
                 {/* Forex Reserves */}
@@ -80,7 +104,7 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
                         <span className="text-xs text-muted-foreground">Reserves</span>
                     </div>
                     <div className="flex items-center gap-1">
-                        <span className="text-lg font-bold">${forexReserves.value as number || 0}B</span>
+                        <span className="text-lg font-bold">${(forexReserves.value as number) || 0}B</span>
                         {getTrendIcon(forexReserves.trend as string)}
                     </div>
                 </div>
@@ -94,3 +118,4 @@ const EconomicIndicators = ({ economyData }: EconomicIndicatorsProps) => {
 };
 
 export default EconomicIndicators;
+

frontend/app/components/dashboard/TrendingTopics.tsx

@@ -26,10 +26,12 @@ export const TrendingTopics: React.FC = () => {
     const [loading, setLoading] = useState(true);
     const [error, setError] = useState<string | null>(null);
 
+    const API_BASE = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000';
+
     useEffect(() => {
         const fetchTrending = async () => {
             try {
-                const response = await fetch('/api/trending');
+                const response = await fetch(`${API_BASE}/api/trending`);
                 const result = await response.json();
                 setData(result);
                 setError(null);

frontend/app/hooks/use-roger-data.ts

@@ -11,9 +11,10 @@ const API_BASE = process.env.NEXT_PUBLIC_API_URL || 'http://localhost:8000';
 const WS_URL = API_BASE.replace('http', 'ws') + '/ws';
 
 // Timeouts for resilient connection
-const RECONNECT_DELAY = 3000;
+const RECONNECT_DELAY = 1000;  // Reduced from 3s to 1s for faster recovery
 const MAX_LOADING_TIME = 120000; // 2 minutes max loading time
-const INITIAL_FETCH_DELAY = 2000; // Fetch from REST after 2s if no WS data
+const INITIAL_FETCH_DELAY = 1000; // Fetch from REST after 1s if no WS data
+const FALLBACK_POLL_INTERVAL = 2000; // Poll REST every 2s when WS disconnected
 
 export interface RogerEvent {
   event_id: string;
@@ -96,6 +97,7 @@ export function useRogerData() {
   const wsRef = useRef<WebSocket | null>(null);
   const loadingTimeoutRef = useRef<NodeJS.Timeout | null>(null);
   const initialFetchDoneRef = useRef(false);
+  const lastDataTimeRef = useRef<number>(Date.now()); // Track when we last got data
 
   // Fetch rivernet data
   const fetchRiverData = useCallback(async () => {
@@ -213,9 +215,12 @@ export function useRogerData() {
         };
 
         websocket.onclose = () => {
-          console.log('[Roger] WebSocket disconnected. Reconnecting in 3s...');
+          console.log('[Roger] WebSocket disconnected. Reconnecting in 1s...');
           setIsConnected(false);
 
+          // IMMEDIATELY fetch from REST to prevent blank UI
+          fetchInitialData();
+
           // Reconnect after delay
           reconnectTimeout = setTimeout(() => {
             connect();
@@ -288,12 +293,13 @@ export function useRogerData() {
     }
   }, [isConnected]);
 
-  // Fallback polling if WebSocket fails
+  // Fallback polling if WebSocket fails - more aggressive when disconnected
   useEffect(() => {
     if (isConnected) return;
 
-    const interval = setInterval(fetchData, 5000);
-    fetchData(); // Initial fetch
+    console.log('[Roger] WebSocket disconnected - starting aggressive REST polling');
+    const interval = setInterval(fetchData, FALLBACK_POLL_INTERVAL);
+    fetchData(); // Initial fetch immediately
 
     return () => clearInterval(interval);
   }, [isConnected, fetchData]);

frontend/app/pages/Index.tsx

@@ -10,6 +10,7 @@ import WeatherPredictions from "../components/dashboard/WeatherPredictions";
 import CurrencyPrediction from "../components/dashboard/CurrencyPrediction";
 import NationalThreatCard from "../components/dashboard/NationalThreatCard";
 import HistoricalIntel from "../components/dashboard/HistoricalIntel";
+import TrendingTopics from "../components/dashboard/TrendingTopics";
 import SatelliteView from "../components/map/SatelliteView";
 import LoadingScreen from "../components/LoadingScreen";
 import { Activity, Map, Radio, BarChart3, Zap, Brain, Cloud, DollarSign, Satellite } from "lucide-react";
@@ -119,6 +120,7 @@ const Index = () => {
 
           <TabsContent value="overview" className="space-y-6 animate-fade-in">
             <DashboardOverview />
+            <TrendingTopics />
             <div className="grid grid-cols-1 lg:grid-cols-2 gap-6">
               <StockPredictions />
               <CurrencyPrediction />

main.py

@@ -15,7 +15,7 @@ from pydantic import BaseModel
 from typing import Dict, Any, List, Set, Optional
 import asyncio
 import json
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 import sys
 import os
 import logging
@@ -23,6 +23,12 @@ import threading
 import time
 import uuid  # CRITICAL: Was missing, needed for event_id generation
 
+
+def utc_now() -> datetime:
+    """Return current UTC time (Python 3.12+ compatible)."""
+    return datetime.now(timezone.utc)
+
+
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..')))
 
 from src.graphs.combinedAgentGraph import graph
@@ -183,7 +189,7 @@ current_state: Dict[str, Any] = {
         "avg_confidence": 0.0,
         "high_priority_count": 0,
         "total_events": 0,
-        "last_updated": datetime.utcnow().isoformat()
+        "last_updated": utc_now().isoformat()
     },
     "run_count": 0,
     "status": "initializing",
@@ -200,12 +206,12 @@ main_event_loop = None
 # Storage manager
 storage_manager = StorageManager()
 
-# WebSocket settings - RESILIENT for long scraping operations (60s+ graph cycles)
-# Increased intervals to prevent disconnections during lengthy scraping
-HEARTBEAT_INTERVAL = 45.0  # Send ping every 45s (was 25s)
-HEARTBEAT_TIMEOUT = 30.0   # Wait 30s for pong (was 10s) 
-HEARTBEAT_MISS_THRESHOLD = 4  # Allow 4 misses (was 3) = ~3 minutes tolerance
-SEND_TIMEOUT = 10.0  # Increased from 5s
+# WebSocket settings - ULTRA-RESILIENT for long scraping operations
+# Heavy graph cycles can take 2-3 minutes, so we need high tolerance
+HEARTBEAT_INTERVAL = 60.0  # Send ping every 60s (increased from 45s)
+HEARTBEAT_TIMEOUT = 45.0   # Wait 45s for pong (increased from 30s) 
+HEARTBEAT_MISS_THRESHOLD = 5  # Allow 5 misses = ~5 minutes tolerance
+SEND_TIMEOUT = 15.0  # Increased for slow networks/heavy load
 
 class ConnectionManager:
     """Manages active WebSocket with heartbeat"""
@@ -218,7 +224,7 @@ class ConnectionManager:
         async with self._lock:
             meta = {
                 "heartbeat_task": asyncio.create_task(self._heartbeat_loop(websocket)),
-                "last_pong": datetime.utcnow(),
+                "last_pong": utc_now(),
                 "misses": 0
             }
             self.active_connections[websocket] = meta
@@ -276,7 +282,7 @@ class ConnectionManager:
                             if meta is None:
                                 return
                             last_pong = meta.get("last_pong")
-                            if last_pong and (datetime.utcnow() - last_pong).total_seconds() < (HEARTBEAT_INTERVAL + HEARTBEAT_TIMEOUT):
+                            if last_pong and (utc_now() - last_pong).total_seconds() < (HEARTBEAT_INTERVAL + HEARTBEAT_TIMEOUT):
                                 pong_received = True
                                 meta['misses'] = 0
                                 break
@@ -463,7 +469,7 @@ def run_graph_loop():
                             severity = event_data.get("severity", "medium")
                             impact_type = event_data.get("impact_type", "risk")
                             confidence = event_data.get("confidence_score", event_data.get("confidence", 0.5))
-                            timestamp = event_data.get("timestamp", datetime.utcnow().isoformat())
+                            timestamp = event_data.get("timestamp", utc_now().isoformat())
 
                             # Check for duplicates
                             is_dup, _, _ = storage_manager.is_duplicate(summary)
@@ -525,7 +531,7 @@ async def database_polling_loop():
     Runs concurrently with graph thread.
     """
     global current_state
-    last_check = datetime.utcnow()
+    last_check = utc_now()
 
     logger.info("[DB_POLLER] Starting database polling loop")
 
@@ -535,7 +541,7 @@ async def database_polling_loop():
 
             # Get new feeds since last check
             new_feeds = storage_manager.get_feeds_since(last_check)
-            last_check = datetime.utcnow()
+            last_check = utc_now()
 
             if new_feeds:
                 logger.info(f"[DB_POLLER] Found {len(new_feeds)} new feeds")
@@ -556,7 +562,7 @@ async def database_polling_loop():
                     current_state['final_ranked_feed'] = unique_feeds + current_state.get('final_ranked_feed', [])
                     current_state['final_ranked_feed'] = current_state['final_ranked_feed'][:100]  # Keep last 100
                     current_state['status'] = 'operational'
-                    current_state['last_update'] = datetime.utcnow().isoformat()
+                    current_state['last_update'] = utc_now().isoformat()
 
                     # Mark first run as complete (frontend loading screen can now hide)
                     if not current_state.get('first_run_complete'):
@@ -775,6 +781,116 @@ def get_national_threat_score():
             "error": str(e)
         }
 
+# ============================================
+# INTEL CONFIG API - User Keywords & Profiles
+# ============================================
+
+# Global intel config (loaded from file)
+INTEL_CONFIG_PATH = os.path.join(os.path.dirname(__file__), "data", "intel_config.json")
+
+# Default config structure
+DEFAULT_INTEL_CONFIG = {
+    "user_profiles": {
+        "twitter": [],
+        "facebook": [],
+        "linkedin": []
+    },
+    "user_keywords": [],
+    "user_products": []
+}
+
+
+def load_intel_config() -> dict:
+    """Load intel config from JSON file."""
+    try:
+        if os.path.exists(INTEL_CONFIG_PATH):
+            with open(INTEL_CONFIG_PATH, "r", encoding="utf-8") as f:
+                return json.load(f)
+    except Exception as e:
+        logger.warning(f"[Intel Config] Error loading config: {e}")
+    return DEFAULT_INTEL_CONFIG.copy()
+
+
+def save_intel_config(config: dict) -> bool:
+    """Save intel config to JSON file."""
+    try:
+        os.makedirs(os.path.dirname(INTEL_CONFIG_PATH), exist_ok=True)
+        with open(INTEL_CONFIG_PATH, "w", encoding="utf-8") as f:
+            json.dump(config, f, indent=2, ensure_ascii=False)
+        return True
+    except Exception as e:
+        logger.error(f"[Intel Config] Error saving config: {e}")
+        return False
+
+
+# Load config on startup
+intel_config = load_intel_config()
+
+
+@app.get("/api/intel/config")
+def get_intel_config():
+    """
+    Get current intelligence configuration.
+    
+    Returns user-defined keywords, products, and social profiles to monitor.
+    """
+    global intel_config
+    intel_config = load_intel_config()  # Refresh from file
+    return {
+        "status": "success",
+        "config": intel_config
+    }
+
+
+class IntelConfigUpdate(BaseModel):
+    user_profiles: dict = None
+    user_keywords: list = None
+    user_products: list = None
+
+
+@app.post("/api/intel/config")
+def update_intel_config(config_update: IntelConfigUpdate):
+    """
+    Update intelligence configuration.
+    
+    Accepts user-defined keywords, products, and social profiles.
+    Changes take effect on the next agent collection cycle.
+    """
+    global intel_config
+    
+    try:
+        # Update fields if provided
+        if config_update.user_profiles is not None:
+            intel_config["user_profiles"] = config_update.user_profiles
+        if config_update.user_keywords is not None:
+            intel_config["user_keywords"] = config_update.user_keywords
+        if config_update.user_products is not None:
+            intel_config["user_products"] = config_update.user_products
+        
+        # Save to file
+        if save_intel_config(intel_config):
+            logger.info(f"[Intel Config] Updated: {len(intel_config.get('user_keywords', []))} keywords, "
+                       f"{sum(len(v) for v in intel_config.get('user_profiles', {}).values())} profiles")
+            return {
+                "status": "updated",
+                "config": intel_config
+            }
+        else:
+            return {"status": "error", "error": "Failed to save configuration"}
+    except Exception as e:
+        logger.error(f"[Intel Config] Update error: {e}")
+        return {"status": "error", "error": str(e)}
+
+
+def get_user_intel_config() -> dict:
+    """
+    Get the current intel config for use by agents.
+    This function is called by social agents to get user-defined keywords and profiles.
+    """
+    global intel_config
+    return intel_config
+
+
 # ============================================
 # SITUATIONAL AWARENESS API ENDPOINTS (NEW)
 # ============================================
@@ -1096,41 +1212,49 @@ def record_topic_mention(topic: str, source: str = "manual", domain: str = "gene
 # ============================================
 
 # Lazy-loaded anomaly detection components
-_anomaly_model = None
+_anomaly_models = {}  # {language: model}
 _vectorizer = None
 _language_detector = None
 
 
 def _load_anomaly_components():
-    """Load anomaly detection model and vectorizer"""
-    global _anomaly_model, _vectorizer, _language_detector
+    """Load per-language anomaly detection models and vectorizer"""
+    global _anomaly_models, _vectorizer, _language_detector
 
-    if _anomaly_model is not None:
+    if _anomaly_models:
         return True
 
     try:
         import joblib
         from pathlib import Path
 
-        # Model path
-        models_dir = Path(__file__).parent / "models" / "anomaly-detection" / "src" / "components"
+        # Model directories
         output_dir = Path(__file__).parent / "models" / "anomaly-detection" / "output"
+        artifacts_dir = Path(__file__).parent / "models" / "anomaly-detection" / "artifacts" / "model_trainer"
+
+        # Load per-language models
+        for lang in ["english", "sinhala", "tamil"]:
+            for search_dir in [artifacts_dir, output_dir]:
+                model_path = search_dir / f"isolation_forest_{lang}.joblib"
+                if model_path.exists():
+                    _anomaly_models[lang] = joblib.load(model_path)
+                    logger.info(f"[AnomalyAPI] Loaded {lang} model from {model_path.name}")
+                    break
 
-        # Try to load isolation_forest model (best for anomaly detection)
-        model_paths = [
-            output_dir / "isolation_forest_model.joblib",
-            output_dir / "lof_model.joblib",
-            models_dir.parent / "output" / "isolation_forest_model.joblib",
-        ]
-
-        for model_path in model_paths:
-            if model_path.exists():
-                _anomaly_model = joblib.load(model_path)
-                logger.info(f"[AnomalyAPI] Loaded model from {model_path}")
-                break
+        # Fallback to legacy model if no per-language models found
+        if not _anomaly_models:
+            legacy_paths = [
+                output_dir / "isolation_forest_embeddings_only.joblib",
+                output_dir / "isolation_forest_model.joblib",
+            ]
+            for legacy_path in legacy_paths:
+                if legacy_path.exists():
+                    _anomaly_models["english"] = joblib.load(legacy_path)
+                    logger.info(f"[AnomalyAPI] Loaded legacy model: {legacy_path.name}")
+                    break
 
-        if _anomaly_model is None:
-            logger.warning("[AnomalyAPI] No trained model found. Run training first.")
+        if not _anomaly_models:
+            logger.warning("[AnomalyAPI] No trained models found. Run training first.")
             return False
 
         # Load vectorizer and language detector
@@ -1140,7 +1264,7 @@ def _load_anomaly_components():
         _vectorizer = get_vectorizer()
         _language_detector = detect_language
 
-        logger.info("[AnomalyAPI] ✓ All anomaly components loaded")
+        logger.info(f"[AnomalyAPI] ✓ Loaded models for: {list(_anomaly_models.keys())}")
         return True
 
     except Exception as e:
@@ -1151,7 +1275,7 @@ def _load_anomaly_components():
 @app.post("/api/predict")
 def predict_anomaly(texts: List[str] = None, text: str = None):
     """
-    Run anomaly detection on text(s).
+    Run anomaly detection on text(s) using per-language models.
     
     Args:
         texts: List of texts to analyze
@@ -1185,7 +1309,7 @@ def predict_anomaly(texts: List[str] = None, text: str = None):
                 "message": "Model not trained yet. Using default scores."
             }
 
-        # Vectorize texts
+        # Process texts with per-language models
         predictions = []
         for t in texts:
             try:
@@ -1195,15 +1319,32 @@ def predict_anomaly(texts: List[str] = None, text: str = None):
                 # Vectorize
                 vector = _vectorizer.vectorize(t, lang)
 
-                # Predict
-                # Isolation Forest returns -1 for anomalies, 1 for normal
-                prediction = _anomaly_model.predict([vector])[0]
+                # Select appropriate model
+                if lang in _anomaly_models:
+                    model = _anomaly_models[lang]
+                    method = f"isolation_forest_{lang}"
+                elif "english" in _anomaly_models:
+                    model = _anomaly_models["english"]
+                    method = "isolation_forest_english_fallback"
+                else:
+                    # No model available
+                    predictions.append({
+                        "text": t[:100] + "..." if len(t) > 100 else t,
+                        "is_anomaly": False,
+                        "anomaly_score": 0.0,
+                        "language": lang,
+                        "method": "no_model"
+                    })
+                    continue
 
-                # Get anomaly score (decision_function returns negative for anomalies)
-                if hasattr(_anomaly_model, 'decision_function'):
-                    score = -_anomaly_model.decision_function([vector])[0]  # Invert so higher = more anomalous
-                elif hasattr(_anomaly_model, 'score_samples'):
-                    score = -_anomaly_model.score_samples([vector])[0]
+                # Predict: -1 = anomaly, 1 = normal
+                prediction = model.predict([vector])[0]
+
+                # Get anomaly score
+                if hasattr(model, 'decision_function'):
+                    score = -model.decision_function([vector])[0]
+                elif hasattr(model, 'score_samples'):
+                    score = -model.score_samples([vector])[0]
                 else:
                     score = 1.0 if prediction == -1 else 0.0
 
@@ -1212,7 +1353,7 @@ def predict_anomaly(texts: List[str] = None, text: str = None):
                     "is_anomaly": prediction == -1,
                     "anomaly_score": float(score),
                     "language": lang,
-                    "method": "isolation_forest"
+                    "method": method
                 })
 
             except Exception as e:
@@ -1228,7 +1369,8 @@ def predict_anomaly(texts: List[str] = None, text: str = None):
             "predictions": predictions,
             "total": len(predictions),
             "anomalies_found": sum(1 for p in predictions if p.get("is_anomaly")),
-            "model_status": "loaded"
+            "model_status": "loaded",
+            "models_available": list(_anomaly_models.keys())
         }
 
     except Exception as e:
@@ -1302,8 +1444,10 @@ def get_anomalies(limit: int = 20, threshold: float = 0.5):
                 "message": "Using severity + keyword scoring. Train ML model for advanced detection."
             }
 
-        # ML Model is loaded - use it for scoring
+        # ML Models are loaded - use per-language models for scoring
         anomalies = []
+        per_lang_counts = {"english": 0, "sinhala": 0, "tamil": 0}
+
         for feed in feeds:
             summary = feed.get("summary", "")
             if not summary:
@@ -1312,10 +1456,22 @@ def get_anomalies(limit: int = 20, threshold: float = 0.5):
             try:
                 lang, _ = _language_detector(summary)
                 vector = _vectorizer.vectorize(summary, lang)
-                prediction = _anomaly_model.predict([vector])[0]
 
-                if hasattr(_anomaly_model, 'decision_function'):
-                    score = -_anomaly_model.decision_function([vector])[0]
+                # Select appropriate model
+                if lang in _anomaly_models:
+                    model = _anomaly_models[lang]
+                    method = f"isolation_forest_{lang}"
+                elif "english" in _anomaly_models:
+                    model = _anomaly_models["english"]
+                    method = "isolation_forest_english_fallback"
+                else:
+                    continue
+
+                per_lang_counts[lang] = per_lang_counts.get(lang, 0) + 1
+                prediction = model.predict([vector])[0]
+
+                if hasattr(model, 'decision_function'):
+                    score = -model.decision_function([vector])[0]
                 else:
                     score = 1.0 if prediction == -1 else 0.0
 
@@ -1327,7 +1483,8 @@ def get_anomalies(limit: int = 20, threshold: float = 0.5):
                         **feed,
                         "anomaly_score": float(round(normalized_score, 3)),
                         "is_anomaly": prediction == -1,
-                        "language": lang
+                        "language": lang,
+                        "detection_method": method
                     })
 
                     if len(anomalies) >= limit:
@@ -1344,7 +1501,9 @@ def get_anomalies(limit: int = 20, threshold: float = 0.5):
             "anomalies": anomalies,
             "total": len(anomalies),
             "threshold": threshold,
-            "model_status": "ml_active"
+            "model_status": "ml_active",
+            "models_loaded": list(_anomaly_models.keys()),
+            "per_language_counts": per_lang_counts
         }
 
     except Exception as e:
@@ -2200,7 +2359,7 @@ async def websocket_endpoint(websocket: WebSocket):
                     async with manager._lock:
                         meta = manager.active_connections.get(websocket)
                         if meta is not None:
-                            meta['last_pong'] = datetime.utcnow()
+                            meta['last_pong'] = utc_now()
                             meta['misses'] = 0
                     continue
             except json.JSONDecodeError:

models/anomaly-detection/main.py

@@ -1,85 +1,138 @@
 """
-models/anomaly-detection/main.py
-Entry point for the anomaly detection training pipeline
+Anomaly Detection Training Pipeline
+Trains clustering and anomaly detection models on feed data
 """
-import os
 import sys
-import logging  # Import standard library BEFORE path manipulation
+import os
 from pathlib import Path
-
-# CRITICAL: Configure logging BEFORE adding src/ to path
-# (src/logging/ directory would otherwise shadow the standard module)
-logging.basicConfig(
-    level=logging.INFO,
-    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.StreamHandler(),
-        logging.FileHandler("training.log")
-    ]
+from datetime import datetime
+
+# Load environment variables from root .env BEFORE other imports
+from dotenv import load_dotenv
+ROOT_DIR = Path(__file__).parent.parent.parent  # Go to ModelX-Ultimate
+load_dotenv(ROOT_DIR / ".env")  # Load root .env with MLflow credentials
+
+from src.components.data_ingestion import DataIngestion
+from src.components.data_validation import DataValidation
+from src.components.data_transformation import DataTransformation
+from src.components.model_trainer import ModelTrainer
+from src.exception.exception import AnomalyDetectionException
+from src.logging.logger import logging
+from src.entity.config_entity import (
+    DataIngestionConfig, DataValidationConfig,
+    DataTransformationConfig, ModelTrainerConfig, PipelineConfig
 )
-logger = logging.getLogger("main")
-
-# Add src to path - AFTER logging is configured
-sys.path.insert(0, str(Path(__file__).parent / "src"))
+from src.constants.training_pipeline import MODELS_TO_TRAIN, MLFLOW_EXPERIMENT_NAME
 
-from src.pipeline import run_training_pipeline
-from src.entity import PipelineConfig
 
 
-def main():
-    """Run the anomaly detection training pipeline"""
-    logger.info("=" * 60)
-    logger.info("ANOMALY DETECTION PIPELINE")
-    logger.info("=" * 60)
+def train_pipeline(pipeline_config: PipelineConfig = None) -> dict:
+    """
+    Train the anomaly detection pipeline.
+    
+    Args:
+        pipeline_config: Pipeline configuration (optional)
+        
+    Returns:
+        dict with training results
+    """
+    result = {"status": "failed"}
+    
+    if pipeline_config is None:
+        pipeline_config = PipelineConfig()
 
-    # 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}")
+        logging.info("\n" + "=" * 60)
+        logging.info("ANOMALY DETECTION TRAINING PIPELINE")
+        logging.info(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        logging.info(f"Models to train: {MODELS_TO_TRAIN}")
+        logging.info(f"MLflow experiment: {MLFLOW_EXPERIMENT_NAME}")
+        logging.info("=" * 60 + "\n")
+
+        # Data Ingestion
+        data_ingestion_config = pipeline_config.data_ingestion
+        data_ingestion = DataIngestion(data_ingestion_config)
+        logging.info("Starting data ingestion...")
+        data_ingestion_artifact = data_ingestion.initiate_data_ingestion()
+        logging.info("✓ Data ingestion completed")
+
+        # Data Validation
+        data_validation_config = pipeline_config.data_validation
+        data_validation = DataValidation(data_validation_config)
+        logging.info("Starting data validation...")
+        data_validation_artifact = data_validation.initiate_data_validation(
+            data_ingestion_artifact.raw_data_path
+        )
+        logging.info("✓ Data validation completed")
+
+        # Data Transformation
+        data_transformation_config = pipeline_config.data_transformation
+        data_transformation = DataTransformation(data_transformation_config)
+        logging.info("Starting data transformation...")
+        data_transformation_artifact = data_transformation.initiate_data_transformation(
+            data_validation_artifact.validated_data_path
+        )
+        logging.info("✓ Data transformation completed")
+
+        # Model Training
+        model_trainer_config = pipeline_config.model_trainer
+        model_trainer = ModelTrainer(model_trainer_config)
+        logging.info("Starting model training...")
+        model_trainer_artifact = model_trainer.initiate_model_trainer(
+            data_transformation_artifact.feature_store_path
+        )
+        logging.info("✓ Model training completed")
+
+        result = {
+            "status": "success",
+            "best_model": model_trainer_artifact.best_model_name,
+            "best_model_path": model_trainer_artifact.best_model_path,
+            "best_metrics": model_trainer_artifact.best_model_metrics,
+            "n_anomalies": model_trainer_artifact.n_anomalies,
+            "mlflow_run_id": model_trainer_artifact.mlflow_run_id,
+            "data_ingestion": {
+                "total_records": data_ingestion_artifact.total_records,
+                "from_sqlite": data_ingestion_artifact.records_from_sqlite,
+                "from_csv": data_ingestion_artifact.records_from_csv
+            },
+            "data_validation": {
+                "valid_records": data_validation_artifact.valid_records,
+                "validation_status": data_validation_artifact.validation_status
+            },
+            "data_transformation": {
+                "language_distribution": data_transformation_artifact.language_distribution
+            }
+        }
+
+        logging.info("\n" + "=" * 60)
+        logging.info("PIPELINE RESULTS")
+        logging.info("=" * 60)
+        logging.info(f"Status: {result['status']}")
+        logging.info(f"Best model: {result['best_model']}")
+        logging.info(f"Anomalies detected: {result['n_anomalies']}")
+        logging.info(f"MLflow run: {result.get('mlflow_run_id', 'N/A')}")
+        logging.info("=" * 60 + "\n")
+
+        logging.info("✓ Pipeline completed successfully!")
 
-        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}")
+    except Exception as e:
+        logging.error(f"✗ Pipeline failed: {str(e)}")
+        result = {
+            "status": "failed",
+            "error": str(e)
+        }
 
-        logger.info("\n" + "=" * 60)
-        logger.info("PIPELINE COMPLETE")
-        logger.info("=" * 60)
+    return result
 
-        return artifact
 
+if __name__ == '__main__':
+    try:
+        results = train_pipeline()
+        
+        if results["status"] == "failed":
+            logging.error("Pipeline failed - check logs for details")
+            sys.exit(1)
+            
     except Exception as e:
-        logger.error(f"Pipeline failed: {e}")
-        raise
-
-
-if __name__ == "__main__":
-    main()
+        logging.error(f"Pipeline crashed: {e}")
+        raise AnomalyDetectionException(e, sys)

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

@@ -183,7 +183,7 @@ class DataIngestion:
 
         return df
 
-    def ingest(self) -> DataIngestionArtifact:
+    def initiate_data_ingestion(self) -> DataIngestionArtifact:
         """
         Execute data ingestion pipeline.
         
@@ -228,6 +228,9 @@ class DataIngestion:
         output_path = Path(self.config.output_directory) / f"ingested_data_{timestamp}.parquet"
 
         if is_data_available:
+            # Convert timestamp column to datetime to avoid parquet conversion error
+            if "timestamp" in combined_df.columns:
+                combined_df["timestamp"] = pd.to_datetime(combined_df["timestamp"], errors="coerce")
             combined_df.to_parquet(output_path, index=False)
             logger.info(f"[DataIngestion] Saved {total_records} records to {output_path}")
         else:

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

@@ -330,7 +330,7 @@ class DataTransformation:
         logger.info(f"[DataTransformation] Feature matrix shape: {feature_matrix.shape}")
         return feature_matrix
 
-    def transform(self, data_path: str) -> DataTransformationArtifact:
+    def initiate_data_transformation(self, data_path: str) -> DataTransformationArtifact:
         """
         Execute data transformation pipeline.
         Integrates with Vectorizer Agent Graph for LLM-enhanced processing.
@@ -409,6 +409,11 @@ class DataTransformation:
         embeddings_path = Path(self.config.output_directory) / f"embeddings_{timestamp}.npy"
         np.save(embeddings_path, embeddings)
 
+        # Save language labels for per-language model training
+        languages_path = Path(self.config.output_directory) / f"languages_{timestamp}.npy"
+        np.save(languages_path, df["language"].values)
+        logger.info(f"[DataTransformation] Saved language labels to {languages_path.name}")
+
         # Save feature matrix
         features_path = Path(self.config.output_directory) / f"features_{timestamp}.npy"
         np.save(features_path, feature_matrix)

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

@@ -182,7 +182,7 @@ class DataValidation:
 
         return errors
 
-    def validate(self, data_path: str) -> DataValidationArtifact:
+    def initiate_data_validation(self, data_path: str) -> DataValidationArtifact:
         """
         Execute data validation pipeline.
         

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

@@ -358,7 +358,7 @@ class ModelTrainer:
             return func(X, trial)
         return {"error": f"Unknown model: {model_name}"}
 
-    def train(self, feature_path: str) -> ModelTrainerArtifact:
+    def initiate_model_trainer(self, feature_path: str) -> ModelTrainerArtifact:
         """
         Execute model training pipeline.
         
@@ -476,37 +476,88 @@ class ModelTrainer:
         logger.info(f"[ModelTrainer] Best model: {best_model['name'] if best_model else 'N/A'}")
 
         # ============================================
-        # TRAIN EMBEDDING-ONLY MODEL FOR LIVE INFERENCE
+        # TRAIN PER-LANGUAGE MODELS FOR LIVE INFERENCE
         # ============================================
-        # The Vectorizer Agent only has 768-dim embeddings at inference time
-        # (no temporal/engagement features), so we train a separate model
+        # Different BERT models produce embeddings in different vector spaces.
+        # We train separate Isolation Forest models per language to avoid
+        # mixing incompatible embeddings.
         try:
             # Check if features include extra metadata (> 768 dims)
             if X.shape[1] > 768:
-                logger.info("[ModelTrainer] Training embedding-only model for Vectorizer Agent...")
-
-                # Extract only the first 768 dimensions (BERT embeddings)
-                X_embeddings_only = X[:, :768]
-                logger.info(f"[ModelTrainer] Embedding-only shape: {X_embeddings_only.shape}")
-
-                # Train Isolation Forest on embeddings only
-                embedding_model = IsolationForest(
-                    contamination=0.1,
-                    n_estimators=100,
-                    random_state=42,
-                    n_jobs=-1
-                )
-                embedding_model.fit(X_embeddings_only)
-
-                # Save to a dedicated path for the Vectorizer Agent
-                embedding_model_path = Path(self.config.output_directory) / "isolation_forest_embeddings_only.joblib"
-                joblib.dump(embedding_model, embedding_model_path)
-
-                logger.info(f"[ModelTrainer] Embedding-only model saved: {embedding_model_path}")
-                logger.info("[ModelTrainer] This model is for real-time inference by Vectorizer Agent")
+                X_embeddings = X[:, :768]  # Extract BERT embeddings only
             else:
-                logger.info(f"[ModelTrainer] Features are already embedding-only ({X.shape[1]} dims)")
+                X_embeddings = X
+
+            logger.info(f"[ModelTrainer] Training per-language models on {X_embeddings.shape[0]} samples...")
+
+            # Load language labels from the same directory as features
+            feature_dir = Path(feature_path).parent
+            lang_files = list(feature_dir.glob("languages_*.npy"))
+            
+            if lang_files:
+                # Get most recent language file
+                latest_lang_file = max(lang_files, key=lambda p: p.stem)
+                languages = np.load(latest_lang_file, allow_pickle=True)
+                logger.info(f"[ModelTrainer] Loaded language labels from {latest_lang_file.name}")
+            else:
+                # Fallback: try to load from transformed data parquet
+                parquet_files = list(feature_dir.glob("transformed_*.parquet"))
+                if parquet_files:
+                    import pandas as pd
+                    latest_parquet = max(parquet_files, key=lambda p: p.stem)
+                    df_temp = pd.read_parquet(latest_parquet)
+                    if "language" in df_temp.columns:
+                        languages = df_temp["language"].values
+                        logger.info(f"[ModelTrainer] Loaded {len(languages)} language labels from parquet")
+                    else:
+                        languages = np.array(["english"] * len(X_embeddings))
+                        logger.warning("[ModelTrainer] No language column in parquet, defaulting to english")
+                else:
+                    languages = np.array(["english"] * len(X_embeddings))
+                    logger.warning("[ModelTrainer] No language data found, defaulting to english")
+
+            # Train per-language models
+            MIN_SAMPLES_PER_LANGUAGE = 10
+            per_lang_models = {}
+
+            for lang in ["english", "sinhala", "tamil"]:
+                lang_mask = languages == lang
+                X_lang = X_embeddings[lang_mask]
+
+                if len(X_lang) >= MIN_SAMPLES_PER_LANGUAGE:
+                    logger.info(f"[ModelTrainer] Training {lang} model on {len(X_lang)} samples...")
+                    
+                    lang_model = IsolationForest(
+                        contamination=0.1,
+                        n_estimators=100,
+                        random_state=42,
+                        n_jobs=-1
+                    )
+                    lang_model.fit(X_lang)
+
+                    # Save per-language model
+                    model_path = Path(self.config.output_directory) / f"isolation_forest_{lang}.joblib"
+                    joblib.dump(lang_model, model_path)
+                    per_lang_models[lang] = str(model_path)
+
+                    logger.info(f"[ModelTrainer] ✓ Saved: isolation_forest_{lang}.joblib ({len(X_lang)} samples)")
+                else:
+                    logger.warning(f"[ModelTrainer] Skipping {lang}: only {len(X_lang)} samples (min: {MIN_SAMPLES_PER_LANGUAGE})")
+
+            # Also save a legacy "embeddings_only" model for backward compatibility (trained on English)
+            if "english" in per_lang_models:
+                import shutil
+                english_model_path = Path(per_lang_models["english"])
+                legacy_path = Path(self.config.output_directory) / "isolation_forest_embeddings_only.joblib"
+                shutil.copy(english_model_path, legacy_path)
+                logger.info(f"[ModelTrainer] ✓ Legacy model copied: isolation_forest_embeddings_only.joblib")
+
+            logger.info(f"[ModelTrainer] Per-language training complete: {list(per_lang_models.keys())}")
+
         except Exception as e:
-            logger.warning(f"[ModelTrainer] Embedding-only model training failed: {e}")
+            logger.warning(f"[ModelTrainer] Per-language model training failed: {e}")
+            import traceback
+            traceback.print_exc()
 
         return artifact
+

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

@@ -0,0 +1 @@
+from .training_pipeline import *

models/anomaly-detection/src/constants/training_pipeline/__init__.py

@@ -0,0 +1,65 @@
+"""
+Anomaly Detection Training Pipeline Constants
+"""
+import os
+
+# Pipeline configuration
+PIPELINE_NAME: str = "AnomalyDetection"
+ARTIFACT_DIR: str = "artifacts"
+
+# Data sources
+SQLITE_DB_PATH = os.getenv(
+    "SQLITE_DB_PATH",
+    os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "..", "data", "feeds", "feed_cache.db")
+)
+CSV_DIRECTORY = os.path.join(os.path.dirname(__file__), "..", "..", "..", "..", "..", "datasets", "political_feeds")
+
+# Data Ingestion
+DATA_INGESTION_DIR_NAME: str = "data_ingestion"
+DATA_INGESTION_FEATURE_STORE_DIR: str = "feature_store"
+DATA_INGESTION_INGESTED_DIR: str = "ingested"
+FILE_NAME: str = "ingested_data.parquet"
+MIN_TEXT_LENGTH: int = 10
+BATCH_SIZE: int = 1000
+
+# Data Validation
+DATA_VALIDATION_DIR_NAME: str = "data_validation"
+DATA_VALIDATION_VALID_DIR: str = "validated"
+DATA_VALIDATION_INVALID_DIR: str = "invalid"
+SCHEMA_FILE_PATH = os.path.join("data_schema", "schema.yaml")
+REQUIRED_COLUMNS = ["post_id", "timestamp", "platform", "category", "text", "content_hash"]
+
+# Data Transformation
+DATA_TRANSFORMATION_DIR_NAME: str = "data_transformation"
+DATA_TRANSFORMATION_TRANSFORMED_DATA_DIR: str = "transformed"
+FEATURE_STORE_FILE_NAME: str = "features.npy"
+
+# Language Models (Multilingual BERT)
+ENGLISH_MODEL: str = "distilbert-base-uncased"
+SINHALA_MODEL: str = "keshan/SinhalaBERTo"
+TAMIL_MODEL: str = "l3cube-pune/tamil-bert"
+VECTOR_DIM: int = 768
+
+# Model Training
+MODEL_TRAINER_DIR_NAME: str = "model_trainer"
+MODEL_TRAINER_TRAINED_MODEL_DIR: str = "trained_model"
+MODEL_FILE_NAME: str = "model.joblib"
+SAVED_MODEL_DIR = os.path.join("saved_models")
+
+# Models to train
+MODELS_TO_TRAIN = ["dbscan", "kmeans", "hdbscan", "isolation_forest", "lof"]
+
+# Optuna hyperparameter tuning
+N_OPTUNA_TRIALS: int = 50
+OPTUNA_TIMEOUT_SECONDS: int = 3600  # 1 hour
+
+# MLflow configuration
+MLFLOW_TRACKING_URI = os.getenv(
+    "MLFLOW_TRACKING_URI",
+    "https://dagshub.com/sliitguy/Model-X.mlflow"
+)
+MLFLOW_EXPERIMENT_NAME: str = "anomaly_detection_feeds"
+
+# Model thresholds
+MODEL_TRAINER_EXPECTED_SCORE: float = 0.3  # Silhouette score threshold
+MODEL_TRAINER_OVERFITTING_THRESHOLD: float = 0.1

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

@@ -71,7 +71,7 @@ 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_TRACKING_URI", "https://dagshub.com/sliitguy/Model-X.mlflow"
     ))
     mlflow_username: str = field(default_factory=lambda: os.getenv(
         "MLFLOW_TRACKING_USERNAME", ""

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

@@ -0,0 +1 @@
+from .exception import AnomalyDetectionException

models/anomaly-detection/src/exception/exception.py

@@ -0,0 +1,24 @@
+import sys
+
+
+class AnomalyDetectionException(Exception):
+    """Custom exception for Anomaly Detection pipeline."""
+    
+    def __init__(self, error_message, error_details: sys):
+        self.error_message = error_message
+        _, _, exc_tb = error_details.exc_info()
+        
+        self.lineno = exc_tb.tb_lineno
+        self.file_name = exc_tb.tb_frame.f_code.co_filename
+    
+    def __str__(self):
+        return "Error occurred in python script name [{0}] line number [{1}] error message [{2}]".format(
+            self.file_name, self.lineno, str(self.error_message)
+        )
+
+
+if __name__ == '__main__':
+    try:
+        a = 1 / 0
+    except Exception as e:
+        raise AnomalyDetectionException(e, sys)

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

@@ -0,0 +1 @@
+from .logger import logging

models/anomaly-detection/src/logging/logger.py

@@ -0,0 +1,32 @@
+"""
+Logging configuration for Anomaly Detection pipeline.
+Creates timestamped log files in the logs directory.
+"""
+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)
+
+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
+)
+
+# Also add console handler for visibility
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.INFO)
+console_handler.setFormatter(logging.Formatter(
+    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+))
+
+# Get root logger and add console handler
+root_logger = logging.getLogger()
+root_logger.addHandler(console_handler)

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

@@ -53,7 +53,7 @@ class TrainingPipeline:
         logger.info("=" * 50)
 
         ingestion = DataIngestion(self.config.data_ingestion)
-        artifact = ingestion.ingest()
+        artifact = ingestion.initiate_data_ingestion()
 
         if not artifact.is_data_available:
             raise ValueError("No data available for training")
@@ -67,7 +67,7 @@ class TrainingPipeline:
         logger.info("=" * 50)
 
         validation = DataValidation(self.config.data_validation)
-        artifact = validation.validate(ingestion_artifact.raw_data_path)
+        artifact = validation.initiate_data_validation(ingestion_artifact.raw_data_path)
 
         return artifact
 
@@ -78,7 +78,7 @@ class TrainingPipeline:
         logger.info("=" * 50)
 
         transformation = DataTransformation(self.config.data_transformation)
-        artifact = transformation.transform(validation_artifact.validated_data_path)
+        artifact = transformation.initiate_data_transformation(validation_artifact.validated_data_path)
 
         return artifact
 
@@ -89,7 +89,7 @@ class TrainingPipeline:
         logger.info("=" * 50)
 
         trainer = ModelTrainer(self.config.model_trainer)
-        artifact = trainer.train(transformation_artifact.feature_store_path)
+        artifact = trainer.initiate_model_trainer(transformation_artifact.feature_store_path)
 
         return artifact
 

models/currency-volatility-prediction/main.py

@@ -1,87 +1,87 @@
 """
-models/currency-volatility-prediction/main.py
-Entry point for Currency Prediction Pipeline
-Can run data collection, training, or prediction independently
+Currency Volatility Prediction Pipeline - USD/LKR Training
+Follows stock-price-prediction pattern with structured artifact flow
 """
-import os
+from src.components.data_ingestion import CurrencyDataIngestion
+from src.components.model_trainer import CurrencyGRUTrainer
+from src.components.predictor import CurrencyPredictor
+from src.exception.exception import CurrencyPredictionException
+from src.logging.logger import logging
+from src.entity.config_entity import DataIngestionConfig, ModelTrainerConfig
+
 import sys
-import logging  # Import standard library BEFORE path manipulation
+import os
 import argparse
-from pathlib import Path
 from datetime import datetime
 
-# CRITICAL: Configure logging BEFORE adding src/ to path
-# (src/logging/ directory would otherwise shadow the standard module)
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
-logger = logging.getLogger("currency_prediction")
-
-# Setup paths - AFTER logging is configured
-PIPELINE_ROOT = Path(__file__).parent
-sys.path.insert(0, str(PIPELINE_ROOT / "src"))
-
-
-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")
+def train_currency(period: str = "2y", epochs: int = 100) -> dict:
+    """
+    Train the currency prediction model.
+    
+    Follows stock-price-prediction pattern with structured results.
+    
+    Args:
+        period: Data period for yfinance (1y, 2y, 5y)
+        epochs: Number of training epochs
+        
+    Returns:
+        dict with training results or error info
+    """
+    result = {"currency": "USD_LKR", "status": "failed"}
 
-    # Train
-    config = ModelTrainerConfig(epochs=epochs)
-    trainer = CurrencyGRUTrainer(config)
-
-    results = trainer.train(df=df, use_mlflow=False)  # Disabled due to Windows Unicode encoding issues
-
-    logger.info("\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']}")
+    try:
+        logging.info(f"\n{'='*60}")
+        logging.info("CURRENCY PREDICTION PIPELINE - TRAINING")
+        logging.info(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+        logging.info(f"{'='*60}")
+
+        # Step 1: Data Ingestion
+        logging.info("[USD_LKR] Starting data ingestion...")
+        config = DataIngestionConfig(history_period=period)
+        ingestion = CurrencyDataIngestion(config)
+        data_path = ingestion.ingest_all()
+        df = ingestion.load_existing(data_path)
+        logging.info(f"[USD_LKR] ✓ Data ingestion completed: {len(df)} records")
+
+        # Step 2: Model Training
+        logging.info("[USD_LKR] Starting model training...")
+        trainer_config = ModelTrainerConfig(epochs=epochs)
+        trainer = CurrencyGRUTrainer(trainer_config)
+        train_results = trainer.train(df=df, use_mlflow=True)
+        logging.info("[USD_LKR] ✓ Model training completed")
+
+        result = {
+            "currency": "USD_LKR",
+            "status": "success",
+            "model_path": train_results["model_path"],
+            "test_mae": train_results["test_mae"],
+            "rmse": train_results["rmse"],
+            "direction_accuracy": train_results["direction_accuracy"],
+            "epochs_trained": train_results["epochs_trained"]
+        }
+
+        logging.info(f"[USD_LKR] ✓ Pipeline completed successfully!")
 
-    return results
+    except Exception as e:
+        logging.error(f"[USD_LKR] ✗ Pipeline failed: {str(e)}")
+        result = {
+            "currency": "USD_LKR",
+            "status": "failed",
+            "error": str(e)
+        }
 
+    return result
 
-def run_prediction():
-    """Run prediction for next day."""
-    from components.data_ingestion import CurrencyDataIngestion
-    from components.predictor import CurrencyPredictor
 
-    logger.info("Generating prediction...")
+def run_prediction() -> dict:
+    """
+    Run prediction for next day.
+    
+    Returns:
+        Prediction dictionary
+    """
+    logging.info("Generating prediction...")
 
     predictor = CurrencyPredictor()
 
@@ -89,70 +89,78 @@ def run_prediction():
         ingestion = CurrencyDataIngestion()
         df = ingestion.load_existing()
         prediction = predictor.predict(df)
+        logging.info("[USD_LKR] ✓ Prediction generated using trained model")
     except FileNotFoundError:
-        logger.warning("Model not trained, using fallback")
+        logging.warning("[USD_LKR] Model not trained, using fallback")
         prediction = predictor.generate_fallback_prediction()
     except Exception as e:
-        logger.error(f"Error: {e}")
+        logging.error(f"[USD_LKR] 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']}")
+    logging.info(f"\n{'='*50}")
+    logging.info(f"USD/LKR PREDICTION FOR {prediction['prediction_date']}")
+    logging.info(f"{'='*50}")
+    logging.info(f"Current Rate:   {prediction['current_rate']:.2f} LKR/USD")
+    logging.info(f"Predicted Rate: {prediction['predicted_rate']:.2f} LKR/USD")
+    logging.info(f"Expected Change: {prediction['expected_change_pct']:+.3f}%")
+    logging.info(f"Direction: {prediction['direction_emoji']} LKR {prediction['direction']}")
+    logging.info(f"Volatility: {prediction['volatility_class']}")
 
     if prediction.get('weekly_trend'):
-        logger.info(f"Weekly Trend: {prediction['weekly_trend']:+.2f}%")
+        logging.info(f"Weekly Trend: {prediction['weekly_trend']:+.2f}%")
     if prediction.get('monthly_trend'):
-        logger.info(f"Monthly Trend: {prediction['monthly_trend']:+.2f}%")
+        logging.info(f"Monthly Trend: {prediction['monthly_trend']:+.2f}%")
 
-    logger.info(f"{'='*50}")
-    logger.info(f"Saved to: {output_path}")
+    logging.info(f"{'='*50}")
+    logging.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
+    """
+    Run the complete pipeline: train → predict.
+    Following stock-price-prediction pattern.
+    """
+    logging.info("\n" + "="*70)
+    logging.info("CURRENCY PREDICTION PIPELINE - FULL RUN")
+    logging.info(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info("="*70 + "\n")
+
+    # Step 1: Training
+    result = train_currency(period="2y", epochs=100)
+
+    # Step 2: Prediction
     prediction = run_prediction()
 
-    logger.info("=" * 60)
-    logger.info("PIPELINE COMPLETE!")
-    logger.info("=" * 60)
+    # Print summary
+    logging.info("\n" + "="*70)
+    logging.info("TRAINING SUMMARY")
+    logging.info("="*70)
+    
+    if result["status"] == "success":
+        logging.info(f"  ✓ USD_LKR: {result['model_path']}")
+        logging.info(f"       MAE: {result['test_mae']:.4f} LKR")
+        logging.info(f"       RMSE: {result['rmse']:.4f} LKR")
+        logging.info(f"       Direction Accuracy: {result['direction_accuracy']*100:.1f}%")
+    else:
+        logging.info(f"  ✗ USD_LKR: {result.get('error', 'Unknown error')[:50]}")
 
-    return prediction
+    logging.info("="*70)
+    logging.info(f"Completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info("="*70 + "\n")
+
+    return result, prediction
 
 
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Currency Prediction Pipeline")
     parser.add_argument(
         "--mode",
-        choices=["ingest", "train", "predict", "full"],
+        choices=["train", "predict", "full"],
         default="predict",
         help="Pipeline mode to run"
     )
@@ -171,11 +179,17 @@ if __name__ == "__main__":
 
     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()
+    try:
+        if args.mode == "train":
+            result = train_currency(period=args.period, epochs=args.epochs)
+            if result["status"] == "failed":
+                sys.exit(1)
+        elif args.mode == "predict":
+            run_prediction()
+        elif args.mode == "full":
+            result, prediction = run_full_pipeline()
+            if result["status"] == "failed":
+                sys.exit(1)
+    except Exception as e:
+        logging.error(f"Pipeline crashed: {e}")
+        raise CurrencyPredictionException(e, sys)

models/currency-volatility-prediction/src/components/model_trainer.py

@@ -5,6 +5,11 @@ Optimized for 8GB RAM laptops without GPU
 """
 import os
 import sys
+
+# Fix Windows console encoding issue with MLflow emoji output
+if sys.platform == 'win32':
+    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
+
 import logging
 import numpy as np
 import pandas as pd

models/currency-volatility-prediction/src/components/predictor.py

@@ -62,7 +62,11 @@ class CurrencyPredictor:
         if not os.path.exists(model_path):
             raise FileNotFoundError(f"No trained model found at {model_path}")
 
-        self._model = load_model(model_path)
+        # Load with compile=False to avoid Keras 2->3 mse serialization issues
+        # Then recompile with standard metrics
+        self._model = load_model(model_path, compile=False)
+        self._model.compile(optimizer='adam', loss='mse', metrics=['mae'])
+        
         scalers = joblib.load(scaler_path)
 
         self._scalers = {

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

@@ -0,0 +1 @@
+from src.exception.exception import CurrencyPredictionException

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

@@ -1,22 +1,24 @@
 import sys
-from src.log_utils import logger
 
-class NetworkSecurityException(Exception):
-    def __init__(self,error_message,error_details:sys):
-        self.error_message = error_message
-        _,_,exc_tb = error_details.exc_info()
-
-        self.lineno=exc_tb.tb_lineno
-        self.file_name=exc_tb.tb_frame.f_code.co_filename
 
+class CurrencyPredictionException(Exception):
+    """Custom exception for Currency Prediction pipeline."""
+    
+    def __init__(self, error_message, error_details: sys):
+        self.error_message = error_message
+        _, _, exc_tb = error_details.exc_info()
+        
+        self.lineno = exc_tb.tb_lineno
+        self.file_name = exc_tb.tb_frame.f_code.co_filename
+    
     def __str__(self):
-        return "Error occured in python script name [{0}] line number [{1}] error message [{2}]".format(
-        self.file_name, self.lineno, str(self.error_message))
+        return "Error occurred in python script name [{0}] line number [{1}] error message [{2}]".format(
+            self.file_name, self.lineno, str(self.error_message)
+        )
+
 
-if __name__=='__main__':
+if __name__ == '__main__':
     try:
-        logger.logging.info("Enter the try block")
-        a=1/0
-        print("This will not be printed",a)
+        a = 1 / 0
     except Exception as e:
-           raise NetworkSecurityException(e,sys)
+        raise CurrencyPredictionException(e, sys)

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

@@ -0,0 +1 @@
+from src.logging.logger import logging

models/currency-volatility-prediction/src/logging/logger.py

@@ -1,15 +1,18 @@
+"""
+Logging configuration for Currency Prediction pipeline.
+Creates timestamped log files in the logs directory.
+"""
 import logging
 import os
 from datetime import datetime
 
-LOG_FILE=f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
+LOG_FILE = f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
 
-logs_path=os.path.join(os.getcwd(), "logs", LOG_FILE)
+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)
+LOG_FILE_PATH = os.path.join(logs_path, LOG_FILE)
 
 logging.basicConfig(
     filename=LOG_FILE_PATH,
@@ -17,4 +20,13 @@ logging.basicConfig(
     level=logging.INFO
 )
 
+# Also add console handler for visibility
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.INFO)
+console_handler.setFormatter(logging.Formatter(
+    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+))
 
+# Get root logger and add console handler
+root_logger = logging.getLogger()
+root_logger.addHandler(console_handler)

models/weather-prediction/main.py

@@ -1,86 +1,154 @@
 """
-models/weather-prediction/main.py
-Entry point for Weather Prediction Pipeline
-Can run data collection, training, or prediction independently
+Weather Prediction Pipeline - Multi-Station Training
+Follows stock-price-prediction pattern with structured artifact flow
 """
-import os
+from src.components.data_ingestion import DataIngestion
+from src.components.model_trainer import WeatherLSTMTrainer
+from src.components.predictor import WeatherPredictor
+from src.exception.exception import WeatherPredictionException
+from src.logging.logger import logging
+from src.entity.config_entity import DataIngestionConfig, WEATHER_STATIONS
+
 import sys
-import logging  # Import standard library BEFORE path manipulation
+import os
 import argparse
 from pathlib import Path
 from datetime import datetime
 
-# CRITICAL: Configure logging BEFORE adding src/ to path
-# (src/logging/ directory would otherwise shadow the standard module)
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
-)
-logger = logging.getLogger("weather_prediction")
-
-# Setup paths - AFTER logging is configured
 PIPELINE_ROOT = Path(__file__).parent
-sys.path.insert(0, str(PIPELINE_ROOT / "src"))
 
 
-def run_data_ingestion(months: int = 12):
-    """Run data ingestion for all stations."""
-    from components.data_ingestion import DataIngestion
-    from entity.config_entity import DataIngestionConfig
+def train_single_station(station_name: str, epochs: int = 100) -> dict:
+    """
+    Train a model for a single weather station.
+    
+    Follows stock-price-prediction pattern with structured results.
+    
+    Args:
+        station_name: Weather station name (e.g., 'COLOMBO', 'KANDY')
+        epochs: Number of training epochs
+        
+    Returns:
+        dict with training results or error info
+    """
+    result = {"station": station_name, "status": "failed"}
 
-    logger.info(f"Starting data ingestion ({months} months)...")
+    try:
+        logging.info(f"\n{'='*60}")
+        logging.info(f"Training model for: {station_name}")
+        logging.info(f"{'='*60}")
+
+        # Data Ingestion
+        logging.info(f"[{station_name}] Loading data...")
+        ingestion = DataIngestion()
+        df = ingestion.load_existing()
+        logging.info(f"[{station_name}] ✓ Data loaded")
+
+        # Model Training
+        logging.info(f"[{station_name}] Starting model training...")
+        trainer = WeatherLSTMTrainer(
+            sequence_length=30,
+            lstm_units=[64, 32]
+        )
+        
+        train_results = trainer.train(
+            df=df,
+            station_name=station_name,
+            epochs=epochs,
+            use_mlflow=False  # Disabled due to Windows Unicode encoding issues
+        )
+        logging.info(f"[{station_name}] ✓ Model training completed")
+
+        result = {
+            "station": station_name,
+            "status": "success",
+            "model_path": train_results.get("model_path", ""),
+            "test_mae": train_results.get("test_mae", 0),
+            "test_mse": train_results.get("test_mse", 0),
+            "epochs_trained": epochs
+        }
+
+        logging.info(f"[{station_name}] ✓ Pipeline completed successfully!")
 
-    config = DataIngestionConfig(months_to_fetch=months)
-    ingestion = DataIngestion(config)
+    except Exception as e:
+        logging.error(f"[{station_name}] ✗ Pipeline failed: {str(e)}")
+        result = {
+            "station": station_name,
+            "status": "failed",
+            "error": str(e)
+        }
 
-    data_path = ingestion.ingest_all()
+    return result
 
-    df = ingestion.load_existing(data_path)
-    stats = ingestion.get_data_stats(df)
 
-    logger.info("Data Ingestion Complete!")
-    logger.info(f"Total records: {stats['total_records']}")
-    logger.info(f"Stations: {stats['stations']}")
-    logger.info(f"Date range: {stats['date_range']}")
+def train_all_stations(stations: list = None, epochs: int = 100) -> list:
+    """
+    Train models for all weather stations.
+    Each station gets its own model saved separately.
+    
+    Follows stock-price-prediction pattern.
+    """
+    stations_to_train = stations or list(WEATHER_STATIONS.keys())
 
-    return data_path
+    logging.info("\n" + "="*70)
+    logging.info("WEATHER PREDICTION - MULTI-STATION TRAINING PIPELINE")
+    logging.info(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info(f"Stations to train: {stations_to_train}")
+    logging.info("="*70 + "\n")
 
+    results = []
+    successful = 0
+    failed = 0
 
-def run_training(station: str = None, epochs: int = 100):
-    """Run model training."""
-    from components.data_ingestion import DataIngestion
-    from components.model_trainer import WeatherLSTMTrainer
-    from entity.config_entity import WEATHER_STATIONS
+    for station_name in stations_to_train:
+        result = train_single_station(station_name, epochs)
+        results.append(result)
+
+        if result["status"] == "success":
+            successful += 1
+        else:
+            failed += 1
+
+    # Print summary
+    logging.info("\n" + "="*70)
+    logging.info("TRAINING SUMMARY")
+    logging.info("="*70)
+    logging.info(f"Total stations: {len(stations_to_train)}")
+    logging.info(f"Successful: {successful}")
+    logging.info(f"Failed: {failed}")
+    logging.info("-"*70)
+
+    for result in results:
+        if result["status"] == "success":
+            logging.info(f"  ✓ {result['station']}: MAE={result['test_mae']:.3f}")
+        else:
+            logging.info(f"  ✗ {result['station']}: {result.get('error', 'Unknown error')[:50]}")
+
+    logging.info("="*70)
+    logging.info(f"Completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info("="*70 + "\n")
 
-    logger.info("Starting model training...")
+    return results
 
-    ingestion = DataIngestion()
-    df = ingestion.load_existing()
 
-    trainer = WeatherLSTMTrainer(
-        sequence_length=30,
-        lstm_units=[64, 32]
-    )
+def run_data_ingestion(months: int = 12):
+    """Run data ingestion for all stations."""
+    logging.info(f"Starting data ingestion ({months} months)...")
 
-    stations_to_train = [station] if station else list(WEATHER_STATIONS.keys())
-    results = []
+    config = DataIngestionConfig(months_to_fetch=months)
+    ingestion = DataIngestion(config)
 
-    for station_name in stations_to_train:
-        try:
-            logger.info(f"Training {station_name}...")
-            result = trainer.train(
-                df=df,
-                station_name=station_name,
-                epochs=epochs,
-                use_mlflow=False  # Disabled due to Windows Unicode encoding issues
-            )
-            results.append(result)
-            logger.info(f"[OK] {station_name}: MAE={result['test_mae']:.3f}")
-        except Exception as e:
-            logger.error(f"[FAIL] {station_name}: {e}")
+    data_path = ingestion.ingest_all()
 
-    logger.info(f"Training complete! Trained {len(results)} models.")
-    return results
+    df = ingestion.load_existing(data_path)
+    stats = ingestion.get_data_stats(df)
+
+    logging.info("✓ Data Ingestion Complete!")
+    logging.info(f"Total records: {stats['total_records']}")
+    logging.info(f"Stations: {stats['stations']}")
+    logging.info(f"Date range: {stats['date_range']}")
+
+    return data_path
 
 
 def check_and_train_missing_models(priority_only: bool = True, epochs: int = 25):
@@ -95,8 +163,6 @@ def check_and_train_missing_models(priority_only: bool = True, epochs: int = 25)
     Returns:
         List of trained station names
     """
-    from entity.config_entity import WEATHER_STATIONS
-
     models_dir = PIPELINE_ROOT / "artifacts" / "models"
     models_dir.mkdir(parents=True, exist_ok=True)
 
@@ -113,44 +179,36 @@ def check_and_train_missing_models(priority_only: bool = True, epochs: int = 25)
             missing_stations.append(station)
 
     if not missing_stations:
-        logger.info("[AUTO-TRAIN] All required models exist.")
+        logging.info("[AUTO-TRAIN] All required models exist.")
         return []
 
-    logger.info(f"[AUTO-TRAIN] Missing models for: {', '.join(missing_stations)}")
-    logger.info("[AUTO-TRAIN] Starting automatic training...")
+    logging.info(f"[AUTO-TRAIN] Missing models for: {', '.join(missing_stations)}")
+    logging.info("[AUTO-TRAIN] Starting automatic training...")
 
     # Ensure we have data first
     data_path = PIPELINE_ROOT / "artifacts" / "data"
     existing_data = list(data_path.glob("weather_history_*.csv")) if data_path.exists() else []
 
     if not existing_data:
-        logger.info("[AUTO-TRAIN] No training data found, ingesting...")
+        logging.info("[AUTO-TRAIN] No training data found, ingesting...")
         try:
             run_data_ingestion(months=3)
         except Exception as e:
-            logger.error(f"[AUTO-TRAIN] Data ingestion failed: {e}")
-            logger.info("[AUTO-TRAIN] Cannot train without data. Please run: python main.py --mode ingest")
+            logging.error(f"[AUTO-TRAIN] Data ingestion failed: {e}")
+            logging.info("[AUTO-TRAIN] Cannot train without data. Please run: python main.py --mode ingest")
             return []
 
-    # Train missing models
-    trained = []
-    for station in missing_stations:
-        try:
-            logger.info(f"[AUTO-TRAIN] Training {station}...")
-            run_training(station=station, epochs=epochs)
-            trained.append(station)
-        except Exception as e:
-            logger.warning(f"[AUTO-TRAIN] Failed to train {station}: {e}")
-
-    logger.info(f"[AUTO-TRAIN] Auto-training complete. Trained {len(trained)} models: {', '.join(trained)}")
+    # Train missing models using structured function
+    results = train_all_stations(stations=missing_stations, epochs=epochs)
+    
+    trained = [r["station"] for r in results if r["status"] == "success"]
+    logging.info(f"[AUTO-TRAIN] Auto-training complete. Trained {len(trained)} models.")
     return trained
 
 
 def run_prediction():
     """Run prediction for all districts."""
-    from components.predictor import WeatherPredictor
-
-    logger.info("Generating predictions...")
+    logging.info("Generating predictions...")
 
     predictor = WeatherPredictor()
 
@@ -160,9 +218,9 @@ def run_prediction():
         sys.path.insert(0, str(PIPELINE_ROOT.parent.parent / "src"))
         from utils.utils import tool_rivernet_status
         rivernet_data = tool_rivernet_status()
-        logger.info(f"RiverNet data available: {len(rivernet_data.get('rivers', []))} rivers")
+        logging.info(f"✓ RiverNet data available: {len(rivernet_data.get('rivers', []))} rivers")
     except Exception as e:
-        logger.warning(f"RiverNet data unavailable: {e}")
+        logging.warning(f"RiverNet data unavailable: {e}")
 
     predictions = predictor.predict_all_districts(rivernet_data=rivernet_data)
     output_path = predictor.save_predictions(predictions)
@@ -175,48 +233,49 @@ def run_prediction():
         sev = p.get("severity", "normal")
         severity_counts[sev] = severity_counts.get(sev, 0) + 1
 
-    logger.info(f"\n{'='*50}")
-    logger.info(f"PREDICTIONS FOR {predictions['prediction_date']}")
-    logger.info(f"{'='*50}")
-    logger.info(f"Districts: {len(districts)}")
-    logger.info(f"Normal: {severity_counts['normal']}")
-    logger.info(f"Advisory: {severity_counts['advisory']}")
-    logger.info(f"Warning: {severity_counts['warning']}")
-    logger.info(f"Critical: {severity_counts['critical']}")
-    logger.info(f"Output: {output_path}")
+    logging.info(f"\n{'='*50}")
+    logging.info(f"PREDICTIONS FOR {predictions['prediction_date']}")
+    logging.info(f"{'='*50}")
+    logging.info(f"Districts: {len(districts)}")
+    logging.info(f"Normal: {severity_counts['normal']}")
+    logging.info(f"Advisory: {severity_counts['advisory']}")
+    logging.info(f"Warning: {severity_counts['warning']}")
+    logging.info(f"Critical: {severity_counts['critical']}")
+    logging.info(f"Output: {output_path}")
 
     return predictions
 
 
 def run_full_pipeline():
-    """Run the full pipeline: ingest → train → predict."""
-    logger.info("=" * 60)
-    logger.info("WEATHER PREDICTION PIPELINE - FULL RUN")
-    logger.info("=" * 60)
+    """
+    Run the full pipeline: ingest → train → predict.
+    Following stock-price-prediction pattern.
+    """
+    logging.info("\n" + "="*70)
+    logging.info("WEATHER PREDICTION PIPELINE - FULL RUN")
+    logging.info(f"Started at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info("="*70 + "\n")
 
     # Step 1: Data Ingestion
     try:
         run_data_ingestion(months=3)
     except Exception as e:
-        logger.error(f"Data ingestion failed: {e}")
-        logger.info("Attempting to use existing data...")
+        logging.error(f"Data ingestion failed: {e}")
+        logging.info("Attempting to use existing data...")
 
     # Step 2: Training (priority stations only)
     priority_stations = ["COLOMBO", "KANDY", "JAFFNA", "BATTICALOA", "RATNAPURA"]
-    for station in priority_stations:
-        try:
-            run_training(station=station, epochs=50)
-        except Exception as e:
-            logger.warning(f"Training {station} failed: {e}")
+    results = train_all_stations(stations=priority_stations, epochs=50)
 
     # Step 3: Prediction
     predictions = run_prediction()
 
-    logger.info("=" * 60)
-    logger.info("PIPELINE COMPLETE!")
-    logger.info("=" * 60)
+    logging.info("\n" + "="*70)
+    logging.info("PIPELINE COMPLETE!")
+    logging.info(f"Completed at: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+    logging.info("="*70 + "\n")
 
-    return predictions
+    return results, predictions
 
 
 if __name__ == "__main__":
@@ -253,18 +312,34 @@ if __name__ == "__main__":
 
     args = parser.parse_args()
 
-    if args.mode == "ingest":
-        run_data_ingestion(months=args.months)
-    elif args.mode == "train":
-        run_training(station=args.station, epochs=args.epochs)
-    elif args.mode == "auto-train":
-        # Explicitly auto-train missing models
-        check_and_train_missing_models(priority_only=True, epochs=25)
-    elif args.mode == "predict":
-        # Auto-train missing models before prediction (unless skipped)
-        if not args.skip_auto_train:
+    try:
+        if args.mode == "ingest":
+            run_data_ingestion(months=args.months)
+        elif args.mode == "train":
+            if args.station:
+                result = train_single_station(args.station, args.epochs)
+                if result["status"] == "failed":
+                    sys.exit(1)
+            else:
+                results = train_all_stations(epochs=args.epochs)
+                failed = sum(1 for r in results if r["status"] == "failed")
+                if failed > 0:
+                    logging.warning(f"{failed} stations failed to train")
+                    sys.exit(1)
+        elif args.mode == "auto-train":
+            # Explicitly auto-train missing models
             check_and_train_missing_models(priority_only=True, epochs=25)
-        run_prediction()
-    elif args.mode == "full":
-        run_full_pipeline()
-
+        elif args.mode == "predict":
+            # Auto-train missing models before prediction (unless skipped)
+            if not args.skip_auto_train:
+                check_and_train_missing_models(priority_only=True, epochs=25)
+            run_prediction()
+        elif args.mode == "full":
+            results, predictions = run_full_pipeline()
+            failed = sum(1 for r in results if r["status"] == "failed")
+            if failed > 0:
+                logging.warning(f"{failed} stations failed to train")
+                sys.exit(1)
+    except Exception as e:
+        logging.error(f"Pipeline crashed: {e}")
+        raise WeatherPredictionException(e, sys)

models/weather-prediction/src/components/model_trainer.py

@@ -4,6 +4,11 @@ LSTM-based Weather Prediction Model Trainer
 """
 import os
 import sys
+
+# Fix Windows console encoding issue with MLflow emoji output
+if sys.platform == 'win32':
+    sys.stdout.reconfigure(encoding='utf-8', errors='replace')
+
 import logging
 import numpy as np
 import pandas as pd

models/weather-prediction/src/components/predictor.py

@@ -67,7 +67,10 @@ class WeatherPredictor:
             logger.warning(f"[PREDICTOR] No model for {station_name}")
             return None, None
 
-        self._models[station_name] = load_model(model_path)
+        # Load with compile=False to avoid Keras 2->3 mse serialization issues
+        # Then recompile with standard metrics
+        self._models[station_name] = load_model(model_path, compile=False)
+        self._models[station_name].compile(optimizer='adam', loss='mse', metrics=['mae'])
         self._scalers[station_name] = joblib.load(scaler_path)
 
         return self._models[station_name], self._scalers[station_name]

models/weather-prediction/src/exception/__init__.py

@@ -0,0 +1 @@
+from src.exception.exception import WeatherPredictionException

models/weather-prediction/src/exception/exception.py

@@ -1,22 +1,24 @@
 import sys
-from src.log_utils import logger
 
-class NetworkSecurityException(Exception):
-    def __init__(self,error_message,error_details:sys):
-        self.error_message = error_message
-        _,_,exc_tb = error_details.exc_info()
-
-        self.lineno=exc_tb.tb_lineno
-        self.file_name=exc_tb.tb_frame.f_code.co_filename
 
+class WeatherPredictionException(Exception):
+    """Custom exception for Weather Prediction pipeline."""
+    
+    def __init__(self, error_message, error_details: sys):
+        self.error_message = error_message
+        _, _, exc_tb = error_details.exc_info()
+        
+        self.lineno = exc_tb.tb_lineno
+        self.file_name = exc_tb.tb_frame.f_code.co_filename
+    
     def __str__(self):
-        return "Error occured in python script name [{0}] line number [{1}] error message [{2}]".format(
-        self.file_name, self.lineno, str(self.error_message))
+        return "Error occurred in python script name [{0}] line number [{1}] error message [{2}]".format(
+            self.file_name, self.lineno, str(self.error_message)
+        )
+
 
-if __name__=='__main__':
+if __name__ == '__main__':
     try:
-        logger.logging.info("Enter the try block")
-        a=1/0
-        print("This will not be printed",a)
+        a = 1 / 0
     except Exception as e:
-           raise NetworkSecurityException(e,sys)
+        raise WeatherPredictionException(e, sys)

models/weather-prediction/src/logging/__init__.py

@@ -0,0 +1 @@
+from src.logging.logger import logging

models/weather-prediction/src/logging/logger.py

@@ -1,15 +1,18 @@
+"""
+Logging configuration for Weather Prediction pipeline.
+Creates timestamped log files in the logs directory.
+"""
 import logging
 import os
 from datetime import datetime
 
-LOG_FILE=f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
+LOG_FILE = f"{datetime.now().strftime('%m_%d_%Y_%H_%M_%S')}.log"
 
-logs_path=os.path.join(os.getcwd(), "logs", LOG_FILE)
+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)
+LOG_FILE_PATH = os.path.join(logs_path, LOG_FILE)
 
 logging.basicConfig(
     filename=LOG_FILE_PATH,
@@ -17,4 +20,13 @@ logging.basicConfig(
     level=logging.INFO
 )
 
+# Also add console handler for visibility
+console_handler = logging.StreamHandler()
+console_handler.setLevel(logging.INFO)
+console_handler.setFormatter(logging.Formatter(
+    "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+))
 
+# Get root logger and add console handler
+root_logger = logging.getLogger()
+root_logger.addHandler(console_handler)

src/graphs/economicalAgentGraph.py

@@ -60,9 +60,9 @@ class EconomicalGraphBuilder:
 
         main_graph = StateGraph(EconomicalAgentState)
 
-        main_graph.add_node("official_sources_module", official_subgraph.invoke)
-        main_graph.add_node("social_media_module", social_subgraph.invoke)
-        main_graph.add_node("feed_generation_module", feed_subgraph.invoke)
+        main_graph.add_node("official_sources_module", lambda state: official_subgraph.invoke(state))
+        main_graph.add_node("social_media_module", lambda state: social_subgraph.invoke(state))
+        main_graph.add_node("feed_generation_module", lambda state: feed_subgraph.invoke(state))
         main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds)
 
         main_graph.set_entry_point("official_sources_module")

src/graphs/intelligenceAgentGraph.py

@@ -60,9 +60,9 @@ class IntelligenceGraphBuilder:
 
         main_graph = StateGraph(IntelligenceAgentState)
 
-        main_graph.add_node("profile_monitoring_module", profile_subgraph.invoke)
-        main_graph.add_node("competitive_intelligence_module", intelligence_subgraph.invoke)
-        main_graph.add_node("feed_generation_module", feed_subgraph.invoke)
+        main_graph.add_node("profile_monitoring_module", lambda state: profile_subgraph.invoke(state))
+        main_graph.add_node("competitive_intelligence_module", lambda state: intelligence_subgraph.invoke(state))
+        main_graph.add_node("feed_generation_module", lambda state: feed_subgraph.invoke(state))
         main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds)
 
         main_graph.set_entry_point("profile_monitoring_module")

src/graphs/meteorologicalAgentGraph.py

@@ -60,9 +60,9 @@ class MeteorologicalGraphBuilder:
 
         main_graph = StateGraph(MeteorologicalAgentState)
 
-        main_graph.add_node("official_sources_module", official_subgraph.invoke)
-        main_graph.add_node("social_media_module", social_subgraph.invoke)
-        main_graph.add_node("feed_generation_module", feed_subgraph.invoke)
+        main_graph.add_node("official_sources_module", lambda state: official_subgraph.invoke(state))
+        main_graph.add_node("social_media_module", lambda state: social_subgraph.invoke(state))
+        main_graph.add_node("feed_generation_module", lambda state: feed_subgraph.invoke(state))
         main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds)
 
         main_graph.set_entry_point("official_sources_module")

src/graphs/politicalAgentGraph.py

@@ -59,9 +59,9 @@ class PoliticalGraphBuilder:
 
         main_graph = StateGraph(PoliticalAgentState)
 
-        main_graph.add_node("official_sources_module", official_subgraph.invoke)
-        main_graph.add_node("social_media_module", social_subgraph.invoke)
-        main_graph.add_node("feed_generation_module", feed_subgraph.invoke)
+        main_graph.add_node("official_sources_module", lambda state: official_subgraph.invoke(state))
+        main_graph.add_node("social_media_module", lambda state: social_subgraph.invoke(state))
+        main_graph.add_node("feed_generation_module", lambda state: feed_subgraph.invoke(state))
         main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds)
 
         main_graph.set_entry_point("official_sources_module")

src/graphs/socialAgentGraph.py

@@ -51,25 +51,39 @@ class SocialGraphBuilder:
 
         return subgraph.compile()
 
+    def build_user_targets_subgraph(self, node: SocialAgentNode) -> StateGraph:
+        """Build subgraph for user-defined keywords and profiles."""
+        subgraph = StateGraph(SocialAgentState)
+        subgraph.add_node("collect_user_targets", node.collect_user_defined_targets)
+        subgraph.set_entry_point("collect_user_targets")
+        subgraph.add_edge("collect_user_targets", END)
+        return subgraph.compile()
+
     def build_graph(self):
         node = SocialAgentNode(self.llm)
 
         trending_subgraph = self.build_trending_subgraph(node)
         social_subgraph = self.build_social_media_subgraph(node)
+        user_targets_subgraph = self.build_user_targets_subgraph(node)
         feed_subgraph = self.build_feed_generation_subgraph(node)
 
         main_graph = StateGraph(SocialAgentState)
 
-        main_graph.add_node("trending_module", trending_subgraph.invoke)
-        main_graph.add_node("social_media_module", social_subgraph.invoke)
-        main_graph.add_node("feed_generation_module", feed_subgraph.invoke)
+        main_graph.add_node("trending_module", lambda state: trending_subgraph.invoke(state))
+        main_graph.add_node("social_media_module", lambda state: social_subgraph.invoke(state))
+        main_graph.add_node("user_targets_module", lambda state: user_targets_subgraph.invoke(state))
+        main_graph.add_node("feed_generation_module", lambda state: feed_subgraph.invoke(state))
         main_graph.add_node("feed_aggregator", node.aggregate_and_store_feeds)
 
+        # Parallel entry points - all 3 modules start together
         main_graph.set_entry_point("trending_module")
         main_graph.set_entry_point("social_media_module")
+        main_graph.set_entry_point("user_targets_module")
 
+        # All modules converge to feed generation
         main_graph.add_edge("trending_module", "feed_generation_module")
         main_graph.add_edge("social_media_module", "feed_generation_module")
+        main_graph.add_edge("user_targets_module", "feed_generation_module")
         main_graph.add_edge("feed_generation_module", "feed_aggregator")
         main_graph.add_edge("feed_aggregator", END)
 

src/nodes/socialAgentNode.py

@@ -5,23 +5,44 @@ Monitors trending topics, events, people, social intelligence across geographic
 
 Updated: Uses Tool Factory pattern for parallel execution safety.
 Each agent instance gets its own private set of tools.
+
+Updated: Now loads user-defined keywords and profiles from intel config.
 """
 
 import json
 import uuid
-from typing import Dict, Any
+import os
+from typing import Dict, Any, List
 from datetime import datetime
 from src.states.socialAgentState import SocialAgentState
 from src.utils.tool_factory import create_tool_set
 from src.llms.groqllm import GroqLLM
 
 
+def load_intel_config() -> dict:
+    """Load intel config from JSON file (same as main.py)."""
+    config_path = os.path.join(os.path.dirname(__file__), "..", "..", "data", "intel_config.json")
+    default_config = {
+        "user_profiles": {"twitter": [], "facebook": [], "linkedin": []},
+        "user_keywords": [],
+        "user_products": []
+    }
+    try:
+        if os.path.exists(config_path):
+            with open(config_path, "r", encoding="utf-8") as f:
+                return json.load(f)
+    except Exception:
+        pass
+    return default_config
+
+
 class SocialAgentNode:
     """
     Modular Social Agent - Geographic social intelligence collection.
     Module 1: Trending Topics (Sri Lanka specific trends)
     Module 2: Social Media (Sri Lanka, Asia, World scopes)
     Module 3: Feed Generation (Categorize, Summarize, Format)
+    Module 4: User-Defined Keywords & Profiles (from frontend config)
 
     Thread Safety:
         Each SocialAgentNode instance creates its own private ToolSet,
@@ -40,6 +61,15 @@ class SocialAgentNode:
         else:
             self.llm = llm
 
+        # Load user-defined intel config (keywords, profiles, products)
+        self.intel_config = load_intel_config()
+        self.user_keywords = self.intel_config.get("user_keywords", [])
+        self.user_profiles = self.intel_config.get("user_profiles", {})
+        self.user_products = self.intel_config.get("user_products", [])
+        
+        print(f"[SocialAgent] Loaded {len(self.user_keywords)} user keywords, "
+              f"{sum(len(v) for v in self.user_profiles.values())} profiles")
+
         # Geographic scopes
         self.geographic_scopes = {
             "sri_lanka": ["sri lanka", "colombo", "srilanka"],
@@ -375,6 +405,111 @@ class SocialAgentNode:
 
         return {"worker_results": world_results, "social_media_results": world_results}
 
+    def collect_user_defined_targets(self, state: SocialAgentState) -> Dict[str, Any]:
+        """
+        Module 2D: Collect data for USER-DEFINED keywords and profiles.
+        These are configured via the frontend Intelligence Settings UI.
+        """
+        print("[MODULE 2D] Collecting User-Defined Targets")
+        
+        user_results = []
+        
+        # Reload config to get latest user settings
+        self.intel_config = load_intel_config()
+        self.user_keywords = self.intel_config.get("user_keywords", [])
+        self.user_profiles = self.intel_config.get("user_profiles", {})
+        self.user_products = self.intel_config.get("user_products", [])
+        
+        # Skip if no user config
+        if not self.user_keywords and not any(self.user_profiles.values()):
+            print("  ⏭️ No user-defined targets configured")
+            return {"worker_results": [], "user_target_results": []}
+        
+        # ============================================
+        # Scrape USER KEYWORDS across Twitter
+        # ============================================
+        if self.user_keywords:
+            print(f"  📝 Scraping {len(self.user_keywords)} user keywords...")
+            twitter_tool = self.tools.get("scrape_twitter")
+            
+            for keyword in self.user_keywords[:10]:  # Limit to 10 keywords
+                try:
+                    if twitter_tool:
+                        twitter_data = twitter_tool.invoke(
+                            {"query": keyword, "max_items": 5}
+                        )
+                        user_results.append({
+                            "source_tool": "scrape_twitter",
+                            "raw_content": str(twitter_data),
+                            "category": "user_keyword",
+                            "scope": "sri_lanka",
+                            "platform": "twitter",
+                            "keyword": keyword,
+                            "timestamp": datetime.utcnow().isoformat(),
+                        })
+                        print(f"    ✓ Keyword: '{keyword}'")
+                except Exception as e:
+                    print(f"    ⚠️ Keyword '{keyword}' error: {e}")
+        
+        # ============================================
+        # Scrape USER PRODUCTS
+        # ============================================
+        if self.user_products:
+            print(f"  📦 Scraping {len(self.user_products)} user products...")
+            twitter_tool = self.tools.get("scrape_twitter")
+            
+            for product in self.user_products[:5]:  # Limit to 5 products
+                try:
+                    if twitter_tool:
+                        twitter_data = twitter_tool.invoke(
+                            {"query": f"{product} review OR {product} Sri Lanka", "max_items": 3}
+                        )
+                        user_results.append({
+                            "source_tool": "scrape_twitter",
+                            "raw_content": str(twitter_data),
+                            "category": "user_product",
+                            "scope": "sri_lanka",
+                            "platform": "twitter",
+                            "product": product,
+                            "timestamp": datetime.utcnow().isoformat(),
+                        })
+                        print(f"    ✓ Product: '{product}'")
+                except Exception as e:
+                    print(f"    ⚠️ Product '{product}' error: {e}")
+        
+        # ============================================
+        # Scrape USER TWITTER PROFILES
+        # ============================================
+        twitter_profiles = self.user_profiles.get("twitter", [])
+        if twitter_profiles:
+            print(f"  👤 Scraping {len(twitter_profiles)} Twitter profiles...")
+            twitter_tool = self.tools.get("scrape_twitter")
+            
+            for profile in twitter_profiles[:10]:  # Limit to 10 profiles
+                try:
+                    # Clean profile handle
+                    handle = profile.replace("@", "").strip()
+                    if twitter_tool:
+                        # Search for tweets mentioning this profile
+                        twitter_data = twitter_tool.invoke(
+                            {"query": f"from:{handle} OR @{handle}", "max_items": 5}
+                        )
+                        user_results.append({
+                            "source_tool": "scrape_twitter",
+                            "raw_content": str(twitter_data),
+                            "category": "user_profile",
+                            "scope": "sri_lanka",
+                            "platform": "twitter",
+                            "profile": f"@{handle}",
+                            "timestamp": datetime.utcnow().isoformat(),
+                        })
+                        print(f"    ✓ Profile: @{handle}")
+                except Exception as e:
+                    print(f"    ⚠️ Profile @{profile} error: {e}")
+        
+        print(f"  ✅ User targets: {len(user_results)} results collected")
+        return {"worker_results": user_results, "user_target_results": user_results}
+
     # ============================================
     # MODULE 3: FEED GENERATION
     # ============================================

src/rag.py

@@ -42,6 +42,200 @@ except ImportError:
     LANGCHAIN_AVAILABLE = False
     logger.warning("[RAG] LangChain not available")
 
+# Neo4j for graph-based retrieval
+try:
+    from neo4j import GraphDatabase
+    NEO4J_AVAILABLE = True
+except ImportError:
+    NEO4J_AVAILABLE = False
+    logger.warning("[RAG] Neo4j not available")
+
+
+# Keywords that indicate a graph/relationship query
+GRAPH_KEYWORDS = [
+    "connected", "related", "timeline", "before", "after", 
+    "caused by", "followed by", "similar to", "linked", 
+    "what happened", "sequence", "chain of events"
+]
+
+
+def is_graph_query(question: str) -> bool:
+    """Detect if question requires graph traversal."""
+    q_lower = question.lower()
+    return any(kw in q_lower for kw in GRAPH_KEYWORDS)
+
+
+class Neo4jRetriever:
+    """Graph-based retrieval for relationship queries with LAZY initialization."""
+    
+    def __init__(self):
+        self.driver = None
+        self._initialized = False
+        self._init_attempted = False
+        
+    def _lazy_init(self):
+        """Lazy initialization - only connect when actually needed."""
+        if self._init_attempted:
+            return self.driver is not None
+            
+        self._init_attempted = True
+        
+        if not NEO4J_AVAILABLE:
+            logger.info("[Neo4jRetriever] Neo4j package not installed")
+            return False
+            
+        neo4j_uri = os.getenv("NEO4J_URI", "")
+        neo4j_user = os.getenv("NEO4J_USER", "neo4j")
+        neo4j_password = os.getenv("NEO4J_PASSWORD", "")
+        
+        if not neo4j_uri or not neo4j_password:
+            logger.info("[Neo4jRetriever] Neo4j credentials not configured - skipping")
+            return False
+            
+        try:
+            self.driver = GraphDatabase.driver(
+                neo4j_uri, auth=(neo4j_user, neo4j_password)
+            )
+            self.driver.verify_connectivity()
+            self._initialized = True
+            logger.info(f"[Neo4jRetriever] Connected to {neo4j_uri}")
+            return True
+        except Exception as e:
+            logger.warning(f"[Neo4jRetriever] Connection failed (will skip graph queries): {e}")
+            self.driver = None
+            return False
+    
+    def get_related_events(self, keyword: str, limit: int = 5) -> List[Dict[str, Any]]:
+        """Find events containing keyword and their related events."""
+        if not self._lazy_init():
+            return []
+            
+        try:
+            with self.driver.session() as session:
+                query = """
+                MATCH (e:Event)
+                WHERE toLower(e.summary) CONTAINS toLower($keyword)
+                OPTIONAL MATCH (e)-[:SIMILAR_TO]-(related:Event)
+                RETURN e.event_id as event_id,
+                       e.summary as summary,
+                       e.domain as domain,
+                       e.severity as severity,
+                       e.timestamp as timestamp,
+                       COLLECT(DISTINCT related.summary)[0..3] as related_summaries
+                ORDER BY e.timestamp DESC
+                LIMIT $limit
+                """
+                results = session.run(query, keyword=keyword, limit=limit)
+                
+                events = []
+                for record in results:
+                    events.append({
+                        "event_id": record["event_id"],
+                        "content": record["summary"],
+                        "domain": record["domain"],
+                        "severity": record["severity"],
+                        "timestamp": record["timestamp"],
+                        "related": record["related_summaries"],
+                        "source": "neo4j_graph"
+                    })
+                
+                logger.info(f"[Neo4jRetriever] Found {len(events)} events for '{keyword}'")
+                return events
+                
+        except Exception as e:
+            logger.error(f"[Neo4jRetriever] Query error: {e}")
+            return []
+    
+    def get_domain_events(self, domain: str, limit: int = 5) -> List[Dict[str, Any]]:
+        """Get recent events by domain with relationships."""
+        if not self._lazy_init():
+            return []
+            
+        try:
+            with self.driver.session() as session:
+                query = """
+                MATCH (e:Event)-[:BELONGS_TO]->(d:Domain {name: $domain})
+                OPTIONAL MATCH (e)-[:SIMILAR_TO]-(related:Event)
+                RETURN e.event_id as event_id,
+                       e.summary as summary,
+                       e.severity as severity,
+                       e.timestamp as timestamp,
+                       COUNT(related) as related_count
+                ORDER BY e.timestamp DESC
+                LIMIT $limit
+                """
+                results = session.run(query, domain=domain.lower(), limit=limit)
+                
+                events = []
+                for record in results:
+                    events.append({
+                        "event_id": record["event_id"],
+                        "content": record["summary"],
+                        "domain": domain,
+                        "severity": record["severity"],
+                        "timestamp": record["timestamp"],
+                        "related_count": record["related_count"],
+                        "source": "neo4j_graph"
+                    })
+                
+                return events
+                
+        except Exception as e:
+            logger.error(f"[Neo4jRetriever] Domain query error: {e}")
+            return []
+    
+    def get_event_chain(self, keyword: str, depth: int = 3) -> List[Dict[str, Any]]:
+        """Get temporal chain of related events."""
+        if not self._lazy_init():
+            return []
+            
+        try:
+            with self.driver.session() as session:
+                query = """
+                MATCH (start:Event)
+                WHERE toLower(start.summary) CONTAINS toLower($keyword)
+                OPTIONAL MATCH path = (start)-[:FOLLOWS|SIMILAR_TO*1..3]-(chain:Event)
+                WITH start, COLLECT(DISTINCT chain) as chain_events
+                RETURN start.event_id as start_id,
+                       start.summary as start_summary,
+                       start.timestamp as start_time,
+                       [e IN chain_events | {summary: e.summary, time: e.timestamp}][0..5] as chain
+                LIMIT 1
+                """
+                result = session.run(query, keyword=keyword).single()
+                
+                if result:
+                    return [{
+                        "event_id": result["start_id"],
+                        "content": result["start_summary"],
+                        "timestamp": result["start_time"],
+                        "chain": result["chain"],
+                        "source": "neo4j_chain"
+                    }]
+                return []
+                
+        except Exception as e:
+            logger.error(f"[Neo4jRetriever] Chain query error: {e}")
+            return []
+    
+    def get_stats(self) -> Dict[str, Any]:
+        """Get Neo4j graph statistics."""
+        if not self._initialized or not self.driver:
+            return {"status": "not_initialized" if not self._init_attempted else "disconnected"}
+            
+        try:
+            with self.driver.session() as session:
+                event_count = session.run(
+                    "MATCH (e:Event) RETURN COUNT(e) as count"
+                ).single()["count"]
+                
+                return {
+                    "status": "connected",
+                    "total_events": event_count
+                }
+        except Exception as e:
+            return {"status": "error", "error": str(e)}
+
 
 class MultiCollectionRetriever:
     COLLECTIONS = ["Roger_feeds"]
@@ -52,6 +246,10 @@ class MultiCollectionRetriever:
         )
         self.client = None
         self.collections: Dict[str, Any] = {}
+        
+        # Thread pool for parallel queries
+        from concurrent.futures import ThreadPoolExecutor
+        self._executor = ThreadPoolExecutor(max_workers=4)
 
         if not CHROMA_AVAILABLE:
             logger.error("[RAG] ChromaDB not installed")
@@ -90,43 +288,68 @@ class MultiCollectionRetriever:
             logger.error(f"[RAG] ChromaDB initialization error: {e}")
             self.client = None
 
-    def search(
-        self, query: str, n_results: int = 5, domain_filter: Optional[str] = None
+    def _query_single_collection(
+        self, name: str, collection, query: str, n_results: int, domain_filter: Optional[str]
     ) -> List[Dict[str, Any]]:
-        if not self.client:
-            return []
+        """Query a single collection - used for parallel execution."""
+        results_list = []
+        try:
+            where_filter = None
+            if domain_filter:
+                where_filter = {"domain": domain_filter.lower()}
 
-        all_results = []
+            results = collection.query(
+                query_texts=[query], n_results=n_results, where=where_filter
+            )
 
-        for name, collection in self.collections.items():
-            try:
-                where_filter = None
-                if domain_filter:
-                    where_filter = {"domain": domain_filter.lower()}
+            if results["ids"] and results["ids"][0]:
+                for i, doc_id in enumerate(results["ids"][0]):
+                    doc = results["documents"][0][i] if results["documents"] else ""
+                    meta = results["metadatas"][0][i] if results["metadatas"] else {}
+                    distance = results["distances"][0][i] if results["distances"] else 0
 
-                results = collection.query(
-                    query_texts=[query], n_results=n_results, where=where_filter
-                )
+                    similarity = 1.0 - min(distance / 2.0, 1.0)
 
-                if results["ids"] and results["ids"][0]:
-                    for i, doc_id in enumerate(results["ids"][0]):
-                        doc = results["documents"][0][i] if results["documents"] else ""
-                        meta = results["metadatas"][0][i] if results["metadatas"] else {}
-                        distance = results["distances"][0][i] if results["distances"] else 0
+                    results_list.append({
+                        "id": doc_id,
+                        "content": doc,
+                        "metadata": meta,
+                        "similarity": similarity,
+                        "collection": name,
+                        "domain": meta.get("domain", "unknown"),
+                    })
 
-                        similarity = 1.0 - min(distance / 2.0, 1.0)
+        except Exception as e:
+            logger.warning(f"[RAG] Error querying {name}: {e}")
+        
+        return results_list
 
-                        all_results.append({
-                            "id": doc_id,
-                            "content": doc,
-                            "metadata": meta,
-                            "similarity": similarity,
-                            "collection": name,
-                            "domain": meta.get("domain", "unknown"),
-                        })
+    def search(
+        self, query: str, n_results: int = 5, domain_filter: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """Search all collections in PARALLEL for faster results."""
+        if not self.client:
+            return []
 
+        # Submit parallel queries to all collections
+        from concurrent.futures import as_completed
+        
+        futures = {}
+        for name, collection in self.collections.items():
+            future = self._executor.submit(
+                self._query_single_collection, 
+                name, collection, query, n_results, domain_filter
+            )
+            futures[future] = name
+        
+        # Collect results as they complete (fastest first)
+        all_results = []
+        for future in as_completed(futures, timeout=10.0):  # 10s timeout
+            try:
+                results = future.result()
+                all_results.extend(results)
             except Exception as e:
-                logger.warning(f"[RAG] Error querying {name}: {e}")
+                logger.warning(f"[RAG] Parallel query failed for {futures[future]}: {e}")
 
         all_results.sort(key=lambda x: x["similarity"], reverse=True)
         return all_results[: n_results * 2]
@@ -152,6 +375,9 @@ class MultiCollectionRetriever:
 class RogerRAG:
     def __init__(self):
         self.retriever = MultiCollectionRetriever()
+        # Neo4j disabled for faster startup - uncomment when graph DB is configured
+        # self.neo4j_retriever = Neo4jRetriever()  # Graph-based retrieval
+        self.neo4j_retriever = None  # Disabled
         self.llm = None
         self.chat_history: List[Tuple[str, str]] = []
 
@@ -165,29 +391,51 @@ class RogerRAG:
                 logger.error("[RAG] GROQ_API_KEY not set")
                 return
 
+            # Using Llama 4 Maverick 17B for fast, high-quality responses
             self.llm = ChatGroq(
                 api_key=api_key,
-                model="openai/gpt-oss-120b",
+                model="meta-llama/llama-4-maverick-17b-128e-instruct",
                 temperature=0.3,
                 max_tokens=1024,
+                request_timeout=30,  # 30 second timeout
             )
-            logger.info("[RAG] Groq LLM initialized")
+            logger.info("[RAG] Groq LLM initialized with Llama 4 Maverick 17B")
 
         except Exception as e:
             logger.error(f"[RAG] LLM initialization error: {e}")
 
-    def _format_context(self, docs: List[Dict[str, Any]]) -> str:
+    def _extract_keywords(self, question: str) -> List[str]:
+        """Extract key terms from question for graph search."""
+        # Remove common stopwords
+        stopwords = {
+            "what", "when", "where", "who", "why", "how", "is", "are", "was", 
+            "were", "the", "a", "an", "to", "of", "in", "on", "for", "with",
+            "about", "related", "connected", "happened", "after", "before",
+            "show", "me", "tell", "find", "get", "events", "timeline"
+        }
+        
+        words = question.lower().replace("?", "").replace(",", "").split()
+        keywords = [w for w in words if w not in stopwords and len(w) > 2]
+        
+        return keywords[:5]  # Return top 5 keywords
+
+    def _format_context(self, docs: List[Dict[str, Any]], include_graph: bool = False) -> str:
         if not docs:
             return "No relevant intelligence data found."
 
         context_parts = []
         now = datetime.now()
+        
+        # Separate ChromaDB and Neo4j results
+        chroma_docs = [d for d in docs if d.get("source") != "neo4j_graph"]
+        graph_docs = [d for d in docs if d.get("source") == "neo4j_graph"]
 
-        for i, doc in enumerate(docs[:5], 1):
+        # Format ChromaDB results
+        for i, doc in enumerate(chroma_docs[:5], 1):
             meta = doc.get("metadata", {})
-            domain = meta.get("domain", "unknown")
+            domain = meta.get("domain", doc.get("domain", "unknown"))
             platform = meta.get("platform", "")
-            timestamp = meta.get("timestamp", "")
+            timestamp = meta.get("timestamp", doc.get("timestamp", ""))
 
             age_str = "unknown date"
             if timestamp:
@@ -199,7 +447,7 @@ class RogerRAG:
                         "%d/%m/%Y",
                     ]:
                         try:
-                            ts_date = datetime.strptime(timestamp[:19], fmt)
+                            ts_date = datetime.strptime(str(timestamp)[:19], fmt)
                             days_old = (now - ts_date).days
                             if days_old == 0:
                                 age_str = "TODAY"
@@ -224,6 +472,22 @@ class RogerRAG:
                 f"TIMESTAMP: {timestamp} ({age_str})\n"
                 f"{doc['content']}\n"
             )
+        
+        # Format Neo4j graph results (if any)
+        if graph_docs:
+            context_parts.append("\n=== RELATED EVENTS FROM KNOWLEDGE GRAPH ===\n")
+            for i, doc in enumerate(graph_docs[:3], 1):
+                related = doc.get("related", [])
+                related_str = ""
+                if related:
+                    related_str = f"\n  Related events: {', '.join(str(r)[:50] + '...' for r in related[:2])}"
+                    
+                context_parts.append(
+                    f"[Graph {i}] Domain: {doc.get('domain', 'unknown')} | "
+                    f"Severity: {doc.get('severity', 'unknown')}\n"
+                    f"{doc.get('content', '')[:500]}"
+                    f"{related_str}\n"
+                )
 
         return "\n---\n".join(context_parts)
 
@@ -266,11 +530,31 @@ class RogerRAG:
         if use_history and self.chat_history:
             search_question = self._reformulate_question(question)
 
+        # ChromaDB semantic search (always)
         docs = self.retriever.search(
             search_question, n_results=5, domain_filter=domain_filter
         )
-
-        if not docs:
+        
+        # Neo4j graph search (for relationship queries) - only if enabled
+        graph_docs = []
+        used_graph = False
+        if self.neo4j_retriever and is_graph_query(search_question):
+            logger.info(f"[RAG] Graph query detected: '{search_question}'")
+            used_graph = True
+            
+            # Extract keywords for graph search
+            # Simple: use first nouns/keywords from question
+            keywords = self._extract_keywords(search_question)
+            
+            for keyword in keywords[:2]:  # Limit to 2 keywords
+                graph_docs.extend(self.neo4j_retriever.get_related_events(keyword, limit=3))
+                
+            logger.info(f"[RAG] Graph retrieval: {len(graph_docs)} docs from Neo4j")
+        
+        # Merge results (ChromaDB + Neo4j)
+        all_docs = docs + graph_docs
+
+        if not all_docs:
             return {
                 "answer": "I couldn't find any relevant intelligence data to answer your question.",
                 "sources": [],
@@ -278,7 +562,7 @@ class RogerRAG:
                 "reformulated": search_question if search_question != question else None,
             }
 
-        context = self._format_context(docs)
+        context = self._format_context(all_docs, include_graph=used_graph)
 
         if not self.llm:
             return {

src/storage/storage_manager.py

@@ -4,6 +4,7 @@ Unified storage manager orchestrating 3-tier deduplication pipeline
 """
 
 import logging
+import re
 from typing import Dict, Any, List, Optional, Tuple
 import csv
 from datetime import datetime
@@ -16,6 +17,14 @@ from .neo4j_graph import Neo4jGraph
 
 logger = logging.getLogger("storage_manager")
 
+# Trending detection integration
+try:
+    from ..utils.trending_detector import record_topic_mention
+    TRENDING_AVAILABLE = True
+except ImportError:
+    TRENDING_AVAILABLE = False
+    logger.warning("[StorageManager] Trending detector not available")
+
 
 class StorageManager:
     """
@@ -133,6 +142,10 @@ class StorageManager:
                 metadata=metadata,
             )
 
+            # Record keywords for trending detection
+            if TRENDING_AVAILABLE:
+                self._record_trending_mentions(summary, domain, metadata)
+
             self.stats["unique_stored"] += 1
             logger.debug(f"[STORE] Stored event {event_id[:8]}... in all databases")
 
@@ -140,6 +153,87 @@ class StorageManager:
             self.stats["errors"] += 1
             logger.error(f"[STORE] Error storing event: {e}")
 
+    def _extract_keywords(self, text: str, max_keywords: int = 5) -> List[str]:
+        """
+        Extract significant keywords from text for trending detection.
+        
+        Args:
+            text: Text to extract keywords from
+            max_keywords: Maximum number of keywords to return
+            
+        Returns:
+            List of keywords (2-3 word phrases)
+        """
+        # Common stopwords to filter out
+        stopwords = {
+            "the", "is", "at", "which", "on", "a", "an", "and", "or", "but",
+            "in", "with", "to", "for", "of", "as", "by", "from", "that", "this",
+            "be", "are", "was", "were", "been", "being", "have", "has", "had",
+            "do", "does", "did", "will", "would", "could", "should", "may",
+            "might", "must", "shall", "can", "need", "dare", "ought", "used",
+            "सिंहल", "தமிழ்",  # Common Sinhala/Tamil particles
+        }
+        
+        # Clean text
+        text = text.lower()
+        text = re.sub(r'http\S+|www\.\S+', '', text)  # Remove URLs
+        text = re.sub(r'[^\w\s]', ' ', text)  # Remove punctuation
+        
+        # Split into words
+        words = text.split()
+        
+        # Filter stopwords and short words
+        filtered = [w for w in words if w not in stopwords and len(w) > 2]
+        
+        # Extract significant words (prioritize proper nouns, locations, etc.)
+        keywords = []
+        
+        # Single important words (capitalized in original or long words)
+        for word in filtered[:20]:
+            if len(word) > 4:  # Longer words are often more significant
+                keywords.append(word)
+        
+        # Deduplicate and limit
+        seen = set()
+        unique_keywords = []
+        for kw in keywords:
+            if kw not in seen:
+                seen.add(kw)
+                unique_keywords.append(kw)
+        
+        return unique_keywords[:max_keywords]
+
+    def _record_trending_mentions(
+        self, 
+        summary: str, 
+        domain: str, 
+        metadata: Optional[Dict[str, Any]] = None
+    ):
+        """
+        Extract keywords from summary and record them for trending detection.
+        
+        Args:
+            summary: Event summary text
+            domain: Event domain (political, economical, etc.)
+            metadata: Optional metadata with platform info
+        """
+        try:
+            keywords = self._extract_keywords(summary)
+            source = metadata.get("platform", "scraper") if metadata else "scraper"
+            
+            for keyword in keywords:
+                record_topic_mention(
+                    topic=keyword,
+                    source=source,
+                    domain=domain
+                )
+            
+            if keywords:
+                logger.debug(f"[TRENDING] Recorded {len(keywords)} keywords: {keywords[:3]}...")
+                
+        except Exception as e:
+            logger.warning(f"[TRENDING] Error recording mentions: {e}")
+
     def link_similar_events(self, event_id_1: str, event_id_2: str, similarity: float):
         """Create similarity link in Neo4j"""
         self.neo4j.link_similar_events(event_id_1, event_id_2, similarity)

src/utils/.browser_data/linkedin/BrowserMetrics-spare.pma

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

src/utils/.browser_data/linkedin/Crashpad/reports/1bb2b465-675d-47f0-b953-a844af38ce6b.dmp

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

src/utils/.browser_data/linkedin/Crashpad/reports/55792d7f-8397-4730-8518-c50a507a611a.dmp

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

src/utils/.browser_data/linkedin/Crashpad/reports/880fc1e0-3241-4d76-a26b-0f9d6135dcd6.dmp

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