Spaces:
Build error
Build error
| # src/data/indicators.py | |
| import pandas as pd | |
| import numpy as np | |
| import ta | |
| class TechnicalIndicators: | |
| """Engineers a rich set of features from price data, including adaptive indicators.""" | |
| def adaptive_laguerre_rsi(df: pd.DataFrame, gamma: float = 0.5) -> pd.Series: | |
| """ATR-adaptive Laguerre RSI.""" | |
| atr = ta.volatility.average_true_range(df['high'], df['low'], df['close']) | |
| # Normalize ATR to adjust gamma | |
| atr_norm = atr / atr.rolling(20).mean() | |
| adaptive_gamma = gamma * (1 + (atr_norm - 1).clip(-0.3, 0.3)) | |
| price = df['close'] | |
| l0 = l1 = l2 = l3 = 0.0 | |
| l0s = [] | |
| for i in range(len(price)): | |
| gamma_val = adaptive_gamma.iloc[i] if not pd.isna(adaptive_gamma.iloc[i]) else gamma | |
| l0 = (1 - gamma_val) * price.iloc[i] + gamma_val * l1 | |
| l1 = (1 - gamma_val) * l0 + gamma_val * l2 | |
| l2 = (1 - gamma_val) * l1 + gamma_val * l3 | |
| l3 = (1 - gamma_val) * l2 + gamma_val * l3 | |
| l0s.append(l0) | |
| l0_series = pd.Series(l0s, index=price.index) | |
| # Calculate RSI from Laguerre filter | |
| cu = (l0_series - l0_series.shift(1)).clip(lower=0).rolling(14).sum() | |
| cd = (l0_series.shift(1) - l0_series).clip(lower=0).rolling(14).sum() | |
| rsi = 100 - (100 / (1 + cu / cd)) | |
| return rsi | |
| def add_adaptive_indicators(df: pd.DataFrame) -> pd.DataFrame: | |
| df = df.copy() | |
| df['adaptive_laguerre_rsi'] = TechnicalIndicators.adaptive_laguerre_rsi(df) | |
| # Volatility-adjusted momentum | |
| df['vol_adj_momentum'] = df['close'].pct_change(10) / df['atr'] | |
| return df | |
| def add_all_indicators(df: pd.DataFrame) -> pd.DataFrame: | |
| df = df.copy() | |
| # Trend Indicators | |
| df['sma_20'] = ta.trend.sma_indicator(df['close'], window=20) | |
| df['sma_50'] = ta.trend.sma_indicator(df['close'], window=50) | |
| df['ema_20'] = ta.trend.ema_indicator(df['close'], window=20) | |
| df['macd'] = ta.trend.macd_diff(df['close']) | |
| # Momentum Indicators | |
| df['rsi'] = ta.momentum.rsi(df['close']) | |
| df['stoch_k'] = ta.momentum.stoch(df['high'], df['low'], df['close']) | |
| # Volatility Indicators | |
| df['bb_high'] = ta.volatility.bollinger_hband(df['close']) | |
| df['bb_low'] = ta.volatility.bollinger_lband(df['close']) | |
| df['atr'] = ta.volatility.average_true_range(df['high'], df['low'], df['close']) | |
| # Volume Indicators | |
| df['obv'] = ta.volume.on_balance_volume(df['close'], df['volume']) | |
| # Price-derived features | |
| df['returns'] = df['close'].pct_change() | |
| df['log_returns'] = np.log(df['close'] / df['close'].shift(1)) | |
| df['high_low_ratio'] = (df['high'] - df['low']) / df['close'] | |
| # Add adaptive indicators | |
| df = TechnicalIndicators.add_adaptive_indicators(df) | |
| return df.dropna() |