Built-in Calculations

ctx ships a library of calculation helpers — moving averages, oscillators, bands, accumulators. They are the recommended way to do math in an indicator: fast, updated one bar at a time, and they can never read the future.

The pattern

Always the same three steps:

Create the helper once in init, store it on self.
Read its value each bar in on_bar via .value (or a named output like .upper).
Guard for warm-up: finite-window helpers return None until they have enough bars. Check if value is not None: before using it.

class SMA(Indicator):
    name = "SMA"
    params = {"period": {"type": "int", "default": 20, "min": 2, "max": 500}}
    def init(self, ctx):
        self.mean = ctx.rolling_mean(source=ctx.close, period=ctx.params.period)
    def on_bar(self, ctx):
        ctx.plot.series("sma", value=self.mean.value,
                        label=f"SMA({ctx.params.period})")

A source argument accepts a price series (ctx.close, ctx.hl2, …), a source parameter (ctx.params.src), or another helper — that's how you compose. The signal a helper exposes is noted as Outputs below.

Moving averages & rolling statistics

ctx.rolling_mean(source, period, min_periods=None)

Simple moving average (arithmetic mean over a fixed window). min_periods defaults to period; lower it to emit a partial-window value sooner. Outputs: .value.

self.sma = ctx.rolling_mean(source=ctx.close, period=20)

ctx.rolling_std(source, period, ddof=0)

Rolling standard deviation. ddof=0 is population stdev (the Bollinger convention); pass ddof=1 for sample stdev. Outputs: .value.

ctx.rolling_min(source, period, min_periods=None) · ctx.rolling_max(source, period, min_periods=None)

Rolling window extremes (Donchian, Aroon, …). Outputs: .value.

self.hi = ctx.rolling_max(source=ctx.high, period=14)
self.lo = ctx.rolling_min(source=ctx.low, period=14)

ctx.ema(source, period, min_periods=None)

Exponential moving average (adjust=False). min_periods defaults to period; pass min_periods=1 to emit from the first bar. Outputs: .value.

ctx.weighted_mean(source, period)

Linearly-weighted moving average (weights 1, 2, …, period). Outputs: .value.

ctx.hull_ma(source, period=21)

Hull moving average — fast, low-lag. Outputs: .value.

ctx.wilder(source, period, min_periods=None)

Wilder / RMA smoothing (alpha = 1/period). The smoothing inside ATR and RSI; useful directly for Wilder-style averages. Outputs: .value.

ctx.rolling_linreg(source, period, min_periods=None)

Least-squares line fitted over the window. Outputs: .value (the line's value at the current bar — a linear-regression MA), .slope, .intercept.

self.lr = ctx.rolling_linreg(source=ctx.close, period=50)
# on_bar: self.lr.slope > 0  → rising regression channel

ctx.rolling_quantile(source, period, q, min_periods=None)

Rolling q-quantile (q between 0 and 1, linear interpolation). q=0.5 is a rolling median. Outputs: .value.

ctx.rolling_argmax(source, period, min_periods=None) · ctx.rolling_argmin(source, period, min_periods=None)

Bars since the window's most recent high / low (0.0 means the current bar is the extreme). The building block of Aroon and "bars since high" studies. Outputs: .value.

Pivots & fractals

ctx.pivot_high(source, left, right) · ctx.pivot_low(source, left, right)

Swing detection with left bars before and right bars after. A pivot is only known once the right bars have printed, so the helper emits it on the confirming bar. Outputs: .value (the pivot price on the bar it is confirmed, otherwise None) and .offset (how many bars back the pivot itself sits — equal to right).

Draw the marker at the pivot's own bar with ctx.bar_time_ago(...):

def init(self, ctx):
    self.ph = ctx.pivot_high(source=ctx.high, left=5, right=5)
def on_bar(self, ctx):
    v = self.ph.value
    if v is not None:
        ctx.plot.shape("ph", price=v, shape="triangle_down", color="#ef4444",
                       position="above_bar",
                       time_ms=ctx.bar_time_ago(int(self.ph.offset)))

ctx.fractal_up(source=None) · ctx.fractal_down(source=None)

Bill Williams fractals — a pivot with left=right=2. source defaults to ctx.high / ctx.low. Outputs: .value, .offset.

Volatility

ctx.true_range()

Per-bar true range: max(high−low, |high−prev_close|, |low−prev_close|). First bar emits high − low. Outputs: .value. No arguments.

ctx.atr(period)

Average True Range — Wilder-smoothed true range. Outputs: .value. Takes no source (it reads high/low/close itself).

self.atr = ctx.atr(period=14)

ctx.bollinger_bands(source, period=20, std=2.0)

Rolling mean ± std × rolling standard deviation. Outputs: .upper, .mid (alias .middle), .lower, and .values() which returns (upper, mid, lower) in one call.

def init(self, ctx):
    self.bb = ctx.bollinger_bands(source=ctx.close, period=20, std=2.0)
def on_bar(self, ctx):
    if self.bb.mid is None:
        return
    ctx.plot.series("bb_u", value=self.bb.upper, color="#9e9e9e")
    ctx.plot.series("bb_m", value=self.bb.mid,   color="#2196f3")
    ctx.plot.series("bb_l", value=self.bb.lower, color="#9e9e9e")

ctx.keltner_channels(period=20, atr_period=10, multiplier=2.0, source=None)

EMA midline with ATR-scaled bands. source defaults to the close. Outputs: .upper, .mid (alias .middle), .lower.

Momentum

ctx.rsi(source, period)

Relative Strength Index (0–100). Outputs: .value.

def init(self, ctx):
    self.rsi = ctx.rsi(source=ctx.close, period=14)
def on_bar(self, ctx):
    ctx.plot.series("rsi", value=self.rsi.value, pane="sub", label="RSI")

ctx.macd(source, fast=12, slow=26, signal=9)

Moving Average Convergence/Divergence. Outputs: .macd (the MACD line, fast EMA − slow EMA), .signal (EMA of the MACD line), .histogram (MACD − signal).

def init(self, ctx):
    self.macd = ctx.macd(source=ctx.close, fast=12, slow=26, signal=9)
def on_bar(self, ctx):
    ctx.plot.series("macd", value=self.macd.macd,      pane="sub")
    ctx.plot.series("sig",  value=self.macd.signal,    pane="sub")
    ctx.plot.series("hist", value=self.macd.histogram, pane="sub",
                    style="histogram")

📌 The MACD line output is .macd (not .line). Bollinger's centreline is .mid (with .middle as an alias). Getting these names right is the most common source of AttributeError.

Accumulators (from an anchor)

ctx.cumulative_sum(source, *, anchor="data_start", session=None)

Running sum (OBV, A/D line, cumulative deltas). anchor is "data_start" (from the first bar of the dataset) or "session" (restarts at every session boundary — pass the session= helper it should follow). Both keyword-only. Outputs: .value.

self.cvol = ctx.cumulative_sum(source=ctx.volume)
# Session-anchored: restarts at each London open
self.sess = ctx.session_window(anchor="london")
self.svol = ctx.cumulative_sum(source=ctx.volume, anchor="session", session=self.sess)

ctx.cumulative_ratio(numerator, denominator, *, anchor="data_start", session=None)

Running Σnumerator / Σdenominator — the shape of an anchored VWAP. Same anchor / session options. Outputs: .value.

# Daily VWAP
self.day  = ctx.session_window(anchor="daily")
self.vwap = ctx.cumulative_ratio(
    numerator=ctx.derived(lambda c: float(c.hlc3) * float(c.volume)),
    denominator=ctx.volume, anchor="session", session=self.day)

⚠️ Keep running totals in helpers, not on self. Charts are computed in windows, so a hand-rolled self.total += x restarts at the window edge and the line silently steps. The accumulators above carry their anchor across windows for you.

Window & event helpers

ctx.session_window(anchor)

UTC session-boundary detector. anchor is one of "daily", "asia", "london", "new_york". Outputs: .in_session (bool), .just_started (bool — True on the first bar of a new session), .start_time (epoch ms of the current session's start), .current_end_time (epoch ms of its scheduled end — handy for drawing a session box before the session is over).

def init(self, ctx):
    self.sess = ctx.session_window(anchor="london")
def on_bar(self, ctx):
    if self.sess.just_started:
        ctx.plot.shape("open", price=float(ctx.open),
                       shape="flag", color="#e58b39")

ctx.histogram(num_bins, value_area_pct=0.70)

A price-binned volume accumulator that drives ctx.plot.volume_profile. Unlike the others it is not auto-driven — you feed it each bar:

.add(*, high, low, close, volume, direction) — direction is "up" or "down". All arguments are keyword-only.
.snapshot(*, time_ms) — an immutable copy for plotting.
.reset() — empty it (start a new profile at a session boundary).
Read-back: .poc_price, .value_area_low, .value_area_high, .total_volume — so you can also plot the point of control or value-area edges as ordinary series.

def init(self, ctx):
    self.hist = ctx.histogram(num_bins=24)
    self.sess = ctx.session_window(anchor="daily")
def on_bar(self, ctx):
    self.hist.add(
        high=float(ctx.high), low=float(ctx.low), close=float(ctx.close),
        volume=float(ctx.volume),
        direction="up" if ctx.close >= ctx.open else "down",
    )
    if self.sess.just_started and not ctx.is_first_bar:
        ctx.plot.volume_profile(
            "vp", histogram=self.hist,
            anchor=(self.sess.start_time, ctx.bar_time),
        )

Composing helpers

A helper's source can be another helper — the framework chains them and sizes history automatically:

def init(self, ctx):
    self.bandwidth = ctx.rolling_std(source=ctx.close, period=20, ddof=0)
    self.bw_avg    = ctx.rolling_mean(source=self.bandwidth, period=50,
                                      min_periods=10)
def on_bar(self, ctx):
    ctx.plot.series("bw", value=self.bw_avg.value, pane="sub")

If any input helper is still in warm-up and returns None, the composed helper holds and returns None too — so a single if value is not None: guard at the point of use is enough.

ctx.derived(fn) — your own per-bar source

When the input you want isn't a price series or a helper, wrap a small function. It receives ctx and returns a number (or None), and any helper accepts it as source=:

def init(self, ctx):
    body = ctx.derived(lambda c: abs(float(c.close) - float(c.open)))
    self.avg_body = ctx.rolling_mean(source=body, period=20)