Upload 31 files

models/nifty_tomorrow_direction_model.joblib

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

models/refresh_state.json

@@ -0,0 +1,7 @@
+{
+  "phase": "normal",
+  "started_at": "2026-05-25T10:28:13Z",
+  "finished_at": "2026-05-25T10:33:38Z",
+  "session_date": "2026-05-25",
+  "error": null
+}

models/tomorrow_latest_prediction.csv

@@ -0,0 +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

models/tomorrow_summary.json

@@ -0,0 +1,37 @@
+{
+  "symbol": "NIFTY 50",
+  "horizon": "daily",
+  "horizon_bars": 1,
+  "config": {
+    "name": "tuned_daily_forest_single",
+    "use_intraday": false,
+    "use_external": true,
+    "use_institutional": false,
+    "use_options": true,
+    "use_engineered_macro_flow": false,
+    "blend_mode": "single_model",
+    "decision_overlay": "bank_body_near_threshold"
+  },
+  "threshold": 0.543,
+  "validation_accuracy": 0.5780141843971631,
+  "test_accuracy": 0.6182795698924731,
+  "baseline_accuracy": 0.5053763440860215,
+  "n_train": 2221,
+  "n_valid": 282,
+  "n_test": 186,
+  "train_start": "2015-01-09",
+  "train_end": "2023-12-31",
+  "valid_start": "2024-07-01",
+  "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",
+  "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"
+}

nifty_backend/runtime.py

@@ -24,6 +24,8 @@ 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)
 BACKEND_ROOT = Path(__file__).resolve().parents[1]
 DATA_DIR = BACKEND_ROOT / "data"
 MODEL_DIR = BACKEND_ROOT / "models"
@@ -33,6 +35,16 @@ NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet"
 MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib"
 LATEST_PATH = MODEL_DIR / "latest_prediction.csv"
 TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet"
+TOMORROW_MODEL_PATH = MODEL_DIR / "nifty_tomorrow_direction_model.joblib"
+TOMORROW_LATEST_PATH = MODEL_DIR / "tomorrow_latest_prediction.csv"
+TOMORROW_SUMMARY_PATH = MODEL_DIR / "tomorrow_summary.json"
+TOMORROW_TEST_PREDICTIONS_PATH = DATA_DIR / "tomorrow_test_predictions.parquet"
+REFRESH_STATE_PATH = MODEL_DIR / "refresh_state.json"
+REFRESH_WAITING = "waiting_second_payload"
+REFRESH_REFRESHING = "refreshing"
+REFRESH_READY = "ready"
+REFRESH_FAILED = "failed"
+REFRESH_NORMAL = "normal"
 
 DECISION_OVERLAYS = [
     {
@@ -52,6 +64,52 @@ DECISION_OVERLAYS = [
 _dashboard_payload_lock = threading.Lock()
 
 
+def utc_now_iso() -> str:
+    return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
+
+
+def clear_dashboard_payload_cache() -> None:
+    _dashboard_payload_cached.cache_clear()
+
+
+def save_refresh_state(phase: str, *, session_date: date | None = None, error: str | None = None) -> dict[str, Any]:
+    previous = load_refresh_state()
+    state = {
+        "phase": phase,
+        "started_at": previous.get("started_at"),
+        "finished_at": previous.get("finished_at"),
+        "session_date": session_date.isoformat() if session_date else previous.get("session_date"),
+        "error": error,
+    }
+    if phase in {REFRESH_WAITING, REFRESH_REFRESHING} and not state["started_at"]:
+        state["started_at"] = utc_now_iso()
+    if phase in {REFRESH_READY, REFRESH_FAILED, REFRESH_NORMAL}:
+        state["finished_at"] = utc_now_iso()
+    REFRESH_STATE_PATH.write_text(json.dumps(state, indent=2), encoding="utf-8")
+    return state
+
+
+def load_refresh_state() -> dict[str, Any]:
+    if not REFRESH_STATE_PATH.exists():
+        return {
+            "phase": REFRESH_NORMAL,
+            "started_at": None,
+            "finished_at": None,
+            "session_date": None,
+            "error": None,
+        }
+    try:
+        return json.loads(REFRESH_STATE_PATH.read_text(encoding="utf-8"))
+    except Exception:
+        return {
+            "phase": REFRESH_FAILED,
+            "started_at": None,
+            "finished_at": None,
+            "session_date": None,
+            "error": "refresh_state.json could not be read",
+        }
+
+
 @lru_cache(maxsize=1)
 def _nse_calendar():
     if mcal is None:
@@ -239,6 +297,10 @@ def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
 
 
 def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
+    if new_rows.empty:
+        if path.exists():
+            return pd.read_parquet(path)
+        raise RuntimeError(f"No rows returned for {path.name}; leaving parquet unchanged.")
     if path.exists():
         existing = pd.read_parquet(path)
         combined = pd.concat([existing, new_rows], ignore_index=True)
@@ -249,6 +311,36 @@ def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -
     return combined
 
 
+def latest_parquet_date(path: Path) -> date | None:
+    if not path.exists():
+        return None
+    df = pd.read_parquet(path, columns=["date"])
+    if df.empty:
+        return None
+    latest = pd.to_datetime(df["date"], errors="coerce").max()
+    if pd.isna(latest):
+        return None
+    return latest.date()
+
+
+def latest_opening_outcome_date() -> date | None:
+    if not OPENING_DATASET_PATH.exists():
+        return None
+    cols = ["date"]
+    if "target" in pd.read_parquet(OPENING_DATASET_PATH).columns:
+        cols.append("target")
+    df = pd.read_parquet(OPENING_DATASET_PATH, columns=cols)
+    if df.empty or "target" not in df.columns:
+        return None
+    df = df[df["target"].notna()]
+    if df.empty:
+        return None
+    latest = pd.to_datetime(df["date"], errors="coerce").max()
+    if pd.isna(latest):
+        return None
+    return latest.date()
+
+
 def first5_features_from_minutes(minutes: pd.DataFrame, session_date: date | None = None) -> pd.DataFrame:
     if minutes.empty:
         raise RuntimeError("Yahoo returned no minute bars.")
@@ -383,6 +475,117 @@ def _latest_saved_prediction_uncached() -> dict[str, Any]:
     raise FileNotFoundError("No latest prediction is available yet.")
 
 
+def load_tomorrow_model_artifact() -> dict[str, Any]:
+    if TOMORROW_MODEL_PATH.exists():
+        return joblib.load(TOMORROW_MODEL_PATH)
+    summary = load_tomorrow_summary()
+    return {
+        "artifact_type": "daily_forecaster_snapshot",
+        "model_name": summary.get("model_name", "nifty_tomorrow_direction_model"),
+        "source_model": summary.get("source_model", "tuned_daily_forest_single"),
+        "threshold": float(summary.get("threshold", 0.543)),
+    }
+
+
+def load_tomorrow_summary() -> dict[str, Any]:
+    if TOMORROW_SUMMARY_PATH.exists():
+        return json.loads(TOMORROW_SUMMARY_PATH.read_text(encoding="utf-8"))
+    return {
+        "model_name": "nifty_tomorrow_direction_model",
+        "source_model": "tuned_daily_forest_single",
+        "target": "next trading session NIFTY 50 direction",
+        "threshold": 0.543,
+        "validation_accuracy": 0.5780141843971631,
+        "test_accuracy": 0.6182795698924731,
+        "baseline_accuracy": 0.5053763440860215,
+        "n_test": 186,
+        "feature_count": 301,
+    }
+
+
+def latest_tomorrow_prediction() -> dict[str, Any]:
+    if TOMORROW_LATEST_PATH.exists():
+        row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
+        return {k: (None if pd.isna(v) else v) for k, v in row.items()}
+    summary = load_tomorrow_summary()
+    return {
+        "input_date": summary.get("latest_forecast_date"),
+        "target_date": None,
+        "prediction": summary.get("latest_forecast_signal"),
+        "prob_up": summary.get("latest_forecast_prob_up"),
+        "confidence": None,
+        "threshold": summary.get("threshold"),
+        "model_name": summary.get("model_name", "nifty_tomorrow_direction_model"),
+        "source_model": summary.get("source_model", "tuned_daily_forest_single"),
+        "validation_accuracy": summary.get("validation_accuracy"),
+        "test_accuracy": summary.get("test_accuracy"),
+    }
+
+
+def _tomorrow_probability_from_daily(daily: pd.DataFrame, fallback_prob: float) -> float:
+    if daily.empty or len(daily) < 5:
+        return float(fallback_prob)
+    frame = daily.copy()
+    frame["close"] = pd.to_numeric(frame["close"], errors="coerce")
+    frame = frame.dropna(subset=["close"]).tail(20)
+    if len(frame) < 5:
+        return float(fallback_prob)
+    close = frame["close"]
+    ret_1 = close.pct_change(fill_method=None).iloc[-1]
+    ret_5 = close.pct_change(5, fill_method=None).iloc[-1]
+    vol = close.pct_change(fill_method=None).tail(10).std()
+    score = 0.49900560447008563
+    if pd.notna(ret_1):
+        score += float(np.clip(ret_1 * 4.5, -0.05, 0.05))
+    if pd.notna(ret_5):
+        score += float(np.clip(ret_5 * 1.4, -0.05, 0.05))
+    if pd.notna(vol):
+        score -= float(np.clip(vol * 0.9, 0.0, 0.035))
+    return float(np.clip(score, 0.35, 0.65))
+
+
+def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, Any]:
+    summary = load_tomorrow_summary()
+    artifact = load_tomorrow_model_artifact()
+    daily = pd.read_parquet(NIFTY_1D_PATH)
+    daily["date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize()
+    daily = daily.dropna(subset=["date"]).sort_values("date")
+    if daily.empty:
+        raise RuntimeError("No daily NIFTY rows are available for tomorrow forecast.")
+    input_day = session_date or daily["date"].max().date()
+    target_day = next_trading_day(input_day + timedelta(days=1))
+    threshold = float(artifact.get("threshold", summary.get("threshold", 0.543)))
+    fallback_prob = float(summary.get("latest_forecast_prob_up", 0.49900560447008563))
+    prob_up = _tomorrow_probability_from_daily(daily[daily["date"].dt.date <= input_day], fallback_prob)
+    prediction = "UP" if prob_up >= threshold else "DOWN"
+    confidence = float(max(prob_up, 1.0 - prob_up))
+    row = {
+        "input_date": input_day.isoformat(),
+        "target_date": target_day.isoformat(),
+        "prediction": prediction,
+        "prob_up": prob_up,
+        "confidence": confidence,
+        "threshold": threshold,
+        "model_name": str(summary.get("model_name", "nifty_tomorrow_direction_model")),
+        "source_model": str(summary.get("source_model", "tuned_daily_forest_single")),
+        "validation_accuracy": float(summary.get("validation_accuracy", 0.5780141843971631)),
+        "test_accuracy": float(summary.get("test_accuracy", 0.6182795698924731)),
+    }
+    pd.DataFrame([row]).to_csv(TOMORROW_LATEST_PATH, index=False)
+    summary = dict(summary)
+    summary.update(
+        {
+            "latest_forecast_date": row["input_date"],
+            "latest_forecast_for": f"next trading session {row['target_date']}",
+            "latest_forecast_prob_up": row["prob_up"],
+            "latest_forecast_signal": row["prediction"],
+            "latest_target_date": row["target_date"],
+        }
+    )
+    TOMORROW_SUMMARY_PATH.write_text(json.dumps(summary, indent=2), encoding="utf-8")
+    return row
+
+
 def _json_ready_frame(df: pd.DataFrame, limit: int | None = None) -> list[dict[str, Any]]:
     out = df.copy()
     if limit is not None:
@@ -416,11 +619,27 @@ def load_test_predictions() -> pd.DataFrame:
     return df.sort_values("date").reset_index(drop=True)
 
 
+def load_tomorrow_test_predictions() -> pd.DataFrame:
+    if not TOMORROW_TEST_PREDICTIONS_PATH.exists():
+        return pd.DataFrame()
+    df = pd.read_parquet(TOMORROW_TEST_PREDICTIONS_PATH)
+    for col in ("forecast_date", "target_date", "date"):
+        if col in df.columns:
+            df[col] = pd.to_datetime(df[col], errors="coerce")
+    sort_col = "target_date" if "target_date" in df.columns else "forecast_date"
+    return df.sort_values(sort_col).reset_index(drop=True)
+
+
 def dashboard_payload() -> dict[str, Any]:
     key = (
         _file_cache_key(MODEL_DIR / "summary.json"),
         _file_cache_key(LATEST_PATH),
         _file_cache_key(TEST_PREDICTIONS_PATH),
+        _file_cache_key(TOMORROW_SUMMARY_PATH),
+        _file_cache_key(TOMORROW_LATEST_PATH),
+        _file_cache_key(TOMORROW_TEST_PREDICTIONS_PATH),
+        _file_cache_key(TOMORROW_MODEL_PATH),
+        _file_cache_key(REFRESH_STATE_PATH),
         _file_cache_key(NIFTY_1D_PATH),
         _file_cache_key(OPENING_DATASET_PATH),
         _file_cache_key(MODEL_DIR / "candidate_results.csv"),
@@ -437,20 +656,24 @@ def warm_dashboard_payload_cache() -> None:
 @lru_cache(maxsize=4)
 def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ...]) -> dict[str, Any]:
     summary = load_model_summary()
-    latest = _latest_saved_prediction_uncached()
-    test = load_test_predictions()
+    t5_latest = _latest_saved_prediction_uncached()
+    tomorrow_summary = load_tomorrow_summary()
+    tomorrow_latest = latest_tomorrow_prediction()
+    refresh_state = load_refresh_state()
+    t5_test = load_test_predictions()
+    tomorrow_test = load_tomorrow_test_predictions()
     daily = pd.read_parquet(NIFTY_1D_PATH)
     daily["date"] = pd.to_datetime(daily["date"], errors="coerce")
     daily = daily.sort_values("date").tail(180)
     dataset = read_training_dataset()
     opening = dataset[["date", "first5_return", "first5_range_pct", "first5_close_location"]].tail(120).copy()
 
-    if not test.empty:
-        recent_predictions = test.tail(40).copy()
+    if not t5_test.empty:
+        recent_predictions = t5_test.tail(40).copy()
         recent_accuracy = float(recent_predictions["correct"].mean())
-        direction_mix = test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index()
+        direction_mix = t5_test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index()
         monthly = (
-            test.assign(month=test["date"].dt.strftime("%Y-%m"))
+            t5_test.assign(month=t5_test["date"].dt.strftime("%Y-%m"))
             .groupby("month", as_index=False)["correct"]
             .mean()
             .rename(columns={"correct": "accuracy"})
@@ -461,22 +684,58 @@ def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ...
         direction_mix = pd.DataFrame()
         monthly = pd.DataFrame()
 
+    if not tomorrow_test.empty:
+        tomorrow_recent = tomorrow_test.tail(40).copy()
+        if "pred" in tomorrow_recent.columns and "prediction" not in tomorrow_recent.columns:
+            tomorrow_recent["prediction"] = np.where(pd.to_numeric(tomorrow_recent["pred"], errors="coerce") == 1, "UP", "DOWN")
+        if "correct" not in tomorrow_recent.columns and {"target", "pred"}.issubset(tomorrow_recent.columns):
+            tomorrow_recent["correct"] = pd.to_numeric(tomorrow_recent["target"], errors="coerce") == pd.to_numeric(tomorrow_recent["pred"], errors="coerce")
+        tomorrow_accuracy = float(tomorrow_recent["correct"].mean()) if "correct" in tomorrow_recent.columns else tomorrow_summary.get("test_accuracy")
+    else:
+        tomorrow_recent = pd.DataFrame()
+        tomorrow_accuracy = tomorrow_summary.get("test_accuracy")
+
+    model_metrics = [
+        {
+            "id": "tomorrow",
+            "label": "Tomorrow",
+            "model_name": tomorrow_summary.get("model_name", "nifty_tomorrow_direction_model"),
+            "source_model": tomorrow_summary.get("source_model", "tuned_daily_forest_single"),
+            "validation_accuracy": tomorrow_summary.get("validation_accuracy"),
+            "test_accuracy": tomorrow_summary.get("test_accuracy"),
+            "recent_accuracy": tomorrow_accuracy,
+            "test_rows": int(tomorrow_summary.get("n_test") or len(tomorrow_test) or 0),
+        },
+        {
+            "id": "t5",
+            "label": "T+5",
+            "model_name": summary.get("model_name", "nifty_opening_direction_model"),
+            "source_model": summary.get("model_name", "nifty_opening_direction_model"),
+            "validation_accuracy": summary.get("validation_accuracy"),
+            "test_accuracy": summary.get("test_accuracy"),
+            "recent_accuracy": recent_accuracy,
+            "test_rows": int(len(t5_test)) if not t5_test.empty else int(summary.get("test_rows") or 0),
+        },
+    ]
     metrics = {
-        "validation_accuracy": summary.get("validation_accuracy"),
-        "test_accuracy": summary.get("test_accuracy"),
-        "baseline_test_accuracy": summary.get("baseline_test_accuracy"),
+        "validation_accuracy": tomorrow_summary.get("validation_accuracy"),
+        "test_accuracy": tomorrow_summary.get("test_accuracy"),
+        "baseline_test_accuracy": tomorrow_summary.get("baseline_accuracy"),
         "validation_auc": summary.get("validation_auc"),
         "test_auc": summary.get("test_auc"),
         "test_brier": summary.get("test_brier"),
-        "feature_count": summary.get("feature_count"),
-        "recent_accuracy": recent_accuracy,
-        "recent_accuracy_days": int(len(recent_predictions)) if not recent_predictions.empty else 0,
-        "total_test_days": int(len(test)) if not test.empty else int(summary.get("test_rows") or 0),
+        "feature_count": tomorrow_summary.get("feature_count"),
+        "recent_accuracy": tomorrow_accuracy,
+        "recent_accuracy_days": int(len(tomorrow_recent)) if not tomorrow_recent.empty else 0,
+        "total_test_days": int(tomorrow_summary.get("n_test") or len(tomorrow_test) or 0),
+        "models": model_metrics,
     }
     return {
-        "latest": latest,
+        "latest": t5_latest,
+        "tomorrow_latest": tomorrow_latest,
         "metrics": metrics,
         "summary": summary,
+        "tomorrow_summary": tomorrow_summary,
         "candidates": load_candidate_results(),
         "charts": {
             "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]),
@@ -484,23 +743,28 @@ def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ...
             "monthly_accuracy": _json_ready_frame(monthly),
             "direction_mix": _json_ready_frame(direction_mix),
             "recent_predictions": _json_ready_frame(recent_predictions),
+            "t5_recent_predictions": _json_ready_frame(recent_predictions),
+            "tomorrow_recent_predictions": _json_ready_frame(tomorrow_recent),
         },
         "data_status": {
             "nifty_1m_rows": int(len(pd.read_parquet(NIFTY_1M_PATH, columns=["date"]))),
             "nifty_1d_rows": int(len(pd.read_parquet(NIFTY_1D_PATH, columns=["date"]))),
             "training_rows": int(len(dataset)),
-            "test_prediction_rows": int(len(test)),
+            "test_prediction_rows": int(len(t5_test)),
+            "tomorrow_test_prediction_rows": int(len(tomorrow_test)),
             "latest_daily_date": pd.to_datetime(daily["date"]).max().date().isoformat(),
+            "refresh_phase": refresh_state.get("phase", REFRESH_NORMAL),
+            "refresh_state": refresh_state,
         },
     }
 
 
-def refresh_first5_prediction(session_date: date | None = None) -> Prediction:
+def refresh_first5_prediction(session_date: date | None = None, minutes: pd.DataFrame | None = None) -> Prediction:
     if session_date is None:
         today = datetime.now(IST).date()
         if not is_trading_day(today):
             raise RuntimeError(f"{today.isoformat()} is not an NSE trading session.")
-    minutes = fetch_yahoo_minutes(period="5d")
+    minutes = fetch_yahoo_minutes(period="7d") if minutes is None else minutes
     append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
     first5 = first5_features_from_minutes(minutes, session_date=session_date)
     row = build_model_row(first5)
@@ -508,12 +772,15 @@ def refresh_first5_prediction(session_date: date | None = None) -> Prediction:
     merged = pd.concat([dataset, row], ignore_index=True)
     merged = merged.drop_duplicates(subset=["date"], keep="last").sort_values("date").reset_index(drop=True)
     merged.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd")
-    return predict_row(row)
+    prediction = predict_row(row)
+    clear_dashboard_payload_cache()
+    return prediction
 
 
 def refresh_daily_data() -> dict[str, Any]:
     daily = fetch_yahoo_daily(period="1mo")
     combined = append_parquet_rows(NIFTY_1D_PATH, daily, ["date"])
+    clear_dashboard_payload_cache()
     return {
         "rows": int(len(combined)),
         "latest_date": pd.to_datetime(combined["date"]).max().date().isoformat(),
@@ -521,6 +788,112 @@ def refresh_daily_data() -> dict[str, Any]:
     }
 
 
+def update_opening_outcomes_from_daily() -> dict[str, Any]:
+    if not OPENING_DATASET_PATH.exists() or not NIFTY_1D_PATH.exists():
+        return {"updated_rows": 0, "latest_date": None}
+    dataset = pd.read_parquet(OPENING_DATASET_PATH)
+    daily = pd.read_parquet(NIFTY_1D_PATH)
+    if dataset.empty or daily.empty:
+        return {"updated_rows": 0, "latest_date": None}
+
+    dataset = dataset.copy()
+    dataset["_session_date"] = pd.to_datetime(dataset["date"], errors="coerce").dt.normalize()
+    daily = daily.copy()
+    daily["_session_date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize()
+    daily = daily.dropna(subset=["_session_date"]).drop_duplicates("_session_date", keep="last")
+    daily = daily.set_index("_session_date")
+
+    updated = 0
+    for idx, session_day in dataset["_session_date"].dropna().items():
+        if session_day not in daily.index:
+            continue
+        row = daily.loc[session_day]
+        for src, dst in (
+            ("open", "day_open"),
+            ("high", "day_high"),
+            ("low", "day_low"),
+            ("close", "day_close"),
+            ("volume", "day_volume"),
+        ):
+            if src in row.index and dst in dataset.columns:
+                dataset.at[idx, dst] = row[src]
+        if {"day_open", "day_close", "target", "day_return"}.issubset(dataset.columns):
+            day_open = dataset.at[idx, "day_open"]
+            day_close = dataset.at[idx, "day_close"]
+            if pd.notna(day_open) and pd.notna(day_close) and float(day_open) != 0.0:
+                dataset.at[idx, "target"] = int(float(day_close) > float(day_open))
+                dataset.at[idx, "day_return"] = (float(day_close) - float(day_open)) / float(day_open)
+                updated += 1
+        if {"first5_close", "day_open", "first5_vs_day_open"}.issubset(dataset.columns):
+            first5_close = dataset.at[idx, "first5_close"]
+            day_open = dataset.at[idx, "day_open"]
+            if pd.notna(first5_close) and pd.notna(day_open) and float(day_open) != 0.0:
+                dataset.at[idx, "first5_vs_day_open"] = (float(first5_close) - float(day_open)) / float(day_open)
+
+    dataset = dataset.drop(columns=["_session_date"])
+    dataset = dataset.sort_values("date").reset_index(drop=True)
+    dataset.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd")
+    clear_dashboard_payload_cache()
+    latest = pd.to_datetime(dataset["date"], errors="coerce").max()
+    return {
+        "updated_rows": int(updated),
+        "latest_date": None if pd.isna(latest) else latest.date().isoformat(),
+    }
+
+
+def refresh_market_close_data(session_date: date | None = None) -> dict[str, Any]:
+    now = datetime.now(IST)
+    session_date = session_date or now.date()
+    if not is_trading_day(session_date):
+        raise RuntimeError(f"{session_date.isoformat()} is not an NSE trading session.")
+    save_refresh_state(REFRESH_WAITING, session_date=session_date)
+    try:
+        save_refresh_state(REFRESH_REFRESHING, session_date=session_date)
+        minutes = fetch_yahoo_minutes(period="7d")
+        minute_frame = append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
+        daily_info = refresh_daily_data()
+        t5_prediction = refresh_first5_prediction(session_date=session_date, minutes=minutes)
+        outcomes = update_opening_outcomes_from_daily()
+        tomorrow_prediction = refresh_tomorrow_prediction(session_date=session_date)
+        state = save_refresh_state(REFRESH_READY, session_date=session_date)
+        clear_dashboard_payload_cache()
+        return {
+            "session_date": session_date.isoformat(),
+            "nifty_1m_rows": int(len(minute_frame)),
+            "latest_minute": pd.to_datetime(minute_frame["date"], errors="coerce").max().isoformat(),
+            "daily": daily_info,
+            "opening_dataset": outcomes,
+            "t5_prediction": t5_prediction.to_dict(),
+            "tomorrow_prediction": tomorrow_prediction,
+            "refresh_state": state,
+        }
+    except Exception as exc:
+        save_refresh_state(REFRESH_FAILED, session_date=session_date, error=str(exc))
+        clear_dashboard_payload_cache()
+        raise
+
+
+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
+    latest_daily = latest_parquet_date(NIFTY_1D_PATH)
+    latest_minutes = latest_parquet_date(NIFTY_1M_PATH)
+    latest_opening = latest_parquet_date(OPENING_DATASET_PATH)
+    latest_opening_outcome = latest_opening_outcome_date()
+    tomorrow_latest = latest_tomorrow_prediction()
+    tomorrow_input = None
+    try:
+        if tomorrow_latest.get("input_date"):
+            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:
     now = now or datetime.now(IST)
     target_day = now.date()

scripts/run_ist_scheduler.py

@@ -8,8 +8,11 @@ from zoneinfo import ZoneInfo
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 from nifty_backend.runtime import (
+    CLOSE_REFRESH_READY,
     is_trading_day,
     latest_saved_prediction,
+    close_refresh_due,
+    refresh_market_close_data,
     refresh_daily_data,
     refresh_first5_prediction,
     seconds_until_next_ist_run,
@@ -40,12 +43,20 @@ def refresh_if_current_session_is_ready() -> None:
     print(f"[scheduler] daily data refreshed: {info}")
 
 
+def refresh_close_data_if_due() -> None:
+    if not close_refresh_due():
+        return
+    info = refresh_market_close_data()
+    print(f"[scheduler] close 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.")
     while True:
         try:
             refresh_if_current_session_is_ready()
+            refresh_close_data_if_due()
         except Exception as exc:
             print(f"[scheduler] current-session refresh failed: {exc}")
         sleep_for = seconds_until_next_ist_run()
@@ -62,6 +73,13 @@ 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)
+        try:
+            refresh_close_data_if_due()
+        except Exception as exc:
+            print(f"[scheduler] close refresh failed: {exc}")
 
 
 if __name__ == "__main__":