Connectors SDK Guide

Connect Reamer Py to any data source

A data source connector — a loader — is a plain Python function that fetches bars from wherever your data actually lives and calls reamer_py.write_bin(). That's the whole integration surface. write_bin and OhlcvBar are core reamer_py bindings, present in every install; nothing extra to pull in, no plugin system, no registration step, no subclassing.

No loader ships pre-installed with reamer_py, including Reamer Labs' own reference examples below. A loader is code you own, sitting in your own project, imported the normal Python way. That's true whether it's a vendor API you connect this afternoon or one of the three worked examples on this page — same shape, same friction, no exceptions. A "some sources are more built-in than others" story is exactly what this page exists to avoid.

The interface

Two things make up the entire contract between your loader and reamer_py — everything else on this page is how you get data into this shape, not part of the interface itself.

OhlcvBar — one bar. Every loader on this page builds a list of these, one per row of source data:

import reamer_py

bar = reamer_py.OhlcvBar()
bar.timestamp = "2024-01-15 09:30:00"   # "YYYY-MM-DD HH:MM:SS", literal UTC
bar.open = 100.0
bar.high = 101.5
bar.low = 99.25
bar.close = 101.0
bar.volume = 12_400.0

Once you have a list of these for one ticker, there are two ways to hand it to reamer_py — pick whichever fits how you're using the data:

Both are covered below: writing a file first, or passing bars directly.

Writing a file: write_bin

reamer_py.write_bin(bars, ticker) takes your list of OhlcvBars and writes them to disk under ticker's own file — a direct, synchronous function call, not a socket or a background service. It writes to reamer_py's data directory (wherever reamer_py.set_data_dir(path) last pointed it, ./data if you never called it), replacing anything already stored under that ticker name, not merging with it. The call returns once the write is finished — nothing to poll or wait on.

import reamer_py

bars = []
bar = reamer_py.OhlcvBar()
bar.timestamp = "2024-01-15 09:30:00"
bar.open, bar.high, bar.low, bar.close = 100.0, 101.5, 99.25, 101.0
bar.volume = 12_400.0
bars.append(bar)

# ... repeat for every bar you have, then write them all at once:
reamer_py.write_bin(bars, "AAPL")

From here, "AAPL" is available to any later reamer_py call in this process — most directly, run_backtest(data=["AAPL"], ...) reads back exactly the bars you just wrote, in order:

result = reamer_py.run_backtest(data=["AAPL"], strategy=MyStrategy())

Passing bars directly: run_backtest(bars=...)

If you'd rather skip the file entirely, run_backtest takes a bars argument instead of data — a dictionary mapping each ticker name to its own list of OhlcvBars, built the same way as above, just handed straight to the backtest instead of written to disk first.

Here's a complete example — pulling rows out of an existing SQL database and running a backtest against them, with no intermediate file at any point. Swap the database part for whatever you actually have (a different database, a REST call, an internal service — the rest of the code doesn't change):

import sqlite3
import reamer_py

# Your existing data source. Any database works the same way -- this example
# uses SQLite, but a Postgres/MySQL/kdb+ client's query result looks the same
# once it's rows in a loop.
conn = sqlite3.connect("my_market_data.db")
rows = conn.execute(
    "SELECT timestamp, open, high, low, close, volume "
    "FROM ohlcv WHERE ticker = 'AAPL' ORDER BY timestamp ASC"
).fetchall()

# Turn each row into an OhlcvBar -- this loop is the only part that's
# specific to your data source's shape.
bars = []
for timestamp, open_, high, low, close, volume in rows:
    bar = reamer_py.OhlcvBar()
    bar.timestamp = timestamp   # must already be "YYYY-MM-DD HH:MM:SS", UTC
    bar.open = open_
    bar.high = high
    bar.low = low
    bar.close = close
    bar.volume = volume
    bars.append(bar)

# Hand the list straight to run_backtest. No .bin file is written anywhere.
result = reamer_py.run_backtest(
    bars={"AAPL": bars},
    strategy=MyStrategy(),
)

That's the whole pattern: however your data actually gets into Python — a database cursor, a paginated API response, a live feed — the only two things that matter are (1) build one OhlcvBar per row, in ascending timestamp order, and (2) put each ticker's list under its name in the bars dict. Everything after that is identical to a normal reamer_py backtest.

The shape

Every loader follows the same five pieces, whether it's one of Reamer Labs' own reference examples or one you write from scratch:

1. Lazy import — import the vendor's SDK inside a function, not at module load time, so nothing about reamer_py itself requires it. Raise ImportError with the exact pip install command if it's missing.
2. Credential resolution — accept credentials as keyword arguments, fall back to an environment variable, raise ValueError with a specific message (which env var, what's missing) if neither is present.
3. Record-to-bar mapping — the one function that knows the vendor's actual response shape. Every field translation (unit scaling, timestamp parsing) happens exactly once, here.
4. Single-ticker load function — fetch, map every record, raise RuntimeError if the result is empty (never silently ingest zero bars), then reamer_py.write_bin(bars, ticker) (or return bars for the in-memory path above).
5. Portfolio variant — same shape, looping over a ticker -> vendor_symbol mapping, one write_bin call per ticker.

Five complete, working examples follow — three vendor APIs, one local file format, and one aggregation pattern. Copy whichever is closest to your own source and adapt the record-to-bar mapping — that's the entire process for connecting a sixth source, a seventh, or an in-house feed nobody outside your firm has ever heard of.

Databento

Vendor SDK, nanosecond timestamps, fixed-point prices.

pip install databento
export DATABENTO_API_KEY=db-...

# databento_loader.py
import os
import sys
from datetime import datetime, timezone

import reamer_py as _r

# Reamer only consumes OHLCV bars — restricting to these schemas up front means a
# schema typo (e.g. "trades", "mbo") fails for free instead of running up a bill on
# data Reamer can't use anyway. These four are Databento's actual full set of OHLCV
# schemas (confirmed via client.metadata.list_schemas() against XNAS.ITCH, GLBX.MDP3,
# and OPRA.PILLAR) — there is no "ohlcv-eod" schema on any dataset; it doesn't exist.
_ALLOWED_SCHEMAS = {"ohlcv-1s", "ohlcv-1m", "ohlcv-1h", "ohlcv-1d"}

_COST_WARNING_THRESHOLD_USD = 1.0

# Databento OHLCV price fields are fixed-point int64, scaled by 1e9.
_FIXED_PRICE_SCALE = 1e9


def _import_databento():
    try:
        import databento
    except ImportError:
        raise ImportError("pip install databento to use this feature") from None
    return databento


def _resolve_api_key(api_key):
    key = api_key or os.environ.get("DATABENTO_API_KEY")
    if not key:
        raise ValueError(
            "No Databento API key provided — pass api_key= or set the "
            "DATABENTO_API_KEY environment variable"
        )
    return key


def _check_schema(schema):
    if schema not in _ALLOWED_SCHEMAS:
        raise ValueError(
            f"schema must be one of {sorted(_ALLOWED_SCHEMAS)}, got {schema!r} — "
            "Reamer only supports OHLCV bar data"
        )


def _warn_if_expensive(client, dataset, symbols, schema, start, end, stype_in):
    try:
        cost = client.metadata.get_cost(
            dataset=dataset, symbols=symbols, schema=schema, start=start, end=end,
            stype_in=stype_in,
        )
    except Exception:
        return
    if cost is not None and cost > _COST_WARNING_THRESHOLD_USD:
        print(
            f"warning: estimated Databento cost for this request is "
            f"${cost:.2f} (threshold ${_COST_WARNING_THRESHOLD_USD:.2f})",
            file=sys.stderr,
        )


def _record_to_bar(record):
    # Integer floor-division, not /1e9 — ts_event is nanosecond-epoch (~19 significant
    # digits), well past float64's ~15-17 digit exact-integer range; dividing as a float
    # loses precision and can round to the wrong second.
    ts = datetime.fromtimestamp(record.ts_event // 1_000_000_000, tz=timezone.utc)
    bar = _r.OhlcvBar()
    bar.timestamp = ts.strftime("%Y-%m-%d %H:%M:%S")
    bar.open = record.open / _FIXED_PRICE_SCALE
    bar.high = record.high / _FIXED_PRICE_SCALE
    bar.low = record.low / _FIXED_PRICE_SCALE
    bar.close = record.close / _FIXED_PRICE_SCALE
    bar.volume = float(record.volume)
    return bar


def load_databento(dataset, symbols, schema, start, end, ticker, api_key=None,
                    stype_in="raw_symbol"):
    """Fetch historical OHLCV bars from Databento straight into the data store —
    no CSV involved. Data is UTC-native, sidestepping the timezone/DST problem
    CSV loading has.

    dataset : Databento dataset code, e.g. "XNAS.ITCH", "GLBX.MDP3"
    symbols : list of Databento symbols, e.g. ["AAPL"] (raw_symbol) or ["ES.c.0"]
              (continuous front-month futures — requires stype_in="continuous")
    schema  : one of "ohlcv-1s", "ohlcv-1m", "ohlcv-1h", "ohlcv-1d"
    start, end : range bounds, passed through to databento.Historical.timeseries.get_range
    ticker  : Reamer-side ticker name to ingest the fetched bars under (query it
              back via run_backtest(data=[ticker]))
    api_key : Databento API key; falls back to the DATABENTO_API_KEY env var
    stype_in : Databento symbology type for `symbols` — "raw_symbol" (default,
               plain tickers) or "continuous" (".c.0"-style rolling futures
               contracts). Passing continuous-contract symbols with the default
               "raw_symbol" fails with a 422 from Databento's own API — this
               isn't optional plumbing, it's required to match the symbol format.
    """
    _check_schema(schema)
    databento = _import_databento()
    key = _resolve_api_key(api_key)

    client = databento.Historical(key)
    _warn_if_expensive(client, dataset, symbols, schema, start, end, stype_in)

    data = client.timeseries.get_range(
        dataset=dataset, symbols=symbols, schema=schema, start=start, end=end,
        stype_in=stype_in,
    )

    bars = [_record_to_bar(r) for r in data]
    if not bars:
        raise RuntimeError(
            f"No data returned for {symbols} in dataset {dataset!r}, schema "
            f"{schema!r}, between {start} and {end}"
        )
    _r.write_bin(bars, ticker)


def load_databento_portfolio(dataset, symbols, schema, start, end, api_key=None,
                              stype_in="raw_symbol"):
    """Same as load_databento, but for multiple tickers at once.

    symbols : dict mapping ticker (Reamer-side name) -> Databento symbol
    stype_in : see load_databento — applies uniformly to every symbol in the dict;
               mixing raw_symbol and continuous tickers in one call isn't supported
    """
    _check_schema(schema)
    databento = _import_databento()
    key = _resolve_api_key(api_key)

    client = databento.Historical(key)

    for ticker, symbol in symbols.items():
        _warn_if_expensive(client, dataset, [symbol], schema, start, end, stype_in)

        data = client.timeseries.get_range(
            dataset=dataset, symbols=[symbol], schema=schema, start=start, end=end,
            stype_in=stype_in,
        )
        bars = [_record_to_bar(r) for r in data]
        if not bars:
            raise RuntimeError(
                f"No data returned for {ticker!r} ({symbol!r}) in dataset "
                f"{dataset!r}, schema {schema!r}, between {start} and {end}"
            )

        _r.write_bin(bars, ticker)

Using it:

from databento_loader import load_databento

load_databento(
    dataset="XNAS.ITCH", symbols=["AAPL"], schema="ohlcv-1m",
    start="2024-01-01", end="2024-02-01", ticker="AAPL",
)

# Continuous front-month futures need stype_in="continuous" — the default,
# "raw_symbol", fails with a 422 against a ".c.0"-style symbol.
load_databento(
    dataset="GLBX.MDP3", symbols=["ES.c.0"], schema="ohlcv-1d",
    start="2024-01-01", end="2024-06-01", ticker="ES", stype_in="continuous",
)

Databento serves fixed UTC epoch nanoseconds by default — one real, concrete benefit of this specific vendor, not a claim about the loader pattern in general: no DST bug is possible through this path, unlike a CSV export in local exchange time. api_key can also be passed explicitly as a keyword argument; it otherwise falls back to DATABENTO_API_KEY. Query the ingested tickers back via run_backtest(data=[ticker, ...]).

Alpaca

REST API, paginated.

pip install requests
export ALPACA_API_KEY_ID=...
export ALPACA_API_SECRET_KEY=...

# alpaca_loader.py
import os
from datetime import datetime, timezone

import reamer_py as _r

_BARS_URL = "https://data.alpaca.markets/v2/stocks/bars"


def _import_requests():
    try:
        import requests
    except ImportError:
        raise ImportError("pip install requests to use this feature") from None
    return requests


def _resolve_keys(api_key_id, api_secret_key):
    key_id = api_key_id or os.environ.get("ALPACA_API_KEY_ID")
    secret = api_secret_key or os.environ.get("ALPACA_API_SECRET_KEY")
    if not key_id or not secret:
        raise ValueError(
            "No Alpaca API credentials provided — pass api_key_id=/api_secret_key= "
            "or set ALPACA_API_KEY_ID and ALPACA_API_SECRET_KEY env vars"
        )
    return key_id, secret


def _parse_ts(ts_str):
    # RFC-3339, e.g. "2022-01-03T09:00:00Z" or "...T09:00:00.123456789Z" — Reamer's
    # bar format is per-second resolution, so truncate any fractional part rather
    # than parse it exactly.
    core = ts_str.replace("Z", "").split(".")[0]
    dt = datetime.strptime(core, "%Y-%m-%dT%H:%M:%S").replace(tzinfo=timezone.utc)
    return dt.strftime("%Y-%m-%d %H:%M:%S")


def _fetch_bars(symbols, timeframe, start, end, key_id, secret, feed, limit):
    requests = _import_requests()
    headers = {"APCA-API-KEY-ID": key_id, "APCA-API-SECRET-KEY": secret}
    params = {"symbols": ",".join(symbols), "timeframe": timeframe,
              "start": start, "end": end, "limit": limit}
    if feed:
        params["feed"] = feed

    by_symbol = {s: [] for s in symbols}
    page_token = None
    while True:
        if page_token:
            params["page_token"] = page_token
        resp = requests.get(_BARS_URL, headers=headers, params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        for sym, bars in (data.get("bars") or {}).items():
            by_symbol.setdefault(sym, []).extend(bars)
        page_token = data.get("next_page_token")
        if not page_token:
            break
    return by_symbol


def _record_to_bar(record):
    bar = _r.OhlcvBar()
    bar.timestamp = _parse_ts(record["t"])
    bar.open = float(record["o"])
    bar.high = float(record["h"])
    bar.low = float(record["l"])
    bar.close = float(record["c"])
    bar.volume = float(record["v"])
    return bar


def load_alpaca(symbols, timeframe, start, end, ticker, api_key_id=None,
                 api_secret_key=None, feed=None, limit=10000):
    """Fetch historical OHLCV bars from Alpaca's Market Data API straight into
    the data store — no CSV involved.

    symbols  : list of Alpaca stock symbols, e.g. ["AAPL"]
    timeframe: Alpaca aggregation string, e.g. "1Min", "1Hour", "1Day"
    start, end: RFC-3339 or YYYY-MM-DD, passed through to the bars endpoint
    ticker   : Reamer-side ticker name to ingest the fetched bars under (single
               symbol only — use load_alpaca_portfolio for more than one)
    api_key_id, api_secret_key: falls back to ALPACA_API_KEY_ID / ALPACA_API_SECRET_KEY
    feed     : optional Alpaca data feed override ("sip", "iex", "boats", "otc")
    """
    if len(symbols) != 1:
        raise ValueError("load_alpaca takes exactly one symbol — use load_alpaca_portfolio for more")
    key_id, secret = _resolve_keys(api_key_id, api_secret_key)
    by_symbol = _fetch_bars(symbols, timeframe, start, end, key_id, secret, feed, limit)

    bars = [_record_to_bar(r) for r in by_symbol.get(symbols[0], [])]
    if not bars:
        raise RuntimeError(f"No data returned for {symbols[0]!r} between {start} and {end}")
    _r.write_bin(bars, ticker)


def load_alpaca_portfolio(symbols, timeframe, start, end, api_key_id=None,
                           api_secret_key=None, feed=None, limit=10000):
    """Same as load_alpaca, but for multiple tickers at once.

    symbols : list of Alpaca stock symbols — also used as the Reamer-side ticker name
    """
    key_id, secret = _resolve_keys(api_key_id, api_secret_key)
    by_symbol = _fetch_bars(symbols, timeframe, start, end, key_id, secret, feed, limit)

    for sym in symbols:
        bars = [_record_to_bar(r) for r in by_symbol.get(sym, [])]
        if not bars:
            raise RuntimeError(f"No data returned for {sym!r} between {start} and {end}")
        _r.write_bin(bars, sym)

Using it:

from alpaca_loader import load_alpaca, load_alpaca_portfolio

load_alpaca(["AAPL"], timeframe="1Min", start="2024-01-01", end="2024-02-01", ticker="AAPL")

# Portfolio — Alpaca symbols are also used as the Reamer-side ticker names
load_alpaca_portfolio(["AAPL", "MSFT"], timeframe="1Day", start="2024-01-01", end="2024-06-01")

The pagination loop above (accumulate into a list, follow next_page_token until it's empty) is the shape any paginated vendor API needs — worth checking for before assuming one request covers a whole range.

Massive.com

Simple REST, Bearer auth, no vendor SDK required. Massive.com is Polygon.io's post-October-2025 rebrand — same aggregates endpoint shape, different domain and a Bearer token instead of an apiKey query parameter.

pip install requests
export MASSIVE_API_KEY=...

# massive_loader.py
from datetime import datetime, timezone
import os

import reamer_py as _r

_BASE_URL = "https://api.massive.com"


def _import_requests():
    try:
        import requests
    except ImportError:
        raise ImportError("pip install requests to use this feature") from None
    return requests


def _resolve_api_key(api_key):
    key = api_key or os.environ.get("MASSIVE_API_KEY")
    if not key:
        raise ValueError(
            "No Massive API key provided — pass api_key= or set the "
            "MASSIVE_API_KEY environment variable"
        )
    return key


def _fetch_bars(ticker, multiplier, timespan, start, end, api_key, adjusted, sort, limit):
    requests = _import_requests()
    headers = {"Authorization": f"Bearer {api_key}"}
    url = f"{_BASE_URL}/v2/aggs/ticker/{ticker}/range/{multiplier}/{timespan}/{start}/{end}"
    params = {"adjusted": str(adjusted).lower(), "sort": sort, "limit": limit}

    results = []
    while url:
        resp = requests.get(url, headers=headers, params=params, timeout=30)
        resp.raise_for_status()
        data = resp.json()
        results.extend(data.get("results") or [])
        url = data.get("next_url")
        params = None  # next_url already carries its own query string
    return results


def _record_to_bar(record):
    # Integer floor-division, not /1000 as a float — "t" is millisecond-epoch;
    # dividing as an int keeps this exact rather than risking float rounding.
    ts = datetime.fromtimestamp(record["t"] // 1000, tz=timezone.utc)
    bar = _r.OhlcvBar()
    bar.timestamp = ts.strftime("%Y-%m-%d %H:%M:%S")
    bar.open = float(record["o"])
    bar.high = float(record["h"])
    bar.low = float(record["l"])
    bar.close = float(record["c"])
    bar.volume = float(record["v"])
    return bar


def load_massive(ticker, multiplier, timespan, start, end, api_key=None,
                  adjusted=True, sort="asc", limit=50000):
    """Fetch historical OHLCV bars from Massive.com's aggregates API straight into
    the data store — no CSV involved. (Massive.com is Polygon.io's post-Oct-2025
    rebrand; the REST API surface — /v2/aggs/ticker/.../range/... — is unchanged.)

    ticker    : Massive stock ticker, e.g. "AAPL" — also the Reamer-side ticker
                name the fetched bars are ingested under
    multiplier: size of the timespan window, e.g. 1
    timespan  : "minute", "hour", "day", etc.
    start, end: YYYY-MM-DD or millisecond epoch, passed through to the aggs endpoint
    api_key   : falls back to the MASSIVE_API_KEY env var
    """
    key = _resolve_api_key(api_key)
    records = _fetch_bars(ticker, multiplier, timespan, start, end, key, adjusted, sort, limit)

    bars = [_record_to_bar(r) for r in records]
    if not bars:
        raise RuntimeError(f"No data returned for {ticker!r} between {start} and {end}")
    _r.write_bin(bars, ticker)


def load_massive_portfolio(tickers, multiplier, timespan, start, end,
                            api_key=None, adjusted=True, sort="asc", limit=50000):
    """Same as load_massive, but for multiple tickers at once.

    tickers : list of Massive stock tickers — also used as the Reamer-side ticker name
    """
    key = _resolve_api_key(api_key)

    for ticker in tickers:
        records = _fetch_bars(ticker, multiplier, timespan, start, end, key, adjusted, sort, limit)
        bars = [_record_to_bar(r) for r in records]
        if not bars:
            raise RuntimeError(f"No data returned for {ticker!r} between {start} and {end}")
        _r.write_bin(bars, ticker)

Using it:

import reamer_py
from massive_loader import load_massive, load_massive_portfolio

load_massive("AAPL", 1, "minute", "2024-01-01", "2024-02-01")

result = reamer_py.run_backtest(data=["AAPL"], strategy=MyStrategy())

Massive/Polygon's aggregates endpoint paginates via a next_url returned in the response body, not an offset/page_token parameter — the loop above follows it directly (and drops the original query params after the first request, since next_url already carries its own).

Parquet

The odd one out among these four: no vendor to authenticate against, no network call, and the data is already typed — no delimiter/header guessing the way load_csv needs. The only dependency is a reader library (pyarrow), which most firms already have installed if they're storing research data as Parquet in the first place.

pip install pyarrow

# parquet_loader.py
from datetime import timezone

import reamer_py as _r

_DEFAULT_COLUMNS = {
    "timestamp": "timestamp", "open": "open", "high": "high",
    "low": "low", "close": "close", "volume": "volume",
}


def _import_pyarrow():
    try:
        import pyarrow.parquet as pq
    except ImportError:
        raise ImportError("pip install pyarrow to use this feature") from None
    return pq


def _record_to_bar(record, columns):
    ts = record[columns["timestamp"]]
    if getattr(ts, "tzinfo", None) is not None:
        ts = ts.astimezone(timezone.utc)
    bar = _r.OhlcvBar()
    bar.timestamp = ts.strftime("%Y-%m-%d %H:%M:%S")
    bar.open = float(record[columns["open"]])
    bar.high = float(record[columns["high"]])
    bar.low = float(record[columns["low"]])
    bar.close = float(record[columns["close"]])
    bar.volume = float(record[columns["volume"]])
    return bar


def load_parquet(path, ticker, columns=None):
    """Read OHLCV bars from a local Parquet file straight into the data store —
    no vendor SDK, no credentials, no network call.

    path    : Parquet file path
    ticker  : Reamer-side ticker name to ingest the read bars under (query it
              back via run_backtest(data=[ticker]))
    columns : optional dict overriding the default column-name mapping
              ({"timestamp": ..., "open": ..., "high": ..., "low": ...,
              "close": ..., "volume": ...}) for files whose column names
              don't match the defaults
    """
    pq = _import_pyarrow()
    columns = columns or _DEFAULT_COLUMNS
    parquet_file = pq.ParquetFile(path)

    bars = []
    for batch in parquet_file.iter_batches(columns=list(columns.values())):
        for record in batch.to_pylist():
            bars.append(_record_to_bar(record, columns))

    if not bars:
        raise RuntimeError(f"No rows read from {path!r}")
    _r.write_bin(bars, ticker)


def load_parquet_portfolio(paths, columns=None):
    """Same as load_parquet, but for multiple tickers at once.

    paths : dict mapping ticker (Reamer-side name) -> Parquet file path, for
            the common one-file-per-ticker layout
    """
    for ticker, path in paths.items():
        load_parquet(path, ticker, columns)


def load_parquet_dataset(path, ticker_column="symbol", columns=None):
    """Read a single file or a partitioned directory holding multiple tickers,
    distinguished by a symbol/ticker column — the shape most data-lake
    Parquet exports actually ship in, as opposed to one file per ticker.

    path          : a Parquet file, or a directory of Parquet files
    ticker_column : the column identifying which ticker each row belongs to
    columns       : see load_parquet
    """
    try:
        import pyarrow.dataset as ds
    except ImportError:
        raise ImportError("pip install pyarrow to use this feature") from None
    columns = columns or _DEFAULT_COLUMNS
    dataset = ds.dataset(path)
    read_columns = list(columns.values()) + [ticker_column]

    grouped = {}
    for batch in dataset.to_batches(columns=read_columns):
        for record in batch.to_pylist():
            grouped.setdefault(record[ticker_column], []).append(_record_to_bar(record, columns))

    if not grouped:
        raise RuntimeError(f"No rows read from {path!r}")
    for grouped_ticker, bars in grouped.items():
        _r.write_bin(bars, grouped_ticker)

Using it:

from parquet_loader import load_parquet, load_parquet_dataset

# One file per ticker
load_parquet("AAPL_1m.parquet", ticker="AAPL")

# Column names that don't match the default? pass your own mapping
load_parquet("aapl.parquet", ticker="AAPL",
             columns={"timestamp": "ts", "open": "o", "high": "h",
                      "low": "l", "close": "c", "volume": "v"})

# One partitioned dataset covering many tickers, split by a symbol column
load_parquet_dataset("s3://research-data/ohlcv/", ticker_column="symbol")

iter_batches/to_batches read the file (or dataset) one row-group at a time rather than materializing the whole Arrow table up front, which keeps peak memory during the read bounded even for large files. The bars list handed to write_bin is still fully built in memory before that one call, though — write_bin itself has no chunked/streaming form yet, so this doesn't make arbitrarily large single-ticker Parquet files free; it only avoids doubling up on the Arrow-side decode buffer while building that list.

Activity/imbalance bars

Every example above maps one vendor record to one bar directly. This one is structurally different: it aggregates many raw trades into fewer, coarser bars — the shape any tick bar, volume bar, dollar bar, or imbalance bar actually takes. Reamer Py doesn't build activity bars itself (this loader is exactly why: the aggregation logic lives in your own code, not the engine) — it only needs the result in OhlcvBar shape, with tick_count/notional set to the real counts your aggregation produced, not left at their defaults.

That last part matters mechanically, not just for bookkeeping: leave tick_count unset (-1, the default) and reamer_py falls back to a gap-derived estimate — roughly one synthetic tick per second until the next bar. That's a reasonable stand-in for a genuine time bar, where a one-minute bar really did take about sixty seconds to form. It's meaningless for a bar that was actually sealed after 500 trades or $1M of notional in nine seconds, or in nine minutes — the real tick/trade count is data your aggregation already has; setting it costs nothing and makes every synthetic tick generated inside that bar reflect how it actually formed, not a guess based on wall-clock time that has no relationship to the bar's real formation.

Two variants shown — tick bars (seal every N trades) and dollar bars (seal every $N of notional traded) — sharing one accumulator loop with a different seal condition. Volume bars are the same pattern again, just sealing on acc["volume"] instead of acc["notional"].

# activity_bar_loader.py
import csv
from datetime import datetime, timezone

import reamer_py


def _read_trades(path):
    """path: CSV with a header row (timestamp, price, size) -- the
    vendor-agnostic shape any tick-level source can be normalized to.
    Timestamp is any ISO-8601-ish string; treated as literal UTC, same as
    every other loader on this page."""
    with open(path, newline="") as f:
        for row in csv.DictReader(f):
            ts = datetime.fromisoformat(row["timestamp"]).replace(tzinfo=timezone.utc)
            yield ts, float(row["price"]), float(row["size"])


def _new_accumulator(ts, price):
    return {"open": price, "high": price, "low": price, "close": price,
            "volume": 0.0, "notional": 0.0, "tick_count": 0, "last_ts": ts}


def _seal_bar(acc):
    # A sealed bar's timestamp is necessarily second-resolution (reamer_py's
    # native bar precision) even though the raw trades feeding it are
    # typically sub-second -- this is the last trade's timestamp truncated
    # to the second, not an oversight.
    bar = reamer_py.OhlcvBar()
    bar.timestamp = acc["last_ts"].strftime("%Y-%m-%d %H:%M:%S")
    bar.open, bar.high, bar.low, bar.close = acc["open"], acc["high"], acc["low"], acc["close"]
    bar.volume = acc["volume"]
    bar.notional = acc["notional"]      # real dollar volume, not a synthetic estimate
    bar.tick_count = acc["tick_count"]  # real trade count, not a gap-derived guess
    return bar


def _aggregate(path, seal_when):
    """seal_when(acc) -> bool decides when the current bar closes -- this
    one loop is every activity bar type, just with a different predicate."""
    bars, acc = [], None
    for ts, price, size in _read_trades(path):
        if acc is None:
            acc = _new_accumulator(ts, price)
        acc["high"]  = max(acc["high"], price)
        acc["low"]   = min(acc["low"], price)
        acc["close"] = price
        acc["volume"]     += size
        acc["notional"]   += price * size
        acc["tick_count"] += 1
        acc["last_ts"] = ts
        if seal_when(acc):
            bars.append(_seal_bar(acc))
            acc = None
    if acc is not None and acc["tick_count"] > 0:
        bars.append(_seal_bar(acc))  # final partial bar
    return bars


def load_tick_bars(path, ticker, ticks_per_bar):
    """Seal a new bar every ticks_per_bar trades -- tick_count on every
    bar is always exactly ticks_per_bar (or fewer, for the final bar)."""
    bars = _aggregate(path, lambda acc: acc["tick_count"] >= ticks_per_bar)
    if not bars:
        raise RuntimeError(f"No trades read from {path!r}")
    reamer_py.write_bin(bars, ticker)


def load_dollar_bars(path, ticker, threshold_notional):
    """Seal a new bar every time cumulative notional (price * size, summed
    since the last seal) crosses threshold_notional -- tick_count on each
    bar is however many trades it actually took to get there."""
    bars = _aggregate(path, lambda acc: acc["notional"] >= threshold_notional)
    if not bars:
        raise RuntimeError(f"No trades read from {path!r}")
    reamer_py.write_bin(bars, ticker)

Using it:

from activity_bar_loader import load_tick_bars, load_dollar_bars

# A bar every 500 trades -- tick_count == 500 on every bar but the last
load_tick_bars("trades.csv", ticker="AAPL_TICK500", ticks_per_bar=500)

# A bar every $1M of notional traded -- tick_count varies bar to bar
load_dollar_bars("trades.csv", ticker="AAPL_DOLLAR1M", threshold_notional=1_000_000.0)

No vendor SDK, no credentials — the only real dependency is having a raw trade stream to aggregate in the first place, which is a different problem from writing the loader itself (a vendor's tick-level API, a local tick file, or a redistributed feed all normalize to the same (timestamp, price, size) shape _read_trades expects).

Testing your loader

Mock the vendor SDK, not the network — sys.modules["your_vendor"] gets replaced with a fake exposing whatever surface your loader calls, so nothing here needs real credentials or a real request:

import sys
import types
import reamer_py


def test_field_mapping(monkeypatch, tmp_path):
    fake_response = types.SimpleNamespace(
        json=lambda: {"results": [{"t": 1704067200000, "o": 100.0, "h": 105.0,
                                    "l": 98.0, "c": 103.0, "v": 1_500_000}]},
        raise_for_status=lambda: None,
    )
    fake_requests = types.SimpleNamespace(get=lambda *a, **k: fake_response)
    monkeypatch.setitem(sys.modules, "requests", fake_requests)
    monkeypatch.setenv("MASSIVE_API_KEY", "test-key")
    reamer_py.set_data_dir(str(tmp_path))

    from massive_loader import load_massive
    load_massive("AAPL", 1, "minute", "2024-01-01", "2024-01-02")

    assert (tmp_path / "AAPL.bin").exists()

To check the mapped values themselves rather than just that a file appeared, run a one-bar probe strategy through run_backtest instead of parsing the on-disk bar format by hand:

class _Probe:
    lookback = 1
    seen = []

    def __init__(self, config_path=None):
        pass

    def on_bar(self, data):
        tv = data["AAPL"]
        if tv.valid:
            _Probe.seen.append(float(tv.close[-1]))
        return None

reamer_py.run_backtest(data=["AAPL"], strategy=_Probe())
assert _Probe.seen == [103.0]

Also worth a dedicated test, once real credentials are available: a fixed package-shape check — construct one real vendor record object, or a minimal real API response, locally, no live call, and confirm the record-to-bar mapping still holds. This is what catches a vendor silently renaming or rescaling a field out from under a loader that mocked too loosely to notice.

Common mistakes

Losing precision on nanosecond/microsecond timestamps via float division: ts_event / 1e9 loses precision at nanosecond-epoch magnitudes (~19 significant digits, past float64's exact-integer range) and can round into the wrong second. Use integer floor division (ts_event // 1_000_000_000) when the vendor's native resolution is finer than seconds.

Ingesting zero bars silently: an empty result from the vendor should raise, not call write_bin([]). A loader that silently no-ops on an empty range turns a typo'd symbol or date range into a confusing "ticker not found" three steps later in run_backtest, instead of a clear error at the loader call site.

Assuming the vendor is already UTC: if the vendor's timestamps carry a timezone or are exchange-local, convert to UTC in the record-to-bar mapping before formatting — reamer_py treats every parsed timestamp as literal UTC wall-clock, unconditionally, the same as CSV loading.

Calling write_bin once per bar: each call replaces the ticker's entire series. Accumulate a list, then call it once.

Reamer Server accepts order intents from strategy processes over a documented wire protocol, not a required SDK — the same principle as a data source loader, applied to submitting orders instead of ingesting bars. Any process that can speak the protocol is a strategy connector, in any language, including one you write yourself rather than a client library Reamer Labs ships.

Full protocol documentation lands once Reamer Server ships. See Reamer Server for where it stands today.

Reamer Server's OMS and sequencer depend on a defined connector interface, not on any one broker's wire format — FIX is the reference implementation, not the only permitted path. A firm whose broker speaks REST, SBE, or its own gateway protocol builds a connector against that same interface rather than waiting on Reamer Labs to add native support for it.

Full interface documentation lands once Reamer Server ships. See Reamer Server for where it stands today.