Stream an indicator live from a crypto exchange

You do not need a hosted service or an API key to get live indicator values. LumiBot reads public candles from an exchange through CCXT and updates your indicators on every poll.

Run it:

python -m lumibot.example_indicators.live_crypto_stream kraken BTC/USD 15m

Output (one line per update; forming lines change until the candle closes):

2026-10-25 20:15 forming close=67412.10 rsi=55.2 ema=67350.8
2026-10-25 20:15 closed  close=67420.00 rsi=55.6 ema=67351.6

The code

"""Stream indicators live from a public crypto exchange, in plain Python.

No hosted service, no API key: CCXT reads public candles from the exchange,
:func:`~lumibot.tools.candles.choose_candle_source` uses the exchange's own
timeframe when it has one (otherwise 1 minute candles are built up with
:class:`~lumibot.tools.candles.CandleBuilder`), and every indicator gets a
forming value on each poll and a committed value when the candle closes.

Run it::

    python -m lumibot.example_indicators.live_crypto_stream kraken BTC/USD 15m

Press Ctrl+C to stop.
"""

from __future__ import annotations

import sys
import time

from lumibot.indicators import registry
from lumibot.tools.candles import CandleBuilder, choose_candle_source


def stream(exchange, symbol: str, timeframe: str, indicators: dict, *, history: int = 300, polls: int | None = None,
           poll_seconds: float = 5.0, sleep=time.sleep):
    """Yield ``(candle, values)`` for every candle update.

    ``exchange`` is a CCXT exchange object (or anything with ``timeframes``
    and ``fetch_ohlcv``). ``indicators`` maps a name to an indicator instance.
    ``values`` maps each name to its outputs. ``candle.closed`` tells you
    whether the values are final (closed candle) or live (forming).
    ``polls=None`` runs until stopped.
    """
    source_tf, build = choose_candle_source(timeframe, exchange.timeframes or {timeframe: timeframe})
    builder = CandleBuilder(timeframe, source_timeframe=source_tf)

    # 1. History: every candle except the newest one is closed.
    limit = history * (builder.timeframe // builder.source_timeframe) if build else history
    rows = exchange.fetch_ohlcv(symbol, source_tf, limit=limit)
    for row in rows[:-1]:
        for candle in builder.update(row, closed=True):
            if candle.closed:
                for indicator in indicators.values():
                    indicator.update(candle, closed=True)
    last_seen = rows[-1][0] if rows else None
    if rows:
        rows = rows[-1:]

    # 2. Live: poll the newest source candles. The last one is still forming.
    count = 0
    while polls is None or count < polls:
        for i, row in enumerate(rows):
            forming_source = i == len(rows) - 1
            for candle in builder.update(row, closed=not forming_source):
                values = {name: ind.update(candle, closed=candle.closed) for name, ind in indicators.items()}
                yield candle, values
        count += 1
        if polls is not None and count >= polls:
            break
        sleep(poll_seconds)
        fresh = exchange.fetch_ohlcv(symbol, source_tf, limit=3)
        rows = [row for row in fresh if last_seen is None or row[0] >= last_seen]
        if rows:
            last_seen = rows[-1][0]


def main(argv=None) -> None:  # pragma: no cover - manual entry point
    import ccxt

    args = list(sys.argv[1:] if argv is None else argv)
    exchange_id, symbol, timeframe = (args + ["kraken", "BTC/USD", "15m"][len(args):])[:3]
    exchange = getattr(ccxt, exchange_id)({"enableRateLimit": True})
    indicators = {"rsi": registry.create("rsi", length=14), "ema": registry.create("ema", length=20)}
    try:
        for candle, values in stream(exchange, symbol, timeframe, indicators):
            state = "closed " if candle.closed else "forming"
            print(f"{candle.time:%Y-%m-%d %H:%M} {state} close={candle.close:.2f} "
                  f"rsi={values['rsi']['rsi']} ema={values['ema']['ema']}")
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":  # pragma: no cover
    main()

Native candles first, built candles as a fallback

choose_candle_source(timeframe, exchange.timeframes) returns:

  • the exchange’s own timeframe when it has one (("15m", False)), or

  • ("1m", True) when it does not, and LumiBot builds the candles from 1 minute candles.

Native candles are the exchange’s own numbers, so prefer them.

The candle builder

CandleBuilder turns small candles into bigger ones, one at a time:

  • Buckets are cut on UTC boundaries, like exchanges do: 15 minute candles start at :00, :15, :30 and :45 UTC; daily candles at UTC midnight.

  • The forming candle updates on every incoming bar, wicks included.

  • When the exchange re-sends the current minute, it replaces the earlier copy instead of counting it twice.

  • The candle locks when a bar from the next bucket arrives, when the bucket’s last bar arrives closed, or when you call close_through(now) after the bucket ends (quiet markets).

  • Nothing is invented. An empty bucket makes no candle; missing volume stays missing.

For a DataFrame of 1 minute bars, build_candles(df, "15m") does the same in one call, also on UTC boundaries.

from lumibot.tools.candles import CandleBuilder

builder = CandleBuilder("15m")                 # from 1 minute bars
for candle in builder.update(minute_bar):      # minute_bar: dict or CCXT row
    values = rsi.update(candle, closed=candle.closed)

Which exchange?

Any CCXT exchange with public candles works. Some exchanges refuse connections from some countries; if one fails, try another (Kraken, Coinbase and OKX all publish 1 minute and 15 minute candles). LumiBot’s tests run this script against a fake exchange in CCXT’s real data shape, and once by hand against Kraken’s public API.

class lumibot.tools.candles.CandleBuilder(timeframe, source_timeframe='1m')

Build timeframe candles one source bar at a time.

update(bar, closed=False) takes one source bar (a dict with t/o/h/l/c/v or time/open/..., an object with those attributes, or a CCXT [ms, o, h, l, c, v] row). The same source bar may be sent again while it is still forming (exchanges re-send the current minute); it replaces the earlier copy instead of being counted twice.

It returns a list of Candle: any candles that just closed, then the current forming candle (if any). A candle locks when

  • a source bar from a later bucket arrives,

  • the bucket’s last source bar arrives with closed=True, or

  • close_through() is called with a time at or after the bucket end.

Buckets are cut on UTC boundaries. source_timeframe must divide timeframe; when they are equal (a native exchange candle) bars pass straight through as forming/closed candles.

close_through(now) → list

Lock the forming candle if now is at or past its bucket end.

Call it from a clock when the exchange may not send another bar soon (quiet markets). Returns the closed candle in a list, or [].

property forming: Candle | None

The candle still being built, or None.

reset() → None

Forget everything (for example after a reconnect gap).

update(bar, closed: bool = False) → list

Apply one source bar. See the class docstring.

lumibot.tools.candles.choose_candle_source(timeframe, native_timeframes: Iterable | Mapping, fallback='1m') → tuple[str, bool]

Pick where candles for timeframe should come from.

native_timeframes is what the exchange serves directly (for a CCXT exchange, exchange.timeframes; a dict’s keys are used). Returns (source_timeframe, needs_building):

  • the target itself when the exchange has it natively: ("15m", False);

  • otherwise fallback (1 minute) when available: ("1m", True);

  • otherwise the largest native timeframe that divides the target evenly.

Raises ValueError when nothing native can build the target.

lumibot.tools.candles.build_candles(df: DataFrame, timeframe, *, include_forming: bool = True, source_timeframe='1m') → DataFrame

Build timeframe candles from finer OHLCV bars, cut on UTC boundaries.

The index may be in any time zone (naive means UTC); the result keeps the input’s zone. With include_forming=False the last candle is dropped unless its bucket is complete, judged by the last source bar’s end time (source_timeframe long).