From 15304106ac3118b035082746a11e18cd085b102e Mon Sep 17 00:00:00 2001 From: "Bobby (aider)" Date: Sat, 8 Feb 2025 20:05:29 -0800 Subject: [PATCH] refactor: standardize ATR EMA scanner to scan entire date range --- src/screener/t_atr_ema.py | 118 +++++++++++++++++++++----------------- 1 file changed, 66 insertions(+), 52 deletions(-) diff --git a/src/screener/t_atr_ema.py b/src/screener/t_atr_ema.py index 7e33e87..a421324 100644 --- a/src/screener/t_atr_ema.py +++ b/src/screener/t_atr_ema.py @@ -7,41 +7,53 @@ from trading.position_calculator import PositionCalculator from utils.data_utils import get_stock_data, validate_signal_date, print_signal, save_signals_to_csv from indicators.three_atr_ema import ThreeATREMAIndicator -def check_atr_ema_bullish_signal(df: pd.DataFrame) -> bool: - """Check for bullish signal based on ATR EMA indicator""" - # Get latest values from DataFrame - last_price = df.iloc[-1] - previous_price = df.iloc[-2] # Get the previous row for comparison +def check_entry_signal(df: pd.DataFrame) -> list: + """ + Check for entry signals based on Three ATR EMA strategy throughout the date range + + Args: + df (pd.DataFrame): DataFrame with price data + + Returns: + list: List of tuples (signal, date, signal_data) for each signal found + """ + if len(df) < 2: # Need at least 2 bars for comparison + return [] + indicator = ThreeATREMAIndicator() results = indicator.calculate(df) - indicator = ThreeATREMAIndicator() - results = indicator.calculate(df) - last_bands = results.iloc[-1] - print(f"\nSunnyBands Indicators:") - print(f"DMA: ${last_bands['dma']:.2f}") - print(f"Upper Band: ${last_bands['upper_band']:.2f}") - print(f"Lower Band: ${last_bands['lower_band']:.2f}") - print(f"Bullish Signal: {'Yes' if last_bands['signal'] else 'No'}") - -def check_atr_ema_buy_condition(df: pd.DataFrame) -> tuple: - """Check if price is below EMA and moving up through lower ATR band""" - # Get latest values from DataFrame - last_price = df.iloc[-1] - previous_price = df.iloc[-2] # Get the previous row for comparison - results = ThreeATREMAIndicator().calculate(df) # Ensure results are calculated here - - # Check if price is below EMA and has started moving up - ema = results['ema'].iloc[-1] - lower_band = results['lower_band'].iloc[-1] + if len(results) < 2: + return [] - signal = ( - last_price['close'] < ema and - previous_price['close'] <= lower_band and - last_price['close'] > previous_price['close'] - ) + signals = [] - return signal, last_price['date'] if signal else None, results.iloc[-1] + # 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 + ema = results['ema'].iloc[i] + lower_band = results['lower_band'].iloc[i] + prev_lower_band = results['lower_band'].iloc[i-1] + + # Entry conditions + entry_signal = ( + current['close'] < ema and + previous['close'] <= prev_lower_band and + current['close'] > previous['close'] + ) + + if entry_signal: + signal_data = { + 'price': current['close'], + 'ema': ema, + 'lower_band': lower_band + } + signals.append((True, current['date'], signal_data)) + + return signals def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, portfolio_size: float = None) -> None: print(f"\nScanning for stocks ${min_price:.2f}-${max_price:.2f} with min volume {min_volume:,}") @@ -110,30 +122,32 @@ def run_atr_ema_scanner(min_price: float, max_price: float, min_volume: int, por results = indicator.calculate(df) - # Check for signals - signal, signal_date, indicator_values = check_atr_ema_buy_condition(df) - if signal: - target_price = indicator_values['upper_band'] + # Check for signals throughout the date range + signals = check_entry_signal(df) + for signal, signal_date, signal_data in signals: + entry_data = { + 'ticker': ticker, + 'entry_price': signal_data['price'], + 'target_price': signal_data['ema'], + 'volume': current_volume, + 'signal_date': signal_date, + 'last_update': datetime.fromtimestamp(last_update/1000000000) + } if calculator: - position = calculator.calculate_position_size(current_price, target_price) - if position['shares'] > 0: - signal_data = { - 'ticker': ticker, - 'entry_price': current_price, - 'target_price': target_price, - 'signal_date': signal_date, - 'volume': current_volume, - 'last_update': datetime.fromtimestamp(last_update/1000000000), - 'shares': position['shares'], - 'position_size': position['position_value'], - 'stop_loss': position['stop_loss'], - 'risk_amount': position['potential_loss'], - 'profit_amount': position['potential_profit'], - 'risk_reward_ratio': position['risk_reward_ratio'] - } - bullish_signals.append(signal_data) - print_signal(signal_data, "🟢") + position = calculator.calculate_position_size(entry_data['entry_price']) + potential_profit = (entry_data['target_price'] - entry_data['entry_price']) * position['shares'] + entry_data.update({ + 'shares': position['shares'], + 'position_size': position['position_value'], + 'stop_loss': position['stop_loss'], + 'risk_amount': position['potential_loss'], + 'profit_amount': potential_profit, + 'risk_reward_ratio': abs(potential_profit / position['potential_loss']) if position['potential_loss'] != 0 else 0 + }) + + bullish_signals.append(entry_data) + print_signal(entry_data, "🟢") except Exception as e: print(f"Error processing {ticker}: {str(e)}")