Spaces:
Sleeping
Sleeping
File size: 9,031 Bytes
a9237fb | 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 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | import pandas as pd
import numpy as np
import ta
from ta.utils import dropna
import warnings
warnings.filterwarnings('ignore')
class TechnicalIndicators:
def __init__(self):
pass
def calculate_all_indicators(self, data):
"""Calculate comprehensive set of technical indicators"""
try:
df = data.copy()
# Ensure we have required columns
if not all(col in df.columns for col in ['Open', 'High', 'Low', 'Close', 'Volume']):
print("Missing required OHLCV columns")
return df
# Clean data and ensure sufficient data points
df = df.dropna()
if len(df) < 30: # Reduced minimum for testing
print(f"Insufficient data: {len(df)} rows, need at least 30")
return df
print(f"Calculating indicators for {len(df)} data points...")
# Basic Moving Averages
df = self.add_moving_averages(df)
# RSI
df = self.add_rsi(df)
# MACD
df = self.add_macd(df)
# Bollinger Bands
df = self.add_bollinger_bands(df)
# Stochastic Oscillator
df = self.add_stochastic(df)
# Williams %R
df = self.add_williams_r(df)
# Average True Range (ATR)
df = self.add_atr(df)
# Additional indicators
df = self.add_additional_indicators(df)
# Volume indicators
df = self.add_volume_indicators(df)
# Count new indicators added
original_cols = ['Open', 'High', 'Low', 'Close', 'Volume', 'Dividends', 'Stock Splits']
new_indicators = [col for col in df.columns if col not in original_cols]
print(f"Successfully calculated {len(new_indicators)} technical indicators")
return df
except Exception as e:
print(f"Error calculating technical indicators: {str(e)}")
return data
def add_moving_averages(self, df):
"""Add various moving averages"""
try:
# Simple Moving Averages using pandas rolling
df['SMA_10'] = df['Close'].rolling(window=10).mean()
df['SMA_20'] = df['Close'].rolling(window=20).mean()
df['SMA_50'] = df['Close'].rolling(window=50).mean()
# Exponential Moving Averages using pandas ewm
df['EMA_12'] = df['Close'].ewm(span=12).mean()
df['EMA_20'] = df['Close'].ewm(span=20).mean()
df['EMA_26'] = df['Close'].ewm(span=26).mean()
return df
except Exception as e:
print(f"Error adding moving averages: {str(e)}")
return df
def add_rsi(self, df, window=14):
"""Add Relative Strength Index"""
try:
# Calculate RSI manually
delta = df['Close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=window).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=window).mean()
rs = gain / loss
df['RSI'] = 100 - (100 / (1 + rs))
return df
except Exception as e:
print(f"Error adding RSI: {str(e)}")
return df
def add_macd(self, df):
"""Add MACD indicators"""
try:
# Calculate MACD manually
ema_12 = df['Close'].ewm(span=12).mean()
ema_26 = df['Close'].ewm(span=26).mean()
df['MACD'] = ema_12 - ema_26
df['MACD_signal'] = df['MACD'].ewm(span=9).mean()
df['MACD_histogram'] = df['MACD'] - df['MACD_signal']
return df
except Exception as e:
print(f"Error adding MACD: {str(e)}")
return df
def add_bollinger_bands(self, df, window=20, std=2):
"""Add Bollinger Bands"""
try:
# Calculate Bollinger Bands manually
df['BB_middle'] = df['Close'].rolling(window=window).mean()
bb_std = df['Close'].rolling(window=window).std()
df['BB_upper'] = df['BB_middle'] + (bb_std * std)
df['BB_lower'] = df['BB_middle'] - (bb_std * std)
df['BB_width'] = (df['BB_upper'] - df['BB_lower']) / df['BB_middle']
df['BB_position'] = (df['Close'] - df['BB_lower']) / (df['BB_upper'] - df['BB_lower'])
return df
except Exception as e:
print(f"Error adding Bollinger Bands: {str(e)}")
return df
def add_stochastic(self, df, k_window=14, d_window=3):
"""Add Stochastic Oscillator"""
try:
stoch = ta.momentum.StochasticOscillator(df['High'], df['Low'], df['Close'],
window=k_window, smooth_window=d_window)
df['Stoch_K'] = stoch.stoch()
df['Stoch_D'] = stoch.stoch_signal()
return df
except Exception as e:
print(f"Error adding Stochastic: {str(e)}")
return df
def add_williams_r(self, df, window=14):
"""Add Williams %R"""
try:
df['Williams_R'] = ta.momentum.williams_r(df['High'], df['Low'], df['Close'], window=window)
return df
except Exception as e:
print(f"Error adding Williams %R: {str(e)}")
return df
def add_atr(self, df, window=14):
"""Add Average True Range"""
try:
df['ATR'] = ta.volatility.average_true_range(df['High'], df['Low'], df['Close'], window=window)
df['ATR_ratio'] = df['ATR'] / df['Close'] # Normalized ATR
return df
except Exception as e:
print(f"Error adding ATR: {str(e)}")
return df
def add_additional_indicators(self, df):
"""Add additional technical indicators"""
try:
# Commodity Channel Index
df['CCI'] = ta.trend.cci(df['High'], df['Low'], df['Close'], window=20)
# Rate of Change
df['ROC'] = ta.momentum.roc(df['Close'], window=12)
# Money Flow Index
df['MFI'] = ta.volume.money_flow_index(df['High'], df['Low'], df['Close'], df['Volume'], window=14)
# Parabolic SAR
df['PSAR'] = ta.trend.psar_down(df['High'], df['Low'], df['Close'])
# Aroon indicators
aroon = ta.trend.AroonIndicator(df['High'], df['Low'], window=25)
df['Aroon_Up'] = aroon.aroon_up()
df['Aroon_Down'] = aroon.aroon_down()
return df
except Exception as e:
print(f"Error adding additional indicators: {str(e)}")
return df
def add_volume_indicators(self, df):
"""Add volume-based indicators"""
try:
# Volume Moving Average
df['Volume_SMA'] = ta.volume.volume_sma(df['Close'], df['Volume'], window=20)
# Volume Weighted Average Price (VWAP)
df['VWAP'] = ta.volume.volume_weighted_average_price(df['High'], df['Low'], df['Close'], df['Volume'])
# On Balance Volume
df['OBV'] = ta.volume.on_balance_volume(df['Close'], df['Volume'])
# Accumulation/Distribution Line
df['ADL'] = ta.volume.acc_dist_index(df['High'], df['Low'], df['Close'], df['Volume'])
# Chaikin Money Flow
df['CMF'] = ta.volume.chaikin_money_flow(df['High'], df['Low'], df['Close'], df['Volume'], window=20)
return df
except Exception as e:
print(f"Error adding volume indicators: {str(e)}")
return df
def get_indicator_summary(self, df):
"""Get summary of all calculated indicators"""
try:
indicator_columns = [col for col in df.columns if col not in ['Open', 'High', 'Low', 'Close', 'Volume']]
summary = {
'total_indicators': len(indicator_columns),
'trend_indicators': [col for col in indicator_columns if any(x in col for x in ['SMA', 'EMA', 'MACD', 'PSAR', 'Aroon'])],
'momentum_indicators': [col for col in indicator_columns if any(x in col for x in ['RSI', 'Stoch', 'Williams', 'ROC', 'CCI'])],
'volatility_indicators': [col for col in indicator_columns if any(x in col for x in ['BB', 'ATR'])],
'volume_indicators': [col for col in indicator_columns if any(x in col for x in ['Volume', 'VWAP', 'OBV', 'ADL', 'CMF', 'MFI'])]
}
return summary
except Exception as e:
print(f"Error getting indicator summary: {str(e)}")
return {}
|