Skip to content
Single .md

RSI, MACD and Bollinger Bands

Pure Python implementations of three indicators common in quant strategies, with no dependency on pandas_ta.

One math brain — reached with no import. RSI, MACD and Bollinger Bands are all in the injected Indicator layer — Indicator.rsi(data, period), Indicator.macd(data, fast, slow, signal) and Indicator.bollinger(data, period, std_dev) are the parity-true path (no import). They call the same tesstrade_core kernels the live chart renders with, so what you compute equals what the chart draws (Wilder RSI; pandas_ta-style MACD; SMA-based bands). In the per-bar branch, feed a bounded sdk.candles[-N:] window so each call stays O(1) in history length (the O(1) rule).

ATR is not in the Indicator catalogue — there is no Indicator.atr. For a chart-matching ATR use pandas_ta (ta.atr) or the native atr_last / _atr helper below.

The pure-Python references on this page remain valuable for two reasons: to read and extend the math, and to cover ATR and any variant outside the catalogue. New to the file layout? Start with Anatomy of a custom indicator — colors and params first, then the math.

RSI -- Relative Strength Index

Oscillator between 0 and 100. Above 70 = overbought; below 30 = oversold.

Formula

gain[i]  = max(close[i] - close[i-1], 0)
loss[i]  = max(close[i-1] - close[i], 0)

avg_gain[period] = mean(gains[1..period])
avg_loss[period] = mean(losses[1..period])

For i > period (Wilder smoothing):
  avg_gain[i] = (avg_gain[i-1] * (period-1) + gain[i]) / period
  avg_loss[i] = (avg_loss[i-1] * (period-1) + loss[i]) / period

rs = avg_gain / avg_loss
rsi = 100 - 100 / (1 + rs)

The parity-true path (chart math)

If you just want RSI that matches the chart, reach for the injected Indicator global (no import) — it runs the same Wilder kernel the chart renders with, so there is no drift between your series and the drawn line:

python
# df= branch (whole series, runs once):
rsi = Indicator.rsi(df, 14)                     # list, len(df), None during warm-up

# per-bar branch (last value over a bounded window — O(1) in history length):
rsi_now = Indicator.rsi(sdk.candles[-300:], 14)[-1]

The reference implementations below reproduce that same Wilder math in pure Python — keep them when you want to read, tweak, or fork the formula.

Implementation -- series

python
def rsi_series(closes, period=14):
    """Wilder RSI. Returns None for the first `period` points."""
    if len(closes) < period + 1:
        return [None] * len(closes)

    out = [None] * period
    gains = 0.0
    losses = 0.0

    # Seed: simple means of the first `period` deltas
    for i in range(1, period + 1):
        delta = closes[i] - closes[i - 1]
        if delta > 0:
            gains += delta
        else:
            losses += -delta
    avg_gain = gains / period
    avg_loss = losses / period
    if avg_loss == 0:
        out.append(100.0)
    else:
        rs = avg_gain / avg_loss
        out.append(100.0 - 100.0 / (1.0 + rs))

    # Wilder smoothing for the rest
    for i in range(period + 1, len(closes)):
        delta = closes[i] - closes[i - 1]
        gain = delta if delta > 0 else 0.0
        loss = -delta if delta < 0 else 0.0
        avg_gain = (avg_gain * (period - 1) + gain) / period
        avg_loss = (avg_loss * (period - 1) + loss) / period
        if avg_loss == 0:
            out.append(100.0)
        else:
            rs = avg_gain / avg_loss
            out.append(100.0 - 100.0 / (1.0 + rs))

    return out

Implementation -- last point (approximation)

python
def rsi_last_approx(closes, period=14):
    """Simplified RSI: arithmetic mean of deltas over the last period.
    Less precise than Wilder, but O(period). Suitable for on_bar_strategy."""
    if len(closes) < period + 1:
        return None
    gains = 0.0
    losses = 0.0
    for i in range(len(closes) - period, len(closes)):
        delta = closes[i] - closes[i - 1]
        if delta > 0:
            gains += delta
        else:
            losses += -delta
    if losses == 0:
        return 100.0
    rs = (gains / period) / (losses / period)
    return 100.0 - 100.0 / (1.0 + rs)

Practical difference: Wilder is more precise for canonical RSI; the simplified version is off by ~1-3 points in most cases. To match the chart (and TradingView), use rsi_series and take [-1] — or, for the same Wilder kernel the chart renders with, call Indicator.rsi(sdk.candles[-300:], 14)[-1] (no import).

Incremental RSI with sdk.state

The most efficient approach: store avg_gain and avg_loss in state and update on every candle:

python
def rsi_incremental(sdk, period=14):
    candles = sdk.candles  # sdk.state is already a persistent dict — no init needed
    if len(candles) < period + 1:
        return None

    if "rsi_ag" not in sdk.state or "rsi_al" not in sdk.state:
        # Seed once, from the first `period` deltas. This IS the canonical
        # Wilder value at index `period` — so on the seeding frame we return
        # it WITHOUT applying a Wilder step. Applying one here would
        # double-count the last delta and permanently shift every later RSI
        # by one smoothing step relative to rsi_series()/Indicator.rsi().
        # The full comprehension runs only on this one seeding frame.
        seed_closes = [c["close"] for c in candles[:period + 1]]
        gains = losses = 0.0
        for i in range(1, period + 1):
            d = seed_closes[i] - seed_closes[i - 1]
            if d > 0: gains += d
            else: losses += -d
        sdk.state["rsi_ag"] = gains / period
        sdk.state["rsi_al"] = losses / period
    else:
        # Steady state: O(1) — only the last two closes matter. No need to
        # materialise the whole `closes` list every bar (that would be O(n)
        # per bar, O(n²) over the backtest).
        d = candles[-1]["close"] - candles[-2]["close"]
        gain = d if d > 0 else 0.0
        loss = -d if d < 0 else 0.0
        sdk.state["rsi_ag"] = (sdk.state["rsi_ag"] * (period - 1) + gain) / period
        sdk.state["rsi_al"] = (sdk.state["rsi_al"] * (period - 1) + loss) / period

    if sdk.state["rsi_al"] == 0:
        return 100.0
    rs = sdk.state["rsi_ag"] / sdk.state["rsi_al"]
    return 100.0 - 100.0 / (1.0 + rs)

Two shortcuts, both no import: for the exact Wilder value with no bookkeeping, read a bounded window — Indicator.rsi(sdk.candles[-300:], 14)[-1] (converges to the full-history RSI). For a bit-identical recursive value, keep avg_gain/avg_loss in sdk.state as above. See Indicator.

MACD -- Moving Average Convergence Divergence

Three lines: MACD line, Signal line, and histogram.

Formula

MACD line   = EMA(close, fast) - EMA(close, slow)     # typical: fast=12, slow=26
Signal line = EMA(MACD line, signal)                   # typical: signal=9
Histogram   = MACD line - Signal line

The parity-true path (chart math)

Indicator.macd returns the three lines from the same kernel the chart draws (pandas_ta-style MACD), so the histogram you cross on equals the histogram on screen (no import):

python
# df= branch (whole series):
macd, signal, hist = Indicator.macd(df, 12, 26, 9)        # three lists, len(df)

# per-bar branch — the last two histogram values over a bounded window:
_, _, hist = Indicator.macd(sdk.candles[-300:], 12, 26, 9)
hist_prev, hist_curr = hist[-2], hist[-1]

The pure-Python reference below reproduces the same formula when you want to read or extend it.

Implementation

python
def ema_series(values, period):
    """Simple EMA (see SMA/EMA docs)."""
    if not values:
        return []
    alpha = 2.0 / (period + 1.0)
    out = [float(values[0])]
    for v in values[1:]:
        out.append(alpha * v + (1.0 - alpha) * out[-1])
    return out


def macd_series(closes, fast=12, slow=26, signal=9):
    """Returns (macd_line, signal_line, hist) -- all aligned with closes."""
    fast_ema = ema_series(closes, fast)
    slow_ema = ema_series(closes, slow)
    macd_line = [f - s for f, s in zip(fast_ema, slow_ema)]
    signal_line = ema_series(macd_line, signal)
    hist = [m - s for m, s in zip(macd_line, signal_line)]
    return macd_line, signal_line, hist

Detecting a histogram cross

python
def macd_hist_cross_up(closes, fast=12, slow=26, signal=9):
    _, _, hist = macd_series(closes, fast, slow, signal)
    if len(hist) < 2:
        return False
    return hist[-2] <= 0 and hist[-1] > 0

A complete MACD template is in the MACD Momentum strategy.

Incremental MACD

EMAs are naturally incremental. Cache each one in sdk.state:

python
def macd_incremental(sdk, fast=12, slow=26, signal=9):
    if not isinstance(sdk.state, dict):
        sdk.state = {}

    close = sdk.candles[-1]["close"]
    alpha_fast = 2.0 / (fast + 1.0)
    alpha_slow = 2.0 / (slow + 1.0)
    alpha_sig = 2.0 / (signal + 1.0)

    # Seed
    if "ema_fast" not in sdk.state:
        sdk.state["ema_fast"] = close
        sdk.state["ema_slow"] = close
        sdk.state["signal"] = 0.0
        return None

    sdk.state["ema_fast"] = alpha_fast * close + (1 - alpha_fast) * sdk.state["ema_fast"]
    sdk.state["ema_slow"] = alpha_slow * close + (1 - alpha_slow) * sdk.state["ema_slow"]
    macd_value = sdk.state["ema_fast"] - sdk.state["ema_slow"]
    sdk.state["signal"] = alpha_sig * macd_value + (1 - alpha_sig) * sdk.state["signal"]
    hist = macd_value - sdk.state["signal"]

    return macd_value, sdk.state["signal"], hist

Same two shortcuts as RSI, both no import: for the exact kernel with no manual EMA caching, read a bounded window — Indicator.macd(sdk.candles[-300:], 12, 26, 9) gives (macd, signal, hist) and you take hist[-1]. For a bit-identical recursive value, keep the three EMAs in sdk.state as above.

Bollinger Bands

Moving average plus or minus N standard deviations. Measures volatility.

In the catalogue. Bollinger Bands is one of the injected Indicator methods: Indicator.bollinger(data, period, std_dev) returns (upper, middle, lower) from the same kernel the chart renders with — no import. In the per-bar branch, feed a bounded window (Indicator.bollinger(sdk.candles[-300:], 20, 2.0)) so it stays O(1) in history length. The reference implementation below stays useful for reading or forking the math (or use pandas_ta.bbands).

The parity-true path (chart math)

python
# df= branch (whole series):
upper, middle, lower = Indicator.bollinger(df, 20, 2.0)   # three lists, len(df)

# per-bar branch (last values over a bounded window):
upper, middle, lower = Indicator.bollinger(sdk.candles[-300:], 20, 2.0)
u, m, l = upper[-1], middle[-1], lower[-1]

Formula

middle = SMA(close, period)            # typically period=20
std    = stdev(close[-period:])
upper  = middle + (std_mult * std)      # typically std_mult=2.0
lower  = middle - (std_mult * std)

Implementation

python
def bbands_series(closes, period=20, std_mult=2.0):
    """Returns (middle, upper, lower) -- all aligned with closes."""
    middle = []
    upper = []
    lower = []
    for i in range(len(closes)):
        if i + 1 < period:
            middle.append(None)
            upper.append(None)
            lower.append(None)
            continue
        window = closes[i - period + 1 : i + 1]
        mean = sum(window) / period
        var = sum((x - mean) ** 2 for x in window) / period
        std = var ** 0.5
        middle.append(mean)
        upper.append(mean + std_mult * std)
        lower.append(mean - std_mult * std)
    return middle, upper, lower

Usage -- reversion when the band is touched

python
def on_bar_strategy(sdk, params):
    period = int((params or {}).get("period", 20))
    std_mult = float((params or {}).get("std_mult", 2.0))

    closes = [c["close"] for c in sdk.candles]
    if len(closes) < period:
        return

    window = closes[-period:]
    mean = sum(window) / period
    var = sum((x - mean) ** 2 for x in window) / period
    std = var ** 0.5
    upper = mean + std_mult * std
    lower = mean - std_mult * std
    close = closes[-1]

    if sdk.position == 0:
        if close <= lower:
            sdk.buy(action="buy_to_open", qty=1, order_type="market")
        elif close >= upper:
            sdk.sell(action="sell_short_to_open", qty=1, order_type="market")
    elif sdk.position > 0 and close >= mean:
        sdk.sell(action="sell_to_close", qty=abs(sdk.position), order_type="market")
    elif sdk.position < 0 and close <= mean:
        sdk.buy(action="buy_to_cover", qty=abs(sdk.position), order_type="market")

Classic bollinger-reversion strategy: buys when the lower band is touched, sells when price returns to the mean. (Reads only a bounded closes[-period:] window, so it is O(period) per bar — or swap the hand-rolled math for Indicator.bollinger(sdk.candles[-300:], period, std_mult) to get the chart-matching bands.)

ATR -- Average True Range

A volatility indicator, useful for dynamic stops. True Range = maximum of:

  • high - low
  • |high - previous_close|
  • |low - previous_close|

Not in the Indicator catalogue. There is no Indicator.atr. For a chart-matching ATR series, use pandas_ta (ta.atr); for a single trailing value per bar, use the native _atr helper below (Wilder smoothing over a bounded window, O(period) per bar, no import).

python
def _atr(candles, period=14):
    """Wilder ATR over a bounded window — O(period) per bar, no import."""
    rows = candles[-(period * 4):]                 # bounded window converges to full-history ATR
    if len(rows) < period + 1:
        return None
    trs = []
    for prev, cur in zip(rows, rows[1:]):
        trs.append(max(cur["high"] - cur["low"],
                       abs(cur["high"] - prev["close"]),
                       abs(cur["low"] - prev["close"])))
    atr = sum(trs[:period]) / period               # seed
    for tr in trs[period:]:                        # Wilder smoothing over the window
        atr = (atr * (period - 1) + tr) / period
    return atr

For a chart-matching series in the df= branch, pandas_ta is the way:

python
import pandas_ta as ta   # allowlisted; also pre-injected as `ta`

atr = ta.atr(df["high"], df["low"], df["close"], length=14)   # Series, len(df)
latest_atr = float(atr.iloc[-1])

A simpler (non-Wilder) last-point helper, if you only need a rough trailing value:

python
def atr_last(candles, period=14):
    """Simple mean of the last `period` true ranges. None if insufficient."""
    if len(candles) < period + 1:
        return None
    trs = []
    for i in range(len(candles) - period, len(candles)):
        h = candles[i]["high"]
        l = candles[i]["low"]
        cp = candles[i - 1]["close"]
        trs.append(max(h - l, abs(h - cp), abs(l - cp)))
    return sum(trs) / period

Common usage: stop = close - 2 * _atr(sdk.candles). _atr reads a bounded window, so it is safe to call every bar.

Next steps