Technical Indicators with Python — Deep Dive

Building indicators from first principles

Understanding the math behind each indicator prevents black-box usage and enables custom variations tailored to specific instruments or timeframes.

Exponential Moving Average internals

The EMA uses a smoothing factor (alpha) derived from the period. Each new value blends the current price with the previous EMA:

import numpy as np
import pandas as pd

def ema(prices: pd.Series, period: int) -> pd.Series:
    """EMA with the standard Wilder smoothing convention."""
    alpha = 2.0 / (period + 1)
    result = np.empty(len(prices))
    result[:period] = np.nan
    result[period - 1] = prices.iloc[:period].mean()  # seed with SMA
    
    for i in range(period, len(prices)):
        result[i] = alpha * prices.iloc[i] + (1 - alpha) * result[i - 1]
    
    return pd.Series(result, index=prices.index)

The choice of seed value matters for short series. For production use, the difference between SMA-seeded and first-value-seeded EMA diminishes after roughly 3× the period length.

Wilder’s RSI vs. Cutler’s RSI

The original RSI by Wilder uses an exponentially weighted moving average for gain and loss, while Cutler’s variant uses a simple moving average. The difference is subtle but affects signal timing:

def wilder_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
    """RSI using Wilder's smoothing (EWMA with alpha = 1/period)."""
    delta = prices.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)
    
    # Wilder smoothing: alpha = 1/period
    avg_gain = gain.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
    avg_loss = loss.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
    
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

def cutler_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
    """RSI using simple moving average — no path dependency."""
    delta = prices.diff()
    gain = delta.clip(lower=0)
    loss = -delta.clip(upper=0)
    
    avg_gain = gain.rolling(period).mean()
    avg_loss = loss.rolling(period).mean()
    
    rs = avg_gain / avg_loss
    return 100 - (100 / (1 + rs))

Wilder’s version is path-dependent (the current value depends on the entire history), while Cutler’s only looks at the most recent window. For backtesting reproducibility, Cutler’s variant is simpler to validate.

MACD with histogram divergence detection

Beyond basic crossover signals, MACD histogram divergence — where price makes a new high but the histogram does not — is a more nuanced signal:

def macd_full(prices: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> pd.DataFrame:
    ema_fast = prices.ewm(span=fast, adjust=False).mean()
    ema_slow = prices.ewm(span=slow, adjust=False).mean()
    macd_line = ema_fast - ema_slow
    signal_line = macd_line.ewm(span=signal, adjust=False).mean()
    histogram = macd_line - signal_line
    
    return pd.DataFrame({
        "macd": macd_line,
        "signal": signal_line,
        "histogram": histogram,
    })

def detect_bearish_divergence(prices: pd.Series, histogram: pd.Series, lookback: int = 20) -> pd.Series:
    """Flag bars where price makes a higher high but MACD histogram makes a lower high."""
    price_hh = prices.rolling(lookback).max() == prices
    hist_peaks = histogram.rolling(lookback).max()
    hist_declining = histogram < hist_peaks.shift(lookback)
    
    return price_hh & hist_declining

Vectorized computation at scale

When computing indicators across thousands of symbols, loop-based implementations become bottlenecks. NumPy vectorization and Numba JIT compilation provide significant speedups.

Vectorized Bollinger Bands across a panel

import numpy as np

def bollinger_panel(close_matrix: np.ndarray, period: int = 20, num_std: float = 2.0) -> tuple:
    """
    Compute Bollinger Bands for a (time × symbols) matrix in one pass.
    Returns upper, middle, lower bands as matrices.
    """
    # Rolling mean via convolution
    kernel = np.ones(period) / period
    middle = np.apply_along_axis(lambda x: np.convolve(x, kernel, mode='same'), axis=0, arr=close_matrix)
    
    # Rolling std via the variance identity: std = sqrt(E[x²] - E[x]²)
    sq_mean = np.apply_along_axis(lambda x: np.convolve(x**2, kernel, mode='same'), axis=0, arr=close_matrix)
    variance = sq_mean - middle**2
    std = np.sqrt(np.maximum(variance, 0))
    
    upper = middle + num_std * std
    lower = middle - num_std * std
    
    return upper, middle, lower

Numba-accelerated ATR

import numpy as np
from numba import njit

@njit
def atr_numba(high: np.ndarray, low: np.ndarray, close: np.ndarray, period: int = 14) -> np.ndarray:
    """ATR with Numba JIT — 50-100× faster than pure Python loops."""
    n = len(high)
    tr = np.empty(n)
    atr = np.empty(n)
    
    tr[0] = high[0] - low[0]
    for i in range(1, n):
        hl = high[i] - low[i]
        hc = abs(high[i] - close[i - 1])
        lc = abs(low[i] - close[i - 1])
        tr[i] = max(hl, hc, lc)
    
    atr[:period] = np.nan
    atr[period - 1] = np.mean(tr[:period])
    
    for i in range(period, n):
        atr[i] = (atr[i - 1] * (period - 1) + tr[i]) / period
    
    return atr

Custom composite indicators

Production trading systems often create custom indicators that combine multiple signals into a single score:

import pandas as pd
import numpy as np

def trend_strength_score(df: pd.DataFrame) -> pd.Series:
    """
    Composite score from 0 (no trend) to 100 (strong trend).
    Combines ADX, moving average alignment, and slope.
    """
    # ADX component (0-100 by definition)
    adx = compute_adx(df, period=14)
    
    # Moving average alignment: how many MAs are in order
    sma_20 = df["close"].rolling(20).mean()
    sma_50 = df["close"].rolling(50).mean()
    sma_100 = df["close"].rolling(100).mean()
    sma_200 = df["close"].rolling(200).mean()
    
    alignment = (
        (sma_20 > sma_50).astype(float) +
        (sma_50 > sma_100).astype(float) +
        (sma_100 > sma_200).astype(float)
    ) / 3 * 100
    
    # Slope of 50-day SMA (normalized)
    slope = sma_50.diff(5) / df["close"] * 1000
    slope_score = slope.clip(-50, 50) + 50  # map to 0-100
    
    # Weighted composite
    return 0.4 * adx + 0.3 * alignment + 0.3 * slope_score

def compute_adx(df: pd.DataFrame, period: int = 14) -> pd.Series:
    """Average Directional Index — measures trend strength regardless of direction."""
    high, low, close = df["high"], df["low"], df["close"]
    
    plus_dm = high.diff().clip(lower=0)
    minus_dm = (-low.diff()).clip(lower=0)
    
    # Zero out when the other DM is larger
    plus_dm[plus_dm < minus_dm] = 0
    minus_dm[minus_dm < plus_dm] = 0
    
    atr = compute_atr_series(high, low, close, period)
    
    plus_di = 100 * plus_dm.ewm(alpha=1/period, adjust=False).mean() / atr
    minus_di = 100 * minus_dm.ewm(alpha=1/period, adjust=False).mean() / atr
    
    dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
    adx = dx.ewm(alpha=1/period, adjust=False).mean()
    
    return adx

def compute_atr_series(high, low, close, period):
    tr = pd.concat([
        high - low,
        (high - close.shift(1)).abs(),
        (low - close.shift(1)).abs(),
    ], axis=1).max(axis=1)
    return tr.ewm(alpha=1/period, adjust=False).mean()

Regime-adaptive indicators

A fixed 14-period RSI works differently in low-volatility and high-volatility regimes. Adaptive indicators adjust their parameters dynamically:

def adaptive_rsi(prices: pd.Series, min_period: int = 6, max_period: int = 28) -> pd.Series:
    """RSI where the lookback period adapts to recent volatility."""
    vol = prices.pct_change().rolling(20).std()
    vol_rank = vol.rank(pct=True)
    
    # High volatility → shorter period (more responsive)
    # Low volatility → longer period (less noise)
    adaptive_period = (max_period - (vol_rank * (max_period - min_period))).round().astype(int)
    adaptive_period = adaptive_period.clip(min_period, max_period)
    
    result = pd.Series(np.nan, index=prices.index)
    
    for i in range(max_period, len(prices)):
        p = adaptive_period.iloc[i]
        window = prices.iloc[i - p : i + 1]
        delta = window.diff()
        gain = delta.clip(lower=0).mean()
        loss = -delta.clip(upper=0).mean()
        if loss == 0:
            result.iloc[i] = 100.0
        else:
            rs = gain / loss
            result.iloc[i] = 100 - (100 / (1 + rs))
    
    return result

Testing indicator robustness

Before trusting any indicator signal in a strategy, validate it:

1. Stability across instruments: does the signal work on stocks, ETFs, futures, and forex, or only on one asset class?
2. Parameter sensitivity: plot strategy performance as a heatmap over a range of indicator parameters. Smooth surfaces indicate robust signals; spiky surfaces indicate overfitting.
3. Regime analysis: segment history into bull, bear, and sideways periods. An indicator that only works in one regime has limited value.
4. Signal decay: measure how quickly the indicator’s predictive power fades. An RSI signal that predicts 1-day returns but not 5-day returns has a short shelf life.
5. Correlation with other indicators: if your RSI signal and MACD signal are 0.9 correlated, they add no diversification to a multi-signal strategy.

Practical architecture for indicator pipelines

from abc import ABC, abstractmethod
from typing import Any

class Indicator(ABC):
    @abstractmethod
    def compute(self, df: pd.DataFrame) -> pd.Series:
        ...
    
    @abstractmethod
    def name(self) -> str:
        ...

class RSIIndicator(Indicator):
    def __init__(self, period: int = 14):
        self.period = period
    
    def compute(self, df: pd.DataFrame) -> pd.Series:
        return wilder_rsi(df["close"], self.period)
    
    def name(self) -> str:
        return f"rsi_{self.period}"

class IndicatorPipeline:
    def __init__(self, indicators: list[Indicator]):
        self.indicators = indicators
    
    def run(self, df: pd.DataFrame) -> pd.DataFrame:
        result = df.copy()
        for ind in self.indicators:
            result[ind.name()] = ind.compute(df)
        return result

This pattern keeps indicator logic testable and composable — each indicator is a unit-tested class, and the pipeline orchestrates computation order and handles missing data consistently.

The one thing to remember: The real value of implementing indicators in Python is not the formulas themselves — it is the ability to test parameter sensitivity, detect regime dependence, and build custom composites that no off-the-shelf charting tool provides.