126 lines
4.5 KiB
Python
126 lines
4.5 KiB
Python
import pandas as pd
|
|
import talib
|
|
from datetime import datetime
|
|
from utils.data_utils import (
|
|
get_stock_data, validate_signal_date, print_signal,
|
|
save_signals_to_csv, get_qualified_stocks
|
|
)
|
|
from utils.scanner_utils import initialize_scanner, process_signal_data
|
|
from indicators.sunny_bands import SunnyBands
|
|
|
|
def check_entry_signal(df: pd.DataFrame) -> list:
|
|
"""
|
|
Check for entry signals based on combined Sunny Bands and SMA strategy
|
|
|
|
Conditions:
|
|
1. 21 SMA below lowest Sunny Band
|
|
2. Price crosses above 21 SMA
|
|
3. Price still below Sunny Bands
|
|
|
|
Args:
|
|
df (pd.DataFrame): DataFrame with price data
|
|
|
|
Returns:
|
|
list: List of tuples (signal, date, signal_data) for each signal found
|
|
"""
|
|
if len(df) < 21: # Need at least 21 bars for SMA
|
|
return []
|
|
|
|
# Calculate Sunny Bands
|
|
sunny = SunnyBands()
|
|
sunny_results = sunny.calculate(df)
|
|
|
|
# Calculate 21 day SMA
|
|
sma21 = talib.SMA(df['close'].values, timeperiod=21)
|
|
|
|
signals = []
|
|
|
|
# Start from index 21 to ensure we have enough data for SMA
|
|
for i in range(21, len(df)):
|
|
current = df.iloc[i]
|
|
prev = df.iloc[i-1]
|
|
current_bands = sunny_results.iloc[i]
|
|
current_sma = sma21[i]
|
|
|
|
# Check conditions:
|
|
# 1. SMA below lower band
|
|
sma_below_band = current_sma < current_bands['lower_band']
|
|
|
|
# 2. Price crosses above SMA
|
|
price_cross_sma = (current['close'] > current_sma) and (prev['close'] < sma21[i-1])
|
|
|
|
# 3. Price still below lower band
|
|
price_below_band = current['close'] < current_bands['lower_band']
|
|
|
|
if sma_below_band and price_cross_sma and price_below_band:
|
|
signal_data = {
|
|
'price': current['close'],
|
|
'sma21': current_sma,
|
|
'upper_band': current_bands['upper_band'],
|
|
'lower_band': current_bands['lower_band'],
|
|
'dma': current_bands['dma']
|
|
}
|
|
signals.append((True, current['date'], signal_data))
|
|
|
|
return signals
|
|
|
|
def run_sunny_sma_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:
|
|
"""
|
|
Run scanner combining Sunny Bands and 21 SMA strategy
|
|
"""
|
|
try:
|
|
# Initialize scanner components
|
|
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) < 50: # Need at least 50 bars for the indicators
|
|
continue
|
|
|
|
signals = check_entry_signal(df)
|
|
for signal, signal_date, signal_data in signals:
|
|
# Custom print for Sunny-SMA signals
|
|
print(f"🌞 {ticker}: SMA-21 Cross at ${signal_data['price']:.2f} on {signal_date.strftime('%Y-%m-%d')}")
|
|
print(f" SMA: ${signal_data['sma21']:.2f}")
|
|
print(f" Lower Band: ${signal_data['lower_band']:.2f}")
|
|
|
|
entry_data = {
|
|
'ticker': ticker,
|
|
'signal_date': signal_date,
|
|
'entry_price': signal_data['price'],
|
|
'sma21': signal_data['sma21'],
|
|
'lower_band': signal_data['lower_band'],
|
|
'volume': current_volume,
|
|
'last_update': last_update,
|
|
'stock_type': stock_type
|
|
}
|
|
bullish_signals.append(entry_data)
|
|
|
|
except Exception as e:
|
|
print(f"Error processing {ticker}: {str(e)}")
|
|
continue
|
|
|
|
save_signals_to_csv(bullish_signals, 'sunny_sma')
|
|
return bullish_signals
|
|
|
|
except Exception as e:
|
|
print(f"Error during scan: {str(e)}")
|
|
return []
|