sridattapradeep/India_Equity_Scanner
0
1"""2backtest.py — walk-forward screener backtest (Phase 12).3 4Replays the live screener rules over a multi-year OHLCV window and measures5what actually happened next, with NO lookahead:6 7 • Momentum — on each day a stock NEWLY enters the Minervini template8 (incl. a point-in-time cross-sectional RS rank), measure9 forward 5/10/20/60-session returns vs NIFTY. Plus a daily-10 rebalanced equal-weight equity curve of all template members.11 • Swing — replays the exact live setup (entry = signal-day close,12 SL = entry − 3.0×ATR14, T1 = entry + 1.5×risk; exit-tuned13 2026-06-10 via scripts/tune_swing_exits.py) bar by bar.14 If SL and T1 are both touched on the same bar, the STOP is15 assumed to fill first (conservative).16 • Reversal — forward returns from the day a stock first completes17 `reversal_qualify_days` consecutive closes above its 200 DMA18 (the day it would first appear on the live watchlist).19 20Point-in-time discipline:21 - All indicators on day t use data through t only (the live scanner is also22 close-based at EOD). Entries are at day-t close; outcomes start at t+1.23 - RS Rating is re-ranked cross-sectionally for EVERY day from each stock's24 own composite return — the same formula compute_rs_ratings() uses live.25 26Honest limitations (also surfaced in the API payload):27 - The universe is TODAY'S Nifty 500 — stocks that fell out of the index in28 the past are missing, which biases results upward (survivorship bias).29 - SMC and VCP are not replayed (their per-day detection is loop-heavy);30 this covers momentum / swing / reversal.31 32Like breadth, results are a pure function of the price window: recomputed33once per trading day in memory, so they survive Railway's ephemeral DB.34"""35from __future__ import annotations36 37import logging38from datetime import datetime, timezone39from typing import Optional40 41import numpy as np42import pandas as pd43 44logger = logging.getLogger(__name__)45 46HORIZONS = (5, 10, 20, 60) # forward sessions for the event studies47SWING_MAX_HOLD = 40 # bars before a swing trade times out48REVERSAL_QUALIFY_DAYS = 15 # consecutive closes above 200 DMA49 50 51# ── helpers ───────────────────────────────────────────────────────────────────52 53def _round(v, d=2) -> Optional[float]:54 try:55 f = float(v)56 return None if (np.isnan(f) or np.isinf(f)) else round(f, d)57 except (TypeError, ValueError):58 return None59 60 61def _wilder_atr(df: pd.DataFrame, period: int = 14) -> pd.Series:62 """Same ATR the live scanner uses (scanner._compute_indicators)."""63 high, low, close = df["high"], df["low"], df["close"]64 tr1 = high - low65 tr2 = (high - close.shift(1)).abs()66 tr3 = (low - close.shift(1)).abs()67 tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)68 return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()69 70 71def _consecutive_true(arr: np.ndarray) -> np.ndarray:72 """Length of the running True streak ending at each position."""73 n = len(arr)74 idx = np.arange(n)75 reset = np.where(~arr, idx, -1)76 last_reset = np.maximum.accumulate(reset)77 return np.where(arr, idx - last_reset, 0)78 79 80def _horizon_stats(signal_mask: pd.DataFrame, closes: pd.DataFrame,81 bench: pd.Series, horizons=HORIZONS) -> dict:82 """Forward absolute + NIFTY-excess return stats at each horizon, measured83 from the close of every True cell in signal_mask."""84 out = {}85 sig = signal_mask.fillna(False).to_numpy()86 for h in horizons:87 fwd = closes.shift(-h) / closes - 188 bench_fwd = bench.shift(-h) / bench - 189 excess = fwd.sub(bench_fwd, axis=0)90 91 abs_vals = fwd.to_numpy()[sig]92 exc_vals = excess.to_numpy()[sig]93 abs_vals = abs_vals[~np.isnan(abs_vals)]94 exc_vals = exc_vals[~np.isnan(exc_vals)]95 96 out[str(h)] = {97 "events": int(len(exc_vals)),98 "win_rate": _round(100 * np.mean(abs_vals > 0)) if len(abs_vals) else None,99 "win_rate_excess": _round(100 * np.mean(exc_vals > 0)) if len(exc_vals) else None,100 "avg_return": _round(100 * np.mean(abs_vals)) if len(abs_vals) else None,101 "avg_excess": _round(100 * np.mean(exc_vals)) if len(exc_vals) else None,102 "median_excess": _round(100 * np.median(exc_vals)) if len(exc_vals) else None,103 }104 return out105 106 107def _max_drawdown_pct(curve: pd.Series) -> Optional[float]:108 if curve.empty:109 return None110 return _round(100 * (curve / curve.cummax() - 1).min())111 112 113# ── shared panels ─────────────────────────────────────────────────────────────114 115def _template_panel(dfs: dict[str, pd.DataFrame], rs_min: float = 70.0) -> dict:116 """Point-in-time Minervini-template panels shared by the production117 backtest and the exit-tuning harness. Returns118 {closes, sma200, template, entries} (all date × symbol DataFrames)."""119 closes = pd.DataFrame({s: d["close"] for s, d in dfs.items()}).sort_index()120 121 sma50 = closes.rolling(50).mean()122 sma150 = closes.rolling(150).mean()123 sma200 = closes.rolling(200).mean()124 sma200_rising = sma200 > sma200.shift(20)125 h52 = closes.rolling(252).max()126 l52 = closes.rolling(252).min()127 128 composite = (129 closes.pct_change(63) * 0.40130 + closes.pct_change(126) * 0.20131 + closes.pct_change(189) * 0.20132 + closes.pct_change(252) * 0.20133 )134 # Cross-sectional percentile per day — same formula as compute_rs_ratings.135 # Stocks with an incomplete composite (NaN) are simply ineligible that day.136 rs = composite.rank(axis=1, pct=True) * 100137 138 template = (139 (closes > sma50) & (closes > sma150) & (closes > sma200)140 & (sma50 > sma150) & (sma150 > sma200)141 & sma200_rising142 & (closes >= 0.75 * h52)143 & (closes >= 1.30 * l52)144 & (rs >= rs_min)145 )146 entries = template & ~template.shift(1, fill_value=False)147 return {"closes": closes, "sma200": sma200, "rs": rs,148 "template": template, "entries": entries}149 150 151# ── swing trade simulation (parameterised — used for exit-rule tuning) ────────152 153def simulate_swing_trades(154 dfs: dict[str, pd.DataFrame],155 entries: pd.DataFrame,156 *,157 stop_atr: float = 1.5,158 target_r: Optional[float] = 2.5,159 max_hold: int = SWING_MAX_HOLD,160 breakeven_at_r: Optional[float] = None,161 trail_atr: Optional[float] = None,162 collect_trades: bool = False,163) -> dict:164 """165 Bar-by-bar replay of swing trades opened at the close of each True cell in166 `entries`.167 168 Exit rules (all measured against risk = stop_atr × ATR14 at entry):169 stop_atr initial stop = entry − stop_atr × ATR170 target_r fixed target = entry + target_r × risk (None = no target)171 breakeven_at_r once the high (from the NEXT bar on) reaches entry +172 N × risk, raise the stop to entry173 trail_atr chandelier trail: stop = max(stop, highest high since174 entry − trail_atr × ATR-at-entry), evaluated with highs175 through the PREVIOUS bar (no same-bar lookahead)176 max_hold exit at close after this many bars177 178 Same-bar conservatism: if a bar touches both the stop and the target, the179 STOP is assumed to fill first. R is always relative to the INITIAL risk.180 """181 closed_r: list[float] = []182 hold_bars: list[int] = []183 trade_log: list[dict] = []184 n_target = n_stop = n_timeout = n_open = 0185 186 for sym in entries.columns:187 df = dfs.get(sym)188 if df is None:189 continue190 col = entries[sym]191 ent_dates = col.index[col.fillna(False)]192 if len(ent_dates) == 0:193 continue194 atr = _wilder_atr(df)195 highs_a = df["high"].to_numpy(dtype=float)196 lows_a = df["low"].to_numpy(dtype=float)197 closes_a = df["close"].to_numpy(dtype=float)198 n = len(df)199 busy_until = -1 # skip overlapping signals while a trade is open200 201 for dt in ent_dates:202 try:203 p = df.index.get_loc(dt)204 except KeyError:205 continue206 if p <= busy_until:207 continue208 entry = closes_a[p]209 a = float(atr.iloc[p]) if not np.isnan(atr.iloc[p]) else 0.0210 if a <= 0 or entry <= 0:211 continue212 risk = stop_atr * a213 stop = entry - risk214 target = entry + target_r * risk if target_r is not None else None215 be_trigger = entry + breakeven_at_r * risk if breakeven_at_r is not None else None216 217 outcome = None218 exit_bar = None219 r_val = 0.0220 highest_high = -np.inf # highs AFTER entry only (no day-p high)221 last_bar = min(p + max_hold, n - 1)222 223 for j in range(p + 1, last_bar + 1):224 # Raise the stop using information through bar j-1 only.225 if trail_atr is not None and np.isfinite(highest_high):226 stop = max(stop, highest_high - trail_atr * a)227 if be_trigger is not None and highest_high >= be_trigger:228 stop = max(stop, entry)229 230 if lows_a[j] <= stop: # stop first — conservative231 outcome, exit_bar = "stop", j232 r_val = (stop - entry) / risk233 break234 if target is not None and highs_a[j] >= target:235 outcome, exit_bar = "target", j236 r_val = float(target_r)237 break238 highest_high = max(highest_high, highs_a[j])239 240 if outcome is None:241 if last_bar == p + max_hold:242 outcome, exit_bar = "timeout", last_bar243 r_val = (closes_a[last_bar] - entry) / risk244 else:245 n_open += 1 # ran out of data — still open246 busy_until = n # nothing later can be evaluated247 continue248 249 closed_r.append(r_val)250 if outcome == "target": n_target += 1251 elif outcome == "stop": n_stop += 1252 else: n_timeout += 1253 hold_bars.append(exit_bar - p)254 if collect_trades:255 trade_log.append({"symbol": sym, "entry_date": dt,256 "r": r_val, "hold": exit_bar - p,257 "outcome": outcome,258 "entry": entry, "stop": entry - risk,259 "target": target,260 "exit_date": df.index[exit_bar]})261 busy_until = exit_bar262 263 r = np.array(closed_r, dtype=float)264 n_closed = len(r)265 gross_win = float(r[r > 0].sum()) if n_closed else 0.0266 gross_loss = float(-r[r < 0].sum()) if n_closed else 0.0267 stats = {268 "trades_closed": n_closed,269 "trades_open": n_open,270 "win_rate": _round(100 * np.mean(r > 0)) if n_closed else None,271 "avg_r": _round(float(r.mean())) if n_closed else None,272 "total_r": _round(float(r.sum()), 1) if n_closed else None,273 "profit_factor": _round(gross_win / gross_loss) if gross_loss > 0 else None,274 "target_pct": _round(100 * n_target / n_closed) if n_closed else None,275 "stop_pct": _round(100 * n_stop / n_closed) if n_closed else None,276 "timeout_pct": _round(100 * n_timeout / n_closed) if n_closed else None,277 "avg_hold_bars": _round(float(np.mean(hold_bars)), 1) if hold_bars else None,278 }279 if collect_trades:280 stats["trades"] = trade_log281 return stats282 283 284# ── single-stock backtest (on-demand, no storage) ────────────────────────────285 286def backtest_single_stock(287 symbol: str,288 df: pd.DataFrame,289 horizons: tuple[int, ...] = HORIZONS,290 stop_atr: float = 3.0,291 target_r: float = 1.5,292 max_hold: int = SWING_MAX_HOLD,293 max_bars: int = 2500,294) -> dict:295 """296 Walk-forward backtest of ONE stock over up to `max_bars` (~10y) of its own297 history. Pure function of the price series — nothing is persisted.298 299 Template = the 8 PRICE criteria of the Minervini trend template. The RS300 percentile criterion is cross-sectional (needs the whole universe per day)301 and is excluded here, exactly as compute_breadth_history does — flagged in302 the payload so the UI can say so.303 304 Swing trades replay the LIVE exit rules (stop_atr/target_r mirror305 scanner.scan_swing_setups). The exposure curve compares "long only while306 the stock is in the template (decided at the prior close)" against buy &307 hold over the same window.308 """309 df = df.tail(max_bars)310 if len(df) < 300:311 return {"status": "insufficient_history", "bars": int(len(df))}312 313 close = df["close"]314 sma50 = close.rolling(50).mean()315 sma150 = close.rolling(150).mean()316 sma200 = close.rolling(200).mean()317 sma200_rising = sma200 > sma200.shift(20)318 h52 = close.rolling(252).max()319 l52 = close.rolling(252).min()320 321 template = (322 (close > sma50) & (close > sma150) & (close > sma200)323 & (sma50 > sma150) & (sma150 > sma200)324 & sma200_rising325 & (close >= 0.75 * h52)326 & (close >= 1.30 * l52)327 ).fillna(False)328 entries = template & ~template.shift(1, fill_value=False)329 330 # ── swing replay with the live exit rules ────────────────────────────────331 sim = simulate_swing_trades(332 {symbol: df}, entries.to_frame(symbol),333 stop_atr=stop_atr, target_r=target_r, max_hold=max_hold,334 collect_trades=True,335 )336 trade_log = sim.pop("trades", [])337 trades = [338 {339 "entry_date": pd.Timestamp(t["entry_date"]).strftime("%Y-%m-%d"),340 "exit_date": pd.Timestamp(t["exit_date"]).strftime("%Y-%m-%d"),341 "entry": _round(t["entry"]), "stop": _round(t["stop"]),342 "target": _round(t["target"]),343 "outcome": t["outcome"], "r": _round(t["r"]), "hold": int(t["hold"]),344 }345 for t in trade_log346 ]347 trades.reverse() # newest first for display348 349 # ── forward returns from each template entry (absolute — single stock) ───350 ent_mask = entries.to_numpy()351 horizon_stats = {}352 for h in horizons:353 fwd = (close.shift(-h) / close - 1).to_numpy()[ent_mask]354 fwd = fwd[~np.isnan(fwd)]355 horizon_stats[str(h)] = {356 "events": int(len(fwd)),357 "win_rate": _round(100 * np.mean(fwd > 0)) if len(fwd) else None,358 "avg_return": _round(100 * np.mean(fwd)) if len(fwd) else None,359 "median_return": _round(100 * np.median(fwd)) if len(fwd) else None,360 }361 362 # ── template-exposure curve vs buy & hold (weekly-sampled for payload) ───363 ret = close.pct_change().fillna(0.0)364 in_pos = template.shift(1, fill_value=False)365 strat = (1 + ret.where(in_pos, 0.0)).cumprod() * 100366 bh = (1 + ret).cumprod() * 100367 idx = list(range(0, len(strat), 5))368 if idx[-1] != len(strat) - 1:369 idx.append(len(strat) - 1)370 curve = [371 {"date": strat.index[i].strftime("%Y-%m-%d"),372 "strategy": _round(float(strat.iloc[i])),373 "benchmark": _round(float(bh.iloc[i]))}374 for i in idx375 ]376 377 return {378 "status": "ok",379 "symbol": symbol,380 "computed_at": datetime.now(timezone.utc).isoformat(),381 "data_from": df.index[0].strftime("%Y-%m-%d"),382 "data_through": df.index[-1].strftime("%Y-%m-%d"),383 "bars": int(len(df)),384 "pct_days_in_template": _round(100 * float(template.mean()), 1),385 "entry_signals": int(entries.sum()),386 "swing": {**sim, "rules": f"SL = entry − {stop_atr}×ATR14 · "387 f"T1 = entry + {target_r}×risk · timeout {max_hold} bars"},388 "horizons": horizon_stats,389 "exposure": {390 "strategy_return_pct": _round(float(strat.iloc[-1]) - 100),391 "benchmark_return_pct": _round(float(bh.iloc[-1]) - 100),392 "strategy_max_dd_pct": _max_drawdown_pct(strat),393 "benchmark_max_dd_pct": _max_drawdown_pct(bh),394 },395 "equity_curve": curve,396 "trades": trades[:60],397 "caveats": [398 "Single-stock template uses the 8 price criteria only — the RS "399 "percentile criterion needs the whole universe per day and is "400 "excluded (same approach as the breadth reconstruction).",401 "Returns are absolute (not NIFTY-excess) and pre-costs.",402 ],403 }404 405 406# ── core ──────────────────────────────────────────────────────────────────────407 408def run_backtest(409 dfs: dict[str, pd.DataFrame],410 benchmark_df: Optional[pd.DataFrame],411 horizons: tuple[int, ...] = HORIZONS,412 rs_min: float = 70.0,413 swing_max_hold: int = SWING_MAX_HOLD,414 reversal_qualify_days: int = REVERSAL_QUALIFY_DAYS,415) -> dict:416 """Replay the screeners over `dfs` (symbol → OHLCV with lowercase columns,417 DatetimeIndex). Returns a JSON-safe dict; {"status": "no_data"} when the418 inputs are unusable."""419 if not dfs:420 return {"status": "no_data"}421 422 panel = _template_panel(dfs, rs_min=rs_min)423 closes = panel["closes"]424 if closes.empty or len(closes) < 300:425 return {"status": "no_data"}426 427 # ── benchmark, aligned to the union calendar ─────────────────────────────428 bench = None429 if benchmark_df is not None and len(benchmark_df):430 b = benchmark_df.copy()431 b.columns = [c.lower() for c in b.columns]432 bench = b["close"].reindex(closes.index).ffill()433 if bench is None or bench.dropna().empty:434 # Degraded mode: equal-weight universe average stands in for NIFTY.435 bench = closes.mean(axis=1)436 437 # ── point-in-time template panel (shared with the tuning harness) ────────438 sma200 = panel["sma200"]439 rs = panel["rs"]440 template = panel["template"]441 entries = panel["entries"]442 443 # ── momentum: entry-event study ──────────────────────────────────────────444 momentum = {445 "signals": int(entries.to_numpy().sum()),446 "horizons": _horizon_stats(entries, closes, bench, horizons),447 }448 449 # ── momentum: equal-weight equity curve (membership decided at t-1 close) ─450 ret = closes.pct_change()451 members_lag = template.shift(1, fill_value=False)452 active = members_lag & ret.notna()453 port_ret = ret.where(active).mean(axis=1)454 member_count = active.sum(axis=1)455 456 has_members = member_count > 0457 equity_curve: list[dict] = []458 curve_stats: dict = {}459 if has_members.any():460 start = has_members.idxmax()461 pr = port_ret.loc[start:].fillna(0.0)462 strat = (1 + pr).cumprod() * 100463 br = bench.loc[start:].pct_change().fillna(0.0)464 bcurve = (1 + br).cumprod() * 100465 466 n_days = max(len(strat) - 1, 1)467 curve_stats = {468 "start": start.strftime("%Y-%m-%d"),469 "trading_days": int(n_days),470 "strategy_return_pct": _round(strat.iloc[-1] - 100),471 "benchmark_return_pct": _round(bcurve.iloc[-1] - 100),472 "strategy_ann_pct": _round(100 * ((strat.iloc[-1] / 100) ** (252 / n_days) - 1)),473 "benchmark_ann_pct": _round(100 * ((bcurve.iloc[-1] / 100) ** (252 / n_days) - 1)),474 "strategy_max_dd_pct": _max_drawdown_pct(strat),475 "benchmark_max_dd_pct": _max_drawdown_pct(bcurve),476 "avg_holdings": _round(float(member_count.loc[start:].mean()), 1),477 }478 equity_curve = [479 {480 "date": ts.strftime("%Y-%m-%d"),481 "strategy": _round(sv),482 "benchmark": _round(bv),483 }484 for ts, sv, bv in zip(strat.index, strat.values, bcurve.values)485 ]486 momentum["portfolio"] = curve_stats487 488 # ── swing: bar-by-bar trade replay on template entry days ────────────────489 # Parameters MUST mirror scanner.scan_swing_setups (exit-tuned 2026-06-10).490 swing = simulate_swing_trades(491 dfs, entries,492 stop_atr=3.0, target_r=1.5, max_hold=swing_max_hold,493 )494 swing["rules"] = (495 "entry = signal close · SL = entry − 3.0×ATR14 · T1 = entry + 1.5×risk · "496 f"timeout {swing_max_hold} bars · same-bar SL+T1 counts as stop"497 )498 499 # ── reversal: first day with N consecutive closes above 200 DMA ──────────500 above200 = (closes > sma200).fillna(False)501 qual = pd.DataFrame(502 {s: _consecutive_true(above200[s].to_numpy()) for s in above200.columns},503 index=above200.index,504 ) == reversal_qualify_days505 reversal = {506 "signals": int(qual.to_numpy().sum()),507 "qualify_days": reversal_qualify_days,508 "horizons": _horizon_stats(qual, closes, bench, horizons),509 }510 511 first_eval = rs.dropna(how="all").index512 return {513 "status": "ok",514 "computed_at": datetime.now(timezone.utc).isoformat(),515 "data_from": closes.index[0].strftime("%Y-%m-%d"),516 "data_through": closes.index[-1].strftime("%Y-%m-%d"),517 "eval_from": first_eval[0].strftime("%Y-%m-%d") if len(first_eval) else None,518 "symbols": int(closes.shape[1]),519 "horizons": list(horizons),520 "momentum": momentum,521 "swing": swing,522 "reversal": reversal,523 "equity_curve": equity_curve,524 "caveats": [525 "Backtest replays TODAY'S Nifty 500 constituents — past index "526 "dropouts are missing (survivorship bias inflates results).",527 "Signals use end-of-day closes only, matching the live scanner.",528 "SMC and VCP screeners are not replayed in this version.",529 ],530 }531