diff --git "a/runtime.py" "b/runtime.py" --- "a/runtime.py" +++ "b/runtime.py" @@ -1,29 +1,29 @@ -from __future__ import annotations - +from __future__ import annotations + import json import copy import os import sys import threading -from dataclasses import dataclass -from datetime import date, datetime, time, timedelta -from functools import lru_cache -from pathlib import Path -from typing import Any -from zoneinfo import ZoneInfo - +from dataclasses import dataclass +from datetime import date, datetime, time, timedelta +from functools import lru_cache +from pathlib import Path +from typing import Any +from zoneinfo import ZoneInfo + 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 -except ImportError: # pragma: no cover - production dependency, local fallback below. - mcal = None - - -IST = ZoneInfo("Asia/Kolkata") + +try: + import pandas_market_calendars as mcal +except ImportError: # pragma: no cover - production dependency, local fallback below. + mcal = None + + +IST = ZoneInfo("Asia/Kolkata") YAHOO_NIFTY_SYMBOL = "^NSEI" MARKET_CLOSE = time(15, 30) FIRST5_READY = time(9, 20) @@ -34,12 +34,12 @@ 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" -MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib" -LATEST_PATH = MODEL_DIR / "latest_prediction.csv" -TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet" +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" +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" @@ -66,247 +66,247 @@ REFRESH_WAITING = "waiting_second_payload" REFRESH_REFRESHING = "refreshing" REFRESH_READY = "ready" REFRESH_FAILED = "failed" -REFRESH_NORMAL = "normal" -LIVE_ACCURACY_PATH = MODEL_DIR / "live_accuracy.json" - -DECISION_OVERLAYS = [ - { - "name": "fifth_minute_momentum_flip", - "feature": "m5_ret_1m", - "op": ">=", - "value": 0.0005085411885759201, - }, - { - "name": "vix_stretch_flip", - "feature": "india_vix_close_vs_sma_20", - "op": ">=", - "value": 0.24641908937959742, - }, -] - +REFRESH_NORMAL = "normal" +LIVE_ACCURACY_PATH = MODEL_DIR / "live_accuracy.json" + +DECISION_OVERLAYS = [ + { + "name": "fifth_minute_momentum_flip", + "feature": "m5_ret_1m", + "op": ">=", + "value": 0.0005085411885759201, + }, + { + "name": "vix_stretch_flip", + "feature": "india_vix_close_vs_sma_20", + "op": ">=", + "value": 0.24641908937959742, + }, +] + _dashboard_payload_lock = threading.Lock() _stale_refresh_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: - return None - for name in ("XNSE", "NSE", "BSE"): - try: - return mcal.get_calendar(name) - except Exception: - continue - return None - - -@lru_cache(maxsize=64) -def trading_schedule(start: date, end: date) -> pd.DataFrame: - calendar = _nse_calendar() - if calendar is None: - days = pd.date_range(start=start, end=end, freq="B") - return pd.DataFrame(index=days) - return calendar.schedule(start_date=start, end_date=end) - - -def is_trading_day(day: date) -> bool: - schedule = trading_schedule(day, day) - return not schedule.empty - - -def next_trading_day(start: date) -> date: - end = start + timedelta(days=14) - schedule = trading_schedule(start, end) - if schedule.empty: - day = start - while not is_trading_day(day): - day += timedelta(days=1) - return day - return pd.Timestamp(schedule.index[0]).date() - - -def previous_trading_day(start: date) -> date: - begin = start - timedelta(days=14) - schedule = trading_schedule(begin, start) - if schedule.empty: - day = start - while not is_trading_day(day): - day -= timedelta(days=1) - return day - return pd.Timestamp(schedule.index[-1]).date() - - -class ProbabilityBlend: - def __init__(self, models: list[Any], weights: np.ndarray): - self.models = models - self.weights = np.asarray(weights, dtype="float64") - self.weights = self.weights / self.weights.sum() - - def predict_proba(self, x: pd.DataFrame) -> np.ndarray: - probs = np.column_stack([predict_proba_up(model, x) for model in self.models]) - prob_up = probs @ self.weights - return np.column_stack([1.0 - prob_up, prob_up]) - - -@dataclass(frozen=True) -class Prediction: - input_date: str - first5_start: str - first5_end: str - prediction: str - prob_up: float - confidence: float - threshold: float - model_name: str - is_overridden: bool = False - - def to_dict(self) -> dict[str, Any]: - return { - "input_date": self.input_date, - "first5_start": self.first5_start, - "first5_end": self.first5_end, - "prediction": self.prediction, - "prob_up": self.prob_up, - "confidence": self.confidence, - "threshold": self.threshold, - "model_name": self.model_name, - "is_overridden": getattr(self, "is_overridden", False), - } - - -def predict_proba_up(model: Any, x: pd.DataFrame) -> np.ndarray: - return np.asarray(model.predict_proba(x)[:, 1], dtype="float64") - - -def safe_div(numer: pd.Series | np.ndarray, denom: pd.Series | np.ndarray) -> pd.Series: - n = pd.Series(numer, copy=False) - d = pd.Series(denom, copy=False) - out = pd.Series(np.nan, index=n.index, dtype="float64") - mask = d.notna() & np.isfinite(d.to_numpy(dtype="float64")) & (d != 0) - out.loc[mask] = n.loc[mask].to_numpy(dtype="float64") / d.loc[mask].to_numpy(dtype="float64") - return out - - -def load_model() -> dict[str, Any]: - # Existing artifact was trained as a script, so its custom blend class - # resolves through __main__ when unpickled. - sys.modules["__main__"].ProbabilityBlend = ProbabilityBlend - sys.modules["__main__"].predict_proba_up = predict_proba_up - payload = joblib.load(MODEL_PATH) - payload.setdefault("decision_overlays", DECISION_OVERLAYS) - payload.setdefault("model_name", "nifty_opening_direction_model") - return payload - - -def overlay_mask(frame: pd.DataFrame, overlay: dict[str, object]) -> np.ndarray: - feature = str(overlay["feature"]) - if feature not in frame.columns: - return np.zeros(len(frame), dtype=bool) - series = pd.to_numeric(frame[feature], errors="coerce") - value = float(overlay["value"]) - if overlay["op"] == ">=": - return (series >= value).fillna(False).to_numpy(dtype=bool) - if overlay["op"] == "<=": - return (series <= value).fillna(False).to_numpy(dtype=bool) - raise ValueError(f"Unsupported overlay op: {overlay['op']}") - - -def apply_decision_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, object]]) -> np.ndarray: - adjusted = np.asarray(pred, dtype="int64").copy() - for overlay in overlays: - mask = overlay_mask(frame, overlay) - adjusted[mask] = 1 - adjusted[mask] - return adjusted - - -def directional_confidence(prob_up: np.ndarray, pred: np.ndarray, threshold: float) -> np.ndarray: - prob_up = np.asarray(prob_up, dtype="float64") - pred = np.asarray(pred, dtype="int64") - base_side_prob = np.where(pred == 1, prob_up, 1.0 - prob_up) - threshold_distance = np.abs(prob_up - float(threshold)) - return np.clip(0.50 + threshold_distance, base_side_prob, 0.99) - - -def read_training_dataset() -> pd.DataFrame: - df = pd.read_parquet(OPENING_DATASET_PATH) - for col in ("date", "first5_start", "first5_end"): - if col in df.columns: - df[col] = pd.to_datetime(df[col], errors="coerce") - return df.sort_values("date").reset_index(drop=True) - - + + +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: + return None + for name in ("XNSE", "NSE", "BSE"): + try: + return mcal.get_calendar(name) + except Exception: + continue + return None + + +@lru_cache(maxsize=64) +def trading_schedule(start: date, end: date) -> pd.DataFrame: + calendar = _nse_calendar() + if calendar is None: + days = pd.date_range(start=start, end=end, freq="B") + return pd.DataFrame(index=days) + return calendar.schedule(start_date=start, end_date=end) + + +def is_trading_day(day: date) -> bool: + schedule = trading_schedule(day, day) + return not schedule.empty + + +def next_trading_day(start: date) -> date: + end = start + timedelta(days=14) + schedule = trading_schedule(start, end) + if schedule.empty: + day = start + while not is_trading_day(day): + day += timedelta(days=1) + return day + return pd.Timestamp(schedule.index[0]).date() + + +def previous_trading_day(start: date) -> date: + begin = start - timedelta(days=14) + schedule = trading_schedule(begin, start) + if schedule.empty: + day = start + while not is_trading_day(day): + day -= timedelta(days=1) + return day + return pd.Timestamp(schedule.index[-1]).date() + + +class ProbabilityBlend: + def __init__(self, models: list[Any], weights: np.ndarray): + self.models = models + self.weights = np.asarray(weights, dtype="float64") + self.weights = self.weights / self.weights.sum() + + def predict_proba(self, x: pd.DataFrame) -> np.ndarray: + probs = np.column_stack([predict_proba_up(model, x) for model in self.models]) + prob_up = probs @ self.weights + return np.column_stack([1.0 - prob_up, prob_up]) + + +@dataclass(frozen=True) +class Prediction: + input_date: str + first5_start: str + first5_end: str + prediction: str + prob_up: float + confidence: float + threshold: float + model_name: str + is_overridden: bool = False + + def to_dict(self) -> dict[str, Any]: + return { + "input_date": self.input_date, + "first5_start": self.first5_start, + "first5_end": self.first5_end, + "prediction": self.prediction, + "prob_up": self.prob_up, + "confidence": self.confidence, + "threshold": self.threshold, + "model_name": self.model_name, + "is_overridden": getattr(self, "is_overridden", False), + } + + +def predict_proba_up(model: Any, x: pd.DataFrame) -> np.ndarray: + return np.asarray(model.predict_proba(x)[:, 1], dtype="float64") + + +def safe_div(numer: pd.Series | np.ndarray, denom: pd.Series | np.ndarray) -> pd.Series: + n = pd.Series(numer, copy=False) + d = pd.Series(denom, copy=False) + out = pd.Series(np.nan, index=n.index, dtype="float64") + mask = d.notna() & np.isfinite(d.to_numpy(dtype="float64")) & (d != 0) + out.loc[mask] = n.loc[mask].to_numpy(dtype="float64") / d.loc[mask].to_numpy(dtype="float64") + return out + + +def load_model() -> dict[str, Any]: + # Existing artifact was trained as a script, so its custom blend class + # resolves through __main__ when unpickled. + sys.modules["__main__"].ProbabilityBlend = ProbabilityBlend + sys.modules["__main__"].predict_proba_up = predict_proba_up + payload = joblib.load(MODEL_PATH) + payload.setdefault("decision_overlays", DECISION_OVERLAYS) + payload.setdefault("model_name", "nifty_opening_direction_model") + return payload + + +def overlay_mask(frame: pd.DataFrame, overlay: dict[str, object]) -> np.ndarray: + feature = str(overlay["feature"]) + if feature not in frame.columns: + return np.zeros(len(frame), dtype=bool) + series = pd.to_numeric(frame[feature], errors="coerce") + value = float(overlay["value"]) + if overlay["op"] == ">=": + return (series >= value).fillna(False).to_numpy(dtype=bool) + if overlay["op"] == "<=": + return (series <= value).fillna(False).to_numpy(dtype=bool) + raise ValueError(f"Unsupported overlay op: {overlay['op']}") + + +def apply_decision_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, object]]) -> np.ndarray: + adjusted = np.asarray(pred, dtype="int64").copy() + for overlay in overlays: + mask = overlay_mask(frame, overlay) + adjusted[mask] = 1 - adjusted[mask] + return adjusted + + +def directional_confidence(prob_up: np.ndarray, pred: np.ndarray, threshold: float) -> np.ndarray: + prob_up = np.asarray(prob_up, dtype="float64") + pred = np.asarray(pred, dtype="int64") + base_side_prob = np.where(pred == 1, prob_up, 1.0 - prob_up) + threshold_distance = np.abs(prob_up - float(threshold)) + return np.clip(0.50 + threshold_distance, base_side_prob, 0.99) + + +def read_training_dataset() -> pd.DataFrame: + df = pd.read_parquet(OPENING_DATASET_PATH) + for col in ("date", "first5_start", "first5_end"): + if col in df.columns: + df[col] = pd.to_datetime(df[col], errors="coerce") + return df.sort_values("date").reset_index(drop=True) + + 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): - df.columns = [str(c[0]).lower() for c in df.columns] - else: - df.columns = [str(c).lower().replace(" ", "_") for c in df.columns] - df = df.reset_index() - date_col = next((c for c in df.columns if c.lower() in {"datetime", "date"}), df.columns[0]) - df["date"] = pd.to_datetime(df[date_col], errors="coerce") - if df["date"].dt.tz is None: - df["date"] = df["date"].dt.tz_localize("UTC").dt.tz_convert(IST) - else: - df["date"] = df["date"].dt.tz_convert(IST) - rename = { - "open": "open", - "high": "high", - "low": "low", - "close": "close", - "adj_close": "close", - "volume": "volume", - } - out = pd.DataFrame({"date": df["date"].dt.tz_localize(None)}) - 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") + if df.empty: + return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"]) + if isinstance(df.columns, pd.MultiIndex): + df.columns = [str(c[0]).lower() for c in df.columns] + else: + df.columns = [str(c).lower().replace(" ", "_") for c in df.columns] + df = df.reset_index() + date_col = next((c for c in df.columns if c.lower() in {"datetime", "date"}), df.columns[0]) + df["date"] = pd.to_datetime(df[date_col], errors="coerce") + if df["date"].dt.tz is None: + df["date"] = df["date"].dt.tz_localize("UTC").dt.tz_convert(IST) + else: + df["date"] = df["date"].dt.tz_convert(IST) + rename = { + "open": "open", + "high": "high", + "low": "low", + "close": "close", + "adj_close": "close", + "volume": "volume", + } + out = pd.DataFrame({"date": df["date"].dt.tz_localize(None)}) + 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") @@ -382,18 +382,18 @@ def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame: 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: 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) - else: - combined = new_rows.copy() + if path.exists(): + existing = pd.read_parquet(path) + combined = pd.concat([existing, new_rows], ignore_index=True) + else: + combined = new_rows.copy() combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True) combined.to_parquet(path, index=False, compression="zstd") return combined @@ -402,135 +402,135 @@ def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) - def append_prediction_history(path: Path, row: dict[str, Any], subset: list[str]) -> pd.DataFrame: frame = pd.DataFrame([row]) return append_parquet_rows(path, frame, subset) - - -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.") - bars = minutes.copy() - bars["dt"] = pd.to_datetime(bars["date"], errors="coerce") - bars["session_date"] = bars["dt"].dt.normalize() - if session_date is None: - session_ts = bars["session_date"].max() - else: - session_ts = pd.Timestamp(session_date).normalize() - day = bars[bars["session_date"] == session_ts].sort_values("dt").copy() - start_dt = pd.Timestamp.combine(session_ts.date(), time(9, 15)) - end_dt = pd.Timestamp.combine(session_ts.date(), time(9, 19)) - first5 = day[(day["dt"] >= start_dt) & (day["dt"] <= end_dt)].head(5).copy() - if len(first5) < 5: - raise RuntimeError(f"Need 5 opening bars for {session_ts.date()}, got {len(first5)}.") - first5["minute_index"] = np.arange(len(first5)) - first5["ret_1m"] = first5["close"].pct_change(fill_method=None) - first5["range_pct_1m"] = safe_div(first5["high"] - first5["low"], first5["open"]) - first5["body_pct_1m"] = safe_div(first5["close"] - first5["open"], first5["open"]) - row = { - "date": session_ts, - "first5_start": first5["dt"].iloc[0], - "first5_end": first5["dt"].iloc[-1], - "first5_open": first5["open"].iloc[0], - "first5_high": first5["high"].max(), - "first5_low": first5["low"].min(), - "first5_close": first5["close"].iloc[-1], - "first5_volume": first5["volume"].sum() if "volume" in first5 else 0.0, - "first5_bars": len(first5), - "first5_last_1m_ret": first5["ret_1m"].iloc[-1], - "first5_ret_std": first5["ret_1m"].std(), - } - row["first5_return"] = (row["first5_close"] - row["first5_open"]) / row["first5_open"] - row["first5_range_pct"] = (row["first5_high"] - row["first5_low"]) / row["first5_open"] - first5_range = row["first5_high"] - row["first5_low"] - row["first5_body_to_range"] = (row["first5_close"] - row["first5_open"]) / first5_range if first5_range else np.nan - row["first5_close_location"] = (row["first5_close"] - row["first5_low"]) / first5_range if first5_range else np.nan - for idx, (_, candle) in enumerate(first5.iterrows(), start=1): - for field in ("open", "high", "low", "close", "ret_1m", "range_pct_1m", "body_pct_1m"): - row[f"m{idx}_{field}"] = candle[field] - row[f"m{idx}_close_vs_first5_open"] = (candle["close"] - row["first5_open"]) / row["first5_open"] - row[f"m{idx}_range_share"] = (candle["high"] - candle["low"]) / first5_range if first5_range else np.nan - row["first5_return_accel"] = row["m5_ret_1m"] - row["m2_ret_1m"] - row["first5_last2_return"] = (row["m5_close"] - row["m4_open"]) / row["m4_open"] - row["first5_first2_return"] = (row["m2_close"] - row["m1_open"]) / row["m1_open"] - row["first5_reversal"] = np.sign(row["first5_first2_return"]) * -np.sign(row["first5_last2_return"]) - row["dow"] = session_ts.dayofweek - row["dom"] = session_ts.day - row["month"] = session_ts.month - return pd.DataFrame([row]) - - -def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame: - dataset = read_training_dataset() - latest_context = dataset.iloc[[-1]].copy() - output = latest_context.copy() - for col in first5_row.columns: - output[col] = first5_row[col].iloc[0] - if {"first5_open", "nifty_close"}.issubset(output.columns): - output["first5_gap_from_prev_close"] = (output["first5_open"] - output["nifty_close"]) / output["nifty_close"] - output["first5_close_vs_prev_close"] = (output["first5_close"] - output["nifty_close"]) / output["nifty_close"] - if {"first5_range_pct", "nifty_range_pct"}.issubset(output.columns): - output["first5_range_vs_prev_range"] = output["first5_range_pct"] / output["nifty_range_pct"] - if {"first5_return", "nifty_ret_1"}.issubset(output.columns): - output["first5_return_x_prev_ret"] = output["first5_return"] * output["nifty_ret_1"] - output["gap_x_prev_ret"] = output["first5_gap_from_prev_close"] * output["nifty_ret_1"] - if {"first5_return", "banknifty_ret_1"}.issubset(output.columns): - output["first5_return_x_bank_ret_1"] = output["first5_return"] * output["banknifty_ret_1"] - if {"first5_range_pct", "india_vix_ret_1"}.issubset(output.columns): - output["first5_range_x_vix_ret_1"] = output["first5_range_pct"] * output["india_vix_ret_1"] - output["target"] = np.nan - output["day_return"] = np.nan - return output - - + + +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.") + bars = minutes.copy() + bars["dt"] = pd.to_datetime(bars["date"], errors="coerce") + bars["session_date"] = bars["dt"].dt.normalize() + if session_date is None: + session_ts = bars["session_date"].max() + else: + session_ts = pd.Timestamp(session_date).normalize() + day = bars[bars["session_date"] == session_ts].sort_values("dt").copy() + start_dt = pd.Timestamp.combine(session_ts.date(), time(9, 15)) + end_dt = pd.Timestamp.combine(session_ts.date(), time(9, 19)) + first5 = day[(day["dt"] >= start_dt) & (day["dt"] <= end_dt)].head(5).copy() + if len(first5) < 5: + raise RuntimeError(f"Need 5 opening bars for {session_ts.date()}, got {len(first5)}.") + first5["minute_index"] = np.arange(len(first5)) + first5["ret_1m"] = first5["close"].pct_change(fill_method=None) + first5["range_pct_1m"] = safe_div(first5["high"] - first5["low"], first5["open"]) + first5["body_pct_1m"] = safe_div(first5["close"] - first5["open"], first5["open"]) + row = { + "date": session_ts, + "first5_start": first5["dt"].iloc[0], + "first5_end": first5["dt"].iloc[-1], + "first5_open": first5["open"].iloc[0], + "first5_high": first5["high"].max(), + "first5_low": first5["low"].min(), + "first5_close": first5["close"].iloc[-1], + "first5_volume": first5["volume"].sum() if "volume" in first5 else 0.0, + "first5_bars": len(first5), + "first5_last_1m_ret": first5["ret_1m"].iloc[-1], + "first5_ret_std": first5["ret_1m"].std(), + } + row["first5_return"] = (row["first5_close"] - row["first5_open"]) / row["first5_open"] + row["first5_range_pct"] = (row["first5_high"] - row["first5_low"]) / row["first5_open"] + first5_range = row["first5_high"] - row["first5_low"] + row["first5_body_to_range"] = (row["first5_close"] - row["first5_open"]) / first5_range if first5_range else np.nan + row["first5_close_location"] = (row["first5_close"] - row["first5_low"]) / first5_range if first5_range else np.nan + for idx, (_, candle) in enumerate(first5.iterrows(), start=1): + for field in ("open", "high", "low", "close", "ret_1m", "range_pct_1m", "body_pct_1m"): + row[f"m{idx}_{field}"] = candle[field] + row[f"m{idx}_close_vs_first5_open"] = (candle["close"] - row["first5_open"]) / row["first5_open"] + row[f"m{idx}_range_share"] = (candle["high"] - candle["low"]) / first5_range if first5_range else np.nan + row["first5_return_accel"] = row["m5_ret_1m"] - row["m2_ret_1m"] + row["first5_last2_return"] = (row["m5_close"] - row["m4_open"]) / row["m4_open"] + row["first5_first2_return"] = (row["m2_close"] - row["m1_open"]) / row["m1_open"] + row["first5_reversal"] = np.sign(row["first5_first2_return"]) * -np.sign(row["first5_last2_return"]) + row["dow"] = session_ts.dayofweek + row["dom"] = session_ts.day + row["month"] = session_ts.month + return pd.DataFrame([row]) + + +def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame: + dataset = read_training_dataset() + latest_context = dataset.iloc[[-1]].copy() + output = latest_context.copy() + for col in first5_row.columns: + output[col] = first5_row[col].iloc[0] + if {"first5_open", "nifty_close"}.issubset(output.columns): + output["first5_gap_from_prev_close"] = (output["first5_open"] - output["nifty_close"]) / output["nifty_close"] + output["first5_close_vs_prev_close"] = (output["first5_close"] - output["nifty_close"]) / output["nifty_close"] + if {"first5_range_pct", "nifty_range_pct"}.issubset(output.columns): + output["first5_range_vs_prev_range"] = output["first5_range_pct"] / output["nifty_range_pct"] + if {"first5_return", "nifty_ret_1"}.issubset(output.columns): + output["first5_return_x_prev_ret"] = output["first5_return"] * output["nifty_ret_1"] + output["gap_x_prev_ret"] = output["first5_gap_from_prev_close"] * output["nifty_ret_1"] + if {"first5_return", "banknifty_ret_1"}.issubset(output.columns): + output["first5_return_x_bank_ret_1"] = output["first5_return"] * output["banknifty_ret_1"] + if {"first5_range_pct", "india_vix_ret_1"}.issubset(output.columns): + output["first5_range_x_vix_ret_1"] = output["first5_range_pct"] * output["india_vix_ret_1"] + output["target"] = np.nan + output["day_return"] = np.nan + return output + + def predict_row(row: pd.DataFrame) -> Prediction: - payload = load_model() - model = payload["model"] - features = payload["features"] - threshold = float(payload["threshold"]) - missing = [c for c in features if c not in row.columns] - if missing: - raise RuntimeError(f"Feature row is missing {len(missing)} features; first missing: {missing[:5]}") - prob_up = predict_proba_up(model, row[features]) - raw_pred = (prob_up >= threshold).astype("int64") - pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS)) - is_overridden = bool(raw_pred[0] != pred[0]) - confidence = directional_confidence(prob_up, pred, threshold) + payload = load_model() + model = payload["model"] + features = payload["features"] + threshold = float(payload["threshold"]) + missing = [c for c in features if c not in row.columns] + if missing: + raise RuntimeError(f"Feature row is missing {len(missing)} features; first missing: {missing[:5]}") + prob_up = predict_proba_up(model, row[features]) + raw_pred = (prob_up >= threshold).astype("int64") + pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS)) + is_overridden = bool(raw_pred[0] != pred[0]) + confidence = directional_confidence(prob_up, pred, threshold) prediction = Prediction( input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(), first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])), first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])), prediction="UP" if int(pred[0]) == 1 else "DOWN", - prob_up=float(prob_up[0]), - confidence=float(confidence[0]), - threshold=threshold, + prob_up=float(prob_up[0]), + confidence=float(confidence[0]), + threshold=threshold, model_name=str(payload.get("model_name", "nifty_opening_direction_model")), is_overridden=is_overridden, ) @@ -545,37 +545,37 @@ def predict_row(row: pd.DataFrame) -> Prediction: ["target_date"], ) return prediction - - -def _file_cache_key(path: Path) -> tuple[str, int | None, int | None]: - try: - stat = path.stat() - except FileNotFoundError: - return (str(path), None, None) - return (str(path), stat.st_mtime_ns, stat.st_size) - - -@lru_cache(maxsize=16) -def _latest_saved_prediction_cached(latest_key: tuple[str, int | None, int | None], summary_key: tuple[str, int | None, int | None]) -> dict[str, Any]: - latest_path = Path(latest_key[0]) - if latest_path.exists(): - return pd.read_csv(latest_path).iloc[-1].to_dict() - summary_path = Path(summary_key[0]) - if summary_path.exists(): - return json.loads(summary_path.read_text(encoding="utf-8")) - raise FileNotFoundError("No latest prediction is available yet.") - - -def latest_saved_prediction() -> dict[str, Any]: - return dict(_latest_saved_prediction_cached(_file_cache_key(LATEST_PATH), _file_cache_key(MODEL_DIR / "summary.json"))) - - + + +def _file_cache_key(path: Path) -> tuple[str, int | None, int | None]: + try: + stat = path.stat() + except FileNotFoundError: + return (str(path), None, None) + return (str(path), stat.st_mtime_ns, stat.st_size) + + +@lru_cache(maxsize=16) +def _latest_saved_prediction_cached(latest_key: tuple[str, int | None, int | None], summary_key: tuple[str, int | None, int | None]) -> dict[str, Any]: + latest_path = Path(latest_key[0]) + if latest_path.exists(): + return pd.read_csv(latest_path).iloc[-1].to_dict() + summary_path = Path(summary_key[0]) + if summary_path.exists(): + return json.loads(summary_path.read_text(encoding="utf-8")) + raise FileNotFoundError("No latest prediction is available yet.") + + +def latest_saved_prediction() -> dict[str, Any]: + return dict(_latest_saved_prediction_cached(_file_cache_key(LATEST_PATH), _file_cache_key(MODEL_DIR / "summary.json"))) + + def _latest_saved_prediction_uncached() -> dict[str, Any]: - if LATEST_PATH.exists(): - return pd.read_csv(LATEST_PATH).iloc[-1].to_dict() - summary_path = MODEL_DIR / "summary.json" - if summary_path.exists(): - return json.loads(summary_path.read_text(encoding="utf-8")) + if LATEST_PATH.exists(): + return pd.read_csv(LATEST_PATH).iloc[-1].to_dict() + summary_path = MODEL_DIR / "summary.json" + if summary_path.exists(): + return json.loads(summary_path.read_text(encoding="utf-8")) raise FileNotFoundError("No latest prediction is available yet.") @@ -644,6 +644,86 @@ def _read_daily_forecaster_latest(summary: dict[str, Any]) -> dict[str, Any] | N } +def _parse_iso_date(value: Any) -> date | None: + try: + return date.fromisoformat(str(value or "")[:10]) + except Exception: + return None + + +def _archive_tomorrow_latest_to_history() -> None: + if not TOMORROW_LATEST_PATH.exists(): + return + try: + row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() + cleaned = {k: (None if pd.isna(v) else v) for k, v in row.items()} + pred = str(cleaned.get("prediction", "")).upper() + if pred in {"UP", "DOWN"} and cleaned.get("target_date"): + _record_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH, cleaned, ["target_date"]) + except Exception: + pass + + +def _find_tomorrow_prediction_for_target(target_day: date) -> dict[str, Any] | None: + """Return the Tomorrow prediction that targets ``target_day``.""" + target_iso = target_day.isoformat() + tom_history = _load_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH) + if not tom_history.empty: + for col in ("date", "input_date", "target_date", "forecast_date"): + if col in tom_history.columns: + tom_history[col] = pd.to_datetime(tom_history[col], errors="coerce") + + if not tom_history.empty and "target_date" in tom_history.columns: + rows = tom_history[tom_history["target_date"].dt.date == target_day] + if not rows.empty: + return rows.iloc[-1].to_dict() + + if TOMORROW_LATEST_PATH.exists(): + try: + row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() + if _parse_iso_date(row.get("target_date")) == target_day: + return row + except Exception: + pass + + input_day = previous_trading_day(target_day - timedelta(days=1)) + if not tom_history.empty and "input_date" in tom_history.columns: + rows = tom_history[tom_history["input_date"].dt.date == input_day] + if not rows.empty: + row = rows.iloc[-1].to_dict() + if _parse_iso_date(row.get("target_date")) in {None, target_day}: + return row + + if TOMORROW_LATEST_PATH.exists(): + try: + row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() + if _parse_iso_date(row.get("input_date")) == input_day: + return row + except Exception: + pass + + tomorrow_test = load_tomorrow_test_predictions() + if not tomorrow_test.empty: + for col in ("target_date", "date"): + if col in tomorrow_test.columns: + test_dates = pd.to_datetime(tomorrow_test[col], errors="coerce").dt.date + rows = tomorrow_test[test_dates == target_day] + if not rows.empty: + return rows.iloc[-1].to_dict() + + ledger = load_live_accuracy() + for entry in ledger.get("tomorrow", {}).get("entries", []): + if str(entry.get("date", ""))[:10] == target_iso: + pred = str(entry.get("prediction", "")).upper() + if pred in {"UP", "DOWN"}: + return { + "target_date": target_iso, + "prediction": pred, + "source": entry.get("source", "live"), + } + return None + + def sync_daily_forecaster_outputs() -> dict[str, Any] | None: summary = _read_daily_forecaster_summary() if summary is None: @@ -651,7 +731,25 @@ def sync_daily_forecaster_outputs() -> dict[str, Any] | None: latest = _read_daily_forecaster_latest(summary) TOMORROW_SUMMARY_PATH.write_text(json.dumps(summary, indent=2), encoding="utf-8") if latest is not None: - pd.DataFrame([latest]).to_csv(TOMORROW_LATEST_PATH, index=False) + _archive_tomorrow_latest_to_history() + keep_existing = False + if TOMORROW_LATEST_PATH.exists(): + try: + existing = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() + existing_input = _parse_iso_date(existing.get("input_date")) + forecaster_input = _parse_iso_date(latest.get("input_date")) + existing_pred = str(existing.get("prediction", "")).upper() + if ( + existing_input is not None + and forecaster_input is not None + and existing_input > forecaster_input + and existing_pred in {"UP", "DOWN"} + ): + keep_existing = True + except Exception: + keep_existing = False + if not keep_existing: + pd.DataFrame([latest]).to_csv(TOMORROW_LATEST_PATH, index=False) if DAILY_FORECASTER_PREDICTIONS_PATH.exists(): predictions = pd.read_csv(DAILY_FORECASTER_PREDICTIONS_PATH) if "symbol" in predictions.columns: @@ -689,14 +787,14 @@ 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)), - } - - + 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]: synced = sync_daily_forecaster_outputs() if synced is not None and TOMORROW_SUMMARY_PATH.exists(): @@ -714,8 +812,8 @@ def load_tomorrow_summary() -> dict[str, Any]: "n_test": 186, "feature_count": 204, } - - + + def latest_tomorrow_prediction() -> dict[str, Any]: sync_daily_forecaster_outputs() latest_daily = latest_parquet_date(NIFTY_1D_PATH) @@ -754,161 +852,161 @@ def latest_tomorrow_prediction() -> dict[str, Any]: 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 load_tplus1_summary() -> dict[str, Any]: - if TPLUS1_SUMMARY_PATH.exists(): - return json.loads(TPLUS1_SUMMARY_PATH.read_text(encoding="utf-8")) - return { - "model_name": "logistic_regression_l1_C0.35_balanced", - "target": "T+1 NIFTY 50 close greater than T 14:20 close", - "window_start": "14:00", - "window_end": "14:20", - "threshold": 0.578, - "validation_accuracy": 0.66, - "test_accuracy": 0.6368421052631579, - "baseline_test_accuracy": 0.5052631578947369, - "test_rows": 190, - "feature_count": 40, - } - - -def latest_tplus1_prediction() -> dict[str, Any]: - if TPLUS1_LATEST_PATH.exists(): - row = pd.read_csv(TPLUS1_LATEST_PATH).iloc[-1].to_dict() - return {k: (None if pd.isna(v) else v) for k, v in row.items()} - summary = load_tplus1_summary() - return { - "input_date": summary.get("latest_input_date"), - "target_date": None, - "forecast_for": summary.get("latest_forecast_for"), - "prediction": summary.get("latest_prediction"), - "prob_up": summary.get("latest_prob_up"), - "confidence": summary.get("latest_confidence"), - "threshold": summary.get("threshold"), - "model_name": summary.get("model_name", "logistic_regression_l1_C0.35_balanced"), - "validation_accuracy": summary.get("validation_accuracy"), - "test_accuracy": summary.get("test_accuracy"), - } - - -def _minute_frame_for_tplus1() -> pd.DataFrame: - minute = pd.read_parquet(NIFTY_1M_PATH) - minute = minute.copy() - minute["dt"] = pd.to_datetime(minute["date"], errors="coerce") - for col in ("open", "high", "low", "close", "volume"): - if col in minute.columns: - minute[col] = pd.to_numeric(minute[col], errors="coerce") - minute = minute.dropna(subset=["dt", "open", "high", "low", "close"]).sort_values("dt").reset_index(drop=True) - minute["session_date"] = minute["dt"].dt.normalize() - minute["time"] = minute["dt"].dt.strftime("%H:%M") - return minute - - -def _build_tplus1_session_features(minute: pd.DataFrame) -> pd.DataFrame: - window = minute[(minute["time"] >= "14:00") & (minute["time"] <= "14:20")].copy() - window["minute_offset"] = window.groupby("session_date", sort=True).cumcount() - grouped = window.groupby("session_date", sort=True) - base = grouped.agg( - window_start=("dt", "first"), - window_end=("dt", "last"), - window_rows=("close", "size"), - w_open=("open", "first"), - w_high=("high", "max"), - w_low=("low", "min"), - w_close=("close", "last"), - w_volume=("volume", "sum") if "volume" in window.columns else ("close", "size"), - ).reset_index().rename(columns={"session_date": "date"}) - base = base[base["window_rows"] == 21].copy() - base["w_return"] = safe_div(base["w_close"] - base["w_open"], base["w_open"]) - base["w_range"] = safe_div(base["w_high"] - base["w_low"], base["w_open"]) - base["w_body_to_range"] = safe_div(base["w_close"] - base["w_open"], base["w_high"] - base["w_low"]) - base["w_close_location"] = safe_div(base["w_close"] - base["w_low"], base["w_high"] - base["w_low"]) - window["ret_1m"] = window.groupby("session_date")["close"].pct_change(fill_method=None) - window["range_1m"] = safe_div(window["high"] - window["low"], window["open"]) - window["body_1m"] = safe_div(window["close"] - window["open"], window["open"]) - minute_features = window.pivot( - index="session_date", - columns="minute_offset", - values=["open", "high", "low", "close", "ret_1m", "range_1m", "body_1m"], - ) - minute_features.columns = [f"m{int(offset):02d}_{field}" for field, offset in minute_features.columns] - minute_features = minute_features.reset_index().rename(columns={"session_date": "date"}) - session_close = ( - minute.groupby("session_date", sort=True) - .agg(day_close=("close", "last")) - .reset_index() - .rename(columns={"session_date": "date"}) - ) - frame = base.merge(minute_features, on="date", how="left").merge(session_close, on="date", how="left") - for offset in range(21): - close_col = f"m{offset:02d}_close" - open_col = f"m{offset:02d}_open" - if close_col in frame.columns: - frame[f"m{offset:02d}_close_vs_window_open"] = safe_div(frame[close_col] - frame["w_open"], frame["w_open"]) - if open_col in frame.columns and close_col in frame.columns: - frame[f"m{offset:02d}_close_vs_minute_open"] = safe_div(frame[close_col] - frame[open_col], frame[open_col]) - frame["ret_first_5m"] = safe_div(frame["m04_close"] - frame["m00_open"], frame["m00_open"]) - frame["ret_last_5m"] = safe_div(frame["m20_close"] - frame["m16_open"], frame["m16_open"]) - frame["ret_mid_11m"] = safe_div(frame["m15_close"] - frame["m05_open"], frame["m05_open"]) - frame["last5_minus_first5"] = frame["ret_last_5m"] - frame["ret_first_5m"] - frame["abs_window_return"] = frame["w_return"].abs() - frame["dow"] = frame["date"].dt.dayofweek - frame["dom"] = frame["date"].dt.day - frame["month"] = frame["date"].dt.month - return frame.sort_values("date").reset_index(drop=True) - - -def _add_tplus1_target_features(features: pd.DataFrame) -> pd.DataFrame: - frame = features.copy() - frame["target_date"] = frame["date"].shift(-1) - frame["target_close"] = frame["day_close"].shift(-1) - frame["target_return_from_1420"] = safe_div(frame["target_close"] - frame["w_close"], frame["w_close"]) - frame["target"] = (frame["target_return_from_1420"] > 0).astype("float64") - frame.loc[frame["target_close"].isna(), "target"] = np.nan - for lag in (1, 2, 3, 5, 10): - frame[f"prev_target_lag{lag}"] = frame["target"].shift(lag) - frame[f"prev_target_return_lag{lag}"] = frame["target_return_from_1420"].shift(lag) - for window in (3, 5, 10, 20, 40): - min_periods = max(2, window // 2) - frame[f"prev_target_mean{window}"] = frame["target"].shift(1).rolling(window, min_periods=min_periods).mean() - shifted_return = frame["target_return_from_1420"].shift(1) - frame[f"prev_target_return_mean{window}"] = shifted_return.rolling(window, min_periods=min_periods).mean() - frame[f"prev_target_return_std{window}"] = shifted_return.rolling(window, min_periods=min_periods).std() - return frame - - -def _apply_tplus1_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, Any]]) -> np.ndarray: - adjusted = np.asarray(pred, dtype="int64").copy() - for overlay in overlays: - feature = str(overlay.get("feature", "")) - if feature not in frame.columns: - continue - series = pd.to_numeric(frame[feature], errors="coerce") - value = float(overlay.get("value", 0.0)) - if overlay.get("op") == "<=": - mask = (series <= value).fillna(False).to_numpy(dtype=bool) - else: - mask = (series >= value).fillna(False).to_numpy(dtype=bool) - action = overlay.get("action") - if action == "up": - adjusted[mask] = 1 - elif action == "down": - adjusted[mask] = 0 - elif action == "flip": - adjusted[mask] = 1 - adjusted[mask] - return adjusted - - + "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 load_tplus1_summary() -> dict[str, Any]: + if TPLUS1_SUMMARY_PATH.exists(): + return json.loads(TPLUS1_SUMMARY_PATH.read_text(encoding="utf-8")) + return { + "model_name": "logistic_regression_l1_C0.35_balanced", + "target": "T+1 NIFTY 50 close greater than T 14:20 close", + "window_start": "14:00", + "window_end": "14:20", + "threshold": 0.578, + "validation_accuracy": 0.66, + "test_accuracy": 0.6368421052631579, + "baseline_test_accuracy": 0.5052631578947369, + "test_rows": 190, + "feature_count": 40, + } + + +def latest_tplus1_prediction() -> dict[str, Any]: + if TPLUS1_LATEST_PATH.exists(): + row = pd.read_csv(TPLUS1_LATEST_PATH).iloc[-1].to_dict() + return {k: (None if pd.isna(v) else v) for k, v in row.items()} + summary = load_tplus1_summary() + return { + "input_date": summary.get("latest_input_date"), + "target_date": None, + "forecast_for": summary.get("latest_forecast_for"), + "prediction": summary.get("latest_prediction"), + "prob_up": summary.get("latest_prob_up"), + "confidence": summary.get("latest_confidence"), + "threshold": summary.get("threshold"), + "model_name": summary.get("model_name", "logistic_regression_l1_C0.35_balanced"), + "validation_accuracy": summary.get("validation_accuracy"), + "test_accuracy": summary.get("test_accuracy"), + } + + +def _minute_frame_for_tplus1() -> pd.DataFrame: + minute = pd.read_parquet(NIFTY_1M_PATH) + minute = minute.copy() + minute["dt"] = pd.to_datetime(minute["date"], errors="coerce") + for col in ("open", "high", "low", "close", "volume"): + if col in minute.columns: + minute[col] = pd.to_numeric(minute[col], errors="coerce") + minute = minute.dropna(subset=["dt", "open", "high", "low", "close"]).sort_values("dt").reset_index(drop=True) + minute["session_date"] = minute["dt"].dt.normalize() + minute["time"] = minute["dt"].dt.strftime("%H:%M") + return minute + + +def _build_tplus1_session_features(minute: pd.DataFrame) -> pd.DataFrame: + window = minute[(minute["time"] >= "14:00") & (minute["time"] <= "14:20")].copy() + window["minute_offset"] = window.groupby("session_date", sort=True).cumcount() + grouped = window.groupby("session_date", sort=True) + base = grouped.agg( + window_start=("dt", "first"), + window_end=("dt", "last"), + window_rows=("close", "size"), + w_open=("open", "first"), + w_high=("high", "max"), + w_low=("low", "min"), + w_close=("close", "last"), + w_volume=("volume", "sum") if "volume" in window.columns else ("close", "size"), + ).reset_index().rename(columns={"session_date": "date"}) + base = base[base["window_rows"] == 21].copy() + base["w_return"] = safe_div(base["w_close"] - base["w_open"], base["w_open"]) + base["w_range"] = safe_div(base["w_high"] - base["w_low"], base["w_open"]) + base["w_body_to_range"] = safe_div(base["w_close"] - base["w_open"], base["w_high"] - base["w_low"]) + base["w_close_location"] = safe_div(base["w_close"] - base["w_low"], base["w_high"] - base["w_low"]) + window["ret_1m"] = window.groupby("session_date")["close"].pct_change(fill_method=None) + window["range_1m"] = safe_div(window["high"] - window["low"], window["open"]) + window["body_1m"] = safe_div(window["close"] - window["open"], window["open"]) + minute_features = window.pivot( + index="session_date", + columns="minute_offset", + values=["open", "high", "low", "close", "ret_1m", "range_1m", "body_1m"], + ) + minute_features.columns = [f"m{int(offset):02d}_{field}" for field, offset in minute_features.columns] + minute_features = minute_features.reset_index().rename(columns={"session_date": "date"}) + session_close = ( + minute.groupby("session_date", sort=True) + .agg(day_close=("close", "last")) + .reset_index() + .rename(columns={"session_date": "date"}) + ) + frame = base.merge(minute_features, on="date", how="left").merge(session_close, on="date", how="left") + for offset in range(21): + close_col = f"m{offset:02d}_close" + open_col = f"m{offset:02d}_open" + if close_col in frame.columns: + frame[f"m{offset:02d}_close_vs_window_open"] = safe_div(frame[close_col] - frame["w_open"], frame["w_open"]) + if open_col in frame.columns and close_col in frame.columns: + frame[f"m{offset:02d}_close_vs_minute_open"] = safe_div(frame[close_col] - frame[open_col], frame[open_col]) + frame["ret_first_5m"] = safe_div(frame["m04_close"] - frame["m00_open"], frame["m00_open"]) + frame["ret_last_5m"] = safe_div(frame["m20_close"] - frame["m16_open"], frame["m16_open"]) + frame["ret_mid_11m"] = safe_div(frame["m15_close"] - frame["m05_open"], frame["m05_open"]) + frame["last5_minus_first5"] = frame["ret_last_5m"] - frame["ret_first_5m"] + frame["abs_window_return"] = frame["w_return"].abs() + frame["dow"] = frame["date"].dt.dayofweek + frame["dom"] = frame["date"].dt.day + frame["month"] = frame["date"].dt.month + return frame.sort_values("date").reset_index(drop=True) + + +def _add_tplus1_target_features(features: pd.DataFrame) -> pd.DataFrame: + frame = features.copy() + frame["target_date"] = frame["date"].shift(-1) + frame["target_close"] = frame["day_close"].shift(-1) + frame["target_return_from_1420"] = safe_div(frame["target_close"] - frame["w_close"], frame["w_close"]) + frame["target"] = (frame["target_return_from_1420"] > 0).astype("float64") + frame.loc[frame["target_close"].isna(), "target"] = np.nan + for lag in (1, 2, 3, 5, 10): + frame[f"prev_target_lag{lag}"] = frame["target"].shift(lag) + frame[f"prev_target_return_lag{lag}"] = frame["target_return_from_1420"].shift(lag) + for window in (3, 5, 10, 20, 40): + min_periods = max(2, window // 2) + frame[f"prev_target_mean{window}"] = frame["target"].shift(1).rolling(window, min_periods=min_periods).mean() + shifted_return = frame["target_return_from_1420"].shift(1) + frame[f"prev_target_return_mean{window}"] = shifted_return.rolling(window, min_periods=min_periods).mean() + frame[f"prev_target_return_std{window}"] = shifted_return.rolling(window, min_periods=min_periods).std() + return frame + + +def _apply_tplus1_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, Any]]) -> np.ndarray: + adjusted = np.asarray(pred, dtype="int64").copy() + for overlay in overlays: + feature = str(overlay.get("feature", "")) + if feature not in frame.columns: + continue + series = pd.to_numeric(frame[feature], errors="coerce") + value = float(overlay.get("value", 0.0)) + if overlay.get("op") == "<=": + mask = (series <= value).fillna(False).to_numpy(dtype=bool) + else: + mask = (series >= value).fillna(False).to_numpy(dtype=bool) + action = overlay.get("action") + if action == "up": + adjusted[mask] = 1 + elif action == "down": + adjusted[mask] = 0 + elif action == "flip": + adjusted[mask] = 1 - adjusted[mask] + 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}") @@ -930,28 +1028,28 @@ def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any 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]}") - prob_up = predict_proba_up(payload["model"], row[features]) - raw_pred = (prob_up >= threshold).astype("int64") - overlay_payload = payload.get("decision_overlay") - overlays = overlay_payload.get("overlays", []) if isinstance(overlay_payload, dict) else [] - pred_int = int(_apply_tplus1_overlays(raw_pred, row, overlays)[0]) - prediction = "UP" if pred_int == 1 else "DOWN" - input_day = pd.to_datetime(row["date"].iloc[0]).date() - target_day = next_trading_day(input_day + timedelta(days=1)) - summary = load_tplus1_summary() + 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]}") + prob_up = predict_proba_up(payload["model"], row[features]) + raw_pred = (prob_up >= threshold).astype("int64") + overlay_payload = payload.get("decision_overlay") + overlays = overlay_payload.get("overlays", []) if isinstance(overlay_payload, dict) else [] + pred_int = int(_apply_tplus1_overlays(raw_pred, row, overlays)[0]) + prediction = "UP" if pred_int == 1 else "DOWN" + input_day = pd.to_datetime(row["date"].iloc[0]).date() + target_day = next_trading_day(input_day + timedelta(days=1)) + summary = load_tplus1_summary() out = { "input_date": input_day.isoformat(), "target_date": target_day.isoformat(), "forecast_for": f"next trading session after {input_day.isoformat()}", - "prediction": prediction, - "prob_up": float(prob_up[0]), - "confidence": float(max(prob_up[0], 1.0 - prob_up[0])), - "threshold": threshold, - "model_name": str(payload.get("model_name", summary.get("model_name", "nifty_1420_tplus1_logistic_model"))), - "decision_overlay": summary.get("decision_overlay"), + "prediction": prediction, + "prob_up": float(prob_up[0]), + "confidence": float(max(prob_up[0], 1.0 - prob_up[0])), + "threshold": threshold, + "model_name": str(payload.get("model_name", summary.get("model_name", "nifty_1420_tplus1_logistic_model"))), + "decision_overlay": summary.get("decision_overlay"), "validation_accuracy": summary.get("validation_accuracy"), "test_accuracy": summary.get("test_accuracy"), "accuracy_goal": summary.get("accuracy_goal"), @@ -961,31 +1059,32 @@ def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any _record_prediction_history(TPLUS1_PREDICTION_HISTORY_PATH, out, ["target_date"]) clear_dashboard_payload_cache() return out - - -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 _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]: + _archive_tomorrow_latest_to_history() synced = sync_daily_forecaster_outputs() if synced is not None and TOMORROW_LATEST_PATH.exists(): latest = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() @@ -1004,18 +1103,18 @@ def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, A return cleaned 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)) + 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(), @@ -1035,95 +1134,96 @@ def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, A 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"], + "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") clear_dashboard_payload_cache() 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: - out = out.tail(limit) - for col in out.columns: - if pd.api.types.is_datetime64_any_dtype(out[col]): - out[col] = out[col].dt.strftime("%Y-%m-%d %H:%M:%S") - out = out.replace({np.nan: None}) - return out.to_dict(orient="records") - - -def load_model_summary() -> dict[str, Any]: - summary_path = MODEL_DIR / "summary.json" - if not summary_path.exists(): - return {} - return json.loads(summary_path.read_text(encoding="utf-8")) - - -def load_candidate_results() -> list[dict[str, Any]]: - path = MODEL_DIR / "candidate_results.csv" - if not path.exists(): - return [] - return _json_ready_frame(pd.read_csv(path).head(12)) - - -def load_test_predictions() -> pd.DataFrame: - if not TEST_PREDICTIONS_PATH.exists(): - return pd.DataFrame() - df = pd.read_parquet(TEST_PREDICTIONS_PATH) - df["date"] = pd.to_datetime(df["date"], errors="coerce") - 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 load_tplus1_test_predictions() -> pd.DataFrame: - if not TPLUS1_TEST_PREDICTIONS_PATH.exists(): - return pd.DataFrame() - df = pd.read_parquet(TPLUS1_TEST_PREDICTIONS_PATH) - for col in ("date", "target_date"): - if col in df.columns: - df[col] = pd.to_datetime(df[col], errors="coerce") - return df.sort_values("date").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(TPLUS1_SUMMARY_PATH), - _file_cache_key(TPLUS1_LATEST_PATH), - _file_cache_key(TPLUS1_TEST_PREDICTIONS_PATH), - _file_cache_key(TPLUS1_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"), - _file_cache_key(NIFTY_1M_PATH), - _file_cache_key(LIVE_ACCURACY_PATH), - ) - with _dashboard_payload_lock: - return copy.deepcopy(_dashboard_payload_cached(key)) - - + + +def _json_ready_frame(df: pd.DataFrame, limit: int | None = None) -> list[dict[str, Any]]: + out = df.copy() + if limit is not None: + out = out.tail(limit) + for col in out.columns: + if pd.api.types.is_datetime64_any_dtype(out[col]): + out[col] = out[col].dt.strftime("%Y-%m-%d %H:%M:%S") + out = out.replace({np.nan: None}) + return out.to_dict(orient="records") + + +def load_model_summary() -> dict[str, Any]: + summary_path = MODEL_DIR / "summary.json" + if not summary_path.exists(): + return {} + return json.loads(summary_path.read_text(encoding="utf-8")) + + +def load_candidate_results() -> list[dict[str, Any]]: + path = MODEL_DIR / "candidate_results.csv" + if not path.exists(): + return [] + return _json_ready_frame(pd.read_csv(path).head(12)) + + +def load_test_predictions() -> pd.DataFrame: + if not TEST_PREDICTIONS_PATH.exists(): + return pd.DataFrame() + df = pd.read_parquet(TEST_PREDICTIONS_PATH) + df["date"] = pd.to_datetime(df["date"], errors="coerce") + 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 load_tplus1_test_predictions() -> pd.DataFrame: + if not TPLUS1_TEST_PREDICTIONS_PATH.exists(): + return pd.DataFrame() + df = pd.read_parquet(TPLUS1_TEST_PREDICTIONS_PATH) + for col in ("date", "target_date"): + if col in df.columns: + df[col] = pd.to_datetime(df[col], errors="coerce") + return df.sort_values("date").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(TPLUS1_SUMMARY_PATH), + _file_cache_key(TPLUS1_LATEST_PATH), + _file_cache_key(TPLUS1_TEST_PREDICTIONS_PATH), + _file_cache_key(TPLUS1_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"), + _file_cache_key(NIFTY_1M_PATH), + _file_cache_key(LIVE_ACCURACY_PATH), + _file_cache_key(TOMORROW_PREDICTION_HISTORY_PATH), + ) + with _dashboard_payload_lock: + return copy.deepcopy(_dashboard_payload_cached(key)) + + def warm_dashboard_payload_cache() -> None: dashboard_payload() @@ -1137,6 +1237,7 @@ def build_prediction_track_record( t5_latest: dict[str, Any], tomorrow_latest: dict[str, Any], tplus1_latest: dict[str, Any], + live_accuracy: dict[str, Any] | None = None, ) -> list[dict[str, Any]]: daily_rows = daily.copy() daily_rows["date"] = pd.to_datetime(daily_rows["date"], errors="coerce").dt.normalize() @@ -1193,6 +1294,19 @@ def build_prediction_track_record( add_prediction(tomorrow_latest.get("target_date"), tomorrow_latest.get("prediction"), "Tomorrow", 40, {"prob_up": tomorrow_latest.get("prob_up")}) + ledger = live_accuracy or load_live_accuracy() + for entry in ledger.get("tomorrow", {}).get("entries", []): + day = str(entry.get("date", ""))[:10] + pred = str(entry.get("prediction", "")).upper() + if day and pred in {"UP", "DOWN"}: + add_prediction( + day, + pred, + "Tomorrow", + 50, + {"source": entry.get("source", "live")}, + ) + records: list[dict[str, Any]] = [] for _, row in daily_rows.tail(20).iterrows(): day = row["date"].date().isoformat() @@ -1202,6 +1316,14 @@ def build_prediction_track_record( actual_direction = "UP" if actual_move >= 0 else "DOWN" pred = predictions_by_date.get(day) prediction = pred.get("prediction") if pred else None + if prediction is None: + found = _find_tomorrow_prediction_for_target(row["date"].date()) + if found: + prediction = str(found.get("prediction", "")).upper() + if prediction not in {"UP", "DOWN"}: + prediction = None + else: + pred = {"prediction": prediction, "source": found.get("source", "Tomorrow"), "prob_up": found.get("prob_up")} records.append( { "date": day, @@ -1218,95 +1340,98 @@ def build_prediction_track_record( @lru_cache(maxsize=4) def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ...]) -> dict[str, Any]: - summary = load_model_summary() - t5_latest = _latest_saved_prediction_uncached() - tomorrow_summary = load_tomorrow_summary() - tomorrow_latest = latest_tomorrow_prediction() - tplus1_summary = load_tplus1_summary() - tplus1_latest = latest_tplus1_prediction() - refresh_state = load_refresh_state() + summary = load_model_summary() + t5_latest = _latest_saved_prediction_uncached() + tomorrow_summary = load_tomorrow_summary() + tomorrow_latest = latest_tomorrow_prediction() + tplus1_summary = load_tplus1_summary() + tplus1_latest = latest_tplus1_prediction() + refresh_state = load_refresh_state() t5_test = load_test_predictions() tomorrow_test = load_tomorrow_test_predictions() tomorrow_history = _load_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH) tplus1_test = load_tplus1_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 t5_test.empty: - recent_predictions = t5_test.tail(40).copy() - recent_accuracy = float(recent_predictions["correct"].mean()) - direction_mix = t5_test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index() - monthly = ( - t5_test.assign(month=t5_test["date"].dt.strftime("%Y-%m")) - .groupby("month", as_index=False)["correct"] - .mean() - .rename(columns={"correct": "accuracy"}) - ) - else: - recent_predictions = pd.DataFrame() - recent_accuracy = 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": "tplus1", - "label": "T+1", - "model_name": tplus1_summary.get("model_name", "nifty_1420_tplus1_logistic_model"), - "source_model": "14:00-14:20 logistic forecaster", - "validation_accuracy": tplus1_summary.get("validation_accuracy"), - "test_accuracy": tplus1_summary.get("test_accuracy"), - "recent_accuracy": float(tplus1_test.tail(40)["correct"].mean()) if not tplus1_test.empty and "correct" in tplus1_test.columns else tplus1_summary.get("test_accuracy"), - "test_rows": int(tplus1_summary.get("test_rows") or len(tplus1_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), - }, - ] + 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 t5_test.empty: + recent_predictions = t5_test.tail(40).copy() + recent_accuracy = float(recent_predictions["correct"].mean()) + direction_mix = t5_test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index() + monthly = ( + t5_test.assign(month=t5_test["date"].dt.strftime("%Y-%m")) + .groupby("month", as_index=False)["correct"] + .mean() + .rename(columns={"correct": "accuracy"}) + ) + else: + recent_predictions = pd.DataFrame() + recent_accuracy = 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": "tplus1", + "label": "T+1", + "model_name": tplus1_summary.get("model_name", "nifty_1420_tplus1_logistic_model"), + "source_model": "14:00-14:20 logistic forecaster", + "validation_accuracy": tplus1_summary.get("validation_accuracy"), + "test_accuracy": tplus1_summary.get("test_accuracy"), + "recent_accuracy": float(tplus1_test.tail(40)["correct"].mean()) if not tplus1_test.empty and "correct" in tplus1_test.columns else tplus1_summary.get("test_accuracy"), + "test_rows": int(tplus1_summary.get("test_rows") or len(tplus1_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": 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": tomorrow_summary.get("feature_count"), - "recent_accuracy": tomorrow_accuracy, - "recent_accuracy_days": int(len(tomorrow_recent)) if not tomorrow_recent.empty else 0, + "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": 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, } + live_accuracy = load_live_accuracy() + ensure_completed_sessions_scored(datetime.now(IST), live_accuracy) + live_accuracy = load_live_accuracy() track_record = build_prediction_track_record( daily, t5_test, @@ -1316,22 +1441,23 @@ def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ... t5_latest, tomorrow_latest, tplus1_latest, + live_accuracy=live_accuracy, ) return { - "latest": t5_latest, - "tomorrow_latest": tomorrow_latest, - "tplus1_latest": tplus1_latest, - "live_accuracy": load_live_accuracy(), - "metrics": metrics, - "summary": summary, - "tomorrow_summary": tomorrow_summary, - "tplus1_summary": tplus1_summary, - "candidates": load_candidate_results(), - "charts": { - "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]), - "opening_features": _json_ready_frame(opening), - "monthly_accuracy": _json_ready_frame(monthly), - "direction_mix": _json_ready_frame(direction_mix), + "latest": t5_latest, + "tomorrow_latest": tomorrow_latest, + "tplus1_latest": tplus1_latest, + "live_accuracy": load_live_accuracy(), + "metrics": metrics, + "summary": summary, + "tomorrow_summary": tomorrow_summary, + "tplus1_summary": tplus1_summary, + "candidates": load_candidate_results(), + "charts": { + "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]), + "opening_features": _json_ready_frame(opening), + "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), @@ -1339,102 +1465,102 @@ def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ... "tplus1_recent_predictions": _json_ready_frame(tplus1_test.tail(40)), "track_record": track_record, }, - "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(t5_test)), - "tomorrow_test_prediction_rows": int(len(tomorrow_test)), - "tplus1_test_prediction_rows": int(len(tplus1_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, 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="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) - dataset = read_training_dataset() - 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") - 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(), - "path": str(NIFTY_1D_PATH), - } - - -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(), - } - - + "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(t5_test)), + "tomorrow_test_prediction_rows": int(len(tomorrow_test)), + "tplus1_test_prediction_rows": int(len(tplus1_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, 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="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) + dataset = read_training_dataset() + 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") + 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(), + "path": str(NIFTY_1D_PATH), + } + + +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 load_live_accuracy() -> dict[str, Any]: """Load the live accuracy ledger from disk.""" default = { @@ -1496,6 +1622,35 @@ def _record_prediction_history(path: Path, row: dict[str, Any], subset: list[str append_prediction_history(path, row, subset) +def ensure_completed_sessions_scored( + now: datetime | None = None, + ledger: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Score any completed Tomorrow sessions that have daily close data but no ledger entry.""" + now = now or datetime.now(IST) + completed = expected_completed_daily_date(now) + if not is_trading_day(completed): + return ledger or load_live_accuracy() + + ledger = ledger or load_live_accuracy() + logged_tom = {e["date"] for e in ledger.get("tomorrow", {}).get("entries", [])} + if completed.isoformat() in logged_tom: + return ledger + + daily = pd.read_parquet(NIFTY_1D_PATH) + daily["_date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize() + day_rows = daily[daily["_date"].dt.date == completed] + if day_rows.empty: + return ledger + + day_open = float(day_rows.iloc[-1]["open"]) + day_close = float(day_rows.iloc[-1]["close"]) + if not (np.isfinite(day_open) and np.isfinite(day_close) and day_open != 0): + return ledger + + return update_live_accuracy(completed) + + def update_live_accuracy(session_date: date) -> dict[str, Any]: """Score today's predictions against actual outcomes and update the ledger. @@ -1548,31 +1703,20 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]: except Exception: pass - # --- Tomorrow: yesterday's prediction targeting today vs close > open --- + # --- Tomorrow: yesterday's prediction targeting today vs close vs open --- logged_tom = {e["date"] for e in ledger["tomorrow"]["entries"]} if session_iso not in logged_tom: - tom_history = _load_prediction_history(TOMORROW_PREDICTION_HISTORY_PATH) - if not tom_history.empty and "target_date" in tom_history.columns: - tom_rows = tom_history[tom_history["target_date"].dt.date == session_date] - else: - tom_rows = pd.DataFrame() - if tom_rows.empty and TOMORROW_LATEST_PATH.exists(): - try: - tom_row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict() - if str(tom_row.get("target_date", ""))[:10] == session_iso: - tom_rows = pd.DataFrame([tom_row]) - except Exception: - tom_rows = pd.DataFrame() - if not tom_rows.empty: + actual_tomorrow = "UP" if day_close >= day_open else "DOWN" + tom_row = _find_tomorrow_prediction_for_target(session_date) + if tom_row: try: - tom_row = tom_rows.iloc[-1].to_dict() pred = str(tom_row.get("prediction", "")).upper() if pred in ("UP", "DOWN"): ledger["tomorrow"]["entries"].append({ "date": session_iso, "prediction": pred, - "actual": actual_close_gt_open, - "correct": pred == actual_close_gt_open, + "actual": actual_tomorrow, + "correct": pred == actual_tomorrow, "source": "live", }) except Exception: @@ -1634,60 +1778,60 @@ def update_live_accuracy(session_date: date) -> dict[str, Any]: return ledger -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() - # Score live predictions BEFORE they get overwritten by fresh ones - try: - update_live_accuracy(session_date) - except Exception as exc: - print(f"[close-refresh] live accuracy update failed: {exc}", flush=True) - t5_prediction = refresh_first5_prediction(session_date=session_date, minutes=minutes) - tplus1_prediction = refresh_tplus1_prediction(session_date=session_date) - 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(), - "tplus1_prediction": tplus1_prediction, - "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 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() + # Score live predictions BEFORE they get overwritten by fresh ones + try: + update_live_accuracy(session_date) + except Exception as exc: + print(f"[close-refresh] live accuracy update failed: {exc}", flush=True) + t5_prediction = refresh_first5_prediction(session_date=session_date, minutes=minutes) + tplus1_prediction = refresh_tplus1_prediction(session_date=session_date) + 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(), + "tplus1_prediction": tplus1_prediction, + "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 + 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) @@ -1795,6 +1939,18 @@ def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]: outcomes = update_opening_outcomes_from_daily() actions.append({"name": "daily", **daily_info}) actions.append({"name": "opening_outcomes", **outcomes}) + try: + completed = date.fromisoformat(status["expected_daily_date"]) + scored = update_live_accuracy(completed) + actions.append( + { + "name": "tomorrow_live_accuracy", + "session_date": completed.isoformat(), + "live_count": scored.get("tomorrow", {}).get("live_count"), + } + ) + except Exception as exc: + actions.append({"name": "tomorrow_live_accuracy", "error": str(exc)}) if status["daily_stale"] or status["tomorrow_stale"]: try: @@ -1819,15 +1975,15 @@ def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]: 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): - target_day += timedelta(days=1) - target_day = next_trading_day(target_day) - return datetime.combine(target_day, run_time, tzinfo=IST) - - -def seconds_until_next_ist_run(run_time: time = time(9, 20)) -> float: - now = datetime.now(IST) - target = next_ist_run_at(run_time, now=now) - return max(1.0, (target - now).total_seconds()) + now = now or datetime.now(IST) + target_day = now.date() + if now >= datetime.combine(target_day, run_time, tzinfo=IST): + target_day += timedelta(days=1) + target_day = next_trading_day(target_day) + return datetime.combine(target_day, run_time, tzinfo=IST) + + +def seconds_until_next_ist_run(run_time: time = time(9, 20)) -> float: + now = datetime.now(IST) + target = next_ist_run_at(run_time, now=now) + return max(1.0, (target - now).total_seconds())