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