netflypsb's picture
Create signals/strategy.py
f382718 verified
raw
history blame
3.11 kB
# signals/strategy.py
def indicator_signal_ema(data, short_period=12, long_period=26):
"""Calculate EMA signals: buy, sell, neutral."""
ema_short = data['Close'].ewm(span=short_period, adjust=False).mean()
ema_long = data['Close'].ewm(span=long_period, adjust=False).mean()
data['EMA_Short'] = ema_short
data['EMA_Long'] = ema_long
data['EMA_Signal'] = 'neutral'
data.loc[ema_short > ema_long, 'EMA_Signal'] = 'buy'
data.loc[ema_short < ema_long, 'EMA_Signal'] = 'sell'
return data
def indicator_signal_macd(data, fast_period=12, slow_period=26, signal_period=9):
"""Calculate MACD signals: buy, sell, neutral."""
exp1 = data['Close'].ewm(span=fast_period, adjust=False).mean()
exp2 = data['Close'].ewm(span=slow_period, adjust=False).mean()
macd = exp1 - exp2
signal_line = macd.ewm(span=signal_period, adjust=False).mean()
data['MACD'] = macd
data['MACD_Signal_Line'] = signal_line
data['MACD_Signal'] = 'neutral'
data.loc[macd > signal_line, 'MACD_Signal'] = 'buy'
data.loc[macd < signal_line, 'MACD_Signal'] = 'sell'
return data
def indicator_signal_rsi(data, period=14, overbought=70, oversold=30):
"""Calculate RSI signals: buy, sell, neutral."""
delta = data['Close'].diff(1)
gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
rs = gain / loss
rsi = 100 - (100 / (1 + rs))
data['RSI'] = rsi
data['RSI_Signal'] = 'neutral'
data.loc[rsi < oversold, 'RSI_Signal'] = 'buy'
data.loc[rsi > overbought, 'RSI_Signal'] = 'sell'
return data
def indicator_signal_bollinger_bands(data, period=20, std_dev=2):
"""Calculate Bollinger Band signals: buy, sell, neutral."""
sma = data['Close'].rolling(window=period).mean()
std = data['Close'].rolling(window=period).std()
upper_band = sma + (std_dev * std)
lower_band = sma - (std_dev * std)
data['BB_Upper'] = upper_band
data['BB_Lower'] = lower_band
data['BB_Signal'] = 'neutral'
data.loc[data['Close'] < lower_band, 'BB_Signal'] = 'buy'
data.loc[data['Close'] > upper_band, 'BB_Signal'] = 'sell'
return data
def generate_combined_signals(data):
"""Combine signals from all indicators and generate final buy/sell signals."""
data = indicator_signal_ema(data)
data = indicator_signal_macd(data)
data = indicator_signal_rsi(data)
data = indicator_signal_bollinger_bands(data)
# Analyze signals from all indicators
signals = ['EMA_Signal', 'MACD_Signal', 'RSI_Signal', 'BB_Signal']
data['Combined_Signal'] = data.apply(lambda row: 'buy' if sum(row[signal] == 'buy' for signal in signals) >= 3
else ('sell' if sum(row[signal] == 'sell' for signal in signals) >= 3 else 'neutral'), axis=1)
return data
# Note: This script assumes the 'data' DataFrame contains a 'Close' column with the closing prices.
# You will need to replace the placeholder calculation functions with the actual implementations
# from your indicators package.