107 lines
3.8 KiB
Python
107 lines
3.8 KiB
Python
from datetime import datetime
|
|
import pandas as pd
|
|
from utils.scanner_utils import initialize_scanner, process_signal_data
|
|
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_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)
|
|
|
|
if len(results) < 2:
|
|
return []
|
|
|
|
signals = []
|
|
|
|
# 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:
|
|
# 1. Previous close was below the lower band
|
|
# 2. Current close is at or above the lower band
|
|
# 3. Price is moving up (current close > previous close)
|
|
# 4. Price is still below EMA (maintaining downtrend context)
|
|
entry_signal = (
|
|
previous['close'] < prev_lower_band and
|
|
current['close'] >= lower_band and
|
|
current['close'] > previous['close'] and
|
|
current['close'] < ema
|
|
)
|
|
|
|
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, interval: str = "1d",
|
|
start_date: datetime = None, end_date: datetime = None) -> None:
|
|
try:
|
|
# Initialize scanner components with all parameters
|
|
interval, start_date, end_date, qualified_stocks, calculator = initialize_scanner(
|
|
min_price=min_price,
|
|
max_price=max_price,
|
|
min_volume=min_volume,
|
|
portfolio_size=portfolio_size,
|
|
interval=interval,
|
|
start_date=start_date,
|
|
end_date=end_date
|
|
)
|
|
|
|
if not qualified_stocks:
|
|
return
|
|
|
|
bullish_signals = []
|
|
|
|
for ticker, current_price, current_volume, last_update, stock_type in qualified_stocks:
|
|
try:
|
|
df = get_stock_data(ticker, start_date, end_date, interval)
|
|
|
|
if df.empty or len(df) < 21: # Need at least 21 bars for EMA
|
|
continue
|
|
|
|
signals = check_entry_signal(df)
|
|
for signal, signal_date, signal_data in signals:
|
|
signal_data['date'] = signal_date
|
|
entry_data = process_signal_data(
|
|
ticker, signal_data, current_volume,
|
|
last_update, stock_type, calculator
|
|
)
|
|
bullish_signals.append(entry_data)
|
|
print_signal(entry_data, "🟢")
|
|
|
|
except Exception as e:
|
|
print(f"Error processing {ticker}: {str(e)}")
|
|
continue
|
|
|
|
save_signals_to_csv(bullish_signals, 'atr_ema')
|
|
return bullish_signals
|
|
|
|
except Exception as e:
|
|
print(f"Error during scan: {str(e)}")
|
|
return []
|