signals/strategy.py

# 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.