Indicator examples¶
Each example is a real file in lumibot/example_indicators/. LumiBot’s test
suite (tests/test_example_indicators.py) runs every one: values against a
direct calculation, live updates against a full recompute, forming bars
against rollback, and a check that no value uses a future bar. They are
teaching examples, not trading advice.
Use one in a strategy:
from lumibot.example_indicators import up_down_ratio # registers "up_down_ratio"
ratio = self.indicators.get("up_down_ratio", asset, length=14)
1. Donchian channel (on the price chart)¶
Highest high, lowest low and the midline of the last length bars.
"""Donchian channel midline: the middle of the highest high and lowest low.
Draws on the price chart (``overlay = True``).
"""
from collections import deque
from lumibot.indicators import Indicator, InputSpec, OutputSpec, register
@register
class DonchianMidline(Indicator):
key = "donchian_mid"
version = 1
title = "Donchian Channel Midline"
short_title = "DCM"
overlay = True # same pane as the price
inputs = [InputSpec("length", "int", 20, min=1, label="Length")]
outputs = [
OutputSpec("upper", "Upper", "line", "#22C55E"),
OutputSpec("mid", "Midline", "line", "#F59E0B"),
OutputSpec("lower", "Lower", "line", "#EF4444"),
]
def init_state(self):
n = self.inputs["length"]
return {"highs": deque(maxlen=n), "lows": deque(maxlen=n)}
def step(self, state, bar):
state["highs"].append(bar.high)
state["lows"].append(bar.low)
if len(state["highs"]) < self.inputs["length"]:
return {"upper": None, "mid": None, "lower": None}
upper, lower = max(state["highs"]), min(state["lows"])
return {"upper": upper, "mid": (upper + lower) / 2, "lower": lower}
2. Up/down ratio (RSI-style, own pane)¶
The share of recent movement that went up, 0 to 100, with plain sums.
"""Up/down ratio: an RSI-style oscillator from 0 to 100.
It is the share of recent price movement that went up:
``100 * gains / (gains + losses)`` over the last ``length`` bars, using plain
sums (Cutler's variant of RSI). It draws in its own pane with guide lines at
30 and 70.
"""
from collections import deque
from lumibot.indicators import Indicator, InputSpec, Level, OutputSpec, register
@register
class UpDownRatio(Indicator):
key = "up_down_ratio"
version = 1
title = "Up/Down Ratio"
short_title = "UDR"
overlay = False # its own pane under the price chart
inputs = [
InputSpec("length", "int", 14, min=1, label="Length"),
InputSpec("source", "source", "close", label="Source"),
]
outputs = [OutputSpec("ratio", "Ratio", "line", "#7C3AED")]
levels = [Level(70, "High"), Level(30, "Low")]
def init_state(self):
return {"prev": None, "moves": deque(maxlen=self.inputs["length"]), "gains": 0.0, "losses": 0.0}
def step(self, state, bar):
price = bar.source(self.inputs["source"])
prev, state["prev"] = state["prev"], price
if prev is None:
return {"ratio": None}
moves = state["moves"]
if len(moves) == moves.maxlen: # drop the oldest move from the running sums
old = moves[0]
state["gains"] -= max(old, 0.0)
state["losses"] -= max(-old, 0.0)
change = price - prev
moves.append(change)
state["gains"] += max(change, 0.0)
state["losses"] += max(-change, 0.0)
if len(moves) < moves.maxlen:
return {"ratio": None}
total = state["gains"] + state["losses"]
return {"ratio": 50.0 if total <= 0 else 100.0 * state["gains"] / total}
3. Volume-weighted trend¶
How far the volume-weighted average sits above or below the plain average.
"""Volume-weighted trend: how far the volume-weighted average sits above or
below the plain average, in percent.
Positive means recent high-volume bars traded higher than the average bar
(buyers in control); negative means the opposite. Bars without volume are
skipped, never counted as zero.
"""
from collections import deque
from lumibot.indicators import Indicator, InputSpec, Level, OutputSpec, register
@register
class VolumeWeightedTrend(Indicator):
key = "vw_trend"
version = 1
title = "Volume-Weighted Trend"
short_title = "VWT"
overlay = False
inputs = [InputSpec("length", "int", 20, min=2, label="Length")]
outputs = [OutputSpec("trend", "Trend %", "histogram", "#0EA5E9")]
levels = [Level(0, "Zero", "solid")]
def init_state(self):
return {"window": deque(maxlen=self.inputs["length"])}
def step(self, state, bar):
if bar.volume is None:
return {"trend": None}
window = state["window"]
window.append((bar.close, bar.volume))
if len(window) < window.maxlen:
return {"trend": None}
volume = sum(v for _, v in window)
sma = sum(c for c, _ in window) / len(window)
if volume <= 0 or sma == 0:
return {"trend": None}
vwma = sum(c * v for c, v in window) / volume
return {"trend": 100.0 * (vwma - sma) / sma}
4. Funding rate from a public exchange API (outside data)¶
Explained on Using outside data in an indicator.
"""Perpetual futures funding rate from a public exchange API, as an indicator.
This shows how an indicator can use outside data. It downloads the funding
rate history of a perpetual swap from OKX's public REST API (no API key) and,
for each bar, returns the latest funding rate that was already settled at the
bar's time. It never uses a funding rate from after the bar, so the same code
is safe in a backtest and live.
The download is cached on the indicator object (not in ``state``): it is
outside data, not something computed from bars, so forming-bar copies do not
need to copy it. To use another exchange or a recorded file, pass a different
``loader`` (any function returning ``[(epoch_ms, rate), ...]``).
"""
from __future__ import annotations
import bisect
import time
import requests
from lumibot.indicators import Indicator, InputSpec, Level, OutputSpec, register
OKX_FUNDING_HISTORY_URL = "https://www.okx.com/api/v5/public/funding-rate-history"
def fetch_okx_funding_history(inst_id: str, pages: int = 3, session=None) -> list:
"""Funding history for ``inst_id`` (for example ``BTC-USDT-SWAP``), oldest
first, as ``[(funding_time_ms, rate), ...]``. Each page holds up to 100
funding events (about 33 days at 8-hour funding)."""
http = session or requests
rows, cursor = [], None
for _ in range(pages):
params = {"instId": inst_id, "limit": 100}
if cursor is not None:
params["after"] = cursor # OKX: return records older than this time
response = http.get(OKX_FUNDING_HISTORY_URL, params=params, timeout=10)
response.raise_for_status()
payload = response.json()
if str(payload.get("code")) != "0":
raise RuntimeError(f"OKX funding history error: {payload.get('msg') or payload.get('code')}")
page = payload.get("data") or []
if not page:
break
for item in page:
rate = item.get("realizedRate") or item.get("fundingRate")
if rate not in (None, ""):
rows.append((int(item["fundingTime"]), float(rate)))
cursor = page[-1]["fundingTime"]
return sorted(set(rows))
@register
class FundingRate(Indicator):
key = "funding_rate"
version = 1
title = "Funding Rate (OKX)"
short_title = "FUND"
overlay = False
inputs = [
InputSpec("inst_id", "string", "BTC-USDT-SWAP", label="Perpetual swap"),
InputSpec("pages", "int", 3, min=1, max=50, label="History pages"),
]
outputs = [OutputSpec("rate_pct", "Funding %", "histogram", "#F97316")]
levels = [Level(0, "Zero", "solid")]
#: Where funding history comes from. Replace it (in a subclass or with
#: ``indicator.loader = fn``) for another exchange or for tests.
loader = staticmethod(fetch_okx_funding_history)
#: Live charts refresh the download at most this often (seconds).
refresh_seconds = 300
def init_state(self):
self._times, self._rates, self._fetched_at = [], [], None
return {}
def _ensure_history(self, bar_ms: int) -> None:
stale = self._fetched_at is None or (
(not self._times or bar_ms > self._times[-1] + 8 * 3600 * 1000)
and time.monotonic() - self._fetched_at >= self.refresh_seconds
)
if stale:
rows = self.loader(self.inputs["inst_id"], self.inputs["pages"])
self._times = [t for t, _ in rows]
self._rates = [r for _, r in rows]
self._fetched_at = time.monotonic()
def step(self, state, bar):
if bar.time is None:
return {"rate_pct": None}
bar_ms = bar.time.value // 1_000_000
self._ensure_history(bar_ms)
i = bisect.bisect_right(self._times, bar_ms) - 1 # latest funding at or before the bar
return {"rate_pct": None if i < 0 else 100.0 * self._rates[i]}
5. Machine-learning direction classifier (numpy)¶
A logistic regression trained on past returns only, retrained every few
bars. Put any model inside an indicator the same way: train or load it in
state, predict in step.
"""A small machine-learning indicator: probability that the next bar closes up.
It trains a logistic regression (plain numpy, no extra packages) on the
returns of the bars it has already seen, and retrains every
``retrain_every`` bars. The features are the last ``lookback`` bar returns;
the label is whether the following bar went up. Training only uses bars at or
before the current one, so the indicator never sees the future.
This is a teaching example of how to put a model inside an indicator, not a
trading signal. Expect a value near 50 most of the time on real markets.
"""
from __future__ import annotations
from collections import deque
import numpy as np
from lumibot.indicators import Indicator, InputSpec, Level, OutputSpec, register
def _fit_logistic(x: np.ndarray, y: np.ndarray, iterations: int = 200, rate: float = 0.5, l2: float = 1e-2):
"""Gradient-descent logistic regression. Returns (weights, bias, mean, std)."""
mean, std = x.mean(axis=0), x.std(axis=0)
std[std == 0] = 1.0
z = (x - mean) / std
w, b = np.zeros(z.shape[1]), 0.0
for _ in range(iterations):
p = 1.0 / (1.0 + np.exp(-(z @ w + b)))
error = p - y
w -= rate * (z.T @ error / len(y) + l2 * w)
b -= rate * error.mean()
return w, b, mean, std
@register
class MLDirection(Indicator):
key = "ml_direction"
version = 1
title = "ML Next-Bar Direction"
short_title = "MLD"
overlay = False
inputs = [
InputSpec("lookback", "int", 5, min=1, max=50, label="Return lags"),
InputSpec("train_window", "int", 200, min=20, label="Training bars"),
InputSpec("retrain_every", "int", 20, min=1, label="Retrain every N bars"),
]
outputs = [OutputSpec("prob_up", "Probability up %", "line", "#10B981")]
levels = [Level(50, "Coin flip")]
def init_state(self):
n = self.inputs["train_window"] + self.inputs["lookback"] + 1
return {"closes": deque(maxlen=n), "model": None, "since_train": 0}
def step(self, state, bar):
closes = state["closes"]
closes.append(bar.close)
lookback = self.inputs["lookback"]
returns = np.diff(np.log(np.asarray(closes, dtype=float)))
if len(returns) < lookback:
return {"prob_up": None}
state["since_train"] += 1
samples = len(returns) - lookback
if samples >= 2 * lookback + 10 and (state["model"] is None or state["since_train"] >= self.inputs["retrain_every"]):
# row k: returns[k:k+lookback] -> did returns[k+lookback] go up?
x = np.lib.stride_tricks.sliding_window_view(returns, lookback)[:samples]
y = (returns[lookback:] > 0).astype(float)
if 0 < y.mean() < 1:
state["model"] = _fit_logistic(x, y)
state["since_train"] = 0
if state["model"] is None:
return {"prob_up": None}
w, b, mean, std = state["model"]
latest = (returns[-lookback:] - mean) / std
return {"prob_up": 100.0 / (1.0 + np.exp(-(latest @ w + b)))}
To use a model you trained elsewhere (scikit-learn, LightGBM, ONNX), load it
once on the indicator object and call it in step with features from
closed bars only.