Computing all indicators in one pipeline

Chapters 101-111 built each indicator as a standalone function. This chapter wires all of them into a single pipeline that takes raw OHLCV data and returns one DataFrame with every indicator computed — the input that chapters 113-117's confluence logic actually consumes.

# indicators/pipeline.py
import pandas as pd

def compute_all_indicators(df: pd.DataFrame) -> pd.DataFrame:
    """df: OHLCV DataFrame, datetime-indexed. Returns df with every indicator appended as columns."""
    out = df.copy()

    # Trend (ch 101)
    out["sma_20"] = sma(out["close"], 20)
    out["sma_50"] = sma(out["close"], 50)
    out["sma_200"] = sma(out["close"], 200)
    out["ema_9"] = ema(out["close"], 9)
    out["ema_21"] = ema(out["close"], 21)
    out["ema_50"] = ema(out["close"], 50)

    # MACD (ch 102)
    macd_df = macd(out)
    out["macd"], out["macd_signal"], out["macd_hist"] = macd_df["macd"], macd_df["signal"], macd_df["histogram"]

    # RSI (ch 103)
    out["rsi_14"] = rsi(out["close"], 14)

    # Bollinger Bands (ch 104)
    bb = bollinger_bands(out["close"], 20, 2.0)
    out["bb_upper"], out["bb_middle"], out["bb_lower"] = bb["upper"], bb["middle"], bb["lower"]
    out["bb_percent_b"] = percent_b(out["close"], bb)
    out["bb_bandwidth"] = bollinger_bandwidth(bb)

    # Stochastic (ch 105)
    stoch = stochastic(out)
    out["stoch_k"], out["stoch_d"] = stoch["k"], stoch["d"]

    # ADX/DMI (ch 106)
    adx_df = adx_dmi(out)
    out["adx"], out["plus_di"], out["minus_di"] = adx_df["adx"], adx_df["plus_di"], adx_df["minus_di"]

    # ATR / Supertrend (ch 107)
    out["atr_14"] = atr(out, 14)
    st = supertrend(out)
    out["supertrend"], out["supertrend_dir"] = st["supertrend"], st["direction"]

    # VWAP (ch 108, intraday only — skip for daily data)
    if is_intraday(out):
        out["vwap"] = vwap_per_session(out)

    # Volume (ch 110)
    out["obv"] = obv(out)

    return out

Performance — don't recompute the whole history on every new bar

def compute_incremental(df: pd.DataFrame, existing: pd.DataFrame, lookback_buffer: int = 250) -> pd.DataFrame:
    """Only recompute indicators for the tail of the series that changed, using enough
    trailing history (lookback_buffer) for the slowest indicator (e.g. SMA 200) to be valid."""
    tail = df.iloc[-lookback_buffer:]
    recomputed_tail = compute_all_indicators(tail)
    return pd.concat([existing.iloc[:-lookback_buffer], recomputed_tail])

Recomputing 200+ columns of indicators across years of history on every single new live tick is wasteful and, for anything running on 1-minute bars, can become a real latency problem — recompute only a trailing window large enough for your slowest indicator's period, and append.

Registry pattern — for adding/removing indicators without editing the pipeline function

INDICATOR_REGISTRY = {
    "sma_20": lambda df: sma(df["close"], 20),
    "rsi_14": lambda df: rsi(df["close"], 14),
    "adx": lambda df: adx_dmi(df)["adx"],
    # ... register every indicator here
}

def compute_selected(df: pd.DataFrame, names: list[str]) -> pd.DataFrame:
    out = df.copy()
    for name in names:
        out[name] = INDICATOR_REGISTRY[name](out)
    return out

Useful once you're running parameter sweeps (chapter 84) or only need a subset of indicators for a specific strategy — avoids paying the computation cost of the full 20+ indicator set when a strategy only uses three of them.

Validate every column before trusting the pipeline

def validate_indicator_output(df: pd.DataFrame) -> dict:
    return {col: {"nan_count": df[col].isna().sum(), "inf_count": (df[col] == float("inf")).sum()}
            for col in df.columns if df[col].dtype in ("float64", "int64")}

A silently NaN-heavy or inf-containing column (from a division edge case, chapter 103's zero-loss RSI trap, or insufficient warm-up history for a 200-period SMA) will quietly corrupt any confluence score built on top of it in chapter 114 — check this pipeline's output before building signal logic on it, not after a strategy misbehaves.

Next: 113 — Generic crossover/intersection detection