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
Indicatorlayer —Indicator.rsi(data, period),Indicator.macd(data, fast, slow, signal)andIndicator.bollinger(data, period, std_dev)are the parity-true path (no import). They call the sametesstrade_corekernels 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 boundedsdk.candles[-N:]window so each call staysO(1)in history length (the O(1) rule).ATR is not in the
Indicatorcatalogue — there is noIndicator.atr. For a chart-matching ATR usepandas_ta(ta.atr) or the nativeatr_last/_atrhelper 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:
# 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
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 outImplementation -- last point (approximation)
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:
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, keepavg_gain/avg_lossinsdk.stateas above. SeeIndicator.
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 lineThe 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):
# 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
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, histDetecting a histogram cross
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] > 0A complete MACD template is in the MACD Momentum strategy.
Incremental MACD
EMAs are naturally incremental. Cache each one in sdk.state:
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"], histSame 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 takehist[-1]. For a bit-identical recursive value, keep the three EMAs insdk.stateas above.
Bollinger Bands
Moving average plus or minus N standard deviations. Measures volatility.
In the catalogue. Bollinger Bands is one of the injected
Indicatormethods: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 staysO(1)in history length. The reference implementation below stays useful for reading or forking the math (or usepandas_ta.bbands).
The parity-true path (chart math)
# 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
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, lowerUsage -- reversion when the band is touched
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
Indicatorcatalogue. There is noIndicator.atr. For a chart-matching ATR series, usepandas_ta(ta.atr); for a single trailing value per bar, use the native_atrhelper below (Wilder smoothing over a bounded window,O(period)per bar, no import).
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 atrFor a chart-matching series in the df= branch, pandas_ta is the way:
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:
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) / periodCommon usage: stop = close - 2 * _atr(sdk.candles). _atr reads a bounded window, so it is safe to call every bar.
Next steps
- Anatomy of a custom indicator -- the canonical file shape: colors and params first, then math, then the dispatcher.
Indicator— the native layer -- the injected math brain: the five methods (rsi/macd/bollingeramong them), the O(1) rule, and how they match the chart (no import).- Implementing SMA/EMA -- the foundation for the indicators above.
- SMA Crossover, RSI Mean Reversion, MACD Momentum -- templates using the indicators on this page.