#!/usr/bin/env python3 """ research/new_features_backtest.py — Backtest impact of feature modules. Tests 2 hypotheses against historical NSE data (2024-01-01 to 2025-06-01): H1 — Fundamentals filter: Do fundamental_score >= 60 trades outperform baseline? H2 — Sector rotation: Do trades in 'leading_sectors' beat trades in 'lagging_sectors'? Method: - Fetch OHLCV + indicators for each stock at each test date - Compute 3D and 5D forward returns (close-to-close) - Classify each date with fundamentals score and sector position - Compare win rates and avg returns across filtered vs unfiltered populations - Output: plain text table (no LLM calls — this is a pure signal test) Usage: cd /Users/videkhanna/Documents/Projects/PaperTrade python research/new_features_backtest.py """ from __future__ import annotations import sys, os sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import warnings warnings.filterwarnings("ignore") import math from datetime import datetime, timedelta from typing import Optional import numpy as np import pandas as pd import yfinance as yf # ── CONFIG ───────────────────────────────────────────────────────────────────── UNIVERSE = [ "RELIANCE.NS", "TCS.NS", "HDFCBANK.NS", "BAJFINANCE.NS", "SUNPHARMA.NS", "WIPRO.NS", ] START = "2023-07-01" # pull data from mid-2023 to cover 2024 test dates END = "2025-06-01" TEST_START = "2024-01-01" TEST_END = "2025-06-01" STEP = 40 # every 40 trading days # ── DATA LOADING ─────────────────────────────────────────────────────────────── def load_prices(tickers: list[str]) -> dict[str, pd.Series]: print(f" Downloading OHLCV for {len(tickers)} stocks ({START} → {END})...") all_prices: dict[str, pd.Series] = {} for tk in tickers: try: df = yf.download(tk, start=START, end=END, progress=False, auto_adjust=True) if df.empty: print(f" [skip] {tk}: no data") continue close = df["Close"] if isinstance(close, pd.DataFrame): close = close.iloc[:, 0] all_prices[tk] = close.dropna() print(f" {tk}: {len(all_prices[tk])} bars") except Exception as e: print(f" [error] {tk}: {e}") return all_prices def get_test_dates(close: pd.Series) -> list[pd.Timestamp]: """Sample every STEP-th bar between TEST_START and TEST_END.""" idx = close.loc[TEST_START:TEST_END].index return list(idx[::STEP]) def forward_return(close: pd.Series, date: pd.Timestamp, n_days: int) -> Optional[float]: """Close-to-close return n_days forward from date.""" try: future_idx = close.index[close.index > date] if len(future_idx) < n_days: return None future_close = float(close.loc[future_idx[n_days - 1]]) current_close = float(close.loc[date]) return (future_close / current_close - 1) * 100 except Exception: return None # ── FEATURE: FUNDAMENTALS SCORE ──────────────────────────────────────────────── def batch_fundamentals(tickers: list[str]) -> dict[str, dict]: """Fetch fundamentals once per ticker (cached by fundamentals.py).""" print(" Fetching fundamentals for each ticker...") from fundamentals import get_fundamentals results = {} for tk in tickers: try: f = get_fundamentals(tk) results[tk] = f score = f.get("fundamental_score", "?") pe_lbl = f.get("pe_relative", "?") print(f" {tk}: score={score}/100 PE={pe_lbl}") except Exception as e: print(f" [warn] {tk}: {e}") results[tk] = {"fundamental_score": 50} return results # ── FEATURE: SECTOR PULSE ────────────────────────────────────────────────────── def get_sector_pulse_at(date: pd.Timestamp, sector_data: dict[str, pd.Series]) -> dict: """ Build a simplified 'leading/lagging' classification at a historical date. Uses pre-fetched sector index price series. """ leading, lagging = [], [] results = [] for name, series in sector_data.items(): try: past = series.loc[:date].dropna() if len(past) < 6: continue latest = float(past.iloc[-1]) p5 = float(past.iloc[-6]) if len(past) >= 6 else latest chg_5d = (latest / p5 - 1) * 100 if p5 > 0 else 0 results.append((name, chg_5d)) except Exception: continue if results: results.sort(key=lambda x: x[1], reverse=True) leading = [r[0] for r in results[:3]] lagging = [r[0] for r in results[-3:]] return {"leading_sectors": leading, "lagging_sectors": lagging} def load_sector_series() -> dict[str, pd.Series]: """Pre-fetch NSE sector index prices for historical backtesting.""" from sector_pulse import _SECTORS print(" Downloading NSE sector indices for backtesting...") out = {} for s in _SECTORS: try: df = yf.download(s["ticker"], start=START, end=END, progress=False, auto_adjust=True) if df.empty: continue close = df["Close"] if isinstance(close, pd.DataFrame): close = close.iloc[:, 0] out[s["name"]] = close.dropna() print(f" {s['name']} ({s['ticker']}): {len(out[s['name']])} bars") except Exception as e: print(f" [skip] {s['name']}: {e}") return out def get_ticker_sector(ticker: str) -> Optional[str]: from sector_pulse import get_sector_for_ticker return get_sector_for_ticker(ticker) # ── ANALYSIS ────────────────────────────────────────────────────────────────── def _stats(returns: list[float]) -> dict: if not returns: return {"n": 0, "win_rate": 0, "avg_ret": 0, "avg_win": 0, "avg_loss": 0} wins = [r for r in returns if r > 0] losses= [r for r in returns if r <= 0] return { "n": len(returns), "win_rate": round(len(wins) / len(returns) * 100, 1), "avg_ret": round(sum(returns) / len(returns), 2), "avg_win": round(sum(wins) / len(wins), 2) if wins else 0, "avg_loss": round(sum(losses) / len(losses), 2) if losses else 0, } def _print_comparison(label_a: str, a: dict, label_b: str, b: dict, tf: str) -> None: print(f"\n [{tf}] {label_a:35s} n={a['n']:>3} win={a['win_rate']:>5.1f}% avg={a['avg_ret']:>+5.2f}%") print(f" [{tf}] {label_b:35s} n={b['n']:>3} win={b['win_rate']:>5.1f}% avg={b['avg_ret']:>+5.2f}%") # ── MAIN ────────────────────────────────────────────────────────────────────── def run_backtest(): print("\n" + "═" * 70) print(" NEW FEATURES BACKTEST — PaperTrade") print(f" Universe: {', '.join(UNIVERSE)}") print(f" Dates: {TEST_START} → {TEST_END}, step={STEP} bars") print("═" * 70) # 1. Load price data prices = load_prices(UNIVERSE) if not prices: print("[error] No price data loaded — check internet connection") return # 2. Fundamentals (fetched once — time-stable) fund = batch_fundamentals(UNIVERSE) # 3. Sector series for historical pulse sector_series = load_sector_series() # 4. Build observation matrix records = [] for tk, close in prices.items(): dates = get_test_dates(close) tk_sector = get_ticker_sector(tk) f = fund.get(tk, {}) fund_score = f.get("fundamental_score", 50) for date in dates: r3 = forward_return(close, date, 3) r5 = forward_return(close, date, 5) if r3 is None and r5 is None: continue sector_ctx = get_sector_pulse_at(date, sector_series) if sector_series else {} leading = sector_ctx.get("leading_sectors", []) lagging = sector_ctx.get("lagging_sectors", []) is_leading = tk_sector in leading if tk_sector else None is_lagging = tk_sector in lagging if tk_sector else None records.append({ "ticker": tk, "date": date, "ret_3d": r3, "ret_5d": r5, "fund_score": fund_score, "is_leading": is_leading, "is_lagging": is_lagging, }) if not records: print("[error] No observation records built — check data") return print(f"\n Built {len(records)} observations across {len(prices)} tickers") # ── HYPOTHESIS TESTS ────────────────────────────────────────────────────── print("\n" + "═" * 70) print(" H1: FUNDAMENTALS FILTER (score >= 60)") print(" Do stocks with strong fundamentals outperform?") print("═" * 70) for tf, col in [("3D", "ret_3d"), ("5D", "ret_5d")]: all_r = [r[col] for r in records if r[col] is not None] strong = [r[col] for r in records if r[col] is not None and r["fund_score"] >= 60] weak = [r[col] for r in records if r[col] is not None and r["fund_score"] < 60] _print_comparison( "All tickers (baseline)", _stats(all_r), "Fundamental score >= 60", _stats(strong), tf ) if weak: print(f" [{tf}] {'Fundamental score < 60':35s} n={_stats(weak)['n']:>3} " f"win={_stats(weak)['win_rate']:>5.1f}% avg={_stats(weak)['avg_ret']:>+5.2f}%") print("\n" + "═" * 70) print(" H2: SECTOR ROTATION — LEADING vs LAGGING") print(" Do leading-sector trades outperform lagging-sector trades?") print("═" * 70) for tf, col in [("3D", "ret_3d"), ("5D", "ret_5d")]: all_r = [r[col] for r in records if r[col] is not None] lead_r = [r[col] for r in records if r[col] is not None and r["is_leading"] is True] lag_r = [r[col] for r in records if r[col] is not None and r["is_lagging"] is True] unmapped = sum(1 for r in records if r[col] is not None and r["is_leading"] is None) _print_comparison( "Leading sector trades", _stats(lead_r), "Lagging sector trades", _stats(lag_r), tf ) if all_r: print(f" [{tf}] {'Baseline (all)':35s} n={_stats(all_r)['n']:>3} " f"win={_stats(all_r)['win_rate']:>5.1f}% avg={_stats(all_r)['avg_ret']:>+5.2f}%") if unmapped > 0: print(f" [{tf}] ({unmapped} observations with unmapped sector — excluded from H2)") # ── COMBINED FILTER ─────────────────────────────────────────────────────── print("\n" + "═" * 70) print(" COMBINED: fundamentals >= 60 + leading sector") print("═" * 70) for tf, col in [("3D", "ret_3d"), ("5D", "ret_5d")]: all_r = [r[col] for r in records if r[col] is not None] combined = [ r[col] for r in records if r[col] is not None and r["fund_score"] >= 60 and r["is_leading"] is True ] _print_comparison( "All (baseline)", _stats(all_r), "Both filters active", _stats(combined), tf ) print("\n" + "═" * 70) print(" INTERPRETATION GUIDE") print(" win_rate > baseline win_rate → filter ADDS value (use it)") print(" avg_ret > baseline avg_ret → filter improves expected return") print(" n < 20 obs → insufficient data (interpret cautiously)") print("═" * 70) print(f"\n Completed: {datetime.now().strftime('%Y-%m-%d %H:%M')}\n") if __name__ == "__main__": run_backtest()