130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
import pandas as pd
|
|
import pandas_ta as ta
|
|
import yfinance as yf
|
|
from datetime import datetime, time, timedelta
|
|
import pytz
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class TouchTurnStrategy:
|
|
"""
|
|
Implements the 'Touch & Turn' (Opening Range Reversal) strategy.
|
|
|
|
1. Captures the 09:30-09:45 EST 15m candle.
|
|
2. Checks if range > 25% of 14-day ATR.
|
|
3. Bearish candle -> Long at Low, Target @ 38.2% Fib.
|
|
4. Bullish candle -> Short at High, Target @ 38.2% Fib.
|
|
"""
|
|
|
|
def __init__(self, ticker: str, risk_percent_atr: float = 25.0, rr_ratio: float = 2.0):
|
|
self.ticker = ticker
|
|
self.risk_percent_atr = risk_percent_atr
|
|
self.rr_ratio = rr_ratio
|
|
self.tz = pytz.timezone('US/Eastern')
|
|
|
|
self.valid_setup = False
|
|
self.direction = 0 # 1 for Short (from High), -1 for Long (from Low)
|
|
self.entry_price = 0.0
|
|
self.target_price = 0.0
|
|
self.stop_loss = 0.0
|
|
|
|
def get_market_data(self):
|
|
"""Fetches 15m candles and daily data for ATR."""
|
|
now = datetime.now(self.tz)
|
|
|
|
# 1. Fetch Daily Data for ATR (need at least 15+ days)
|
|
daily_data = yf.download(self.ticker, period="1mo", interval="1d", progress=False)
|
|
if daily_data.empty or len(daily_data) < 15:
|
|
logger.error(f"Insufficient daily data for {self.ticker}")
|
|
return None, None
|
|
|
|
# Clean column names (yfinance multi-index issue)
|
|
if isinstance(daily_data.columns, pd.MultiIndex):
|
|
daily_data.columns = daily_data.columns.droplevel(1)
|
|
|
|
# Calculate ATR
|
|
daily_data.ta.atr(length=14, append=True)
|
|
daily_atr = daily_data['ATRr_14'].iloc[-2] # Use yesterday's ATR
|
|
|
|
# 2. Fetch 15m Candle for today's opening (09:30 - 09:45)
|
|
# Note: yfinance 15m candles are labeled by start time.
|
|
start_date = now.strftime('%Y-%m-%d')
|
|
intraday_data = yf.download(self.ticker, start=start_date, interval="15m", progress=False)
|
|
|
|
if intraday_data.empty:
|
|
logger.warning(f"No intraday data yet for {self.ticker}")
|
|
return daily_atr, None
|
|
|
|
if isinstance(intraday_data.columns, pd.MultiIndex):
|
|
intraday_data.columns = intraday_data.columns.droplevel(1)
|
|
|
|
# The first candle of the session (09:30)
|
|
opening_candle = intraday_data.between_time('09:30', '09:30')
|
|
if opening_candle.empty:
|
|
logger.warning(f"Opening 15m candle (09:30) not yet available for {self.ticker}")
|
|
return daily_atr, None
|
|
|
|
return daily_atr, opening_candle.iloc[0]
|
|
|
|
def check_setup(self):
|
|
"""Evaluates the strategy criteria based on the opening candle."""
|
|
daily_atr, opening_candle = self.get_market_data()
|
|
|
|
if daily_atr is None or opening_candle is None:
|
|
return False
|
|
|
|
high = opening_candle['High']
|
|
low = opening_candle['Low']
|
|
open_p = opening_candle['Open']
|
|
close_p = opening_candle['Close']
|
|
range_size = high - low
|
|
|
|
# 1. Liquidity Filter
|
|
if range_size < (daily_atr * self.risk_percent_atr / 100):
|
|
logger.info(f"Setup invalid: Range ({range_size:.2f}) < 25% of ATR ({daily_atr:.2f})")
|
|
self.valid_setup = False
|
|
return False
|
|
|
|
# 2. Determine Direction
|
|
if close_p < open_p:
|
|
self.direction = -1 # Bearish candle -> Long setup
|
|
self.entry_price = low
|
|
logger.info(f"Bearish opening candle detected. Preparing LONG at {self.entry_price:.2f}")
|
|
else:
|
|
self.direction = 1 # Bullish candle -> Short setup
|
|
self.entry_price = high
|
|
logger.info(f"Bullish opening candle detected. Preparing SHORT at {self.entry_price:.2f}")
|
|
|
|
# 3. Calculate Fibonacci 38.2% Target
|
|
# For LONG (from Low): target is 38.2% up from the Low
|
|
# For SHORT (from High): target is 38.2% down from the High
|
|
if self.direction == -1: # LONG
|
|
self.target_price = low + (range_size * 0.382)
|
|
target_distance = self.target_price - self.entry_price
|
|
stop_distance = target_distance / self.rr_ratio
|
|
self.stop_loss = self.entry_price - stop_distance
|
|
else: # SHORT
|
|
self.target_price = high - (range_size * 0.382)
|
|
target_distance = self.entry_price - self.target_price
|
|
stop_distance = target_distance / self.rr_ratio
|
|
self.stop_loss = self.entry_price + stop_distance
|
|
|
|
|
|
self.valid_setup = True
|
|
logger.info(f"Valid Setup! Entry: {self.entry_price:.2f}, Target: {self.target_price:.2f}, SL: {self.stop_loss:.2f}")
|
|
return True
|
|
|
|
def get_trade_params(self):
|
|
"""Returns the parameters for the trade execution."""
|
|
if not self.valid_setup:
|
|
return None
|
|
|
|
return {
|
|
"ticker": self.ticker,
|
|
"direction": "BUY" if self.direction == -1 else "SELL",
|
|
"entry_price": self.entry_price,
|
|
"target_price": self.target_price,
|
|
"stop_loss": self.stop_loss
|
|
}
|