""" AkDevz AI Market Intelligence — Technical Indicators Engine Pure NumPy vector calculations for zero-lag intraday metrics. """ import numpy as np from typing import Dict, Any, Tuple class TechnicalEngine: @staticmethod def calculate_ema(prices: np.ndarray, period: int) -> np.ndarray: if len(prices) < period: return np.full_like(prices, np.nan) ema = np.empty_like(prices) ema[:] = np.nan # Simple moving average for initial seed ema[period - 1] = np.mean(prices[:period]) multiplier = 2.0 / (period + 1.0) for i in range(period, len(prices)): ema[i] = (prices[i] - ema[i - 1]) * multiplier + ema[i - 1] return ema @staticmethod def calculate_sma(prices: np.ndarray, period: int) -> np.ndarray: if len(prices) < period: return np.full_like(prices, np.nan) weights = np.repeat(1.0, period) / period sma = np.convolve(prices, weights, 'valid') padding = np.full(period - 1, np.nan) return np.concatenate((padding, sma)) @staticmethod def calculate_vwap(high: np.ndarray, low: np.ndarray, close: np.ndarray, volume: np.ndarray) -> np.ndarray: typical_price = (high + low + close) / 3.0 cum_tp_vol = np.cumsum(typical_price * volume) cum_vol = np.cumsum(volume) # Avoid zero division with np.errstate(divide='ignore', invalid='ignore'): vwap = np.where(cum_vol > 0, cum_tp_vol / cum_vol, typical_price) return vwap @staticmethod def calculate_rsi(close: np.ndarray, period: int = 14) -> np.ndarray: if len(close) <= period: return np.full_like(close, np.nan) diff = np.diff(close) gains = np.where(diff > 0, diff, 0.0) losses = np.where(diff < 0, -diff, 0.0) rsi = np.full_like(close, np.nan) avg_gain = np.mean(gains[:period]) avg_loss = np.mean(losses[:period]) if avg_loss == 0: rsi[period] = 100.0 else: rs = avg_gain / avg_loss rsi[period] = 100.0 - (100.0 / (1.0 + rs)) for i in range(period + 1, len(close)): avg_gain = (avg_gain * (period - 1) + gains[i - 1]) / period avg_loss = (avg_loss * (period - 1) + losses[i - 1]) / period if avg_loss == 0: rsi[i] = 100.0 else: rs = avg_gain / avg_loss rsi[i] = 100.0 - (100.0 / (1.0 + rs)) return rsi @staticmethod def calculate_atr(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray: if len(close) < period + 1: return np.full_like(close, np.nan) tr1 = high[1:] - low[1:] tr2 = np.abs(high[1:] - close[:-1]) tr3 = np.abs(low[1:] - close[:-1]) tr = np.maximum(tr1, np.maximum(tr2, tr3)) atr = np.full_like(close, np.nan) atr[period] = np.mean(tr[:period]) for i in range(period + 1, len(close)): atr[i] = (atr[i - 1] * (period - 1) + tr[i - 1]) / period return atr @staticmethod def calculate_bollinger_bands(close: np.ndarray, period: int = 20, num_std: float = 2.0) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: if len(close) < period: nan_arr = np.full_like(close, np.nan) return nan_arr, nan_arr, nan_arr sma = TechnicalEngine.calculate_sma(close, period) rolling_std = np.empty_like(close) rolling_std[:] = np.nan for i in range(period - 1, len(close)): rolling_std[i] = np.std(close[i - period + 1:i + 1]) upper = sma + (rolling_std * num_std) lower = sma - (rolling_std * num_std) return upper, sma, lower @staticmethod def calculate_supertrend(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 10, multiplier: float = 3.0) -> Tuple[np.ndarray, np.ndarray]: """ Calculates Supertrend and direction (1 for Bullish, -1 for Bearish) """ n = len(close) if n < period: return np.full(n, np.nan), np.zeros(n) atr = TechnicalEngine.calculate_atr(high, low, close, period) hl2 = (high + low) / 2.0 upper_band = hl2 + (multiplier * atr) lower_band = hl2 - (multiplier * atr) supertrend = np.full(n, np.nan) direction = np.zeros(n, dtype=int) # Initialize from first valid ATR bar start_idx = period supertrend[start_idx] = upper_band[start_idx] direction[start_idx] = 1 if close[start_idx] > upper_band[start_idx] else -1 for i in range(start_idx + 1, n): # Trailing bounds if lower_band[i] > lower_band[i - 1] or close[i - 1] < lower_band[i - 1]: final_lower = lower_band[i] else: final_lower = lower_band[i - 1] if upper_band[i] < upper_band[i - 1] or close[i - 1] > upper_band[i - 1]: final_upper = upper_band[i] else: final_upper = upper_band[i - 1] # Flip logic if direction[i - 1] == 1: if close[i] < final_lower: direction[i] = -1 supertrend[i] = final_upper else: direction[i] = 1 supertrend[i] = final_lower else: if close[i] > final_upper: direction[i] = 1 supertrend[i] = final_lower else: direction[i] = -1 supertrend[i] = final_upper return supertrend, direction @staticmethod def calculate_adx(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> Tuple[np.ndarray, np.ndarray, np.ndarray]: n = len(close) if n < (period * 2): nan_arr = np.full(n, np.nan) return nan_arr, nan_arr, nan_arr up_move = high[1:] - high[:-1] down_move = low[:-1] - low[1:] plus_dm = np.where((up_move > down_move) & (up_move > 0), up_move, 0.0) minus_dm = np.where((down_move > up_move) & (down_move > 0), down_move, 0.0) tr1 = high[1:] - low[1:] tr2 = np.abs(high[1:] - close[:-1]) tr3 = np.abs(low[1:] - close[:-1]) tr = np.maximum(tr1, np.maximum(tr2, tr3)) atr = np.zeros(n - 1) smooth_plus = np.zeros(n - 1) smooth_minus = np.zeros(n - 1) atr[period - 1] = np.sum(tr[:period]) smooth_plus[period - 1] = np.sum(plus_dm[:period]) smooth_minus[period - 1] = np.sum(minus_dm[:period]) for i in range(period, n - 1): atr[i] = atr[i - 1] - (atr[i - 1] / period) + tr[i] smooth_plus[i] = smooth_plus[i - 1] - (smooth_plus[i - 1] / period) + plus_dm[i] smooth_minus[i] = smooth_minus[i - 1] - (smooth_minus[i - 1] / period) + minus_dm[i] with np.errstate(divide='ignore', invalid='ignore'): plus_di = 100.0 * (smooth_plus / atr) minus_di = 100.0 * (smooth_minus / atr) dx = 100.0 * np.abs(plus_di - minus_di) / (plus_di + minus_di) adx = np.full(n, np.nan) adx[2 * period] = np.mean(dx[period:2 * period]) for i in range(2 * period + 1, n): adx[i] = (adx[i - 1] * (period - 1) + dx[i - 1]) / period # Pad DI outputs to match input array length pad_plus = np.concatenate(([np.nan], plus_di)) pad_minus = np.concatenate(([np.nan], minus_di)) return adx, pad_plus, pad_minus