Spaces:
Sleeping
Sleeping
| 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 | |