File size: 3,315 Bytes
46cedba | 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 | """Advanced Technical Indicators - Ichimoku, Supertrend, Keltner, VPVR"""
import numpy as np
import pandas as pd
class AdvancedTechnical:
"""Advanced technical indicators"""
@staticmethod
def ichimoku(close, high, low, tenkan=9, kijun=26, senkou_b=52):
features = pd.DataFrame(index=close.index)
tenkan_s = (high.rolling(tenkan).max() + low.rolling(tenkan).min()) / 2
kijun_s = (high.rolling(kijun).max() + low.rolling(kijun).min()) / 2
senkou_a = (tenkan_s + kijun_s) / 2
senkou_b = (high.rolling(senkou_b).max() + low.rolling(senkou_b).min()) / 2
features['ichimoku_tk_cross'] = (tenkan_s - kijun_s) / close
features['ichimoku_price_kumo'] = (close - senkou_a) / close
features['ichimoku_cloud_width'] = (senkou_a - senkou_b) / close
return features
@staticmethod
def supertrend(close, high, low, period=10, multiplier=3.0):
features = pd.DataFrame(index=close.index)
tr = pd.concat([high - low, (high - close.shift(1)).abs(), (low - close.shift(1)).abs()], axis=1).max(axis=1)
atr = tr.rolling(period).mean()
hl2 = (high + low) / 2
upper = hl2 + multiplier * atr
lower = hl2 - multiplier * atr
# Determine trend direction
supertrend_val = pd.Series(index=close.index, dtype=float)
direction = pd.Series(1.0, index=close.index)
supertrend_val.iloc[0] = lower.iloc[0]
for i in range(1, len(close)):
if close.iloc[i] > upper.iloc[i-1]:
supertrend_val.iloc[i] = lower.iloc[i]
direction.iloc[i] = 1.0
elif close.iloc[i] < lower.iloc[i-1]:
supertrend_val.iloc[i] = upper.iloc[i]
direction.iloc[i] = -1.0
else:
supertrend_val.iloc[i] = supertrend_val.iloc[i-1]
direction.iloc[i] = direction.iloc[i-1]
features['supertrend_dist'] = (close - supertrend_val) / close
features['supertrend_signal'] = direction
return features
@staticmethod
def keltner_channels(close, high, low, ema_period=20, atr_period=10, multiplier=2.0):
features = pd.DataFrame(index=close.index)
ema = close.ewm(span=ema_period).mean()
tr = pd.concat([high - low, (high - close.shift(1)).abs(), (low - close.shift(1)).abs()], axis=1).max(axis=1)
atr = tr.rolling(atr_period).mean()
upper = ema + multiplier * atr
lower = ema - multiplier * atr
features['keltner_position'] = (close - lower) / (upper - lower).replace(0, np.nan)
features['keltner_width'] = (upper - lower) / close
return features
@staticmethod
def volume_profile(close, volume, high, low, window=21):
features = pd.DataFrame(index=close.index)
vwp = (close * volume).rolling(window).sum() / volume.rolling(window).sum().replace(0, np.nan)
features['vwp_deviation'] = (close - vwp) / close
up_vol = volume.where(close > close.shift(1), 0)
down_vol = volume.where(close < close.shift(1), 0)
features['volume_delta'] = (up_vol.rolling(window).sum() - down_vol.rolling(window).sum()) / \
volume.rolling(window).sum().replace(0, np.nan)
return features |