import yfinance as yf import pandas as pd import pandas_ta as ta from datetime import datetime, time, timedelta import pytz import logging from prettytable import PrettyTable logging.basicConfig(level=logging.WARNING) # Set to INFO for detailed trade logs logger = logging.getLogger(__name__) def backtest_ticker(ticker: str, days: int = 59, risk_percent_atr: float = 25.0, rr_ratio: float = 2.0, quiet: bool = False): """ Backtests the Touch & Turn strategy on a single ticker over the maximum available 15m history. """ if not quiet: print(f"\nRunning Backtest for {ticker} over {days} days...") tz = pytz.timezone('US/Eastern') # Fetch Daily data (Need roughly days + 30 for 14-day ATR buffer) daily_data = yf.download(ticker, period="6mo", interval="1d", progress=False) if daily_data.empty: if not quiet: print(f"Failed to fetch daily data for {ticker}") return None # Clean up multi-index columns if present if isinstance(daily_data.columns, pd.MultiIndex): daily_data.columns = daily_data.columns.droplevel(1) daily_data.ta.atr(length=14, append=True) daily_data.dropna(subset=['ATRr_14'], inplace=True) # Fetch 15m intraday data intraday_data = yf.download(ticker, period=f"{days}d", interval="15m", progress=False) if intraday_data.empty: if not quiet: print(f"Failed to fetch 15m data for {ticker}") return None if isinstance(intraday_data.columns, pd.MultiIndex): intraday_data.columns = intraday_data.columns.droplevel(1) # Convert index to US/Eastern timezone to reliably filter by time if intraday_data.index.tz is None: intraday_data.index = intraday_data.index.tz_localize('UTC').tz_convert(tz) else: intraday_data.index = intraday_data.index.tz_convert(tz) unique_dates = pd.Series(intraday_data.index.date).unique() # Metrics total_trades = 0 wins = 0 losses = 0 time_exits = 0 total_pnl_r = 0.0 # PnL measured in R multiples trade_log = [] for trade_date in unique_dates: # 1. Get Yesterday's ATR try: # Get daily data up to the day before `trade_date` prior_daily = daily_data[daily_data.index.date < trade_date] if prior_daily.empty: continue yesterday_atr = prior_daily['ATRr_14'].iloc[-1] except IndexError: continue # 2. Get today's 15m data day_data = intraday_data[intraday_data.index.date == trade_date] # 3. Find 09:30 candle opening_candles = day_data.between_time('09:30', '09:30') if opening_candles.empty: continue open_c = opening_candles.iloc[0] high, low, open_p, close_p = open_c['High'], open_c['Low'], open_c['Open'], open_c['Close'] range_size = high - low # 4. Filter for Liquidity/Volatility if range_size < (yesterday_atr * risk_percent_atr / 100): continue # Market not volatile enough today # 5. Determine Setup direction = -1 if close_p < open_p else 1 # -1 for Long, 1 for Short if direction == -1: # LONG setup (from Bearish open) entry_price = low target_price = low + (range_size * 0.382) stop_distance = (target_price - entry_price) / rr_ratio stop_loss = entry_price - stop_distance trade_dir = "LONG" else: # SHORT setup (from Bullish open) entry_price = high target_price = high - (range_size * 0.382) stop_distance = (entry_price - target_price) / rr_ratio stop_loss = entry_price + stop_distance trade_dir = "SHORT" # 6. Simulate Execution Window (09:45 to 10:45 inclusive - since 11:00 is forced exit) sim_data = day_data.between_time('09:45', '10:45') if sim_data.empty: continue position_open = False trade_result = None pnl_r = 0.0 exit_price = 0.0 exit_time = None exit_reason = "" for timestamp, candle in sim_data.iterrows(): c_high, c_low, c_close = candle['High'], candle['Low'], candle['Close'] # Check for Entry if not position_open: if (trade_dir == "LONG" and c_low <= entry_price) or (trade_dir == "SHORT" and c_high >= entry_price): position_open = True logger.info(f"{trade_date} - FILLED {trade_dir} at {entry_price:.2f}") if trade_dir == "LONG": if c_low <= stop_loss: trade_result = "LOSS" exit_price = stop_loss exit_time = timestamp exit_reason = "SL Hit" pnl_r = -1.0 position_open = False elif c_high >= target_price: trade_result = "WIN" exit_price = target_price exit_time = timestamp exit_reason = "TP Hit" pnl_r = rr_ratio position_open = False else: # SHORT if c_high >= stop_loss: trade_result = "LOSS" exit_price = stop_loss exit_time = timestamp exit_reason = "SL Hit" pnl_r = -1.0 position_open = False elif c_low <= target_price: trade_result = "WIN" exit_price = target_price exit_time = timestamp exit_reason = "TP Hit" pnl_r = rr_ratio position_open = False continue # If position is already open, check SL/TP for current candle if position_open: if trade_dir == "LONG": if c_low <= stop_loss: trade_result = "LOSS" exit_price = stop_loss exit_time = timestamp exit_reason = "SL Hit" pnl_r = -1.0 position_open = False elif c_high >= target_price: trade_result = "WIN" exit_price = target_price exit_time = timestamp exit_reason = "TP Hit" pnl_r = rr_ratio position_open = False else: # SHORT if c_high >= stop_loss: trade_result = "LOSS" exit_price = stop_loss exit_time = timestamp exit_reason = "SL Hit" pnl_r = -1.0 position_open = False elif c_low <= target_price: trade_result = "WIN" exit_price = target_price exit_time = timestamp exit_reason = "TP Hit" pnl_r = rr_ratio position_open = False if not position_open: break # 7. Force Exit at 11:00 (Close of 10:45 candle) if position_open: last_candle = sim_data.iloc[-1] exit_price = last_candle['Close'] exit_time = sim_data.index[-1] exit_reason = "11:00 Time Exit" if trade_dir == "LONG": distance = exit_price - entry_price pnl_r = distance / stop_distance else: distance = entry_price - exit_price pnl_r = distance / stop_distance if pnl_r > 0: trade_result = "WIN (Time)" else: trade_result = "LOSS (Time)" position_open = False # 8. Record Trade if trade_result is not None: total_trades += 1 if "WIN" in trade_result: wins += 1 else: losses += 1 if "Time" in trade_result: time_exits += 1 total_pnl_r += pnl_r trade_log.append([ trade_date.strftime("%Y-%m-%d"), trade_dir, f"{entry_price:.2f}", f"{target_price:.2f}", f"{stop_loss:.2f}", trade_result, f"{exit_price:.2f}", f"{pnl_r:.2f} R" ]) win_rate = (wins / total_trades) * 100 if total_trades > 0 else 0 if not quiet and total_trades > 0: table = PrettyTable() table.field_names = ["Date", "Dir", "Entry", "TP", "SL", "Result", "Exit Px", "PnL (R)"] for row in trade_log: table.add_row(row) print(table) print(f"\n--- Backtest Results: {ticker} ---") print(f"Total Setups Triggered: {total_trades}") print(f"Wins: {wins} | Losses: {losses}") print(f"Win Rate: {win_rate:.2f}%") print(f"Time Exits (11:00 EST): {time_exits}") print(f"Net PnL: {total_pnl_r:.2f} R") elif not quiet: print(f"No valid setups triggered for {ticker} in the last {days} days.") return { "Ticker": ticker, "Trades": total_trades, "Wins": wins, "Losses": losses, "Win Rate (%)": round(win_rate, 2), "Time Exits": time_exits, "Net PnL (R)": round(total_pnl_r, 2) } if __name__ == "__main__": import os if os.path.exists("isa_watchlist.csv"): print("Reading tickers from isa_watchlist.csv...") df = pd.read_csv("isa_watchlist.csv") tickers_to_test = df['Ticker'].tolist() else: print("isa_watchlist.csv not found, using default list...") tickers_to_test = ["TSLA", "NVDA", "NFLX", "AMD"] results = [] print(f"Running batch backtest for {len(tickers_to_test)} tickers. This may take a minute...") for t in tickers_to_test: res = backtest_ticker(t, quiet=True) if res and res["Trades"] > 0: results.append(res) if results: results_df = pd.DataFrame(results) results_df = results_df.sort_values(by="Net PnL (R)", ascending=False).reset_index(drop=True) print("\n" + "="*80) print("🚀 BACKTEST LEADERBOARD (LAST ~60 DAYS) 🚀") print("="*80) table = PrettyTable() table.field_names = results_df.columns for _, row in results_df.iterrows(): table.add_row(row.tolist()) print(table) print("\n* PnL is measured in 'R' (Risk Multiples). +2.0 R means you made twice what you risked.") else: print("No trades generated for any tickers.")