File size: 3,133 Bytes
5cd2b89 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 |
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
|