import pandas as pd import numpy as np import ta from typing import List, Dict, Any def calculate_indicators(df: pd.DataFrame, indicators: List[str]) -> pd.DataFrame: """ Calculate technical indicators for the given OHLCV data Args: df: DataFrame with OHLCV data indicators: List of indicators to calculate Returns: DataFrame with calculated indicators """ # Make a copy to avoid modifying the original DataFrame result = df.copy() # Ensure we have required columns required_columns = ['timestamp', 'datetime', 'open', 'high', 'low', 'close', 'volume'] for col in required_columns: if col not in result.columns: if col == 'datetime' and 'timestamp' in result.columns: # We can calculate datetime from timestamp continue else: raise ValueError(f"Required column {col} not found in DataFrame") # Calculate RSI if 'rsi' in indicators: result['rsi'] = ta.momentum.RSIIndicator( close=result['close'], window=14 ).rsi() # Calculate MACD if 'macd' in indicators: macd = ta.trend.MACD( close=result['close'], window_fast=12, window_slow=26, window_sign=9 ) result['macd'] = macd.macd() result['macd_signal'] = macd.macd_signal() result['macd_histogram'] = macd.macd_diff() # Calculate Bollinger Bands if 'bollinger' in indicators: bollinger = ta.volatility.BollingerBands( close=result['close'], window=20, window_dev=2 ) result['bb_upper'] = bollinger.bollinger_hband() result['bb_middle'] = bollinger.bollinger_mavg() result['bb_lower'] = bollinger.bollinger_lband() # Calculate Exponential Moving Averages if 'ema' in indicators: result['ema_9'] = ta.trend.EMAIndicator( close=result['close'], window=9 ).ema_indicator() result['ema_21'] = ta.trend.EMAIndicator( close=result['close'], window=21 ).ema_indicator() result['ema_50'] = ta.trend.EMAIndicator( close=result['close'], window=50 ).ema_indicator() result['ema_200'] = ta.trend.EMAIndicator( close=result['close'], window=200 ).ema_indicator() # Calculate Average True Range (ATR) if 'atr' in indicators: result['atr'] = ta.volatility.AverageTrueRange( high=result['high'], low=result['low'], close=result['close'], window=14 ).average_true_range() # Calculate Stochastic Oscillator if 'stoch' in indicators: stoch = ta.momentum.StochasticOscillator( high=result['high'], low=result['low'], close=result['close'], window=14, smooth_window=3 ) result['stoch_k'] = stoch.stoch() result['stoch_d'] = stoch.stoch_signal() return result