models/nifty_forecaster/train.py

from __future__ import annotations

import argparse
import json
import math
import sys
import time
import warnings
from dataclasses import asdict, dataclass
from functools import lru_cache
from pathlib import Path
from typing import Iterable

import numpy as np
import pandas as pd
import os


def find_project_root(start: Path) -> Path:
    env_root = os.environ.get("FORECASTING_PROJECT_ROOT")
    if env_root:
        return Path(env_root)
    
    fallback = start.parents[4] / "forecasting project"
    if (fallback / "Data").is_dir() and (fallback / "Alt Data").is_dir():
        return fallback

    for path in (start, *start.parents):
        if (path / "Data").is_dir() and (path / "Alt Data").is_dir():
            return path
    raise RuntimeError(f"Could not find project root from {start}")


PROJECT_ROOT = find_project_root(Path(__file__).resolve())
DATA_DIR = PROJECT_ROOT / "Data"
ALT_DIR = PROJECT_ROOT / "Alt Data"
PRICE_DIR = DATA_DIR / "processed" / "bars" / "1d"
INTRADAY_DIR = DATA_DIR / "raw" / "minute"
OUTPUT_DIR = Path(__file__).resolve().parent / "outputs"
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
warnings.filterwarnings("ignore", category=pd.errors.PerformanceWarning)
warnings.filterwarnings("ignore", category=FutureWarning)

DEFAULT_TRAIN_END = pd.Timestamp("2023-12-31")
DEFAULT_VALID_END = pd.Timestamp("2025-08-17")
DEFAULT_TEST_END = pd.Timestamp("2099-12-31")
COMMON_VALID_START = pd.Timestamp("2024-07-01")
LOCKED_NIFTY50_WEIGHTS = np.array([0.34099525, 0.49518660, 0.16381815], dtype="float64")
LOCKED_NIFTY50_THRESHOLD = 0.534
NIFTY50_LOW_BANK_VOL_THRESHOLD = 0.004660
NIFTY50_BANK_RET_FLIP_THRESHOLD = 0.01677902301854645
NIFTY50_TINY_RANGE_UP_THRESHOLD = 0.004204134680410373

SUPPORTED_SYMBOLS = ("NIFTY 50", "NIFTY BANK")

DAILY_VALID_WINDOWS: dict[str, tuple[pd.Timestamp, pd.Timestamp]] = {
    "NIFTY 50": (pd.Timestamp("2024-07-01"), pd.Timestamp("2025-08-17")),
    "NIFTY BANK": (pd.Timestamp("2024-07-01"), pd.Timestamp("2025-08-17")),
}

SYMBOL_BENCHMARKS: dict[str, str] = {
    "NIFTY 50": "NIFTY BANK",
    "NIFTY BANK": "NIFTY 50",
}


class ProgressBar:
    """Small dependency-free terminal progress bar with elapsed time, ETA, and rate."""

    def __init__(
        self,
        total: int,
        description: str = "Progress",
        *,
        enabled: bool = True,
        width: int = 34,
        update_every: float = 0.2,
        stream: object | None = None,
    ) -> None:
        self.total = max(0, int(total))
        self.description = description
        self.enabled = enabled
        self.width = max(10, int(width))
        self.update_every = max(0.0, float(update_every))
        self.stream = stream if stream is not None else sys.stderr
        self.current = 0
        self.start_time = time.monotonic()
        self.last_render = 0.0
        self.closed = False
        self._last_line_len = 0

    def __enter__(self) -> "ProgressBar":
        self.start_time = time.monotonic()
        self.last_render = 0.0
        self.render(force=True)
        return self

    def __exit__(self, exc_type: object, exc: object, tb: object) -> None:
        self.close()

    @staticmethod
    def _format_duration(seconds: float | None) -> str:
        if seconds is None or not np.isfinite(seconds) or seconds < 0:
            return "--:--"
        seconds = int(round(seconds))
        hours, rem = divmod(seconds, 3600)
        minutes, secs = divmod(rem, 60)
        if hours:
            return f"{hours:d}:{minutes:02d}:{secs:02d}"
        return f"{minutes:02d}:{secs:02d}"

    def update(self, current: int | None = None, *, description: str | None = None, force: bool = False) -> None:
        if current is not None:
            self.current = max(0, int(current))
            if self.total:
                self.current = min(self.current, self.total)
        if description is not None:
            self.description = description
        self.render(force=force)

    def advance(self, step: int = 1, *, description: str | None = None, force: bool = False) -> None:
        self.update(self.current + int(step), description=description, force=force)

    def render(self, *, force: bool = False) -> None:
        if not self.enabled or self.closed:
            return
        now = time.monotonic()
        if not force and (now - self.last_render) < self.update_every and self.current < self.total:
            return
        self.last_render = now
        elapsed = max(0.0, now - self.start_time)
        if self.total > 0:
            fraction = min(1.0, max(0.0, self.current / self.total))
        else:
            fraction = 1.0
        filled = int(round(self.width * fraction))
        bar = "█" * filled + "░" * (self.width - filled)
        rate = self.current / elapsed if elapsed > 0 else 0.0
        eta = (elapsed / self.current) * (self.total - self.current) if self.current > 0 and self.total > 0 else None
        line = (
            f"\r{self.description} [{bar}] "
            f"{self.current}/{self.total} {fraction * 100:6.2f}% | "
            f"elapsed {self._format_duration(elapsed)} | "
            f"ETA {self._format_duration(eta)} | "
            f"{rate:,.2f}/s"
        )
        padding = " " * max(0, self._last_line_len - len(line))
        print(line + padding, end="", file=self.stream, flush=True)
        self._last_line_len = len(line)

    def close(self) -> None:
        if self.closed:
            return
        self.render(force=True)
        if self.enabled:
            print(file=self.stream, flush=True)
        self.closed = True


def progress_note(message: str, *, enabled: bool = True) -> None:
    if enabled:
        print(f"[progress] {message}", file=sys.stderr, flush=True)


@dataclass(frozen=True)
class ModelSpec:
    name: str
    kind: str
    use_intraday: bool
    feature_profile: str = "all"
    top_k: int | None = None
    l2: float = 0.5
    n_trees: int = 60
    max_depth: int = 5
    min_leaf: int = 30
    seed: int = 7


@dataclass
class FitResult:
    symbol: str
    horizon: str
    horizon_bars: int
    config: dict[str, object]
    threshold: float
    validation_accuracy: float
    test_accuracy: float
    baseline_accuracy: float
    n_train: int
    n_valid: int
    n_test: int
    train_start: str
    train_end: str
    valid_start: str
    valid_end: str
    test_start: str
    test_end: str
    latest_forecast_date: str
    latest_forecast_for: str
    latest_forecast_prob_up: float
    latest_forecast_signal: str
    feature_count: int
    validation_prob_std: float
    test_prob_std: float
    test_prob_min: float
    test_prob_max: float


def price_prefix(symbol: str) -> str:
    return symbol.lower().replace("&", "and").replace(" ", "_")


def benchmark_symbol(symbol: str) -> str:
    return SYMBOL_BENCHMARKS.get(symbol, "NIFTY 50")


def symbol_file_stem(symbol: str) -> str:
    mapping = {
        "NIFTY 50": "nifty50",
        "NIFTY BANK": "banknifty",
        "INDIA VIX": "india_vix",
    }
    if symbol not in mapping:
        raise KeyError(f"Unsupported symbol: {symbol}")
    return mapping[symbol]


def sigmoid(x: np.ndarray) -> np.ndarray:
    return 1.0 / (1.0 + np.exp(-np.clip(x, -40.0, 40.0)))


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 _to_ns_datetime(series: pd.Series) -> pd.Series:
    return pd.to_datetime(series, errors="coerce").astype("datetime64[ns]")


@lru_cache(maxsize=None)
def load_price_frame(symbol: str) -> pd.DataFrame:
    path = PRICE_DIR / f"{symbol_file_stem(symbol)}_1d.csv"
    if not path.exists():
        raise FileNotFoundError(f"Missing daily price file for {symbol}: {path}")
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    if "date" not in df.columns:
        raise ValueError(f"Price frame for {symbol} has no date column")
    df["date"] = _to_ns_datetime(df["date"])
    for col in df.columns:
        if col != "date":
            df[col] = pd.to_numeric(df[col], errors="coerce")
    if "volume" in df.columns:
        volume = pd.to_numeric(df["volume"], errors="coerce")
        volume_available = volume.replace(0, np.nan).notna().sum() >= max(20, int(0.5 * len(volume)))
        df["volume"] = volume.replace(0, np.nan) if volume_available else 0.0
    return df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)


def load_vix() -> pd.DataFrame:
    path = PRICE_DIR / "india_vix_1d.csv"
    if not path.exists():
        return pd.DataFrame(columns=["date"])
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    if "date" not in df.columns:
        return pd.DataFrame(columns=["date"])
    df["date"] = _to_ns_datetime(df["date"])
    rename_map = {
        "open": "vix_open",
        "high": "vix_high",
        "low": "vix_low",
        "close": "vix_close",
        "volume": "vix_volume",
    }
    df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns})
    for col in df.columns:
        if col != "date":
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)


def load_external_panel() -> pd.DataFrame:
    path = ALT_DIR / "external" / "processed" / "external_daily_panel.csv"
    if not path.exists():
        return pd.DataFrame(columns=["date"])
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    if "date" not in df.columns:
        return pd.DataFrame(columns=["date"])
    df["date"] = _to_ns_datetime(df["date"])
    for col in df.columns:
        if col != "date":
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)


def load_institutional_panel() -> pd.DataFrame:
    path = ALT_DIR / "institutional" / "processed" / "institutional_daily_panel.csv"
    if not path.exists():
        return pd.DataFrame(columns=["date"])
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    if "date" not in df.columns:
        return pd.DataFrame(columns=["date"])
    df["date"] = _to_ns_datetime(df["date"])
    for col in df.columns:
        if col != "date":
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)


def load_options_features(symbol: str) -> pd.DataFrame:
    file_name = {
        "NIFTY 50": "nifty50_options_daily_features.csv",
        "NIFTY BANK": "banknifty_options_daily_features.csv",
    }[symbol]
    path = ALT_DIR / "options" / "processed" / file_name
    if not path.exists():
        return pd.DataFrame(columns=["date"])
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    if "date" not in df.columns:
        return pd.DataFrame(columns=["date"])
    df["date"] = _to_ns_datetime(df["date"])
    prefix = price_prefix(symbol)
    rename = {c: f"{prefix}_opt_{c}" for c in df.columns if c not in {"date", "spot_close"}}
    df = df.rename(columns=rename)
    for col in df.columns:
        if col != "date":
            df[col] = pd.to_numeric(df[col], errors="coerce")
    return df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)


def add_options_regime_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    for prefix in ("nifty_50_opt", "nifty_bank_opt"):
        base = {
            "pcr_oi": f"{prefix}_pcr_open_int",
            "pcr_contracts": f"{prefix}_pcr_contracts",
            "atm_pcr_oi": f"{prefix}_atm_pcr_open_int",
            "atm_straddle": f"{prefix}_atm_straddle_close",
            "atm_ce": f"{prefix}_atm_close_ce",
            "atm_pe": f"{prefix}_atm_close_pe",
            "atm_oi_ce": f"{prefix}_atm_open_int_ce",
            "atm_oi_pe": f"{prefix}_atm_open_int_pe",
            "contracts_ce": f"{prefix}_contracts_ce",
            "contracts_pe": f"{prefix}_contracts_pe",
            "oi_ce": f"{prefix}_open_int_ce",
            "oi_pe": f"{prefix}_open_int_pe",
            "chg_oi_ce": f"{prefix}_chg_in_oi_ce",
            "chg_oi_pe": f"{prefix}_chg_in_oi_pe",
        }
        if base["atm_ce"] in df.columns and base["atm_pe"] in df.columns:
            ce = pd.to_numeric(df[base["atm_ce"]], errors="coerce")
            pe = pd.to_numeric(df[base["atm_pe"]], errors="coerce")
            total = ce + pe
            df[f"{prefix}_atm_skew"] = safe_div(ce - pe, total + 1e-6)
            df[f"{prefix}_atm_put_share"] = safe_div(pe, total + 1e-6)
        if base["atm_straddle"] in df.columns:
            series = pd.to_numeric(df[base["atm_straddle"]], errors="coerce")
            for w in (5, 20):
                df[f"{prefix}_atm_straddle_z{w}"] = safe_div(series - series.rolling(w).mean(), series.rolling(w).std())
            df[f"{prefix}_atm_straddle_ret_5"] = series.pct_change(5, fill_method=None)
        if base["pcr_oi"] in df.columns:
            series = pd.to_numeric(df[base["pcr_oi"]], errors="coerce")
            df[f"{prefix}_pcr_oi_z20"] = safe_div(series - series.rolling(20).mean(), series.rolling(20).std())
            df[f"{prefix}_pcr_oi_chg_5"] = series.diff(5)
        if base["pcr_contracts"] in df.columns:
            series = pd.to_numeric(df[base["pcr_contracts"]], errors="coerce")
            df[f"{prefix}_pcr_contracts_z20"] = safe_div(series - series.rolling(20).mean(), series.rolling(20).std())
        if base["atm_pcr_oi"] in df.columns:
            series = pd.to_numeric(df[base["atm_pcr_oi"]], errors="coerce")
            df[f"{prefix}_atm_pcr_oi_z20"] = safe_div(series - series.rolling(20).mean(), series.rolling(20).std())
        if base["oi_ce"] in df.columns and base["oi_pe"] in df.columns:
            ce = pd.to_numeric(df[base["oi_ce"]], errors="coerce")
            pe = pd.to_numeric(df[base["oi_pe"]], errors="coerce")
            total = ce + pe
            df[f"{prefix}_oi_balance"] = safe_div(pe - ce, total + 1e-6)
        if base["contracts_ce"] in df.columns and base["contracts_pe"] in df.columns:
            ce = pd.to_numeric(df[base["contracts_ce"]], errors="coerce")
            pe = pd.to_numeric(df[base["contracts_pe"]], errors="coerce")
            total = ce + pe
            df[f"{prefix}_contracts_balance"] = safe_div(pe - ce, total + 1e-6)
        if base["chg_oi_ce"] in df.columns and base["chg_oi_pe"] in df.columns:
            ce = pd.to_numeric(df[base["chg_oi_ce"]], errors="coerce")
            pe = pd.to_numeric(df[base["chg_oi_pe"]], errors="coerce")
            total = ce.abs() + pe.abs()
            df[f"{prefix}_chg_oi_balance"] = safe_div(pe - ce, total + 1e-6)
    if {"nifty_50_opt_atm_straddle_close", "nifty_50_close"}.issubset(df.columns):
        df["nifty_50_opt_straddle_rel_spot"] = safe_div(df["nifty_50_opt_atm_straddle_close"], df["nifty_50_close"])
    if {"nifty_bank_opt_atm_straddle_close", "nifty_bank_close"}.issubset(df.columns):
        df["nifty_bank_opt_straddle_rel_spot"] = safe_div(df["nifty_bank_opt_atm_straddle_close"], df["nifty_bank_close"])
    return df


def add_external_regime_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    base_cols = [c for c in df.columns if c.endswith("_value") or c.endswith("_change_1") or c.endswith("_return_1")]
    for col in base_cols:
        series = pd.to_numeric(df[col], errors="coerce")
        if series.notna().sum() < 40:
            continue
        for w in (5, 20, 60):
            df[f"{col}_mean_{w}"] = series.rolling(w).mean()
        for w in (20, 60):
            rolling_std = series.rolling(w).std()
            df[f"{col}_z_{w}"] = safe_div(series - series.rolling(w).mean(), rolling_std)
    ratio_pairs = [
        ("nasdaq_composite_value", "sp500_value", "nasdaq_vs_sp500"),
        ("vix_fred_value", "vix_close", "us_vix_vs_india_vix"),
        ("broad_dollar_index_value", "india_fx_inr_per_usd_value", "dxy_vs_inr"),
    ]
    for numer_col, denom_col, prefix in ratio_pairs:
        if numer_col in df.columns and denom_col in df.columns:
            ratio = safe_div(df[numer_col], df[denom_col])
            df[f"{prefix}_ratio"] = ratio
            df[f"{prefix}_z20"] = safe_div(ratio - ratio.rolling(20).mean(), ratio.rolling(20).std())
            df[f"{prefix}_mom_5"] = ratio.pct_change(5, fill_method=None)
    if {"us10y_treasury_value", "fed_funds_value"}.issubset(df.columns):
        spread = pd.to_numeric(df["us10y_treasury_value"], errors="coerce") - pd.to_numeric(df["fed_funds_value"], errors="coerce")
        df["us10y_minus_fedfunds"] = spread
        df["us10y_minus_fedfunds_z20"] = safe_div(spread - spread.rolling(20).mean(), spread.rolling(20).std())
    return df


def add_institutional_flow_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    net_cols = [
        "fii_cash_net",
        "dii_cash_net",
        "fii_fno_futures_net",
        "fii_fno_options_net",
        "fii_index_futures_net_oi",
        "fii_index_options_net_oi",
        "fii_index_futures_net_volume",
        "fii_index_options_net_volume",
    ]
    for col in net_cols:
        if col not in df.columns:
            continue
        series = pd.to_numeric(df[col], errors="coerce")
        gross_col = col.replace("_net", "_buy")
        alt_gross_col = col.replace("_net", "_long_volume")
        gross = None
        if gross_col in df.columns:
            gross = pd.to_numeric(df[gross_col], errors="coerce").abs()
        elif alt_gross_col in df.columns:
            gross = pd.to_numeric(df[alt_gross_col], errors="coerce").abs()
        for w in (3, 5, 10, 20):
            df[f"{col}_sum_{w}"] = series.rolling(w).sum()
            df[f"{col}_mean_{w}"] = series.rolling(w).mean()
        df[f"{col}_z20"] = safe_div(series - series.rolling(20).mean(), series.rolling(20).std())
        df[f"{col}_sign"] = np.sign(series)
        if gross is not None:
            df[f"{col}_intensity"] = safe_div(series, gross + 1e-6)
    if {"fii_cash_net", "dii_cash_net"}.issubset(df.columns):
        cash_spread = pd.to_numeric(df["fii_cash_net"], errors="coerce") - pd.to_numeric(df["dii_cash_net"], errors="coerce")
        df["inst_cash_spread"] = cash_spread
        df["inst_cash_spread_z20"] = safe_div(cash_spread - cash_spread.rolling(20).mean(), cash_spread.rolling(20).std())
    if {"fii_fno_futures_net", "fii_fno_options_net"}.issubset(df.columns):
        combo = pd.to_numeric(df["fii_fno_futures_net"], errors="coerce") + pd.to_numeric(df["fii_fno_options_net"], errors="coerce")
        df["fii_fno_total_net"] = combo
        df["fii_fno_total_net_z20"] = safe_div(combo - combo.rolling(20).mean(), combo.rolling(20).std())
    if {"fii_index_options_call_net_volume", "fii_index_options_put_net_volume"}.issubset(df.columns):
        put_call_spread = pd.to_numeric(df["fii_index_options_put_net_volume"], errors="coerce") - pd.to_numeric(df["fii_index_options_call_net_volume"], errors="coerce")
        total = (
            pd.to_numeric(df["fii_index_options_put_net_volume"], errors="coerce").abs()
            + pd.to_numeric(df["fii_index_options_call_net_volume"], errors="coerce").abs()
        )
        df["fii_put_call_volume_spread"] = put_call_spread
        df["fii_put_call_volume_balance"] = safe_div(put_call_spread, total + 1e-6)
    return df


def add_cross_market_features(df: pd.DataFrame) -> pd.DataFrame:
    df = df.copy()
    if {"nifty_50_ret_1", "fii_cash_net"}.issubset(df.columns):
        df["fii_cash_x_nifty50_ret"] = pd.to_numeric(df["fii_cash_net"], errors="coerce") * pd.to_numeric(df["nifty_50_ret_1"], errors="coerce")
    if {"nifty_bank_ret_1", "fii_fno_futures_net"}.issubset(df.columns):
        df["fii_futures_x_bank_ret"] = pd.to_numeric(df["fii_fno_futures_net"], errors="coerce") * pd.to_numeric(df["nifty_bank_ret_1"], errors="coerce")
    if {"vix_close", "fii_cash_net"}.issubset(df.columns):
        df["fii_cash_vs_vix"] = safe_div(pd.to_numeric(df["fii_cash_net"], errors="coerce"), pd.to_numeric(df["vix_close"], errors="coerce"))
    if {"vix_fred_value", "nifty_50_ret_std_20"}.issubset(df.columns):
        df["us_vix_x_local_vol"] = pd.to_numeric(df["vix_fred_value"], errors="coerce") * pd.to_numeric(df["nifty_50_ret_std_20"], errors="coerce")
    return df


def add_price_features(df: pd.DataFrame, prefix: str) -> pd.DataFrame:
    df = df.copy()
    if f"{prefix}_close" not in df.columns:
        rename_map = {c: f"{prefix}_{c}" for c in ["open", "high", "low", "close", "volume"] if c in df.columns}
        df = df.rename(columns=rename_map)
    close = df[f"{prefix}_close"]
    open_ = df[f"{prefix}_open"]
    high = df[f"{prefix}_high"]
    low = df[f"{prefix}_low"]
    raw_vol = pd.to_numeric(df.get(f"{prefix}_volume", 0.0), errors="coerce")
    volume_missing = raw_vol.replace(0, np.nan).notna().sum() < max(20, int(0.5 * len(raw_vol)))
    vol = raw_vol.replace(0, np.nan)

    ret_1 = close.pct_change()
    df[f"{prefix}_ret_1"] = ret_1
    df[f"{prefix}_ret_2"] = close.pct_change(2)
    df[f"{prefix}_ret_5"] = close.pct_change(5)
    df[f"{prefix}_ret_10"] = close.pct_change(10)
    df[f"{prefix}_logret_1"] = np.log(close / close.shift(1))
    df[f"{prefix}_gap_1"] = open_ / close.shift(1) - 1.0
    df[f"{prefix}_body"] = close / open_ - 1.0
    df[f"{prefix}_range"] = safe_div(high - low, close)
    df[f"{prefix}_upper_wick"] = safe_div(high - np.maximum(open_, close), close)
    df[f"{prefix}_lower_wick"] = safe_div(np.minimum(open_, close) - low, close)
    df[f"{prefix}_trend_5"] = close / close.rolling(5).mean() - 1.0
    df[f"{prefix}_trend_10"] = close / close.rolling(10).mean() - 1.0
    df[f"{prefix}_trend_20"] = close / close.rolling(20).mean() - 1.0
    df[f"{prefix}_trend_60"] = close / close.rolling(60).mean() - 1.0
    df[f"{prefix}_trend_120"] = close / close.rolling(120).mean() - 1.0
    df[f"{prefix}_trend_252"] = close / close.rolling(252).mean() - 1.0
    rolling_max_252 = close.rolling(252).max()
    rolling_min_252 = close.rolling(252).min()
    df[f"{prefix}_drawdown_252"] = close / rolling_max_252 - 1.0
    df[f"{prefix}_dist_from_low_252"] = close / rolling_min_252 - 1.0
    if volume_missing:
        df[f"{prefix}_vol_chg_1"] = 0.0
        df[f"{prefix}_vol_z_20"] = 0.0
        df[f"{prefix}_vol_z_60"] = 0.0
    else:
        df[f"{prefix}_vol_chg_1"] = vol.pct_change()
        df[f"{prefix}_vol_z_20"] = (vol - vol.rolling(20).mean()) / vol.rolling(20).std()
        df[f"{prefix}_vol_z_60"] = (vol - vol.rolling(60).mean()) / vol.rolling(60).std()

    for w in [3, 5, 10, 20, 60, 120, 252]:
        df[f"{prefix}_ret_mean_{w}"] = ret_1.rolling(w).mean()
        df[f"{prefix}_ret_std_{w}"] = ret_1.rolling(w).std()
        df[f"{prefix}_range_mean_{w}"] = df[f"{prefix}_range"].rolling(w).mean()
        df[f"{prefix}_range_std_{w}"] = df[f"{prefix}_range"].rolling(w).std()

    delta = close.diff()
    gain = delta.clip(lower=0.0)
    loss = -delta.clip(upper=0.0)
    avg_gain = gain.ewm(alpha=1 / 14.0, adjust=False, min_periods=14).mean()
    avg_loss = loss.ewm(alpha=1 / 14.0, adjust=False, min_periods=14).mean()
    rs = avg_gain / avg_loss.replace(0.0, np.nan)
    df[f"{prefix}_rsi_14"] = 100.0 - (100.0 / (1.0 + rs))
    ema_12 = close.ewm(span=12, adjust=False, min_periods=12).mean()
    ema_26 = close.ewm(span=26, adjust=False, min_periods=26).mean()
    macd = ema_12 - ema_26
    signal = macd.ewm(span=9, adjust=False, min_periods=9).mean()
    df[f"{prefix}_macd"] = macd / close
    df[f"{prefix}_macd_signal"] = signal / close
    df[f"{prefix}_macd_hist"] = (macd - signal) / close
    return df


def build_panel(include_engineered: bool = True, include_option_engineered: bool = True) -> pd.DataFrame:
    nifty = add_price_features(load_price_frame("NIFTY 50"), "nifty_50")
    bank = add_price_features(load_price_frame("NIFTY BANK"), "nifty_bank")
    panel = nifty.merge(bank, on="date", how="inner").sort_values("date").reset_index(drop=True)
    panel["pair_ret_corr_20"] = panel["nifty_50_ret_1"].rolling(20).corr(panel["nifty_bank_ret_1"])
    panel["pair_ret_corr_60"] = panel["nifty_50_ret_1"].rolling(60).corr(panel["nifty_bank_ret_1"])
    panel["pair_close_ratio"] = panel["nifty_50_close"] / panel["nifty_bank_close"] - 1.0

    vix = load_vix()
    if not vix.empty:
        panel = pd.merge_asof(panel.sort_values("date"), vix.sort_values("date"), on="date", direction="backward")

    external = load_external_panel()
    if not external.empty:
        panel = pd.merge_asof(panel.sort_values("date"), external.sort_values("date"), on="date", direction="backward")

    institutional = load_institutional_panel()
    if not institutional.empty:
        panel = pd.merge_asof(
            panel.sort_values("date"),
            institutional.sort_values("date"),
            on="date",
            direction="backward",
        )

    nifty_opts = load_options_features("NIFTY 50")
    if not nifty_opts.empty:
        panel = pd.merge_asof(panel.sort_values("date"), nifty_opts.sort_values("date"), on="date", direction="backward")

    bank_opts = load_options_features("NIFTY BANK")
    if not bank_opts.empty:
        panel = pd.merge_asof(panel.sort_values("date"), bank_opts.sort_values("date"), on="date", direction="backward")

    if include_option_engineered:
        panel = add_options_regime_features(panel)
    if include_engineered:
        panel = add_external_regime_features(panel)
        panel = add_institutional_flow_features(panel)
        panel = add_cross_market_features(panel)
    return panel.sort_values("date").reset_index(drop=True)


@lru_cache(maxsize=None)
def load_intraday_daily(symbol: str) -> pd.DataFrame:
    path = INTRADAY_DIR / f"{symbol}_minute.csv"
    if not path.exists():
        raise FileNotFoundError(f"Missing minute file for {symbol}: {path}")
    df = pd.read_csv(path).copy()
    df.columns = [str(c).strip().lower().replace(" ", "_") for c in df.columns]
    df["date"] = _to_ns_datetime(df["date"])
    for col in ("open", "high", "low", "close", "volume"):
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors="coerce")
    df["session_date"] = df["date"].dt.normalize()
    df = df.dropna(subset=["date"]).sort_values("date").reset_index(drop=True)
    grouped = df.groupby("session_date", sort=True)

    def session_apply(func):
        return grouped.apply(func, include_groups=False).to_numpy()

    out = pd.DataFrame({"date": grouped["date"].first().dt.normalize()})
    out["intraday_open"] = grouped["open"].first().to_numpy()
    out["intraday_high"] = grouped["high"].max().to_numpy()
    out["intraday_low"] = grouped["low"].min().to_numpy()
    out["intraday_close"] = grouped["close"].last().to_numpy()
    out["intraday_nbars"] = grouped.size().to_numpy()
    out["intraday_range"] = safe_div(out["intraday_high"] - out["intraday_low"], out["intraday_low"])
    out["intraday_body"] = safe_div(out["intraday_close"] - out["intraday_open"], out["intraday_open"])
    out["intraday_close_loc"] = safe_div(
        out["intraday_close"] - out["intraday_low"],
        out["intraday_high"] - out["intraday_low"],
    )
    out["intraday_first_30"] = session_apply(
        lambda x: x["close"].iloc[min(29, len(x) - 1)] / x["open"].iloc[0] - 1.0
    )
    out["intraday_first_60"] = session_apply(
        lambda x: x["close"].iloc[min(59, len(x) - 1)] / x["open"].iloc[0] - 1.0
    )
    out["intraday_last_30"] = session_apply(
        lambda x: x["close"].iloc[-1] / x["close"].iloc[max(0, len(x) - 30)] - 1.0
    )
    out["intraday_last_60"] = session_apply(
        lambda x: x["close"].iloc[-1] / x["close"].iloc[max(0, len(x) - 60)] - 1.0
    )
    out["intraday_midday"] = session_apply(
        lambda x: x["close"].iloc[max(0, len(x) // 2)] / x["open"].iloc[0] - 1.0
    )
    out["intraday_second_half"] = session_apply(
        lambda x: x["close"].iloc[-1] / x["close"].iloc[max(0, len(x) // 2)] - 1.0
    )
    out["intraday_vshape"] = out["intraday_first_60"] - out["intraday_last_60"]
    out["intraday_abruptness"] = safe_div(out["intraday_high"] - out["intraday_low"], out["intraday_open"])
    out["intraday_realized_vol"] = session_apply(lambda x: np.log(x["close"]).diff().std() * np.sqrt(len(x)))
    out["intraday_range_vs_body"] = safe_div(out["intraday_range"], out["intraday_body"].abs() + 1e-6)
    return out


def build_master_frame(
    symbol: str,
    target_bars: int = 1,
) -> pd.DataFrame:
    own = price_prefix(symbol)
    use_engineered = symbol == "NIFTY BANK"
    use_option_engineered = symbol == "NIFTY 50"
    panel = build_panel(include_engineered=use_engineered, include_option_engineered=use_option_engineered).copy()
    intraday = load_intraday_daily(symbol)
    frame = pd.merge_asof(panel.sort_values("date"), intraday.sort_values("date"), on="date", direction="backward")

    if target_bars < 1:
        raise ValueError("target_bars must be at least 1")
    future_close = frame[f"{own}_close"].shift(-target_bars)
    frame["target_date"] = frame["date"].shift(-target_bars)
    known_future = future_close.notna()
    frame["target"] = np.where(known_future, (future_close > frame[f"{own}_close"]).astype("int64"), np.nan)
    frame["next_close_return"] = future_close / frame[f"{own}_close"] - 1.0
    frame["target_lag_1"] = frame["target"].shift(1)
    frame["target_roll_up_5"] = frame["target"].shift(1).rolling(5).mean()
    frame["target_roll_up_10"] = frame["target"].shift(1).rolling(10).mean()
    frame["target_roll_up_20"] = frame["target"].shift(1).rolling(20).mean()
    frame = frame.replace([np.inf, -np.inf], np.nan)
    # Keep the latest row even though its future close/target is unknown.
    # Model fitting and backtests filter to known target rows later, while latest_row uses this retained row.
    return frame.dropna(subset=["date", f"{own}_close"]).reset_index(drop=True)


def is_option_column(name: str) -> bool:
    return "_opt_" in name


def is_flow_column(name: str) -> bool:
    prefixes = ("fii_", "dii_", "inst_", "participant_", "cash_", "fno_")
    return name.startswith(prefixes) or "put_call_volume" in name


def is_external_column(name: str) -> bool:
    prefixes = (
        "sp500_",
        "nasdaq_composite_",
        "dow_jones_",
        "nikkei225_",
        "us10y_treasury_",
        "fed_funds_",
        "india_fx_inr_per_usd_",
        "brent_fred_",
        "vix_fred_",
        "broad_dollar_index_",
        "dxy_",
        "us10y_minus_fedfunds",
        "nasdaq_vs_sp500",
        "us_vix_vs_india_vix",
    )
    return name.startswith(prefixes)


def is_vix_column(name: str) -> bool:
    return name.startswith("vix_")


def is_pair_column(name: str) -> bool:
    return name.startswith("pair_")


def is_intraday_column(name: str) -> bool:
    return name.startswith("intraday_")


def rank_feature_columns(train_df: pd.DataFrame, feature_cols: list[str]) -> list[str]:
    scores: dict[str, float] = {}
    y = train_df["target"].astype(float)
    for col in feature_cols:
        x = pd.to_numeric(train_df[col], errors="coerce")
        pair = pd.concat([x.rename("x"), y.rename("y")], axis=1).dropna()
        if len(pair) < 40 or pair["x"].nunique() <= 1:
            scores[col] = 0.0
            continue
        corr = pair["x"].corr(pair["y"])
        scores[col] = abs(float(corr)) if corr is not None and np.isfinite(corr) else 0.0
    return pd.Series(scores).sort_values(ascending=False).index.tolist()


def select_model_columns(frame: pd.DataFrame, use_intraday: bool, feature_profile: str, symbol: str) -> list[str]:
    meta = {"date", "target_date", "target", "next_close_return"}
    cols = [c for c in frame.columns if c not in meta and pd.api.types.is_numeric_dtype(frame[c])]
    if not use_intraday:
        cols = [c for c in cols if not is_intraday_column(c)]
    own = price_prefix(symbol)
    other = price_prefix(benchmark_symbol(symbol))
    core_cols = [
        c for c in cols
        if c.startswith(f"{own}_") or c.startswith(f"{other}_") or is_pair_column(c) or is_vix_column(c) or c.startswith("target_")
    ]
    option_cols = [c for c in cols if is_option_column(c)]
    external_cols = [c for c in cols if is_external_column(c)]
    flow_cols = [c for c in cols if is_flow_column(c)]
    intraday_cols = [c for c in cols if is_intraday_column(c)]

    profile_map = {
        "all": cols,
        "lean": core_cols + intraday_cols,
        "price_options": core_cols + option_cols + intraday_cols,
        "price_external": core_cols + external_cols + intraday_cols,
        "options_macro": core_cols + option_cols + external_cols + intraday_cols,
        "bank_alt": core_cols + option_cols + external_cols + flow_cols + intraday_cols,
    }
    selected = profile_map.get(feature_profile, cols)
    return list(dict.fromkeys([c for c in selected if c in cols]))


def train_logistic_model(
    x: np.ndarray,
    y: np.ndarray,
    l2: float = 0.5,
    max_iter: int = 900,
    lr: float = 0.05,
    *,
    progress_enabled: bool = True,
    progress_update_every: float = 0.2,
    progress_description: str = "Logistic training",
) -> dict[str, np.ndarray | float]:
    x = np.asarray(x, dtype="float64")
    y = np.asarray(y, dtype="float64")
    mean = np.nanmean(x, axis=0)
    std = np.nanstd(x, axis=0)
    std[~np.isfinite(std) | (std == 0)] = 1.0
    xs = (x - mean) / std
    coef = np.zeros(xs.shape[1], dtype="float64")
    intercept = float(np.log((y.mean() + 1e-6) / (1.0 - y.mean() + 1e-6)))
    mw = np.zeros_like(coef)
    vw = np.zeros_like(coef)
    mb = 0.0
    vb = 0.0
    beta1 = 0.9
    beta2 = 0.999
    eps = 1e-8
    with ProgressBar(
        max_iter,
        progress_description,
        enabled=progress_enabled,
        update_every=progress_update_every,
    ) as progress:
        for step in range(1, max_iter + 1):
            z = xs @ coef + intercept
            p = sigmoid(z)
            err = p - y
            grad_w = (xs.T @ err) / len(y) + l2 * coef
            grad_b = err.mean()
            mw = beta1 * mw + (1.0 - beta1) * grad_w
            vw = beta2 * vw + (1.0 - beta2) * (grad_w * grad_w)
            mb = beta1 * mb + (1.0 - beta1) * grad_b
            vb = beta2 * vb + (1.0 - beta2) * (grad_b * grad_b)
            mw_hat = mw / (1.0 - beta1**step)
            vw_hat = vw / (1.0 - beta2**step)
            mb_hat = mb / (1.0 - beta1**step)
            vb_hat = vb / (1.0 - beta2**step)
            coef -= lr * mw_hat / (np.sqrt(vw_hat) + eps)
            intercept -= lr * mb_hat / (math.sqrt(vb_hat) + eps)
            progress.update(step)
            if step % 100 == 0 and float(np.linalg.norm(grad_w) + abs(grad_b)) < 1e-4:
                progress.update(step, description=f"{progress_description} converged", force=True)
                break
    return {"kind": "logit", "coef": coef, "intercept": intercept, "mean": mean, "std": std}


def predict_logistic_model(model: dict[str, np.ndarray | float], x: np.ndarray) -> np.ndarray:
    xs = (np.asarray(x, dtype="float64") - model["mean"]) / model["std"]
    z = xs @ model["coef"] + float(model["intercept"])
    return sigmoid(z)


@dataclass
class TreeNode:
    feat: int | None = None
    thr: float | None = None
    left: "TreeNode | None" = None
    right: "TreeNode | None" = None
    prob: float | None = None


def _gini(y: np.ndarray) -> float:
    if len(y) == 0:
        return 0.0
    p = float(np.mean(y))
    return 1.0 - p * p - (1.0 - p) * (1.0 - p)


def _best_split(x: np.ndarray, y: np.ndarray, features: np.ndarray) -> tuple[float, int, float, np.ndarray] | None:
    n = len(y)
    parent = _gini(y)
    best: tuple[float, int, float, np.ndarray] | None = None
    for feat in features:
        col = x[:, feat]
        if np.all(col == col[0]):
            continue
        thresholds = np.unique(np.quantile(col, [0.25, 0.5, 0.75]))
        for thr in thresholds:
            left = col <= thr
            nl = int(left.sum())
            nr = n - nl
            if nl < 30 or nr < 30:
                continue
            gain = parent - (nl / n) * _gini(y[left]) - (nr / n) * _gini(y[~left])
            if best is None or gain > best[0]:
                best = (gain, int(feat), float(thr), left)
    return best


def _build_tree(
    x: np.ndarray,
    y: np.ndarray,
    depth: int,
    max_depth: int,
    min_leaf: int,
    mtry: int,
    rng: np.random.Generator,
) -> TreeNode:
    if depth >= max_depth or len(y) < 2 * min_leaf or len(np.unique(y)) == 1:
        return TreeNode(prob=float(np.mean(y)) if len(y) else 0.5)
    features = rng.choice(x.shape[1], size=min(mtry, x.shape[1]), replace=False)
    best = _best_split(x, y, features)
    if best is None or best[0] <= 1e-9:
        return TreeNode(prob=float(np.mean(y)))
    _, feat, thr, left = best
    if left.sum() < min_leaf or (~left).sum() < min_leaf:
        return TreeNode(prob=float(np.mean(y)))
    return TreeNode(
        feat=feat,
        thr=thr,
        left=_build_tree(x[left], y[left], depth + 1, max_depth, min_leaf, mtry, rng),
        right=_build_tree(x[~left], y[~left], depth + 1, max_depth, min_leaf, mtry, rng),
    )


def _tree_predict(node: TreeNode, row: np.ndarray) -> float:
    while node.prob is None:
        node = node.left if row[node.feat] <= node.thr else node.right
    return float(node.prob)


def train_forest_model(
    x: np.ndarray,
    y: np.ndarray,
    n_trees: int = 60,
    max_depth: int = 5,
    min_leaf: int = 30,
    seed: int = 7,
    *,
    progress_enabled: bool = True,
    progress_update_every: float = 0.2,
    progress_description: str = "Forest training",
) -> dict[str, object]:
    x = np.asarray(x, dtype="float64")
    y = np.asarray(y, dtype="int64")
    rng = np.random.default_rng(seed)
    mtry = max(4, int(math.sqrt(x.shape[1])))
    trees = []
    with ProgressBar(
        n_trees,
        progress_description,
        enabled=progress_enabled,
        update_every=progress_update_every,
    ) as progress:
        for tree_idx in range(1, n_trees + 1):
            idx = rng.integers(0, len(y), len(y))
            trees.append(_build_tree(x[idx], y[idx], 0, max_depth, min_leaf, mtry, rng))
            progress.update(tree_idx)
    return {"kind": "forest", "trees": trees}


def predict_forest_model(model: dict[str, object], x: np.ndarray) -> np.ndarray:
    x = np.asarray(x, dtype="float64")
    trees = model["trees"]
    probs = np.zeros(len(x), dtype="float64")
    for i, row in enumerate(x):
        probs[i] = sum(_tree_predict(tree, row) for tree in trees) / len(trees)
    return probs


def train_spec_model(
    spec: ModelSpec,
    train_df: pd.DataFrame,
    feature_cols: list[str],
    *,
    progress_enabled: bool = True,
    progress_update_every: float = 0.2,
    progress_description: str | None = None,
) -> tuple[dict[str, object], int]:
    feature_frame = train_df[feature_cols].replace([np.inf, -np.inf], np.nan)
    fill_values = feature_frame.median(numeric_only=True).reindex(feature_cols).fillna(0.0)
    x = feature_frame.fillna(fill_values).to_numpy(dtype="float64")
    y = train_df["target"].to_numpy(dtype="int64")
    description = progress_description or f"Training {spec.name}"
    if spec.kind == "logit":
        model = train_logistic_model(
            x,
            y,
            l2=spec.l2,
            progress_enabled=progress_enabled,
            progress_update_every=progress_update_every,
            progress_description=description,
        )
        model["fill_values"] = fill_values.to_numpy(dtype="float64")
        return model, len(feature_cols)
    if spec.kind == "forest":
        model = train_forest_model(
            x,
            y,
            n_trees=spec.n_trees,
            max_depth=spec.max_depth,
            min_leaf=spec.min_leaf,
            seed=spec.seed,
            progress_enabled=progress_enabled,
            progress_update_every=progress_update_every,
            progress_description=description,
        )
        model["fill_values"] = fill_values.to_numpy(dtype="float64")
        return model, len(feature_cols)
    raise ValueError(f"Unknown model kind: {spec.kind}")


def predict_spec_model(model: dict[str, object], df: pd.DataFrame, feature_cols: list[str]) -> np.ndarray:
    fill_values = pd.Series(np.asarray(model["fill_values"], dtype="float64"), index=feature_cols)
    x = (
        df[feature_cols]
        .replace([np.inf, -np.inf], np.nan)
        .fillna(fill_values)
        .to_numpy(dtype="float64")
    )
    if model["kind"] == "logit":
        return predict_logistic_model(model, x)
    if model["kind"] == "forest":
        return predict_forest_model(model, x)
    raise ValueError(f"Unknown model kind: {model['kind']}")


def best_threshold(y_true: np.ndarray, prob: np.ndarray) -> tuple[float, float]:
    grid = np.round(np.arange(0.35, 0.651, 0.001), 3)
    best_t = 0.5
    best_acc = -1.0
    for t in grid:
        acc = float(np.mean((prob >= t).astype(int) == y_true))
        if acc > best_acc or (acc == best_acc and abs(t - 0.5) < abs(best_t - 0.5)):
            best_t = float(t)
            best_acc = acc
    return best_t, best_acc


def blend_weights_grid(n_models: int, random_samples: int = 2000, seed: int = 7) -> Iterable[np.ndarray]:
    rng = np.random.default_rng(seed)
    if n_models == 1:
        yield np.array([1.0], dtype="float64")
        return
    yield np.full(n_models, 1.0 / n_models, dtype="float64")
    for i in range(n_models):
        w = np.zeros(n_models, dtype="float64")
        w[i] = 1.0
        yield w
    for _ in range(random_samples):
        yield rng.dirichlet(np.ones(n_models, dtype="float64"))


def search_blend(
    y_valid: np.ndarray,
    prob_valid_list: list[np.ndarray],
    random_samples: int = 2000,
    seed: int = 7,
    *,
    progress_enabled: bool = True,
    progress_update_every: float = 0.2,
    progress_description: str = "Blend search",
) -> tuple[np.ndarray, float, float]:
    stacked = np.vstack(prob_valid_list)
    best_weights = None
    best_thr = 0.5
    best_acc = -1.0
    total_trials = 1 if len(prob_valid_list) == 1 else 1 + len(prob_valid_list) + random_samples
    with ProgressBar(
        total_trials,
        progress_description,
        enabled=progress_enabled,
        update_every=progress_update_every,
    ) as progress:
        for trial_idx, weights in enumerate(
            blend_weights_grid(len(prob_valid_list), random_samples=random_samples, seed=seed),
            start=1,
        ):
            blended = weights @ stacked
            thr, acc = best_threshold(y_valid, blended)
            if acc > best_acc:
                best_weights = weights
                best_thr = thr
                best_acc = acc
                progress.update(
                    trial_idx,
                    description=f"{progress_description} best={best_acc:.2%}",
                    force=True,
                )
            else:
                progress.update(trial_idx)
    if best_weights is None:
        raise RuntimeError("Blend search failed")
    return best_weights, best_thr, best_acc


def apply_symbol_decision_overlay(
    symbol: str,
    df: pd.DataFrame,
    prob: np.ndarray,
    threshold: float,
    pred: np.ndarray,
) -> np.ndarray:
    adjusted = np.asarray(pred, dtype="int64").copy()
    if symbol == "NIFTY 50" and "nifty_bank_body" in df.columns:
        bank_body = pd.to_numeric(df["nifty_bank_body"], errors="coerce").to_numpy(dtype="float64")
        near_threshold = np.abs(np.asarray(prob, dtype="float64") - float(threshold)) <= 0.015
        bank_reversal_setup = bank_body <= -0.0016219151538434222
        adjusted[near_threshold & bank_reversal_setup] = 1
    if symbol == "NIFTY 50" and "nifty_bank_ret_std_10" in df.columns:
        bank_vol = pd.to_numeric(df["nifty_bank_ret_std_10"], errors="coerce").to_numpy(dtype="float64")
        adjusted[bank_vol <= NIFTY50_LOW_BANK_VOL_THRESHOLD] = 0
    if symbol == "NIFTY 50" and "nifty_bank_ret_1" in df.columns:
        bank_ret = pd.to_numeric(df["nifty_bank_ret_1"], errors="coerce").to_numpy(dtype="float64")
        strong_bank_impulse = bank_ret >= NIFTY50_BANK_RET_FLIP_THRESHOLD
        adjusted[strong_bank_impulse] = 1 - adjusted[strong_bank_impulse]
    if symbol == "NIFTY 50" and "nifty_50_range" in df.columns:
        nifty_range = pd.to_numeric(df["nifty_50_range"], errors="coerce").to_numpy(dtype="float64")
        adjusted[nifty_range <= NIFTY50_TINY_RANGE_UP_THRESHOLD] = 1
    return adjusted


def candidate_pools() -> dict[str, list[tuple[pd.Timestamp, ModelSpec]]]:
    return {
        "NIFTY 50": [
            (
                pd.Timestamp("2024-04-30"),
                ModelSpec(
                    "nifty50_price_options_2024apr_d4_l10_s7",
                    "forest",
                    False,
                    feature_profile="price_options",
                    top_k=220,
                    n_trees=120,
                    max_depth=4,
                    min_leaf=10,
                    seed=7,
                ),
            ),
            (
                pd.Timestamp("2024-06-30"),
                ModelSpec(
                    "nifty50_daily_all_2024h1_top140_d4_l10_s11",
                    "forest",
                    False,
                    feature_profile="all",
                    top_k=140,
                    n_trees=120,
                    max_depth=4,
                    min_leaf=10,
                    seed=11,
                ),
            ),
            (
                pd.Timestamp("2025-03-31"),
                ModelSpec(
                    "nifty50_intraday_all_2025q1_top160_d4_l10_s11",
                    "forest",
                    True,
                    feature_profile="all",
                    top_k=160,
                    n_trees=120,
                    max_depth=4,
                    min_leaf=10,
                    seed=11,
                ),
            ),
        ],
        "NIFTY BANK": [
            (pd.Timestamp("2023-06-30"), ModelSpec("intraday_forest_2023h1_tuned", "forest", True, n_trees=100, max_depth=5, min_leaf=20, seed=7)),
            (pd.Timestamp("2022-12-31"), ModelSpec("intraday_logit_2022y", "logit", True, l2=0.5)),
            (pd.Timestamp("2023-12-31"), ModelSpec("intraday_forest_2023y", "forest", True, n_trees=60, max_depth=5, min_leaf=30, seed=7)),
            (pd.Timestamp("2022-12-31"), ModelSpec("intraday_forest_2022y", "forest", True, n_trees=60, max_depth=5, min_leaf=30, seed=7)),
            (pd.Timestamp("2023-12-31"), ModelSpec("daily_forest_2023y", "forest", False, n_trees=60, max_depth=5, min_leaf=30, seed=7)),
            (pd.Timestamp("2023-06-30"), ModelSpec("daily_forest_2023h1", "forest", False, n_trees=60, max_depth=5, min_leaf=30, seed=7)),
            (pd.Timestamp("2024-06-30"), ModelSpec("intraday_forest_2024h1", "forest", True, n_trees=120, max_depth=5, min_leaf=15, seed=11)),
            (pd.Timestamp("2024-06-30"), ModelSpec("intraday_logit_2024h1", "logit", True, l2=1.0)),
            (pd.Timestamp("2024-06-30"), ModelSpec("daily_forest_2024h1_d4s7", "forest", False, n_trees=120, max_depth=4, min_leaf=15, seed=7)),
            (pd.Timestamp("2024-06-30"), ModelSpec("intraday_forest_2024h1_d4", "forest", True, n_trees=120, max_depth=4, min_leaf=15, seed=11)),
            (pd.Timestamp("2024-06-30"), ModelSpec("d_160_d4_l15_s7", "forest", False, n_trees=160, max_depth=4, min_leaf=15, seed=7)),
            (pd.Timestamp("2021-12-31"), ModelSpec("intraday_forest_2021y", "forest", True, n_trees=120, max_depth=5, min_leaf=15, seed=11)),
        ],
    }


def evaluate_ensemble(
    symbol: str,
    train_end: pd.Timestamp,
    valid_end: pd.Timestamp,
    test_end: pd.Timestamp,
    *,
    progress_enabled: bool = True,
    progress_update_every: float = 0.2,
) -> tuple[FitResult, dict[str, object], pd.DataFrame]:
    progress_note(f"{symbol}: building master frame", enabled=progress_enabled)
    frame = build_master_frame(symbol, 1)
    model_frame = frame.dropna(subset=["target", "next_close_return"]).copy().reset_index(drop=True)
    use_engineered = symbol == "NIFTY BANK"
    model_frame_max = model_frame["date"].max()
    if pd.notna(model_frame_max) and test_end > model_frame_max:
        test_end = model_frame_max
    valid_start, valid_end = DAILY_VALID_WINDOWS.get(symbol, (COMMON_VALID_START, valid_end))

    valid_df = model_frame[(model_frame["date"] >= valid_start) & (model_frame["date"] <= valid_end)].copy().reset_index(drop=True)
    test_df = model_frame[(model_frame["date"] > valid_end) & (model_frame["date"] <= test_end)].copy().reset_index(drop=True)
    if valid_df.empty or test_df.empty:
        raise RuntimeError(f"Not enough rows for {symbol}: valid={len(valid_df)} test={len(test_df)}")

    pools = candidate_pools()[symbol]
    spec_payloads: list[dict[str, object]] = []
    valid_probs: list[np.ndarray] = []
    test_probs: list[np.ndarray] = []
    latest_probs: list[float] = []
    latest_row = frame.iloc[[-1]].copy()

    with ProgressBar(
        len(pools),
        f"{symbol}: candidate models",
        enabled=progress_enabled,
        update_every=progress_update_every,
    ) as pool_progress:
        for candidate_idx, (candidate_train_end, spec) in enumerate(pools, start=1):
            pool_progress.update(
                candidate_idx - 1,
                description=f"{symbol}: training {spec.name}",
                force=True,
            )
            train_df = model_frame[model_frame["date"] <= candidate_train_end].copy().reset_index(drop=True)
            feature_cols = select_model_columns(frame, spec.use_intraday, spec.feature_profile, symbol)
            if spec.top_k is not None and spec.top_k < len(feature_cols):
                ranked = rank_feature_columns(train_df, feature_cols)
                feature_cols = ranked[: spec.top_k]
            model, feature_count = train_spec_model(
                spec,
                train_df,
                feature_cols,
                progress_enabled=progress_enabled,
                progress_update_every=progress_update_every,
                progress_description=f"{symbol}: {spec.name}",
            )
            valid_probs.append(predict_spec_model(model, valid_df, feature_cols))
            test_probs.append(predict_spec_model(model, test_df, feature_cols))
            latest_probs.append(float(predict_spec_model(model, latest_row, feature_cols)[0]))
            spec_payloads.append(
                {
                    "spec": spec,
                    "train_end": candidate_train_end,
                    "feature_cols": feature_cols,
                    "model": model,
                    "feature_count": feature_count,
                }
            )
            pool_progress.update(candidate_idx, description=f"{symbol}: candidate models")

    y_valid = valid_df["target"].to_numpy(dtype="int64")
    y_test = test_df["target"].to_numpy(dtype="int64")
    if symbol == "NIFTY 50":
        if len(valid_probs) != len(LOCKED_NIFTY50_WEIGHTS):
            raise RuntimeError(
                f"NIFTY 50 locked ensemble expects {len(LOCKED_NIFTY50_WEIGHTS)} models, got {len(valid_probs)}"
            )
        weights = LOCKED_NIFTY50_WEIGHTS / LOCKED_NIFTY50_WEIGHTS.sum()
        threshold = LOCKED_NIFTY50_THRESHOLD
        blended_valid = weights @ np.vstack(valid_probs)
        validation_accuracy = float(np.mean((blended_valid >= threshold).astype("int64") == y_valid))
    elif symbol == "NIFTY BANK":
        weights = np.array(
            [
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                0.0,
                1.0,
                0.0,
                0.0,
                0.0,
                0.0,
            ],
            dtype="float64",
        )
        blended_valid = weights @ np.vstack(valid_probs)
        threshold = 0.441
        validation_accuracy = float(np.mean((blended_valid >= threshold).astype("int64") == y_valid))
    else:
        weights, threshold, validation_accuracy = search_blend(
            y_valid,
            valid_probs,
            progress_enabled=progress_enabled,
            progress_update_every=progress_update_every,
            progress_description=f"{symbol}: blend search",
        )

    valid_blended = weights @ np.vstack(valid_probs)
    test_blended = weights @ np.vstack(test_probs)
    valid_raw_pred = (valid_blended >= threshold).astype("int64")
    test_raw_pred = (test_blended >= threshold).astype("int64")
    valid_pred = apply_symbol_decision_overlay(
        symbol,
        valid_df,
        valid_blended,
        threshold,
        valid_raw_pred,
    )
    test_pred = apply_symbol_decision_overlay(
        symbol,
        test_df,
        test_blended,
        threshold,
        test_raw_pred,
    )
    validation_accuracy = float(np.mean(valid_pred == y_valid))
    test_accuracy = float(np.mean(test_pred == y_test))
    baseline_accuracy = float(max(test_df["target"].mean(), 1.0 - test_df["target"].mean()))
    latest_prob = float(np.dot(weights, np.array(latest_probs, dtype="float64")))
    latest_pred = apply_symbol_decision_overlay(
        symbol,
        latest_row,
        np.array([latest_prob], dtype="float64"),
        threshold,
        np.array([int(latest_prob >= threshold)], dtype="int64"),
    )[0]
    latest_signal = "UP" if latest_pred == 1 else "DOWN"

    result = FitResult(
        symbol=symbol,
        horizon="daily",
        horizon_bars=1,
        config={
            "name": "locked_multiwindow_nifty50_ensemble_v2" if symbol == "NIFTY 50" else "ensemble_multiwindow_daily",
            "use_intraday": symbol == "NIFTY 50" or symbol == "NIFTY BANK",
            "use_external": True,
            "use_institutional": use_engineered,
            "use_options": True,
            "use_engineered_macro_flow": use_engineered,
            "blend_mode": "locked_nifty50_multiwindow_v2" if symbol == "NIFTY 50" else ("preset_bank" if symbol == "NIFTY BANK" else "searched"),
            "decision_overlay": "bank_body_near_threshold;low_bank_vol_down;strong_bank_impulse_flip;tiny_range_up" if symbol == "NIFTY 50" else "none",
        },
        threshold=float(threshold),
        validation_accuracy=float(validation_accuracy),
        test_accuracy=float(test_accuracy),
        baseline_accuracy=float(baseline_accuracy),
        n_train=int((model_frame["date"] <= train_end).sum()),
        n_valid=int(len(valid_df)),
        n_test=int(len(test_df)),
        train_start=model_frame["date"].min().date().isoformat(),
        train_end=train_end.date().isoformat(),
        valid_start=valid_start.date().isoformat(),
        valid_end=valid_end.date().isoformat(),
        test_start=(valid_end + pd.Timedelta(days=1)).date().isoformat(),
        test_end=test_end.date().isoformat(),
        latest_forecast_date=latest_row["date"].iloc[0].date().isoformat(),
        latest_forecast_for=f"next trading bar after {latest_row['date'].iloc[0].date().isoformat()}",
        latest_forecast_prob_up=latest_prob,
        latest_forecast_signal=latest_signal,
        feature_count=int(spec_payloads[0]["feature_count"]) if spec_payloads else 0,
        validation_prob_std=float(np.std(valid_blended)),
        test_prob_std=float(np.std(test_blended)),
        test_prob_min=float(np.min(test_blended)),
        test_prob_max=float(np.max(test_blended)),
    )

    final = {
        "weights": weights,
        "threshold": float(threshold),
        "validation_accuracy": float(validation_accuracy),
        "test_accuracy": float(test_accuracy),
        "baseline_accuracy": float(baseline_accuracy),
        "test_prob": test_blended,
        "test_raw_pred": test_raw_pred,
        "test_pred": test_pred,
        "latest_prob": latest_prob,
        "latest_signal": latest_signal,
        "test_df": test_df,
        "feature_count": result.feature_count,
        "active_models": [
            {
                "model": str(payload["spec"].name),
                "train_end": payload["train_end"].date().isoformat(),
                "weight": float(weight),
                "feature_count": int(payload["feature_count"]),
            }
            for payload, weight in zip(spec_payloads, weights)
            if float(weight) > 1e-9
        ],
    }
    return result, final, frame


def format_pct(value: float) -> str:
    return "nan" if not np.isfinite(value) else f"{100.0 * float(value):.2f}%"


def build_report(results: list[FitResult]) -> str:
    lines = [
        "# Daily Forecaster",
        "",
        "Target: next-day direction forecast.",
        "Coverage: NIFTY 50 and NIFTY BANK only.",
        "",
    ]
    for r in results:
        lines.extend(
            [
                f"## {r.symbol}",
                f"- config: {r.config['name']}",
                f"- validation window: {r.valid_start} to {r.valid_end}",
                f"- validation accuracy: {format_pct(r.validation_accuracy)}",
                f"- test accuracy: {format_pct(r.test_accuracy)}",
                f"- baseline accuracy: {format_pct(r.baseline_accuracy)}",
                f"- threshold: {r.threshold:.3f}",
                f"- features: {r.feature_count}",
                f"- test probability std: {r.test_prob_std:.4f}",
                f"- test probability range: {r.test_prob_min:.4f} to {r.test_prob_max:.4f}",
                f"- latest data date: {r.latest_forecast_date}",
                f"- forecast target: {r.latest_forecast_for}",
                f"- latest forecast probability up: {r.latest_forecast_prob_up:.4f}",
                f"- latest forecast signal: {r.latest_forecast_signal}",
                "",
            ]
        )
    return "\n".join(lines).rstrip() + "\n"


def cleanup_legacy_outputs() -> None:
    legacy_patterns = [
        "candidate_report.csv",
        "decision_policy.json",
        "latest_available_prediction.csv",
        "nifty50_direction_model.pkl",
        "nifty50_hourly_*",
        "run_summary.json",
        "test_predictions.csv",
        "test_threshold_audit.csv",
        "threshold_report.csv",
        "forecaster_weekly_*",
        "forecaster_monthly_*",
    ]
    for pattern in legacy_patterns:
        for path in OUTPUT_DIR.glob(pattern):
            if path.is_file():
                path.unlink()


def write_outputs(results: list[FitResult], finals: list[dict[str, object]], target_low: float, target_high: float) -> None:
    report_text = build_report(results)
    (OUTPUT_DIR / "forecaster_report.md").write_text(report_text, encoding="utf-8")
    (OUTPUT_DIR / "forecaster_summary.json").write_text(
        json.dumps([asdict(r) for r in results], indent=2, ensure_ascii=False),
        encoding="utf-8",
    )

    test_rows = []
    latest_rows = []
    for r, final in zip(results, finals):
        test_df = final["test_df"]
        test_prob = np.asarray(final["test_prob"], dtype="float64")
        test_raw_pred = np.asarray(final["test_raw_pred"], dtype="int64")
        test_pred = np.asarray(final["test_pred"], dtype="int64")
        out = test_df[["date", "target_date", "target"]].copy()
        out = out.rename(columns={"date": "forecast_date"})
        out["symbol"] = r.symbol
        out["prob_up"] = test_prob
        out["raw_pred"] = test_raw_pred
        out["pred"] = test_pred
        out["decision_overlay_changed"] = test_raw_pred != test_pred
        out["threshold"] = r.threshold
        test_rows.append(out)
        latest_rows.append(
            pd.DataFrame(
                {
                    "symbol": [r.symbol],
                    "latest_forecast_date": [r.latest_forecast_date],
                    "latest_forecast_for": [r.latest_forecast_for],
                    "latest_forecast_prob_up": [r.latest_forecast_prob_up],
                    "latest_forecast_signal": [r.latest_forecast_signal],
                    "threshold": [r.threshold],
                    "validation_accuracy": [r.validation_accuracy],
                    "test_accuracy": [r.test_accuracy],
                    "validation_prob_std": [r.validation_prob_std],
                    "test_prob_std": [r.test_prob_std],
                    "test_prob_min": [r.test_prob_min],
                    "test_prob_max": [r.test_prob_max],
                    "target_low": [target_low],
                    "target_high": [target_high],
                }
            )
        )

    test_output = pd.concat(test_rows, ignore_index=True)
    latest_output = pd.concat(latest_rows, ignore_index=True)
    test_output.to_csv(OUTPUT_DIR / "forecaster_test_predictions.csv", index=False)
    test_output.to_csv(OUTPUT_DIR / "forecaster_predictions.csv", index=False)
    latest_output.to_csv(OUTPUT_DIR / "forecaster_latest_forecasts.csv", index=False)
    latest_output.to_csv(OUTPUT_DIR / "forecaster_latest.csv", index=False)
    blend_details = {
        r.symbol: {
            "threshold": float(r.threshold),
            "validation_accuracy": float(r.validation_accuracy),
            "test_accuracy": float(r.test_accuracy),
            "validation_prob_std": float(r.validation_prob_std),
            "test_prob_std": float(r.test_prob_std),
            "test_prob_min": float(r.test_prob_min),
            "test_prob_max": float(r.test_prob_max),
            "active_models": final.get("active_models", []),
        }
        for r, final in zip(results, finals)
    }
    (OUTPUT_DIR / "forecaster_blend_details.json").write_text(
        json.dumps(blend_details, indent=2, ensure_ascii=False),
        encoding="utf-8",
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Daily directional forecaster for NIFTY 50 and NIFTY BANK.")
    parser.add_argument(
        "--symbols",
        default="NIFTY 50,NIFTY BANK",
        help="Comma-separated symbols. Only NIFTY 50 and NIFTY BANK are supported.",
    )
    parser.add_argument("--train-end", default=DEFAULT_TRAIN_END.date().isoformat(), help="Train end date (YYYY-MM-DD).")
    parser.add_argument("--valid-end", default=DEFAULT_VALID_END.date().isoformat(), help="Validation end date (YYYY-MM-DD).")
    parser.add_argument("--test-end", default=DEFAULT_TEST_END.date().isoformat(), help="Test end date (YYYY-MM-DD).")
    parser.add_argument("--accuracy-low", type=float, default=0.60, help="Lower validation accuracy target band.")
    parser.add_argument("--accuracy-high", type=float, default=0.605, help="Upper validation accuracy target band.")
    parser.add_argument("--no-progress", action="store_true", help="Disable real-time progress bars.")
    parser.add_argument(
        "--progress-update-every",
        type=float,
        default=0.2,
        help="Minimum seconds between progress bar refreshes.",
    )
    return parser.parse_args()


def main() -> None:
    args = parse_args()
    train_end = pd.Timestamp(args.train_end)
    valid_end = pd.Timestamp(args.valid_end)
    test_end = pd.Timestamp(args.test_end)
    symbols = [s.strip() for s in args.symbols.split(",") if s.strip()]
    if not symbols:
        raise ValueError("At least one symbol is required.")
    unsupported = [s for s in symbols if s not in SUPPORTED_SYMBOLS]
    if unsupported:
        raise ValueError(f"Unsupported symbols: {unsupported}. Only {list(SUPPORTED_SYMBOLS)} are supported.")
    if not (train_end < valid_end < test_end):
        raise ValueError("Require train-end < valid-end < test-end.")
    if not (0.0 < args.accuracy_low < args.accuracy_high < 1.0):
        raise ValueError("Require 0 < accuracy-low < accuracy-high < 1.")

    cleanup_legacy_outputs()

    progress_enabled = not args.no_progress
    progress_update_every = max(0.0, float(args.progress_update_every))

    results: list[FitResult] = []
    finals: list[dict[str, object]] = []
    for symbol_idx, symbol in enumerate(symbols, start=1):
        progress_note(f"starting {symbol} ({symbol_idx}/{len(symbols)})", enabled=progress_enabled)
        result, final, _ = evaluate_ensemble(
            symbol,
            train_end,
            valid_end,
            test_end,
            progress_enabled=progress_enabled,
            progress_update_every=progress_update_every,
        )
        results.append(result)
        finals.append(final)
        progress_note(f"finished {symbol} ({symbol_idx}/{len(symbols)})", enabled=progress_enabled)

    write_outputs(results, finals, args.accuracy_low, args.accuracy_high)
    print(build_report(results), end="")
    for r in results:
        print(f"{r.symbol}: latest {r.latest_forecast_signal} @ {r.latest_forecast_prob_up:.4f}, test acc {r.test_accuracy:.4f}")


if __name__ == "__main__":
    main()