#!/usr/bin/env python3 """ research/entry_validation_backtest.py ────────────────────────────────────── Validates the entry-price open-buffer change against 7 years of NSE data. Questions answered ────────────────── Q1 Entry achievability — Do BUY-signal stocks actually gap UP at next open? Is the timeframe-aware buffer enough to model realistic entry? Q2 Target hit (strict) — Does price reach target_hi WITHIN the predicted TF? Reaching target on day-6 for a 5D prediction = FAIL. Q3 Time-to-target — For successful trades, how many days did it take? Distribution: same-day vs within-TF vs late vs never. Q4 Entry impact on R:R — Old entry (= close) vs new entry (TF buffer). Does the dynamic buffer meaningfully hurt hit rate? Q5 Failure decomposition — Why did target NOT get hit within TF? Gap-up miss at entry | stalled price | reversed. Data source: research/ai_prompt_accuracy_iter64.csv (7 years, 828 rows) + actual OHLCV with Open fetched from Yahoo Finance. Usage: python research/entry_validation_backtest.py python research/entry_validation_backtest.py --no-cache # re-download OHLCV """ from __future__ import annotations import sys, os, argparse, pickle sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) import warnings warnings.filterwarnings("ignore") import numpy as np import pandas as pd import yfinance as yf # ── CONFIG ──────────────────────────────────────────────────────────────────── CSV_PATH = os.path.join(os.path.dirname(__file__), "ai_prompt_accuracy_iter64.csv") CACHE_FILE = os.path.join(os.path.dirname(__file__), "cache", "ev_ohlcv_with_open.pkl") ENTRY_BUFFER_BY_TF = { "1D": 0.002, # 0.20% "3D": 0.003, # 0.30% "5D": 0.005, # 0.50% } SEP = "=" * 72 SEP2 = "─" * 72 TICKERS = [ "RELIANCE.NS", "TCS.NS", "HDFCBANK.NS", "BAJFINANCE.NS", "SUNPHARMA.NS", "WIPRO.NS", ] # ── OHLCV DOWNLOAD (includes Open) ─────────────────────────────────────────── def _download_ohlcv(tickers: list[str], no_cache: bool) -> dict[str, pd.DataFrame]: """Returns {ticker: DataFrame(Date, Open, High, Low, Close)}.""" os.makedirs(os.path.dirname(CACHE_FILE), exist_ok=True) if not no_cache and os.path.exists(CACHE_FILE): with open(CACHE_FILE, "rb") as f: print(" [cache] Loaded OHLCV with Open from disk.") return pickle.load(f) print(f" Downloading {len(tickers)} tickers 2017–2025 from Yahoo Finance…") raw = yf.download(tickers, start="2017-01-01", end="2025-06-01", auto_adjust=True, progress=True) result = {} for tk in tickers: try: df = pd.DataFrame({ "Open": raw["Open"][tk], "High": raw["High"][tk], "Low": raw["Low"][tk], "Close": raw["Close"][tk], }).dropna() df.index = pd.to_datetime(df.index) result[tk] = df except Exception as e: print(f" WARNING: {tk} failed — {e}") with open(CACHE_FILE, "wb") as f: pickle.dump(result, f, protocol=pickle.HIGHEST_PROTOCOL) print(f" Saved OHLCV cache → {CACHE_FILE}") return result # ── PER-ROW ANALYSIS ───────────────────────────────────────────────────────── TF_DAYS = {"1D": 1, "3D": 3, "5D": 5} def _analyse_row(row: pd.Series, ohlcv: dict[str, pd.DataFrame]) -> dict | None: ticker = row["ticker"] tf = str(row["timeframe"]) n_days = TF_DAYS.get(tf) if n_days is None: return None df = ohlcv.get(ticker) if df is None or df.empty: return None # Find signal date in OHLCV index sig_date = pd.to_datetime(row["date"]) future = df[df.index > sig_date].head(n_days + 1) # +1 for entry day check if len(future) < 2: return None # not enough forward data entry_day = future.iloc[0] # the NEXT trading day (entry day) close_price = float(row["entry_price"]) # previous day's close (signal bar) target_hi = float(row["target_price_hi"]) target_lo = float(row["target_price_lo"]) direction = str(row["direction"]) # BULLISH / BEARISH / NEUTRAL # ── Entry prices ───────────────────────────────────────────────────────── entry_old = close_price # old behaviour: last close buf = ENTRY_BUFFER_BY_TF.get(tf, 0.003) if direction in ("BULLISH", "BUY", "STRONG BUY"): entry_new = round(close_price * (1 + buf), 4) elif direction in ("BEARISH", "SELL"): entry_new = round(close_price * (1 - buf), 4) else: entry_new = close_price next_open = float(entry_day["Open"]) gap_pct = (next_open - close_price) / close_price * 100 # Was entry achievable at open? if direction in ("BULLISH", "BUY", "STRONG BUY"): entry_old_filled = next_open <= entry_old entry_new_filled = next_open <= entry_new elif direction in ("BEARISH", "SELL"): entry_old_filled = next_open >= entry_old entry_new_filled = next_open >= entry_new else: entry_old_filled = False entry_new_filled = False # ── Target prices (both anchored to close, same as production) ─────────── ret_hi_pct = (target_hi / close_price - 1) * 100 ret_lo_pct = (target_lo / close_price - 1) * 100 target_hi_old = target_hi target_hi_new = target_hi # ── All highs during holding period (entry day + holding days) ──────────── all_highs = list(future["High"]) # day1..dayN all_lows = list(future["Low"]) all_dates = list(future.index) # Days-to-target: first day high touches target_hi_new (within TF) days_to_target_old = None days_to_target_new = None for i, (h, d) in enumerate(zip(all_highs, all_dates)): day_num = i + 1 # 1-indexed if direction in ("BULLISH", "BUY", "STRONG BUY"): if days_to_target_old is None and h >= target_hi_old: days_to_target_old = day_num if days_to_target_new is None and h >= target_hi_new: days_to_target_new = day_num elif direction in ("BEARISH", "SELL"): # For BEARISH: target is the lo (price goes down) if days_to_target_old is None and all_lows[i] <= target_lo: days_to_target_old = day_num if days_to_target_new is None and all_lows[i] <= target_lo: days_to_target_new = day_num # ── Target hit within TF ───────────────────────────────────────────────── hit_old_within_tf = days_to_target_old is not None and days_to_target_old <= n_days hit_new_within_tf = days_to_target_new is not None and days_to_target_new <= n_days # ── Actual P&L at end of TF (close of last holding day) ────────────────── exit_close = float(future.iloc[-1]["Close"]) pnl_old_pct = (exit_close - entry_old) / entry_old * 100 if direction in ("BEARISH", "SELL"): pnl_new_pct = (entry_new - exit_close) / entry_new * 100 else: pnl_new_pct = (exit_close - entry_new) / entry_new * 100 # ── Failure mode classification ─────────────────────────────────────────── # Only for BULLISH predictions that missed target_new within TF failure_mode = "N/A" if direction in ("BULLISH", "BUY", "STRONG BUY") and not hit_new_within_tf: if not entry_new_filled: failure_mode = "entry_gap_miss" # stock gapped past entry elif max(all_highs) >= target_hi_new: failure_mode = "hit_but_late" # hit target after TF expired elif pnl_new_pct > 0: failure_mode = "stalled_short" # moved right direction but not enough elif pnl_new_pct < -1: failure_mode = "reversed" # went the wrong way else: failure_mode = "flat" # barely moved return { "ticker": ticker, "date": sig_date, "tf": tf, "n_days": n_days, "direction": direction, "close_price": close_price, "next_open": next_open, "gap_pct": round(gap_pct, 3), "entry_old": entry_old, "entry_new": entry_new, "entry_old_filled": entry_old_filled, "entry_new_filled": entry_new_filled, "target_hi_old": round(target_hi_old, 3), "target_hi_new": round(target_hi_new, 3), "ret_hi_pct": round(ret_hi_pct, 3), "days_to_target_old": days_to_target_old, "days_to_target_new": days_to_target_new, "hit_old_within_tf": hit_old_within_tf, "hit_new_within_tf": hit_new_within_tf, "pnl_old_pct": round(pnl_old_pct, 3), "pnl_new_pct": round(pnl_new_pct, 3), "failure_mode": failure_mode, "max_up_for_tf": float(row.get("max_up_for_tf", 0)), "min_down_for_tf": float(row.get("min_down_for_tf", 0)), "confidence": row.get("confidence", ""), } # ── REPORT PRINTER ──────────────────────────────────────────────────────────── def _pct(n: int, d: int) -> str: return f"{n/d*100:.1f}%" if d else "N/A" def print_report(df: pd.DataFrame) -> None: bull = df[df["direction"].isin(["BULLISH", "BUY", "STRONG BUY"])] bear = df[df["direction"].isin(["BEARISH", "SELL"])] print(f"\n{SEP}") print(" ENTRY PRICE VALIDATION BACKTEST") print(f" Data: {df['date'].min().date()} → {df['date'].max().date()} " f"| {len(df)} predictions | 6 NSE stocks | 3 timeframes") print(SEP) # ── Q1: Gap-up behavior at next open ───────────────────────────────────── print(f"\n{'Q1 ENTRY ACHIEVABILITY — Gap at Next Open':^72}") print(SEP2) print(f"{'Metric':<45} {'BULLISH':>12} {'BEARISH':>12}") print(SEP2) for label, subset in [("BULLISH", bull), ("BEARISH", bear)]: gaps = subset["gap_pct"] if not len(gaps): continue print(f" Avg gap next-open vs close ({label:<8}) {gaps.mean():>+.3f}%") print(f" Median gap {gaps.median():>+.3f}%") print(f" % that gap UP > 0 {_pct((gaps > 0).sum(), len(gaps))}") print(f" % that gap > 0.3% (misses buffer) {_pct((gaps > 0.3).sum(), len(gaps))}") print(f" % that gap > 0.5% {_pct((gaps > 0.5).sum(), len(gaps))}") print(f" % that gap > 1.0% {_pct((gaps > 1.0).sum(), len(gaps))}") print() print(f" Entry fill rate — old (at close) : {_pct(bull['entry_old_filled'].sum(), len(bull))}") print(f" Entry fill rate — new (TF policy): {_pct(bull['entry_new_filled'].sum(), len(bull))}") print(f" Improvement in fill rate : " f"+{(bull['entry_new_filled'].mean() - bull['entry_old_filled'].mean())*100:.1f}pp") # ── Q2: Strict TF target hit rate ──────────────────────────────────────── print(f"\n{SEP2}") print(f"{'Q2 STRICT TARGET HIT — within predicted timeframe':^72}") print(SEP2) print(f"{'TF':<6} {'N':>5} {'OldHit%':>9} {'NewHit%':>9} {'Delta':>8} {'AvgRetHi%':>10}") print(SEP2) for tf in ["1D", "3D", "5D"]: sub = bull[bull["tf"] == tf] if sub.empty: continue old_hit = sub["hit_old_within_tf"].mean() * 100 new_hit = sub["hit_new_within_tf"].mean() * 100 avg_ret = sub["ret_hi_pct"].mean() print(f" {tf:<4} {len(sub):>5} {old_hit:>8.1f}% {new_hit:>8.1f}% " f"{new_hit-old_hit:>+7.1f}pp {avg_ret:>9.2f}%") print() print(f" Key insight: Target hit rate should stay the same old vs new.") print(f" Entry policy changes fill quality and P&L, not market target reachability.") # ── Q3: Time-to-target distribution ────────────────────────────────────── print(f"\n{SEP2}") print(f"{'Q3 TIME TO TARGET (BULLISH predictions that hit)':^72}") print(SEP2) print(f"{'TF':<6} {'Hit same-day':>14} {'Hit day 2':>11} {'Hit day 3':>11} " f"{'Hit day 4-5':>12} {'Never/Late':>11}") print(SEP2) for tf in ["1D", "3D", "5D"]: n_days = TF_DAYS[tf] sub = bull[bull["tf"] == tf].copy() total = len(sub) d = sub["days_to_target_new"].copy() day1 = (d == 1).sum() day2 = (d == 2).sum() day3 = (d == 3).sum() day45 = ((d >= 4) & (d <= 5)).sum() late = total - day1 - day2 - day3 - day45 def pp(n): return f"{_pct(n,total):>10}" print(f" {tf:<4} {pp(day1)} {pp(day2)} {pp(day3)} {pp(day45)} {pp(late)}") # Average days-to-target for successful trades print() hits = bull[bull["hit_new_within_tf"]] if len(hits): avg_days = hits["days_to_target_new"].mean() med_days = hits["days_to_target_new"].median() print(f" Avg days-to-target (successful trades): {avg_days:.1f} days") print(f" Median days-to-target : {med_days:.0f} days") # Prediction accuracy by TF: reached target exactly on the last day vs early print() for tf in ["1D", "3D", "5D"]: n_days = TF_DAYS[tf] sub = bull[bull["tf"] == tf] hits_ = sub[sub["hit_new_within_tf"]] on_last_day = (hits_["days_to_target_new"] == n_days).sum() early = (hits_["days_to_target_new"] < n_days).sum() print(f" {tf}: {len(hits_)} hits — " f"{_pct(early, len(hits_))} hit EARLY (before TF end), " f"{_pct(on_last_day, len(hits_))} hit ON the final day") # ── Q4: P&L impact of entry buffer ─────────────────────────────────────── print(f"\n{SEP2}") print(f"{'Q4 R:R IMPACT — old entry (close) vs new entry (TF buffer)':^72}") print(SEP2) print(f"{'TF':<6} {'Avg P&L old':>13} {'Avg P&L new':>13} {'Delta':>8} " f"{'Win% old':>10} {'Win% new':>10}") print(SEP2) for tf in ["1D", "3D", "5D"]: sub = bull[bull["tf"] == tf] if sub.empty: continue pnl_o = sub["pnl_old_pct"].mean() pnl_n = sub["pnl_new_pct"].mean() win_o = (sub["pnl_old_pct"] > 0).mean() * 100 win_n = (sub["pnl_new_pct"] > 0).mean() * 100 print(f" {tf:<4} {pnl_o:>+12.2f}% {pnl_n:>+12.2f}% {pnl_n-pnl_o:>+7.3f}pp " f"{win_o:>9.1f}% {win_n:>9.1f}%") # ── Q5: Failure decomposition ───────────────────────────────────────────── print(f"\n{SEP2}") print(f"{'Q5 FAILURE MODE — why BULLISH trades missed target within TF':^72}") print(SEP2) misses = bull[~bull["hit_new_within_tf"]] modes = misses["failure_mode"].value_counts() total_misses = len(misses) print(f" Total missed: {total_misses} / {len(bull)} BULLISH predictions") print() labels = { "entry_gap_miss": "Gap-up past entry (unfillable at open)", "hit_but_late": "Price DID hit target but AFTER TF expired", "stalled_short": "Moved right direction, fell short of target", "reversed": "Price reversed (down > -1%)", "flat": "Price barely moved (±1%)", } for mode, count in modes.items(): desc = labels.get(mode, mode) print(f" {_pct(count, total_misses):>6} {count:>4} {desc}") # Sub-breakdown: "hit_but_late" — how late? late_hits = misses[misses["failure_mode"] == "hit_but_late"] if len(late_hits): print() print(f" Of the {len(late_hits)} 'hit but late' cases:") for tf in ["1D", "3D", "5D"]: sub_late = late_hits[late_hits["tf"] == tf] if len(sub_late): med_d = sub_late["days_to_target_new"].median() print(f" {tf}: {len(sub_late)} trades — median {med_d:.0f} days to target " f"(vs {TF_DAYS[tf]}-day window)") # ── SUMMARY TABLE ───────────────────────────────────────────────────────── print(f"\n{SEP}") print(f"{'SUMMARY':^72}") print(SEP) print() print(f" OLD SYSTEM (entry = last close): " f"entry fills {_pct(bull['entry_old_filled'].sum(), len(bull))} of the time") print(f" NEW SYSTEM (entry = close ± TF buffer): " f"entry fills {_pct(bull['entry_new_filled'].sum(), len(bull))} of the time") print() for tf in ["1D", "3D", "5D"]: sub = bull[bull["tf"] == tf] old_hit = sub["hit_old_within_tf"].mean() * 100 new_hit = sub["hit_new_within_tf"].mean() * 100 print(f" {tf} strict hit rate: old={old_hit:.1f}% new={new_hit:.1f}% " f"delta={new_hit-old_hit:+.1f}pp") late_pct = _pct( bull[bull["failure_mode"] == "hit_but_late"].shape[0], len(bull[~bull["hit_new_within_tf"]]) ) print() print(f" {late_pct} of all misses DID eventually hit target — just not within TF") print(f" → These are 'false failures' where timeframe was too tight") print() gap_miss_pct = _pct( bull[bull["failure_mode"] == "entry_gap_miss"].shape[0], len(bull) ) print(f" {gap_miss_pct} of BULLISH trades had an unfillable gap-up at open") print(f" → TF-aware buffer absorbs many of these; beyond policy threshold = skip") print(SEP) # ── MAIN ────────────────────────────────────────────────────────────────────── def main(): parser = argparse.ArgumentParser() parser.add_argument("--no-cache", action="store_true", help="Re-download OHLCV") parser.add_argument("--csv", default=CSV_PATH, help="Input CSV path") parser.add_argument("--save", default="", help="Save results to CSV path") args = parser.parse_args() print(f"\n{SEP}") print(" Loading prediction CSV…") df = pd.read_csv(args.csv) df["date"] = pd.to_datetime(df["date"]) print(f" {len(df)} rows loaded from {os.path.basename(args.csv)}") print("\n Loading OHLCV with Open prices…") ohlcv = _download_ohlcv(TICKERS, no_cache=args.no_cache) print("\n Running per-row analysis…") results = [] for _, row in df.iterrows(): rec = _analyse_row(row, ohlcv) if rec: results.append(rec) out = pd.DataFrame(results) print(f" Analysed {len(out)} rows (skipped {len(df) - len(out)} — insufficient fwd data)") if args.save: out.to_csv(args.save, index=False) print(f" Results saved → {args.save}") print_report(out) if __name__ == "__main__": main()