"""Common technical indicators as pure numpy functions. All functions take a 1-D price array (or H/L/C arrays) and return an array of the same length, with leading ``NaN`` where the lookback window is not yet full. Add strategy-specific indicators (your custom range filter, regime detector, …) alongside these as you need them. These are vectorized with numpy; for the hot path compile them with numba if profiling demands it (doc 01 — numba is in the stack for that reason). """ from __future__ import annotations import numpy as np def sma(prices: np.ndarray, period: int) -> np.ndarray: """Simple moving average. ``NaN`` until ``period`` values are seen.""" if period <= 0: raise ValueError("period must be > 0") out = np.full(prices.shape, np.nan, dtype=float) if len(prices) < period: return out csum = np.cumsum(prices, dtype=float) csum[period:] = csum[period:] - csum[:-period] out[period - 1:] = csum[period - 1:] / period return out def ema(prices: np.ndarray, period: int) -> np.ndarray: """Exponential moving average (seeded with the first ``period`` SMA).""" if period <= 0: raise ValueError("period must be > 0") out = np.full(prices.shape, np.nan, dtype=float) if len(prices) < period: return out alpha = 2.0 / (period + 1.0) out[period - 1] = prices[:period].mean() for i in range(period, len(prices)): out[i] = alpha * prices[i] + (1.0 - alpha) * out[i - 1] return out def rsi(prices: np.ndarray, period: int = 14) -> np.ndarray: """Relative Strength Index (Wilder's smoothing). Range ``[0, 100]``.""" if period <= 0: raise ValueError("period must be > 0") out = np.full(prices.shape, np.nan, dtype=float) if len(prices) <= period: return out deltas = np.diff(prices, prepend=prices[0]) gains = np.where(deltas > 0, deltas, 0.0) losses = np.where(deltas < 0, -deltas, 0.0) avg_gain = gains[1:period + 1].mean() avg_loss = losses[1:period + 1].mean() for i in range(period, len(prices)): avg_gain = (avg_gain * (period - 1) + gains[i]) / period avg_loss = (avg_loss * (period - 1) + losses[i]) / period rs = avg_gain / avg_loss if avg_loss != 0 else np.inf out[i] = 100.0 - 100.0 / (1.0 + rs) return out def atr( high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14, ) -> np.ndarray: """Average True Range (Wilder's smoothing).""" if not (len(high) == len(low) == len(close)): raise ValueError("high/low/close must have equal length") if period <= 0: raise ValueError("period must be > 0") out = np.full(close.shape, np.nan, dtype=float) if len(close) <= period: return out prev_close = np.concatenate(([close[0]], close[:-1])) tr = np.maximum.reduce([ high - low, np.abs(high - prev_close), np.abs(low - prev_close), ]) atr_val = tr[1:period + 1].mean() out[period] = atr_val for i in range(period + 1, len(close)): atr_val = (atr_val * (period - 1) + tr[i]) / period out[i] = atr_val return out