Upload technical_indicators.py

technical_indicators.py

@@ -0,0 +1,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