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 {}