Data Feeds¶

ml4t-live exposes five primary feed classes plus BarAggregator for resampling tick-style feeds. Each feed yields (timestamp, data, context) tuples through DataFeedProtocol.

AlpacaDataFeed¶

from ml4t.live import AlpacaDataFeed

feed = AlpacaDataFeed(
    api_key="...",
    secret_key="...",
    symbols=["AAPL", "BTC/USD"],
    data_type="bars",  # "bars", "quotes", or "trades"
    feed="iex",        # "iex" or "sip"
)

Use this feed for Alpaca-native stocks and crypto. Stock symbols and .../USD crypto symbols can be mixed in one feed. The feed itself does not implement a standalone reconnect loop; use LiveEngine(auto_recover=True) if you want watchdog-driven restart behavior.

IBDataFeed¶

from ml4t.live import IBBroker, IBDataFeed

broker = IBBroker(port=7497)
await broker.connect()

feed = IBDataFeed(
    broker.ib,
    symbols=["SPY", "QQQ"],
    exchange="SMART",
    currency="USD",
    tick_throttle_ms=1000,
)

IBDataFeed emits tick-style updates shaped like {symbol: {"price", "size"}}. Wrap it in BarAggregator if your strategy expects OHLCV bars. The feed does not own a reconnect loop; watchdog-driven stop/restart belongs in LiveEngine when enabled.

DataBentoFeed¶

Historical replay:

from ml4t.live import DataBentoFeed

feed = DataBentoFeed.from_file(
    "data/ES_202401.dbn",
    symbols=["ES.FUT"],
    replay_speed=10.0,
)

Live streaming:

feed = DataBentoFeed.from_live(
    api_key="...",
    dataset="GLBX.MDP3",
    schema="ohlcv-1s",
    symbols=["ES.c.0", "NQ.c.0"],
)

DataBentoFeed requires the optional databento package.

CryptoFeed¶

from ml4t.live import CryptoFeed

feed = CryptoFeed(
    exchange="binance",
    symbols=["BTC/USDT", "ETH/USDT"],
    timeframe="1m",
    stream_ohlcv=True,
)

CryptoFeed uses ccxt or ccxt.pro depending on availability and supports both trade streaming and OHLCV streaming.

OKXFundingFeed¶

from ml4t.live import OKXFundingFeed

feed = OKXFundingFeed(
    symbols=["BTC-USDT-SWAP", "ETH-USDT-SWAP"],
    timeframe="1m",  # also supports "1H", "4H", and "1D"
    poll_interval_seconds=5.0,
)

OKXFundingFeed combines OHLCV bars from /api/v5/market/candles with funding-rate context from /api/v5/public/funding-rate. Minute granularity is supported, and emitted timestamps align to UTC minute boundaries.

BarAggregator¶

Use BarAggregator to convert tick or sub-minute feeds into larger bars:

from ml4t.live import BarAggregator

feed = BarAggregator(
    source_feed=raw_feed,
    bar_size_minutes=1,
    flush_timeout_seconds=2.0,
)

Optional symbol filtering:

feed = BarAggregator(raw_feed, bar_size_minutes=5, assets=["SPY", "QQQ"])

Using a Feed With LiveEngine¶

engine = LiveEngine(strategy, safe_broker, feed)
await engine.connect()
await engine.run()

LiveEngine.connect() starts the feed for you, so normal engine usage does not require a manual feed.start() call. If you configure auto_recover=True, the engine watchdog can also stop and restart the broker/feed pair after feed_silent or broker_disconnected events.

Choosing a Feed¶

Use AlpacaDataFeed for Alpaca-native stocks and crypto.
Use IBDataFeed for direct market data from TWS or IB Gateway.
Use DataBentoFeed for replay or institutional market-data workflows.
Use CryptoFeed for exchange-agnostic crypto streaming through CCXT.
Use OKXFundingFeed for perpetual-swap strategies that depend on funding-rate context.
Use BarAggregator when your upstream feed is tick-oriented but your strategy expects bars.