remove unwanted files and create proper sql tables

Dockerfile

@@ -2,7 +2,7 @@ FROM python:3.12-slim
 
 WORKDIR /app
 
-# build-essential: needed for xgboost/duckdb wheels.
+# build-essential: needed for xgboost wheels.
 # The rest: Playwright Chromium runtime libs (from the scrapper service).
 RUN apt-get update \
     && apt-get install -y --no-install-recommends \
@@ -42,16 +42,14 @@ RUN python -m playwright install chromium
 COPY api ./api
 COPY app ./app
 COPY database ./database
-COPY docker ./docker
 COPY evaluation ./evaluation
 COPY feature_engine ./feature_engine
 COPY inference ./inference
 COPY jobs ./jobs
 COPY static_data ./static_data
-COPY tests ./tests
 COPY scrapper_service ./scrapper_service
 
-RUN mkdir -p /tmp/duckdb_feature_engine /data/db
+RUN mkdir -p /data/db
 
 EXPOSE 8000
 

README.md

@@ -499,6 +499,19 @@ assert result.actual_return == pytest.approx(0.03)
 
 # 9. Architecture: SQLite vs Market Data
 
+## Current SQLite feature pipeline
+
+Raw data is separated into `ohlcv` (India stocks), `nifty_ohlcv`,
+`india_vix_ohlcv`, `us_stocks_ohlcv`, and `macro_ohlcv`. Macro rows use the
+`macro_type` values `gold`, `brent_crude_oil`, and `usd_inr`; the
+`all_market_ohlcv` view is the unified read path.
+
+`jobs.rebuild_feature_stores` persists the five source feature tables, the
+model-ready `merged_features` table, and the D+1-open / next-five-closes
+`training_labels` table in SQLite. No DuckDB dependency is used. The guarded
+`POST /predict?prediction_date=YYYY-MM-DD` endpoint returns
+`not_a_trading_day` when no NSE candle exists for the requested date.
+
 A critical architectural detail:
 
 ## The database does NOT contain the engineered feature matrix
@@ -1831,7 +1844,7 @@ ORDER BY created_at DESC;
 Run:
 
 ```bash
-pytest -q tests/test_resolver.py
+python -m jobs.resolve_predictions
 ```
 
 Expected:

app/config.py

@@ -39,7 +39,7 @@ class Settings:
     STATIC_DATA_DIR = Path(
         os.getenv(
             "STATIC_DATA_DIR",
-            "/app/static_data",
+            str(Path(__file__).resolve().parents[1] / "static_data"),
         )
     )
 
@@ -115,4 +115,4 @@ class Settings:
     CORR_LOOKBACK_DAYS = 70
 
 
-settings = Settings()
+settings = Settings()

app/main.py

@@ -1,3 +1,5 @@
+from contextlib import asynccontextmanager
+from datetime import date
 from fastapi import Depends, FastAPI, HTTPException
 from fastapi.middleware.gzip import GZipMiddleware
 from dotenv import load_dotenv
@@ -24,10 +26,28 @@ from jobs.daily_prediction import (
 )
 
 
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Keep the market universe and its SQLite cache ready before requests.
+    # Ingestion starts at each ticker's first missing date, so restarts do not
+    # repeatedly download the initial 800-day history.
+    from app.config import settings
+    from database.connection import connect
+    from jobs.market_ingestion import ingest_market_data, sync_static_tickers
+    from jobs.rebuild_feature_stores import rebuild_feature_stores
+
+    with connect(settings.DATABASE_URL) as conn:
+        sync_static_tickers(conn, settings.STATIC_DATA_DIR)
+        ingest_market_data(conn)
+    rebuild_feature_stores()
+    async with scrapper_lifespan(app):
+        yield
+
+
 app = FastAPI(
     title="Stock Prediction Engine",
     version="1.0.0",
-    lifespan=scrapper_lifespan,
+    lifespan=lifespan,
 )
 
 app.add_middleware(GZipMiddleware, minimum_size=1000)
@@ -42,18 +62,38 @@ def health():
 
 
 @app.post("/predict", dependencies=[Depends(require_pipeline_guid)])
-def predict():
+def predict(prediction_date: date | None = None):
 
     try:
 
-        result = run_daily_prediction()
+        target_date = prediction_date or pd.Timestamp.now(tz="Asia/Kolkata").date()
+        from app.config import settings
+        from database.connection import connect
+
+        # NSE candles are the trading-calendar authority.  This also handles
+        # exchange holidays without maintaining a second hard-coded calendar.
+        with connect(settings.DATABASE_URL) as conn:
+            with conn.cursor() as cur:
+                cur.execute("SELECT 1 FROM ohlcv WHERE trading_date = ? LIMIT 1", (target_date,))
+                if cur.fetchone() is None:
+                    return {"status": "not_a_trading_day", "date": target_date.isoformat()}
+
+        result = run_daily_prediction(pd.Timestamp(target_date))
 
         return {
             "status": "success",
             "rows": len(result),
             "date": str(
-                result["timestamp"].iloc[0]
+                target_date
             ),
+            "predictions": [
+                {
+                    "symbol": row.symbol,
+                    "predicted_probability": float(row.predicted_probability),
+                    "rank": int(row.rank),
+                }
+                for row in result.itertuples(index=False)
+            ],
         }
 
     except Exception as exc:

database/market_repository.py

@@ -40,15 +40,35 @@ def delete_ticker(conn, ticker_id: int) -> bool:
         return cur.rowcount == 1
 
 
-def upsert_ohlcv(conn, ticker_id: int, candle: dict[str, Any]) -> None:
+def _ohlcv_table(ticker: dict[str, Any]) -> tuple[str, str | None]:
+    if ticker["market"] == "NSE":
+        return "ohlcv", None
+    if ticker["market"] == "US":
+        return "us_stocks_ohlcv", None
+    if ticker["symbol"] == "NIFTY":
+        return "nifty_ohlcv", None
+    if ticker["symbol"] == "INDIA_VIX":
+        return "india_vix_ohlcv", None
+    macro_types = {"GOLD": "gold", "BRENT": "brent_crude_oil", "USDINR": "usd_inr"}
+    return "macro_ohlcv", macro_types[ticker["symbol"]]
+
+
+def upsert_ohlcv(conn, ticker: dict[str, Any], candle: dict[str, Any]) -> None:
+    table, macro_type = _ohlcv_table(ticker)
     with conn.cursor() as cur:
+        columns = "ticker_id, trading_date, open, high, low, close, volume"
+        values = ":ticker_id, :trading_date, :open, :high, :low, :close, :volume"
+        params = {"ticker_id": ticker["id"], **candle}
+        if macro_type:
+            columns = "ticker_id, macro_type, trading_date, open, high, low, close, volume"
+            values = ":ticker_id, :macro_type, :trading_date, :open, :high, :low, :close, :volume"
+            params["macro_type"] = macro_type
         cur.execute("""
-            INSERT INTO market_ohlcv (ticker_id, trading_date, open, high, low, close, volume)
-            VALUES (:ticker_id, :trading_date, :open, :high, :low, :close, :volume)
+            INSERT INTO {table} ({columns}) VALUES ({values})
             ON CONFLICT (ticker_id, trading_date) DO UPDATE SET
                 open = excluded.open, high = excluded.high, low = excluded.low,
                 close = excluded.close, volume = excluded.volume, fetched_at = STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')
-        """, {"ticker_id": ticker_id, **candle})
+        """.format(table=table, columns=columns, values=values), params)
 
 
 def list_ohlcv(
@@ -62,7 +82,7 @@ def list_ohlcv(
     with conn.cursor() as cur:
         cur.execute("""
             SELECT t.symbol, t.market, o.trading_date, o.open, o.high, o.low, o.close, o.volume
-            FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
+            FROM all_market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
             WHERE (:symbol IS NULL OR t.symbol = :symbol)
               AND (:market IS NULL OR t.market = :market)
               AND (:start IS NULL OR o.trading_date >= :start)
@@ -78,5 +98,5 @@ def latest_ohlcv_dates(conn) -> dict[int, object]:
     # column type for it (unlike a bare column reference) and the PARSE_DECLTYPES
     # date converter never fires here — parse the returned ISO text explicitly.
     with conn.cursor() as cur:
-        cur.execute("SELECT ticker_id, MAX(trading_date) AS trading_date FROM market_ohlcv GROUP BY ticker_id")
+        cur.execute("SELECT ticker_id, MAX(trading_date) AS trading_date FROM all_market_ohlcv GROUP BY ticker_id")
         return {row["ticker_id"]: date.fromisoformat(row["trading_date"]) for row in cur.fetchall()}

database/resolution_repository.py

@@ -21,7 +21,7 @@ def fetch_future_ohlcv(conn, symbol: str, prediction_date: date, forward_days: i
     with conn.cursor() as cur:
         cur.execute("""
             SELECT o.trading_date AS timestamp, o.open, o.close
-            FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
+            FROM ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
             WHERE t.symbol = ? AND t.market = 'NSE' AND o.trading_date > ?
             ORDER BY o.trading_date ASC LIMIT ?
         """, (symbol, prediction_date, forward_days))

database/schema.sql

@@ -250,3 +250,93 @@ CREATE TABLE IF NOT EXISTS market_ohlcv (
 
 CREATE INDEX IF NOT EXISTS idx_market_ohlcv_ticker_date
     ON market_ohlcv (ticker_id, trading_date DESC);
+
+-- ============================================================
+-- 6. MARKET DATA BY DATASET
+--
+-- The legacy market_ohlcv table remains only as a migration source for an
+-- existing installation.  New reads and writes use these dataset tables.
+-- ============================================================
+
+CREATE TABLE IF NOT EXISTS ohlcv (
+    ticker_id INTEGER NOT NULL REFERENCES market_tickers(id) ON DELETE CASCADE,
+    trading_date DATE NOT NULL,
+    open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL,
+    volume INTEGER, source TEXT NOT NULL DEFAULT 'yfinance',
+    fetched_at TIMESTAMP NOT NULL DEFAULT (STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (ticker_id, trading_date)
+);
+
+CREATE TABLE IF NOT EXISTS nifty_ohlcv (
+    ticker_id INTEGER NOT NULL REFERENCES market_tickers(id) ON DELETE CASCADE,
+    trading_date DATE NOT NULL,
+    open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL,
+    volume INTEGER, source TEXT NOT NULL DEFAULT 'yfinance',
+    fetched_at TIMESTAMP NOT NULL DEFAULT (STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (ticker_id, trading_date)
+);
+
+CREATE TABLE IF NOT EXISTS india_vix_ohlcv (
+    ticker_id INTEGER NOT NULL REFERENCES market_tickers(id) ON DELETE CASCADE,
+    trading_date DATE NOT NULL,
+    open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL,
+    volume INTEGER, source TEXT NOT NULL DEFAULT 'yfinance',
+    fetched_at TIMESTAMP NOT NULL DEFAULT (STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (ticker_id, trading_date)
+);
+
+CREATE TABLE IF NOT EXISTS us_stocks_ohlcv (
+    ticker_id INTEGER NOT NULL REFERENCES market_tickers(id) ON DELETE CASCADE,
+    trading_date DATE NOT NULL,
+    open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL,
+    volume INTEGER, source TEXT NOT NULL DEFAULT 'yfinance',
+    fetched_at TIMESTAMP NOT NULL DEFAULT (STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (ticker_id, trading_date)
+);
+
+CREATE TABLE IF NOT EXISTS macro_ohlcv (
+    ticker_id INTEGER NOT NULL REFERENCES market_tickers(id) ON DELETE CASCADE,
+    macro_type TEXT NOT NULL CHECK (macro_type IN ('gold', 'brent_crude_oil', 'usd_inr')),
+    trading_date DATE NOT NULL,
+    open REAL NOT NULL, high REAL NOT NULL, low REAL NOT NULL, close REAL NOT NULL,
+    volume INTEGER, source TEXT NOT NULL DEFAULT 'yfinance',
+    fetched_at TIMESTAMP NOT NULL DEFAULT (STRFTIME('%Y-%m-%dT%H:%M:%fZ', 'now')),
+    PRIMARY KEY (ticker_id, trading_date)
+);
+
+CREATE VIEW IF NOT EXISTS all_market_ohlcv AS
+SELECT ticker_id, trading_date, open, high, low, close, volume, source, fetched_at FROM ohlcv
+UNION ALL SELECT ticker_id, trading_date, open, high, low, close, volume, source, fetched_at FROM nifty_ohlcv
+UNION ALL SELECT ticker_id, trading_date, open, high, low, close, volume, source, fetched_at FROM india_vix_ohlcv
+UNION ALL SELECT ticker_id, trading_date, open, high, low, close, volume, source, fetched_at FROM us_stocks_ohlcv
+UNION ALL SELECT ticker_id, trading_date, open, high, low, close, volume, source, fetched_at FROM macro_ohlcv;
+
+-- Copy legacy data exactly once when upgrading an existing database.
+INSERT OR IGNORE INTO ohlcv SELECT o.* FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id WHERE t.market = 'NSE';
+INSERT OR IGNORE INTO nifty_ohlcv SELECT o.* FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id WHERE t.symbol = 'NIFTY';
+INSERT OR IGNORE INTO india_vix_ohlcv SELECT o.* FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id WHERE t.symbol = 'INDIA_VIX';
+INSERT OR IGNORE INTO us_stocks_ohlcv SELECT o.* FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id WHERE t.market = 'US';
+INSERT OR IGNORE INTO macro_ohlcv (ticker_id, macro_type, trading_date, open, high, low, close, volume, source, fetched_at)
+SELECT o.ticker_id, CASE t.symbol WHEN 'GOLD' THEN 'gold' WHEN 'BRENT' THEN 'brent_crude_oil' WHEN 'USDINR' THEN 'usd_inr' END,
+       o.trading_date, o.open, o.high, o.low, o.close, o.volume, o.source, o.fetched_at
+FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id WHERE t.symbol IN ('GOLD', 'BRENT', 'USDINR');
+
+-- Feature tables are populated by jobs/rebuild_feature_stores.py in SQLite.
+-- Their dynamic model columns replace these minimal bootstrap definitions.
+CREATE TABLE IF NOT EXISTS ohlcv_features (timestamp DATE NOT NULL, symbol TEXT NOT NULL);
+CREATE TABLE IF NOT EXISTS nifty_features (timestamp DATE NOT NULL);
+CREATE TABLE IF NOT EXISTS vix_features (timestamp DATE NOT NULL);
+CREATE TABLE IF NOT EXISTS macro_features (timestamp DATE NOT NULL);
+CREATE TABLE IF NOT EXISTS us_features (timestamp DATE NOT NULL, symbol TEXT NOT NULL);
+CREATE TABLE IF NOT EXISTS merged_features (timestamp DATE NOT NULL, symbol TEXT NOT NULL);
+
+CREATE TABLE IF NOT EXISTS training_labels (
+    timestamp DATE NOT NULL,
+    symbol TEXT NOT NULL,
+    entry_date DATE,
+    entry_open REAL,
+    max_close_5d REAL,
+    actual_return REAL,
+    label INTEGER CHECK (label IN (0, 1)),
+    PRIMARY KEY (timestamp, symbol)
+);

docker/entrypoint.sh

@@ -1,8 +0,0 @@
-#!/bin/sh
-set -eu
-
-echo "Production scheduler starting..."
-echo "Timezone: ${TZ:-Asia/Kolkata}"
-echo "Schedule: Monday-Friday at 16:00 IST"
-
-exec python -m jobs.scheduler

evaluation/README.md

@@ -1,40 +0,0 @@
-# Prediction Resolver
-
-The production contract is fixed:
-
-```text
-Prediction D
-    |
-    +--> D+1 OPEN = entry price
-    |
-    +--> D+1 CLOSE
-    +--> D+2 CLOSE
-    +--> D+3 CLOSE
-    +--> D+4 CLOSE
-    +--> D+5 CLOSE
-              |
-              v
-        MAX(CLOSE)
-              |
-              v
-actual_return = (max_close / entry_open) - 1
-              |
-              v
-actual_label = 1 if actual_return >= 0.03 else 0
-```
-
-Important:
-
-- `D` itself is never part of the return window.
-- Entry is always the **next trading day's OPEN**.
-- The maximum is taken over the **five closes D+1 through D+5**.
-- A prediction with fewer than five future trading sessions remains unresolved.
-- Exactly `3.0%` is a positive label.
-- This resolver does not recalculate model features. It only resolves the outcome.
-- The OHLCV price source must use the same adjustment convention as training.
-
-Run tests from the project root:
-
-```bash
-pytest -q tests/test_resolver.py
-```

evaluation/resolver.py

@@ -1,9 +0,0 @@
-"""Compatibility import for the canonical resolution calculation layer."""
-
-from jobs.resolver import (
-    FORWARD_TRADING_DAYS,
-    TARGET_RETURN,
-    Resolution,
-    resolve_from_ohlcv_rows,
-    resolve_prediction_rows,
-)

feature_engine/engine.py

@@ -1,8 +1,8 @@
+"""Build and persist model features using SQLite-backed feature stores only."""
+
 from __future__ import annotations
 
 import json
-
-import duckdb
 import pandas as pd
 
 from app.config import settings
@@ -16,78 +16,61 @@ class FeatureEngine:
         with settings.METADATA_PATH.open() as handle:
             self.required_features = set(json.load(handle)["feature_cols"])
 
-    def _connection(self):
-        con = duckdb.connect()
-        con.execute("PRAGMA threads=2")
-        con.execute("PRAGMA memory_limit='4GB'")
-        con.execute("PRAGMA temp_directory='/tmp/duckdb_feature_engine'")
-        return con
-
     @staticmethod
-    def _read_table(sqlite_conn, table_name: str) -> pd.DataFrame:
-        return pd.read_sql(f"SELECT * FROM {table_name}", sqlite_conn, parse_dates=["timestamp"])
+    def _read_table(conn, table_name: str) -> pd.DataFrame:
+        return pd.read_sql_query(f'SELECT * FROM "{table_name}"', conn, parse_dates=["timestamp"])
 
-    @staticmethod
-    def _columns(con, table_name: str):
-        return [row[0] for row in con.execute(f"DESCRIBE SELECT * FROM {table_name}").fetchall()]
+    def _merge(self, ohlcv_df, nifty_df, vix_df, macro_df, us_features_df) -> pd.DataFrame:
+        """Attach only information available on or before each NSE session."""
+        us_wide = build_us_wide(us_features_df)
+        base = ohlcv_df.sort_values(["timestamp", "symbol"]).copy()
 
-    def build_features_for_date(self, prediction_date: pd.Timestamp) -> pd.DataFrame:
-        prediction_date = pd.Timestamp(prediction_date)
-        start_date = prediction_date - pd.Timedelta(days=settings.FEATURE_LOOKBACK_DAYS)
+        def attach(left, right, columns):
+            if not columns:
+                return left
+            right = right[["timestamp", *columns]].sort_values("timestamp")
+            return pd.merge_asof(left.sort_values("timestamp"), right, on="timestamp", direction="backward")
+
+        nifty_cols = [c for c in nifty_df if c != "timestamp" and f"nifty_{c}" in self.required_features]
+        nifty = nifty_df.rename(columns={c: f"nifty_{c}" for c in nifty_cols})
+        base = attach(base, nifty, [f"nifty_{c}" for c in nifty_cols])
 
-        sqlite_conn = connect(settings.DATABASE_URL)
-        # pandas' sqlite reader assumes plain tuple rows to infer column dtypes;
-        # our shared connect() sets a dict row_factory for the rest of the app,
-        # which silently corrupts pd.read_sql results if left in place here.
-        sqlite_conn.row_factory = None
+        vix_cols = [c for c in vix_df if c != "timestamp" and c in self.required_features]
+        if "close_1" in self.required_features and "close" in vix_df:
+            vix_df = vix_df.rename(columns={"close": "close_1"})
+            vix_cols.append("close_1")
+        base = attach(base, vix_df, list(dict.fromkeys(vix_cols)))
+        base = attach(base, macro_df, [c for c in macro_df if c != "timestamp" and c in self.required_features])
+        base = attach(base, us_wide, [c for c in us_wide if c != "timestamp" and c in self.required_features])
+        return add_correlation_features(base.sort_values(["symbol", "timestamp"]).reset_index(drop=True))
+
+    def build_all_features(self, conn=None) -> pd.DataFrame:
+        owns_connection = conn is None
+        conn = conn or connect(settings.DATABASE_URL)
+        original_row_factory = conn.row_factory
+        conn.row_factory = None
         try:
-            ohlcv_df = self._read_table(sqlite_conn, settings.OHLCV_FEATURES_TABLE)
-            nifty_df = self._read_table(sqlite_conn, settings.NIFTY_FEATURES_TABLE)
-            vix_df = self._read_table(sqlite_conn, settings.VIX_FEATURES_TABLE)
-            macro_df = self._read_table(sqlite_conn, settings.MACRO_FEATURES_TABLE)
-            us_features_df = self._read_table(sqlite_conn, settings.US_FEATURES_TABLE)
+            frames = [self._read_table(conn, name) for name in (
+                settings.OHLCV_FEATURES_TABLE, settings.NIFTY_FEATURES_TABLE,
+                settings.VIX_FEATURES_TABLE, settings.MACRO_FEATURES_TABLE,
+                settings.US_FEATURES_TABLE,
+            )]
         finally:
-            sqlite_conn.close()
+            conn.row_factory = original_row_factory
+            if owns_connection:
+                conn.close()
+        return self._merge(*frames)
 
-        us_wide = build_us_wide(us_features_df, start_date, prediction_date)
-        df = self._load_and_merge(ohlcv_df, nifty_df, vix_df, macro_df, us_wide, start_date, prediction_date)
-        df = add_correlation_features(df)
-        df = df[df["timestamp"].dt.normalize() == prediction_date.normalize()].copy()
+    def build_features_for_date(self, prediction_date: pd.Timestamp) -> pd.DataFrame:
+        prediction_date = pd.Timestamp(prediction_date).normalize()
+        conn = connect(settings.DATABASE_URL)
+        conn.row_factory = None
+        try:
+            df = pd.read_sql_query('SELECT * FROM merged_features WHERE timestamp = ?', conn,
+                                   params=(prediction_date.date().isoformat(),), parse_dates=["timestamp"])
+        finally:
+            conn.close()
         missing = sorted(self.required_features - set(df.columns))
         if missing:
             raise RuntimeError(f"Feature-store merge is missing model columns: {missing}")
         return df
-
-    def _load_and_merge(self, ohlcv_df, nifty_df, vix_df, macro_df, us_wide, start_date, end_date):
-        con = self._connection()
-        try:
-            con.register("ohlcv_df", ohlcv_df)
-            con.register("nifty_df", nifty_df)
-            con.register("vix_df", vix_df)
-            con.register("macro_df", macro_df)
-            con.register("us_wide", us_wide)
-
-            nifty_columns = self._columns(con, "nifty_df")
-            vix_columns = self._columns(con, "vix_df")
-            macro_columns = self._columns(con, "macro_df")
-
-            nifty_select = [f'n."{col}" AS "nifty_{col}"' for col in nifty_columns if f"nifty_{col}" in self.required_features]
-            vix_select = [f'v."{col}"' for col in vix_columns if col != "timestamp" and col in self.required_features]
-            if "close_1" in self.required_features and "close" in vix_columns:
-                vix_select.append('v."close" AS "close_1"')
-            macro_select = [f'm."{col}"' for col in macro_columns if col != "timestamp" and col in self.required_features]
-            us_select = [f'u."{col}"' for col in us_wide.columns if col != "timestamp" and col in self.required_features]
-            selects = ["o.*", *nifty_select, *vix_select, *macro_select, *us_select]
-            query = f"""
-                SELECT {", ".join(selects)}
-                FROM ohlcv_df o
-                ASOF LEFT JOIN nifty_df n ON o.timestamp >= n.timestamp
-                ASOF LEFT JOIN vix_df v ON o.timestamp >= v.timestamp
-                ASOF LEFT JOIN macro_df m ON o.timestamp >= m.timestamp
-                ASOF LEFT JOIN us_wide u ON o.timestamp >= u.timestamp
-                WHERE o.timestamp >= ? AND o.timestamp <= ?
-                ORDER BY o.symbol, o.timestamp
-            """
-            return con.execute(query, [start_date.date(), end_date.date()]).df()
-        finally:
-            con.close()

feature_engine/nifty.py

@@ -1,27 +0,0 @@
-from pathlib import Path
-import duckdb
-import pandas as pd
-
-
-def load_nifty(
-    path: Path,
-    start_date: pd.Timestamp,
-    end_date: pd.Timestamp,
-) -> pd.DataFrame:
-
-    con = duckdb.connect()
-
-    try:
-        return con.execute(
-            f"""
-            SELECT *
-            FROM read_parquet('{path.as_posix()}')
-            WHERE
-                timestamp <= '{end_date.date()}'
-                AND timestamp >= '{start_date.date()}'
-            ORDER BY timestamp
-            """
-        ).df()
-
-    finally:
-        con.close()

feature_engine/stock.py

@@ -1,29 +0,0 @@
-from pathlib import Path
-
-import duckdb
-import pandas as pd
-
-
-def load_stock_features(
-    path: Path,
-    start_date: pd.Timestamp,
-    end_date: pd.Timestamp,
-) -> pd.DataFrame:
-
-    con = duckdb.connect()
-
-    try:
-
-        query = f"""
-            SELECT *
-            FROM read_parquet('{path.as_posix()}')
-            WHERE
-                timestamp >= '{start_date.date()}'
-                AND timestamp <= '{end_date.date()}'
-            ORDER BY symbol, timestamp
-        """
-
-        return con.execute(query).df()
-
-    finally:
-        con.close()

feature_engine/us_market.py

@@ -1,76 +1,17 @@
-import duckdb
 import pandas as pd
 
 
-def build_us_wide(
-    us_features_df: pd.DataFrame,
-    start_date: pd.Timestamp,
-    end_date: pd.Timestamp,
-) -> pd.DataFrame:
-
-    con = duckdb.connect()
-    con.register("us_features", us_features_df)
-
-    query = f"""
-        SELECT
-            timestamp,
-
-            AVG(ret_1d) AS us_ret_1d,
-            AVG(ret_5d) AS us_ret_5d,
-            AVG(ret_20d) AS us_ret_20d,
-            AVG(ret_60d) AS us_ret_60d,
-
-            AVG(
-                CASE
-                    WHEN ret_1d > 0 THEN 1.0
-                    ELSE 0.0
-                END
-            ) AS us_breadth_1d,
-
-            AVG(
-                CASE
-                    WHEN ret_5d > 0 THEN 1.0
-                    ELSE 0.0
-                END
-            ) AS us_breadth_5d,
-
-            AVG(
-                CASE
-                    WHEN ret_20d > 0 THEN 1.0
-                    ELSE 0.0
-                END
-            ) AS us_breadth_20d,
-
-            STDDEV(ret_1d) AS us_dispersion_1d,
-            STDDEV(ret_5d) AS us_dispersion_5d,
-
-            AVG(vol_ratio) AS us_avg_vol_ratio,
-
-            AVG(
-                CASE
-                    WHEN close_sma20_ratio > 0
-                    THEN 1.0
-                    ELSE 0.0
-                END
-            ) AS us_pct_above_sma20,
-
-            AVG(rsi_14) AS us_avg_rsi,
-
-            AVG(distance_from_52w_high)
-                AS us_avg_dist_52w_high
-
-        FROM us_features
-
-        WHERE
-            timestamp >= '{start_date.date()}'
-            AND timestamp <= '{end_date.date()}'
-
-        GROUP BY timestamp
-
-        ORDER BY timestamp
-    """
-
-    try:
-        return con.execute(query).df()
-    finally:
-        con.close()
+def build_us_wide(us_features_df: pd.DataFrame) -> pd.DataFrame:
+    """Aggregate US stock features by session without DuckDB."""
+    return us_features_df.groupby("timestamp", as_index=False).agg(
+        us_ret_1d=("ret_1d", "mean"), us_ret_5d=("ret_5d", "mean"),
+        us_ret_20d=("ret_20d", "mean"), us_ret_60d=("ret_60d", "mean"),
+        us_breadth_1d=("ret_1d", lambda s: (s > 0).mean()),
+        us_breadth_5d=("ret_5d", lambda s: (s > 0).mean()),
+        us_breadth_20d=("ret_20d", lambda s: (s > 0).mean()),
+        us_dispersion_1d=("ret_1d", "std"), us_dispersion_5d=("ret_5d", "std"),
+        us_avg_vol_ratio=("vol_ratio", "mean"),
+        us_pct_above_sma20=("close_sma20_ratio", lambda s: (s > 0).mean()),
+        us_avg_rsi=("rsi_14", "mean"),
+        us_avg_dist_52w_high=("distance_from_52w_high", "mean"),
+    ).sort_values("timestamp")

jobs/healthcheck.py

@@ -1,34 +0,0 @@
-"""
-Small process-level health check for container/orchestrator use.
-"""
-
-from __future__ import annotations
-
-import os
-import sys
-
-from database.connection import connect
-
-
-def main() -> int:
-    database_url = os.getenv("DATABASE_URL")
-
-    if not database_url:
-        print("UNHEALTHY: DATABASE_URL is not configured")
-        return 1
-
-    try:
-        with connect(database_url) as conn:
-            with conn.cursor() as cur:
-                cur.execute("SELECT 1")
-                cur.fetchone()
-    except Exception as exc:
-        print(f"UNHEALTHY: SQLite connection failed: {exc}")
-        return 1
-
-    print("HEALTHY")
-    return 0
-
-
-if __name__ == "__main__":
-    sys.exit(main())

jobs/market_ingestion.py

@@ -61,7 +61,9 @@ def ingest_market_data(conn, lookback_days: int = 800, overlap_days: int = 7) ->
     inserted = 0
     for ticker in rows:
         latest_date = latest_dates.get(ticker["id"])
-        start = (latest_date - timedelta(days=overlap_days)) if latest_date else (date.today() - timedelta(days=lookback_days))
+        # A populated ticker is fetched only from the first missing session;
+        # the initial history window is used once, not every day.
+        start = (latest_date + timedelta(days=1)) if latest_date else (date.today() - timedelta(days=lookback_days))
         history = yf.Ticker(ticker["yfinance_symbol"]).history(start=start.isoformat(), auto_adjust=False)
         if history.empty:
             continue
@@ -132,9 +134,7 @@ def ingest_market_data(conn, lookback_days: int = 800, overlap_days: int = 7) ->
                 volume_value = int(volume)
 
             upsert_ohlcv(
-                conn,
-                ticker["id"],
-                {
+                conn, ticker, {
                     "trading_date": timestamp.date(),
                     "open": open_price,
                     "high": high_price,

jobs/rebuild_feature_stores.py

@@ -7,6 +7,7 @@ from scrapper_service.logger import logger
 from app.config import settings
 from database.connection import connect
 from feature_engine.training_transforms import macro, nifty, ohlcv, vix
+from feature_engine.engine import FeatureEngine
 
 
 def _load_candles(conn, market: str | None = None, symbols: tuple[str, ...] | None = None) -> pd.DataFrame:
@@ -20,7 +21,7 @@ def _load_candles(conn, market: str | None = None, symbols: tuple[str, ...] | No
         params.extend(symbols)
     query = f"""
         SELECT o.trading_date AS timestamp, t.symbol, o.open, o.high, o.low, o.close, o.volume
-        FROM market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
+        FROM all_market_ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
         WHERE {" AND ".join(clauses)}
         ORDER BY t.symbol, o.trading_date
     """
@@ -40,6 +41,24 @@ def _write_table(df: pd.DataFrame, conn, table_name: str) -> None:
         raise
 
 
+def _build_labels(conn) -> pd.DataFrame:
+    """Create auditable D+1-open / next-five-closes training labels in SQLite."""
+    raw = pd.read_sql_query("""
+        SELECT t.symbol, o.trading_date AS timestamp, o.open, o.close
+        FROM ohlcv o JOIN market_tickers t ON t.id = o.ticker_id
+        WHERE t.active ORDER BY t.symbol, o.trading_date
+    """, conn, parse_dates=["timestamp"])
+    raw["entry_date"] = raw.groupby("symbol")["timestamp"].shift(-1)
+    raw["entry_open"] = raw.groupby("symbol")["open"].shift(-1)
+    future_closes = [raw.groupby("symbol")["close"].shift(-i) for i in range(1, 6)]
+    raw["max_close_5d"] = pd.concat(future_closes, axis=1).max(axis=1)
+    complete = pd.concat(future_closes, axis=1).notna().all(axis=1) & raw["entry_open"].gt(0)
+    raw["actual_return"] = raw["max_close_5d"] / raw["entry_open"] - 1
+    raw["label"] = pd.NA
+    raw.loc[complete, "label"] = (raw.loc[complete, "actual_return"] >= settings.TARGET_THRESHOLD).astype(int)
+    return raw[["timestamp", "symbol", "entry_date", "entry_open", "max_close_5d", "actual_return", "label"]]
+
+
 def rebuild_feature_stores() -> dict[str, int]:
 
     logger.info(
@@ -211,6 +230,12 @@ def rebuild_feature_stores() -> dict[str, int]:
                 settings.MACRO_FEATURES_TABLE,
             )
 
+            # The merged matrix is also persisted in SQLite so serving never
+            # needs an in-memory DuckDB query.
+            merged_features = FeatureEngine().build_all_features(conn)
+            _write_table(merged_features, conn, "merged_features")
+            _write_table(_build_labels(conn), conn, "training_labels")
+
             conn.commit()
 
         except Exception:
@@ -251,6 +276,8 @@ def rebuild_feature_stores() -> dict[str, int]:
                 settings.NIFTY_FEATURES_TABLE,
                 settings.VIX_FEATURES_TABLE,
                 settings.MACRO_FEATURES_TABLE,
+                "merged_features",
+                "training_labels",
             ]
 
             missing = [
@@ -272,6 +299,7 @@ def rebuild_feature_stores() -> dict[str, int]:
         "nifty": len(nifty_features),
         "vix": len(vix_features),
         "macro": len(macro_features),
+        "merged": len(merged_features),
     }
 
 

jobs/resolve_predictions.py

@@ -1,4 +1,4 @@
-"""Resolve mature predictions using the market_ohlcv table."""
+"""Resolve mature predictions using the India-stock OHLCV table."""
 
 from __future__ import annotations
 
@@ -16,6 +16,3 @@ def main() -> None:
         resolved = resolve_pending_predictions(conn)
     print(f"Resolution complete: {resolved} predictions resolved")
 
-
-if __name__ == "__main__":
-    main()

jobs/run_once.py

@@ -1,13 +0,0 @@
-"""
-Run the complete Phase 5 pipeline once.
-
-Useful for:
-    - deployment smoke tests
-    - manually replaying a production day
-    - debugging without waiting for 16:00 IST
-"""
-
-from jobs.daily_pipeline import run_pipeline
-
-if __name__ == "__main__":
-    run_pipeline()

pyproject.toml

@@ -1,3 +0,0 @@
-[tool.pytest.ini_options]
-pythonpath = ["."]
-testpaths = ["tests"]

requirements.txt

@@ -20,7 +20,6 @@ python-dotenv==1.2.2
 
 pandas
 numpy
-duckdb
 xgboost
 scikit-learn
 
@@ -80,4 +79,4 @@ lxml==6.1.0
 # ------------------------------------------------------------
 
 uvloop==0.22.1 ; sys_platform != "win32"
-huggingface_hub
+huggingface_hub

tests/manual_test.py

@@ -1,29 +0,0 @@
-def run_manual_test(prediction_date):
-
-    # 1. Generate today's feature rows
-    features = build_live_features(
-        as_of_date=prediction_date
-    )
-
-    # 2. Validate against model metadata
-    validate_features(features)
-
-    # 3. Run XGBoost
-    predictions = predict(features)
-
-    # 4. Rank
-    predictions = rank_predictions(predictions)
-
-    # 5. Store
-    save_predictions(predictions)
-
-    # 6. Resolve using historical future data
-    resolve_predictions(
-        as_of_date=prediction_date
-    )
-
-    # 7. Calculate metrics
-    calculate_metrics()
-
-    # 8. Check retraining condition
-    check_retrain()

tests/test_metrics.py

@@ -1,40 +0,0 @@
-from datetime import date
-import pytest
-from evaluation.metrics import (
-    precision_at_threshold, hit_rate, positive_rate,
-    average_return, binary_pr_auc, rank_deciles, calculate_metrics
-)
-
-def test_precision():
-    assert precision_at_threshold([.9,.8,.4,.2], [1,0,1,0], .5) == .5
-
-def test_rates():
-    assert hit_rate([1,0,1,0,0]) == .4
-    assert positive_rate([1,0,1,0,0]) == .4
-
-def test_average_return():
-    assert average_return([.03,.05,-.01]) == pytest.approx(.0233333333)
-
-def test_pr_auc():
-    assert binary_pr_auc([.9,.8,.2,.1], [1,1,0,0]) == pytest.approx(1.0)
-
-def test_deciles():
-    probs = list(reversed([i/100 for i in range(1,101)]))
-    deciles = rank_deciles(probs, [1]*100, [.03]*100)
-    assert len(deciles) == 10
-    assert all(d["count"] == 10 for d in deciles)
-
-def test_complete_metrics():
-    rows = [
-        {"prediction_date":date(2026,7,1),"symbol":"AAA","predicted_probability":.9,"actual_label":1,"actual_return":.05},
-        {"prediction_date":date(2026,7,1),"symbol":"BBB","predicted_probability":.4,"actual_label":0,"actual_return":-.02},
-        {"prediction_date":date(2026,7,2),"symbol":"AAA","predicted_probability":.8,"actual_label":1,"actual_return":.04},
-        {"prediction_date":date(2026,7,2),"symbol":"BBB","predicted_probability":.3,"actual_label":0,"actual_return":-.01},
-    ]
-    result = calculate_metrics(rows, threshold=.5)
-    assert result.prediction_days == 2
-    assert result.resolved_predictions == 4
-    assert result.precision == 1.0
-    assert result.hit_rate == .5
-    assert result.average_return == pytest.approx(.015)
-    assert len(result.deciles) == 10

tests/test_resolver.py

@@ -1,76 +0,0 @@
-from datetime import date
-
-# from evaluation.resolver import resolve_prediction_rows
-
-from jobs import resolver
-
-
-def test_positive_prediction():
-    result = resolver.resolve_prediction_rows(
-        prediction_date=date(2026, 8, 3),
-        symbol="ABC",
-        future_rows=[
-            {"date": date(2026, 8, 4), "open": 100.0, "close": 101.0},
-            {"date": date(2026, 8, 5), "open": 101.0, "close": 102.0},
-            {"date": date(2026, 8, 6), "open": 102.0, "close": 103.0},
-            {"date": date(2026, 8, 7), "open": 103.0, "close": 104.0},
-            {"date": date(2026, 8, 10), "open": 104.0, "close": 104.0},
-        ],
-    )
-
-    assert result is not None
-    assert result.entry_date == date(2026, 8, 4)
-    assert result.entry_open == 100.0
-    assert result.max_close == 104.0
-    assert result.actual_return == 0.04
-    assert result.actual_label == 1
-
-
-def test_exact_three_percent_is_positive():
-    result = resolver.resolve_prediction_rows(
-        prediction_date=date(2026, 8, 3),
-        symbol="ABC",
-        future_rows=[
-            {"date": date(2026, 8, 4), "open": 100.0, "close": 103.0},
-            {"date": date(2026, 8, 5), "open": 103.0, "close": 101.0},
-            {"date": date(2026, 8, 6), "open": 101.0, "close": 100.0},
-            {"date": date(2026, 8, 7), "open": 100.0, "close": 99.0},
-            {"date": date(2026, 8, 10), "open": 99.0, "close": 98.0},
-        ],
-    )
-
-    assert result is not None
-    assert result.actual_return == 0.03
-    assert result.actual_label == 1
-
-
-def test_below_three_percent_is_negative():
-    result = resolver.resolve_prediction_rows(
-        prediction_date=date(2026, 8, 3),
-        symbol="ABC",
-        future_rows=[
-            {"date": date(2026, 8, 4), "open": 100.0, "close": 102.99},
-            {"date": date(2026, 8, 5), "open": 102.0, "close": 102.5},
-            {"date": date(2026, 8, 6), "open": 102.0, "close": 101.0},
-            {"date": date(2026, 8, 7), "open": 101.0, "close": 100.0},
-            {"date": date(2026, 8, 10), "open": 100.0, "close": 99.0},
-        ],
-    )
-
-    assert result is not None
-    assert result.actual_return < 0.03
-    assert result.actual_label == 0
-
-
-def test_incomplete_window_stays_unresolved():
-    result = resolver.resolve_prediction_rows(
-        prediction_date=date(2026, 8, 3),
-        symbol="ABC",
-        future_rows=[
-            {"date": date(2026, 8, 4), "open": 100.0, "close": 105.0},
-            {"date": date(2026, 8, 5), "open": 105.0, "close": 105.0},
-            {"date": date(2026, 8, 6), "open": 105.0, "close": 105.0},
-        ],
-    )
-
-    assert result is None

tests/test_retrain_trigger.py

@@ -1,27 +0,0 @@
-import pytest
-from evaluation.retrain_trigger import (
-    calculate_deterioration, evaluate_retrain_trigger
-)
-
-def test_15_percent_deterioration_triggers():
-    result = evaluate_retrain_trigger(0.60, 0.51)
-    assert result.should_flag is True
-    assert result.deterioration == pytest.approx(0.15)
-
-def test_less_than_15_percent_deterioration_does_not_trigger():
-    result = evaluate_retrain_trigger(0.60, 0.52)
-    assert result.should_flag is False
-    assert result.deterioration == pytest.approx(0.1333333333)
-
-def test_improvement_does_not_trigger():
-    result = evaluate_retrain_trigger(0.60, 0.65)
-    assert result.should_flag is False
-    assert result.deterioration < 0
-
-def test_no_metric_does_not_trigger():
-    result = evaluate_retrain_trigger(0.60, None)
-    assert result.should_flag is False
-    assert result.deterioration is None
-
-def test_deterioration_formula():
-    assert calculate_deterioration(0.50, 0.425) == pytest.approx(0.15)