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