""" Pure-function technical indicators. All operate on pandas Series/DataFrame and return pandas objects so NaN propagation is handled automatically. """ import numpy as np import pandas as pd def ema(series: pd.Series, period: int) -> pd.Series: return series.ewm(span=period, min_periods=period, adjust=False).mean() def rsi(close: pd.Series, period: int = 14) -> pd.Series: delta = close.diff() gain = delta.clip(lower=0) loss = (-delta).clip(lower=0) avg_g = gain.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() avg_l = loss.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() rs = avg_g / avg_l.replace(0, np.inf) return 100 - (100 / (1 + rs)) def atr(df: pd.DataFrame, period: int = 14) -> pd.Series: hi, lo, pc = df["high"], df["low"], df["close"].shift(1) tr = pd.concat([(hi - lo), (hi - pc).abs(), (lo - pc).abs()], axis=1).max(axis=1) return tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() def adx(df: pd.DataFrame, period: int = 14) -> pd.Series: """Average Directional Index.""" hi, lo, pc = df["high"], df["low"], df["close"].shift(1) tr = pd.concat([(hi - lo), (hi - pc).abs(), (lo - pc).abs()], axis=1).max(axis=1) dm_pos = (hi - hi.shift(1)).clip(lower=0).where((hi - hi.shift(1)) > (lo.shift(1) - lo), 0) dm_neg = (lo.shift(1) - lo).clip(lower=0).where((lo.shift(1) - lo) > (hi - hi.shift(1)), 0) atr_s = tr.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() di_pos = 100 * dm_pos.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() / atr_s di_neg = 100 * dm_neg.ewm(alpha=1 / period, min_periods=period, adjust=False).mean() / atr_s dx = (100 * (di_pos - di_neg).abs() / (di_pos + di_neg).replace(0, np.nan)) return dx.ewm(alpha=1 / period, min_periods=period, adjust=False).mean()