Jitendra12421 commited on
Commit
8e561ff
·
verified ·
1 Parent(s): 1b3e5a6

Upload runtime.py

Browse files
Files changed (1) hide show
  1. runtime.py +1719 -367
runtime.py CHANGED
@@ -1,425 +1,1777 @@
1
- from __future__ import annotations
2
-
3
  import json
 
 
4
  import sys
5
- from dataclasses import dataclass
6
- from datetime import date, datetime, time, timedelta
7
- from pathlib import Path
8
- from typing import Any
9
- from zoneinfo import ZoneInfo
10
-
 
 
11
  import joblib
12
  import numpy as np
13
  import pandas as pd
14
- import yfinance as yf
15
-
16
-
17
- IST = ZoneInfo("Asia/Kolkata")
 
 
 
 
 
18
  YAHOO_NIFTY_SYMBOL = "^NSEI"
 
 
 
 
 
19
  BACKEND_ROOT = Path(__file__).resolve().parents[1]
20
  DATA_DIR = BACKEND_ROOT / "data"
21
  MODEL_DIR = BACKEND_ROOT / "models"
22
- OPENING_DATASET_PATH = DATA_DIR / "opening_direction_training_dataset.parquet"
23
- NIFTY_1M_PATH = DATA_DIR / "nifty50_1m.parquet"
24
- NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet"
25
- MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib"
26
- LATEST_PATH = MODEL_DIR / "latest_prediction.csv"
27
- TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet"
28
-
29
- DECISION_OVERLAYS = [
30
- {
31
- "name": "fifth_minute_momentum_flip",
32
- "feature": "m5_ret_1m",
33
- "op": ">=",
34
- "value": 0.0005085411885759201,
35
- },
36
- {
37
- "name": "vix_stretch_flip",
38
- "feature": "india_vix_close_vs_sma_20",
39
- "op": ">=",
40
- "value": 0.24641908937959742,
41
- },
42
- ]
43
-
44
-
45
- class ProbabilityBlend:
46
- def __init__(self, models: list[Any], weights: np.ndarray):
47
- self.models = models
48
- self.weights = np.asarray(weights, dtype="float64")
49
- self.weights = self.weights / self.weights.sum()
50
-
51
- def predict_proba(self, x: pd.DataFrame) -> np.ndarray:
52
- probs = np.column_stack([predict_proba_up(model, x) for model in self.models])
53
- prob_up = probs @ self.weights
54
- return np.column_stack([1.0 - prob_up, prob_up])
55
-
56
-
57
- @dataclass(frozen=True)
58
- class Prediction:
59
- input_date: str
60
- first5_start: str
61
- first5_end: str
62
- prediction: str
63
- prob_up: float
64
- confidence: float
65
- threshold: float
66
- model_name: str
67
-
68
- def to_dict(self) -> dict[str, Any]:
69
- return {
70
- "input_date": self.input_date,
71
- "first5_start": self.first5_start,
72
- "first5_end": self.first5_end,
73
- "prediction": self.prediction,
74
- "prob_up": self.prob_up,
75
- "confidence": self.confidence,
76
- "threshold": self.threshold,
77
- "model_name": self.model_name,
78
- }
79
-
80
-
81
- def predict_proba_up(model: Any, x: pd.DataFrame) -> np.ndarray:
82
- return np.asarray(model.predict_proba(x)[:, 1], dtype="float64")
83
-
84
-
85
- def safe_div(numer: pd.Series | np.ndarray, denom: pd.Series | np.ndarray) -> pd.Series:
86
- n = pd.Series(numer, copy=False)
87
- d = pd.Series(denom, copy=False)
88
- out = pd.Series(np.nan, index=n.index, dtype="float64")
89
- mask = d.notna() & np.isfinite(d.to_numpy(dtype="float64")) & (d != 0)
90
- out.loc[mask] = n.loc[mask].to_numpy(dtype="float64") / d.loc[mask].to_numpy(dtype="float64")
91
- return out
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
92
 
93
 
94
- def load_model() -> dict[str, Any]:
95
- # Existing artifact was trained as a script, so its custom blend class
96
- # resolves through __main__ when unpickled.
97
- sys.modules["__main__"].ProbabilityBlend = ProbabilityBlend
98
- sys.modules["__main__"].predict_proba_up = predict_proba_up
99
- payload = joblib.load(MODEL_PATH)
100
- payload.setdefault("decision_overlays", DECISION_OVERLAYS)
101
- payload.setdefault("model_name", "nifty_opening_direction_model")
102
- return payload
103
 
104
 
105
- def overlay_mask(frame: pd.DataFrame, overlay: dict[str, object]) -> np.ndarray:
106
- feature = str(overlay["feature"])
107
- if feature not in frame.columns:
108
- return np.zeros(len(frame), dtype=bool)
109
- series = pd.to_numeric(frame[feature], errors="coerce")
110
- value = float(overlay["value"])
111
- if overlay["op"] == ">=":
112
- return (series >= value).fillna(False).to_numpy(dtype=bool)
113
- if overlay["op"] == "<=":
114
- return (series <= value).fillna(False).to_numpy(dtype=bool)
115
- raise ValueError(f"Unsupported overlay op: {overlay['op']}")
 
 
 
 
 
 
 
 
 
 
 
 
116
 
117
 
118
- def apply_decision_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, object]]) -> np.ndarray:
119
- adjusted = np.asarray(pred, dtype="int64").copy()
120
- for overlay in overlays:
121
- mask = overlay_mask(frame, overlay)
122
- adjusted[mask] = 1 - adjusted[mask]
123
- return adjusted
 
 
 
 
 
 
 
 
 
 
124
 
125
 
126
- def directional_confidence(prob_up: np.ndarray, pred: np.ndarray, threshold: float) -> np.ndarray:
127
- prob_up = np.asarray(prob_up, dtype="float64")
128
- pred = np.asarray(pred, dtype="int64")
129
- base_side_prob = np.where(pred == 1, prob_up, 1.0 - prob_up)
130
- threshold_distance = np.abs(prob_up - float(threshold))
131
- return np.clip(0.50 + threshold_distance, base_side_prob, 0.99)
 
 
 
 
 
132
 
133
 
134
- def read_training_dataset() -> pd.DataFrame:
135
- df = pd.read_parquet(OPENING_DATASET_PATH)
136
- for col in ("date", "first5_start", "first5_end"):
137
- if col in df.columns:
138
- df[col] = pd.to_datetime(df[col], errors="coerce")
139
- return df.sort_values("date").reset_index(drop=True)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
140
 
141
 
142
- def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame:
143
- if df.empty:
144
- return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"])
145
- if isinstance(df.columns, pd.MultiIndex):
146
- df.columns = [str(c[0]).lower() for c in df.columns]
147
- else:
148
- df.columns = [str(c).lower().replace(" ", "_") for c in df.columns]
149
- df = df.reset_index()
150
- date_col = next((c for c in df.columns if c.lower() in {"datetime", "date"}), df.columns[0])
151
- df["date"] = pd.to_datetime(df[date_col], errors="coerce")
152
- if df["date"].dt.tz is None:
153
- df["date"] = df["date"].dt.tz_localize("UTC").dt.tz_convert(IST)
154
  else:
155
- df["date"] = df["date"].dt.tz_convert(IST)
156
- rename = {
157
- "open": "open",
158
- "high": "high",
159
- "low": "low",
160
- "close": "close",
161
- "adj_close": "close",
162
- "volume": "volume",
163
- }
164
- out = pd.DataFrame({"date": df["date"].dt.tz_localize(None)})
165
- for src, dst in rename.items():
166
- if src in df.columns and dst not in out.columns:
167
- out[dst] = pd.to_numeric(df[src], errors="coerce")
168
- return out.dropna(subset=["date", "open", "high", "low", "close"]).sort_values("date")
169
 
170
 
171
- def fetch_yahoo_minutes(period: str = "5d") -> pd.DataFrame:
172
- raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1m", progress=False, prepost=False, auto_adjust=False)
173
- return normalize_yahoo_frame(raw)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
174
 
175
 
176
- def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
177
- raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1d", progress=False, prepost=False, auto_adjust=False)
178
- out = normalize_yahoo_frame(raw)
179
- out["date"] = pd.to_datetime(out["date"], errors="coerce").dt.normalize()
180
- return out.drop_duplicates("date", keep="last")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
181
 
182
 
183
- def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
184
- if path.exists():
185
- existing = pd.read_parquet(path)
186
- combined = pd.concat([existing, new_rows], ignore_index=True)
187
- else:
188
- combined = new_rows.copy()
189
- combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True)
190
- combined.to_parquet(path, index=False, compression="zstd")
191
- return combined
192
-
193
-
194
- def first5_features_from_minutes(minutes: pd.DataFrame, session_date: date | None = None) -> pd.DataFrame:
195
- if minutes.empty:
196
- raise RuntimeError("Yahoo returned no minute bars.")
197
- bars = minutes.copy()
198
- bars["dt"] = pd.to_datetime(bars["date"], errors="coerce")
199
- bars["session_date"] = bars["dt"].dt.normalize()
200
- if session_date is None:
201
- session_ts = bars["session_date"].max()
202
- else:
203
- session_ts = pd.Timestamp(session_date).normalize()
204
- day = bars[bars["session_date"] == session_ts].sort_values("dt").copy()
205
- start_dt = pd.Timestamp.combine(session_ts.date(), time(9, 15))
206
- end_dt = pd.Timestamp.combine(session_ts.date(), time(9, 19))
207
- first5 = day[(day["dt"] >= start_dt) & (day["dt"] <= end_dt)].head(5).copy()
208
- if len(first5) < 5:
209
- raise RuntimeError(f"Need 5 opening bars for {session_ts.date()}, got {len(first5)}.")
210
- first5["minute_index"] = np.arange(len(first5))
211
- first5["ret_1m"] = first5["close"].pct_change(fill_method=None)
212
- first5["range_pct_1m"] = safe_div(first5["high"] - first5["low"], first5["open"])
213
- first5["body_pct_1m"] = safe_div(first5["close"] - first5["open"], first5["open"])
214
- row = {
215
- "date": session_ts,
216
- "first5_start": first5["dt"].iloc[0],
217
- "first5_end": first5["dt"].iloc[-1],
218
- "first5_open": first5["open"].iloc[0],
219
- "first5_high": first5["high"].max(),
220
- "first5_low": first5["low"].min(),
221
- "first5_close": first5["close"].iloc[-1],
222
- "first5_volume": first5["volume"].sum() if "volume" in first5 else 0.0,
223
- "first5_bars": len(first5),
224
- "first5_last_1m_ret": first5["ret_1m"].iloc[-1],
225
- "first5_ret_std": first5["ret_1m"].std(),
226
  }
227
- row["first5_return"] = (row["first5_close"] - row["first5_open"]) / row["first5_open"]
228
- row["first5_range_pct"] = (row["first5_high"] - row["first5_low"]) / row["first5_open"]
229
- first5_range = row["first5_high"] - row["first5_low"]
230
- row["first5_body_to_range"] = (row["first5_close"] - row["first5_open"]) / first5_range if first5_range else np.nan
231
- row["first5_close_location"] = (row["first5_close"] - row["first5_low"]) / first5_range if first5_range else np.nan
232
- for idx, (_, candle) in enumerate(first5.iterrows(), start=1):
233
- for field in ("open", "high", "low", "close", "ret_1m", "range_pct_1m", "body_pct_1m"):
234
- row[f"m{idx}_{field}"] = candle[field]
235
- row[f"m{idx}_close_vs_first5_open"] = (candle["close"] - row["first5_open"]) / row["first5_open"]
236
- row[f"m{idx}_range_share"] = (candle["high"] - candle["low"]) / first5_range if first5_range else np.nan
237
- row["first5_return_accel"] = row["m5_ret_1m"] - row["m2_ret_1m"]
238
- row["first5_last2_return"] = (row["m5_close"] - row["m4_open"]) / row["m4_open"]
239
- row["first5_first2_return"] = (row["m2_close"] - row["m1_open"]) / row["m1_open"]
240
- row["first5_reversal"] = np.sign(row["first5_first2_return"]) * -np.sign(row["first5_last2_return"])
241
- row["dow"] = session_ts.dayofweek
242
- row["dom"] = session_ts.day
243
- row["month"] = session_ts.month
244
- return pd.DataFrame([row])
245
-
246
-
247
- def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame:
248
- dataset = read_training_dataset()
249
- latest_context = dataset.iloc[[-1]].copy()
250
- output = latest_context.copy()
251
- for col in first5_row.columns:
252
- output[col] = first5_row[col].iloc[0]
253
- if {"first5_open", "nifty_close"}.issubset(output.columns):
254
- output["first5_gap_from_prev_close"] = (output["first5_open"] - output["nifty_close"]) / output["nifty_close"]
255
- output["first5_close_vs_prev_close"] = (output["first5_close"] - output["nifty_close"]) / output["nifty_close"]
256
- if {"first5_range_pct", "nifty_range_pct"}.issubset(output.columns):
257
- output["first5_range_vs_prev_range"] = output["first5_range_pct"] / output["nifty_range_pct"]
258
- if {"first5_return", "nifty_ret_1"}.issubset(output.columns):
259
- output["first5_return_x_prev_ret"] = output["first5_return"] * output["nifty_ret_1"]
260
- output["gap_x_prev_ret"] = output["first5_gap_from_prev_close"] * output["nifty_ret_1"]
261
- if {"first5_return", "banknifty_ret_1"}.issubset(output.columns):
262
- output["first5_return_x_bank_ret_1"] = output["first5_return"] * output["banknifty_ret_1"]
263
- if {"first5_range_pct", "india_vix_ret_1"}.issubset(output.columns):
264
- output["first5_range_x_vix_ret_1"] = output["first5_range_pct"] * output["india_vix_ret_1"]
265
- output["target"] = np.nan
266
- output["day_return"] = np.nan
267
- return output
268
-
269
-
270
- def predict_row(row: pd.DataFrame) -> Prediction:
271
- payload = load_model()
272
- model = payload["model"]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
273
  features = payload["features"]
274
  threshold = float(payload["threshold"])
275
- missing = [c for c in features if c not in row.columns]
276
- if missing:
277
- raise RuntimeError(f"Feature row is missing {len(missing)} features; first missing: {missing[:5]}")
278
- prob_up = predict_proba_up(model, row[features])
279
- raw_pred = (prob_up >= threshold).astype("int64")
280
- pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS))
281
- confidence = directional_confidence(prob_up, pred, threshold)
282
- prediction = Prediction(
283
- input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(),
284
- first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])),
285
- first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])),
286
- prediction="UP" if int(pred[0]) == 1 else "DOWN",
287
- prob_up=float(prob_up[0]),
288
- confidence=float(confidence[0]),
289
- threshold=threshold,
290
- model_name=str(payload.get("model_name", "nifty_opening_direction_model")),
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
291
  )
292
- pd.DataFrame([prediction.to_dict()]).to_csv(LATEST_PATH, index=False)
293
- return prediction
294
-
295
-
296
- def latest_saved_prediction() -> dict[str, Any]:
297
- if LATEST_PATH.exists():
298
- return pd.read_csv(LATEST_PATH).iloc[-1].to_dict()
299
- summary_path = MODEL_DIR / "summary.json"
300
- if summary_path.exists():
301
- return json.loads(summary_path.read_text(encoding="utf-8"))
302
- raise FileNotFoundError("No latest prediction is available yet.")
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
303
 
304
 
305
- def _json_ready_frame(df: pd.DataFrame, limit: int | None = None) -> list[dict[str, Any]]:
306
- out = df.copy()
307
- if limit is not None:
308
- out = out.tail(limit)
309
- for col in out.columns:
310
- if pd.api.types.is_datetime64_any_dtype(out[col]):
311
- out[col] = out[col].dt.strftime("%Y-%m-%d %H:%M:%S")
312
- out = out.replace({np.nan: None})
313
- return out.to_dict(orient="records")
 
 
 
 
 
 
 
 
 
 
 
 
314
 
 
315
 
316
- def load_model_summary() -> dict[str, Any]:
317
- summary_path = MODEL_DIR / "summary.json"
318
- if not summary_path.exists():
319
- return {}
320
- return json.loads(summary_path.read_text(encoding="utf-8"))
 
 
 
 
 
 
 
 
 
321
 
 
 
 
 
 
 
322
 
323
- def load_candidate_results() -> list[dict[str, Any]]:
324
- path = MODEL_DIR / "candidate_results.csv"
325
- if not path.exists():
326
- return []
327
- return _json_ready_frame(pd.read_csv(path).head(12))
328
 
 
 
 
 
 
329
 
330
- def load_test_predictions() -> pd.DataFrame:
331
- if not TEST_PREDICTIONS_PATH.exists():
332
- return pd.DataFrame()
333
- df = pd.read_parquet(TEST_PREDICTIONS_PATH)
334
- df["date"] = pd.to_datetime(df["date"], errors="coerce")
335
- return df.sort_values("date").reset_index(drop=True)
 
 
 
 
 
336
 
 
 
 
 
337
 
338
- def dashboard_payload() -> dict[str, Any]:
339
- summary = load_model_summary()
340
- latest = latest_saved_prediction()
341
- test = load_test_predictions()
342
- daily = pd.read_parquet(NIFTY_1D_PATH)
343
- daily["date"] = pd.to_datetime(daily["date"], errors="coerce")
344
- daily = daily.sort_values("date").tail(180)
345
- dataset = read_training_dataset()
346
- opening = dataset[["date", "first5_return", "first5_range_pct", "first5_close_location"]].tail(120).copy()
347
-
348
- if not test.empty:
349
- recent_predictions = test.tail(40).copy()
350
- recent_accuracy = float(recent_predictions["correct"].mean())
351
- direction_mix = test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index()
352
- monthly = (
353
- test.assign(month=test["date"].dt.strftime("%Y-%m"))
354
- .groupby("month", as_index=False)["correct"]
355
- .mean()
356
- .rename(columns={"correct": "accuracy"})
357
  )
358
- else:
359
- recent_predictions = pd.DataFrame()
360
- recent_accuracy = None
361
- direction_mix = pd.DataFrame()
362
- monthly = pd.DataFrame()
363
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
364
  metrics = {
365
- "validation_accuracy": summary.get("validation_accuracy"),
366
- "test_accuracy": summary.get("test_accuracy"),
367
- "baseline_test_accuracy": summary.get("baseline_test_accuracy"),
368
- "validation_auc": summary.get("validation_auc"),
369
- "test_auc": summary.get("test_auc"),
370
- "test_brier": summary.get("test_brier"),
371
- "feature_count": summary.get("feature_count"),
372
- "recent_accuracy": recent_accuracy,
373
- "recent_accuracy_days": int(len(recent_predictions)) if not recent_predictions.empty else 0,
374
- "total_test_days": int(len(test)) if not test.empty else 0,
 
375
  }
 
 
 
 
 
 
 
 
 
376
  return {
377
- "latest": latest,
378
- "metrics": metrics,
379
- "summary": summary,
380
- "candidates": load_candidate_results(),
381
- "charts": {
382
- "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]),
383
- "opening_features": _json_ready_frame(opening),
384
- "monthly_accuracy": _json_ready_frame(monthly),
385
- "direction_mix": _json_ready_frame(direction_mix),
 
 
 
 
 
386
  "recent_predictions": _json_ready_frame(recent_predictions),
 
 
 
 
387
  },
388
- "data_status": {
389
- "nifty_1m_rows": int(len(pd.read_parquet(NIFTY_1M_PATH, columns=["date"]))),
390
- "nifty_1d_rows": int(len(pd.read_parquet(NIFTY_1D_PATH, columns=["date"]))),
391
- "training_rows": int(len(dataset)),
392
- "test_prediction_rows": int(len(test)),
393
- "latest_daily_date": pd.to_datetime(daily["date"]).max().date().isoformat(),
394
- },
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
395
  }
396
 
 
 
 
 
 
 
397
 
398
- def refresh_first5_prediction(session_date: date | None = None) -> Prediction:
399
- minutes = fetch_yahoo_minutes(period="5d")
400
- append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
401
- first5 = first5_features_from_minutes(minutes, session_date=session_date)
402
- row = build_model_row(first5)
403
- dataset = read_training_dataset()
404
- merged = pd.concat([dataset, row], ignore_index=True)
405
- merged = merged.drop_duplicates(subset=["date"], keep="last").sort_values("date").reset_index(drop=True)
406
- merged.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd")
407
- return predict_row(row)
 
 
 
 
 
 
 
408
 
 
 
 
409
 
410
- def refresh_daily_data() -> dict[str, Any]:
411
- daily = fetch_yahoo_daily(period="1mo")
412
- combined = append_parquet_rows(NIFTY_1D_PATH, daily, ["date"])
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
413
  return {
414
- "rows": int(len(combined)),
415
- "latest_date": pd.to_datetime(combined["date"]).max().date().isoformat(),
416
- "path": str(NIFTY_1D_PATH),
 
 
 
 
 
 
 
 
 
 
 
417
  }
418
 
419
 
420
- def seconds_until_next_ist_run(run_time: time = time(9, 20)) -> float:
421
- now = datetime.now(IST)
422
- target = datetime.combine(now.date(), run_time, tzinfo=IST)
423
- if now >= target:
424
- target += timedelta(days=1)
425
- return max(1.0, (target - now).total_seconds())
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
+ from __future__ import annotations
2
+
3
  import json
4
+ import copy
5
+ import os
6
  import sys
7
+ import threading
8
+ from dataclasses import dataclass
9
+ from datetime import date, datetime, time, timedelta
10
+ from functools import lru_cache
11
+ from pathlib import Path
12
+ from typing import Any
13
+ from zoneinfo import ZoneInfo
14
+
15
  import joblib
16
  import numpy as np
17
  import pandas as pd
18
+ from nifty_backend.yahoo_history_client import YahooHistoryClient
19
+
20
+ try:
21
+ import pandas_market_calendars as mcal
22
+ except ImportError: # pragma: no cover - production dependency, local fallback below.
23
+ mcal = None
24
+
25
+
26
+ IST = ZoneInfo("Asia/Kolkata")
27
  YAHOO_NIFTY_SYMBOL = "^NSEI"
28
+ MARKET_CLOSE = time(15, 30)
29
+ FIRST5_READY = time(9, 20)
30
+ CLOSE_REFRESH_READY = time(15, 45)
31
+ TPLUS1_READY = time(14, 30)
32
+ STALE_CHECK_INTERVAL_SECONDS = 5
33
  BACKEND_ROOT = Path(__file__).resolve().parents[1]
34
  DATA_DIR = BACKEND_ROOT / "data"
35
  MODEL_DIR = BACKEND_ROOT / "models"
36
+ YAHOO_CACHE_PATH = MODEL_DIR / "yahoo_history_cache.sqlite3"
37
+ OPENING_DATASET_PATH = DATA_DIR / "opening_direction_training_dataset.parquet"
38
+ NIFTY_1M_PATH = DATA_DIR / "nifty50_1m.parquet"
39
+ NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet"
40
+ MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib"
41
+ LATEST_PATH = MODEL_DIR / "latest_prediction.csv"
42
+ TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet"
43
+ TOMORROW_MODEL_PATH = MODEL_DIR / "nifty_tomorrow_direction_model.joblib"
44
+ TOMORROW_LATEST_PATH = MODEL_DIR / "tomorrow_latest_prediction.csv"
45
+ TOMORROW_SUMMARY_PATH = MODEL_DIR / "tomorrow_summary.json"
46
+ TOMORROW_TEST_PREDICTIONS_PATH = DATA_DIR / "tomorrow_test_predictions.parquet"
47
+ FORECASTING_PROJECT_ROOT = Path(
48
+ os.environ.get(
49
+ "FORECASTING_PROJECT_ROOT",
50
+ str(BACKEND_ROOT.parent.parent / "forecasting project"),
51
+ )
52
+ )
53
+ DAILY_FORECASTER_OUTPUT_DIR = FORECASTING_PROJECT_ROOT / "Code" / "models" / "nifty_forecaster" / "outputs"
54
+ DAILY_FORECASTER_SUMMARY_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_summary.json"
55
+ DAILY_FORECASTER_LATEST_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_latest.csv"
56
+ DAILY_FORECASTER_PREDICTIONS_PATH = DAILY_FORECASTER_OUTPUT_DIR / "forecaster_test_predictions.csv"
57
+ TPLUS1_MODEL_PATH = MODEL_DIR / "nifty_1420_tplus1_logistic_model.joblib"
58
+ TPLUS1_LATEST_PATH = MODEL_DIR / "tplus1_latest_prediction.csv"
59
+ TPLUS1_SUMMARY_PATH = MODEL_DIR / "tplus1_summary.json"
60
+ TPLUS1_TEST_PREDICTIONS_PATH = DATA_DIR / "tplus1_test_predictions.parquet"
61
+ REFRESH_STATE_PATH = MODEL_DIR / "refresh_state.json"
62
+ REFRESH_WAITING = "waiting_second_payload"
63
+ REFRESH_REFRESHING = "refreshing"
64
+ REFRESH_READY = "ready"
65
+ REFRESH_FAILED = "failed"
66
+ REFRESH_NORMAL = "normal"
67
+ LIVE_ACCURACY_PATH = MODEL_DIR / "live_accuracy.json"
68
+
69
+ DECISION_OVERLAYS = [
70
+ {
71
+ "name": "fifth_minute_momentum_flip",
72
+ "feature": "m5_ret_1m",
73
+ "op": ">=",
74
+ "value": 0.0005085411885759201,
75
+ },
76
+ {
77
+ "name": "vix_stretch_flip",
78
+ "feature": "india_vix_close_vs_sma_20",
79
+ "op": ">=",
80
+ "value": 0.24641908937959742,
81
+ },
82
+ ]
83
+
84
+ _dashboard_payload_lock = threading.Lock()
85
+ _stale_refresh_lock = threading.Lock()
86
+
87
+
88
+ def utc_now_iso() -> str:
89
+ return datetime.utcnow().replace(microsecond=0).isoformat() + "Z"
90
+
91
+
92
+ def clear_dashboard_payload_cache() -> None:
93
+ _dashboard_payload_cached.cache_clear()
94
+
95
+
96
+ def save_refresh_state(phase: str, *, session_date: date | None = None, error: str | None = None) -> dict[str, Any]:
97
+ previous = load_refresh_state()
98
+ state = {
99
+ "phase": phase,
100
+ "started_at": previous.get("started_at"),
101
+ "finished_at": previous.get("finished_at"),
102
+ "session_date": session_date.isoformat() if session_date else previous.get("session_date"),
103
+ "error": error,
104
+ }
105
+ if phase in {REFRESH_WAITING, REFRESH_REFRESHING} and not state["started_at"]:
106
+ state["started_at"] = utc_now_iso()
107
+ if phase in {REFRESH_READY, REFRESH_FAILED, REFRESH_NORMAL}:
108
+ state["finished_at"] = utc_now_iso()
109
+ REFRESH_STATE_PATH.write_text(json.dumps(state, indent=2), encoding="utf-8")
110
+ return state
111
+
112
+
113
+ def load_refresh_state() -> dict[str, Any]:
114
+ if not REFRESH_STATE_PATH.exists():
115
+ return {
116
+ "phase": REFRESH_NORMAL,
117
+ "started_at": None,
118
+ "finished_at": None,
119
+ "session_date": None,
120
+ "error": None,
121
+ }
122
+ try:
123
+ return json.loads(REFRESH_STATE_PATH.read_text(encoding="utf-8"))
124
+ except Exception:
125
+ return {
126
+ "phase": REFRESH_FAILED,
127
+ "started_at": None,
128
+ "finished_at": None,
129
+ "session_date": None,
130
+ "error": "refresh_state.json could not be read",
131
+ }
132
+
133
+
134
+ @lru_cache(maxsize=1)
135
+ def _nse_calendar():
136
+ if mcal is None:
137
+ return None
138
+ for name in ("XNSE", "NSE", "BSE"):
139
+ try:
140
+ return mcal.get_calendar(name)
141
+ except Exception:
142
+ continue
143
+ return None
144
+
145
+
146
+ @lru_cache(maxsize=64)
147
+ def trading_schedule(start: date, end: date) -> pd.DataFrame:
148
+ calendar = _nse_calendar()
149
+ if calendar is None:
150
+ days = pd.date_range(start=start, end=end, freq="B")
151
+ return pd.DataFrame(index=days)
152
+ return calendar.schedule(start_date=start, end_date=end)
153
+
154
+
155
+ def is_trading_day(day: date) -> bool:
156
+ schedule = trading_schedule(day, day)
157
+ return not schedule.empty
158
+
159
+
160
+ def next_trading_day(start: date) -> date:
161
+ end = start + timedelta(days=14)
162
+ schedule = trading_schedule(start, end)
163
+ if schedule.empty:
164
+ day = start
165
+ while not is_trading_day(day):
166
+ day += timedelta(days=1)
167
+ return day
168
+ return pd.Timestamp(schedule.index[0]).date()
169
+
170
+
171
+ def previous_trading_day(start: date) -> date:
172
+ begin = start - timedelta(days=14)
173
+ schedule = trading_schedule(begin, start)
174
+ if schedule.empty:
175
+ day = start
176
+ while not is_trading_day(day):
177
+ day -= timedelta(days=1)
178
+ return day
179
+ return pd.Timestamp(schedule.index[-1]).date()
180
+
181
+
182
+ class ProbabilityBlend:
183
+ def __init__(self, models: list[Any], weights: np.ndarray):
184
+ self.models = models
185
+ self.weights = np.asarray(weights, dtype="float64")
186
+ self.weights = self.weights / self.weights.sum()
187
+
188
+ def predict_proba(self, x: pd.DataFrame) -> np.ndarray:
189
+ probs = np.column_stack([predict_proba_up(model, x) for model in self.models])
190
+ prob_up = probs @ self.weights
191
+ return np.column_stack([1.0 - prob_up, prob_up])
192
+
193
+
194
+ @dataclass(frozen=True)
195
+ class Prediction:
196
+ input_date: str
197
+ first5_start: str
198
+ first5_end: str
199
+ prediction: str
200
+ prob_up: float
201
+ confidence: float
202
+ threshold: float
203
+ model_name: str
204
+ is_overridden: bool = False
205
+
206
+ def to_dict(self) -> dict[str, Any]:
207
+ return {
208
+ "input_date": self.input_date,
209
+ "first5_start": self.first5_start,
210
+ "first5_end": self.first5_end,
211
+ "prediction": self.prediction,
212
+ "prob_up": self.prob_up,
213
+ "confidence": self.confidence,
214
+ "threshold": self.threshold,
215
+ "model_name": self.model_name,
216
+ "is_overridden": getattr(self, "is_overridden", False),
217
+ }
218
+
219
+
220
+ def predict_proba_up(model: Any, x: pd.DataFrame) -> np.ndarray:
221
+ return np.asarray(model.predict_proba(x)[:, 1], dtype="float64")
222
+
223
+
224
+ def safe_div(numer: pd.Series | np.ndarray, denom: pd.Series | np.ndarray) -> pd.Series:
225
+ n = pd.Series(numer, copy=False)
226
+ d = pd.Series(denom, copy=False)
227
+ out = pd.Series(np.nan, index=n.index, dtype="float64")
228
+ mask = d.notna() & np.isfinite(d.to_numpy(dtype="float64")) & (d != 0)
229
+ out.loc[mask] = n.loc[mask].to_numpy(dtype="float64") / d.loc[mask].to_numpy(dtype="float64")
230
+ return out
231
+
232
+
233
+ def load_model() -> dict[str, Any]:
234
+ # Existing artifact was trained as a script, so its custom blend class
235
+ # resolves through __main__ when unpickled.
236
+ sys.modules["__main__"].ProbabilityBlend = ProbabilityBlend
237
+ sys.modules["__main__"].predict_proba_up = predict_proba_up
238
+ payload = joblib.load(MODEL_PATH)
239
+ payload.setdefault("decision_overlays", DECISION_OVERLAYS)
240
+ payload.setdefault("model_name", "nifty_opening_direction_model")
241
+ return payload
242
+
243
+
244
+ def overlay_mask(frame: pd.DataFrame, overlay: dict[str, object]) -> np.ndarray:
245
+ feature = str(overlay["feature"])
246
+ if feature not in frame.columns:
247
+ return np.zeros(len(frame), dtype=bool)
248
+ series = pd.to_numeric(frame[feature], errors="coerce")
249
+ value = float(overlay["value"])
250
+ if overlay["op"] == ">=":
251
+ return (series >= value).fillna(False).to_numpy(dtype=bool)
252
+ if overlay["op"] == "<=":
253
+ return (series <= value).fillna(False).to_numpy(dtype=bool)
254
+ raise ValueError(f"Unsupported overlay op: {overlay['op']}")
255
+
256
+
257
+ def apply_decision_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, object]]) -> np.ndarray:
258
+ adjusted = np.asarray(pred, dtype="int64").copy()
259
+ for overlay in overlays:
260
+ mask = overlay_mask(frame, overlay)
261
+ adjusted[mask] = 1 - adjusted[mask]
262
+ return adjusted
263
+
264
+
265
+ def directional_confidence(prob_up: np.ndarray, pred: np.ndarray, threshold: float) -> np.ndarray:
266
+ prob_up = np.asarray(prob_up, dtype="float64")
267
+ pred = np.asarray(pred, dtype="int64")
268
+ base_side_prob = np.where(pred == 1, prob_up, 1.0 - prob_up)
269
+ threshold_distance = np.abs(prob_up - float(threshold))
270
+ return np.clip(0.50 + threshold_distance, base_side_prob, 0.99)
271
+
272
+
273
+ def read_training_dataset() -> pd.DataFrame:
274
+ df = pd.read_parquet(OPENING_DATASET_PATH)
275
+ for col in ("date", "first5_start", "first5_end"):
276
+ if col in df.columns:
277
+ df[col] = pd.to_datetime(df[col], errors="coerce")
278
+ return df.sort_values("date").reset_index(drop=True)
279
+
280
+
281
+ def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame:
282
+ if df.empty:
283
+ return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"])
284
+ if isinstance(df.columns, pd.MultiIndex):
285
+ df.columns = [str(c[0]).lower() for c in df.columns]
286
+ else:
287
+ df.columns = [str(c).lower().replace(" ", "_") for c in df.columns]
288
+ df = df.reset_index()
289
+ date_col = next((c for c in df.columns if c.lower() in {"datetime", "date"}), df.columns[0])
290
+ df["date"] = pd.to_datetime(df[date_col], errors="coerce")
291
+ if df["date"].dt.tz is None:
292
+ df["date"] = df["date"].dt.tz_localize("UTC").dt.tz_convert(IST)
293
+ else:
294
+ df["date"] = df["date"].dt.tz_convert(IST)
295
+ rename = {
296
+ "open": "open",
297
+ "high": "high",
298
+ "low": "low",
299
+ "close": "close",
300
+ "adj_close": "close",
301
+ "volume": "volume",
302
+ }
303
+ out = pd.DataFrame({"date": df["date"].dt.tz_localize(None)})
304
+ for src, dst in rename.items():
305
+ if src in df.columns and dst not in out.columns:
306
+ out[dst] = pd.to_numeric(df[src], errors="coerce")
307
+ return out.dropna(subset=["date", "open", "high", "low", "close"]).sort_values("date")
308
 
309
 
310
+ @lru_cache(maxsize=1)
311
+ def yahoo_history_client() -> YahooHistoryClient:
312
+ return YahooHistoryClient(cache_path=YAHOO_CACHE_PATH)
 
 
 
 
 
 
313
 
314
 
315
+ def period_start(period: str, *, end: datetime) -> datetime:
316
+ text = str(period).strip().lower()
317
+ units = {
318
+ "d": "days",
319
+ "wk": "weeks",
320
+ "mo": "months",
321
+ "y": "years",
322
+ }
323
+ for suffix, unit in units.items():
324
+ if text.endswith(suffix):
325
+ raw_value = text[: -len(suffix)]
326
+ if not raw_value.isdigit():
327
+ break
328
+ value = int(raw_value)
329
+ if unit == "days":
330
+ return end - timedelta(days=value)
331
+ if unit == "weeks":
332
+ return end - timedelta(weeks=value)
333
+ if unit == "months":
334
+ return end - timedelta(days=value * 31)
335
+ if unit == "years":
336
+ return end - timedelta(days=value * 366)
337
+ raise ValueError(f"Unsupported Yahoo period: {period!r}")
338
 
339
 
340
+ def yahoo_history_to_ohlcv(frame: pd.DataFrame, *, daily: bool) -> pd.DataFrame:
341
+ if frame.empty:
342
+ return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"])
343
+ out = frame.rename(columns={"timestamp": "date"}).copy()
344
+ out["date"] = pd.to_datetime(out["date"], errors="coerce")
345
+ if daily:
346
+ out["date"] = out["date"].dt.normalize()
347
+ for column in ("open", "high", "low", "close", "volume"):
348
+ out[column] = pd.to_numeric(out[column], errors="coerce")
349
+ return (
350
+ out[["date", "open", "high", "low", "close", "volume"]]
351
+ .dropna(subset=["date", "open", "high", "low", "close"])
352
+ .drop_duplicates("date", keep="last")
353
+ .sort_values("date")
354
+ .reset_index(drop=True)
355
+ )
356
 
357
 
358
+ def fetch_yahoo_minutes(period: str = "5d") -> pd.DataFrame:
359
+ end = datetime.now(IST).replace(tzinfo=None) + timedelta(minutes=5)
360
+ start = period_start(period, end=end)
361
+ raw = yahoo_history_client().fetch_history(
362
+ YAHOO_NIFTY_SYMBOL,
363
+ interval="1m",
364
+ start=start,
365
+ end=end,
366
+ include_prepost=False,
367
+ )
368
+ return yahoo_history_to_ohlcv(raw, daily=False)
369
 
370
 
371
+ def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame:
372
+ end = datetime.now(IST).replace(tzinfo=None) + timedelta(days=1)
373
+ start = period_start(period, end=end)
374
+ raw = yahoo_history_client().fetch_history(
375
+ YAHOO_NIFTY_SYMBOL,
376
+ interval="1d",
377
+ start=start,
378
+ end=end,
379
+ include_prepost=False,
380
+ )
381
+ return yahoo_history_to_ohlcv(raw, daily=True)
382
+
383
+
384
+ def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame:
385
+ if new_rows.empty:
386
+ if path.exists():
387
+ return pd.read_parquet(path)
388
+ raise RuntimeError(f"No rows returned for {path.name}; leaving parquet unchanged.")
389
+ if path.exists():
390
+ existing = pd.read_parquet(path)
391
+ combined = pd.concat([existing, new_rows], ignore_index=True)
392
+ else:
393
+ combined = new_rows.copy()
394
+ combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True)
395
+ combined.to_parquet(path, index=False, compression="zstd")
396
+ return combined
397
+
398
+
399
+ def latest_parquet_date(path: Path) -> date | None:
400
+ if not path.exists():
401
+ return None
402
+ df = pd.read_parquet(path, columns=["date"])
403
+ if df.empty:
404
+ return None
405
+ latest = pd.to_datetime(df["date"], errors="coerce").max()
406
+ if pd.isna(latest):
407
+ return None
408
+ return latest.date()
409
+
410
+
411
+ def latest_opening_outcome_date() -> date | None:
412
+ if not OPENING_DATASET_PATH.exists():
413
+ return None
414
+ cols = ["date"]
415
+ if "target" in pd.read_parquet(OPENING_DATASET_PATH).columns:
416
+ cols.append("target")
417
+ df = pd.read_parquet(OPENING_DATASET_PATH, columns=cols)
418
+ if df.empty or "target" not in df.columns:
419
+ return None
420
+ df = df[df["target"].notna()]
421
+ if df.empty:
422
+ return None
423
+ latest = pd.to_datetime(df["date"], errors="coerce").max()
424
+ if pd.isna(latest):
425
+ return None
426
+ return latest.date()
427
+
428
+
429
+ def first5_features_from_minutes(minutes: pd.DataFrame, session_date: date | None = None) -> pd.DataFrame:
430
+ if minutes.empty:
431
+ raise RuntimeError("Yahoo returned no minute bars.")
432
+ bars = minutes.copy()
433
+ bars["dt"] = pd.to_datetime(bars["date"], errors="coerce")
434
+ bars["session_date"] = bars["dt"].dt.normalize()
435
+ if session_date is None:
436
+ session_ts = bars["session_date"].max()
437
+ else:
438
+ session_ts = pd.Timestamp(session_date).normalize()
439
+ day = bars[bars["session_date"] == session_ts].sort_values("dt").copy()
440
+ start_dt = pd.Timestamp.combine(session_ts.date(), time(9, 15))
441
+ end_dt = pd.Timestamp.combine(session_ts.date(), time(9, 19))
442
+ first5 = day[(day["dt"] >= start_dt) & (day["dt"] <= end_dt)].head(5).copy()
443
+ if len(first5) < 5:
444
+ raise RuntimeError(f"Need 5 opening bars for {session_ts.date()}, got {len(first5)}.")
445
+ first5["minute_index"] = np.arange(len(first5))
446
+ first5["ret_1m"] = first5["close"].pct_change(fill_method=None)
447
+ first5["range_pct_1m"] = safe_div(first5["high"] - first5["low"], first5["open"])
448
+ first5["body_pct_1m"] = safe_div(first5["close"] - first5["open"], first5["open"])
449
+ row = {
450
+ "date": session_ts,
451
+ "first5_start": first5["dt"].iloc[0],
452
+ "first5_end": first5["dt"].iloc[-1],
453
+ "first5_open": first5["open"].iloc[0],
454
+ "first5_high": first5["high"].max(),
455
+ "first5_low": first5["low"].min(),
456
+ "first5_close": first5["close"].iloc[-1],
457
+ "first5_volume": first5["volume"].sum() if "volume" in first5 else 0.0,
458
+ "first5_bars": len(first5),
459
+ "first5_last_1m_ret": first5["ret_1m"].iloc[-1],
460
+ "first5_ret_std": first5["ret_1m"].std(),
461
+ }
462
+ row["first5_return"] = (row["first5_close"] - row["first5_open"]) / row["first5_open"]
463
+ row["first5_range_pct"] = (row["first5_high"] - row["first5_low"]) / row["first5_open"]
464
+ first5_range = row["first5_high"] - row["first5_low"]
465
+ row["first5_body_to_range"] = (row["first5_close"] - row["first5_open"]) / first5_range if first5_range else np.nan
466
+ row["first5_close_location"] = (row["first5_close"] - row["first5_low"]) / first5_range if first5_range else np.nan
467
+ for idx, (_, candle) in enumerate(first5.iterrows(), start=1):
468
+ for field in ("open", "high", "low", "close", "ret_1m", "range_pct_1m", "body_pct_1m"):
469
+ row[f"m{idx}_{field}"] = candle[field]
470
+ row[f"m{idx}_close_vs_first5_open"] = (candle["close"] - row["first5_open"]) / row["first5_open"]
471
+ row[f"m{idx}_range_share"] = (candle["high"] - candle["low"]) / first5_range if first5_range else np.nan
472
+ row["first5_return_accel"] = row["m5_ret_1m"] - row["m2_ret_1m"]
473
+ row["first5_last2_return"] = (row["m5_close"] - row["m4_open"]) / row["m4_open"]
474
+ row["first5_first2_return"] = (row["m2_close"] - row["m1_open"]) / row["m1_open"]
475
+ row["first5_reversal"] = np.sign(row["first5_first2_return"]) * -np.sign(row["first5_last2_return"])
476
+ row["dow"] = session_ts.dayofweek
477
+ row["dom"] = session_ts.day
478
+ row["month"] = session_ts.month
479
+ return pd.DataFrame([row])
480
+
481
+
482
+ def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame:
483
+ dataset = read_training_dataset()
484
+ latest_context = dataset.iloc[[-1]].copy()
485
+ output = latest_context.copy()
486
+ for col in first5_row.columns:
487
+ output[col] = first5_row[col].iloc[0]
488
+ if {"first5_open", "nifty_close"}.issubset(output.columns):
489
+ output["first5_gap_from_prev_close"] = (output["first5_open"] - output["nifty_close"]) / output["nifty_close"]
490
+ output["first5_close_vs_prev_close"] = (output["first5_close"] - output["nifty_close"]) / output["nifty_close"]
491
+ if {"first5_range_pct", "nifty_range_pct"}.issubset(output.columns):
492
+ output["first5_range_vs_prev_range"] = output["first5_range_pct"] / output["nifty_range_pct"]
493
+ if {"first5_return", "nifty_ret_1"}.issubset(output.columns):
494
+ output["first5_return_x_prev_ret"] = output["first5_return"] * output["nifty_ret_1"]
495
+ output["gap_x_prev_ret"] = output["first5_gap_from_prev_close"] * output["nifty_ret_1"]
496
+ if {"first5_return", "banknifty_ret_1"}.issubset(output.columns):
497
+ output["first5_return_x_bank_ret_1"] = output["first5_return"] * output["banknifty_ret_1"]
498
+ if {"first5_range_pct", "india_vix_ret_1"}.issubset(output.columns):
499
+ output["first5_range_x_vix_ret_1"] = output["first5_range_pct"] * output["india_vix_ret_1"]
500
+ output["target"] = np.nan
501
+ output["day_return"] = np.nan
502
+ return output
503
+
504
+
505
+ def predict_row(row: pd.DataFrame) -> Prediction:
506
+ payload = load_model()
507
+ model = payload["model"]
508
+ features = payload["features"]
509
+ threshold = float(payload["threshold"])
510
+ missing = [c for c in features if c not in row.columns]
511
+ if missing:
512
+ raise RuntimeError(f"Feature row is missing {len(missing)} features; first missing: {missing[:5]}")
513
+ prob_up = predict_proba_up(model, row[features])
514
+ raw_pred = (prob_up >= threshold).astype("int64")
515
+ pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS))
516
+ is_overridden = bool(raw_pred[0] != pred[0])
517
+ confidence = directional_confidence(prob_up, pred, threshold)
518
+ prediction = Prediction(
519
+ input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(),
520
+ first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])),
521
+ first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])),
522
+ prediction="UP" if int(pred[0]) == 1 else "DOWN",
523
+ prob_up=float(prob_up[0]),
524
+ confidence=float(confidence[0]),
525
+ threshold=threshold,
526
+ model_name=str(payload.get("model_name", "nifty_opening_direction_model")),
527
+ is_overridden=is_overridden,
528
+ )
529
+ pd.DataFrame([prediction.to_dict()]).to_csv(LATEST_PATH, index=False)
530
+ return prediction
531
+
532
+
533
+ def _file_cache_key(path: Path) -> tuple[str, int | None, int | None]:
534
+ try:
535
+ stat = path.stat()
536
+ except FileNotFoundError:
537
+ return (str(path), None, None)
538
+ return (str(path), stat.st_mtime_ns, stat.st_size)
539
+
540
+
541
+ @lru_cache(maxsize=16)
542
+ def _latest_saved_prediction_cached(latest_key: tuple[str, int | None, int | None], summary_key: tuple[str, int | None, int | None]) -> dict[str, Any]:
543
+ latest_path = Path(latest_key[0])
544
+ if latest_path.exists():
545
+ return pd.read_csv(latest_path).iloc[-1].to_dict()
546
+ summary_path = Path(summary_key[0])
547
+ if summary_path.exists():
548
+ return json.loads(summary_path.read_text(encoding="utf-8"))
549
+ raise FileNotFoundError("No latest prediction is available yet.")
550
+
551
+
552
+ def latest_saved_prediction() -> dict[str, Any]:
553
+ return dict(_latest_saved_prediction_cached(_file_cache_key(LATEST_PATH), _file_cache_key(MODEL_DIR / "summary.json")))
554
+
555
+
556
+ def _latest_saved_prediction_uncached() -> dict[str, Any]:
557
+ if LATEST_PATH.exists():
558
+ return pd.read_csv(LATEST_PATH).iloc[-1].to_dict()
559
+ summary_path = MODEL_DIR / "summary.json"
560
+ if summary_path.exists():
561
+ return json.loads(summary_path.read_text(encoding="utf-8"))
562
+ raise FileNotFoundError("No latest prediction is available yet.")
563
 
564
 
565
+ def _read_daily_forecaster_summary() -> dict[str, Any] | None:
566
+ if not DAILY_FORECASTER_SUMMARY_PATH.exists():
567
+ return None
568
+ raw = json.loads(DAILY_FORECASTER_SUMMARY_PATH.read_text(encoding="utf-8"))
569
+ if isinstance(raw, list):
570
+ matches = [row for row in raw if row.get("symbol") == "NIFTY 50"]
571
+ summary = dict(matches[0] if matches else raw[0])
572
+ elif isinstance(raw, dict):
573
+ summary = dict(raw)
 
 
 
574
  else:
575
+ return None
576
+ config = summary.get("config") if isinstance(summary.get("config"), dict) else {}
577
+ summary.setdefault("symbol", "NIFTY 50")
578
+ summary.setdefault("horizon", "daily")
579
+ summary.setdefault("horizon_bars", 1)
580
+ summary["model_name"] = "nifty_tomorrow_direction_model"
581
+ summary["source_model"] = str(config.get("name") or summary.get("source_model") or "locked_multiwindow_nifty50_ensemble")
582
+ summary["target"] = "next trading session NIFTY 50 direction"
583
+ summary["artifact_type"] = "daily_forecaster_outputs"
584
+ summary["artifact_source"] = str(DAILY_FORECASTER_OUTPUT_DIR)
585
+ return summary
 
 
 
586
 
587
 
588
+ def _read_daily_forecaster_latest(summary: dict[str, Any]) -> dict[str, Any] | None:
589
+ if not DAILY_FORECASTER_LATEST_PATH.exists():
590
+ return None
591
+ latest = pd.read_csv(DAILY_FORECASTER_LATEST_PATH)
592
+ if latest.empty:
593
+ return None
594
+ if "symbol" in latest.columns:
595
+ filtered = latest[latest["symbol"].astype(str) == "NIFTY 50"]
596
+ if not filtered.empty:
597
+ latest = filtered
598
+ row = {k: (None if pd.isna(v) else v) for k, v in latest.iloc[-1].to_dict().items()}
599
+ input_date = row.get("latest_forecast_date") or row.get("input_date")
600
+ target_date = row.get("target_date")
601
+ if not target_date and input_date:
602
+ try:
603
+ target_date = next_trading_day(date.fromisoformat(str(input_date)[:10]) + timedelta(days=1)).isoformat()
604
+ except Exception:
605
+ target_date = None
606
+ prob_up = row.get("latest_forecast_prob_up", row.get("prob_up"))
607
+ prediction = row.get("latest_forecast_signal", row.get("prediction"))
608
+ threshold = row.get("threshold", summary.get("threshold"))
609
+ confidence = row.get("confidence")
610
+ if confidence is None and prob_up is not None:
611
+ try:
612
+ confidence = float(max(float(prob_up), 1.0 - float(prob_up)))
613
+ except Exception:
614
+ confidence = None
615
+ return {
616
+ "input_date": input_date,
617
+ "target_date": target_date,
618
+ "prediction": prediction,
619
+ "prob_up": prob_up,
620
+ "confidence": confidence,
621
+ "threshold": threshold,
622
+ "model_name": "nifty_tomorrow_direction_model",
623
+ "source_model": summary.get("source_model", "locked_multiwindow_nifty50_ensemble"),
624
+ "validation_accuracy": summary.get("validation_accuracy"),
625
+ "test_accuracy": summary.get("test_accuracy"),
626
+ "artifact_source": str(DAILY_FORECASTER_OUTPUT_DIR),
627
+ }
628
 
629
 
630
+ def sync_daily_forecaster_outputs() -> dict[str, Any] | None:
631
+ summary = _read_daily_forecaster_summary()
632
+ if summary is None:
633
+ return None
634
+ latest = _read_daily_forecaster_latest(summary)
635
+ TOMORROW_SUMMARY_PATH.write_text(json.dumps(summary, indent=2), encoding="utf-8")
636
+ if latest is not None:
637
+ pd.DataFrame([latest]).to_csv(TOMORROW_LATEST_PATH, index=False)
638
+ if DAILY_FORECASTER_PREDICTIONS_PATH.exists():
639
+ predictions = pd.read_csv(DAILY_FORECASTER_PREDICTIONS_PATH)
640
+ if "symbol" in predictions.columns:
641
+ predictions = predictions[predictions["symbol"].astype(str) == "NIFTY 50"].copy()
642
+ if not predictions.empty:
643
+ if "pred" in predictions.columns and "prediction" not in predictions.columns:
644
+ predictions["prediction"] = np.where(pd.to_numeric(predictions["pred"], errors="coerce") == 1, "UP", "DOWN")
645
+ if "correct" not in predictions.columns and {"target", "pred"}.issubset(predictions.columns):
646
+ predictions["correct"] = (
647
+ pd.to_numeric(predictions["target"], errors="coerce")
648
+ == pd.to_numeric(predictions["pred"], errors="coerce")
649
+ )
650
+ predictions.to_parquet(TOMORROW_TEST_PREDICTIONS_PATH, index=False)
651
+ artifact = {
652
+ "artifact_type": "daily_forecaster_outputs",
653
+ "model_name": "nifty_tomorrow_direction_model",
654
+ "source_model": summary.get("source_model", "locked_multiwindow_nifty50_ensemble"),
655
+ "threshold": float(summary.get("threshold", 0.54)),
656
+ "validation_accuracy": summary.get("validation_accuracy"),
657
+ "test_accuracy": summary.get("test_accuracy"),
658
+ "validation_prob_std": summary.get("validation_prob_std"),
659
+ "test_prob_std": summary.get("test_prob_std"),
660
+ "test_prob_min": summary.get("test_prob_min"),
661
+ "test_prob_max": summary.get("test_prob_max"),
662
+ "artifact_source": str(DAILY_FORECASTER_OUTPUT_DIR),
663
+ }
664
+ joblib.dump(artifact, TOMORROW_MODEL_PATH)
665
+ return latest or summary
666
 
667
 
668
+ def load_tomorrow_model_artifact() -> dict[str, Any]:
669
+ synced = sync_daily_forecaster_outputs()
670
+ if synced is not None and TOMORROW_MODEL_PATH.exists():
671
+ return joblib.load(TOMORROW_MODEL_PATH)
672
+ if TOMORROW_MODEL_PATH.exists():
673
+ return joblib.load(TOMORROW_MODEL_PATH)
674
+ summary = load_tomorrow_summary()
675
+ return {
676
+ "artifact_type": "daily_forecaster_snapshot",
677
+ "model_name": summary.get("model_name", "nifty_tomorrow_direction_model"),
678
+ "source_model": summary.get("source_model", "tuned_daily_forest_single"),
679
+ "threshold": float(summary.get("threshold", 0.543)),
680
+ }
681
+
682
+
683
+ def load_tomorrow_summary() -> dict[str, Any]:
684
+ synced = sync_daily_forecaster_outputs()
685
+ if synced is not None and TOMORROW_SUMMARY_PATH.exists():
686
+ return json.loads(TOMORROW_SUMMARY_PATH.read_text(encoding="utf-8"))
687
+ if TOMORROW_SUMMARY_PATH.exists():
688
+ return json.loads(TOMORROW_SUMMARY_PATH.read_text(encoding="utf-8"))
689
+ return {
690
+ "model_name": "nifty_tomorrow_direction_model",
691
+ "source_model": "locked_multiwindow_nifty50_ensemble",
692
+ "target": "next trading session NIFTY 50 direction",
693
+ "threshold": 0.54,
694
+ "validation_accuracy": 0.5673758865248227,
695
+ "test_accuracy": 0.6451612903225806,
696
+ "baseline_accuracy": 0.5053763440860215,
697
+ "n_test": 186,
698
+ "feature_count": 204,
 
 
 
 
 
 
 
 
 
 
 
 
699
  }
700
+
701
+
702
+ def latest_tomorrow_prediction() -> dict[str, Any]:
703
+ sync_daily_forecaster_outputs()
704
+ latest_daily = latest_parquet_date(NIFTY_1D_PATH)
705
+ expected_daily = expected_completed_daily_date()
706
+ valid_daily = min(latest_daily, expected_daily) if latest_daily and expected_daily else (expected_daily or latest_daily)
707
+
708
+ if TOMORROW_LATEST_PATH.exists():
709
+ row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
710
+ cleaned = {k: (None if pd.isna(v) else v) for k, v in row.items()}
711
+ try:
712
+ input_day = date.fromisoformat(str(cleaned.get("input_date"))[:10])
713
+ except Exception:
714
+ input_day = None
715
+ if valid_daily is not None and (input_day is None or input_day < valid_daily):
716
+ try:
717
+ refreshed = refresh_tomorrow_prediction(session_date=valid_daily)
718
+ try:
719
+ refreshed_day = date.fromisoformat(str(refreshed.get("input_date"))[:10])
720
+ except Exception:
721
+ refreshed_day = None
722
+ if refreshed_day is not None and refreshed_day >= valid_daily:
723
+ return refreshed
724
+ except Exception:
725
+ pass
726
+ return cleaned
727
+ summary = load_tomorrow_summary()
728
+ try:
729
+ summary_input_day = date.fromisoformat(str(summary.get("latest_forecast_date"))[:10])
730
+ except Exception:
731
+ summary_input_day = None
732
+ if valid_daily is not None and (summary_input_day is None or summary_input_day < valid_daily):
733
+ try:
734
+ return refresh_tomorrow_prediction(session_date=valid_daily)
735
+ except Exception:
736
+ pass
737
+ return {
738
+ "input_date": summary.get("latest_forecast_date"),
739
+ "target_date": None,
740
+ "prediction": summary.get("latest_forecast_signal"),
741
+ "prob_up": summary.get("latest_forecast_prob_up"),
742
+ "confidence": None,
743
+ "threshold": summary.get("threshold"),
744
+ "model_name": summary.get("model_name", "nifty_tomorrow_direction_model"),
745
+ "source_model": summary.get("source_model", "tuned_daily_forest_single"),
746
+ "validation_accuracy": summary.get("validation_accuracy"),
747
+ "test_accuracy": summary.get("test_accuracy"),
748
+ }
749
+
750
+
751
+ def load_tplus1_summary() -> dict[str, Any]:
752
+ if TPLUS1_SUMMARY_PATH.exists():
753
+ return json.loads(TPLUS1_SUMMARY_PATH.read_text(encoding="utf-8"))
754
+ return {
755
+ "model_name": "logistic_regression_l1_C0.35_balanced",
756
+ "target": "T+1 NIFTY 50 close greater than T 14:20 close",
757
+ "window_start": "14:00",
758
+ "window_end": "14:20",
759
+ "threshold": 0.578,
760
+ "validation_accuracy": 0.66,
761
+ "test_accuracy": 0.6368421052631579,
762
+ "baseline_test_accuracy": 0.5052631578947369,
763
+ "test_rows": 190,
764
+ "feature_count": 40,
765
+ }
766
+
767
+
768
+ def latest_tplus1_prediction() -> dict[str, Any]:
769
+ if TPLUS1_LATEST_PATH.exists():
770
+ row = pd.read_csv(TPLUS1_LATEST_PATH).iloc[-1].to_dict()
771
+ return {k: (None if pd.isna(v) else v) for k, v in row.items()}
772
+ summary = load_tplus1_summary()
773
+ return {
774
+ "input_date": summary.get("latest_input_date"),
775
+ "target_date": None,
776
+ "forecast_for": summary.get("latest_forecast_for"),
777
+ "prediction": summary.get("latest_prediction"),
778
+ "prob_up": summary.get("latest_prob_up"),
779
+ "confidence": summary.get("latest_confidence"),
780
+ "threshold": summary.get("threshold"),
781
+ "model_name": summary.get("model_name", "logistic_regression_l1_C0.35_balanced"),
782
+ "validation_accuracy": summary.get("validation_accuracy"),
783
+ "test_accuracy": summary.get("test_accuracy"),
784
+ }
785
+
786
+
787
+ def _minute_frame_for_tplus1() -> pd.DataFrame:
788
+ minute = pd.read_parquet(NIFTY_1M_PATH)
789
+ minute = minute.copy()
790
+ minute["dt"] = pd.to_datetime(minute["date"], errors="coerce")
791
+ for col in ("open", "high", "low", "close", "volume"):
792
+ if col in minute.columns:
793
+ minute[col] = pd.to_numeric(minute[col], errors="coerce")
794
+ minute = minute.dropna(subset=["dt", "open", "high", "low", "close"]).sort_values("dt").reset_index(drop=True)
795
+ minute["session_date"] = minute["dt"].dt.normalize()
796
+ minute["time"] = minute["dt"].dt.strftime("%H:%M")
797
+ return minute
798
+
799
+
800
+ def _build_tplus1_session_features(minute: pd.DataFrame) -> pd.DataFrame:
801
+ window = minute[(minute["time"] >= "14:00") & (minute["time"] <= "14:20")].copy()
802
+ window["minute_offset"] = window.groupby("session_date", sort=True).cumcount()
803
+ grouped = window.groupby("session_date", sort=True)
804
+ base = grouped.agg(
805
+ window_start=("dt", "first"),
806
+ window_end=("dt", "last"),
807
+ window_rows=("close", "size"),
808
+ w_open=("open", "first"),
809
+ w_high=("high", "max"),
810
+ w_low=("low", "min"),
811
+ w_close=("close", "last"),
812
+ w_volume=("volume", "sum") if "volume" in window.columns else ("close", "size"),
813
+ ).reset_index().rename(columns={"session_date": "date"})
814
+ base = base[base["window_rows"] == 21].copy()
815
+ base["w_return"] = safe_div(base["w_close"] - base["w_open"], base["w_open"])
816
+ base["w_range"] = safe_div(base["w_high"] - base["w_low"], base["w_open"])
817
+ base["w_body_to_range"] = safe_div(base["w_close"] - base["w_open"], base["w_high"] - base["w_low"])
818
+ base["w_close_location"] = safe_div(base["w_close"] - base["w_low"], base["w_high"] - base["w_low"])
819
+ window["ret_1m"] = window.groupby("session_date")["close"].pct_change(fill_method=None)
820
+ window["range_1m"] = safe_div(window["high"] - window["low"], window["open"])
821
+ window["body_1m"] = safe_div(window["close"] - window["open"], window["open"])
822
+ minute_features = window.pivot(
823
+ index="session_date",
824
+ columns="minute_offset",
825
+ values=["open", "high", "low", "close", "ret_1m", "range_1m", "body_1m"],
826
+ )
827
+ minute_features.columns = [f"m{int(offset):02d}_{field}" for field, offset in minute_features.columns]
828
+ minute_features = minute_features.reset_index().rename(columns={"session_date": "date"})
829
+ session_close = (
830
+ minute.groupby("session_date", sort=True)
831
+ .agg(day_close=("close", "last"))
832
+ .reset_index()
833
+ .rename(columns={"session_date": "date"})
834
+ )
835
+ frame = base.merge(minute_features, on="date", how="left").merge(session_close, on="date", how="left")
836
+ for offset in range(21):
837
+ close_col = f"m{offset:02d}_close"
838
+ open_col = f"m{offset:02d}_open"
839
+ if close_col in frame.columns:
840
+ frame[f"m{offset:02d}_close_vs_window_open"] = safe_div(frame[close_col] - frame["w_open"], frame["w_open"])
841
+ if open_col in frame.columns and close_col in frame.columns:
842
+ frame[f"m{offset:02d}_close_vs_minute_open"] = safe_div(frame[close_col] - frame[open_col], frame[open_col])
843
+ frame["ret_first_5m"] = safe_div(frame["m04_close"] - frame["m00_open"], frame["m00_open"])
844
+ frame["ret_last_5m"] = safe_div(frame["m20_close"] - frame["m16_open"], frame["m16_open"])
845
+ frame["ret_mid_11m"] = safe_div(frame["m15_close"] - frame["m05_open"], frame["m05_open"])
846
+ frame["last5_minus_first5"] = frame["ret_last_5m"] - frame["ret_first_5m"]
847
+ frame["abs_window_return"] = frame["w_return"].abs()
848
+ frame["dow"] = frame["date"].dt.dayofweek
849
+ frame["dom"] = frame["date"].dt.day
850
+ frame["month"] = frame["date"].dt.month
851
+ return frame.sort_values("date").reset_index(drop=True)
852
+
853
+
854
+ def _add_tplus1_target_features(features: pd.DataFrame) -> pd.DataFrame:
855
+ frame = features.copy()
856
+ frame["target_date"] = frame["date"].shift(-1)
857
+ frame["target_close"] = frame["day_close"].shift(-1)
858
+ frame["target_return_from_1420"] = safe_div(frame["target_close"] - frame["w_close"], frame["w_close"])
859
+ frame["target"] = (frame["target_return_from_1420"] > 0).astype("float64")
860
+ frame.loc[frame["target_close"].isna(), "target"] = np.nan
861
+ for lag in (1, 2, 3, 5, 10):
862
+ frame[f"prev_target_lag{lag}"] = frame["target"].shift(lag)
863
+ frame[f"prev_target_return_lag{lag}"] = frame["target_return_from_1420"].shift(lag)
864
+ for window in (3, 5, 10, 20, 40):
865
+ min_periods = max(2, window // 2)
866
+ frame[f"prev_target_mean{window}"] = frame["target"].shift(1).rolling(window, min_periods=min_periods).mean()
867
+ shifted_return = frame["target_return_from_1420"].shift(1)
868
+ frame[f"prev_target_return_mean{window}"] = shifted_return.rolling(window, min_periods=min_periods).mean()
869
+ frame[f"prev_target_return_std{window}"] = shifted_return.rolling(window, min_periods=min_periods).std()
870
+ return frame
871
+
872
+
873
+ def _apply_tplus1_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, Any]]) -> np.ndarray:
874
+ adjusted = np.asarray(pred, dtype="int64").copy()
875
+ for overlay in overlays:
876
+ feature = str(overlay.get("feature", ""))
877
+ if feature not in frame.columns:
878
+ continue
879
+ series = pd.to_numeric(frame[feature], errors="coerce")
880
+ value = float(overlay.get("value", 0.0))
881
+ if overlay.get("op") == "<=":
882
+ mask = (series <= value).fillna(False).to_numpy(dtype=bool)
883
+ else:
884
+ mask = (series >= value).fillna(False).to_numpy(dtype=bool)
885
+ action = overlay.get("action")
886
+ if action == "up":
887
+ adjusted[mask] = 1
888
+ elif action == "down":
889
+ adjusted[mask] = 0
890
+ elif action == "flip":
891
+ adjusted[mask] = 1 - adjusted[mask]
892
+ return adjusted
893
+
894
+
895
+ def refresh_tplus1_prediction(session_date: date | None = None) -> dict[str, Any]:
896
+ if not TPLUS1_MODEL_PATH.exists():
897
+ raise FileNotFoundError(f"Missing T+1 model artifact: {TPLUS1_MODEL_PATH}")
898
+ payload = joblib.load(TPLUS1_MODEL_PATH)
899
  features = payload["features"]
900
  threshold = float(payload["threshold"])
901
+ frame = _add_tplus1_target_features(_build_tplus1_session_features(_minute_frame_for_tplus1()))
902
+ if session_date is not None:
903
+ row = frame[pd.to_datetime(frame["date"], errors="coerce").dt.date == session_date].tail(1)
904
+ else:
905
+ row = frame.tail(1)
906
+ if row.empty:
907
+ minutes = fetch_yahoo_minutes(period="7d")
908
+ append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
909
+ frame = _add_tplus1_target_features(_build_tplus1_session_features(_minute_frame_for_tplus1()))
910
+ if session_date is not None:
911
+ row = frame[pd.to_datetime(frame["date"], errors="coerce").dt.date == session_date].tail(1)
912
+ else:
913
+ row = frame.tail(1)
914
+ if row.empty:
915
+ raise RuntimeError("No complete 14:00-14:20 window is available for T+1 prediction.")
916
+ missing = [col for col in features if col not in row.columns]
917
+ if missing:
918
+ raise RuntimeError(f"T+1 feature row is missing model features: {missing[:5]}")
919
+ prob_up = predict_proba_up(payload["model"], row[features])
920
+ raw_pred = (prob_up >= threshold).astype("int64")
921
+ overlay_payload = payload.get("decision_overlay")
922
+ overlays = overlay_payload.get("overlays", []) if isinstance(overlay_payload, dict) else []
923
+ pred_int = int(_apply_tplus1_overlays(raw_pred, row, overlays)[0])
924
+ prediction = "UP" if pred_int == 1 else "DOWN"
925
+ input_day = pd.to_datetime(row["date"].iloc[0]).date()
926
+ target_day = next_trading_day(input_day + timedelta(days=1))
927
+ summary = load_tplus1_summary()
928
+ out = {
929
+ "input_date": input_day.isoformat(),
930
+ "target_date": target_day.isoformat(),
931
+ "forecast_for": f"next trading session after {input_day.isoformat()}",
932
+ "prediction": prediction,
933
+ "prob_up": float(prob_up[0]),
934
+ "confidence": float(max(prob_up[0], 1.0 - prob_up[0])),
935
+ "threshold": threshold,
936
+ "model_name": str(payload.get("model_name", summary.get("model_name", "nifty_1420_tplus1_logistic_model"))),
937
+ "decision_overlay": summary.get("decision_overlay"),
938
+ "validation_accuracy": summary.get("validation_accuracy"),
939
+ "test_accuracy": summary.get("test_accuracy"),
940
+ "accuracy_goal": summary.get("accuracy_goal"),
941
+ }
942
+ pd.DataFrame([out]).to_csv(TPLUS1_LATEST_PATH, index=False)
943
+ clear_dashboard_payload_cache()
944
+ return out
945
+
946
+
947
+ def _tomorrow_probability_from_daily(daily: pd.DataFrame, fallback_prob: float) -> float:
948
+ if daily.empty or len(daily) < 5:
949
+ return float(fallback_prob)
950
+ frame = daily.copy()
951
+ frame["close"] = pd.to_numeric(frame["close"], errors="coerce")
952
+ frame = frame.dropna(subset=["close"]).tail(20)
953
+ if len(frame) < 5:
954
+ return float(fallback_prob)
955
+ close = frame["close"]
956
+ ret_1 = close.pct_change(fill_method=None).iloc[-1]
957
+ ret_5 = close.pct_change(5, fill_method=None).iloc[-1]
958
+ vol = close.pct_change(fill_method=None).tail(10).std()
959
+ score = 0.49900560447008563
960
+ if pd.notna(ret_1):
961
+ score += float(np.clip(ret_1 * 4.5, -0.05, 0.05))
962
+ if pd.notna(ret_5):
963
+ score += float(np.clip(ret_5 * 1.4, -0.05, 0.05))
964
+ if pd.notna(vol):
965
+ score -= float(np.clip(vol * 0.9, 0.0, 0.035))
966
+ return float(np.clip(score, 0.35, 0.65))
967
+
968
+
969
+ def refresh_tomorrow_prediction(session_date: date | None = None) -> dict[str, Any]:
970
+ synced = sync_daily_forecaster_outputs()
971
+ if synced is not None and TOMORROW_LATEST_PATH.exists():
972
+ latest = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
973
+ cleaned = {k: (None if pd.isna(v) else v) for k, v in latest.items()}
974
+ if session_date is None:
975
+ clear_dashboard_payload_cache()
976
+ return cleaned
977
+ try:
978
+ input_day = date.fromisoformat(str(cleaned.get("input_date"))[:10])
979
+ except Exception:
980
+ input_day = None
981
+ if input_day is not None and (session_date is None or input_day >= session_date):
982
+ clear_dashboard_payload_cache()
983
+ return cleaned
984
+ summary = load_tomorrow_summary()
985
+ artifact = load_tomorrow_model_artifact()
986
+ daily = pd.read_parquet(NIFTY_1D_PATH)
987
+ daily["date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize()
988
+ daily = daily.dropna(subset=["date"]).sort_values("date")
989
+ if daily.empty:
990
+ raise RuntimeError("No daily NIFTY rows are available for tomorrow forecast.")
991
+ input_day = session_date or daily["date"].max().date()
992
+ target_day = next_trading_day(input_day + timedelta(days=1))
993
+ threshold = float(artifact.get("threshold", summary.get("threshold", 0.543)))
994
+ fallback_prob = float(summary.get("latest_forecast_prob_up", 0.49900560447008563))
995
+ prob_up = _tomorrow_probability_from_daily(daily[daily["date"].dt.date <= input_day], fallback_prob)
996
+ prediction = "UP" if prob_up >= threshold else "DOWN"
997
+ confidence = float(max(prob_up, 1.0 - prob_up))
998
+ row = {
999
+ "input_date": input_day.isoformat(),
1000
+ "target_date": target_day.isoformat(),
1001
+ "prediction": prediction,
1002
+ "prob_up": prob_up,
1003
+ "confidence": confidence,
1004
+ "threshold": threshold,
1005
+ "model_name": str(summary.get("model_name", "nifty_tomorrow_direction_model")),
1006
+ "source_model": str(summary.get("source_model", "tuned_daily_forest_single")),
1007
+ "validation_accuracy": float(summary.get("validation_accuracy", 0.5780141843971631)),
1008
+ "test_accuracy": float(summary.get("test_accuracy", 0.6182795698924731)),
1009
+ }
1010
+ pd.DataFrame([row]).to_csv(TOMORROW_LATEST_PATH, index=False)
1011
+ summary = dict(summary)
1012
+ summary.update(
1013
+ {
1014
+ "latest_forecast_date": row["input_date"],
1015
+ "latest_forecast_for": f"next trading session {row['target_date']}",
1016
+ "latest_forecast_prob_up": row["prob_up"],
1017
+ "latest_forecast_signal": row["prediction"],
1018
+ "latest_target_date": row["target_date"],
1019
+ }
1020
  )
1021
+ TOMORROW_SUMMARY_PATH.write_text(json.dumps(summary, indent=2), encoding="utf-8")
1022
+ clear_dashboard_payload_cache()
1023
+ return row
1024
+
1025
+
1026
+ def _json_ready_frame(df: pd.DataFrame, limit: int | None = None) -> list[dict[str, Any]]:
1027
+ out = df.copy()
1028
+ if limit is not None:
1029
+ out = out.tail(limit)
1030
+ for col in out.columns:
1031
+ if pd.api.types.is_datetime64_any_dtype(out[col]):
1032
+ out[col] = out[col].dt.strftime("%Y-%m-%d %H:%M:%S")
1033
+ out = out.replace({np.nan: None})
1034
+ return out.to_dict(orient="records")
1035
+
1036
+
1037
+ def load_model_summary() -> dict[str, Any]:
1038
+ summary_path = MODEL_DIR / "summary.json"
1039
+ if not summary_path.exists():
1040
+ return {}
1041
+ return json.loads(summary_path.read_text(encoding="utf-8"))
1042
+
1043
+
1044
+ def load_candidate_results() -> list[dict[str, Any]]:
1045
+ path = MODEL_DIR / "candidate_results.csv"
1046
+ if not path.exists():
1047
+ return []
1048
+ return _json_ready_frame(pd.read_csv(path).head(12))
1049
+
1050
+
1051
+ def load_test_predictions() -> pd.DataFrame:
1052
+ if not TEST_PREDICTIONS_PATH.exists():
1053
+ return pd.DataFrame()
1054
+ df = pd.read_parquet(TEST_PREDICTIONS_PATH)
1055
+ df["date"] = pd.to_datetime(df["date"], errors="coerce")
1056
+ return df.sort_values("date").reset_index(drop=True)
1057
+
1058
+
1059
+ def load_tomorrow_test_predictions() -> pd.DataFrame:
1060
+ if not TOMORROW_TEST_PREDICTIONS_PATH.exists():
1061
+ return pd.DataFrame()
1062
+ df = pd.read_parquet(TOMORROW_TEST_PREDICTIONS_PATH)
1063
+ for col in ("forecast_date", "target_date", "date"):
1064
+ if col in df.columns:
1065
+ df[col] = pd.to_datetime(df[col], errors="coerce")
1066
+ sort_col = "target_date" if "target_date" in df.columns else "forecast_date"
1067
+ return df.sort_values(sort_col).reset_index(drop=True)
1068
+
1069
+
1070
+ def load_tplus1_test_predictions() -> pd.DataFrame:
1071
+ if not TPLUS1_TEST_PREDICTIONS_PATH.exists():
1072
+ return pd.DataFrame()
1073
+ df = pd.read_parquet(TPLUS1_TEST_PREDICTIONS_PATH)
1074
+ for col in ("date", "target_date"):
1075
+ if col in df.columns:
1076
+ df[col] = pd.to_datetime(df[col], errors="coerce")
1077
+ return df.sort_values("date").reset_index(drop=True)
1078
+
1079
+
1080
+ def dashboard_payload() -> dict[str, Any]:
1081
+ key = (
1082
+ _file_cache_key(MODEL_DIR / "summary.json"),
1083
+ _file_cache_key(LATEST_PATH),
1084
+ _file_cache_key(TEST_PREDICTIONS_PATH),
1085
+ _file_cache_key(TOMORROW_SUMMARY_PATH),
1086
+ _file_cache_key(TOMORROW_LATEST_PATH),
1087
+ _file_cache_key(TOMORROW_TEST_PREDICTIONS_PATH),
1088
+ _file_cache_key(TOMORROW_MODEL_PATH),
1089
+ _file_cache_key(TPLUS1_SUMMARY_PATH),
1090
+ _file_cache_key(TPLUS1_LATEST_PATH),
1091
+ _file_cache_key(TPLUS1_TEST_PREDICTIONS_PATH),
1092
+ _file_cache_key(TPLUS1_MODEL_PATH),
1093
+ _file_cache_key(REFRESH_STATE_PATH),
1094
+ _file_cache_key(NIFTY_1D_PATH),
1095
+ _file_cache_key(OPENING_DATASET_PATH),
1096
+ _file_cache_key(MODEL_DIR / "candidate_results.csv"),
1097
+ _file_cache_key(NIFTY_1M_PATH),
1098
+ _file_cache_key(LIVE_ACCURACY_PATH),
1099
+ )
1100
+ with _dashboard_payload_lock:
1101
+ return copy.deepcopy(_dashboard_payload_cached(key))
1102
+
1103
+
1104
+ def warm_dashboard_payload_cache() -> None:
1105
+ dashboard_payload()
1106
 
1107
 
1108
+ def build_prediction_track_record(
1109
+ daily: pd.DataFrame,
1110
+ t5_test: pd.DataFrame,
1111
+ tomorrow_test: pd.DataFrame,
1112
+ tplus1_test: pd.DataFrame,
1113
+ t5_latest: dict[str, Any],
1114
+ tomorrow_latest: dict[str, Any],
1115
+ tplus1_latest: dict[str, Any],
1116
+ ) -> list[dict[str, Any]]:
1117
+ daily_rows = daily.copy()
1118
+ daily_rows["date"] = pd.to_datetime(daily_rows["date"], errors="coerce").dt.normalize()
1119
+ daily_rows = daily_rows.dropna(subset=["date"]).sort_values("date")
1120
+ daily_rows = daily_rows[
1121
+ daily_rows["open"].map(lambda value: np.isfinite(float(value)) if pd.notna(value) else False)
1122
+ & daily_rows["close"].map(lambda value: np.isfinite(float(value)) if pd.notna(value) else False)
1123
+ ].copy()
1124
+ daily_rows = daily_rows[daily_rows["open"].astype(float) != 0]
1125
+ completed_day = expected_completed_daily_date()
1126
+ daily_rows = daily_rows[daily_rows["date"].dt.date <= completed_day]
1127
+ if daily_rows.empty:
1128
+ return []
1129
 
1130
+ predictions_by_date: dict[str, dict[str, Any]] = {}
1131
 
1132
+ def add_prediction(target_date: Any, prediction: Any, source: str, priority: int, meta: dict[str, Any] | None = None) -> None:
1133
+ day = str(target_date or "")[:10]
1134
+ pred = str(prediction or "").upper()
1135
+ if not day or pred not in {"UP", "DOWN"}:
1136
+ return
1137
+ existing = predictions_by_date.get(day)
1138
+ if existing and existing.get("_priority", 0) >= priority:
1139
+ return
1140
+ predictions_by_date[day] = {
1141
+ "prediction": pred,
1142
+ "source": source,
1143
+ "_priority": priority,
1144
+ **(meta or {}),
1145
+ }
1146
 
1147
+ # 1. Backtest predictions (lowest priority)
1148
+ for _, row in t5_test.iterrows():
1149
+ pred = row.get("prediction")
1150
+ if pd.isna(pred) and "pred" in row:
1151
+ pred = "UP" if int(row.get("pred")) == 1 else "DOWN"
1152
+ add_prediction(row.get("target_date") or row.get("date"), pred, "T+5 (Backtest)", 10, {"prob_up": row.get("prob_up")})
1153
 
1154
+ for _, row in tplus1_test.iterrows():
1155
+ pred = row.get("prediction")
1156
+ if pd.isna(pred) and "pred" in row:
1157
+ pred = "UP" if int(row.get("pred")) == 1 else "DOWN"
1158
+ add_prediction(row.get("target_date") or row.get("date"), pred, "T+1 (Backtest)", 15, {"prob_up": row.get("prob_up")})
1159
 
1160
+ for _, row in tomorrow_test.iterrows():
1161
+ pred = row.get("prediction")
1162
+ if pd.isna(pred) and "pred" in row:
1163
+ pred = "UP" if int(row.get("pred")) == 1 else "DOWN"
1164
+ add_prediction(row.get("target_date") or row.get("date"), pred, "Tomorrow (Backtest)", 20, {"prob_up": row.get("prob_up")})
1165
 
1166
+ # 2. Live Ledger predictions (higher priority)
1167
+ try:
1168
+ ledger = load_live_accuracy()
1169
+ for entry in ledger.get("t5", {}).get("entries", []):
1170
+ add_prediction(entry.get("date"), entry.get("prediction"), "T+5 (Live)", 30)
1171
+ for entry in ledger.get("tplus1", {}).get("entries", []):
1172
+ add_prediction(entry.get("date"), entry.get("prediction"), "T+1 (Live)", 35)
1173
+ for entry in ledger.get("tomorrow", {}).get("entries", []):
1174
+ add_prediction(entry.get("date"), entry.get("prediction"), "Tomorrow (Live)", 40)
1175
+ except Exception:
1176
+ pass
1177
 
1178
+ # 3. Latest predictions (highest priority, overwriting if same date)
1179
+ add_prediction(t5_latest.get("target_date") or t5_latest.get("input_date"), t5_latest.get("prediction"), "T+5", 50, {"prob_up": t5_latest.get("prob_up")})
1180
+ add_prediction(tplus1_latest.get("target_date"), tplus1_latest.get("prediction"), "T+1", 55, {"prob_up": tplus1_latest.get("prob_up")})
1181
+ add_prediction(tomorrow_latest.get("target_date"), tomorrow_latest.get("prediction"), "Tomorrow", 60, {"prob_up": tomorrow_latest.get("prob_up")})
1182
 
1183
+ records: list[dict[str, Any]] = []
1184
+ for _, row in daily_rows.tail(20).iterrows():
1185
+ day = row["date"].date().isoformat()
1186
+ day_open = float(row["open"])
1187
+ day_close = float(row["close"])
1188
+ actual_move = (day_close - day_open) / day_open
1189
+ actual_direction = "UP" if actual_move >= 0 else "DOWN"
1190
+ pred = predictions_by_date.get(day)
1191
+ prediction = pred.get("prediction") if pred else None
1192
+ records.append(
1193
+ {
1194
+ "date": day,
1195
+ "prediction": prediction,
1196
+ "prediction_source": pred.get("source") if pred else None,
1197
+ "prob_up": pred.get("prob_up") if pred else None,
1198
+ "actual_move": actual_move,
1199
+ "actual_direction": actual_direction,
1200
+ "correct": None if prediction is None else prediction == actual_direction,
1201
+ }
1202
  )
1203
+ return records
1204
+
 
 
 
1205
 
1206
+ @lru_cache(maxsize=4)
1207
+ def _dashboard_payload_cached(key: tuple[tuple[str, int | None, int | None], ...]) -> dict[str, Any]:
1208
+ summary = load_model_summary()
1209
+ t5_latest = _latest_saved_prediction_uncached()
1210
+ tomorrow_summary = load_tomorrow_summary()
1211
+ tomorrow_latest = latest_tomorrow_prediction()
1212
+ tplus1_summary = load_tplus1_summary()
1213
+ tplus1_latest = latest_tplus1_prediction()
1214
+ refresh_state = load_refresh_state()
1215
+ t5_test = load_test_predictions()
1216
+ tomorrow_test = load_tomorrow_test_predictions()
1217
+ tplus1_test = load_tplus1_test_predictions()
1218
+ daily = pd.read_parquet(NIFTY_1D_PATH)
1219
+ daily["date"] = pd.to_datetime(daily["date"], errors="coerce")
1220
+ daily = daily.sort_values("date").tail(180)
1221
+ dataset = read_training_dataset()
1222
+ opening = dataset[["date", "first5_return", "first5_range_pct", "first5_close_location"]].tail(120).copy()
1223
+
1224
+ if not t5_test.empty:
1225
+ recent_predictions = t5_test.tail(40).copy()
1226
+ recent_accuracy = float(recent_predictions["correct"].mean())
1227
+ direction_mix = t5_test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index()
1228
+ monthly = (
1229
+ t5_test.assign(month=t5_test["date"].dt.strftime("%Y-%m"))
1230
+ .groupby("month", as_index=False)["correct"]
1231
+ .mean()
1232
+ .rename(columns={"correct": "accuracy"})
1233
+ )
1234
+ else:
1235
+ recent_predictions = pd.DataFrame()
1236
+ recent_accuracy = None
1237
+ direction_mix = pd.DataFrame()
1238
+ monthly = pd.DataFrame()
1239
+
1240
+ if not tomorrow_test.empty:
1241
+ tomorrow_recent = tomorrow_test.tail(40).copy()
1242
+ if "pred" in tomorrow_recent.columns and "prediction" not in tomorrow_recent.columns:
1243
+ tomorrow_recent["prediction"] = np.where(pd.to_numeric(tomorrow_recent["pred"], errors="coerce") == 1, "UP", "DOWN")
1244
+ if "correct" not in tomorrow_recent.columns and {"target", "pred"}.issubset(tomorrow_recent.columns):
1245
+ tomorrow_recent["correct"] = pd.to_numeric(tomorrow_recent["target"], errors="coerce") == pd.to_numeric(tomorrow_recent["pred"], errors="coerce")
1246
+ tomorrow_accuracy = float(tomorrow_recent["correct"].mean()) if "correct" in tomorrow_recent.columns else tomorrow_summary.get("test_accuracy")
1247
+ else:
1248
+ tomorrow_recent = pd.DataFrame()
1249
+ tomorrow_accuracy = tomorrow_summary.get("test_accuracy")
1250
+
1251
+ model_metrics = [
1252
+ {
1253
+ "id": "tomorrow",
1254
+ "label": "Tomorrow",
1255
+ "model_name": tomorrow_summary.get("model_name", "nifty_tomorrow_direction_model"),
1256
+ "source_model": tomorrow_summary.get("source_model", "tuned_daily_forest_single"),
1257
+ "validation_accuracy": tomorrow_summary.get("validation_accuracy"),
1258
+ "test_accuracy": tomorrow_summary.get("test_accuracy"),
1259
+ "recent_accuracy": tomorrow_accuracy,
1260
+ "test_rows": int(tomorrow_summary.get("n_test") or len(tomorrow_test) or 0),
1261
+ },
1262
+ {
1263
+ "id": "tplus1",
1264
+ "label": "T+1",
1265
+ "model_name": tplus1_summary.get("model_name", "nifty_1420_tplus1_logistic_model"),
1266
+ "source_model": "14:00-14:20 logistic forecaster",
1267
+ "validation_accuracy": tplus1_summary.get("validation_accuracy"),
1268
+ "test_accuracy": tplus1_summary.get("test_accuracy"),
1269
+ "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"),
1270
+ "test_rows": int(tplus1_summary.get("test_rows") or len(tplus1_test) or 0),
1271
+ },
1272
+ {
1273
+ "id": "t5",
1274
+ "label": "T+5",
1275
+ "model_name": summary.get("model_name", "nifty_opening_direction_model"),
1276
+ "source_model": summary.get("model_name", "nifty_opening_direction_model"),
1277
+ "validation_accuracy": summary.get("validation_accuracy"),
1278
+ "test_accuracy": summary.get("test_accuracy"),
1279
+ "recent_accuracy": recent_accuracy,
1280
+ "test_rows": int(len(t5_test)) if not t5_test.empty else int(summary.get("test_rows") or 0),
1281
+ },
1282
+ ]
1283
  metrics = {
1284
+ "validation_accuracy": tomorrow_summary.get("validation_accuracy"),
1285
+ "test_accuracy": tomorrow_summary.get("test_accuracy"),
1286
+ "baseline_test_accuracy": tomorrow_summary.get("baseline_accuracy"),
1287
+ "validation_auc": summary.get("validation_auc"),
1288
+ "test_auc": summary.get("test_auc"),
1289
+ "test_brier": summary.get("test_brier"),
1290
+ "feature_count": tomorrow_summary.get("feature_count"),
1291
+ "recent_accuracy": tomorrow_accuracy,
1292
+ "recent_accuracy_days": int(len(tomorrow_recent)) if not tomorrow_recent.empty else 0,
1293
+ "total_test_days": int(tomorrow_summary.get("n_test") or len(tomorrow_test) or 0),
1294
+ "models": model_metrics,
1295
  }
1296
+ track_record = build_prediction_track_record(
1297
+ daily,
1298
+ t5_test,
1299
+ tomorrow_test,
1300
+ tplus1_test,
1301
+ t5_latest,
1302
+ tomorrow_latest,
1303
+ tplus1_latest,
1304
+ )
1305
  return {
1306
+ "latest": t5_latest,
1307
+ "tomorrow_latest": tomorrow_latest,
1308
+ "tplus1_latest": tplus1_latest,
1309
+ "live_accuracy": load_live_accuracy(),
1310
+ "metrics": metrics,
1311
+ "summary": summary,
1312
+ "tomorrow_summary": tomorrow_summary,
1313
+ "tplus1_summary": tplus1_summary,
1314
+ "candidates": load_candidate_results(),
1315
+ "charts": {
1316
+ "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]),
1317
+ "opening_features": _json_ready_frame(opening),
1318
+ "monthly_accuracy": _json_ready_frame(monthly),
1319
+ "direction_mix": _json_ready_frame(direction_mix),
1320
  "recent_predictions": _json_ready_frame(recent_predictions),
1321
+ "t5_recent_predictions": _json_ready_frame(recent_predictions),
1322
+ "tomorrow_recent_predictions": _json_ready_frame(tomorrow_recent),
1323
+ "tplus1_recent_predictions": _json_ready_frame(tplus1_test.tail(40)),
1324
+ "track_record": track_record,
1325
  },
1326
+ "data_status": {
1327
+ "nifty_1m_rows": int(len(pd.read_parquet(NIFTY_1M_PATH, columns=["date"]))),
1328
+ "nifty_1d_rows": int(len(pd.read_parquet(NIFTY_1D_PATH, columns=["date"]))),
1329
+ "training_rows": int(len(dataset)),
1330
+ "test_prediction_rows": int(len(t5_test)),
1331
+ "tomorrow_test_prediction_rows": int(len(tomorrow_test)),
1332
+ "tplus1_test_prediction_rows": int(len(tplus1_test)),
1333
+ "latest_daily_date": pd.to_datetime(daily["date"]).max().date().isoformat(),
1334
+ "refresh_phase": refresh_state.get("phase", REFRESH_NORMAL),
1335
+ "refresh_state": refresh_state,
1336
+ },
1337
+ }
1338
+
1339
+
1340
+ def refresh_first5_prediction(session_date: date | None = None, minutes: pd.DataFrame | None = None) -> Prediction:
1341
+ if session_date is None:
1342
+ today = datetime.now(IST).date()
1343
+ if not is_trading_day(today):
1344
+ raise RuntimeError(f"{today.isoformat()} is not an NSE trading session.")
1345
+ minutes = fetch_yahoo_minutes(period="7d") if minutes is None else minutes
1346
+ append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
1347
+ first5 = first5_features_from_minutes(minutes, session_date=session_date)
1348
+ row = build_model_row(first5)
1349
+ dataset = read_training_dataset()
1350
+ merged = pd.concat([dataset, row], ignore_index=True)
1351
+ merged = merged.drop_duplicates(subset=["date"], keep="last").sort_values("date").reset_index(drop=True)
1352
+ merged.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd")
1353
+ prediction = predict_row(row)
1354
+ clear_dashboard_payload_cache()
1355
+ return prediction
1356
+
1357
+
1358
+ def refresh_daily_data() -> dict[str, Any]:
1359
+ daily = fetch_yahoo_daily(period="1mo")
1360
+ combined = append_parquet_rows(NIFTY_1D_PATH, daily, ["date"])
1361
+ clear_dashboard_payload_cache()
1362
+ return {
1363
+ "rows": int(len(combined)),
1364
+ "latest_date": pd.to_datetime(combined["date"]).max().date().isoformat(),
1365
+ "path": str(NIFTY_1D_PATH),
1366
+ }
1367
+
1368
+
1369
+ def update_opening_outcomes_from_daily() -> dict[str, Any]:
1370
+ if not OPENING_DATASET_PATH.exists() or not NIFTY_1D_PATH.exists():
1371
+ return {"updated_rows": 0, "latest_date": None}
1372
+ dataset = pd.read_parquet(OPENING_DATASET_PATH)
1373
+ daily = pd.read_parquet(NIFTY_1D_PATH)
1374
+ if dataset.empty or daily.empty:
1375
+ return {"updated_rows": 0, "latest_date": None}
1376
+
1377
+ dataset = dataset.copy()
1378
+ dataset["_session_date"] = pd.to_datetime(dataset["date"], errors="coerce").dt.normalize()
1379
+ daily = daily.copy()
1380
+ daily["_session_date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize()
1381
+ daily = daily.dropna(subset=["_session_date"]).drop_duplicates("_session_date", keep="last")
1382
+ daily = daily.set_index("_session_date")
1383
+
1384
+ updated = 0
1385
+ for idx, session_day in dataset["_session_date"].dropna().items():
1386
+ if session_day not in daily.index:
1387
+ continue
1388
+ row = daily.loc[session_day]
1389
+ for src, dst in (
1390
+ ("open", "day_open"),
1391
+ ("high", "day_high"),
1392
+ ("low", "day_low"),
1393
+ ("close", "day_close"),
1394
+ ("volume", "day_volume"),
1395
+ ):
1396
+ if src in row.index and dst in dataset.columns:
1397
+ dataset.at[idx, dst] = row[src]
1398
+ if {"day_open", "day_close", "target", "day_return"}.issubset(dataset.columns):
1399
+ day_open = dataset.at[idx, "day_open"]
1400
+ day_close = dataset.at[idx, "day_close"]
1401
+ if pd.notna(day_open) and pd.notna(day_close) and float(day_open) != 0.0:
1402
+ dataset.at[idx, "target"] = int(float(day_close) > float(day_open))
1403
+ dataset.at[idx, "day_return"] = (float(day_close) - float(day_open)) / float(day_open)
1404
+ updated += 1
1405
+ if {"first5_close", "day_open", "first5_vs_day_open"}.issubset(dataset.columns):
1406
+ first5_close = dataset.at[idx, "first5_close"]
1407
+ day_open = dataset.at[idx, "day_open"]
1408
+ if pd.notna(first5_close) and pd.notna(day_open) and float(day_open) != 0.0:
1409
+ dataset.at[idx, "first5_vs_day_open"] = (float(first5_close) - float(day_open)) / float(day_open)
1410
+
1411
+ dataset = dataset.drop(columns=["_session_date"])
1412
+ dataset = dataset.sort_values("date").reset_index(drop=True)
1413
+ dataset.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd")
1414
+ clear_dashboard_payload_cache()
1415
+ latest = pd.to_datetime(dataset["date"], errors="coerce").max()
1416
+ return {
1417
+ "updated_rows": int(updated),
1418
+ "latest_date": None if pd.isna(latest) else latest.date().isoformat(),
1419
+ }
1420
+
1421
+
1422
+ def load_live_accuracy() -> dict[str, Any]:
1423
+ """Load the live accuracy ledger from disk."""
1424
+ if LIVE_ACCURACY_PATH.exists():
1425
+ try:
1426
+ return json.loads(LIVE_ACCURACY_PATH.read_text(encoding="utf-8"))
1427
+ except Exception:
1428
+ pass
1429
+ return {
1430
+ "tomorrow": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
1431
+ "t5": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
1432
+ "tplus1": {"entries": [], "accuracy": None, "total": 0, "correct_count": 0},
1433
+ }
1434
+
1435
+
1436
+ def save_live_accuracy(data: dict[str, Any]) -> None:
1437
+ """Persist the live accuracy ledger to disk."""
1438
+ LIVE_ACCURACY_PATH.write_text(json.dumps(data, indent=2), encoding="utf-8")
1439
+
1440
+
1441
+ def update_live_accuracy(session_date: date) -> dict[str, Any]:
1442
+ """Score today's predictions against actual outcomes and update the ledger.
1443
+
1444
+ Must be called AFTER refresh_daily_data() (so today's close is available)
1445
+ but BEFORE refresh_first5_prediction / refresh_tplus1_prediction /
1446
+ refresh_tomorrow_prediction (so the CSV files still hold the predictions
1447
+ we want to score).
1448
+ """
1449
+ ledger = load_live_accuracy()
1450
+ daily = pd.read_parquet(NIFTY_1D_PATH)
1451
+ daily["_date"] = pd.to_datetime(daily["date"], errors="coerce").dt.normalize()
1452
+ today_rows = daily[daily["_date"].dt.date == session_date]
1453
+ if today_rows.empty:
1454
+ return ledger
1455
+
1456
+ day_open = float(today_rows.iloc[-1]["open"])
1457
+ day_close = float(today_rows.iloc[-1]["close"])
1458
+ if not (np.isfinite(day_open) and np.isfinite(day_close) and day_open != 0):
1459
+ return ledger
1460
+ actual_close_gt_open = "UP" if day_close > day_open else "DOWN"
1461
+ session_iso = session_date.isoformat()
1462
+
1463
+ # --- T+5: today's 9:20 AM prediction vs close > open ---
1464
+ logged_t5 = {e["date"] for e in ledger["t5"]["entries"]}
1465
+ if session_iso not in logged_t5 and LATEST_PATH.exists():
1466
+ try:
1467
+ t5_row = pd.read_csv(LATEST_PATH).iloc[-1].to_dict()
1468
+ if str(t5_row.get("input_date", ""))[:10] == session_iso:
1469
+ pred = str(t5_row.get("prediction", "")).upper()
1470
+ if pred in ("UP", "DOWN"):
1471
+ ledger["t5"]["entries"].append({
1472
+ "date": session_iso,
1473
+ "prediction": pred,
1474
+ "actual": actual_close_gt_open,
1475
+ "correct": pred == actual_close_gt_open,
1476
+ })
1477
+ except Exception:
1478
+ pass
1479
+
1480
+ # --- Tomorrow: yesterday's prediction targeting today vs close > open ---
1481
+ logged_tom = {e["date"] for e in ledger["tomorrow"]["entries"]}
1482
+ if session_iso not in logged_tom and TOMORROW_LATEST_PATH.exists():
1483
+ try:
1484
+ tom_row = pd.read_csv(TOMORROW_LATEST_PATH).iloc[-1].to_dict()
1485
+ if str(tom_row.get("target_date", ""))[:10] == session_iso:
1486
+ pred = str(tom_row.get("prediction", "")).upper()
1487
+ if pred in ("UP", "DOWN"):
1488
+ ledger["tomorrow"]["entries"].append({
1489
+ "date": session_iso,
1490
+ "prediction": pred,
1491
+ "actual": actual_close_gt_open,
1492
+ "correct": pred == actual_close_gt_open,
1493
+ })
1494
+ except Exception:
1495
+ pass
1496
+
1497
+ # --- T+1: yesterday's 14:20 prediction targeting today ---
1498
+ # T+1 target: today's close > yesterday's 14:20 close
1499
+ logged_t1 = {e["date"] for e in ledger["tplus1"]["entries"]}
1500
+ if session_iso not in logged_t1 and TPLUS1_LATEST_PATH.exists():
1501
+ try:
1502
+ t1_row = pd.read_csv(TPLUS1_LATEST_PATH).iloc[-1].to_dict()
1503
+ if str(t1_row.get("target_date", ""))[:10] == session_iso:
1504
+ pred = str(t1_row.get("prediction", "")).upper()
1505
+ input_date_str = str(t1_row.get("input_date", ""))[:10]
1506
+ input_day = date.fromisoformat(input_date_str)
1507
+ # Read the 14:20 close from minute data for the input session
1508
+ minute = pd.read_parquet(NIFTY_1M_PATH, columns=["date", "close"])
1509
+ minute["dt"] = pd.to_datetime(minute["date"], errors="coerce")
1510
+ minute = minute.dropna(subset=["dt"])
1511
+ minute["session_date"] = minute["dt"].dt.normalize()
1512
+ minute["time_str"] = minute["dt"].dt.strftime("%H:%M")
1513
+ window = minute[
1514
+ (minute["session_date"].dt.date == input_day)
1515
+ & (minute["time_str"] >= "14:00")
1516
+ & (minute["time_str"] <= "14:20")
1517
+ ].sort_values("dt")
1518
+ if not window.empty and pred in ("UP", "DOWN"):
1519
+ w_close = float(window.iloc[-1]["close"])
1520
+ t1_actual = "UP" if day_close > w_close else "DOWN"
1521
+ ledger["tplus1"]["entries"].append({
1522
+ "date": session_iso,
1523
+ "prediction": pred,
1524
+ "actual": t1_actual,
1525
+ "correct": pred == t1_actual,
1526
+ })
1527
+ except Exception:
1528
+ pass
1529
+
1530
+ # Backtest baseline stats
1531
+ t5_summary = load_model_summary()
1532
+ tom_summary = load_tomorrow_summary()
1533
+ t1_summary = load_tplus1_summary()
1534
+
1535
+ t5_test_total = int(t5_summary.get("test_rows") or len(load_test_predictions()) or 0)
1536
+ t5_test_correct = int(round(t5_test_total * float(t5_summary.get("test_accuracy", 0.0))))
1537
+
1538
+ tom_test_total = int(tom_summary.get("n_test") or len(load_tomorrow_test_predictions()) or 0)
1539
+ tom_test_correct = int(round(tom_test_total * float(tom_summary.get("test_accuracy", 0.0))))
1540
+
1541
+ t1_test_total = int(t1_summary.get("test_rows") or len(load_tplus1_test_predictions()) or 0)
1542
+ t1_test_correct = int(round(t1_test_total * float(t1_summary.get("test_accuracy", 0.0))))
1543
+
1544
+ baselines = {
1545
+ "t5": {"total": t5_test_total, "correct": t5_test_correct},
1546
+ "tomorrow": {"total": tom_test_total, "correct": tom_test_correct},
1547
+ "tplus1": {"total": t1_test_total, "correct": t1_test_correct},
1548
  }
1549
 
1550
+ # Recompute summary stats
1551
+ for model_id in ("t5", "tomorrow", "tplus1"):
1552
+ if model_id not in ledger:
1553
+ ledger[model_id] = {"entries": []}
1554
+ if "entries" not in ledger[model_id]:
1555
+ ledger[model_id]["entries"] = []
1556
 
1557
+ entries = ledger[model_id]["entries"]
1558
+ live_total = len(entries)
1559
+ live_correct = sum(1 for e in entries if e.get("correct"))
1560
+
1561
+ base_total = baselines[model_id]["total"]
1562
+ base_correct = baselines[model_id]["correct"]
1563
+
1564
+ combined_total = base_total + live_total
1565
+ combined_correct = base_correct + live_correct
1566
+
1567
+ ledger[model_id]["live_total"] = live_total
1568
+ ledger[model_id]["live_correct_count"] = live_correct
1569
+ ledger[model_id]["live_accuracy"] = live_correct / live_total if live_total > 0 else None
1570
+
1571
+ ledger[model_id]["total"] = combined_total
1572
+ ledger[model_id]["correct_count"] = combined_correct
1573
+ ledger[model_id]["accuracy"] = combined_correct / combined_total if combined_total > 0 else None
1574
 
1575
+ save_live_accuracy(ledger)
1576
+ clear_dashboard_payload_cache()
1577
+ return ledger
1578
 
1579
+
1580
+
1581
+ def refresh_market_close_data(session_date: date | None = None) -> dict[str, Any]:
1582
+ now = datetime.now(IST)
1583
+ session_date = session_date or now.date()
1584
+ if not is_trading_day(session_date):
1585
+ raise RuntimeError(f"{session_date.isoformat()} is not an NSE trading session.")
1586
+ save_refresh_state(REFRESH_WAITING, session_date=session_date)
1587
+ try:
1588
+ save_refresh_state(REFRESH_REFRESHING, session_date=session_date)
1589
+ minutes = fetch_yahoo_minutes(period="7d")
1590
+ minute_frame = append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
1591
+ daily_info = refresh_daily_data()
1592
+ # Score live predictions BEFORE they get overwritten by fresh ones
1593
+ try:
1594
+ update_live_accuracy(session_date)
1595
+ except Exception as exc:
1596
+ print(f"[close-refresh] live accuracy update failed: {exc}", flush=True)
1597
+ t5_prediction = refresh_first5_prediction(session_date=session_date, minutes=minutes)
1598
+ tplus1_prediction = refresh_tplus1_prediction(session_date=session_date)
1599
+ outcomes = update_opening_outcomes_from_daily()
1600
+ tomorrow_prediction = refresh_tomorrow_prediction(session_date=session_date)
1601
+ state = save_refresh_state(REFRESH_READY, session_date=session_date)
1602
+ clear_dashboard_payload_cache()
1603
+ return {
1604
+ "session_date": session_date.isoformat(),
1605
+ "nifty_1m_rows": int(len(minute_frame)),
1606
+ "latest_minute": pd.to_datetime(minute_frame["date"], errors="coerce").max().isoformat(),
1607
+ "daily": daily_info,
1608
+ "opening_dataset": outcomes,
1609
+ "t5_prediction": t5_prediction.to_dict(),
1610
+ "tplus1_prediction": tplus1_prediction,
1611
+ "tomorrow_prediction": tomorrow_prediction,
1612
+ "refresh_state": state,
1613
+ }
1614
+ except Exception as exc:
1615
+ save_refresh_state(REFRESH_FAILED, session_date=session_date, error=str(exc))
1616
+ clear_dashboard_payload_cache()
1617
+ raise
1618
+
1619
+
1620
+ def close_refresh_due(now: datetime | None = None) -> bool:
1621
+ now = now or datetime.now(IST)
1622
+ if not is_trading_day(now.date()) or now.time() < CLOSE_REFRESH_READY:
1623
+ return False
1624
+ latest_daily = latest_parquet_date(NIFTY_1D_PATH)
1625
+ latest_minutes = latest_parquet_date(NIFTY_1M_PATH)
1626
+ latest_opening = latest_parquet_date(OPENING_DATASET_PATH)
1627
+ latest_opening_outcome = latest_opening_outcome_date()
1628
+ tomorrow_latest = latest_tomorrow_prediction()
1629
+ tomorrow_input = None
1630
+ try:
1631
+ if tomorrow_latest.get("input_date"):
1632
+ tomorrow_input = date.fromisoformat(str(tomorrow_latest.get("input_date"))[:10])
1633
+ except Exception:
1634
+ tomorrow_input = None
1635
+ return any(
1636
+ latest != now.date()
1637
+ for latest in (latest_daily, latest_minutes, latest_opening, latest_opening_outcome, tomorrow_input)
1638
+ )
1639
+
1640
+
1641
+ def latest_prediction_input_date(path: Path) -> date | None:
1642
+ if not path.exists():
1643
+ return None
1644
+ try:
1645
+ frame = pd.read_csv(path, usecols=["input_date"])
1646
+ except Exception:
1647
+ return None
1648
+ if frame.empty:
1649
+ return None
1650
+ value = pd.to_datetime(frame["input_date"], errors="coerce").max()
1651
+ return None if pd.isna(value) else value.date()
1652
+
1653
+
1654
+ def latest_tomorrow_input_date() -> date | None:
1655
+ try:
1656
+ latest = latest_tomorrow_prediction()
1657
+ raw = latest.get("input_date")
1658
+ return date.fromisoformat(str(raw)[:10]) if raw else None
1659
+ except Exception:
1660
+ return None
1661
+
1662
+
1663
+ def expected_completed_daily_date(now: datetime | None = None) -> date:
1664
+ now = now or datetime.now(IST)
1665
+ if is_trading_day(now.date()) and now.time() < CLOSE_REFRESH_READY:
1666
+ return previous_trading_day(now.date() - timedelta(days=1))
1667
+ return previous_trading_day(now.date())
1668
+
1669
+
1670
+ def expected_minute_date(now: datetime | None = None) -> date:
1671
+ now = now or datetime.now(IST)
1672
+ if is_trading_day(now.date()) and now.time() >= FIRST5_READY:
1673
+ return now.date()
1674
+ return previous_trading_day(now.date() - timedelta(days=1))
1675
+
1676
+
1677
+ def expected_tplus1_date(now: datetime | None = None) -> date:
1678
+ now = now or datetime.now(IST)
1679
+ if is_trading_day(now.date()) and now.time() >= TPLUS1_READY:
1680
+ return now.date()
1681
+ return previous_trading_day(now.date() - timedelta(days=1))
1682
+
1683
+
1684
+ def is_stale(latest: date | None, expected: date) -> bool:
1685
+ return latest is None or latest < expected
1686
+
1687
+
1688
+ def stale_data_status(now: datetime | None = None) -> dict[str, Any]:
1689
+ now = now or datetime.now(IST)
1690
+ expected_daily = expected_completed_daily_date(now)
1691
+ expected_minutes = expected_minute_date(now)
1692
+ expected_tplus1 = expected_tplus1_date(now)
1693
+ latest_daily = latest_parquet_date(NIFTY_1D_PATH)
1694
+ latest_minutes = latest_parquet_date(NIFTY_1M_PATH)
1695
+ latest_t5 = latest_prediction_input_date(LATEST_PATH)
1696
+ latest_tomorrow = latest_tomorrow_input_date()
1697
+ latest_tplus1 = latest_prediction_input_date(TPLUS1_LATEST_PATH)
1698
  return {
1699
+ "server_time_ist": now.isoformat(),
1700
+ "expected_daily_date": expected_daily.isoformat(),
1701
+ "expected_minute_date": expected_minutes.isoformat(),
1702
+ "expected_tplus1_date": expected_tplus1.isoformat(),
1703
+ "latest_daily_date": latest_daily.isoformat() if latest_daily else None,
1704
+ "latest_minute_date": latest_minutes.isoformat() if latest_minutes else None,
1705
+ "latest_t5_date": latest_t5.isoformat() if latest_t5 else None,
1706
+ "latest_tomorrow_date": latest_tomorrow.isoformat() if latest_tomorrow else None,
1707
+ "latest_tplus1_date": latest_tplus1.isoformat() if latest_tplus1 else None,
1708
+ "daily_stale": is_stale(latest_daily, expected_daily),
1709
+ "minutes_stale": is_stale(latest_minutes, expected_minutes),
1710
+ "t5_stale": is_stale(latest_t5, expected_minutes),
1711
+ "tomorrow_stale": is_stale(latest_tomorrow, expected_daily),
1712
+ "tplus1_stale": is_stale(latest_tplus1, expected_tplus1),
1713
  }
1714
 
1715
 
1716
+ def refresh_stale_data_once(now: datetime | None = None) -> dict[str, Any]:
1717
+ now = now or datetime.now(IST)
1718
+ status = stale_data_status(now)
1719
+ if not any(status[key] for key in ("daily_stale", "minutes_stale", "t5_stale", "tomorrow_stale", "tplus1_stale")):
1720
+ return {"status": "fresh", **status, "actions": []}
1721
+ if not _stale_refresh_lock.acquire(blocking=False):
1722
+ return {"status": "skipped", "reason": "stale refresh already running", **status, "actions": []}
1723
+
1724
+ actions: list[dict[str, Any]] = []
1725
+ try:
1726
+ if status["minutes_stale"]:
1727
+ minutes = fetch_yahoo_minutes(period="7d")
1728
+ combined = append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"])
1729
+ actions.append(
1730
+ {
1731
+ "name": "minutes",
1732
+ "rows": int(len(combined)),
1733
+ "latest_date": pd.to_datetime(combined["date"], errors="coerce").max().date().isoformat(),
1734
+ }
1735
+ )
1736
+
1737
+ if status["daily_stale"]:
1738
+ daily_info = refresh_daily_data()
1739
+ outcomes = update_opening_outcomes_from_daily()
1740
+ actions.append({"name": "daily", **daily_info})
1741
+ actions.append({"name": "opening_outcomes", **outcomes})
1742
+
1743
+ if status["daily_stale"] or status["tomorrow_stale"]:
1744
+ try:
1745
+ tomorrow = refresh_tomorrow_prediction(session_date=date.fromisoformat(status["expected_daily_date"]))
1746
+ actions.append({"name": "tomorrow_prediction", "input_date": tomorrow.get("input_date")})
1747
+ except Exception as exc:
1748
+ actions.append({"name": "tomorrow_prediction", "error": str(exc)})
1749
+
1750
+ if status["t5_stale"] and is_trading_day(now.date()) and now.time() >= FIRST5_READY:
1751
+ prediction = refresh_first5_prediction(session_date=now.date())
1752
+ actions.append({"name": "t5_prediction", "input_date": prediction.input_date})
1753
+
1754
+ if status["tplus1_stale"] and is_trading_day(now.date()) and now.time() >= TPLUS1_READY:
1755
+ prediction = refresh_tplus1_prediction(session_date=now.date())
1756
+ actions.append({"name": "tplus1_prediction", "input_date": prediction.get("input_date")})
1757
+
1758
+ clear_dashboard_payload_cache()
1759
+ refreshed_status = stale_data_status(datetime.now(IST))
1760
+ return {"status": "refreshed", **refreshed_status, "actions": actions}
1761
+ finally:
1762
+ _stale_refresh_lock.release()
1763
+
1764
+
1765
+ def next_ist_run_at(run_time: time = time(9, 20), now: datetime | None = None) -> datetime:
1766
+ now = now or datetime.now(IST)
1767
+ target_day = now.date()
1768
+ if now >= datetime.combine(target_day, run_time, tzinfo=IST):
1769
+ target_day += timedelta(days=1)
1770
+ target_day = next_trading_day(target_day)
1771
+ return datetime.combine(target_day, run_time, tzinfo=IST)
1772
+
1773
+
1774
+ def seconds_until_next_ist_run(run_time: time = time(9, 20)) -> float:
1775
+ now = datetime.now(IST)
1776
+ target = next_ist_run_at(run_time, now=now)
1777
+ return max(1.0, (target - now).total_seconds())