Files
trading-bot/src/strategy/scanner.py
T

84 lines
3.1 KiB
Python

import yfinance as yf
import pandas_ta as ta
import pandas as pd
import logging
from typing import List, Dict
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
DEFAULT_TICKERS = [
"AAPL", "MSFT", "NVDA", "AMZN", "META", "GOOGL", "TSLA",
"AMD", "NFLX", "QCOM", "INTC", "BA", "DIS", "SPY", "QQQ"
]
def scan_for_candidates(tickers: List[str] = DEFAULT_TICKERS, min_price: float = 20.0, min_volume: int = 2_000_000) -> pd.DataFrame:
"""
Scans a list of tickers to find the best candidates for the Touch & Turn strategy.
Prioritizes high Average True Range (ATR) as a percentage of price, ensuring adequate volume.
"""
logger.info(f"Scanning {len(tickers)} tickers for high volatility/liquidity...")
results = []
# Download daily data for the past 1mo to calculate 14-day ATR and Avg Volume
data = yf.download(tickers, period="1mo", interval="1d", group_by="ticker", progress=False)
for ticker in tickers:
try:
# Handle single ticker vs multi-ticker dataframe structure from yfinance
if len(tickers) == 1:
df = data.copy()
else:
df = data[ticker].copy()
if df.empty or len(df) < 15:
continue
# Clean column names (yfinance multi-index can sometimes leave tuple names)
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.droplevel(1)
df.ta.atr(length=14, append=True)
# Safely get the close price, falling back to yesterday if today's is NaN (common at exactly 09:30)
close_price = df['Close'].iloc[-1]
if pd.isna(close_price) and len(df) > 1:
close_price = df['Close'].iloc[-2]
yesterday_atr = df['ATRr_14'].iloc[-2]
# Safely get avg volume
avg_volume = df['Volume'].tail(14).mean()
if pd.isna(avg_volume):
avg_volume = 0
# Filters
if pd.isna(close_price) or close_price < min_price or avg_volume < min_volume or pd.isna(yesterday_atr):
continue
atr_percent = (yesterday_atr / close_price) * 100
results.append({
"Ticker": ticker,
"Close": round(float(close_price), 2),
"ATR_14": round(float(yesterday_atr), 2),
"ATR_Percent": round(float(atr_percent), 2),
"Avg_Volume": int(avg_volume)
})
except Exception as e:
logger.debug(f"Failed processing {ticker}: {e}")
results_df = pd.DataFrame(results)
if not results_df.empty:
# Sort by ATR Percentage descending (we want the most volatile stocks)
results_df = results_df.sort_values(by="ATR_Percent", ascending=False).reset_index(drop=True)
return results_df
if __name__ == "__main__":
candidates = scan_for_candidates()
print("\nTop Candidates for Touch & Turn Strategy:")
print("-" * 65)
print(candidates.to_string(index=False))