refactor: standardize ATR EMA scanner to scan entire date range
This commit is contained in:
parent
cea43c4d32
commit
15304106ac
@ -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)}")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user