from __future__ import annotations import json import sys from dataclasses import dataclass from datetime import date, datetime, time, timedelta from pathlib import Path from typing import Any from zoneinfo import ZoneInfo import joblib import numpy as np import pandas as pd import yfinance as yf IST = ZoneInfo("Asia/Kolkata") YAHOO_NIFTY_SYMBOL = "^NSEI" BACKEND_ROOT = Path(__file__).resolve().parents[1] DATA_DIR = BACKEND_ROOT / "data" MODEL_DIR = BACKEND_ROOT / "models" OPENING_DATASET_PATH = DATA_DIR / "opening_direction_training_dataset.parquet" NIFTY_1M_PATH = DATA_DIR / "nifty50_1m.parquet" NIFTY_1D_PATH = DATA_DIR / "nifty50_1d.parquet" MODEL_PATH = MODEL_DIR / "nifty_opening_direction_model.joblib" LATEST_PATH = MODEL_DIR / "latest_prediction.csv" TEST_PREDICTIONS_PATH = DATA_DIR / "test_predictions.parquet" DECISION_OVERLAYS = [ { "name": "fifth_minute_momentum_flip", "feature": "m5_ret_1m", "op": ">=", "value": 0.0005085411885759201, }, { "name": "vix_stretch_flip", "feature": "india_vix_close_vs_sma_20", "op": ">=", "value": 0.24641908937959742, }, ] class ProbabilityBlend: def __init__(self, models: list[Any], weights: np.ndarray): self.models = models self.weights = np.asarray(weights, dtype="float64") self.weights = self.weights / self.weights.sum() def predict_proba(self, x: pd.DataFrame) -> np.ndarray: probs = np.column_stack([predict_proba_up(model, x) for model in self.models]) prob_up = probs @ self.weights return np.column_stack([1.0 - prob_up, prob_up]) @dataclass(frozen=True) class Prediction: input_date: str first5_start: str first5_end: str prediction: str prob_up: float confidence: float threshold: float model_name: str def to_dict(self) -> dict[str, Any]: return { "input_date": self.input_date, "first5_start": self.first5_start, "first5_end": self.first5_end, "prediction": self.prediction, "prob_up": self.prob_up, "confidence": self.confidence, "threshold": self.threshold, "model_name": self.model_name, } def predict_proba_up(model: Any, x: pd.DataFrame) -> np.ndarray: return np.asarray(model.predict_proba(x)[:, 1], dtype="float64") def safe_div(numer: pd.Series | np.ndarray, denom: pd.Series | np.ndarray) -> pd.Series: n = pd.Series(numer, copy=False) d = pd.Series(denom, copy=False) out = pd.Series(np.nan, index=n.index, dtype="float64") mask = d.notna() & np.isfinite(d.to_numpy(dtype="float64")) & (d != 0) out.loc[mask] = n.loc[mask].to_numpy(dtype="float64") / d.loc[mask].to_numpy(dtype="float64") return out def load_model() -> dict[str, Any]: # Existing artifact was trained as a script, so its custom blend class # resolves through __main__ when unpickled. sys.modules["__main__"].ProbabilityBlend = ProbabilityBlend sys.modules["__main__"].predict_proba_up = predict_proba_up payload = joblib.load(MODEL_PATH) payload.setdefault("decision_overlays", DECISION_OVERLAYS) payload.setdefault("model_name", "nifty_opening_direction_model") return payload def overlay_mask(frame: pd.DataFrame, overlay: dict[str, object]) -> np.ndarray: feature = str(overlay["feature"]) if feature not in frame.columns: return np.zeros(len(frame), dtype=bool) series = pd.to_numeric(frame[feature], errors="coerce") value = float(overlay["value"]) if overlay["op"] == ">=": return (series >= value).fillna(False).to_numpy(dtype=bool) if overlay["op"] == "<=": return (series <= value).fillna(False).to_numpy(dtype=bool) raise ValueError(f"Unsupported overlay op: {overlay['op']}") def apply_decision_overlays(pred: np.ndarray, frame: pd.DataFrame, overlays: list[dict[str, object]]) -> np.ndarray: adjusted = np.asarray(pred, dtype="int64").copy() for overlay in overlays: mask = overlay_mask(frame, overlay) adjusted[mask] = 1 - adjusted[mask] return adjusted def directional_confidence(prob_up: np.ndarray, pred: np.ndarray, threshold: float) -> np.ndarray: prob_up = np.asarray(prob_up, dtype="float64") pred = np.asarray(pred, dtype="int64") base_side_prob = np.where(pred == 1, prob_up, 1.0 - prob_up) threshold_distance = np.abs(prob_up - float(threshold)) return np.clip(0.50 + threshold_distance, base_side_prob, 0.99) def read_training_dataset() -> pd.DataFrame: df = pd.read_parquet(OPENING_DATASET_PATH) for col in ("date", "first5_start", "first5_end"): if col in df.columns: df[col] = pd.to_datetime(df[col], errors="coerce") return df.sort_values("date").reset_index(drop=True) def normalize_yahoo_frame(df: pd.DataFrame) -> pd.DataFrame: if df.empty: return pd.DataFrame(columns=["date", "open", "high", "low", "close", "volume"]) if isinstance(df.columns, pd.MultiIndex): df.columns = [str(c[0]).lower() for c in df.columns] else: df.columns = [str(c).lower().replace(" ", "_") for c in df.columns] df = df.reset_index() date_col = next((c for c in df.columns if c.lower() in {"datetime", "date"}), df.columns[0]) df["date"] = pd.to_datetime(df[date_col], errors="coerce") if df["date"].dt.tz is None: df["date"] = df["date"].dt.tz_localize("UTC").dt.tz_convert(IST) else: df["date"] = df["date"].dt.tz_convert(IST) rename = { "open": "open", "high": "high", "low": "low", "close": "close", "adj_close": "close", "volume": "volume", } out = pd.DataFrame({"date": df["date"].dt.tz_localize(None)}) for src, dst in rename.items(): if src in df.columns and dst not in out.columns: out[dst] = pd.to_numeric(df[src], errors="coerce") return out.dropna(subset=["date", "open", "high", "low", "close"]).sort_values("date") def fetch_yahoo_minutes(period: str = "5d") -> pd.DataFrame: raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1m", progress=False, prepost=False, auto_adjust=False) return normalize_yahoo_frame(raw) def fetch_yahoo_daily(period: str = "1mo") -> pd.DataFrame: raw = yf.download(YAHOO_NIFTY_SYMBOL, period=period, interval="1d", progress=False, prepost=False, auto_adjust=False) out = normalize_yahoo_frame(raw) out["date"] = pd.to_datetime(out["date"], errors="coerce").dt.normalize() return out.drop_duplicates("date", keep="last") def append_parquet_rows(path: Path, new_rows: pd.DataFrame, subset: list[str]) -> pd.DataFrame: if path.exists(): existing = pd.read_parquet(path) combined = pd.concat([existing, new_rows], ignore_index=True) else: combined = new_rows.copy() combined = combined.drop_duplicates(subset=subset, keep="last").sort_values(subset).reset_index(drop=True) combined.to_parquet(path, index=False, compression="zstd") return combined def first5_features_from_minutes(minutes: pd.DataFrame, session_date: date | None = None) -> pd.DataFrame: if minutes.empty: raise RuntimeError("Yahoo returned no minute bars.") bars = minutes.copy() bars["dt"] = pd.to_datetime(bars["date"], errors="coerce") bars["session_date"] = bars["dt"].dt.normalize() if session_date is None: session_ts = bars["session_date"].max() else: session_ts = pd.Timestamp(session_date).normalize() day = bars[bars["session_date"] == session_ts].sort_values("dt").copy() start_dt = pd.Timestamp.combine(session_ts.date(), time(9, 15)) end_dt = pd.Timestamp.combine(session_ts.date(), time(9, 19)) first5 = day[(day["dt"] >= start_dt) & (day["dt"] <= end_dt)].head(5).copy() if len(first5) < 5: raise RuntimeError(f"Need 5 opening bars for {session_ts.date()}, got {len(first5)}.") first5["minute_index"] = np.arange(len(first5)) first5["ret_1m"] = first5["close"].pct_change(fill_method=None) first5["range_pct_1m"] = safe_div(first5["high"] - first5["low"], first5["open"]) first5["body_pct_1m"] = safe_div(first5["close"] - first5["open"], first5["open"]) row = { "date": session_ts, "first5_start": first5["dt"].iloc[0], "first5_end": first5["dt"].iloc[-1], "first5_open": first5["open"].iloc[0], "first5_high": first5["high"].max(), "first5_low": first5["low"].min(), "first5_close": first5["close"].iloc[-1], "first5_volume": first5["volume"].sum() if "volume" in first5 else 0.0, "first5_bars": len(first5), "first5_last_1m_ret": first5["ret_1m"].iloc[-1], "first5_ret_std": first5["ret_1m"].std(), } row["first5_return"] = (row["first5_close"] - row["first5_open"]) / row["first5_open"] row["first5_range_pct"] = (row["first5_high"] - row["first5_low"]) / row["first5_open"] first5_range = row["first5_high"] - row["first5_low"] row["first5_body_to_range"] = (row["first5_close"] - row["first5_open"]) / first5_range if first5_range else np.nan row["first5_close_location"] = (row["first5_close"] - row["first5_low"]) / first5_range if first5_range else np.nan for idx, (_, candle) in enumerate(first5.iterrows(), start=1): for field in ("open", "high", "low", "close", "ret_1m", "range_pct_1m", "body_pct_1m"): row[f"m{idx}_{field}"] = candle[field] row[f"m{idx}_close_vs_first5_open"] = (candle["close"] - row["first5_open"]) / row["first5_open"] row[f"m{idx}_range_share"] = (candle["high"] - candle["low"]) / first5_range if first5_range else np.nan row["first5_return_accel"] = row["m5_ret_1m"] - row["m2_ret_1m"] row["first5_last2_return"] = (row["m5_close"] - row["m4_open"]) / row["m4_open"] row["first5_first2_return"] = (row["m2_close"] - row["m1_open"]) / row["m1_open"] row["first5_reversal"] = np.sign(row["first5_first2_return"]) * -np.sign(row["first5_last2_return"]) row["dow"] = session_ts.dayofweek row["dom"] = session_ts.day row["month"] = session_ts.month return pd.DataFrame([row]) def build_model_row(first5_row: pd.DataFrame) -> pd.DataFrame: dataset = read_training_dataset() latest_context = dataset.iloc[[-1]].copy() output = latest_context.copy() for col in first5_row.columns: output[col] = first5_row[col].iloc[0] if {"first5_open", "nifty_close"}.issubset(output.columns): output["first5_gap_from_prev_close"] = (output["first5_open"] - output["nifty_close"]) / output["nifty_close"] output["first5_close_vs_prev_close"] = (output["first5_close"] - output["nifty_close"]) / output["nifty_close"] if {"first5_range_pct", "nifty_range_pct"}.issubset(output.columns): output["first5_range_vs_prev_range"] = output["first5_range_pct"] / output["nifty_range_pct"] if {"first5_return", "nifty_ret_1"}.issubset(output.columns): output["first5_return_x_prev_ret"] = output["first5_return"] * output["nifty_ret_1"] output["gap_x_prev_ret"] = output["first5_gap_from_prev_close"] * output["nifty_ret_1"] if {"first5_return", "banknifty_ret_1"}.issubset(output.columns): output["first5_return_x_bank_ret_1"] = output["first5_return"] * output["banknifty_ret_1"] if {"first5_range_pct", "india_vix_ret_1"}.issubset(output.columns): output["first5_range_x_vix_ret_1"] = output["first5_range_pct"] * output["india_vix_ret_1"] output["target"] = np.nan output["day_return"] = np.nan return output def predict_row(row: pd.DataFrame) -> Prediction: payload = load_model() model = payload["model"] features = payload["features"] threshold = float(payload["threshold"]) missing = [c for c in features if c not in row.columns] if missing: raise RuntimeError(f"Feature row is missing {len(missing)} features; first missing: {missing[:5]}") prob_up = predict_proba_up(model, row[features]) raw_pred = (prob_up >= threshold).astype("int64") pred = apply_decision_overlays(raw_pred, row, payload.get("decision_overlays", DECISION_OVERLAYS)) confidence = directional_confidence(prob_up, pred, threshold) prediction = Prediction( input_date=pd.to_datetime(row["date"].iloc[0]).date().isoformat(), first5_start=str(pd.to_datetime(row["first5_start"].iloc[0])), first5_end=str(pd.to_datetime(row["first5_end"].iloc[0])), prediction="UP" if int(pred[0]) == 1 else "DOWN", prob_up=float(prob_up[0]), confidence=float(confidence[0]), threshold=threshold, model_name=str(payload.get("model_name", "nifty_opening_direction_model")), ) pd.DataFrame([prediction.to_dict()]).to_csv(LATEST_PATH, index=False) return prediction def latest_saved_prediction() -> dict[str, Any]: if LATEST_PATH.exists(): return pd.read_csv(LATEST_PATH).iloc[-1].to_dict() summary_path = MODEL_DIR / "summary.json" if summary_path.exists(): return json.loads(summary_path.read_text(encoding="utf-8")) raise FileNotFoundError("No latest prediction is available yet.") def _json_ready_frame(df: pd.DataFrame, limit: int | None = None) -> list[dict[str, Any]]: out = df.copy() if limit is not None: out = out.tail(limit) for col in out.columns: if pd.api.types.is_datetime64_any_dtype(out[col]): out[col] = out[col].dt.strftime("%Y-%m-%d %H:%M:%S") out = out.replace({np.nan: None}) return out.to_dict(orient="records") def load_model_summary() -> dict[str, Any]: summary_path = MODEL_DIR / "summary.json" if not summary_path.exists(): return {} return json.loads(summary_path.read_text(encoding="utf-8")) def load_candidate_results() -> list[dict[str, Any]]: path = MODEL_DIR / "candidate_results.csv" if not path.exists(): return [] return _json_ready_frame(pd.read_csv(path).head(12)) def load_test_predictions() -> pd.DataFrame: if not TEST_PREDICTIONS_PATH.exists(): return pd.DataFrame() df = pd.read_parquet(TEST_PREDICTIONS_PATH) df["date"] = pd.to_datetime(df["date"], errors="coerce") return df.sort_values("date").reset_index(drop=True) def dashboard_payload() -> dict[str, Any]: summary = load_model_summary() latest = latest_saved_prediction() test = load_test_predictions() daily = pd.read_parquet(NIFTY_1D_PATH) daily["date"] = pd.to_datetime(daily["date"], errors="coerce") daily = daily.sort_values("date").tail(180) dataset = read_training_dataset() opening = dataset[["date", "first5_return", "first5_range_pct", "first5_close_location"]].tail(120).copy() if not test.empty: recent_predictions = test.tail(40).copy() recent_accuracy = float(recent_predictions["correct"].mean()) direction_mix = test.groupby("prediction")["correct"].agg(["count", "mean"]).reset_index() monthly = ( test.assign(month=test["date"].dt.strftime("%Y-%m")) .groupby("month", as_index=False)["correct"] .mean() .rename(columns={"correct": "accuracy"}) ) else: recent_predictions = pd.DataFrame() recent_accuracy = None direction_mix = pd.DataFrame() monthly = pd.DataFrame() metrics = { "validation_accuracy": summary.get("validation_accuracy"), "test_accuracy": summary.get("test_accuracy"), "baseline_test_accuracy": summary.get("baseline_test_accuracy"), "validation_auc": summary.get("validation_auc"), "test_auc": summary.get("test_auc"), "test_brier": summary.get("test_brier"), "feature_count": summary.get("feature_count"), "recent_accuracy": recent_accuracy, "recent_accuracy_days": int(len(recent_predictions)) if not recent_predictions.empty else 0, "total_test_days": int(len(test)) if not test.empty else 0, } return { "latest": latest, "metrics": metrics, "summary": summary, "candidates": load_candidate_results(), "charts": { "daily_close": _json_ready_frame(daily[["date", "open", "high", "low", "close"]]), "opening_features": _json_ready_frame(opening), "monthly_accuracy": _json_ready_frame(monthly), "direction_mix": _json_ready_frame(direction_mix), "recent_predictions": _json_ready_frame(recent_predictions), }, "data_status": { "nifty_1m_rows": int(len(pd.read_parquet(NIFTY_1M_PATH, columns=["date"]))), "nifty_1d_rows": int(len(pd.read_parquet(NIFTY_1D_PATH, columns=["date"]))), "training_rows": int(len(dataset)), "test_prediction_rows": int(len(test)), "latest_daily_date": pd.to_datetime(daily["date"]).max().date().isoformat(), }, } def refresh_first5_prediction(session_date: date | None = None) -> Prediction: minutes = fetch_yahoo_minutes(period="5d") append_parquet_rows(NIFTY_1M_PATH, minutes, ["date"]) first5 = first5_features_from_minutes(minutes, session_date=session_date) row = build_model_row(first5) dataset = read_training_dataset() merged = pd.concat([dataset, row], ignore_index=True) merged = merged.drop_duplicates(subset=["date"], keep="last").sort_values("date").reset_index(drop=True) merged.to_parquet(OPENING_DATASET_PATH, index=False, compression="zstd") return predict_row(row) def refresh_daily_data() -> dict[str, Any]: daily = fetch_yahoo_daily(period="1mo") combined = append_parquet_rows(NIFTY_1D_PATH, daily, ["date"]) return { "rows": int(len(combined)), "latest_date": pd.to_datetime(combined["date"]).max().date().isoformat(), "path": str(NIFTY_1D_PATH), } def seconds_until_next_ist_run(run_time: time = time(9, 20)) -> float: now = datetime.now(IST) target = datetime.combine(now.date(), run_time, tzinfo=IST) if now >= target: target += timedelta(days=1) return max(1.0, (target - now).total_seconds())