Upload 39 files

app.py

@@ -12,22 +12,25 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi import FastAPI
 
 sys.path.insert(0, str(Path(__file__).resolve().parent))
-from nifty_backend.runtime import (
-    CLOSE_REFRESH_READY,
-    IST,
-    TPLUS1_READY,
-    close_refresh_due,
-    dashboard_payload,
-    is_trading_day,
-    latest_saved_prediction,
-    latest_tplus1_prediction,
-    next_trading_day,
-    refresh_daily_data,
-    refresh_first5_prediction,
-    refresh_market_close_data,
-    seconds_until_next_ist_run,
-    warm_dashboard_payload_cache,
-)
+from nifty_backend.runtime import (
+    CLOSE_REFRESH_READY,
+    IST,
+    STALE_CHECK_INTERVAL_SECONDS,
+    TPLUS1_READY,
+    close_refresh_due,
+    dashboard_payload,
+    is_trading_day,
+    latest_saved_prediction,
+    latest_tplus1_prediction,
+    next_trading_day,
+    refresh_daily_data,
+    refresh_first5_prediction,
+    refresh_market_close_data,
+    refresh_stale_data_once,
+    refresh_tplus1_prediction,
+    seconds_until_next_ist_run,
+    warm_dashboard_payload_cache,
+)
 
 
 app = FastAPI(title="NIFTY 50 Forecaster Backend")
@@ -40,14 +43,15 @@ app.add_middleware(
 )
 
 
-market_status = "Waiting for next session"
-close_refresh_lock = threading.Lock()
-MARKET_OPEN = time(9, 15)
-FIRST5_READY = time(9, 20)
-MARKET_CLOSE = time(15, 30)
+market_status = "Waiting for next session"
+close_refresh_lock = threading.Lock()
+tplus1_refresh_lock = threading.Lock()
+MARKET_OPEN = time(9, 15)
+FIRST5_READY = time(9, 20)
+MARKET_CLOSE = time(15, 30)
 
 
-def refresh_market_close_data_if_due() -> dict:
+def refresh_market_close_data_if_due() -> dict:
     if not close_refresh_due():
         return {"status": "skipped", "reason": "close refresh is not due"}
     if not close_refresh_lock.acquire(blocking=False):
@@ -55,8 +59,39 @@ def refresh_market_close_data_if_due() -> dict:
     try:
         info = refresh_market_close_data()
         return {"status": "refreshed", **info}
-    finally:
-        close_refresh_lock.release()
+    finally:
+        close_refresh_lock.release()
+
+
+def latest_tplus1_prediction_date(payload: dict | None = None) -> date | None:
+    try:
+        latest = payload if payload is not None else latest_tplus1_prediction()
+        raw = latest.get("input_date")
+        return date.fromisoformat(str(raw)[:10]) if raw else None
+    except Exception:
+        return None
+
+
+def tplus1_refresh_due(now: datetime | None = None, latest_date: date | None = None) -> bool:
+    now = now or datetime.now(IST)
+    if not is_trading_day(now.date()) or not (TPLUS1_READY <= now.time() < MARKET_CLOSE):
+        return False
+    latest_date = latest_date if latest_date is not None else latest_tplus1_prediction_date()
+    return latest_date != now.date()
+
+
+def refresh_tplus1_if_due() -> dict:
+    now = datetime.now(IST)
+    latest_date = latest_tplus1_prediction_date()
+    if not tplus1_refresh_due(now=now, latest_date=latest_date):
+        return {"status": "skipped", "reason": "tplus1 refresh is not due"}
+    if not tplus1_refresh_lock.acquire(blocking=False):
+        return {"status": "skipped", "reason": "tplus1 refresh already running"}
+    try:
+        prediction = refresh_tplus1_prediction(session_date=now.date())
+        return {"status": "refreshed", "prediction": prediction}
+    finally:
+        tplus1_refresh_lock.release()
 
 
 def latest_prediction_date(payload: dict | None = None) -> date | None:
@@ -68,7 +103,7 @@ def latest_prediction_date(payload: dict | None = None) -> date | None:
         return None
 
 
-def current_market_state(now: datetime | None = None) -> dict:
+def current_market_state(now: datetime | None = None) -> dict:
     global market_status
     now = now or datetime.now(IST)
     today = now.date()
@@ -78,11 +113,8 @@ def current_market_state(now: datetime | None = None) -> dict:
     market_is_open_for_t5 = trading_day and FIRST5_READY <= current_time < MARKET_CLOSE
     market_is_open_for_tplus1 = trading_day and TPLUS1_READY <= current_time < MARKET_CLOSE
     has_current_first5 = market_is_open_for_t5 and latest_date == today
-    try:
-        tplus1_latest_date = date.fromisoformat(str(latest_tplus1_prediction().get("input_date"))[:10])
-    except Exception:
-        tplus1_latest_date = None
-    has_current_tplus1 = market_is_open_for_tplus1 and tplus1_latest_date == today
+    tplus1_latest_date = latest_tplus1_prediction_date()
+    has_current_tplus1 = market_is_open_for_tplus1 and tplus1_latest_date == today
     next_session = today if trading_day and current_time < MARKET_CLOSE else next_trading_day(today + timedelta(days=1))
 
     if not trading_day:
@@ -107,11 +139,48 @@ def current_market_state(now: datetime | None = None) -> dict:
         else:
             status = "Prediction Pending"
             detail = "No current-session prediction has been generated yet."
-    else:
-        status = "Market Closed"
-        detail = "Trading session has ended."
-
-    return {
+    else:
+        status = "Market Closed"
+        detail = "Trading session has ended."
+
+    if not trading_day:
+        tplus1_status = "Market Closed"
+        tplus1_detail = f"Next trading session is {next_session.isoformat()}."
+    elif current_time < TPLUS1_READY:
+        tplus1_status = "Waiting for 2:30 PM"
+        tplus1_detail = "The T+1 forecast becomes available at 2:30 PM IST."
+    elif current_time < MARKET_CLOSE:
+        if has_current_tplus1:
+            tplus1_status = "Ready"
+            tplus1_detail = "Today's T+1 prediction is available."
+        else:
+            tplus1_status = "Pending"
+            tplus1_detail = "No current-session T+1 prediction has been generated yet."
+    else:
+        tplus1_status = "Market Closed"
+        tplus1_detail = "Trading session has ended."
+
+    if not trading_day:
+        t5_status = "Market Closed"
+        t5_detail = f"Next trading session is {next_session.isoformat()}."
+    elif current_time < FIRST5_READY:
+        t5_status = "Waiting for 9:20 AM"
+        t5_detail = "The T+5 forecast becomes available after the first five one-minute bars."
+    elif current_time < MARKET_CLOSE:
+        if market_status in {"Fetching T+5 Prediction Data...", "Prediction Failed"}:
+            t5_status = market_status
+            t5_detail = "The first-five-minute prediction job is still resolving."
+        elif has_current_first5:
+            t5_status = "Ready"
+            t5_detail = "Today's first-five-minute prediction is available."
+        else:
+            t5_status = "Pending"
+            t5_detail = "No current-session prediction has been generated yet."
+    else:
+        t5_status = "Market Closed"
+        t5_detail = "Trading session has ended."
+
+    return {
         "market_status": status,
         "market_detail": detail,
         "server_time_ist": now.isoformat(),
@@ -119,12 +188,16 @@ def current_market_state(now: datetime | None = None) -> dict:
         "session_date": today.isoformat(),
         "next_session_date": next_session.isoformat(),
         "latest_prediction_date": latest_date.isoformat() if latest_date else None,
-        "t5_available": has_current_first5,
-        "market_is_open_for_t5": market_is_open_for_t5,
-        "tplus1_available": has_current_tplus1,
-        "market_is_open_for_tplus1": market_is_open_for_tplus1,
-        "latest_tplus1_prediction_date": tplus1_latest_date.isoformat() if tplus1_latest_date else None,
-    }
+        "t5_available": has_current_first5,
+        "t5_status": t5_status,
+        "t5_detail": t5_detail,
+        "market_is_open_for_t5": market_is_open_for_t5,
+        "tplus1_available": has_current_tplus1,
+        "tplus1_status": tplus1_status,
+        "tplus1_detail": tplus1_detail,
+        "market_is_open_for_tplus1": market_is_open_for_tplus1,
+        "latest_tplus1_prediction_date": tplus1_latest_date.isoformat() if tplus1_latest_date else None,
+    }
 
 
 def attach_market_state(payload: dict) -> dict:
@@ -137,9 +210,7 @@ def attach_market_state(payload: dict) -> dict:
     tplus1_latest = payload.get("tplus1_latest") or {}
     t5_available = bool(state["t5_available"] and t5_latest.get("prediction"))
     tplus1_available = bool(state["tplus1_available"] and tplus1_latest.get("prediction"))
-    market_closed = state["market_status"] == "Market Closed"
-    unavailable_reason = "Market Closed" if market_closed else state["market_status"]
-    tomorrow_available = bool(tomorrow_latest.get("prediction"))
+    tomorrow_available = bool(tomorrow_latest.get("prediction"))
     refresh_phase = payload.get("data_status", {}).get("refresh_phase")
     if refresh_phase in {"waiting_second_payload", "refreshing"}:
         tomorrow_status = "WAITING FOR SECOND PAYLOAD"
@@ -163,12 +234,12 @@ def attach_market_state(payload: dict) -> dict:
             "validation_accuracy": tomorrow_latest.get("validation_accuracy"),
             "test_accuracy": tomorrow_latest.get("test_accuracy"),
         },
-        "t5": {
-            "available": t5_available,
-            "status": "Ready" if t5_available else unavailable_reason,
-            "reason": None if t5_available else state["market_detail"],
-            "input_date": t5_latest.get("input_date"),
-            "prediction": t5_latest.get("prediction") if t5_available else None,
+        "t5": {
+            "available": t5_available,
+            "status": "Ready" if t5_available else state["t5_status"],
+            "reason": None if t5_available else state["t5_detail"],
+            "input_date": t5_latest.get("input_date"),
+            "prediction": t5_latest.get("prediction") if t5_available else None,
             "prob_up": t5_latest.get("prob_up") if t5_available else None,
             "confidence": t5_latest.get("confidence") if t5_available else None,
             "threshold": t5_latest.get("threshold") if t5_available else None,
@@ -176,12 +247,12 @@ def attach_market_state(payload: dict) -> dict:
             "validation_accuracy": (payload.get("summary") or {}).get("validation_accuracy"),
             "test_accuracy": (payload.get("summary") or {}).get("test_accuracy"),
         },
-        "tplus1": {
-            "available": tplus1_available,
-            "status": "Ready" if tplus1_available else unavailable_reason,
-            "reason": None if tplus1_available else state["market_detail"],
-            "target_date": tplus1_latest.get("target_date") or state["next_session_date"],
-            "input_date": tplus1_latest.get("input_date"),
+        "tplus1": {
+            "available": tplus1_available,
+            "status": "Ready" if tplus1_available else state["tplus1_status"],
+            "reason": None if tplus1_available else state["tplus1_detail"],
+            "target_date": tplus1_latest.get("target_date") or state["next_session_date"],
+            "input_date": tplus1_latest.get("input_date"),
             "prediction": tplus1_latest.get("prediction") if tplus1_available else None,
             "prob_up": tplus1_latest.get("prob_up") if tplus1_available else None,
             "confidence": tplus1_latest.get("confidence") if tplus1_available else None,
@@ -194,7 +265,7 @@ def attach_market_state(payload: dict) -> dict:
     return payload
 
 
-async def daily_ist_refresh_loop() -> None:
+async def daily_ist_refresh_loop() -> None:
     global market_status
     while True:
         # Wait until 9:00 AM IST
@@ -222,14 +293,22 @@ async def daily_ist_refresh_loop() -> None:
             print(f"[scheduler] first5 refresh failed: {exc}", flush=True)
             market_status = "Prediction Failed"
             
-        try:
-            await asyncio.to_thread(refresh_daily_data)
-        except Exception as exc:
-            print(f"[scheduler] daily refresh failed: {exc}", flush=True)
-
-        await asyncio.sleep(seconds_until_next_ist_run(CLOSE_REFRESH_READY))
-        print("[scheduler] 3:45 PM IST - Refreshing close data", flush=True)
-        try:
+        try:
+            await asyncio.to_thread(refresh_daily_data)
+        except Exception as exc:
+            print(f"[scheduler] daily refresh failed: {exc}", flush=True)
+
+        await asyncio.sleep(seconds_until_next_ist_run(TPLUS1_READY))
+        print("[scheduler] 2:30 PM IST - Refreshing T+1 prediction", flush=True)
+        try:
+            info = await asyncio.to_thread(refresh_tplus1_if_due)
+            print(f"[scheduler] tplus1 refresh result: {info}", flush=True)
+        except Exception as exc:
+            print(f"[scheduler] tplus1 refresh failed: {exc}", flush=True)
+
+        await asyncio.sleep(seconds_until_next_ist_run(CLOSE_REFRESH_READY))
+        print("[scheduler] 3:45 PM IST - Refreshing close data", flush=True)
+        try:
             info = await asyncio.to_thread(refresh_market_close_data_if_due)
             print(f"[scheduler] close refresh result: {info}", flush=True)
         except Exception as exc:
@@ -257,24 +336,44 @@ async def refresh_current_session_once() -> None:
         print(f"[startup] daily refresh failed: {exc}", flush=True)
 
 
-async def refresh_market_close_once_if_due() -> None:
+async def refresh_market_close_once_if_due() -> None:
     try:
         info = await asyncio.to_thread(refresh_market_close_data_if_due)
         if info.get("status") == "refreshed":
             print(f"[startup] close refresh result: {info}", flush=True)
     except Exception as exc:
-        print(f"[startup] close refresh failed: {exc}", flush=True)
-
-
-async def warm_dashboard_payload_cache_once() -> None:
-    try:
-        await asyncio.to_thread(warm_dashboard_payload_cache)
-    except Exception as exc:
-        print(f"[startup] dashboard payload warmup failed: {exc}", flush=True)
-
-
-@app.on_event("startup")
-async def start_scheduler() -> None:
+        print(f"[startup] close refresh failed: {exc}", flush=True)
+
+
+async def refresh_tplus1_once_if_due() -> None:
+    try:
+        info = await asyncio.to_thread(refresh_tplus1_if_due)
+        if info.get("status") == "refreshed":
+            print(f"[startup] tplus1 refresh result: {info}", flush=True)
+    except Exception as exc:
+        print(f"[startup] tplus1 refresh failed: {exc}", flush=True)
+
+
+async def warm_dashboard_payload_cache_once() -> None:
+    try:
+        await asyncio.to_thread(warm_dashboard_payload_cache)
+    except Exception as exc:
+        print(f"[startup] dashboard payload warmup failed: {exc}", flush=True)
+
+
+async def stale_data_watch_loop() -> None:
+    while True:
+        try:
+            info = await asyncio.to_thread(refresh_stale_data_once)
+            if info.get("status") == "refreshed":
+                print(f"[stale-watch] refreshed stale data: {info}", flush=True)
+        except Exception as exc:
+            print(f"[stale-watch] stale refresh failed: {exc}", flush=True)
+        await asyncio.sleep(STALE_CHECK_INTERVAL_SECONDS)
+
+
+@app.on_event("startup")
+async def start_scheduler() -> None:
     global market_status
     # Initialize correct status on startup based on current time
     now = datetime.now(IST).time()
@@ -292,10 +391,12 @@ async def start_scheduler() -> None:
     else:
         market_status = "Prediction Pending"
         
-    asyncio.create_task(refresh_current_session_once())
-    asyncio.create_task(refresh_market_close_once_if_due())
-    asyncio.create_task(warm_dashboard_payload_cache_once())
-    asyncio.create_task(daily_ist_refresh_loop())
+    asyncio.create_task(refresh_current_session_once())
+    asyncio.create_task(refresh_tplus1_once_if_due())
+    asyncio.create_task(refresh_market_close_once_if_due())
+    asyncio.create_task(warm_dashboard_payload_cache_once())
+    asyncio.create_task(stale_data_watch_loop())
+    asyncio.create_task(daily_ist_refresh_loop())
 
 
 @app.get("/health")
@@ -308,18 +409,27 @@ def root() -> dict[str, str]:
     return {"service": "NIFTY 50 Forecaster Backend", "status": "ok"}
 
 
-@app.get("/dashboard")
-def dashboard() -> dict:
-    return attach_market_state(dashboard_payload())
+@app.get("/dashboard")
+def dashboard() -> dict:
+    return attach_market_state(dashboard_payload())
 
 
 @app.get("/cron/keepalive")
-def cron_keepalive(background_tasks: BackgroundTasks) -> dict:
-    close_refresh = {"status": "not_checked"}
-    if close_refresh_due():
-        background_tasks.add_task(refresh_market_close_data_if_due)
-        close_refresh = {"status": "scheduled"}
-    return {"status": "awake", "market": current_market_state(), "close_refresh": close_refresh}
+def cron_keepalive(background_tasks: BackgroundTasks) -> dict:
+    close_refresh = {"status": "not_checked"}
+    tplus1_refresh = {"status": "not_checked"}
+    if tplus1_refresh_due():
+        background_tasks.add_task(refresh_tplus1_if_due)
+        tplus1_refresh = {"status": "scheduled"}
+    if close_refresh_due():
+        background_tasks.add_task(refresh_market_close_data_if_due)
+        close_refresh = {"status": "scheduled"}
+    return {
+        "status": "awake",
+        "market": current_market_state(),
+        "tplus1_refresh": tplus1_refresh,
+        "close_refresh": close_refresh,
+    }
 
 
 @app.get("/prediction/latest")

data/nifty50_1d.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:b57d445258fc7ff258e04869c7b233c2d8e483db82d93917a8c28f16b5ee0d19
-size 78241
+oid sha256:be744722b6c72c2fade81cc25551e32edcbda4737d02e6bc6ff8f0dff4b31d90
+size 78275

data/nifty50_1m.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:eff0ea13a459412466def2b530b482e84e06346ee4f7203519689baf0adce32c
-size 18555782
+oid sha256:216816fb4cb1b022029e3e1ab88b344e2c6dcfb65a51d2e1b70ab76dc3320a45
+size 18580743

data/opening_direction_training_dataset.parquet

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:c21f3c5e7781b67e9849a6d07c9ec28d3f087b74be7e2472785124ceb35a34f4
-size 4462258
+oid sha256:5293909811782086d6c16ea35e8b4313dbcea6c928e6ffc07b342502dc94f466
+size 4463019

models/latest_prediction.csv

@@ -1,2 +1,2 @@
-input_date,first5_start,first5_end,prediction,prob_up,confidence,threshold,model_name
-2026-05-21,2026-05-21 09:15:00,2026-05-21 09:19:00,UP,0.4379885393062092,0.5129885393062092,0.425,blend_extra_trees_tight_logit_overlay
+input_date,first5_start,first5_end,prediction,prob_up,confidence,threshold,model_name,is_overridden
+2026-05-26,2026-05-26 09:15:00,2026-05-26 09:19:00,DOWN,0.6058803888947725,0.6808803888947725,0.425,blend_extra_trees_tight_logit_overlay,True

models/tomorrow_latest_prediction.csv

@@ -1,2 +1,2 @@
 input_date,target_date,prediction,prob_up,confidence,threshold,model_name,source_model,validation_accuracy,test_accuracy
-2026-05-25,2026-05-26,UP,0.5649926672304149,0.5649926672304149,0.543,nifty_tomorrow_direction_model,tuned_daily_forest_single,0.5780141843971631,0.6182795698924731
+2026-05-26,2026-05-27,DOWN,0.4892079705003583,0.5107920294996418,0.543,nifty_tomorrow_direction_model,tuned_daily_forest_single,0.5780141843971631,0.6182795698924731

models/tomorrow_summary.json

@@ -25,13 +25,13 @@
   "valid_end": "2025-08-17",
   "test_start": "2025-08-18",
   "test_end": "2026-05-20",
-  "latest_forecast_date": "2026-05-25",
-  "latest_forecast_for": "next trading session 2026-05-26",
-  "latest_forecast_prob_up": 0.5649926672304149,
-  "latest_forecast_signal": "UP",
+  "latest_forecast_date": "2026-05-26",
+  "latest_forecast_for": "next trading session 2026-05-27",
+  "latest_forecast_prob_up": 0.4892079705003583,
+  "latest_forecast_signal": "DOWN",
   "feature_count": 301,
   "model_name": "nifty_tomorrow_direction_model",
   "source_model": "tuned_daily_forest_single",
   "target": "next trading session NIFTY 50 direction",
-  "latest_target_date": "2026-05-26"
+  "latest_target_date": "2026-05-27"
 }

models/tplus1_latest_prediction.csv

@@ -1,2 +1,2 @@
 input_date,target_date,forecast_for,prediction,prob_up,confidence,threshold,model_name,decision_overlay,validation_accuracy,test_accuracy,accuracy_goal
-2026-05-21,2026-05-22,next trading session after 2026-05-21,UP,0.620604654276822,0.620604654276822,0.578,logistic_regression_l1_C0.35_balanced,prev_target_mean10_le_0.4_up;m02_range_1m_ge_0.000479116_up,0.66,0.6368421052631579,0.63
+2026-05-26,2026-05-27,next trading session after 2026-05-26,UP,0.48291233826421653,0.5170876617357835,0.578,logistic_regression_l1_C0.35_balanced,prev_target_mean10_le_0.4_up;m02_range_1m_ge_0.000479116_up,0.66,0.6368421052631579,0.63

nifty_backend/runtime.py

@@ -11,10 +11,10 @@ from pathlib import Path
 from typing import Any
 from zoneinfo import ZoneInfo
 
-import joblib
-import numpy as np
-import pandas as pd
-import yfinance as yf
+import joblib
+import numpy as np
+import pandas as pd
+from nifty_backend.yahoo_history_client import YahooHistoryClient
 
 try:
     import pandas_market_calendars as mcal
@@ -23,13 +23,16 @@ except ImportError:  # pragma: no cover - production dependency, local fallback
 
 
 IST = ZoneInfo("Asia/Kolkata")
-YAHOO_NIFTY_SYMBOL = "^NSEI"
-MARKET_CLOSE = time(15, 30)
-CLOSE_REFRESH_READY = time(15, 45)
-TPLUS1_READY = time(14, 30)
-BACKEND_ROOT = Path(__file__).resolve().parents[1]
-DATA_DIR = BACKEND_ROOT / "data"
-MODEL_DIR = BACKEND_ROOT / "models"
+YAHOO_NIFTY_SYMBOL = "^NSEI"
+MARKET_CLOSE = time(15, 30)
+FIRST5_READY = time(9, 20)
+CLOSE_REFRESH_READY = time(15, 45)
+TPLUS1_READY = time(14, 30)
+STALE_CHECK_INTERVAL_SECONDS = 5
+BACKEND_ROOT = Path(__file__).resolve().parents[1]
+DATA_DIR = BACKEND_ROOT / "data"
+MODEL_DIR = BACKEND_ROOT / "models"
+YAHOO_CACHE_PATH = MODEL_DIR / "yahoo_history_cache.sqlite3"
 OPENING_DATASET_PATH = DATA_DIR / "opening_direction_training_dataset.parquet"
 NIFTY_1M_PATH = DATA_DIR / "nifty50_1m.parquet"
 NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet"
@@ -66,7 +69,8 @@ DECISION_OVERLAYS = [
     },
 ]
 
-_dashboard_payload_lock = threading.Lock()
+_dashboard_payload_lock = threading.Lock()
+_stale_refresh_lock = threading.Lock()
 
 
 def utc_now_iso() -> str:
@@ -262,7 +266,7 @@ def read_training_dataset() -> pd.DataFrame:
     return df.sort_values("date").reset_index(drop=True)
 
 
-def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame:
+def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame:
     if df.empty:
         return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"])
     if isinstance(df.columns, pd.MultiIndex):
@@ -288,19 +292,81 @@ def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame:
     for src, dst in rename.items():
         if src in df.columns and dst not in out.columns:
             out[dst] = pd.to_numeric(df[src], errors="coerce")
-    return out.dropna(subset=["date", "open", "high", "low", "close"]).sort_values("date")
-
-
-def fetch_yahoo_minutes(period: str = "5d") -> pd.DataFrame:
-    raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1m", progress=False, prepost=False, auto_adjust=False)
-    return normalize_yahoo_frame(raw)
-
-
-def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
-    raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1d", progress=False, prepost=False, auto_adjust=False)
-    out = normalize_yahoo_frame(raw)
-    out["date"] = pd.to_datetime(out["date"], errors="coerce").dt.normalize()
-    return out.drop_duplicates("date", keep="last")
+    return out.dropna(subset=["date", "open", "high", "low", "close"]).sort_values("date")
+
+
+@lru_cache(maxsize=1)
+def yahoo_history_client() -> YahooHistoryClient:
+    return YahooHistoryClient(cache_path=YAHOO_CACHE_PATH)
+
+
+def period_start(period: str, *, end: datetime) -> datetime:
+    text = str(period).strip().lower()
+    units = {
+        "d": "days",
+        "wk": "weeks",
+        "mo": "months",
+        "y": "years",
+    }
+    for suffix, unit in units.items():
+        if text.endswith(suffix):
+            raw_value = text[: -len(suffix)]
+            if not raw_value.isdigit():
+                break
+            value = int(raw_value)
+            if unit == "days":
+                return end - timedelta(days=value)
+            if unit == "weeks":
+                return end - timedelta(weeks=value)
+            if unit == "months":
+                return end - timedelta(days=value * 31)
+            if unit == "years":
+                return end - timedelta(days=value * 366)
+    raise ValueError(f"Unsupported Yahoo period: {period!r}")
+
+
+def yahoo_history_to_ohlcv(frame: pd.DataFrame, *, daily: bool) -> pd.DataFrame:
+    if frame.empty:
+        return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"])
+    out = frame.rename(columns={"timestamp": "date"}).copy()
+    out["date"] = pd.to_datetime(out["date"], errors="coerce")
+    if daily:
+        out["date"] = out["date"].dt.normalize()
+    for column in ("open", "high", "low", "close", "volume"):
+        out[column] = pd.to_numeric(out[column], errors="coerce")
+    return (
+        out[["date", "open", "high", "low", "close", "volume"]]
+        .dropna(subset=["date", "open", "high", "low", "close"])
+        .drop_duplicates("date", keep="last")
+        .sort_values("date")
+        .reset_index(drop=True)
+    )
+
+
+def fetch_yahoo_minutes(period: str = "5d") -> pd.DataFrame:
+    end = datetime.now(IST).replace(tzinfo=None) + timedelta(minutes=5)
+    start = period_start(period, end=end)
+    raw = yahoo_history_client().fetch_history(
+        YAHOO_NIFTY_SYMBOL,
+        interval="1m",
+        start=start,
+        end=end,
+        include_prepost=False,
+    )
+    return yahoo_history_to_ohlcv(raw, daily=False)
+
+
+def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
+    end = datetime.now(IST).replace(tzinfo=None) + timedelta(days=1)
+    start = period_start(period, end=end)
+    raw = yahoo_history_client().fetch_history(
+        YAHOO_NIFTY_SYMBOL,
+        interval="1d",
+        start=start,
+        end=end,
+        include_prepost=False,
+    )
+    return yahoo_history_to_ohlcv(raw, daily=True)
 
 
 def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
@@ -675,19 +741,27 @@ def _apply_tplus1_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list
     return adjusted
 
 
-def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any]:
-    if not TPLUS1_MODEL_PATH.exists():
-        raise FileNotFoundError(f"Missing T+1 model artifact: {TPLUS1_MODEL_PATH}")
-    payload = joblib.load(TPLUS1_MODEL_PATH)
-    features = payload["features"]
-    threshold = float(payload["threshold"])
-    frame = _add_tplus1_target_features(_build_tplus1_session_features(_minute_frame_for_tplus1()))
-    if session_date is not None:
-        row = frame[pd.to_datetime(frame["date"], errors="coerce").dt.date == session_date].tail(1)
-    else:
-        row = frame.tail(1)
-    if row.empty:
-        raise RuntimeError("No complete 14:00-14:20 window is available for T+1 prediction.")
+def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any]:
+    if not TPLUS1_MODEL_PATH.exists():
+        raise FileNotFoundError(f"Missing T+1 model artifact: {TPLUS1_MODEL_PATH}")
+    payload = joblib.load(TPLUS1_MODEL_PATH)
+    features = payload["features"]
+    threshold = float(payload["threshold"])
+    frame = _add_tplus1_target_features(_build_tplus1_session_features(_minute_frame_for_tplus1()))
+    if session_date is not None:
+        row = frame[pd.to_datetime(frame["date"], errors="coerce").dt.date == session_date].tail(1)
+    else:
+        row = frame.tail(1)
+    if row.empty:
+        minutes = fetch_yahoo_minutes(period="7d")
+        append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
+        frame = _add_tplus1_target_features(_build_tplus1_session_features(_minute_frame_for_tplus1()))
+        if session_date is not None:
+            row = frame[pd.to_datetime(frame["date"], errors="coerce").dt.date == session_date].tail(1)
+        else:
+            row = frame.tail(1)
+    if row.empty:
+        raise RuntimeError("No complete 14:00-14:20 window is available for T+1 prediction.")
     missing = [col for col in features if col not in row.columns]
     if missing:
         raise RuntimeError(f"T+1 feature row is missing model features: {missing[:5]}")
@@ -1103,7 +1177,7 @@ def refresh_market_close_data(session_date: date | None = None) -> dict[str, Any
         raise
 
 
-def close_refresh_due(now: datetime | None = None) -> bool:
+def close_refresh_due(now: datetime | None = None) -> bool:
     now = now or datetime.now(IST)
     if not is_trading_day(now.date()) or now.time() < CLOSE_REFRESH_READY:
         return False
@@ -1118,13 +1192,123 @@ def close_refresh_due(now: datetime | None = None) -> bool:
             tomorrow_input = date.fromisoformat(str(tomorrow_latest.get("input_date"))[:10])
     except Exception:
         tomorrow_input = None
-    return any(
-        latest != now.date()
-        for latest in (latest_daily, latest_minutes, latest_opening, latest_opening_outcome, tomorrow_input)
-    )
-
-
-def next_ist_run_at(run_time: time = time(9, 20), now: datetime | None = None) -> datetime:
+    return any(
+        latest != now.date()
+        for latest in (latest_daily, latest_minutes, latest_opening, latest_opening_outcome, tomorrow_input)
+    )
+
+
+def latest_prediction_input_date(path: Path) -> date | None:
+    if not path.exists():
+        return None
+    try:
+        frame = pd.read_csv(path, usecols=["input_date"])
+    except Exception:
+        return None
+    if frame.empty:
+        return None
+    value = pd.to_datetime(frame["input_date"], errors="coerce").max()
+    return None if pd.isna(value) else value.date()
+
+
+def expected_completed_daily_date(now: datetime | None = None) -> date:
+    now = now or datetime.now(IST)
+    if is_trading_day(now.date()) and now.time() < CLOSE_REFRESH_READY:
+        return previous_trading_day(now.date() - timedelta(days=1))
+    return previous_trading_day(now.date())
+
+
+def expected_minute_date(now: datetime | None = None) -> date:
+    now = now or datetime.now(IST)
+    if is_trading_day(now.date()) and now.time() >= FIRST5_READY:
+        return now.date()
+    return previous_trading_day(now.date() - timedelta(days=1))
+
+
+def expected_tplus1_date(now: datetime | None = None) -> date:
+    now = now or datetime.now(IST)
+    if is_trading_day(now.date()) and now.time() >= TPLUS1_READY:
+        return now.date()
+    return previous_trading_day(now.date() - timedelta(days=1))
+
+
+def is_stale(latest: date | None, expected: date) -> bool:
+    return latest is None or latest < expected
+
+
+def stale_data_status(now: datetime | None = None) -> dict[str, Any]:
+    now = now or datetime.now(IST)
+    expected_daily = expected_completed_daily_date(now)
+    expected_minutes = expected_minute_date(now)
+    expected_tplus1 = expected_tplus1_date(now)
+    latest_daily = latest_parquet_date(NIFTY_1D_PATH)
+    latest_minutes = latest_parquet_date(NIFTY_1M_PATH)
+    latest_t5 = latest_prediction_input_date(LATEST_PATH)
+    latest_tplus1 = latest_prediction_input_date(TPLUS1_LATEST_PATH)
+    return {
+        "server_time_ist": now.isoformat(),
+        "expected_daily_date": expected_daily.isoformat(),
+        "expected_minute_date": expected_minutes.isoformat(),
+        "expected_tplus1_date": expected_tplus1.isoformat(),
+        "latest_daily_date": latest_daily.isoformat() if latest_daily else None,
+        "latest_minute_date": latest_minutes.isoformat() if latest_minutes else None,
+        "latest_t5_date": latest_t5.isoformat() if latest_t5 else None,
+        "latest_tplus1_date": latest_tplus1.isoformat() if latest_tplus1 else None,
+        "daily_stale": is_stale(latest_daily, expected_daily),
+        "minutes_stale": is_stale(latest_minutes, expected_minutes),
+        "t5_stale": is_stale(latest_t5, expected_minutes),
+        "tplus1_stale": is_stale(latest_tplus1, expected_tplus1),
+    }
+
+
+def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]:
+    now = now or datetime.now(IST)
+    status = stale_data_status(now)
+    if not any(status[key] for key in ("daily_stale", "minutes_stale", "t5_stale", "tplus1_stale")):
+        return {"status": "fresh", **status, "actions": []}
+    if not _stale_refresh_lock.acquire(blocking=False):
+        return {"status": "skipped", "reason": "stale refresh already running", **status, "actions": []}
+
+    actions: list[dict[str, Any]] = []
+    try:
+        if status["minutes_stale"]:
+            minutes = fetch_yahoo_minutes(period="7d")
+            combined = append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
+            actions.append(
+                {
+                    "name": "minutes",
+                    "rows": int(len(combined)),
+                    "latest_date": pd.to_datetime(combined["date"], errors="coerce").max().date().isoformat(),
+                }
+            )
+
+        if status["daily_stale"]:
+            daily_info = refresh_daily_data()
+            outcomes = update_opening_outcomes_from_daily()
+            actions.append({"name": "daily", **daily_info})
+            actions.append({"name": "opening_outcomes", **outcomes})
+            try:
+                tomorrow = refresh_tomorrow_prediction(session_date=date.fromisoformat(status["expected_daily_date"]))
+                actions.append({"name": "tomorrow_prediction", "input_date": tomorrow.get("input_date")})
+            except Exception as exc:
+                actions.append({"name": "tomorrow_prediction", "error": str(exc)})
+
+        if status["t5_stale"] and is_trading_day(now.date()) and now.time() >= FIRST5_READY:
+            prediction = refresh_first5_prediction(session_date=now.date())
+            actions.append({"name": "t5_prediction", "input_date": prediction.input_date})
+
+        if status["tplus1_stale"] and is_trading_day(now.date()) and now.time() >= TPLUS1_READY:
+            prediction = refresh_tplus1_prediction(session_date=now.date())
+            actions.append({"name": "tplus1_prediction", "input_date": prediction.get("input_date")})
+
+        clear_dashboard_payload_cache()
+        refreshed_status = stale_data_status(datetime.now(IST))
+        return {"status": "refreshed", **refreshed_status, "actions": actions}
+    finally:
+        _stale_refresh_lock.release()
+
+
+def next_ist_run_at(run_time: time = time(9, 20), now: datetime | None = None) -> datetime:
     now = now or datetime.now(IST)
     target_day = now.date()
     if now >= datetime.combine(target_day, run_time, tzinfo=IST):

nifty_backend/yahoo_history_client.py

@@ -0,0 +1,445 @@
+from __future__ import annotations
+
+import gzip
+import hashlib
+import json
+import sqlite3
+import threading
+import time
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from itertools import cycle
+from pathlib import Path
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import pandas as pd
+import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry
+
+
+YAHOO_CHART_HOSTS = (
+    "https://query1.finance.yahoo.com",
+    "https://query2.finance.yahoo.com",
+)
+DEFAULT_HEADERS = {
+    "User-Agent": (
+        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+        "AppleWebKit/537.36 (KHTML, like Gecko) "
+        "Chrome/136.0 Safari/537.36"
+    ),
+    "Accept": "application/json,text/plain,*/*",
+    "Accept-Language": "en-US,en;q=0.9",
+    "Connection": "keep-alive",
+    "Origin": "https://finance.yahoo.com",
+    "Referer": "https://finance.yahoo.com/",
+}
+BAR_COLUMNS = ["timestamp", "open", "high", "low", "close", "adj_close", "volume", "dividend", "split_ratio"]
+
+
+@dataclass(frozen=True)
+class IntervalPolicy:
+    interval: str
+    chunk_days: int
+    min_chunk_days: int
+    retention_days: int | None
+
+
+INTERVAL_POLICIES: dict[str, IntervalPolicy] = {
+    "1m": IntervalPolicy("1m", chunk_days=7, min_chunk_days=1, retention_days=30),
+    "2m": IntervalPolicy("2m", chunk_days=60, min_chunk_days=2, retention_days=60),
+    "5m": IntervalPolicy("5m", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "15m": IntervalPolicy("15m", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "30m": IntervalPolicy("30m", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "60m": IntervalPolicy("60m", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "90m": IntervalPolicy("90m", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "1h": IntervalPolicy("1h", chunk_days=60, min_chunk_days=5, retention_days=60),
+    "1d": IntervalPolicy("1d", chunk_days=3650, min_chunk_days=30, retention_days=None),
+    "5d": IntervalPolicy("5d", chunk_days=3650, min_chunk_days=30, retention_days=None),
+    "1wk": IntervalPolicy("1wk", chunk_days=3650, min_chunk_days=30, retention_days=None),
+    "1mo": IntervalPolicy("1mo", chunk_days=3650, min_chunk_days=30, retention_days=None),
+    "3mo": IntervalPolicy("3mo", chunk_days=3650, min_chunk_days=30, retention_days=None),
+}
+
+
+class YahooHistoryError(RuntimeError):
+    pass
+
+
+class YahooSymbolError(YahooHistoryError):
+    pass
+
+
+class YahooIntervalLimitError(YahooHistoryError):
+    pass
+
+
+class YahooRateLimitError(YahooHistoryError):
+    pass
+
+
+class SqliteResponseCache:
+    def __init__(self, path: Path) -> None:
+        self.path = path
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.Lock()
+        with self._connect() as connection:
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS response_cache (
+                    cache_key TEXT PRIMARY KEY,
+                    fetched_at INTEGER NOT NULL,
+                    payload_gzip BLOB NOT NULL
+                )
+                """
+            )
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(self.path)
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA synchronous=NORMAL")
+        return connection
+
+    def get(self, cache_key: str, ttl_seconds: int) -> dict[str, Any] | None:
+        with self._lock, self._connect() as connection:
+            row = connection.execute(
+                "SELECT fetched_at, payload_gzip FROM response_cache WHERE cache_key = ?",
+                (cache_key,),
+            ).fetchone()
+        if row is None:
+            return None
+        fetched_at, payload_gzip = row
+        if int(time.time()) - int(fetched_at) > ttl_seconds:
+            return None
+        return json.loads(gzip.decompress(payload_gzip).decode("utf-8"))
+
+    def set(self, cache_key: str, payload: dict[str, Any]) -> None:
+        packed = gzip.compress(json.dumps(payload, separators=(",", ":"), ensure_ascii=True).encode("utf-8"))
+        with self._lock, self._connect() as connection:
+            connection.execute(
+                """
+                INSERT INTO response_cache (cache_key, fetched_at, payload_gzip)
+                VALUES (?, ?, ?)
+                ON CONFLICT(cache_key) DO UPDATE SET
+                    fetched_at = excluded.fetched_at,
+                    payload_gzip = excluded.payload_gzip
+                """,
+                (cache_key, int(time.time()), packed),
+            )
+
+
+class RateLimiter:
+    def __init__(self, min_gap_seconds: float) -> None:
+        self.min_gap_seconds = max(0.0, float(min_gap_seconds))
+        self._lock = threading.Lock()
+        self._next_allowed = 0.0
+
+    def wait(self) -> None:
+        with self._lock:
+            delay = self._next_allowed - time.monotonic()
+            if delay > 0:
+                time.sleep(delay)
+            self._next_allowed = time.monotonic() + self.min_gap_seconds
+
+
+class YahooHistoryClient:
+    def __init__(
+        self,
+        *,
+        cache_path: Path,
+        timeout_seconds: float = 25.0,
+        min_request_gap_seconds: float = 0.35,
+        max_retries: int = 5,
+    ) -> None:
+        self.cache = SqliteResponseCache(cache_path)
+        self.timeout_seconds = timeout_seconds
+        self.rate_limiter = RateLimiter(min_request_gap_seconds)
+        self.host_cycle = cycle(YAHOO_CHART_HOSTS)
+        self.session = self._build_session(max_retries=max_retries)
+
+    def _build_session(self, *, max_retries: int) -> requests.Session:
+        retry = Retry(
+            total=max_retries,
+            connect=max_retries,
+            read=max_retries,
+            backoff_factor=0.8,
+            status_forcelist=(429, 500, 502, 503, 504),
+            allowed_methods=("GET",),
+            respect_retry_after_header=True,
+            raise_on_status=False,
+        )
+        adapter = HTTPAdapter(max_retries=retry, pool_connections=16, pool_maxsize=16)
+        session = requests.Session()
+        session.headers.update(DEFAULT_HEADERS)
+        session.mount("https://", adapter)
+        session.mount("http://", adapter)
+        return session
+
+    def fetch_history(
+        self,
+        symbol: str,
+        *,
+        interval: str,
+        start: str | datetime,
+        end: str | datetime,
+        include_prepost: bool = False,
+        adjust_ohlc: bool = False,
+    ) -> pd.DataFrame:
+        policy = self._interval_policy(interval)
+        start_dt = self._coerce_datetime(start, end_of_day=False)
+        end_dt = self._coerce_datetime(end, end_of_day=True)
+        if end_dt <= start_dt:
+            raise ValueError("end must be later than start")
+        self._validate_retention_window(policy=policy, start_dt=start_dt, end_dt=end_dt)
+
+        frames: list[pd.DataFrame] = []
+        for chunk_start, chunk_end in self._iter_chunks(start_dt=start_dt, end_dt=end_dt, chunk_days=policy.chunk_days):
+            chunk = self._fetch_chunk_adaptive(
+                symbol=symbol,
+                interval=policy.interval,
+                chunk_start=chunk_start,
+                chunk_end=chunk_end,
+                min_chunk_days=policy.min_chunk_days,
+                include_prepost=include_prepost,
+                adjust_ohlc=adjust_ohlc,
+            )
+            if not chunk.empty:
+                frames.append(chunk)
+        if not frames:
+            return pd.DataFrame(columns=BAR_COLUMNS)
+        history = pd.concat(frames, ignore_index=True)
+        history = history.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp").reset_index(drop=True)
+        return history[(history["timestamp"] >= start_dt) & (history["timestamp"] <= end_dt)].reset_index(drop=True)
+
+    def _fetch_chunk_adaptive(
+        self,
+        *,
+        symbol: str,
+        interval: str,
+        chunk_start: datetime,
+        chunk_end: datetime,
+        min_chunk_days: int,
+        include_prepost: bool,
+        adjust_ohlc: bool,
+    ) -> pd.DataFrame:
+        try:
+            payload = self._request_chart(
+                symbol=symbol,
+                interval=interval,
+                start_dt=chunk_start,
+                end_dt=chunk_end,
+                include_prepost=include_prepost,
+            )
+            return self._payload_to_frame(payload=payload, adjust_ohlc=adjust_ohlc)
+        except YahooIntervalLimitError:
+            if max((chunk_end - chunk_start).days, 1) <= min_chunk_days:
+                raise
+            midpoint = chunk_start + (chunk_end - chunk_start) / 2
+            left = self._fetch_chunk_adaptive(
+                symbol=symbol,
+                interval=interval,
+                chunk_start=chunk_start,
+                chunk_end=midpoint,
+                min_chunk_days=min_chunk_days,
+                include_prepost=include_prepost,
+                adjust_ohlc=adjust_ohlc,
+            )
+            right = self._fetch_chunk_adaptive(
+                symbol=symbol,
+                interval=interval,
+                chunk_start=midpoint,
+                chunk_end=chunk_end,
+                min_chunk_days=min_chunk_days,
+                include_prepost=include_prepost,
+                adjust_ohlc=adjust_ohlc,
+            )
+            return pd.concat([left, right], ignore_index=True)
+
+    def _request_chart(
+        self,
+        *,
+        symbol: str,
+        interval: str,
+        start_dt: datetime,
+        end_dt: datetime,
+        include_prepost: bool,
+    ) -> dict[str, Any]:
+        params = {
+            "period1": str(int(start_dt.timestamp())),
+            "period2": str(int(end_dt.timestamp())),
+            "interval": interval,
+            "includePrePost": "true" if include_prepost else "false",
+            "events": "div,splits,capitalGains",
+            "includeAdjustedClose": "true",
+        }
+        last_error: Exception | None = None
+        for _ in range(len(YAHOO_CHART_HOSTS)):
+            base_url = next(self.host_cycle)
+            url = f"{base_url}/v8/finance/chart/{requests.utils.quote(symbol, safe='')}"
+            cache_key = self._cache_key(url=url, params=params)
+            cached = self.cache.get(cache_key, ttl_seconds=self._cache_ttl_seconds(end_dt=end_dt))
+            if cached is not None:
+                return cached
+
+            self.rate_limiter.wait()
+            response = self.session.get(url, params=params, timeout=self.timeout_seconds)
+            if response.status_code == 429:
+                last_error = YahooRateLimitError(f"Yahoo rate-limited {symbol} at interval {interval}.")
+                time.sleep(1.5)
+                continue
+            if response.status_code == 404:
+                raise YahooSymbolError(f"Yahoo did not recognize ticker {symbol}.")
+            if response.status_code == 422:
+                raise YahooIntervalLimitError(
+                    f"Yahoo rejected {symbol} {interval} from {start_dt.isoformat()} to {end_dt.isoformat()}."
+                )
+            try:
+                response.raise_for_status()
+            except requests.HTTPError as exc:
+                last_error = exc
+                continue
+
+            payload = response.json()
+            error = payload.get("chart", {}).get("error")
+            if error:
+                description = error.get("description") or error.get("code") or str(error)
+                lowered = description.lower()
+                if "not found" in lowered or "no data found" in lowered or "symbol" in lowered:
+                    raise YahooSymbolError(description)
+                if "range" in lowered or "interval" in lowered or "last" in lowered:
+                    raise YahooIntervalLimitError(description)
+                if "rate limit" in lowered or "too many requests" in lowered:
+                    raise YahooRateLimitError(description)
+                raise YahooHistoryError(description)
+            self.cache.set(cache_key, payload)
+            return payload
+
+        if last_error is not None:
+            raise YahooHistoryError(str(last_error)) from last_error
+        raise YahooHistoryError(f"Yahoo request failed for {symbol} {interval}.")
+
+    def _payload_to_frame(self, *, payload: dict[str, Any], adjust_ohlc: bool) -> pd.DataFrame:
+        result = payload.get("chart", {}).get("result") or []
+        if not result:
+            return pd.DataFrame(columns=BAR_COLUMNS)
+        result0 = result[0]
+        meta = result0.get("meta") or {}
+        timestamps = result0.get("timestamp") or []
+        quote_sets = result0.get("indicators", {}).get("quote") or []
+        if not timestamps or not quote_sets:
+            return pd.DataFrame(columns=BAR_COLUMNS)
+
+        try:
+            timezone = ZoneInfo(meta.get("exchangeTimezoneName") or "UTC")
+        except Exception:
+            timezone = ZoneInfo("UTC")
+        quote = quote_sets[0]
+        adjclose_sets = result0.get("indicators", {}).get("adjclose") or [{}]
+        adj_close = adjclose_sets[0].get("adjclose", []) if adjclose_sets else []
+        events = result0.get("events") or {}
+        dividends = self._event_series(events.get("dividends") or {}, value_key="amount")
+        splits = self._event_series(events.get("splits") or {}, value_key="splitRatio")
+        row_count = len(timestamps)
+
+        frame = pd.DataFrame(
+            {
+                "timestamp": pd.to_datetime(timestamps, unit="s", utc=True).tz_convert(timezone).tz_localize(None),
+                "open": self._normalize_values(quote.get("open", []), row_count),
+                "high": self._normalize_values(quote.get("high", []), row_count),
+                "low": self._normalize_values(quote.get("low", []), row_count),
+                "close": self._normalize_values(quote.get("close", []), row_count),
+                "adj_close": self._normalize_values(adj_close, row_count),
+                "volume": self._normalize_values(quote.get("volume", []), row_count),
+            }
+        )
+        for column in ("open", "high", "low", "close", "adj_close", "volume"):
+            frame[column] = pd.to_numeric(frame[column], errors="coerce")
+        frame = frame.dropna(subset=["timestamp", "close"]).reset_index(drop=True)
+        frame["volume"] = frame["volume"].fillna(0.0)
+
+        frame["epoch"] = (frame["timestamp"].astype("int64") // 1_000_000_000).astype("int64")
+        frame["dividend"] = frame["epoch"].map(dividends).fillna(0.0)
+        frame["split_ratio"] = frame["epoch"].map(splits).fillna(1.0)
+        frame = frame.drop(columns=["epoch"])
+
+        if adjust_ohlc:
+            ratio = frame["adj_close"].where(frame["close"] != 0, frame["close"]) / frame["close"].replace(0, pd.NA)
+            ratio = ratio.fillna(1.0)
+            for column in ("open", "high", "low", "close"):
+                frame[column] = frame[column] * ratio
+        return frame.drop_duplicates(subset=["timestamp"], keep="last").sort_values("timestamp").reset_index(drop=True)[BAR_COLUMNS]
+
+    @staticmethod
+    def _event_series(events: dict[str, Any], *, value_key: str) -> dict[int, float]:
+        output: dict[int, float] = {}
+        for event in events.values():
+            timestamp = event.get("date")
+            value = event.get(value_key)
+            if timestamp is None or value is None:
+                continue
+            try:
+                output[int(timestamp)] = float(value)
+            except (TypeError, ValueError):
+                continue
+        return output
+
+    @staticmethod
+    def _normalize_values(values: list[Any] | tuple[Any, ...], size: int) -> list[Any]:
+        normalized = list(values[:size])
+        if len(normalized) < size:
+            normalized.extend([None] * (size - len(normalized)))
+        return normalized
+
+    @staticmethod
+    def _cache_key(*, url: str, params: dict[str, str]) -> str:
+        material = json.dumps({"url": url, "params": params}, sort_keys=True, separators=(",", ":"))
+        return hashlib.sha256(material.encode("utf-8")).hexdigest()
+
+    @staticmethod
+    def _cache_ttl_seconds(*, end_dt: datetime) -> int:
+        now_utc = datetime.utcnow()
+        if end_dt < now_utc - timedelta(days=2):
+            return 7 * 24 * 60 * 60
+        if end_dt < now_utc - timedelta(hours=12):
+            return 60 * 60
+        return 90
+
+    @staticmethod
+    def _coerce_datetime(value: str | datetime, *, end_of_day: bool) -> datetime:
+        timestamp = pd.Timestamp(value)
+        if timestamp.tzinfo is not None:
+            timestamp = timestamp.tz_convert("UTC").tz_localize(None)
+        dt = timestamp.to_pydatetime()
+        if end_of_day and dt.hour == 0 and dt.minute == 0 and dt.second == 0 and dt.microsecond == 0:
+            return dt + timedelta(days=1)
+        return dt
+
+    @staticmethod
+    def _iter_chunks(*, start_dt: datetime, end_dt: datetime, chunk_days: int) -> list[tuple[datetime, datetime]]:
+        chunks: list[tuple[datetime, datetime]] = []
+        cursor = start_dt
+        while cursor < end_dt:
+            next_edge = min(cursor + timedelta(days=chunk_days), end_dt)
+            chunks.append((cursor, next_edge))
+            cursor = next_edge
+        return chunks
+
+    @staticmethod
+    def _interval_policy(interval: str) -> IntervalPolicy:
+        normalized = interval.strip()
+        if normalized not in INTERVAL_POLICIES:
+            allowed = ", ".join(sorted(INTERVAL_POLICIES))
+            raise ValueError(f"Unsupported interval {interval!r}. Allowed values: {allowed}")
+        return INTERVAL_POLICIES[normalized]
+
+    @staticmethod
+    def _validate_retention_window(*, policy: IntervalPolicy, start_dt: datetime, end_dt: datetime) -> None:
+        if policy.retention_days is None:
+            return
+        earliest = datetime.utcnow() - timedelta(days=policy.retention_days)
+        if start_dt < earliest or end_dt < earliest:
+            cutoff = earliest.strftime("%Y-%m-%d")
+            raise YahooIntervalLimitError(
+                f"Yahoo only serves {policy.interval} history back to about {cutoff}. Use 1d or coarser for deeper history."
+            )

requirements.txt

@@ -1,7 +1,7 @@
 pandas
 pandas_market_calendars
 pyarrow
-yfinance
+requests
 fastapi
 uvicorn
 joblib

scripts/run_ist_scheduler.py

@@ -9,12 +9,14 @@ from zoneinfo import ZoneInfo
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from nifty_backend.runtime import (
     CLOSE_REFRESH_READY,
+    STALE_CHECK_INTERVAL_SECONDS,
     is_trading_day,
     latest_saved_prediction,
     close_refresh_due,
     refresh_market_close_data,
     refresh_daily_data,
     refresh_first5_prediction,
+    refresh_stale_data_once,
     seconds_until_next_ist_run,
 )
 
@@ -50,6 +52,12 @@ def refresh_close_data_if_due() -> None:
     print(f"[scheduler] close data refreshed: {info}")
 
 
+def refresh_stale_data_if_due() -> None:
+    info = refresh_stale_data_once()
+    if info.get("status") == "refreshed":
+        print(f"[scheduler] stale data refreshed: {info}")
+
+
 def main() -> None:
     print("[scheduler] NIFTY first-five-minute scheduler started.")
     print("[scheduler] Runs the opening prediction after 09:20 IST so the 09:15-09:19 candles are complete.")
@@ -57,12 +65,14 @@ def main() -> None:
         try:
             refresh_if_current_session_is_ready()
             refresh_close_data_if_due()
+            refresh_stale_data_if_due()
         except Exception as exc:
             print(f"[scheduler] current-session refresh failed: {exc}")
-        sleep_for = seconds_until_next_ist_run()
-        target = datetime.now(IST).timestamp() + sleep_for
-        print(f"[scheduler] sleeping {sleep_for / 60:.1f} minutes; next wake timestamp={target:.0f}")
-        time.sleep(sleep_for)
+        next_first5 = seconds_until_next_ist_run()
+        if next_first5 > STALE_CHECK_INTERVAL_SECONDS:
+            time.sleep(STALE_CHECK_INTERVAL_SECONDS)
+            continue
+        time.sleep(next_first5)
         try:
             prediction = refresh_first5_prediction()
             print(f"[scheduler] first5 prediction refreshed: {prediction.to_dict()}")
@@ -73,9 +83,11 @@ def main() -> None:
             print(f"[scheduler] daily data refreshed: {info}")
         except Exception as exc:
             print(f"[scheduler] daily refresh failed: {exc}")
-        sleep_for = seconds_until_next_ist_run(CLOSE_REFRESH_READY)
-        print(f"[scheduler] sleeping {sleep_for / 60:.1f} minutes until close refresh.")
-        time.sleep(sleep_for)
+        next_close = seconds_until_next_ist_run(CLOSE_REFRESH_READY)
+        if next_close > STALE_CHECK_INTERVAL_SECONDS:
+            time.sleep(STALE_CHECK_INTERVAL_SECONDS)
+            continue
+        time.sleep(next_close)
         try:
             refresh_close_data_if_due()
         except Exception as exc: