runtime.py

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())