Change it to only a scrapper

.dockerignore

@@ -0,0 +1,36 @@
+# Virtualenvs (start.sh creates scrapper_env locally)
+scrapper_env/
+*_env/
+venv/
+.venv/
+
+# Python caches / build artifacts
+__pycache__/
+*.py[cod]
+*.egg-info/
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/
+
+# Local databases — must not be baked into the image
+*.db
+*.sqlite
+*.sqlite3
+
+# Secrets / local env
+.env
+.env.*
+
+# Playwright browser cache (installed inside the image via the build step)
+.cache/
+ms-playwright/
+
+# Logs and OS cruft
+*.log
+.DS_Store
+
+# VCS / tooling
+.git/
+.gitignore
+.dockerignore
+Dockerfile

Dockerfile

@@ -1,72 +1,95 @@
 FROM python:3.10-slim
 
-# =========================
-# System dependencies
-# =========================
+# ----------------------------------------------------
+# System dependencies (Playwright Chromium runtime libs)
+# ----------------------------------------------------
 RUN apt-get update && apt-get install -y \
-    curl wget git build-essential ca-certificates \
-    libnss3 libatk-bridge2.0-0 libdrm2 libxkbcommon0 \
-    libgtk-3-0 libgbm1 libasound2 libxshmfence1 \
-    libxcomposite1 libxdamage1 libxfixes3 libxrandr2 \
-    libxext6 libx11-6 libxcb1 libxrender1 libxi6 \
+    curl \
+    wget \
+    ca-certificates \
+    libnss3 \
+    libatk-bridge2.0-0 \
+    libdrm2 \
+    libxkbcommon0 \
+    libgtk-3-0 \
+    libgbm1 \
+    libasound2 \
+    libxshmfence1 \
+    libxcomposite1 \
+    libxdamage1 \
+    libxfixes3 \
+    libxrandr2 \
+    libxext6 \
+    libx11-6 \
+    libxcb1 \
+    libxrender1 \
+    libxi6 \
     && rm -rf /var/lib/apt/lists/*
 
-# =========================
-# User (HF Spaces requirement)
-# =========================
+# ----------------------------------------------------
+# Create application user
+# ----------------------------------------------------
 RUN useradd -m -u 1000 user
 
-# =========================
-# Directories (IMPORTANT: /data persists)
-# =========================
-RUN mkdir -p /home/user/app \
+# ----------------------------------------------------
+# Application directories
+# NOTE: Do NOT create /data here.
+# Hugging Face mounts /data at runtime.
+# ----------------------------------------------------
+RUN mkdir -p \
+    /home/user/app \
     /home/user/app/models \
     /home/user/app/ollama \
-    /data/hf-cache \
-    /data/logs \
-    && chown -R user:user /home/user /data
+    && chown -R user:user /home/user
+
+# ----------------------------------------------------
+# Runtime environment
+# ----------------------------------------------------
+ENV HF_HOME=/data/hf-cache
+ENV HF_HUB_CACHE=/data/hf-cache
+ENV TRANSFORMERS_CACHE=/data/hf-cache
 
-# =========================
-# Env
-# =========================
 ENV HOME=/home/user
 ENV PATH=/home/user/.local/bin:$PATH
 ENV PYTHONUNBUFFERED=1
+ENV PYTHONDONTWRITEBYTECODE=1
 
-ENV HF_HOME=/data/hf-cache
-ENV HF_HUB_CACHE=/data/hf-cache
-ENV TRANSFORMERS_CACHE=/data/hf-cache
-ENV PLAYWRIGHT_BROWSERS_PATH=/data/hf-cache/playwright
+# ----------------------------------------------------
+# Switch to non-root user
+# ----------------------------------------------------
+USER user
 
 WORKDIR /home/user/app
 
-# =========================
-# COPY requirements FIRST (IMPORTANT)
-# =========================
+# ----------------------------------------------------
+# Python dependencies
+# ----------------------------------------------------
 COPY --chown=user:user requirements.txt .
 
-# =========================
-# Install Python deps
-# =========================
 RUN pip install --no-cache-dir --upgrade pip && \
     pip install --no-cache-dir -r requirements.txt
 
-# =========================
-# Playwright install
-# =========================
-RUN python -m playwright install --with-deps chromium
-RUN pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu
+# ----------------------------------------------------
+# Playwright Chromium
+# ----------------------------------------------------
+RUN python -m playwright install chromium
 
-# =========================
-# Copy app code
-# =========================
+# ----------------------------------------------------
+# Application code
+# ----------------------------------------------------
 COPY --chown=user:user . .
 
-# =========================
-# Switch to user
-# =========================
-USER user
-
 EXPOSE 7860
 
+# ----------------------------------------------------
+# Health check
+# ----------------------------------------------------
+HEALTHCHECK --interval=30s --timeout=5s --start-period=30s --retries=3 \
+    CMD curl -fsS -X POST http://localhost:7860/action \
+        -H "Content-Type: application/json" \
+        -d '{"action":"HEALTH_CHECK"}' || exit 1
+
+# ----------------------------------------------------
+# Start application
+# ----------------------------------------------------
 CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

actions.py

@@ -11,21 +11,16 @@ from database import SessionLocal
 from config import get_auth_token, reload_env
 from db_helpers import (
     add_source,
-    get_analyzed_articles,
-    get_raw_html,
-    get_sector_sentiments,
+    get_one_raw_html,
+    delete_raw_html,
     get_stats,
 )
 
-# Import the new per-worker control API from the refactored workers module.
 from workers import (
     start_workers,
     stop_workers,
-    stop_worker,
-    restart_worker,
     worker_status,
     shutdown_server,
-    _WORKER_REGISTRY,
 )
 
 # =========================
@@ -44,11 +39,7 @@ class ActionRequest(BaseModel):
 
 
 def verify_auth(x_auth_token: Optional[str] = Header(None)):
-    """Reject requests when BACKEND_AUTH_TOKEN is set and the header doesn't match.
-
-    Reads get_auth_token() each call so a reload_env() in handle_start()
-    immediately changes the accepted token without a process restart.
-    """
+    """Reject requests when BACKEND_AUTH_TOKEN is set and the header doesn't match."""
 
     expected = get_auth_token()
 
@@ -79,21 +70,8 @@ def _ping_database():
         return False
 
 
-def _worker_name_from_action(action: str, prefix: str) -> Optional[str]:
-    """
-    Derive the internal worker name from an action string.
-
-    Examples
-    --------
-    _worker_name_from_action("START_SCRAPER",   "START_")   → "scraper"
-    _worker_name_from_action("STOP_HEADLINE",   "STOP_")    → "headline"
-    _worker_name_from_action("RESTART_CLEANUP", "RESTART_") → "cleanup"
-    """
-    return action[len(prefix) :].lower()
-
-
 # =========================
-# BULK ACTION HANDLERS
+# LIFECYCLE ACTION HANDLERS
 # =========================
 
 
@@ -119,9 +97,7 @@ def handle_start(payload=None):
 
 
 def handle_shutdown(payload=None):
-    """Stop ALL workers gracefully. The server process keeps running so the
-    user can edit .env and call START again — for full process exit, use EXIT.
-    """
+    """Stop ALL workers gracefully; server process keeps running."""
 
     logger.info("handle_shutdown: invoked")
 
@@ -163,118 +139,39 @@ def handle_exit(payload=None):
 
 
 # =========================
-# PER-WORKER ACTION HANDLERS
+# DATA / HANDOFF ACTION HANDLERS
 # =========================
 
 
-def handle_start_worker(payload=None, *, name: str):
-    """Start a single named worker. No-op (with warning) if already running."""
-
-    logger.info(f"handle_start_worker: invoked name={name!r}")
-
-    if name not in _WORKER_REGISTRY:
-        return {
-            "success": False,
-            "error": f"Unknown worker {name!r}. Valid names: {sorted(_WORKER_REGISTRY)}",
-        }
-
-    status = worker_status()
-    if status.get(name):
-        logger.info(f"handle_start_worker: '{name}' already running")
-        return {
-            "success": False,
-            "message": f"Worker '{name}' is already running",
-        }
-
-    start_workers([name])
-    logger.info(f"handle_start_worker: '{name}' started")
-    return {
-        "success": True,
-        "message": f"Worker '{name}' started",
-    }
-
-
-def handle_stop_worker(payload=None, *, name: str):
-    """Stop a single named worker. No-op (with warning) if not running."""
-
-    logger.info(f"handle_stop_worker: invoked name={name!r}")
-
-    if name not in _WORKER_REGISTRY:
-        return {
-            "success": False,
-            "error": f"Unknown worker {name!r}. Valid names: {sorted(_WORKER_REGISTRY)}",
-        }
-
-    status = worker_status()
-    if not status.get(name):
-        logger.info(f"handle_stop_worker: '{name}' is not running")
-        return {
-            "success": False,
-            "message": f"Worker '{name}' is not running",
-        }
-
-    stop_worker(name)
-    logger.info(f"handle_stop_worker: '{name}' stopped")
-    return {
-        "success": True,
-        "message": f"Worker '{name}' stopped",
-    }
-
-
-def handle_restart_worker(payload=None, *, name: str):
-    """Stop a named worker (if running) then immediately restart it."""
-
-    logger.info(f"handle_restart_worker: invoked name={name!r}")
-
-    if name not in _WORKER_REGISTRY:
-        return {
-            "success": False,
-            "error": f"Unknown worker {name!r}. Valid names: {sorted(_WORKER_REGISTRY)}",
-        }
-
-    restart_worker(name)
-    logger.info(f"handle_restart_worker: '{name}' restarted")
-    return {
-        "success": True,
-        "message": f"Worker '{name}' restarted",
-    }
-
+def handle_get_raw_html(payload=None):
+    """Return ONE raw_html row for the LLM service to ingest.
 
-# =========================
-# OTHER ACTION HANDLERS
-# =========================
+    Optional payload: {"type": "news" | "stock"}.
+    """
 
+    logger.info(f"handle_get_raw_html: invoked payload={payload!r}")
 
-def handle_get_analyzed(payload=None):
-    """Return all analyzed (non-soft-deleted) articles as JSON."""
+    type_filter = None
+    if isinstance(payload, dict):
+        type_filter = payload.get("type")
 
-    logger.info("handle_get_analyzed: invoked")
-    articles = get_analyzed_articles()
-    return {
-        "success": True,
-        "count": len(articles),
-        "articles": articles,
-    }
+    data = get_one_raw_html(type_filter)
+    return {"success": True, "data": data}
 
 
-def handle_get_raw(payload=None):
-    "Return all raw html stored in the database"
+def handle_delete_raw_html(payload=None):
+    """Delete a raw_html row once the consumer has stored its own copy.
 
-    logger.info("handle_get_raw: invoked")
-    raw_html = get_raw_html()
-    return {"success": True, "count": len(raw_html), "data": raw_html}
+    Payload: {"id": "<uuid>"}.
+    """
 
+    logger.info(f"handle_delete_raw_html: invoked payload={payload!r}")
 
-def handle_get_sector(payload=None):
-    """Return the sector rollup table (one row per sector) as JSON."""
+    if not isinstance(payload, dict) or not payload.get("id"):
+        return {"success": False, "error": "payload must include an 'id'"}
 
-    logger.info("handle_get_sector: invoked")
-    sectors = get_sector_sentiments()
-    return {
-        "success": True,
-        "count": len(sectors),
-        "sectors": sectors,
-    }
+    deleted = delete_raw_html(payload["id"])
+    return {"success": True, "deleted": deleted}
 
 
 def handle_add_sources(payload=None):
@@ -311,7 +208,6 @@ def handle_health_check(payload=None):
     elif all(workers.values()):
         status = "ok"
     else:
-        # Some workers running, some not.
         status = "degraded"
 
     logger.info(f"handle_health_check: status={status} workers={workers} db_ok={db_ok}")
@@ -327,57 +223,27 @@ def handle_health_check(payload=None):
 # =========================
 # ACTIONS DISPATCH
 # =========================
-# Bulk / lifecycle actions are registered directly.
-# Per-worker actions (START_*, STOP_*, RESTART_*) are built dynamically from
-# _WORKER_REGISTRY so adding a new worker to workers.py automatically exposes
-# all three control actions here with zero extra boilerplate.
 
 
 def _build_actions() -> dict:
-    actions = {
-        # --- lifecycle ---
+    # The scrapper runs all its workers (scraper, stock_enhancer, macro) as a
+    # single unit: START launches them all, SHUTDOWN stops them all. There is no
+    # per-worker control here by design — see the LLM service for granular
+    # per-worker actions.
+    return {
+        # --- lifecycle (all workers together) ---
         "START": handle_start,
         "SHUTDOWN": handle_shutdown,
-        "EXIT": handle_exit,
-        # --- data ---
-        # this is for sentiment model - pipeline
-        "GET_ANALYZED": handle_get_analyzed,
-        # add news sources
+        # "EXIT": handle_exit,
+        # --- handoff API (consumed by the LLM service) ---
+        "GET_RAW_HTML": handle_get_raw_html,
+        "DELETE_RAW_HTML": handle_delete_raw_html,
+        # --- sources ---
         "ADD_SOURCES": handle_add_sources,
-        # this is for raw html model - pipeline
-        "GET_RAW_DATA": handle_get_raw,
-        # "GET_SECTOR": handle_get_sector,
         # --- diagnostics ---
         "HEALTH_CHECK": handle_health_check,
         "GET_STATS": get_work_done,
     }
 
-    # Dynamically register START_<WORKER>, STOP_<WORKER>, RESTART_<WORKER>
-    # for every entry in the worker registry.
-    #
-    # e.g. _WORKER_REGISTRY keys: scraper, headline, sentiment, cleanup, purge
-    #
-    # Registered actions:
-    #   START_SCRAPER   / STOP_SCRAPER   / RESTART_SCRAPER
-    #   START_HEADLINE  / STOP_HEADLINE  / RESTART_HEADLINE
-    #   START_SENTIMENT / STOP_SENTIMENT / RESTART_SENTIMENT
-    #   START_CLEANUP   / STOP_CLEANUP   / RESTART_CLEANUP
-    #   START_PURGE     / STOP_PURGE     / RESTART_PURGE
-
-    for worker_name in _WORKER_REGISTRY:
-        key = worker_name.upper()
-
-        actions[f"START_{key}"] = lambda p=None, n=worker_name: handle_start_worker(
-            p, name=n
-        )
-        actions[f"STOP_{key}"] = lambda p=None, n=worker_name: handle_stop_worker(
-            p, name=n
-        )
-        actions[f"RESTART_{key}"] = lambda p=None, n=worker_name: handle_restart_worker(
-            p, name=n
-        )
-
-    return actions
-
 
 ACTIONS = _build_actions()

config.py

@@ -1,70 +1,73 @@
 import os
+
 from dotenv import load_dotenv
-from huggingface_hub import hf_hub_download
 
 # =========================
 # PATHS
 # =========================
 
-MODEL_PATH = "Qwen/Qwen2.5-3B-Instruct"
-
 BACKEND_DIR = os.path.dirname(os.path.abspath(__file__))
-# MODEL_PATH = os.path.join(
-#     BACKEND_DIR, "model", "Qwen2.5-3B-Instruct-GGUF/qwen2.5-3b-instruct-q4_k_m.gguf"
-# )
-
-# Our own sentiment model is an ONNX graph run directly via ONNX Runtime
-# (NOT through Ollama). MODEL_SENTIMENT_OWN is the .onnx file; its tokenizer
-# files (HuggingFace export layout) are expected alongside it in the same dir.
-MODEL_REPO = "sbasu2512/financial_sentiment_model/finbert_financial_model"
-# MODEL_SENTIMENT_OWN = hf_hub_download(
-#     repo_id=MODEL_REPO,
-#     filename="model/sentiment_analyzer_onnx/model.onnx",
-# )
-# MODEL_SENTIMENT_OWN_DIR = os.path.dirname(MODEL_SENTIMENT_OWN)
-
-# Max token length used when tokenizing a headline for the ONNX sentiment model.
-MODEL_SENTIMENT_OWN_MAX_LEN = 300
-STOCK_PATH = os.path.join(BACKEND_DIR, "ind_nifty_list.csv")
+
+# One CSV per market. Each must share the same columns
+# (Company Name, Industry, Symbol, Series, ISIN Code, Country, Exchange).
+# load_stocks_to_db() loads every path here, so adding a new market is just
+# dropping in a CSV and listing it below.
+STOCK_PATHS = [
+    os.path.join(BACKEND_DIR, "ind_nifty_list.csv"),
+    os.path.join(BACKEND_DIR, "us_stocks_list.csv"),
+]
 SOURCES_PATH = os.path.join(BACKEND_DIR, "news_sources.json")
+
+# Config-driven macro/market series (FII/DII, VIX, commodities, indices, FX,
+# bond yields). One row per series; see load_macro_sources_to_db().
+MACRO_SOURCES_PATH = os.path.join(BACKEND_DIR, "macro_sources.csv")
+
 DOTENV_PATH = os.path.join(BACKEND_DIR, ".env")
-HF_model_cache = "/data/hf-cache"
+
+
+# =========================
+# MARKETS
+# =========================
+# Drives both the close-time scheduler (utils.py builds ZoneInfo from `tz`) and
+# the per-market "fetch after close" gating in the stock/macro workers. Adding a
+# market is a one-line change here. tz is kept as a string so config has no
+# import dependency on utils/zoneinfo.
+
+MARKET_CONFIG = {
+    "INDIA": {
+        "tz": "Asia/Kolkata",
+        "close": (15, 30),  # NSE/BSE close 15:30 IST
+        "exchanges": ["NSE", "BSE"],
+    },
+    "US": {
+        "tz": "America/New_York",
+        "close": (16, 0),  # NYSE/NASDAQ close 16:00 ET
+        "exchanges": ["NASDAQ", "NYSE"],
+    },
+}
+
 
 # =========================
 # TIMING / THRESHOLDS
 # =========================
 
-# Re-fetch each source 3 hours after its last attempt; check every 60s.
-SCRAPE_INTERVAL_MS = 3 * 60 * 60 * 1000
-SCRAPER_TICK_SECONDS = 60 * 10
+# News: re-fetch each source once an hour. The loop wakes more often than the
+# interval so a source becomes eligible promptly after its hour elapses (don't
+# set the tick == interval, or the effective cadence drifts).
+SCRAPE_INTERVAL_MS = 60 * 60 * 1000
+SCRAPER_TICK_SECONDS = 60 * 5
 
 # Disable a source after this many consecutive failures.
 MAX_CONSECUTIVE_FAILURES = 5
 
-# Hard-delete soft-deleted rows once they've been marked for this long.
-PURGE_AFTER_MS = 24 * 60 * 60 * 1000
-PURGE_TICK_SECONDS = 60 * 60
-
-# Idle / backoff intervals
-HEADLINE_IDLE_SECONDS = 45 * 60
-SENTIMENT_IDLE_SECONDS = 25 * 60
-SENTIMENT_DATASET_IDLE_SECONDS = 25 * 60
-SENTIMENT_TABLE_IDLE_SECONDS = 25 * 60
-CLEANUP_TICK_SECONDS = 60 * 60
+# Stock enrichment: how long the stock worker idles when no stock is due, and
+# how long after the last fetch a stock becomes eligible again.
+STOCK_IDLE_SECONDS = 45 * 60
 ENRICHMENT_RETRY_MS = 6 * 60 * 60 * 1000
-# A clean_headlines row is eligible for soft-delete once it has been analyzed
-# and fetched more than this many times via GET_CLEAN_ANALYZED.
-CLEAN_HEADLINE_MAX_FETCHES = 3
+
 WORKER_ERROR_BACKOFF_SECONDS = 30
 HTTP_REQUEST_TIMEOUT_SECONDS = 10
 
-# Sector sentiment rollup (sector_worker): aggregate the sentiment table once
-# per 24h. "today" is the trailing 24h; the 7-day average spans the trailing
-# 7 days (inclusive). Both are confidence-weighted over raw sentiment rows.
-SECTOR_TICK_SECONDS = 24 * 60 * 60
-SECTOR_TODAY_WINDOW_MS = 24 * 60 * 60 * 1000
-SECTOR_7D_WINDOW_MS = 7 * 24 * 60 * 60 * 1000
-
 # =========================
 # ENV / DOTENV
 # =========================

database.py

@@ -1,7 +1,7 @@
 from sqlalchemy import create_engine, event
 from sqlalchemy.orm import sessionmaker, declarative_base
 
-DATABASE_URL = "sqlite:////data/webscrapper.db"
+DATABASE_URL = "sqlite:///./scrapper.db"
 
 engine = create_engine(
     DATABASE_URL,
@@ -25,6 +25,10 @@ def _set_sqlite_pragma(dbapi_connection, connection_record):
     cursor.close()
 
 
-SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+)
 
 Base = declarative_base()

db_helpers.py

@@ -1,86 +1,25 @@
 import json
-import re
 import time
-from datetime import datetime, timedelta, timezone
-from bs4 import BeautifulSoup
+from datetime import datetime, timedelta
+
 from sqlalchemy import or_, text, func
 from database import engine, SessionLocal
 import pandas as pd
-import numpy as np
 from logger import logger
 from models import (
     RawHTML,
-    ParsedArticle,
     NewsSource,
-    Sentiment,
-    Sector,
-    Stock,
     StockSources,
+    MacroSource,
 )
 from config import (
-    ENRICHMENT_RETRY_MS,
     SOURCES_PATH,
     MAX_CONSECUTIVE_FAILURES,
-    SECTOR_TODAY_WINDOW_MS,
-    SECTOR_7D_WINDOW_MS,
-    STOCK_PATH,
+    STOCK_PATHS,
+    MACRO_SOURCES_PATH,
+    MARKET_CONFIG,
 )
-
-
-def clean_headline_text(text: str) -> str:
-    if not text:
-        logger.warning("clean_headline_text: empty/invalid headline text")
-        return ""
-
-    text = text.strip()
-
-    # Remove markdown bullets
-    text = re.sub(r"^\s*[-*•]+\s*", "", text)
-    text = re.sub(r"^\s*[-–—•*]+\s*", "", text)
-
-    # Remove numbered lists
-    text = re.sub(r"^\s*\d+[.)]\s*", "", text)
-
-    # Remove lettered lists
-    text = re.sub(r"^\s*[a-zA-Z][.)]\s*", "", text)
-
-    # Remove surrounding quotes
-    text = text.strip("'\"`")
-
-    # Collapse whitespace
-    text = re.sub(r"\s+", " ", text)
-
-    return text
-
-
-def looks_like_headline(text: str) -> bool:
-    if not text:
-        return False
-
-    if len(text) < 20:
-        return False
-
-    if len(text) > 250:
-        return False
-
-    if text.count(" ") < 2:
-        return False
-
-    blacklist = [
-        "click here",
-        "read more",
-        "subscribe",
-        "advertisement",
-        "cookie policy",
-        "sign in",
-    ]
-
-    lower = text.lower()
-
-    if any(x in lower for x in blacklist):
-        return False
-
-    return True
+from utils import get_market_day_start_ms
 
 
 # =========================
@@ -102,33 +41,39 @@ def migrate_schema():
         return
 
     needed = {
+        "raw_html": [
+            ("country", "TEXT"),
+        ],
         "news_sources": [
             ("last_attempted_at", "INTEGER"),
             ("failure_count", "INTEGER NOT NULL DEFAULT 0"),
             ("failure_reason", "TEXT"),
             ("type", "TEXT"),
         ],
-        "parsed_articles": [
-            ("is_cleaned", "BOOLEAN NOT NULL DEFAULT 0"),
-            ("is_deleted", "BOOLEAN NOT NULL DEFAULT 0"),
-            ("deleted_at", "INTEGER"),
-        ],
-        "raw_html": [
-            ("is_deleted", "BOOLEAN NOT NULL DEFAULT 0"),
-            ("deleted_at", "INTEGER"),
+        "stock_sources": [
+            ("url", "TEXT"),
+            ("last_attempted_at", "INTEGER"),
+            ("last_fetched_at", "INTEGER"),
+            ("failure_count", "INTEGER NOT NULL DEFAULT 0"),
+            ("failure_reason", "TEXT"),
+            ("updated_at", "INTEGER"),
         ],
-        "sector": [
-            ("sector", "TEXT NOT NULL UNIQUE"),
-            ("sector_sentiment", "REAL"),
-            ("sector_sentiment_7d_avg", "REAL"),
-            ("sector_momentum", "REAL"),
-            ("is_analyzed", "BOOLEAN NOT NULL DEFAULT 0"),
+        # create_all builds macro_sources fresh; these entries make partial
+        # upgrades idempotent.
+        "macro_sources": [
+            ("name", "TEXT"),
+            ("type", "TEXT"),
+            ("country", "TEXT"),
+            ("market", "TEXT"),
+            ("exchange", "TEXT"),
+            ("url", "TEXT"),
+            ("last_fetched_at", "INTEGER"),
+            ("last_attempted_at", "INTEGER"),
+            ("failure_count", "INTEGER NOT NULL DEFAULT 0"),
+            ("failure_reason", "TEXT"),
             ("created_at", "INTEGER"),
             ("updated_at", "INTEGER"),
         ],
-        "sentiment": [
-            ("headline_id", "TEXT"),
-        ],
     }
 
     added = 0
@@ -151,7 +96,24 @@ def migrate_schema():
 
 
 # =========================
-# LOAD SOURCES JSON TO DB
+# GOOGLE FINANCE URL
+# =========================
+
+
+def build_google_finance_url(ticker: str, exchange: str) -> str:
+    """Construct the Google Finance quote URL for a ticker/exchange pair.
+
+    Built once at load time and stored on StockSources.url so the stock worker
+    doesn't reconstruct it every tick.
+    """
+
+    ticker = (ticker or "").strip().upper()
+    exchange = (exchange or "NSE").strip().upper() or "NSE"
+    return f"https://www.google.com/finance/quote/{ticker}:{exchange}?tab=overview"
+
+
+# =========================
+# LOAD NEWS SOURCES JSON TO DB
 # =========================
 
 
@@ -203,62 +165,171 @@ def load_sources_to_db():
 
 
 # =========================
-# LOAD SOURCES JSON TO DB
+# LOAD STOCK SOURCES CSV TO DB
 # =========================
 
 
 def load_stocks_to_db():
+    """Load every market CSV in STOCK_PATHS into stock_sources.
 
-    logger.info(f"load_stocks_to_db: reading {STOCK_PATH}")
+    Each row's Google Finance URL is built from its own exchange, so NSE,
+    NASDAQ, NYSE, etc. are all handled the same way. Rows are de-duplicated on
+    (ticker, exchange) so the same symbol can legitimately exist on two markets.
+    """
 
     db = SessionLocal()
 
     try:
 
-        stocks_df = pd.read_csv(STOCK_PATH)
-
         inserted = 0
 
-        for _, row in stocks_df.iterrows():
+        for path in STOCK_PATHS:
+
+            logger.info(f"load_stocks_to_db: reading {path}")
+
+            try:
+                stocks_df = pd.read_csv(path)
+            except FileNotFoundError:
+                logger.warning(f"load_stocks_to_db: {path} not found; skipping")
+                continue
+
+            for _, row in stocks_df.iterrows():
+
+                company_name = str(row["Company Name"]).strip().lower()
+
+                ticker = str(row["Symbol"]).strip().upper()
+
+                sector = str(row["Industry"]).strip().upper()
+
+                country = str(row["Country"]).strip().upper()
+
+                exchange = str(row["Exchange"]).strip().upper()
+
+                existing = (
+                    db.query(StockSources)
+                    .filter(
+                        StockSources.ticker == ticker,
+                        StockSources.exchange == exchange,
+                    )
+                    .first()
+                )
+
+                if existing:
+                    # Backfill the URL on a pre-existing row that predates this column.
+                    if not existing.url:
+                        existing.url = build_google_finance_url(ticker, exchange)
+                    continue
+
+                stock = StockSources(
+                    company_name=company_name,
+                    ticker=ticker,
+                    sector=sector,
+                    country=country,
+                    exchange=exchange,
+                    url=build_google_finance_url(ticker, exchange),
+                )
+
+                db.add(stock)
+
+                inserted += 1
+
+        db.commit()
+
+        logger.info(f"load_stocks_to_db: inserted " f"{inserted} stocks")
+
+    except Exception:
+
+        db.rollback()
+
+        logger.exception("load_stocks_to_db failed")
+
+        raise
+
+    finally:
+
+        db.close()
+
+
+# =========================
+# LOAD MACRO SOURCES CSV TO DB
+# =========================
+
+
+def load_macro_sources_to_db():
+    """Load macro/market series from MACRO_SOURCES_PATH into MacroSource.
+
+    CSV columns: name, type, country, market, exchange, gf_ticker, url.
+    For Google-Finance series leave `url` blank and supply `gf_ticker`+`exchange`
+    (the URL is built via build_google_finance_url); for raw pages (e.g. NSE
+    FII/DII) supply `url` directly. De-duplicated on (name, type).
+    """
+
+    logger.info(f"load_macro_sources_to_db: reading {MACRO_SOURCES_PATH}")
+
+    db = SessionLocal()
+
+    try:
 
-            company_name = str(row["Company Name"]).strip().lower()
+        try:
+            df = pd.read_csv(MACRO_SOURCES_PATH).fillna("")
+        except FileNotFoundError:
+            logger.warning(
+                f"load_macro_sources_to_db: {MACRO_SOURCES_PATH} not found; skipping"
+            )
+            return
 
-            ticker = str(row["Symbol"]).strip().upper()
+        inserted = 0
 
-            sector = str(row["Industry"]).strip().upper()
+        for _, row in df.iterrows():
 
-            country = str(row["Country"]).strip().upper()
+            name = str(row["name"]).strip()
+            type_ = str(row["type"]).strip().lower()
+            country = str(row.get("country", "")).strip().upper() or None
+            market = str(row["market"]).strip().upper()
+            exchange = str(row.get("exchange", "")).strip().upper() or None
+            gf_ticker = str(row.get("gf_ticker", "")).strip().upper()
+            raw_url = str(row.get("url", "")).strip()
 
-            exchange = str(row["Exchange"]).strip().upper()
+            # Materialize the fetch URL: explicit url wins; otherwise build a
+            # Google Finance quote URL from gf_ticker:exchange.
+            url = raw_url or (
+                build_google_finance_url(gf_ticker, exchange) if gf_ticker else None
+            )
 
             existing = (
-                db.query(StockSources).filter(StockSources.ticker == ticker).first()
+                db.query(MacroSource)
+                .filter(MacroSource.name == name, MacroSource.type == type_)
+                .first()
             )
 
             if existing:
+                # Backfill a URL onto a pre-existing row that predates this load.
+                if not existing.url and url:
+                    existing.url = url
                 continue
 
-            stock = StockSources(
-                company_name=company_name,
-                ticker=ticker,
-                sector=sector,
-                country=country,
-                exchange=exchange,
+            db.add(
+                MacroSource(
+                    name=name,
+                    type=type_,
+                    country=country,
+                    market=market,
+                    exchange=exchange,
+                    url=url,
+                )
             )
 
-            db.add(stock)
-
             inserted += 1
 
         db.commit()
 
-        logger.info(f"load_stocks_to_db: inserted " f"{inserted} stocks")
+        logger.info(f"load_macro_sources_to_db: inserted {inserted} macro sources")
 
     except Exception:
 
         db.rollback()
 
-        logger.exception("load_stocks_to_db failed")
+        logger.exception("load_macro_sources_to_db failed")
 
         raise
 
@@ -268,7 +339,7 @@ def load_stocks_to_db():
 
 
 # =========================
-# ADD ONE SOURCE
+# ADD ONE NEWS SOURCE
 # =========================
 
 
@@ -304,7 +375,7 @@ def add_source(name: str, website: str):
                 ),
             }
 
-        news_source = NewsSource(name=name, website=website)
+        news_source = NewsSource(name=name, website=website, type="news")
         db.add(news_source)
         db.commit()
 
@@ -324,7 +395,7 @@ def add_source(name: str, website: str):
 
 
 # =========================
-# READ SOURCES
+# READ NEWS SOURCES
 # =========================
 
 
@@ -339,7 +410,7 @@ def read_sources(filter_option: str):
 
 
 # =========================
-# UPDATE last_fetched_at
+# UPDATE last_fetched_at (news)
 # =========================
 
 
@@ -367,7 +438,7 @@ def update_last_fetched(source_id: str):
 
 
 # =========================
-# RECORD FAILURE
+# RECORD FAILURE (news)
 # =========================
 
 
@@ -406,650 +477,167 @@ def record_failure(source_id: str, reason: str):
 # =========================
 
 
-def store_raw_html(source: str, html: str, type: str):
-    db = SessionLocal()
-    try:
-        soup = BeautifulSoup(html, "lxml")
+def store_raw_html(source: str, html: str, type: str, country: str | None = None):
+    """Persist a fetched page verbatim (true raw HTML — no stripping).
 
-        for tag in soup(
-            ["script", "style", "nav", "footer", "header", "noscript", "svg"]
-        ):
-            tag.decompose()
+    `country` is set for stock pages (carried over from the StockSources row) and
+    left NULL for news. The LLM service does its own BeautifulSoup cleaning when
+    it parses.
+    """
 
-        cleaned_html = soup.get_text(" ", strip=True)
-        entry = RawHTML(source=source, html=cleaned_html, type=type)
+    db = SessionLocal()
+    try:
+        entry = RawHTML(source=source, html=html, type=type, country=country)
         db.add(entry)
         db.commit()
         # entry.id is populated by the model's uuid default before INSERT,
         # so no refresh is needed.
         logger.info(
             f"store_raw_html: inserted {entry.id} "
-            f"(source={source}, {len(cleaned_html)} chars)"
+            f"(source={source}, country={country}, {len(html)} chars)"
         )
         return entry.id
     finally:
         db.close()
 
 
-def get_unparsed_record(type: str):
-    db = SessionLocal()
-    try:
-        base_query = db.query(RawHTML).filter(
-            RawHTML.parsed.is_(False),
-            RawHTML.type.is_(type),
-        )
-
-        total = base_query.count()
-        logger.info(f"get_unparsed_record: found {total} unparsed article(s)")
-
-        record = base_query.order_by(RawHTML.created_at).first()
-
-        if record is None:
-            logger.info("get_unparsed_record: no unparsed rows")
-        else:
-            logger.info(
-                f"get_unparsed_record: returning {record.id} "
-                f"(source={record.source})"
-            )
-        return record
-    finally:
-        db.close()
-
-
-def mark_as_parsed(record_id: str):
-    db = SessionLocal()
-    try:
-        record = db.query(RawHTML).filter(RawHTML.id == record_id).first()
-        if record:
-            record.parsed = True
-            db.commit()
-            logger.info(f"mark_as_parsed: {record_id} -> parsed=True")
-        else:
-            logger.warning(f"mark_as_parsed: {record_id} not found")
-    finally:
-        db.close()
-
-
-# =========================
-# PARSED ARTICLES
-# =========================
-
-
-def store_headlines(raw_html_id: str, headlines: list):
-    db = SessionLocal()
-    inserted_count = 0  # Track how many were actually inserted
-
-    try:
-        for headline in headlines:
-            clean_headlines = clean_headline_text(headline)
-
-            # If it doesn't look like a headline, skip it immediately (Guard Clause)
-            if not looks_like_headline(clean_headlines):
-                continue
-
-            article = ParsedArticle(
-                raw_html_id=raw_html_id,
-                title=clean_headlines,  # Reused the variable instead of calling the function again
-                sentiment=None,
-                category=None,
-                analyzed=False,
-            )
-            db.add(article)
-            inserted_count += 1
-
-        # Commit everything in ONE batch after the loop finishes successfully
-        db.commit()
-
-        logger.info(
-            f"store_headlines: successfully inserted {inserted_count} of "
-            f"{len(headlines)} headlines for raw_html {raw_html_id}"
-        )
-
-    except Exception as e:
-        db.rollback()  # Undo pending changes if an error occurs
-        logger.error(f"Failed to store headlines for raw_html {raw_html_id}: {e}")
-        raise
-
-    finally:
-        db.close()
-
-
-def get_unanalyzed_article():
-    db = SessionLocal()
-    try:
-        article = (
-            db.query(ParsedArticle)
-            .filter(
-                ParsedArticle.analyzed.is_(False), ParsedArticle.is_deleted.is_(False)
-            )
-            .order_by(ParsedArticle.created_at)
-            .first()
-        )
-        if article is None:
-            logger.info("get_unanalyzed_article: no unanalyzed rows")
-        else:
-            preview = (article.title or "")[:60]
-            logger.info(
-                f"get_unanalyzed_article: returning {article.id} " f"title={preview!r}"
-            )
-        return article
-    finally:
-        db.close()
-
-
-def get_analyzed_articles():
-    """Return all analyzed, non-soft-deleted articles as a list of dicts."""
-
-    db = SessionLocal()
-    try:
-        rows = (
-            db.query(ParsedArticle)
-            .filter(
-                ParsedArticle.analyzed.is_(True),
-                ParsedArticle.is_deleted.is_(False),
-            )
-            .order_by(ParsedArticle.created_at)
-            .all()
-        )
-
-        articles = [
-            {
-                "id": r.id,
-                "raw_html_id": r.raw_html_id,
-                "title": r.title,
-                "sentiment": r.sentiment,
-                "category": r.category,
-                "created_at": r.created_at,
-            }
-            for r in rows
-        ]
-
-        logger.info(f"get_analyzed_articles: returning {len(articles)} rows")
-        return articles
-    finally:
-        db.close()
+def get_one_raw_html(type: str | None = None):
+    """Return the oldest raw_html row as a dict (or None).
 
-
-def get_analyzed_article():
-    """Return one analyzed, not-yet-cleaned, non soft deleted article.
-
-    Used by the cleaning_headlines worker, so it excludes rows it has already
-    turned into clean_headlines (is_cleaned=True).
+    Optional `type` filter ("news" / "stock"). Single-consumer + delete-after-
+    ingest means no locking is needed — the consumer deletes the row right after
+    storing its own copy.
     """
-    db = SessionLocal()
-    try:
-        article = (
-            db.query(ParsedArticle)
-            .filter(
-                ParsedArticle.analyzed.is_(True),
-                ParsedArticle.is_cleaned.is_(False),
-                ParsedArticle.is_deleted.is_(False),
-                ParsedArticle.category.is_("business"),
-            )
-            .order_by(ParsedArticle.created_at)
-            .first()
-        )
-
-        if article:
-            logger.info(f"get_analyzed_article found: {article.id}")
-            return article
-        else:
-            logger.info("get_analyzed_article: no uncleaned analyzed article found")
-    finally:
-        db.close()
-
-
-def mark_article_cleaned(article_id: str):
-    """Flag an article as dataset-cleaned so the cleanup worker can retire it."""
 
     db = SessionLocal()
     try:
-        article = db.query(ParsedArticle).filter(ParsedArticle.id == article_id).first()
-        if article:
-            article.is_cleaned = True
-            db.commit()
-            logger.info(f"mark_article_cleaned: {article_id} -> is_cleaned=True")
-        else:
-            logger.warning(f"mark_article_cleaned: {article_id} not found")
-    finally:
-        db.close()
+        query = db.query(RawHTML)
 
+        if type:
+            normalized = str(type).strip().lower()
+            query = query.filter(RawHTML.type == normalized)
 
-def update_article_analysis(article_id: str, sentiment: str, category: str):
-    db = SessionLocal()
-    try:
-        article = db.query(ParsedArticle).filter(ParsedArticle.id == article_id).first()
-        if article:
-            article.sentiment = sentiment.lower()
-            article.category = category.lower()
-            article.analyzed = True
-            db.commit()
-            logger.info(
-                f"update_article_analysis: {article_id} "
-                f"sentiment={sentiment} category={category}"
-            )
-        else:
-            logger.warning(f"update_article_analysis: {article_id} not found")
-    finally:
-        db.close()
-
-
-def soft_delete_invalid_headlines(article_id: str):
-    db = SessionLocal()
-    try:
-        article = db.query(ParsedArticle).filter(ParsedArticle.id == article_id).first()
-        if article:
-            article.analyzed = True
-            article.is_deleted = True
-            db.commit()
-            logger.info(f"soft deleted article: {article_id} - Invalid headline")
-        else:
-            logger.warning(f"soft deleted article: {article_id} not found")
-    finally:
-        db.close()
-
+        record = query.order_by(RawHTML.created_at).first()
 
-# =========================
-# SENTIMENT TABLE
-# =========================
-
-
-def store_sentiment(
-    company: str,
-    sentiment_score: float,
-    confidence: float,
-    sector: str,
-    headline_id: str,
-):
-    """Insert one row into the sentiment results table."""
+        if record is None:
+            logger.info(f"get_one_raw_html: nothing available (type={type!r})")
+            return None
 
-    db = SessionLocal()
-    try:
-        entry = Sentiment(
-            company=company,
-            sentiment_score=sentiment_score,
-            confidence=confidence,
-            sector=sector,
-            headline_id=headline_id,
-        )
-        db.add(entry)
-        db.commit()
         logger.info(
-            f"store_sentiment: inserted {entry.id} "
-            f"(company={company!r}, score={sentiment_score:.4f}, "
-            f"confidence={confidence:.4f})"
+            f"get_one_raw_html: returning {record.id} "
+            f"(source={record.source}, type={record.type})"
         )
-        return entry.id
-    finally:
-        db.close()
-
 
-def get_stats():
-    db = SessionLocal()
-    try:
         return {
-            "raw_html_pending": db.query(RawHTML)
-            .filter(RawHTML.parsed == False)
-            .count(),
-            "articles_pending": db.query(ParsedArticle)
-            .filter(ParsedArticle.analyzed == False)
-            .count(),
-            "articles_analyzed": db.query(ParsedArticle)
-            .filter(ParsedArticle.analyzed == True)
-            .count(),
-            "false_headlines_detected": db.query(ParsedArticle)
-            .filter(
-                ParsedArticle.analyzed.is_(True),
-                ParsedArticle.is_cleaned.is_(True),
-                ParsedArticle.sentiment.is_(None),
-            )
-            .count(),
-            "headlines_cleaned": db.query(ParsedArticle)
-            .filter(
-                ParsedArticle.analyzed.is_(True),
-                ParsedArticle.is_cleaned.is_(True),
-                ParsedArticle.sentiment.is_not(None),
-            )
-            .count(),
+            "id": record.id,
+            "source": record.source,
+            "type": record.type,
+            "country": record.country,
+            "created_at": record.created_at,
+            "html": record.html,
         }
     finally:
         db.close()
 
 
-# =========================
-# SECTOR ROLLUP
-# =========================
-
-
-def _weighted_sector_map(db, cutoff_ms: int) -> dict:
-    """Confidence-weighted sentiment per sector for rows created since cutoff_ms.
+def delete_raw_html(record_id: str) -> bool:
+    """Hard-delete a raw_html row by id. Returns True if a row was removed."""
 
-    Returns {sector: sum(score*confidence)/sum(confidence)}. Sectors with a
-    non-positive confidence sum (e.g. all-neutral) are skipped to avoid div-by-zero.
-    """
-    from llm import is_valid_sector
-
-    rows = (
-        db.query(
-            Sentiment.sector,
-            func.sum(Sentiment.sentiment_score * Sentiment.confidence),
-            func.sum(Sentiment.confidence),
-        )
-        .filter(Sentiment.created_at >= cutoff_ms)
-        .group_by(Sentiment.sector)
-        .all()
-    )
-
-    result = {}
-    for sector, weighted_sum, confidence_sum in rows:
-
-        if not is_valid_sector(sector):
-            continue
-        if confidence_sum is None or confidence_sum <= 0:
-            continue
-        result[sector] = float(weighted_sum) / float(confidence_sum)
-
-    return result
-
-
-def update_sector_table() -> dict:
-    """Recompute the sector rollup from the sentiment table and upsert one row
-    per sector (matched on sector name). Returns {inserted, updated, sectors}.
-    """
-
-    db = SessionLocal()
-    try:
-        now_ms = int(time.time() * 1000)
-
-        today = _weighted_sector_map(db, now_ms - SECTOR_TODAY_WINDOW_MS)
-        seven_day = _weighted_sector_map(db, now_ms - SECTOR_7D_WINDOW_MS)
-
-        logger.info(
-            f"update_sector_table: {len(today)} sector(s) active in last 24h, "
-            f"{len(seven_day)} in last 7d"
-        )
-
-        inserted = 0
-        updated = 0
-
-        for sector, today_value in today.items():
-            avg_7d = seven_day.get(sector, today_value)
-            momentum = today_value - avg_7d
-
-            row = db.query(Sector).filter(Sector.sector == sector).first()
-
-            if row is None:
-                row = Sector(sector=sector)
-                db.add(row)
-                inserted += 1
-            else:
-                updated += 1
-
-            row.sector_sentiment = today_value
-            row.sector_sentiment_7d_avg = avg_7d
-            row.sector_momentum = momentum
-            row.is_analyzed = False
-            row.updated_at = now_ms
-
-        db.commit()
-
-        logger.info(
-            f"update_sector_table: done (inserted={inserted}, updated={updated})"
-        )
-        return {
-            "inserted": inserted,
-            "updated": updated,
-            "sectors": sorted(today.keys()),
-        }
-
-    except Exception:
-        db.rollback()
-        logger.exception("update_sector_table failed")
-        raise
-    finally:
-        db.close()
-
-
-def mark_clean_headlines_past_7d() -> int:
-    """Flag analyzed headlines whose sentiment has aged out of the sector
-    7-day window (created_at older than SECTOR_7D_WINDOW_MS).
+    if not record_id:
+        return False
 
-    Once past the window the row has been fully counted into the last sector
-    7d-average and will not be read again, so set has_been_read_sector=True.
-    Returns the number of rows newly flagged.
-    """
     db = SessionLocal()
     try:
-        cutoff_ms = int(time.time() * 1000) - SECTOR_7D_WINDOW_MS
-
-        # ---> FIX: Chain an .update() call and set synchronize_session=False <---
-        count = (
-            db.query(Sentiment)
-            .filter(
-                ParsedArticle.created_at < cutoff_ms,
-                Sentiment.has_been_read_sector.is_(False),
-            )
-            .update({Sentiment.has_been_read_sector: True}, synchronize_session=False)
-        )
+        record = db.query(RawHTML).filter(RawHTML.id == record_id).first()
+        if not record:
+            logger.warning(f"delete_raw_html: {record_id} not found")
+            return False
 
-        # This now commits the actual UPDATE mutation safely to your database
+        db.delete(record)
         db.commit()
-
-        logger.info(f"mark_clean_headlines_past_7d: flagged {count} row(s)")
-        return count
-
-    except Exception:
-        db.rollback()
-        logger.exception("mark_clean_headlines_past_7d execution failed")
-        raise
+        logger.info(f"delete_raw_html: deleted {record_id}")
+        return True
     finally:
         db.close()
 
 
-def get_sector_sentiments():
-    """Return all sector rollup rows as a list of dicts (highest momentum first)."""
+# =========================
+# SOURCE FETCH SCHEDULING (stock + macro)
+# =========================
+# Both StockSources and MacroSource share the "fetch once per market day" shape,
+# so these helpers are generic over the ORM model. A market closes once a day;
+# rows whose last_fetched_at falls within the current market day are skipped.
 
-    db = SessionLocal()
-    try:
-        rows = db.query(Sector).order_by(Sector.sector_momentum.desc()).all()
-
-        sectors = [
-            {
-                "id": r.id,
-                "sector": r.sector,
-                "sector_sentiment": r.sector_sentiment,
-                "sector_sentiment_7d_avg": r.sector_sentiment_7d_avg,
-                "sector_momentum": r.sector_momentum,
-                "is_analyzed": r.is_analyzed,
-                "created_at": r.created_at,
-                "updated_at": r.updated_at,
-            }
-            for r in rows
-        ]
 
-        logger.info(f"get_sector_sentiments: returning {len(sectors)} rows")
-        return sectors
-    finally:
-        db.close()
+def get_sources_due_for_market(model, market: str):
+    """Return rows of `model` belonging to `market` not yet fetched today.
 
+    StockSources is grouped by `exchange` (from MARKET_CONFIG); MacroSource
+    carries an explicit `market` column. Rows are detached (expunged) so callers
+    can read attributes after the session closes.
+    """
 
-def update_stock_data(source_id: str, data: dict):
+    cfg = MARKET_CONFIG.get(market)
+    if not cfg:
+        logger.warning(f"get_sources_due_for_market: unknown market {market!r}")
+        return []
 
-    logger.debug(f"Incoming dict to save to db: {data}")
+    day_start_ms = get_market_day_start_ms(market)
 
     db = SessionLocal()
-
     try:
+        query = db.query(model)
 
-        source = db.query(StockSources).filter(StockSources.id == source_id).first()
-
-        if not source:
-            logger.warning(f"unable to find stock source id:{source_id}")
-            return
-
-        now_ms = int(time.time() * 1000)
-
-        # UTC day boundaries
-        today = datetime.now(timezone.utc).date()
-
-        existing_stock = (
-            db.query(Stock)
-            .filter(
-                Stock.ticker == source.ticker,
-                Stock.country == source.country,
-                Stock.created_at
-                >= datetime.combine(
-                    today,
-                    datetime.min.time(),
-                    tzinfo=timezone.utc,
-                ),
-            )
-            .order_by(Stock.created_at.desc())
-            .first()
-        )
-
-        if existing_stock:
-
-            existing_stock.open = data.get("open")
-            existing_stock.high = data.get("high")
-            existing_stock.low = data.get("low")
-            existing_stock.close = data.get("close")
-            existing_stock.volume = data.get("volume")
-            existing_stock.is_complete = data.get("is_complete")
-            existing_stock.raw_html_id = data.get("raw_html_id")
-            existing_stock.updated_at = now_ms
-
-            logger.info(f"updated stock row for ticker:{source.ticker}")
-
+        if hasattr(model, "market"):
+            query = query.filter(model.market == market)
         else:
+            query = query.filter(model.exchange.in_(cfg["exchanges"]))
 
-            stock = Stock(
-                ticker=source.ticker,
-                company_name=source.company_name,
-                country=source.country,
-                open=data.get("open"),
-                high=data.get("high"),
-                low=data.get("low"),
-                close=data.get("close"),
-                volume=data.get("volume"),
-                is_complete=data.get("is_complete", False),
-                raw_html_id=data.get("raw_html_id"),
-                created_at=datetime.now(timezone.utc),
-                updated_at=now_ms,
+        query = query.filter(
+            or_(
+                model.last_fetched_at.is_(None),
+                model.last_fetched_at < day_start_ms,
             )
+        ).order_by(model.last_fetched_at.nullsfirst())
 
-            db.add(stock)
+        rows = query.all()
 
-            logger.info(f"created stock row for ticker:{source.ticker}")
+        for row in rows:
+            db.expunge(row)
 
-        db.commit()
-
-    except Exception:
-        db.rollback()
-        raise
+        return rows
 
     finally:
         db.close()
 
 
-def get_today_bounds():
-
-    now = datetime.now()
+def update_source_table(model, id: str, error: str | None = None):
+    """Stamp a fetch attempt on a stock/macro source row.
 
-    start = datetime(
-        now.year,
-        now.month,
-        now.day,
-        0,
-        0,
-        0,
-    )
-
-    end = datetime(
-        now.year,
-        now.month,
-        now.day,
-        23,
-        59,
-        59,
-    )
-
-    return (
-        int(start.timestamp() * 1000),
-        int(end.timestamp() * 1000),
-    )
-
-
-def get_today_stock(
-    db,
-    ticker: str,
-):
-
-    ticker = str(ticker).strip().upper()
-
-    start_ms, end_ms = get_today_bounds()
-
-    return (
-        db.query(Stock)
-        .filter(
-            Stock.ticker == ticker,
-            Stock.created_at >= start_ms,
-            Stock.created_at <= end_ms,
-        )
-        .first()
-    )
-
-
-def get_stock_for_enrichment():
-
-    db = SessionLocal()
-
-    try:
-        six_hours_ago = datetime.utcnow() - timedelta(hours=6)
-
-        stock = (
-            db.query(StockSources)
-            .filter(
-                or_(
-                    StockSources.last_fetched_at.is_(None),
-                    StockSources.last_fetched_at < six_hours_ago,
-                )
-            )
-            .order_by(StockSources.last_fetched_at.nullsfirst())
-            .first()
-        )
-
-        if stock:
-            db.expunge(stock)
-
-        return stock
-
-    finally:
-        db.close()
-
-
-def update_stock_table(id: str, error: str | None = None):
+    On success clears failure state and stamps last_fetched_at; on error bumps
+    the failure counter. Generic over StockSources / MacroSource.
+    """
 
     db = SessionLocal()
 
     try:
 
-        stock = db.query(StockSources).filter(StockSources.id == id).first()
+        row = db.query(model).filter(model.id == id).first()
 
-        if not stock:
+        if not row:
             return
 
         now_ms = int(time.time() * 1000)
 
-        stock.last_attempted_at = now_ms
+        row.last_attempted_at = now_ms
 
         if error is not None:
-            stock.failure_reason = error
-            stock.failure_count = (stock.failure_count or 0) + 1
+            row.failure_reason = error
+            row.failure_count = (row.failure_count or 0) + 1
         else:
-            stock.last_fetched_at = now_ms
-            stock.failure_reason = None
-            stock.failure_count = 0
+            row.last_fetched_at = now_ms
+            row.failure_reason = None
+            row.failure_count = 0
 
         db.commit()
 
@@ -1061,42 +649,29 @@ def update_stock_table(id: str, error: str | None = None):
         db.close()
 
 
-def get_raw_html():
-    db = SessionLocal()
+# =========================
+# STATS
+# =========================
+
 
+def get_stats():
+    db = SessionLocal()
     try:
-        rows = (
-            db.query(
-                RawHTML,
-                ParsedArticle.title,
-            )
-            .outerjoin(
-                ParsedArticle,
-                ParsedArticle.raw_html_id == RawHTML.id,
-            )
-            .filter(RawHTML.parsed.is_(True))
-            .order_by(RawHTML.created_at.desc())
+        by_type = {
+            t: c
+            for t, c in db.query(RawHTML.type, func.count(RawHTML.id))
+            .group_by(RawHTML.type)
             .all()
-        )
-
-        result = {}
-
-        for raw_html, title in rows:
-
-            if raw_html.id not in result:
-                result[raw_html.id] = {
-                    "id": raw_html.id,
-                    "source": raw_html.source,
-                    "type": raw_html.type.name if raw_html.type else None,
-                    "created_at": raw_html.created_at,
-                    "raw_html": raw_html.html,
-                    "headlines": [],
-                }
-
-            if title:
-                result[raw_html.id]["headlines"].append(title)
-
-        return list(result.values())
-
+        }
+        return {
+            # raw_html waiting to be picked up by the LLM service, total + split
+            # by type (news / stock / macro families).
+            "raw_html_total": sum(by_type.values()),
+            "raw_html_by_type": by_type,
+            # configured source counts
+            "news_sources": db.query(NewsSource).count(),
+            "stock_sources": db.query(StockSources).count(),
+            "macro_sources": db.query(MacroSource).count(),
+        }
     finally:
         db.close()

llm.py

@@ -1,1087 +0,0 @@
-import os
-from threading import Lock
-import json
-import re
-import time
-from transformers import (
-    AutoTokenizer,
-    AutoModelForCausalLM,
-    AutoModelForSequenceClassification,
-)
-import torch
-import numpy as np
-import pandas as pd
-import onnxruntime as ort
-from bs4 import BeautifulSoup
-import spacy
-
-from database import SessionLocal
-from db_helpers import clean_headline_text
-from logger import logger
-from config import (
-    MODEL_PATH,
-    MODEL_REPO,
-    MODEL_SENTIMENT_OWN_MAX_LEN,
-)
-from models import StockSources
-from utils import _normalize_na
-
-# =========================
-# PIPELINE LOCK
-# =========================
-
-# Serializes all LLM inference calls so headline_worker and sentiment_worker
-# can't fight over the same in-process pipeline (transformers' generation
-# state is not re-entrant on a single device). Removing this lock without
-# moving to a request-queue / multi-GPU setup will produce CUDA OOMs or
-# corrupted outputs.
-pipeline_lock = Lock()
-
-
-# =========================
-# MODEL STATE
-# =========================
-
-# Our own sentiment model: an ONNX Runtime session + its tokenizer. Loaded once
-# by load_models(); both stay None if the artifacts are missing (predict falls
-# back to pure-neutral).
-_nlp = None
-_ner = None
-_llm_model = None
-_llm_tokenizer = None
-_own_sentiment_model = None
-_own_sentiment_tokenizer = None
-
-
-def load_models():
-    """Load spaCy + transformers + ONNX sentiment model (idempotent)."""
-
-    global _nlp, _ner, _llm_model, _llm_tokenizer, _own_sentiment_model, _own_sentiment_tokenizer
-
-    # ---------------- SPA CY (base pipeline) ----------------
-    if _nlp is None:
-        logger.info("load_models: loading spaCy base pipeline")
-        _nlp = spacy.load("en_core_web_sm", disable=["ner", "tagger", "lemmatizer"])
-        _nlp.max_length = 2000000
-        logger.info("load_models: spaCy ready")
-    else:
-        logger.info("load_models: spaCy already loaded")
-
-    # ---------------- NER (full model) ----------------
-    if _ner is None:
-        logger.info("load_models: loading spaCy NER model")
-        _ner = spacy.load("en_core_web_sm")
-        logger.info("load_models: NER ready")
-    else:
-        logger.info("load_models: NER already loaded")
-
-    # ---------------- TRANSFORMERS LLM (Qwen) ----------------
-    if _llm_model is None:
-        try:
-            logger.info("load_models: loading Qwen model")
-
-            model_id = MODEL_PATH  # or "Qwen/Qwen2.5-3B"
-
-            _llm_tokenizer = AutoTokenizer.from_pretrained(model_id)
-
-            _llm_model = AutoModelForCausalLM.from_pretrained(
-                model_id,
-                device_map="auto",
-                torch_dtype=(
-                    torch.float16 if torch.cuda.is_available() else torch.float32
-                ),
-                low_cpu_mem_usage=True,
-            )
-
-            logger.info("load_models: transformers LLM ready")
-
-        except Exception:
-            _llm_model = None
-            _llm_tokenizer = None
-            logger.exception("load_models: failed to load transformers model")
-    else:
-        logger.info("load_models: transformers already loaded")
-
-    # ---------------- SENTIMENT (TRANSFORMERS) ----------------
-    if _own_sentiment_model is None:
-        try:
-            logger.info("load_models: loading HF sentiment model")
-
-            _own_sentiment_tokenizer = AutoTokenizer.from_pretrained(MODEL_REPO)
-
-            _own_sentiment_model = AutoModelForSequenceClassification.from_pretrained(
-                MODEL_REPO,
-                torch_dtype=(
-                    torch.float16 if torch.cuda.is_available() else torch.float32
-                ),
-            )
-            _own_sentiment_model.to("cuda" if torch.cuda.is_available() else "cpu")
-            _own_sentiment_model.eval()
-
-            logger.info("load_models: sentiment model ready")
-
-        except Exception:
-            _own_sentiment_model = None
-            _own_sentiment_tokenizer = None
-            logger.exception("load_models: sentiment failed (non-fatal)")
-    else:
-        logger.info("load_models: sentiment already loaded")
-
-    logger.info("load_models: done")
-
-
-def analyze(text: str, filter_type: str = "headlines"):
-    limits = {
-        "headlines": 1000,
-        "sentiment": 50,
-        "sector": 20,
-        "stocks": 1000,
-    }
-
-    max_tokens = limits.get(filter_type, 2000)
-
-    if _llm_model is None or _llm_tokenizer is None:
-        raise RuntimeError("LLM model not loaded. Check load_models() logs.")
-
-    messages = [
-        {
-            "role": "system",
-            "content": "You extract structured data and return JSON when needed.",
-        },
-        {"role": "user", "content": text},
-    ]
-
-    prompt = _llm_tokenizer.apply_chat_template(
-        messages, tokenize=False, add_generation_prompt=True
-    )
-
-    inputs = _llm_tokenizer(prompt, return_tensors="pt")
-
-    # move to model device safely
-    device = next(_llm_model.parameters()).device
-    inputs = {k: v.to(device) for k, v in inputs.items()}
-
-    with torch.no_grad():
-        outputs = _llm_model.generate(
-            **inputs, max_new_tokens=max_tokens, temperature=0, do_sample=False
-        )
-
-    decoded = _llm_tokenizer.decode(
-        outputs[0][inputs["input_ids"].shape[-1] :], skip_special_tokens=True
-    )
-
-    return decoded
-
-
-# =========================
-# OWN SENTIMENT MODEL (ONNX Runtime)
-# =========================
-# Our own model is an ONNX sequence-classification graph labelled
-# {0: negative, 1: neutral, 2: positive}. It is run directly via ONNX Runtime
-# (NOT through Ollama): the headline is tokenized with the model's own tokenizer,
-# fed to the session, and the output logits are softmaxed into the three class
-# probabilities. Any failure (missing artifacts, unexpected output) falls back
-# to pure-neutral so the worker loop stays simple.
-
-SENTIMENT_LABELS = {0: "negative", 1: "neutral", 2: "positive"}
-
-
-def _softmax(logits: np.ndarray) -> np.ndarray:
-    """Numerically-stable softmax over a 1-D logit vector."""
-
-    shifted = logits - np.max(logits)
-    exp = np.exp(shifted)
-    return exp / np.sum(exp)
-
-
-def _sentiment_probs(headline: str) -> dict:
-    """Run HF transformers sentiment model and return probabilities."""
-
-    default = {"negative": 0.0, "neutral": 1.0, "positive": 0.0}
-
-    if not headline:
-        return default
-
-    if _own_sentiment_model is None or _own_sentiment_tokenizer is None:
-        logger.error("sentiment model not loaded")
-        return default
-
-    try:
-        inputs = _own_sentiment_tokenizer(
-            headline,
-            return_tensors="pt",
-            truncation=True,
-            max_length=512,
-        )
-
-        inputs = {k: v.to(_own_sentiment_model.device) for k, v in inputs.items()}
-
-        with torch.no_grad():
-            outputs = _own_sentiment_model(**inputs)
-
-        logits = outputs.logits[0].cpu().numpy()
-
-        probs = _softmax(logits)
-
-        return {
-            "negative": float(probs[0]),
-            "neutral": float(probs[1]),
-            "positive": float(probs[2]),
-        }
-
-    except Exception:
-        logger.exception("sentiment inference failed")
-        return default
-
-
-def predict_own_sentiment(headline: str) -> dict:
-    """Score a headline with our own ONNX model; derive sentiment_score/confidence.
-
-    Returns:
-        {
-          "probs": {"negative": p, "neutral": p, "positive": p},
-          "sentiment_score": P(positive) - P(negative),
-          "confidence": 1 - P(neutral),
-        }
-    """
-
-    preview = (headline or "")[:80]
-    logger.info(f"predict_own_sentiment: headline={preview!r}")
-
-    probs = _sentiment_probs(headline)
-
-    sentiment_score = probs["positive"] - probs["negative"]
-    confidence = 1.0 - probs["neutral"]
-
-    logger.info(
-        f"predict_own_sentiment: probs={probs} "
-        f"sentiment_score={sentiment_score:.4f} confidence={confidence:.4f}"
-    )
-
-    return {
-        "probs": probs,
-        "sentiment_score": sentiment_score,
-        "confidence": confidence,
-    }
-
-
-def extract_json(text: str):
-    text = text.replace("```json", "")
-    text = text.replace("```", "")
-    text = text.strip()
-
-    match = re.search(r"\{.*\}", text, re.DOTALL)
-
-    if not match:
-        logger.error("No JSON found for text")
-        logger.error(text)
-        return ""
-
-    return json.loads(match.group(0))
-
-
-def extract_headlines_from_RAW_html(text: str):
-    text = text.replace("```json", "")
-    text = text.replace("```", "")
-    text = text.strip()
-
-    # find first opening brace
-    start = text.find("{")
-
-    if start == -1:
-        logger.error("No JSON found")
-        logger.error(text)
-        raise ValueError("No JSON found")
-
-    decoder = json.JSONDecoder()
-
-    try:
-        obj, end = decoder.raw_decode(text[start:])
-        return obj
-    except Exception:
-        logger.error("Failed parsing JSON")
-        logger.error(text)
-        raise
-
-
-def looks_like_headline(text: str) -> bool:
-    if not text:
-        return False
-
-    if len(text) < 20:
-        return False
-
-    if len(text) > 250:
-        return False
-
-    if text.count(" ") < 2:
-        return False
-
-    blacklist = [
-        "click here",
-        "read more",
-        "subscribe",
-        "advertisement",
-        "cookie policy",
-        "sign in",
-    ]
-
-    lower = text.lower()
-
-    if any(x in lower for x in blacklist):
-        return False
-
-    return True
-
-
-# =========================
-# Split text into semantic chunks
-# =========================
-
-
-def split_into_chunks(raw_text, chunk_size=3000):
-    logger.info(
-        f"split_into_chunks: input={len(raw_text)} chars, " f"chunk_size={chunk_size}"
-    )
-
-    doc = _nlp(raw_text)
-
-    chunks = []
-    current_chunk = ""
-
-    for sentence in doc.sents:
-        sentence_text = sentence.text.strip()
-        if not sentence_text:
-            continue
-
-        if len(current_chunk) + len(sentence_text) + 1 > chunk_size:
-            if current_chunk:
-                chunks.append(current_chunk.strip())
-            current_chunk = sentence_text
-        else:
-            current_chunk += " " + sentence_text
-
-    if current_chunk:
-        chunks.append(current_chunk.strip())
-
-    logger.info(f"split_into_chunks: produced {len(chunks)} chunk(s)")
-    return chunks
-
-
-# =========================
-# Extract Headlines
-# =========================
-def extract_headlines(raw_html: str):
-    """Extract headlines from RAW html content. It also cleans the headline using regex and filters it."""
-    logger.info(f"extract_headlines: parsing {len(raw_html)} chars of HTML")
-    soup = BeautifulSoup(raw_html, "lxml")
-
-    for tag in soup(["script", "style", "nav", "footer", "header", "noscript", "svg"]):
-        tag.decompose()
-
-    cleaned_text = soup.get_text(" ", strip=True)
-
-    logger.info(f"extract_headlines: cleaned text = {len(cleaned_text)} chars")
-
-    text_chunks = split_into_chunks(cleaned_text, chunk_size=3000)
-
-    all_headlines = []
-
-    for chunk_idx, chunk in enumerate(text_chunks, start=1):
-
-        logger.info(
-            f"extract_headlines: chunk {chunk_idx}/{len(text_chunks)} "
-            f"({len(chunk)} chars) -> LLM"
-        )
-
-        prompt = f"""
-Respond ONLY with valid JSON.
-
-{{
-  "headlines": [
-    "headline 1",
-    "headline 2"
-  ]
-}}
-
-Rules:
-- Return only headlines
-- No explanations
-- No numbering
-- No duplicates
-- Ignore advertisements
-- Ignore navigation text
-- Ignore broken text
-- Ignore any lines which you think is not a valid headline
-
-CONTENT:
-{chunk}
-"""
-
-        try:
-            with pipeline_lock:
-                headlines_text = analyze(prompt, filter_type="headlines").strip()
-
-            try:
-                data = extract_headlines_from_RAW_html(headlines_text)
-
-                headlines = []
-
-                for h in data.get("headlines", []):
-                    h = clean_headline_text(str(h).strip())
-
-                    if not looks_like_headline(h):
-                        continue
-
-                    if len(h) < 20:
-                        continue
-
-                    if "<" in h or ">" in h:
-                        continue
-
-                    headlines.append(h)
-
-            except Exception:
-                logger.exception(
-                    f"extract_headlines: invalid JSON from chunk {chunk_idx}"
-                )
-                headlines = []
-
-            logger.info(
-                f"extract_headlines: chunk {chunk_idx} -> "
-                f"{len(headlines)} headlines"
-            )
-
-            all_headlines.extend(headlines)
-
-        except Exception:
-            logger.exception(f"extract_headlines: chunk {chunk_idx} failed")
-
-    unique_headlines = []
-    seen = set()
-
-    for headline in all_headlines:
-        normalized = headline.lower().strip()
-
-        if normalized not in seen:
-            seen.add(normalized)
-            unique_headlines.append(headline)
-
-    logger.info(
-        f"extract_headlines: done; {len(all_headlines)} total, "
-        f"{len(unique_headlines)} after dedup"
-    )
-
-    return unique_headlines
-
-
-def is_valid_sector(sector: str):
-    """Validates a sectort, returns is_valid:true/false"""
-    logger.info("Validating sector: sector={sector}")
-    prompt = f"""
-You are a industry sector validator in India.
-
-        Determine whether the input is a genuine industry sector.
-
-        A valid industry sector:
-        - Section A: Agriculture, Forestry, and Fishing
-        - Section C: Manufacturing (e.g., Textiles, Chemicals, Automobiles)
-        - Section J: Information and Communication (IT, Software, Telecommunications)
-        - Section K: Financial and Insurance Activities
-
-        An invalid industry sector:
-        - Navigation text
-        - Menu items
-        - Advertisement text
-        - Login/signup text
-        - Cookie notices
-        - Random fragments
-        - Category labels
-        - Gibberish
-
-        Respond ONLY with valid JSON.
-
-        If the industry sector is INVALID:
-
-        {{
-            "is_valid": false
-        }}
-
-        If the industry sector is VALID:
-
-        {{
-            "is_valid": true,
-        }}
-
-        Headline:
-        {sector}
-"""
-    global_data = ""
-    try:
-        with pipeline_lock:
-            logger.info("analyze_sector: calling Ollama")
-            text = analyze(prompt, "sector").strip()
-
-        logger.info(f"analyze_sector: raw output length={len(text)} chars")
-        logger.debug(f"analyze_sector: For sector {sector}")
-        logger.debug(f"raw LLM response: {text}")
-        data = extract_json(text)
-        logger.info(f"Extracted json is {data}")
-        global_data = data
-
-        raw_valid = data.get("is_valid", False)
-
-        if isinstance(raw_valid, bool):
-            is_valid = raw_valid
-        else:
-            is_valid = str(raw_valid).strip().lower() == "true"
-
-        if not is_valid:
-            logger.info(f"analyze_article: invalid sector detected: " f"{sector}")
-
-            return {
-                "is_valid": False,
-            }
-
-        return {
-            "is_valid": True,
-        }
-
-    except Exception:
-        logger.exception("sector: failed to parse model response")
-
-        return {
-            "is_valid": False,
-            "sentiment": None,
-            "category": None,
-        }
-
-
-# =========================
-# Analyze Headline
-# =========================
-
-
-def analyze_article(title: str):
-    preview = (title or "")[:80]
-    logger.info(f"analyze_article: title={preview!r}")
-
-    clean_title = clean_headline_text(title)
-
-    prompt = f"""
-        You are a news headline validator and classifier.
-
-        Determine whether the input is a genuine news headline.
-
-        A valid headline:
-        - Describes a real news event, announcement, development, report, statement, market move, policy change, sports result, technology update, etc.
-        - Looks like something published by a news organization.
-
-        An invalid headline:
-        - Navigation text
-        - Menu items
-        - Advertisement text
-        - Login/signup text
-        - Cookie notices
-        - Random fragments
-        - Broken HTML text
-        - Single words
-        - Category labels
-        - Gibberish
-
-        Respond ONLY with valid JSON.
-
-        If the headline is INVALID:
-
-        {{
-            "is_valid": false
-        }}
-
-        If the headline is VALID:
-
-        {{
-            "is_valid": true,
-            "sentiment": "positive|negative|neutral",
-            "category": "business|politics|technology|sports|world|other"
-        }}
-
-        Headline:
-        {clean_title}
-    """
-    global_data = ""
-    try:
-        with pipeline_lock:
-            logger.info("analyze_article: calling Ollama")
-            text = analyze(prompt, "sentiment").strip()
-
-        logger.info(f"analyze_article: raw output length={len(text)} chars")
-        logger.debug(f"analyze_article: For headline {clean_title}")
-        logger.debug(f"raw LLM response: {text}")
-        data = extract_json(text)
-        global_data = data
-
-        raw_valid = data.get("is_valid", False)
-
-        if isinstance(raw_valid, bool):
-            is_valid = raw_valid
-        else:
-            is_valid = str(raw_valid).strip().lower() == "true"
-
-        if not is_valid:
-            logger.info(
-                f"analyze_article: invalid headline detected: " f"{clean_title[:100]}"
-            )
-
-            return {
-                "is_valid": False,
-                "sentiment": None,
-                "category": None,
-            }
-
-        sentiment = str(data.get("sentiment", "neutral")).strip().lower()
-
-        category = str(data.get("category", "other")).strip().lower()
-
-        valid_sentiments = {
-            "positive",
-            "negative",
-            "neutral",
-        }
-
-        valid_categories = {
-            "business",
-            "politics",
-            "technology",
-            "sports",
-            "world",
-            "other",
-        }
-
-        if sentiment not in valid_sentiments:
-            sentiment = "neutral"
-
-        if category not in valid_categories:
-            category = "other"
-
-        logger.info(
-            f"analyze_article: valid headline "
-            f"sentiment={sentiment} "
-            f"category={category}"
-        )
-
-        return {
-            "is_valid": True,
-            "sentiment": sentiment,
-            "category": category,
-        }
-
-    except Exception:
-        logger.exception("analyze_article: failed to parse model response")
-
-        logger.exception(
-            f"{clean_title[:100]}",
-        )
-
-        logger.exception(
-            f"{global_data}",
-        )
-
-        return {
-            "is_valid": False,
-            "sentiment": None,
-            "category": None,
-        }
-
-
-# =========================
-# ML DATASET PREPROCESSING
-# =========================
-
-
-def clean_for_dataset(text: str) -> str:
-    """Local preprocessing for the ML dataset.
-
-    Builds on clean_headline_text (bullets / numbering / quotes) and additionally
-    strips dashes so the headline is normalized before being sent to the model.
-    """
-
-    if not text:
-        return ""
-
-    # Shared cleaning: bullets, numbered/lettered lists, surrounding quotes.
-    text = clean_headline_text(text)
-
-    # Remove dashes (hyphen, en dash, em dash) -> space.
-    text = re.sub(r"[-–—]+", " ", text)
-
-    # Collapse any whitespace introduced above.
-    text = re.sub(r"\s+", " ", text).strip()
-
-    return text
-
-
-def extract_entities(text):
-    doc = _ner(text)
-    # Return a list of tuples containing (Entity Text, Entity Type)
-    return [(ent.text, ent.label_) for ent in doc.ents]
-
-
-def extract_company_sector(headline: str) -> dict:
-    """
-    Resolve a company mentioned in a headline using:
-    1. Direct company name match
-    2. Direct ticker match
-    3. LLM fallback (ticker only)
-    4. DB as single source of truth for company + sector
-    """
-
-    preview = (headline or "")[:80]
-    logger.info(f"extract_company_sector: headline={preview!r}")
-
-    default = {
-        "is_valid": False,
-        "ticker": "NA",
-        "company": "NA",
-        "sector": "NA",
-    }
-
-    if not headline:
-        return default
-
-    headline_lower = headline.lower()
-
-    db = SessionLocal()
-
-    try:
-
-        sources = db.query(
-            StockSources.ticker,
-            StockSources.company_name,
-            StockSources.sector,
-        ).all()
-
-        entities = extract_entities(headline_lower)
-
-        entity_texts = {e[0].lower() for e in entities}
-
-        #
-        # STEP 1: Company name match
-        #
-        for source in sources:
-
-            company_name = (source.company_name or "").strip()
-            if not company_name:
-                continue
-
-            if company_name.lower() in headline_lower:
-
-                logger.info(f"Matched company name: {company_name}")
-
-                return {
-                    "is_valid": True,
-                    "ticker": source.ticker,
-                    "company": source.company_name,  # keep original casing
-                    "sector": source.sector,
-                }
-
-        #
-        # STEP 2: Ticker match
-        #
-        for source in sources:
-            ticker = (source.ticker or "").strip()
-
-            if source.company_name and source.company_name.lower() in entity_texts:
-                return {
-                    "is_valid": True,
-                    "ticker": source.ticker,
-                    "company": source.company_name,
-                    "sector": source.sector,
-                }
-
-            if not ticker:
-                continue
-
-            if ticker.lower() in headline_lower:
-
-                logger.info(f"Matched ticker: {ticker}")
-
-                return {
-                    "is_valid": True,
-                    "ticker": source.ticker,
-                    "company": source.company_name,
-                    "sector": source.sector,
-                }
-
-        #
-        # STEP 3: LLM fallback (ticker only)
-        #
-        logger.info("No direct StockSources match found. Falling back to LLM.")
-
-        company_list = [
-            {
-                "ticker": s.ticker,
-                "company": s.company_name,
-            }
-            for s in sources
-            if s.company_name
-        ]
-
-        prompt = f"""
-You are a financial news classifier. Identify the most relevant stock ticker/company name from the headline.
-
-Return ONLY JSON:
-
-{{
-    "ticker": "ticker",
-    "company_name": "company name",
-    "is_valid": "True|False"
-}}
-
-Headline:
-{headline}
-"""
-
-        with pipeline_lock:
-            text = analyze(prompt, "headlines").strip()
-
-        valid_tickers = {s.ticker.upper() for s in sources if s.ticker}
-
-        valid_companies = {s.company_name.upper() for s in sources if s.company_name}
-
-        logger.debug(f"raw LLM response: {text}")
-
-        data = extract_json(text)
-
-        logger.debug(f"extracted response: {data}")
-
-        if not isinstance(data, dict):
-            return default
-
-        is_valid = None
-
-        if "is_valid" in data:
-            is_valid = (data.get("is_valid") or "").strip().upper()
-
-        if not is_valid:
-            return default
-
-        ticker = None
-        company = None
-
-        # CASE 1: normal format
-        if "ticker" in data:
-            t = (data.get("ticker") or "").strip().upper()
-            if t in valid_tickers:
-                ticker = t
-
-        # CASE 2: weird key-value format {"TICKER": "Company Name"}
-        else:
-            for k, v in data.items():
-                if isinstance(k, str) and k.strip():
-                    # check if it exists in database
-                    if k.upper() in valid_tickers:
-                        ticker = k
-                        break
-                    if k.upper() in valid_companies:
-                        company = k.upper()
-                        break
-
-        ticker = (ticker or "").strip().upper()
-        company = (company or "").strip().lower()
-        source = None
-
-        if (not ticker or ticker == "NA") and (not company or company == "NA"):
-            return default
-
-        if ticker in valid_tickers:
-            source = (
-                db.query(StockSources).filter(StockSources.ticker.ilike(ticker)).first()
-            )
-
-        if company in valid_companies:
-            source = (
-                db.query(StockSources)
-                .filter(StockSources.company_name.ilike(company))
-                .first()
-            )
-
-        if not source:
-            return default
-
-        return {
-            "is_valid": True,
-            "ticker": source.ticker,
-            "company": source.company_name,
-            "sector": source.sector or "NA",
-        }
-
-    except Exception:
-        logger.exception("extract_company_sector: failed")
-        return default
-
-    finally:
-        db.close()
-
-
-# =========================
-# Extract Stock data
-# =========================
-
-
-def metrics_score(metrics):
-
-    score = 0
-
-    for field in (
-        "Open",
-        "High",
-        "Low",
-        "Close",
-        "volume",
-    ):
-
-        value = str(metrics.get(field, "")).strip()
-
-        if value and value != "—":
-            score += 1
-
-    return score
-
-
-import json
-
-
-def extract_stock_metrics_from_html(raw_html: str):
-    """
-    Uses the LLM to extract all key/value metrics from
-    Google Finance style stock information HTML.
-
-    Returns:
-    {
-        "Open": "...",
-        "High": "...",
-        ...
-    }
-    """
-    soup = BeautifulSoup(raw_html, "lxml")
-
-    for tag in soup(["script", "style", "nav", "footer", "header", "noscript", "svg"]):
-        tag.decompose()
-
-    cleaned_text = soup.get_text(" ", strip=True)
-
-    logger.info(f"extract_headlines: cleaned text = {len(cleaned_text)} chars")
-
-    chunks = split_into_chunks(cleaned_text)
-
-    merged_metrics = {}
-
-    logger.info(
-        f"extract_stock_metrics_from_html: " f"Split HTML into {len(chunks)} chunks"
-    )
-
-    best_score = 0
-
-    for chunk_idx, chunk in enumerate(chunks, start=1):
-
-        prompt = f"""
-Respond ONLY with valid JSON.
-
-Desired Format:
-
-{{
-  "metrics": {{
-    "company_name": "company name"
-    "ticker": ticker,
-    "exchange": exchange,
-    "Open": open_price,
-    "High": high_price,
-    "Low": low_price,
-    "Close": close_price,
-    "volume": volume,
-  }}
-}}
- 
-Rules:
-- Parse the HTML.
-- Extract every metric label and its value.
-- Preserve metric names exactly.
-- Remove HTML tags.
-- Ignore styling and javascript attributes.
-- Ignore duplicate metric/value pairs.
-- Return only JSON.
-- No explanations.
-- No markdown.
-
-HTML:
-
-{chunk}
-"""
-
-        try:
-
-            logger.info(
-                f"extract_stock_metrics_from_html: "
-                f"Processing chunk "
-                f"{chunk_idx}/{len(chunks)}"
-            )
-
-            with pipeline_lock:
-                llm_response = analyze(prompt, filter_type="stocks").strip()
-
-            parsed = json.loads(llm_response)
-
-            logger.debug(f"LLM JSON response:\n{llm_response}")
-
-            metrics = parsed.get("metrics", {})
-
-            score = metrics_score(metrics)
-
-            if score > best_score:
-                best_score = score
-                # merge immediately
-                for k, v in metrics.items():
-                    k = str(k).strip()
-                    v = str(v).strip() if v is not None else ""
-
-                    if not k or not v:
-                        continue
-
-                    if v.lower() in {
-                        "company name",
-                        "ticker",
-                        "ticker symbol",
-                        "exchange",
-                        "null",
-                        "none",
-                    }:
-                        continue
-
-                    if v.lower() in {"", "—", "na", "n/a"}:
-                        continue
-
-                    if k not in merged_metrics or merged_metrics[k] in {
-                        "",
-                        "—",
-                        "null",
-                        "none",
-                    }:
-                        merged_metrics[k] = v
-
-            if score >= 5:
-                logger.info("Perfect stock quote found, " "stopping chunk processing")
-                break
-
-        except Exception:
-            logger.exception(
-                f"extract_stock_metrics_from_html: " f"chunk {chunk_idx} failed"
-            )
-
-    logger.info(
-        f"extract_stock_metrics_from_html: " f"Returning {len(merged_metrics)} metrics"
-    )
-    logger.debug(
-        f"extract_stock_metrics_from_html: " f"Returning {(merged_metrics)} metrics"
-    )
-
-    return merged_metrics

logger.py

@@ -8,8 +8,7 @@ class Logger:
 
     def __init__(self):
 
-        self.logs_folder = os.path.join("/data", "logs")
-        os.makedirs(self.logs_folder, exist_ok=True)
+        self.logs_folder = "logs"
 
         self.max_entries = 10000
 
@@ -41,13 +40,24 @@ class Logger:
             except Exception:
                 pass
 
-        timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
+        timestamp = datetime.now().strftime(
+            "%Y-%m-%d_%H-%M-%S"
+        )
 
-        filename = f"backend_logs_{timestamp}_{self.file_index}.txt"
+        filename = (
+            f"backend_logs_{timestamp}_{self.file_index}.txt"
+        )
 
-        self.current_file_path = os.path.join(self.logs_folder, filename)
+        self.current_file_path = os.path.join(
+            self.logs_folder,
+            filename
+        )
 
-        self.current_file = open(self.current_file_path, "a", encoding="utf-8")
+        self.current_file = open(
+            self.current_file_path,
+            "a",
+            encoding="utf-8"
+        )
 
         self.current_entries = 0
 
@@ -65,9 +75,15 @@ class Logger:
 
                 self._create_new_log_file()
 
-            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+            timestamp = datetime.now().strftime(
+                "%Y-%m-%d %H:%M:%S"
+            )
 
-            formatted = f"[{timestamp}] " f"[{level.upper()}] " f"{message}"
+            formatted = (
+                f"[{timestamp}] "
+                f"[{level.upper()}] "
+                f"{message}"
+            )
 
             # Print to terminal
             print(formatted)

macro_sources.csv

@@ -0,0 +1,17 @@
+name,type,country,market,exchange,gf_ticker,url
+Nifty 50,index,INDIA,INDIA,INDEXNSE,NIFTY_50,
+BSE Sensex,index,INDIA,INDIA,INDEXBOM,SENSEX,
+S&P 500,index,UNITED STATES,US,INDEXSP,.INX,
+Nasdaq Composite,index,UNITED STATES,US,INDEXNASDAQ,.IXIC,
+Dow Jones,index,UNITED STATES,US,INDEXDJX,.DJI,
+India VIX,vix,INDIA,INDIA,INDEXNSE,INDIAVIX,
+CBOE VIX,vix,UNITED STATES,US,INDEXCBOE,VIX,
+Gold (GLD),commodity,GLOBAL,US,NYSEARCA,GLD,
+Brent Oil (BNO),commodity,GLOBAL,US,NYSEARCA,BNO,
+WTI Oil (USO),commodity,GLOBAL,US,NYSEARCA,USO,
+Silver (SLV),commodity,GLOBAL,US,NYSEARCA,SLV,
+Copper (CPER),commodity,GLOBAL,US,NYSEARCA,CPER,
+USD/INR,fx,INDIA,INDIA,,,https://www.google.com/finance/quote/USD-INR
+India 10Y Bond,yield,INDIA,INDIA,,,https://www.worldgovernmentbonds.com/bond-historical-data/india/10-years/
+US 10Y Bond,yield,UNITED STATES,US,,,https://www.worldgovernmentbonds.com/bond-historical-data/united-states/10-years/
+FII/DII Activity,fii_dii,INDIA,INDIA,,,https://www.moneycontrol.com/stocks/marketstats/fii_dii_activity/index.php

main.py

@@ -1,16 +1,18 @@
 from contextlib import asynccontextmanager
 
-from fastapi import Depends, FastAPI
-import os
+from fastapi import FastAPI
 import subprocess
 from logger import logger
 from database import engine
 from models import Base
-from db_helpers import load_stocks_to_db, migrate_schema, load_sources_to_db
-from llm import load_models
-from utils import create_ollama_modelfile
+from db_helpers import (
+    load_stocks_to_db,
+    migrate_schema,
+    load_sources_to_db,
+    load_macro_sources_to_db,
+)
 from workers import stop_workers
-from actions import ActionRequest, verify_auth, ACTIONS
+from actions import ActionRequest, ACTIONS
 
 # =========================
 # LIFESPAN
@@ -26,14 +28,13 @@ async def lifespan(app: FastAPI):
     Base.metadata.create_all(bind=engine)
     logger.info("lifespan: schema create_all done")
 
-    create_ollama_modelfile()
     migrate_schema()
-    load_models()
     load_sources_to_db()
     load_stocks_to_db()
+    load_macro_sources_to_db()
+
     logger.info("Checking/Installing Playwright Chromium browser...")
     try:
-        # Executes 'playwright install chromium' programmatically
         subprocess.run(["playwright", "install", "chromium"], check=True)
         logger.info("Playwright browser check complete.")
     except Exception as e:

models.py

@@ -1,33 +1,46 @@
 # =========================
-# models.py
+# models.py (scrapper)
 # =========================
+# The scrapper only needs the tables required to (a) know what to fetch and
+# (b) hand raw HTML off to the LLM service. All parsing / sentiment / sector /
+# stock-metric tables live in the LLM service.
 
 from sqlalchemy import (
     Column,
     String,
     Text,
     Integer,
-    Float,
-    Boolean,
     text as text_func,
-    Enum,
 )
 
 import time
 import uuid
-import enum
 from database import Base
 
 
-# =========================
-# RAW HTML TABLE
-# =========================
-class SourceType(str, enum.Enum):
-    NEWS = "news"
-    STOCK = "stock"
+# raw_html.type is stored as one of these plain lowercase strings.
+NEWS = "news"
+STOCK = "stock"
+
+# Macro/market series types. Each MacroSource row carries one of these as its
+# raw_html.type, so the LLM side can route it to the right processor.
+FII_DII = "fii_dii"
+VIX = "vix"
+COMMODITY = "commodity"
+INDEX = "index"
+FX = "fx"
+YIELD = "yield"
+MACRO_TYPES = (FII_DII, VIX, COMMODITY, INDEX, FX, YIELD)
 
 
 class RawHTML(Base):
+    """A single fetched page. Stored verbatim (true raw HTML) and served to the
+    LLM service via GET_RAW_HTML, then hard-deleted via DELETE_RAW_HTML once the
+    consumer has stored its own copy.
+
+    `type` is a plain lowercase string ("news" / "stock") to keep value handling
+    unambiguous across the HTTP boundary.
+    """
 
     __tablename__ = "raw_html"
 
@@ -39,44 +52,11 @@ class RawHTML(Base):
 
     created_at = Column(Integer, default=lambda: int(time.time() * 1000))
 
-    parsed = Column(Boolean, nullable=False, default=False)
-
-    type = Column(Enum(SourceType), nullable=False)
-
-
-# =========================
-# PARSED ARTICLES TABLE
-# =========================
-
-
-class ParsedArticle(Base):
-
-    __tablename__ = "parsed_articles"
-
-    id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
-
-    raw_html_id = Column(String)
-
-    title = Column(Text)
-
-    sentiment = Column(String)
-
-    category = Column(String)
-
-    analyzed = Column(Boolean, default=False)
-
-    # set by sentiment worker, is set to true after sentiment score is done
-    is_cleaned = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
-    )
-    # set by analyzed headlines, is set to true when headline is not valid
-    is_deleted = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
-    )
-
-    deleted_at = Column(Integer, nullable=True)
+    type = Column(String, nullable=False)
 
-    created_at = Column(Integer, default=lambda: int(time.time() * 1000))
+    # Market country for stock pages (e.g. "INDIA", "UNITED STATES"), derived
+    # from the StockSources row at fetch time. NULL for news pages.
+    country = Column(String, nullable=True)
 
 
 # =========================
@@ -92,6 +72,7 @@ class NewsSource(Base):
 
     name = Column(String, nullable=False, unique=True)
 
+    # `website` is the URL the scraper fetches for this news source.
     website = Column(String, nullable=False)
 
     created_at = Column(Integer, default=lambda: int(time.time() * 1000))
@@ -110,76 +91,10 @@ class NewsSource(Base):
 
 
 # =========================
-# SENTIMENT TABLE
+# STOCK SOURCES TABLE
 # =========================
 
 
-class Sentiment(Base):
-
-    __tablename__ = "sentiment"
-
-    id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
-
-    company = Column(String)
-
-    # P(positive) - P(negative), range -1..1
-    sentiment_score = Column(Float)
-
-    # 1 - P(neutral)
-    confidence = Column(Float)
-
-    sector = Column(String)
-
-    has_been_read_company = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
-    )
-
-    has_been_read_sector = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
-    )
-
-    created_at = Column(Integer, default=lambda: int(time.time() * 1000))
-
-    headline_id = Column(String)
-
-
-# =========================
-# SECTOR ROLLUP TABLE
-# =========================
-# Derived sector-level summary, refreshed once per 24h by sector_worker from the
-# sentiment table. One row per sector (upserted in place): metrics are
-# confidence-weighted averages, sum(score*confidence)/sum(confidence).
-
-
-class Sector(Base):
-
-    __tablename__ = "sector"
-
-    id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
-
-    sector = Column(String, nullable=False, unique=True, index=True)
-
-    # Weighted sentiment over the trailing 24h ("today").
-    sector_sentiment = Column(Float)
-
-    # Weighted sentiment over the trailing 7 days (inclusive).
-    sector_sentiment_7d_avg = Column(Float)
-
-    # sector_sentiment - sector_sentiment_7d_avg
-    sector_momentum = Column(Float)
-
-    # Forward-looking flag for a downstream consumer; reset to False on each
-    # recompute since the row's numbers change.
-    is_analyzed = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
-    )
-
-    created_at = Column(Integer, default=lambda: int(time.time() * 1000))
-
-    # Stamped each time the row is recomputed/upserted.
-    updated_at = Column(Integer, nullable=True)
-
-
 class StockSources(Base):
 
     __tablename__ = "stock_sources"
@@ -193,6 +108,10 @@ class StockSources(Base):
     exchange = Column(String, nullable=False)
     sector = Column(String, nullable=False)
 
+    # Pre-built Google Finance URL, computed once when the CSV is loaded so the
+    # stock_enhancer worker just fetches it instead of constructing it per tick.
+    url = Column(String, nullable=True)
+
     last_fetched_at = Column(Integer, nullable=True)
     last_attempted_at = Column(Integer, nullable=True)
     failure_count = Column(
@@ -201,22 +120,36 @@ class StockSources(Base):
     failure_reason = Column(Text, nullable=True)
 
 
-class Stock(Base):
+# =========================
+# MACRO SOURCES TABLE
+# =========================
+# Config-driven market/macro series (FII/DII, VIX, commodities, indices, FX,
+# bond yields). Mirrors StockSources' fetch/failure shape but carries a `type`
+# (one of MACRO_TYPES) and a `market` (key into config.MARKET_CONFIG) instead of
+# ticker/sector semantics. The same generic fetch-after-close worker drives it.
+
 
-    __tablename__ = "stock"
+class MacroSource(Base):
+
+    __tablename__ = "macro_sources"
 
     id = Column(String, primary_key=True, index=True, default=lambda: str(uuid.uuid4()))
     created_at = Column(Integer, default=lambda: int(time.time() * 1000))
     updated_at = Column(Integer, nullable=True)
-    raw_html_id = Column(String, nullable=False)
-    company_name = Column(String, nullable=False)
-    ticker = Column(String, nullable=False)
-    close = Column(Float)
-    volume = Column(Float)
-    country = Column(String)
-    open = Column(Float)
-    high = Column(Float)
-    low = Column(Float)
-    is_complete = Column(
-        Boolean, nullable=False, default=False, server_default=text_func("0")
+
+    name = Column(String, nullable=False)
+    type = Column(String, nullable=False)  # one of MACRO_TYPES
+    country = Column(String, nullable=True)
+    market = Column(String, nullable=False)  # key into MARKET_CONFIG
+    exchange = Column(String, nullable=True)
+
+    # Pre-built fetch URL (Google Finance quote URL or a raw page URL), computed
+    # once when the CSV is loaded.
+    url = Column(String, nullable=True)
+
+    last_fetched_at = Column(Integer, nullable=True)
+    last_attempted_at = Column(Integer, nullable=True)
+    failure_count = Column(
+        Integer, nullable=False, default=0, server_default=text_func("0")
     )
+    failure_reason = Column(Text, nullable=True)

requirements.txt

@@ -1,88 +1,32 @@
 # =========================
-# CORE API
+# Scrapper service deps
 # =========================
+# Fetches pages and stores raw HTML. No ML / LLM stack here.
+
 fastapi==0.136.1
 uvicorn==0.47.0
 starlette==1.0.0
+pydantic==2.13.4
+pydantic_core==2.46.4
 python-dotenv==1.2.2
-SQLAlchemy
 
+# DB
+SQLAlchemy==2.0.49
+greenlet==3.5.0
 
-# =========================
-# HTTP / SCRAPING
-# =========================
-requests==2.34.2
-httpx>=0.26.0,<0.27.0
+# Fetching
+curl_cffi==0.15.0
+playwright
+certifi==2026.4.22
+charset-normalizer==3.4.7
+idna==3.15
 urllib3==2.7.0
+
+# HTML / CSV
 beautifulsoup4==4.14.3
+soupsieve==2.8.3
 lxml==6.1.0
-lxml_html_clean==0.4.4
-cssselect==1.4.0
-curl_cffi==0.15.0
-feedparser==6.0.12
-feedfinder2==0.0.4
-tldextract==5.3.1
-newspaper3k==0.2.8
-
-# =========================
-# NLP CORE (STABLE STACK)
-# =========================
-spacy==3.8.14
-thinc==8.3.13
-blis==1.3.3
-preshed==3.0.13
-catalogue==2.0.10
-confection==1.3.3
-srsly==2.5.3
-wasabi==1.1.3
-
-nltk==3.9.4
-regex==2026.5.9
-jieba3k==0.35.1
-tinysegmenter==0.3
-
-# spaCy model
-en_core_web_sm @ https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl
-
-# =========================
-# ML / AI STACK (SAFE FOR PY310)
-# =========================
-numpy==1.26.4
-networkx>=3.3,<3.6
-sympy==1.14.0
-accelerate>=0.26.0 
-torch>=2.3.1
-transformers==4.57.6
-tokenizers>=0.21
-safetensors>=0.4
-huggingface_hub>=0.30
-onnxruntime==1.17.3
-
-# =========================
-# DATA / UTILS
-# =========================
 pandas
-scikit-learn
-joblib==1.5.3
-diskcache==5.6.3
-python-dateutil==2.9.0.post0
-tqdm==4.67.3
-rich==15.0.0
 
-# =========================
-# WEB / SERVER UTILITIES
-# =========================
-anyio==4.13.0
-h11==0.16.0
-httptools==0.7.1
-watchfiles==1.2.0
-websockets==16.0
-typing_extensions==4.15.0
-annotated-types==0.7.0
-pydantic==2.13.4
-pydantic_core==2.46.4
-
-# =========================
-# PLAYWRIGHT
-# =========================
-playwright
+# perf (non-windows)
+uvloop==0.22.1 ; sys_platform != "win32"

us_stocks_list.csv

@@ -0,0 +1,41 @@
+Company Name,Industry,Symbol,Series,ISIN Code,Country,Exchange
+Apple Inc.,Technology,AAPL,,,United States,NASDAQ
+Microsoft Corporation,Technology,MSFT,,,United States,NASDAQ
+Alphabet Inc.,Communication Services,GOOGL,,,United States,NASDAQ
+Amazon.com Inc.,Consumer Discretionary,AMZN,,,United States,NASDAQ
+Meta Platforms Inc.,Communication Services,META,,,United States,NASDAQ
+NVIDIA Corporation,Technology,NVDA,,,United States,NASDAQ
+Tesla Inc.,Consumer Discretionary,TSLA,,,United States,NASDAQ
+Broadcom Inc.,Technology,AVGO,,,United States,NASDAQ
+PepsiCo Inc.,Consumer Staples,PEP,,,United States,NASDAQ
+Costco Wholesale Corporation,Consumer Staples,COST,,,United States,NASDAQ
+Adobe Inc.,Technology,ADBE,,,United States,NASDAQ
+Netflix Inc.,Communication Services,NFLX,,,United States,NASDAQ
+Advanced Micro Devices Inc.,Technology,AMD,,,United States,NASDAQ
+Intel Corporation,Technology,INTC,,,United States,NASDAQ
+Comcast Corporation,Communication Services,CMCSA,,,United States,NASDAQ
+QUALCOMM Incorporated,Technology,QCOM,,,United States,NASDAQ
+Texas Instruments Incorporated,Technology,TXN,,,United States,NASDAQ
+Amgen Inc.,Healthcare,AMGN,,,United States,NASDAQ
+Intuit Inc.,Technology,INTU,,,United States,NASDAQ
+Starbucks Corporation,Consumer Discretionary,SBUX,,,United States,NASDAQ
+JPMorgan Chase & Co.,Financial Services,JPM,,,United States,NYSE
+Visa Inc.,Financial Services,V,,,United States,NYSE
+Johnson & Johnson,Healthcare,JNJ,,,United States,NYSE
+Walmart Inc.,Consumer Staples,WMT,,,United States,NYSE
+Procter & Gamble Company,Consumer Staples,PG,,,United States,NYSE
+Mastercard Incorporated,Financial Services,MA,,,United States,NYSE
+The Home Depot Inc.,Consumer Discretionary,HD,,,United States,NYSE
+Bank of America Corporation,Financial Services,BAC,,,United States,NYSE
+Exxon Mobil Corporation,Energy,XOM,,,United States,NYSE
+Chevron Corporation,Energy,CVX,,,United States,NYSE
+The Coca-Cola Company,Consumer Staples,KO,,,United States,NYSE
+Pfizer Inc.,Healthcare,PFE,,,United States,NYSE
+The Walt Disney Company,Communication Services,DIS,,,United States,NYSE
+Merck & Co. Inc.,Healthcare,MRK,,,United States,NYSE
+AbbVie Inc.,Healthcare,ABBV,,,United States,NYSE
+Salesforce Inc.,Technology,CRM,,,United States,NYSE
+NIKE Inc.,Consumer Discretionary,NKE,,,United States,NYSE
+McDonald's Corporation,Consumer Discretionary,MCD,,,United States,NYSE
+Oracle Corporation,Technology,ORCL,,,United States,NYSE
+UnitedHealth Group Incorporated,Healthcare,UNH,,,United States,NYSE

utils.py

@@ -4,7 +4,8 @@ from datetime import datetime, timedelta
 from zoneinfo import ZoneInfo
 
 IST = ZoneInfo("Asia/Kolkata")
-from config import HTTP_REQUEST_TIMEOUT_SECONDS, MODEL_PATH
+ET = ZoneInfo("America/New_York")
+from config import HTTP_REQUEST_TIMEOUT_SECONDS, MARKET_CONFIG
 from logger import logger
 
 
@@ -116,20 +117,6 @@ def fetch_html(url: str):
             pass
 
 
-def _normalize_na(value) -> str:
-    """Coerce empty / placeholder model output to the literal "NA"."""
-
-    if value is None:
-        return "NA"
-
-    s = str(value).strip()
-
-    if not s or s.lower() in {"na", "n/a", "none", "null", "unknown", "-", ""}:
-        return "NA"
-
-    return s
-
-
 def seconds_until_next_monday():
     now = datetime.now(IST)
 
@@ -146,7 +133,64 @@ def seconds_until_next_monday():
     return max(1, int((next_monday - now).total_seconds()))
 
 
-def create_ollama_modelfile():
-    path = "/home/user/app/ollama/Modelfile-q4"
-    with open(path, "w") as f:
-        f.write(f"FROM {MODEL_PATH}\n")
+# =========================
+# MARKET-CLOSE SCHEDULING
+# =========================
+# Generalizes seconds_until_next_monday() to "next weekday close for a market".
+# The stock/macro workers wake at the soonest close across markets, fetch the
+# rows for whichever market closed today, then sleep again to the next close.
+
+
+def _market_tz(market: str) -> ZoneInfo:
+    """ZoneInfo for a market key in MARKET_CONFIG (defaults to IST)."""
+
+    cfg = MARKET_CONFIG.get(market)
+    return ZoneInfo(cfg["tz"]) if cfg else IST
+
+
+def _market_close_today(market: str, now: datetime) -> datetime:
+    """Today's close datetime (tz-aware) for `market`, on the date of `now`."""
+
+    hour, minute = MARKET_CONFIG[market]["close"]
+    return now.replace(hour=hour, minute=minute, second=0, microsecond=0)
+
+
+def market_already_closed_today(market: str) -> bool:
+    """True if `market` is on a weekday and its close time has already passed."""
+
+    now = datetime.now(_market_tz(market))
+
+    # Saturday=5, Sunday=6 — markets shut.
+    if now.weekday() in (5, 6):
+        return False
+
+    return now >= _market_close_today(market, now)
+
+
+def seconds_until_next_market_close(market: str) -> int:
+    """Seconds until `market`'s next close, skipping weekends.
+
+    If today is a weekday and the close hasn't happened yet, that's the target.
+    Otherwise advance day-by-day to the next weekday's close.
+    """
+
+    now = datetime.now(_market_tz(market))
+    close = _market_close_today(market, now)
+
+    # Move to the next valid close: future + on a weekday.
+    while close <= now or close.weekday() in (5, 6):
+        close = close + timedelta(days=1)
+
+    return max(1, int((close - now).total_seconds()))
+
+
+def get_market_day_start_ms(market: str) -> int:
+    """Epoch-ms for the start (00:00) of the current day in `market`'s tz.
+
+    Used to gate "already fetched today": a row whose last_fetched_at is >= this
+    value has been fetched during the current market day and is skipped.
+    """
+
+    now = datetime.now(_market_tz(market))
+    start = now.replace(hour=0, minute=0, second=0, microsecond=0)
+    return int(start.timestamp() * 1000)

workers.py

@@ -1,56 +1,31 @@
 import os
 import signal
 import time
-import re
 from threading import Thread, Lock, Event
-from datetime import datetime, timedelta
-from zoneinfo import ZoneInfo
 
-IST = ZoneInfo("Asia/Kolkata")
 from logger import logger
-from models import Stock
+from models import StockSources, MacroSource, NEWS, STOCK
 from db_helpers import (
-    get_stock_for_enrichment,
-    mark_clean_headlines_past_7d,
     read_sources,
     update_last_fetched,
     record_failure,
     store_raw_html,
-    get_unparsed_record,
-    mark_as_parsed,
-    store_headlines,
-    get_unanalyzed_article,
-    update_article_analysis,
-    soft_delete_invalid_headlines,
-    get_analyzed_article,
-    mark_article_cleaned,
-    store_sentiment,
-    update_sector_table,
-    mark_clean_headlines_past_7d,
-    update_stock_data,
-    update_stock_table,
-)
-from llm import (
-    extract_headlines,
-    analyze_article,
-    extract_company_sector,
-    extract_stock_metrics_from_html,
-    predict_own_sentiment,
+    get_sources_due_for_market,
+    update_source_table,
 )
 from config import (
     SCRAPER_TICK_SECONDS,
     SCRAPE_INTERVAL_MS,
     MAX_CONSECUTIVE_FAILURES,
-    HEADLINE_IDLE_SECONDS,
-    SENTIMENT_IDLE_SECONDS,
-    SENTIMENT_TABLE_IDLE_SECONDS,
     WORKER_ERROR_BACKOFF_SECONDS,
-    HTTP_REQUEST_TIMEOUT_SECONDS,
-    SECTOR_TICK_SECONDS,
+    MARKET_CONFIG,
+)
+from utils import (
+    fetch_html,
+    seconds_until_next_market_close,
+    market_already_closed_today,
 )
-from utils import fetch_html, seconds_until_next_monday
 
-last_nse_top_fetched = 0
 
 # =========================
 # WORKER LIFECYCLE STATE
@@ -62,65 +37,6 @@ workers_lock = Lock()
 _workers: dict[str, dict] = {}
 
 
-# =========================
-# HELPER - FINANCIAL SPECIFIC
-# =========================
-
-
-def parse_market_number(value):
-
-    if value is None:
-        return None
-
-    value = str(value).strip()
-
-    if not value:
-        return None
-
-    if value in {"—", "-", "N/A", "NA", "null", "None"}:
-        return None
-
-    value = value.replace(",", "")
-    value = value.replace("₹", "")
-    value = value.replace("$", "")
-
-    match = re.match(
-        r"^([-+]?\d*\.?\d+)\s*([a-zA-Z]*)$",
-        value,
-    )
-
-    if not match:
-        return None
-
-    number = float(match.group(1))
-
-    suffix = match.group(2).lower()
-
-    multipliers = {
-        "k": 1_000,
-        "m": 1_000_000,
-        "mn": 1_000_000,
-        "million": 1_000_000,
-        "b": 1_000_000_000,
-        "bn": 1_000_000_000,
-        "billion": 1_000_000_000,
-        "t": 1_000_000_000_000,
-        "tn": 1_000_000_000_000,
-        "l": 100_000,
-        "lac": 100_000,
-        "lakh": 100_000,
-        "cr": 10_000_000,
-        "crore": 10_000_000,
-    }
-
-    multiplier = multipliers.get(
-        suffix,
-        1,
-    )
-
-    return number * multiplier
-
-
 # =========================
 # WORKER LOOP HARNESS
 # =========================
@@ -128,9 +44,6 @@ def parse_market_number(value):
 # private *_iteration that runs exactly one pass. The harness loops the
 # iteration, catching unhandled exceptions so a surprise error in any one
 # pass logs + backs off + continues, rather than killing the thread silently.
-#
-# Each worker receives its own stop_event so it can be halted independently
-# without affecting any other running worker.
 
 
 def _safe_worker_loop(name: str, iteration_fn, stop_event: Event):
@@ -148,14 +61,13 @@ def _safe_worker_loop(name: str, iteration_fn, stop_event: Event):
 
 
 # =========================
-# SCRAPER WORKER
+# SCRAPER WORKER (news)
 # =========================
 
 
 def _scraping_iteration(stop_event: Event):
-    """
-    This worker is suppposed to scrape news sources for news. Stores RAW HTML into the database.
-    """
+    """Scrape news sources and store their RAW HTML into the database."""
+
     logger.info("scraper tick: reading source list")
     sources = read_sources("news")
     now_ms = int(time.time() * 1000)
@@ -194,7 +106,6 @@ def _scraping_iteration(stop_event: Event):
 
         source_name = source.name
         url = source.website
-        type = source.type
         attempted += 1
 
         logger.info(f"scraper: fetching {source_name} ({url})")
@@ -205,10 +116,10 @@ def _scraping_iteration(stop_event: Event):
             if html is None:
                 logger.error(f"{source_name} fetch failed")
                 logger.error(f"due to {error_reason}")
-                record_failure(source.id, str(e))
+                record_failure(source.id, str(error_reason))
                 continue
 
-            raw_id = store_raw_html(source_name, html, type)
+            raw_id = store_raw_html(source_name, html, NEWS)
             update_last_fetched(source.id)
             logger.info(f"Stored {source_name} -> {raw_id}")
 
@@ -230,387 +141,143 @@ def scraping_worker(stop_event: Event):
 
 
 # =========================
-# HEADLINE WORKER
+# SHARED FETCH HELPER (stock + macro)
 # =========================
+# Both the stock and macro workers fetch a pre-built URL and store the raw HTML;
+# only the source label and raw_html.type differ. update_source_table stamps
+# last_fetched_at so the row drops out of "due" for the rest of the market day.
 
 
-def _headline_iteration(stop_event: Event):
-    """
-    This worker is supposed to query database for unparsed RAW HTML from news sources and get headlines out of it and store in Parsed Article table
-    """
+def _fetch_and_store(row, model, source_label: str, type_: str):
+    """Fetch row.url and store it as raw_html of `type_`. Stamps the source row."""
 
-    logger.info("Headline tick: looking for unparsed raw_html")
-    record = get_unparsed_record("NEWS")
+    ident = getattr(row, "name", None) or getattr(row, "ticker", None) or row.id
+    country = getattr(row, "country", None)
+    url = row.url
 
-    if not record:
-        logger.info(
-            f"headline tick: nothing to do; " f"sleeping {HEADLINE_IDLE_SECONDS}s"
-        )
-        stop_event.wait(HEADLINE_IDLE_SECONDS)
+    if not url:
+        logger.error(f"{ident} has no pre-built url; skipping")
+        update_source_table(model, row.id, "missing url")
         return
 
-    logger.info(
-        f"headline tick: processing raw_html {record.id} "
-        f"(source={record.source}, {len(record.html)} chars)"
-    )
+    logger.info(f"fetching {ident} ({type_})")
 
     try:
-        started = time.time()
-        headlines = extract_headlines(record.html)
-        logger.info(
-            f"headline: extract_headlines returned {len(headlines)} headlines "
-            f"in {time.time() - started:.1f}s for {record.id}"
-        )
-        store_headlines(record.id, headlines)
-        mark_as_parsed(record.id)
-        logger.info(f"headline tick done: {record.id} parsed")
-
-    except Exception:
-        logger.exception(f"Headline extraction failed for {record.id}")
-        # Mark as parsed so a poison record can't hot-loop the worker.
-        mark_as_parsed(record.id)
-        stop_event.wait(WORKER_ERROR_BACKOFF_SECONDS)
-
-
-def headline_worker(stop_event: Event):
-    _safe_worker_loop("headline_worker", _headline_iteration, stop_event)
-
-
-# =========================
-# SENTIMENT WORKER
-# =========================
-
-
-def _sentiment_iteration(stop_event: Event):
-    """
-    This worker is suppossed to query database for unanalyzed article and get sentiment out of it and update the Parsed Article table
-    """
-    logger.info("sentiment tick: looking for unanalyzed article")
-    article = get_unanalyzed_article()
-
-    if not article:
-        logger.info(
-            f"sentiment tick: nothing to do; " f"sleeping {SENTIMENT_IDLE_SECONDS}s"
-        )
-        stop_event.wait(SENTIMENT_IDLE_SECONDS)
-        return
-
-    preview = (article.title or "")[:80]
-
-    logger.info(f"sentiment tick: analyzing article " f"{article.id} title={preview!r}")
-
-    try:
-        started = time.time()
-
-        result = analyze_article(article.title)
-
-        elapsed = time.time() - started
-
-        logger.info(
-            f"sentiment: result for {article.id} "
-            f"is_valid={result['is_valid']} "
-            f"({elapsed:.1f}s)"
-        )
-
-        if not result["is_valid"]:
-
-            logger.info(f"sentiment: soft deleting invalid headline " f"{article.id}")
-
-            # soft delete invalid headlines
-            soft_delete_invalid_headlines(article.id)
-
-            logger.info(f"sentiment tick done: {article.id} soft deleted")
+        html, error = fetch_html(url)
 
+        if not html:
+            update_source_table(model, row.id, error)
+            logger.error(f"fetch failed for {ident}: {error}")
             return
 
-        logger.info(
-            f"sentiment: article {article.id} "
-            f"sentiment={result['sentiment']} "
-            f"category={result['category']}"
-        )
-
-        update_article_analysis(article.id, result["sentiment"], result["category"])
-
-        logger.info(f"sentiment tick done: {article.id} analyzed")
+        store_raw_html(source_label, html, type_, country=country)
+        update_source_table(model, row.id)
+        logger.info(f"{ident}: raw html stored ({type_})")
 
     except Exception:
+        logger.exception(f"fetch failed for {ident}")
+        update_source_table(model, row.id, "fetch exception")
 
-        logger.exception(f"Sentiment analysis failed for {article.id}")
-
-        stop_event.wait(WORKER_ERROR_BACKOFF_SECONDS)
 
+def _sleep_until_next_close(stop_event: Event, label: str):
+    """Sleep until the soonest market close across all markets (weekend-aware)."""
 
-def sentiment_worker(stop_event: Event):
-    _safe_worker_loop("sentiment_worker", _sentiment_iteration, stop_event)
+    sleep_s = min(seconds_until_next_market_close(m) for m in MARKET_CONFIG)
+    logger.info(
+        f"{label} tick done; sleeping {sleep_s / 3600:.1f}h until next market close"
+    )
+    stop_event.wait(sleep_s)
 
 
 # =========================
-# SENTIMENT TABLE WORKER (own model)
+# STOCK ENHANCER WORKER (fetch-only)
 # =========================
-# Reads parsed articles, scores each with our own sentiment model (probabilities
-# for {0: negative, 1: neutral, 2: positive}), and writes a row to the sentiment
-# table: company, sentiment_score = P(pos)-P(neg), confidence = 1-P(neutral),
-# sector. The source clean_headlines row is flagged is_analyzed=True so the
-# cleanup worker can retire it once it has also been fetched enough times.
-
-
-def _sentiment_table_iteration(stop_event: Event):
-
-    logger.info("sentiment_table tick: looking for unscored headline")
-    headline = get_analyzed_article()
-
-    if not headline:
-        logger.info(
-            f"sentiment_table tick: nothing to do; "
-            f"sleeping {SENTIMENT_TABLE_IDLE_SECONDS}s"
-        )
-        stop_event.wait(SENTIMENT_TABLE_IDLE_SECONDS)
-    else:
-        headline_id = headline.id
-        text = headline.title
-        result = extract_company_sector(text)
-
-        is_valid, ticker, company, sector = (
-            result["is_valid"],
-            result["ticker"],
-            result["company"],
-            result["sector"],
-        )
+# Once per market close (weekday), fetch every stock for that market that hasn't
+# been fetched today, store its Google Finance HTML (type=stock), then sleep to
+# the next close. All metric extraction happens on the LLM side.
 
-        if is_valid:
-            preview = (text or "")[:80]
-            logger.info(f"sentiment_table tick: scoring {headline_id} text={preview!r}")
 
-            try:
-                started = time.time()
+def _stock_market_iteration(stop_event: Event):
 
-                result = predict_own_sentiment(text)
+    for market in MARKET_CONFIG:
 
-                store_sentiment(
-                    company=company,
-                    sentiment_score=result["sentiment_score"],
-                    confidence=result["confidence"],
-                    sector=sector,
-                    headline_id=headline_id,
-                )
+        if not market_already_closed_today(market):
+            logger.info(f"stock: {market} not closed yet today (or weekend); skipping")
+            continue
 
-                mark_article_cleaned(headline_id)
+        due = get_sources_due_for_market(StockSources, market)
+        logger.info(f"stock: {market} closed; {len(due)} stock(s) due")
 
-                logger.info(
-                    f"sentiment_table tick done: {headline_id} -> sentiment "
-                    f"(score={result['sentiment_score']:.4f} "
-                    f"confidence={result['confidence']:.4f}, "
-                    f"{time.time() - started:.1f}s)"
-                )
+        for stock in due:
+            if stop_event.is_set():
+                logger.info("stock tick: shutdown signaled mid-loop, returning")
+                return
+            _fetch_and_store(stock, StockSources, "Google Finance", STOCK)
 
-            except Exception:
-                # Transient failure (model/DB) must not lose the row — leave it
-                # unscored so it is retried on the next tick.
-                mark_article_cleaned(headline_id)
-                logger.exception(f"sentiment_table failed for {headline_id}")
-                stop_event.wait(WORKER_ERROR_BACKOFF_SECONDS)
-        else:
-            mark_article_cleaned(headline_id)
-            logger.exception(f"sentiment_table failed for {headline_id}")
+    _sleep_until_next_close(stop_event, "stock")
 
 
-def sentiment_table_worker(stop_event: Event):
-    _safe_worker_loop("sentiment_table_worker", _sentiment_table_iteration, stop_event)
+def stock_enrichment_worker(stop_event: Event):
+    _safe_worker_loop("stock_enrichment_worker", _stock_market_iteration, stop_event)
 
 
 # =========================
-# SECTOR ROLLUP WORKER (24h)
+# MACRO WORKER (fetch-only)
 # =========================
-# Once per 24h, recompute the sector rollup table from the sentiment table
-# (confidence-weighted sentiment over the trailing 24h and 7 days, plus
-# momentum). Upserts one row per sector. Runs immediately on start, then waits.
-
-
-def _sector_iteration(stop_event: Event):
-
-    logger.info("sector tick: recomputing sector rollup from sentiment table")
-    result = update_sector_table()
-    # Flag clean_headlines whose sentiment has aged out of the 7-day window so
-    # the cleanup worker can retire them (they won't be read again).
-    flagged = mark_clean_headlines_past_7d()
-
-    logger.info(
-        f"sector tick done: inserted={result['inserted']} "
-        f"updated={result['updated']} clean_headlines_flagged={flagged}; "
-        f"sleeping {SECTOR_TICK_SECONDS}s"
-    )
-    stop_event.wait(SECTOR_TICK_SECONDS)
-
-
-def sector_worker(stop_event: Event):
-    _safe_worker_loop("sector_worker", _sector_iteration, stop_event)
-
-
-def get_indian_stock_detail_from_google(
-    stock: Stock,
-):
-
-    if not stock:
-        return
+# Same once-per-close cadence as the stock worker, but over MacroSource rows
+# (FII/DII, VIX, commodities, indices, FX, yields). The raw_html.type is taken
+# from each row so one worker covers every macro family.
 
-    ticker = stock.ticker
-    stock_id = stock.id
-    logger.info(f"enriching stock " f"{ticker}")
 
-    try:
-
-        url = f"https://www.google.com/finance/quote/{ticker}:NSE?tab=overview"
-
-        stock_data_raw, error = fetch_html(url)
-        if not stock_data_raw:
-            update_stock_table(stock_id, error)
-            logger.error(f"unable to get stock info due to error:{error}")
-            return
+def _macro_iteration(stop_event: Event):
 
-        update_stock_table(stock_id)
-        raw_html_id = store_raw_html("Google Finance", stock_data_raw, "STOCK")
-        parsed = extract_stock_metrics_from_html(stock_data_raw)
-
-        if not parsed:
-
-            logger.warning(f"no google metrics " f"for {ticker}")
-
-            return
-
-        logger.debug(f"Returning metrics from html are {parsed}")
-
-        open_val = parse_market_number(parsed.get("Open"))
-        high_val = parse_market_number(parsed.get("High"))
-        low_val = parse_market_number(parsed.get("Low"))
-        close_val = parse_market_number(parsed.get("Close"))
-        volume_val = parse_market_number(parsed.get("volume"))
-
-        is_complete = all(
-            x is not None
-            for x in (
-                open_val,
-                high_val,
-                low_val,
-                close_val,
-                volume_val,
-            )
-        )
-
-        update_stock_data(
-            stock.id,
-            {
-                "open": open_val,
-                "high": high_val,
-                "low": low_val,
-                "close": close_val,
-                "volume": volume_val,
-                "is_complete": is_complete,
-                "raw_html_id": raw_html_id,
-            },
-        )
-
-    except Exception:
+    for market in MARKET_CONFIG:
 
-        logger.exception(f"google enrichment failed " f"for {ticker}")
-
-
-day_map = {
-    0: "Monday",
-    1: "Tuesday",
-    2: "Wednesday",
-    3: "Thrusday",
-    4: "Friday",
-    5: "Saturday",
-    6: "Sunday",
-}
-
-
-def stock_enrichment_iteration(
-    stop_event: Event,
-):
-    now = datetime.now(IST)
-    day = day_map[now.weekday()]
-
-    #
-    # Sunday
-    #
-    if now.weekday() in (5, 6):
-        sleep_seconds = seconds_until_next_monday()
-
-        logger.info(
-            f"Today is {day}. Markets are closed. "
-            f"Sleeping stock enrichment worker for "
-            f"{sleep_seconds / 3600:.1f} hours."
-        )
-
-        stop_event.wait(sleep_seconds)
-        return
-
-    logger.info(f"Today is {day}. ")
-    stock = get_stock_for_enrichment()
-
-    if not stock:
-
-        logger.info(f"No stock found for enrichment")
-        stop_event.wait(HEADLINE_IDLE_SECONDS)
-
-        return
-
-    try:
-
-        get_indian_stock_detail_from_google(stock)
-
-    except Exception:
+        if not market_already_closed_today(market):
+            logger.info(f"macro: {market} not closed yet today (or weekend); skipping")
+            continue
 
-        logger.exception("stock enrichment failed")
+        due = get_sources_due_for_market(MacroSource, market)
+        logger.info(f"macro: {market} closed; {len(due)} series due")
 
-        stop_event.wait(WORKER_ERROR_BACKOFF_SECONDS)
+        for src in due:
+            if stop_event.is_set():
+                logger.info("macro tick: shutdown signaled mid-loop, returning")
+                return
+            _fetch_and_store(src, MacroSource, src.name, src.type)
 
+    _sleep_until_next_close(stop_event, "macro")
 
-def stock_enrichment_worker(
-    stop_event: Event,
-):
 
-    _safe_worker_loop(
-        "stock_enrichment_worker",
-        stock_enrichment_iteration,
-        stop_event,
-    )
+def macro_worker(stop_event: Event):
+    _safe_worker_loop("macro_worker", _macro_iteration, stop_event)
 
 
 # =========================
 # WORKER REGISTRY
 # =========================
 # Maps a stable name → the worker's entry-point function.
-# Add new workers here and they automatically work with all control functions.
 # Registered actions:
-#   START_SCRAPER   / STOP_SCRAPER   / RESTART_SCRAPER
-#   START_HEADLINE  / STOP_HEADLINE  / RESTART_HEADLINE
-#   START_SENTIMENT / STOP_SENTIMENT / RESTART_SENTIMENT
-#   START_CLEANUP   / STOP_CLEANUP   / RESTART_CLEANUP (Retired)
-#   START_PURGE     / STOP_PURGE     / RESTART_PURGE (Retired)
+#   START_SCRAPER         / STOP_SCRAPER         / RESTART_SCRAPER
+#   START_STOCK_ENHANCER  / STOP_STOCK_ENHANCER  / RESTART_STOCK_ENHANCER
+#   START_MACRO           / STOP_MACRO           / RESTART_MACRO
 
 _WORKER_REGISTRY: dict[str, callable] = {
     "scraper": scraping_worker,
-    "headline": headline_worker,
-    "sentiment": sentiment_worker,
-    # Exposed as START_SENTIMENT_TABLE / STOP_* / RESTART_*
-    "sentiment_table": sentiment_table_worker,
-    # Exposed as START_SECTOR / STOP_SECTOR / RESTART_SECTOR
-    # "sector": sector_worker,
     "stock_enhancer": stock_enrichment_worker,
+    "macro": macro_worker,
 }
 
+
 # =========================
 # INTERNAL HELPERS
 # =========================
 
 
 def _launch_worker(name: str):
-    """
-    Spin up a single named worker and register it in _workers.
-    Caller must hold workers_lock.
-    """
+    """Spin up a single named worker and register it in _workers.
+    Caller must hold workers_lock."""
+
     if name not in _WORKER_REGISTRY:
         raise ValueError(
             f"Unknown worker: {name!r}. Valid names: {list(_WORKER_REGISTRY)}"
@@ -630,6 +297,7 @@ def _launch_worker(name: str):
 
 def _signal_and_join(name: str, stop_event: Event, thread: Thread, timeout: float):
     """Signal a worker's event and wait for its thread to exit."""
+
     logger.info(f"stop: signaling '{name}'")
     stop_event.set()
     thread.join(timeout=timeout)
@@ -645,15 +313,9 @@ def _signal_and_join(name: str, stop_event: Event, thread: Thread, timeout: floa
 
 
 def start_workers(names: list[str] | None = None):
-    """
-    Start workers. If *names* is omitted, all registered workers are started.
-    Already-running workers are skipped with a warning.
-
-    Examples
-    --------
-    start_workers()                        # start everything
-    start_workers(["scraper", "headline"]) # start a subset
-    """
+    """Start workers. If *names* is omitted, all registered workers are started.
+    Already-running workers are skipped with a warning."""
+
     targets = names or list(_WORKER_REGISTRY.keys())
 
     with workers_lock:
@@ -668,14 +330,8 @@ def start_workers(names: list[str] | None = None):
 
 
 def stop_worker(name: str, timeout: float = 5.0):
-    """
-    Stop a single named worker and remove it from the registry.
-
-    Examples
-    --------
-    stop_worker("scraper")
-    stop_worker("headline", timeout=10.0)
-    """
+    """Stop a single named worker and remove it from the registry."""
+
     with workers_lock:
         entry = _workers.get(name)
         if not entry:
@@ -687,14 +343,8 @@ def stop_worker(name: str, timeout: float = 5.0):
 
 
 def stop_workers(names: list[str] | None = None, timeout: float = 5.0):
-    """
-    Stop workers. If *names* is omitted, all running workers are stopped.
-
-    Examples
-    --------
-    stop_workers()                          # full shutdown
-    stop_workers(["scraper", "sentiment"])  # stop a subset
-    """
+    """Stop workers. If *names* is omitted, all running workers are stopped."""
+
     targets = names or list(_workers.keys())
     for name in targets:
         stop_worker(name, timeout=timeout)
@@ -702,13 +352,8 @@ def stop_workers(names: list[str] | None = None, timeout: float = 5.0):
 
 
 def restart_worker(name: str, timeout: float = 5.0):
-    """
-    Stop a worker (if running) then immediately restart it.
+    """Stop a worker (if running) then immediately restart it."""
 
-    Examples
-    --------
-    restart_worker("scraper")
-    """
     with workers_lock:
         entry = _workers.get(name)
         if entry:
@@ -719,14 +364,8 @@ def restart_worker(name: str, timeout: float = 5.0):
 
 
 def worker_status() -> dict[str, bool]:
-    """
-    Return a snapshot of {worker_name: is_alive} for every registered worker.
-
-    Examples
-    --------
-    status = worker_status()
-    # {"scraper": True, "headline": True, "sentiment": False, ...}
-    """
+    """Return a snapshot of {worker_name: is_alive} for every running worker."""
+
     with workers_lock:
         return {name: entry["thread"].is_alive() for name, entry in _workers.items()}