fix bugs

README.md

@@ -53,6 +53,19 @@ Once it's up:
 - API health check: <http://localhost:8000/health>
 - Metrics: <http://localhost:8000/metrics/twenty-day>
 
+### Prediction API and authorization
+
+`GET /predict?prediction_date=YYYY-MM-DD` is public and returns the stored
+predictions for that NSE trading day. Saturdays, Sundays, exchange holidays,
+and dates with no NSE candle return `{"status":"not_a_trading_day"}`. A
+prediction is available after the weekday scheduler has completed its run.
+
+`POST /predict` is the protected manual catch-up route; it accepts the same
+date parameter, incrementally ingests missing candles, rebuilds features, and
+writes predictions. All POST, PATCH, and DELETE routes require
+`X-Pipeline-Guid`; GET routes, including `/dashboard` and `GET /predict`, do
+not require it.
+
 On a first-ever run the ticker registry and `predictions` table are empty, so the dashboard will show "No predictions yet." until either the 16:00 IST scheduler fires or you trigger the pipeline manually:
 
 ```bash
@@ -2286,7 +2299,7 @@ Ticker source files are `static_data/ind_nifty_list.csv` and `static_data/us_sto
 
 All state-changing API routes require the `X-Pipeline-Guid` request header. Its value must equal `PIPELINE_ADMIN_GUID` in `.env`. Do not expose that value in browser code or source control.
 
-At 16:00 IST on weekdays, the scheduler runs the full lifecycle. It initializes the ticker registry from the CSV files only when empty, then incrementally upserts yfinance OHLCV into SQLite tables `market_tickers` and `market_ohlcv`. `GET /dashboard` presents the prediction and five-session outcome table.
+At 16:00 IST on weekdays, the scheduler runs market-data ingestion, feature rebuild, prediction, resolution, metrics, and retrain checks in that order. It incrementally upserts yfinance candles into the SQLite dataset tables, then `GET /dashboard` presents the resulting prediction and five-session outcome table.
 
 ### Feature-contract note
 
@@ -2299,8 +2312,8 @@ The production scheduler invokes the full lifecycle at **16:00 IST, Monday throu
 ### Incremental yfinance policy
 
 - A ticker with no stored OHLCV history is backfilled with up to 800 calendar days, providing sufficient warm-up history for the 200-day indicators.
-- A ticker already in `market_ohlcv` is refreshed from its latest stored trading date minus seven days. This overlap makes daily runs idempotent and allows yfinance corrections to be upserted without re-downloading the full history.
-- All returned candles are upserted into `market_ohlcv` by `(ticker_id, trading_date)`.
+- A populated ticker is fetched from its first missing calendar date only; daily runs do not re-download the 800-day warm-up history.
+- All returned candles are upserted by `(ticker_id, trading_date)` into their dataset-specific SQLite tables.
 - The market-data registry is imported from the CSV files only when it is empty, or when an administrator calls `POST /market/tickers/sync`. Daily runs preserve ticker additions, edits, deactivations, and deletions made through the protected API.
 
 ### Tables

api/dashboard.py

@@ -44,7 +44,7 @@ def _fetch_default_prediction(conn):
     with conn.cursor() as cur:
         cur.execute(f"""
             SELECT {PREDICTION_COLUMNS} FROM predictions
-            ORDER BY resolved_at IS NULL, resolved_at DESC, prediction_date DESC, rank ASC
+            ORDER BY prediction_date DESC, rank ASC
             LIMIT 1
         """)
         return cur.fetchone()

api/market.py

@@ -92,3 +92,4 @@ def ingest():
     with connect(settings.DATABASE_URL) as conn:
         rows = ingest_market_data(conn)
     return {"status": "ingested", "candles_upserted": rows}
+

app/main.py

@@ -28,18 +28,14 @@ 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.
+    # Startup seeds only the local ticker registry. Market-network fetching is
+    # deliberately performed by the guarded prediction request, never here.
     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
+    from jobs.market_ingestion import sync_static_tickers
 
     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
 
@@ -61,6 +57,36 @@ def health():
     }
 
 
+@app.get("/predict")
+def get_prediction(prediction_date: date | None = None):
+    """Return scheduler-produced predictions for a requested NSE trading day."""
+    target_date = prediction_date or pd.Timestamp.now(tz="Asia/Kolkata").date()
+    if target_date.weekday() >= 5:
+        return {"status": "not_a_trading_day", "date": target_date.isoformat()}
+
+    from app.config import settings
+    from database.connection import connect
+    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()}
+            cur.execute("""
+                SELECT symbol, predicted_probability, "rank", prediction_close
+                FROM predictions WHERE prediction_date = ? ORDER BY "rank"
+            """, (target_date,))
+            rows = cur.fetchall()
+
+    if not rows:
+        raise HTTPException(status_code=404, detail="Prediction has not been generated for this trading day yet")
+    return {
+        "status": "success",
+        "date": target_date.isoformat(),
+        "rows": len(rows),
+        "predictions": rows,
+    }
+
+
 @app.post("/predict", dependencies=[Depends(require_pipeline_guid)])
 def predict(prediction_date: date | None = None):
 
@@ -69,9 +95,26 @@ def predict(prediction_date: date | None = None):
         target_date = prediction_date or pd.Timestamp.now(tz="Asia/Kolkata").date()
         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
+
+        # Never call yfinance for weekends. This is intentionally before any
+        # ticker/database/network work.
+        if target_date.weekday() >= 5:
+            return {"status": "not_a_trading_day", "date": target_date.isoformat()}
+
+        today_ist = pd.Timestamp.now(tz="Asia/Kolkata").date()
+        if target_date > today_ist:
+            return {"status": "not_a_trading_day", "date": target_date.isoformat()}
+
+        # Protected manual catch-up: fetch only missing candles through the
+        # requested date, then rebuild the local SQLite feature/label tables.
+        with connect(settings.DATABASE_URL) as conn:
+            sync_static_tickers(conn, settings.STATIC_DATA_DIR)
+            ingest_market_data(conn, end_date=target_date)
+        rebuild_feature_stores()
 
-        # NSE candles are the trading-calendar authority.  This also handles
-        # exchange holidays without maintaining a second hard-coded calendar.
+        # NSE candles are the holiday-aware trading-calendar authority.
         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,))

jobs/daily_pipeline.py

@@ -5,9 +5,9 @@ Runs at 16:00 IST in this exact order:
 
     1. market-data and feature rebuild
     2. prediction
-    2. resolution
-    3. metrics
-    4. retrain trigger
+    3. resolution
+    4. metrics
+    5. retrain trigger
 
 Each step must succeed before the next step starts.
 
@@ -257,10 +257,3 @@ def run_pipeline() -> None:
         "============================================================"
     )
 
-
-# ============================================================
-# DIRECT EXECUTION
-# ============================================================
-
-if __name__ == "__main__":
-    run_pipeline()

jobs/market_ingestion.py

@@ -55,7 +55,9 @@ def sync_static_tickers(conn, static_dir: Path) -> int:
     return len(tickers)
 
 
-def ingest_market_data(conn, lookback_days: int = 800, overlap_days: int = 7) -> int:
+def ingest_market_data(conn, *, end_date: date | None = None, lookback_days: int = 800) -> int:
+    """Fetch each ticker only through the requested final calendar date."""
+    end_date = min(end_date or date.today(), date.today())
     rows = [ticker for ticker in list_tickers(conn) if ticker["active"]]
     latest_dates = latest_ohlcv_dates(conn)
     inserted = 0
@@ -63,8 +65,15 @@ def ingest_market_data(conn, lookback_days: int = 800, overlap_days: int = 7) ->
         latest_date = latest_dates.get(ticker["id"])
         # 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)
+        start = (latest_date + timedelta(days=1)) if latest_date else (end_date - timedelta(days=lookback_days))
+        # A restored database can contain candles dated after today's market
+        # session. yfinance rejects a start date after its end date.
+        if start > end_date:
+            continue
+        # yfinance's end date is exclusive, so include the requested day.
+        history = yf.Ticker(ticker["yfinance_symbol"]).history(
+            start=start.isoformat(), end=(end_date + timedelta(days=1)).isoformat(), auto_adjust=False,
+        )
         if history.empty:
             continue
         for timestamp, candle in history.iterrows():

jobs/rebuild_feature_stores.py

@@ -88,7 +88,7 @@ def rebuild_feature_stores() -> dict[str, int]:
 
             logger.info(
                 f"Tables BEFORE rebuild: "
-                f"{[row[0] for row in cur.fetchall()]}"
+                f"{[row['name'] for row in cur.fetchall()]}"
             )
 
         # -------------------------------------------------
@@ -262,7 +262,7 @@ def rebuild_feature_stores() -> dict[str, int]:
             """)
 
             tables = [
-                row[0]
+                row["name"]
                 for row in cur.fetchall()
             ]