feat: Scan entire date range for signals in ATR EMA strategy

This commit is contained in:
Bobby (aider) 2025-02-08 19:52:55 -08:00
parent 0ca47d3c3d
commit 7e046a1e33

View File

@ -7,33 +7,36 @@ from utils.data_utils import get_stock_data, validate_signal_date, print_signal,
from screener.user_input import get_interval_choice, get_date_range from screener.user_input import get_interval_choice, get_date_range
from indicators.three_atr_ema import ThreeATREMAIndicator from indicators.three_atr_ema import ThreeATREMAIndicator
def check_entry_signal(df: pd.DataFrame) -> tuple: def check_entry_signal(df: pd.DataFrame) -> list:
""" """
Check for entry signal based on Three ATR EMA strategy Check for entry signals based on Three ATR EMA strategy throughout the date range
Args: Args:
df (pd.DataFrame): DataFrame with price data df (pd.DataFrame): DataFrame with price data
Returns: Returns:
tuple: (bool, datetime, dict) Entry signal, signal date, and signal data list: List of tuples (signal, date, signal_data) for each signal found
""" """
if len(df) < 2: # Need at least 2 bars for comparison if len(df) < 2: # Need at least 2 bars for comparison
return False, None, None return []
indicator = ThreeATREMAIndicator() indicator = ThreeATREMAIndicator()
results = indicator.calculate(df) results = indicator.calculate(df)
if len(results) < 2: if len(results) < 2:
return False, None, None return []
# Get latest values signals = []
current = df.iloc[-1]
previous = df.iloc[-2] # Start from index 1 to compare with previous
for i in range(1, len(df)):
current = df.iloc[i]
previous = df.iloc[i-1]
# Get indicator values # Get indicator values
ema = results['ema'].iloc[-1] ema = results['ema'].iloc[i]
lower_band = results['lower_band'].iloc[-1] lower_band = results['lower_band'].iloc[i]
prev_lower_band = results['lower_band'].iloc[-2] prev_lower_band = results['lower_band'].iloc[i-1]
# Entry conditions from Pine script: # Entry conditions from Pine script:
entry_signal = ( entry_signal = (
@ -42,13 +45,15 @@ def check_entry_signal(df: pd.DataFrame) -> tuple:
current['close'] > previous['close'] current['close'] > previous['close']
) )
if entry_signal:
signal_data = { signal_data = {
'price': current['close'], 'price': current['close'],
'ema': ema, 'ema': ema,
'lower_band': lower_band 'lower_band': lower_band
} if entry_signal else None }
signals.append((True, current['date'], signal_data))
return entry_signal, current['date'] if entry_signal else None, signal_data return signals
def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None: def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None:
""" """
@ -122,8 +127,8 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
if df.empty or len(df) < 21: # Need at least 21 bars for EMA if df.empty or len(df) < 21: # Need at least 21 bars for EMA
continue continue
signal, signal_date, signal_data = check_entry_signal(df) signals = check_entry_signal(df)
if signal: for signal, signal_date, signal_data in signals:
entry_data = { entry_data = {
'ticker': ticker, 'ticker': ticker,
'entry_price': signal_data['price'], 'entry_price': signal_data['price'],
@ -146,8 +151,6 @@ def run_atr_ema_scanner_v2(min_price: float, max_price: float, min_volume: int,
}) })
entry_signals.append(entry_data) entry_signals.append(entry_data)
# Print signal information with date
print_signal(entry_data) print_signal(entry_data)
except Exception as e: except Exception as e: