Spaces:
Build error
Build error
File size: 3,015 Bytes
d82ffcc | 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 | # 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."""
@staticmethod
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
@staticmethod
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
@staticmethod
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() |