292 lines
11 KiB
Python
292 lines
11 KiB
Python
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.")
|
|
|