ParallelLLC/algorithmic_trading

1"""Vectorised technical indicators.2 3Every function takes and returns pandas objects aligned to the input index, and4every one of them is causal: the value at bar ``t`` uses only data up to and5including ``t``. That property is what makes the backtest engine's single6``shift`` enough to guarantee no look-ahead.7"""8 9from __future__ import annotations10 11import numpy as np12import pandas as pd13 14__all__ = [15    "sma",16    "ema",17    "rsi",18    "macd",19    "bollinger",20    "atr",21    "donchian",22    "zscore",23    "roc",24    "realised_vol",25]26 27 28def sma(series: pd.Series, window: int) -> pd.Series:29    return series.rolling(window, min_periods=window).mean()30 31 32def ema(series: pd.Series, window: int) -> pd.Series:33    return series.ewm(span=window, adjust=False, min_periods=window).mean()34 35 36def rsi(series: pd.Series, window: int = 14) -> pd.Series:37    """Wilder's RSI."""38    delta = series.diff()39    gain = delta.clip(lower=0.0)40    loss = -delta.clip(upper=0.0)41    avg_gain = gain.ewm(alpha=1.0 / window, adjust=False, min_periods=window).mean()42    avg_loss = loss.ewm(alpha=1.0 / window, adjust=False, min_periods=window).mean()43    rs = avg_gain / avg_loss.replace(0.0, np.nan)44    out = 100.0 - (100.0 / (1.0 + rs))45    # avg_loss == 0 leaves rs undefined: an all-gain window is RSI 100, and a46    # perfectly flat window (no gains either) is RSI 50.47    flat = (avg_gain == 0.0) & (avg_loss == 0.0)48    out = out.mask((avg_loss == 0.0) & (avg_gain > 0.0), 100.0)49    out = out.mask(flat, 50.0)50    return out.where(avg_gain.notna())51 52 53def macd(54    series: pd.Series, fast: int = 12, slow: int = 26, signal: int = 955) -> tuple[pd.Series, pd.Series, pd.Series]:56    """Returns ``(macd_line, signal_line, histogram)``."""57    macd_line = ema(series, fast) - ema(series, slow)58    signal_line = macd_line.ewm(span=signal, adjust=False, min_periods=signal).mean()59    return macd_line, signal_line, macd_line - signal_line60 61 62def bollinger(63    series: pd.Series, window: int = 20, k: float = 2.064) -> tuple[pd.Series, pd.Series, pd.Series]:65    """Returns ``(lower, middle, upper)``."""66    mid = sma(series, window)67    sd = series.rolling(window, min_periods=window).std(ddof=0)68    return mid - k * sd, mid, mid + k * sd69 70 71def atr(df: pd.DataFrame, window: int = 14) -> pd.Series:72    prev_close = df["close"].shift(1)73    tr = pd.concat(74        [75            df["high"] - df["low"],76            (df["high"] - prev_close).abs(),77            (df["low"] - prev_close).abs(),78        ],79        axis=1,80    ).max(axis=1)81    return tr.ewm(alpha=1.0 / window, adjust=False, min_periods=window).mean()82 83 84def donchian(df: pd.DataFrame, window: int = 20) -> tuple[pd.Series, pd.Series]:85    """Rolling channel excluding the current bar, so a breakout test is causal."""86    upper = df["high"].rolling(window, min_periods=window).max().shift(1)87    lower = df["low"].rolling(window, min_periods=window).min().shift(1)88    return lower, upper89 90 91def zscore(series: pd.Series, window: int = 20) -> pd.Series:92    mean = series.rolling(window, min_periods=window).mean()93    sd = series.rolling(window, min_periods=window).std(ddof=0)94    return (series - mean) / sd.replace(0.0, np.nan)95 96 97def roc(series: pd.Series, window: int = 20) -> pd.Series:98    return series.pct_change(window)99 100 101def realised_vol(returns: pd.Series, window: int = 20, periods_per_year: int = 252) -> pd.Series:102    return returns.rolling(window, min_periods=window).std(ddof=0) * np.sqrt(periods_per_year)103