Skip to content

Data API

The lumina_lob.data package downloads and calibrates against real tick data.

lumina_lob.data.polygon

Polygon.io downloader for end-of-day bars and tick/trade data.

Classes

PolygonClient

Lightweight client for Polygon.io market data.

Parameters

api_key: Polygon API key. If omitted, read from the POLYGON_API_KEY environment variable. base_url: Polygon API base URL. cache_dir: Directory to cache raw JSON responses.

Source code in lumina_lob/data/polygon.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
class PolygonClient:
    """Lightweight client for Polygon.io market data.

    Parameters
    ----------
    api_key:
        Polygon API key. If omitted, read from the ``POLYGON_API_KEY``
        environment variable.
    base_url:
        Polygon API base URL.
    cache_dir:
        Directory to cache raw JSON responses.
    """

    def __init__(
        self,
        api_key: str | None = None,
        base_url: str = "https://api.polygon.io",
        cache_dir: str = ".cache/polygon",
    ) -> None:
        self.api_key = api_key if api_key is not None else os.environ.get("POLYGON_API_KEY")
        if not self.api_key:
            raise ValueError("Polygon API key is required (pass api_key or set POLYGON_API_KEY)")
        self.base_url = base_url.rstrip("/")
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)

    def get_daily_bars(
        self, ticker: str, start_date: str, end_date: str
    ) -> pd.DataFrame:
        """Fetch daily OHLCV bars for ``ticker`` between ``start_date`` and ``end_date``.

        Dates must be in ``YYYY-MM-DD`` format.  Results are cached by ticker
        and date range.

        Returns
        -------
        DataFrame with columns:
        ``timestamp``, ``open``, ``high``, ``low``, ``close``, ``volume``,
        ``vwap``, ``transactions``.
        """
        endpoint = f"{self.base_url}/v2/aggs/ticker/{ticker}/range/1/day/{start_date}/{end_date}"
        params = {"apiKey": self.api_key}
        cache_key = f"{ticker.upper()}_{start_date}_{end_date}_bars.json"
        data = self._fetch_json(endpoint, params, cache_key)
        return self._bars_to_dataframe(data.get("results", []))

    def get_trades(self, ticker: str, date: str) -> pd.DataFrame:
        """Fetch trades (ticks) for ``ticker`` on ``date``.

        Returns
        -------
        DataFrame with columns:
        ``timestamp``, ``price``, ``size``, ``exchange``, ``conditions``.
        """
        endpoint = f"{self.base_url}/v3/trades/{ticker}"
        params = {"timestamp": date, "apiKey": self.api_key}
        cache_key = f"{ticker.upper()}_{date}_trades.json"
        data = self._fetch_json(endpoint, params, cache_key)
        return self._trades_to_dataframe(data.get("results", []))

    def _fetch_json(
        self, url: str, params: dict[str, Any], cache_key: str
    ) -> dict[str, Any]:
        """Return JSON from cache or fetch from Polygon and cache it."""
        cache_path = self.cache_dir / cache_key.replace("/", "_")
        if cache_path.exists():
            with cache_path.open("r", encoding="utf-8") as fh:
                return cast(dict[str, Any], json.load(fh))

        response = requests.get(url, params=params, timeout=30)
        if response.status_code != 200:
            raise RuntimeError(
                f"Polygon API error {response.status_code}: {response.text}"
            )
        data: dict[str, Any] = cast(dict[str, Any], response.json())
        cache_path.write_text(response.text, encoding="utf-8")
        return data

    @staticmethod
    def _bars_to_dataframe(results: list[dict[str, Any]]) -> pd.DataFrame:
        if not results:
            return pd.DataFrame(
                columns=[
                    "timestamp",
                    "open",
                    "high",
                    "low",
                    "close",
                    "volume",
                    "vwap",
                    "transactions",
                ]
            )
        rows = []
        for r in results:
            rows.append(
                {
                    "timestamp": pd.to_datetime(r["t"], unit="ms", utc=True),
                    "open": float(r["o"]),
                    "high": float(r["h"]),
                    "low": float(r["l"]),
                    "close": float(r["c"]),
                    "volume": int(r["v"]),
                    "vwap": float(r.get("vw", 0.0)),
                    "transactions": int(r.get("n", 0)),
                }
            )
        df = pd.DataFrame(rows)
        return df.sort_values("timestamp").reset_index(drop=True)

    @staticmethod
    def _trades_to_dataframe(results: list[dict[str, Any]]) -> pd.DataFrame:
        if not results:
            return pd.DataFrame(
                columns=["timestamp", "price", "size", "exchange", "conditions"]
            )
        rows = []
        for r in results:
            rows.append(
                {
                    "timestamp": pd.to_datetime(r["sip_timestamp"], unit="ns", utc=True),
                    "price": float(r["price"]),
                    "size": int(r["size"]),
                    "exchange": int(r.get("exchange", 0)),
                    "conditions": r.get("conditions", []),
                }
            )
        df = pd.DataFrame(rows)
        return df.sort_values("timestamp").reset_index(drop=True)
Methods:
get_daily_bars(ticker, start_date, end_date)

Fetch daily OHLCV bars for ticker between start_date and end_date.

Dates must be in YYYY-MM-DD format. Results are cached by ticker and date range.

Returns

DataFrame with columns: timestamp, open, high, low, close, volume, vwap, transactions.

Source code in lumina_lob/data/polygon.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def get_daily_bars(
    self, ticker: str, start_date: str, end_date: str
) -> pd.DataFrame:
    """Fetch daily OHLCV bars for ``ticker`` between ``start_date`` and ``end_date``.

    Dates must be in ``YYYY-MM-DD`` format.  Results are cached by ticker
    and date range.

    Returns
    -------
    DataFrame with columns:
    ``timestamp``, ``open``, ``high``, ``low``, ``close``, ``volume``,
    ``vwap``, ``transactions``.
    """
    endpoint = f"{self.base_url}/v2/aggs/ticker/{ticker}/range/1/day/{start_date}/{end_date}"
    params = {"apiKey": self.api_key}
    cache_key = f"{ticker.upper()}_{start_date}_{end_date}_bars.json"
    data = self._fetch_json(endpoint, params, cache_key)
    return self._bars_to_dataframe(data.get("results", []))
get_trades(ticker, date)

Fetch trades (ticks) for ticker on date.

Returns

DataFrame with columns: timestamp, price, size, exchange, conditions.

Source code in lumina_lob/data/polygon.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
def get_trades(self, ticker: str, date: str) -> pd.DataFrame:
    """Fetch trades (ticks) for ``ticker`` on ``date``.

    Returns
    -------
    DataFrame with columns:
    ``timestamp``, ``price``, ``size``, ``exchange``, ``conditions``.
    """
    endpoint = f"{self.base_url}/v3/trades/{ticker}"
    params = {"timestamp": date, "apiKey": self.api_key}
    cache_key = f"{ticker.upper()}_{date}_trades.json"
    data = self._fetch_json(endpoint, params, cache_key)
    return self._trades_to_dataframe(data.get("results", []))

lumina_lob.data.databento

Databento downloader for historical trades and quotes.

Classes

DatabentoClient

Lightweight client for Databento historical market data.

Parameters

api_key: Databento API key. If omitted, read from the DATABENTO_API_KEY environment variable. cache_dir: Directory to cache raw DBN files.

Source code in lumina_lob/data/databento.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
class DatabentoClient:
    """Lightweight client for Databento historical market data.

    Parameters
    ----------
    api_key:
        Databento API key. If omitted, read from the ``DATABENTO_API_KEY``
        environment variable.
    cache_dir:
        Directory to cache raw DBN files.
    """

    def __init__(
        self,
        api_key: str | None = None,
        cache_dir: str = ".cache/databento",
    ) -> None:
        self.api_key = api_key if api_key is not None else os.environ.get("DATABENTO_API_KEY")
        if not self.api_key:
            raise ValueError("Databento API key is required (pass api_key or set DATABENTO_API_KEY)")
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self._client = db.Historical(key=self.api_key)

    def get_trades(
        self,
        symbol: str,
        start: str,
        end: str,
        dataset: str = "XNAS.ITCH",
    ) -> pd.DataFrame:
        """Fetch trade records for ``symbol`` between ``start`` and ``end``.

        Returns
        -------
        DataFrame with Databento ``trades`` schema columns.
        """
        return self._get_range(dataset, symbol, "trades", start, end)

    def get_quotes(
        self,
        symbol: str,
        start: str,
        end: str,
        dataset: str = "XNAS.ITCH",
        schema: str = "bbo",
    ) -> pd.DataFrame:
        """Fetch quote records for ``symbol`` between ``start`` and ``end``.

        Default schema is ``bbo`` (best bid/offer). Use ``mbo`` for market-by-order.

        Returns
        -------
        DataFrame with the requested quote schema columns.
        """
        if schema not in {"bbo", "mbo", "mbp-1", "mbp-10", "tbbo"}:
            raise ValueError(f"unsupported quote schema: {schema}")
        return self._get_range(dataset, symbol, schema, start, end)

    def _get_range(
        self,
        dataset: str,
        symbol: str,
        schema: str,
        start: str,
        end: str,
    ) -> pd.DataFrame:
        cache_key = f"{dataset}_{symbol}_{schema}_{start}_{end}.dbn"
        cache_path = self.cache_dir / cache_key.replace("/", "_").replace(":", "-")

        if cache_path.exists():
            store = db.DBNStore.from_file(str(cache_path))
            return store.to_df()

        store = self._client.timeseries.get_range(
            dataset=dataset,
            symbols=symbol,
            schema=schema,
            start=start,
            end=end,
            stype_in="raw_symbol",
            path=str(cache_path),
        )
        return store.to_df()
Methods:
get_quotes(symbol, start, end, dataset='XNAS.ITCH', schema='bbo')

Fetch quote records for symbol between start and end.

Default schema is bbo (best bid/offer). Use mbo for market-by-order.

Returns

DataFrame with the requested quote schema columns.

Source code in lumina_lob/data/databento.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def get_quotes(
    self,
    symbol: str,
    start: str,
    end: str,
    dataset: str = "XNAS.ITCH",
    schema: str = "bbo",
) -> pd.DataFrame:
    """Fetch quote records for ``symbol`` between ``start`` and ``end``.

    Default schema is ``bbo`` (best bid/offer). Use ``mbo`` for market-by-order.

    Returns
    -------
    DataFrame with the requested quote schema columns.
    """
    if schema not in {"bbo", "mbo", "mbp-1", "mbp-10", "tbbo"}:
        raise ValueError(f"unsupported quote schema: {schema}")
    return self._get_range(dataset, symbol, schema, start, end)
get_trades(symbol, start, end, dataset='XNAS.ITCH')

Fetch trade records for symbol between start and end.

Returns

DataFrame with Databento trades schema columns.

Source code in lumina_lob/data/databento.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def get_trades(
    self,
    symbol: str,
    start: str,
    end: str,
    dataset: str = "XNAS.ITCH",
) -> pd.DataFrame:
    """Fetch trade records for ``symbol`` between ``start`` and ``end``.

    Returns
    -------
    DataFrame with Databento ``trades`` schema columns.
    """
    return self._get_range(dataset, symbol, "trades", start, end)

lumina_lob.data.calibration

Calibration utilities for fitting agent parameters to real market data.

Classes

CalibratedParams dataclass

Parameters estimated from real market data for simulation agents.

Attributes

arrival_rate: Expected number of events per time_unit (e.g., per second). size_dist_method: Either "lognormal" or "empirical". size_lognorm_mu: Mean of log-sizes when size_dist_method == "lognormal". size_lognorm_sigma: Standard deviation of log-sizes when size_dist_method == "lognormal". size_hist: Normalized empirical size histogram when size_dist_method == "empirical". mean_spread: Average bid-ask spread in price units, if quote data was supplied. tick_size: Minimum price increment inferred from the spread / price grid. time_unit: Unit used for arrival_rate ("S" = seconds, "min" = minutes, etc.). permanent_impact: Permanent price-impact coefficient per unit signed volume, if price data was supplied. temporary_impact: Temporary price-impact coefficient per unit signed volume, if price data was supplied. impact_decay: Decay factor for the propagator-style temporary impact residual.

Source code in lumina_lob/data/calibration.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
@dataclass
class CalibratedParams:
    """Parameters estimated from real market data for simulation agents.

    Attributes
    ----------
    arrival_rate:
        Expected number of events per ``time_unit`` (e.g., per second).
    size_dist_method:
        Either ``"lognormal"`` or ``"empirical"``.
    size_lognorm_mu:
        Mean of log-sizes when ``size_dist_method == "lognormal"``.
    size_lognorm_sigma:
        Standard deviation of log-sizes when ``size_dist_method == "lognormal"``.
    size_hist:
        Normalized empirical size histogram when ``size_dist_method == "empirical"``.
    mean_spread:
        Average bid-ask spread in price units, if quote data was supplied.
    tick_size:
        Minimum price increment inferred from the spread / price grid.
    time_unit:
        Unit used for ``arrival_rate`` (``"S"`` = seconds, ``"min"`` = minutes, etc.).
    permanent_impact:
        Permanent price-impact coefficient per unit signed volume, if price data
        was supplied.
    temporary_impact:
        Temporary price-impact coefficient per unit signed volume, if price data
        was supplied.
    impact_decay:
        Decay factor for the propagator-style temporary impact residual.
    """

    arrival_rate: float
    size_dist_method: str
    size_lognorm_mu: float | None = None
    size_lognorm_sigma: float | None = None
    size_hist: pd.Series | None = None
    mean_spread: float | None = None
    tick_size: float = 0.01
    time_unit: str = "S"
    permanent_impact: float | None = None
    temporary_impact: float | None = None
    impact_decay: float | None = None

Functions:

calibrate(trades, quotes=None, size_method='lognormal', time_unit='S', bid_col='bid_px_00', ask_col='ask_px_00', price_col='price', side_col='side')

Estimate agent parameters from a trades DataFrame and optional quotes.

Parameters

trades: DataFrame with at least timestamp and size columns. If price_col and side_col are present, propagator-style impact coefficients are also estimated. quotes: Optional DataFrame with at least timestamp, bid_col and ask_col. size_method: "lognormal" fits log(size); "empirical" returns a normalized histogram. time_unit: Unit for arrival_rate. Pandas frequency string, e.g. "S", "min", "H". bid_col, ask_col: Column names for best bid and ask in quotes. price_col: Column name for trade prices in trades. side_col: Column name for trade side in trades.

Returns

CalibratedParams

Source code in lumina_lob/data/calibration.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
def calibrate(
    trades: pd.DataFrame,
    quotes: pd.DataFrame | None = None,
    size_method: str = "lognormal",
    time_unit: str = "S",
    bid_col: str = "bid_px_00",
    ask_col: str = "ask_px_00",
    price_col: str = "price",
    side_col: str = "side",
) -> CalibratedParams:
    """Estimate agent parameters from a trades DataFrame and optional quotes.

    Parameters
    ----------
    trades:
        DataFrame with at least ``timestamp`` and ``size`` columns. If ``price_col``
        and ``side_col`` are present, propagator-style impact coefficients are also
        estimated.
    quotes:
        Optional DataFrame with at least ``timestamp``, ``bid_col`` and ``ask_col``.
    size_method:
        ``"lognormal"`` fits ``log(size)``; ``"empirical"`` returns a normalized
        histogram.
    time_unit:
        Unit for ``arrival_rate``.  Pandas frequency string, e.g. ``"S"``,
        ``"min"``, ``"H"``.
    bid_col, ask_col:
        Column names for best bid and ask in ``quotes``.
    price_col:
        Column name for trade prices in ``trades``.
    side_col:
        Column name for trade side in ``trades``.

    Returns
    -------
    CalibratedParams
    """
    if "timestamp" not in trades.columns or "size" not in trades.columns:
        raise ValueError("trades DataFrame must contain 'timestamp' and 'size' columns")
    if trades.empty:
        raise ValueError("trades DataFrame is empty")

    arrival_rate = _estimate_arrival_rate(trades["timestamp"], time_unit)
    size_result = _fit_size_distribution(trades["size"], size_method)
    spread = None
    tick_size = 0.01
    impact_result = {}
    if quotes is not None:
        _validate_quote_columns(quotes, bid_col, ask_col)
        tick_size = _estimate_tick_size(quotes, bid_col, ask_col)
        spread = _estimate_mean_spread(quotes, bid_col, ask_col)
    if price_col in trades.columns and side_col in trades.columns:
        impact_result = _fit_propagator_impact(trades[price_col], _signed_volume(trades, side_col))

    return CalibratedParams(
        arrival_rate=arrival_rate,
        size_dist_method=size_method,
        mean_spread=spread,
        tick_size=tick_size,
        time_unit=time_unit,
        **size_result,
        **impact_result,
    )

lumina_lob.data.replay

Replay historical tick events through the matching engine.

Classes

ReplayEngine

Replay a merged stream of quote updates and trades through the engine.

For every quote event, the previous synthetic best-bid/best-ask limit orders are cancelled and replaced by fresh ones at the new quoted prices. For every trade event, a market order in the trade direction is submitted with the traded quantity. After each event the current best spread is recorded, allowing a comparison between the real quote spread distribution and the spread distribution produced by the engine.

Source code in lumina_lob/data/replay.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
class ReplayEngine:
    """Replay a merged stream of quote updates and trades through the engine.

    For every quote event, the previous synthetic best-bid/best-ask limit orders
    are cancelled and replaced by fresh ones at the new quoted prices.  For
    every trade event, a market order in the trade direction is submitted with
    the traded quantity.  After each event the current best spread is recorded,
    allowing a comparison between the real quote spread distribution and the
    spread distribution produced by the engine.
    """

    def __init__(self, book: OrderBook | None = None, engine: MatchingEngine | None = None, default_quote_size: int = 100):
        self.book = book if book is not None else OrderBook()
        self.engine = engine if engine is not None else MatchingEngine(self.book)
        self.default_quote_size = int(default_quote_size)
        self._order_id = 0
        self._quote_order_ids: list[int] = []
        self._timestamps: list[pd.Timestamp] = []
        self._spreads: list[float] = []
        self._best_bids: list[float | None] = []
        self._best_asks: list[float | None] = []
        self._trade_count = 0
        self._trade_volume = 0.0

    def replay(
        self,
        events: pd.DataFrame,
        event_col: str = "event_type",
        bid_col: str = "bid_px",
        ask_col: str = "ask_px",
        side_col: str = "side",
        size_col: str = "size",
    ) -> pd.DataFrame:
        """Process ``events`` in row order and return a metrics DataFrame.

        Parameters
        ----------
        events:
            DataFrame with at least ``timestamp`` and ``event_type`` columns.
            ``event_type`` values must be ``"quote"`` or ``"trade"``.  Quote rows
            must contain ``bid_col`` and ``ask_col``.  Trade rows must contain
            ``side_col`` and ``size_col``.
        event_col:
            Name of the column that distinguishes quote and trade events.
        bid_col, ask_col:
            Column names for quoted bid/ask prices.
        side_col, size_col:
            Column names for trade side and trade size.

        Returns
        -------
        pandas.DataFrame
            One row per event with columns ``timestamp``, ``spread``,
            ``best_bid``, and ``best_ask``.
        """
        if events.empty:
            raise ValueError("events DataFrame is empty")
        if "timestamp" not in events.columns or event_col not in events.columns:
            raise ValueError(f"events DataFrame must contain 'timestamp' and '{event_col}' columns")

        for _, row in events.iterrows():
            kind = str(row[event_col]).lower()
            self._clear_quotes()
            if kind == "quote":
                self._post_quote(row, bid_col, ask_col)
            elif kind == "trade":
                self._execute_trade(row, side_col, size_col)
            else:
                raise ValueError(f"unsupported event_type: {row[event_col]}")
            self._record(row["timestamp"])

        return pd.DataFrame(
            {
                "timestamp": self._timestamps,
                "spread": self._spreads,
                "best_bid": self._best_bids,
                "best_ask": self._best_asks,
            }
        )

    def _clear_quotes(self) -> None:
        for order_id in self._quote_order_ids:
            self.book.cancel(order_id)
        self._quote_order_ids = []

    def _post_quote(self, row: pd.Series, bid_col: str, ask_col: str) -> None:
        bid = row.get(bid_col)
        ask = row.get(ask_col)
        if pd.notna(bid):
            order_id = self._next_order_id()
            order = Order(
                order_id=order_id,
                side=Side.BID,
                price=float(bid),
                qty=self.default_quote_size,
                order_type=OrderType.LIMIT,
            )
            self.engine.process(order)
            self._quote_order_ids.append(order_id)
        if pd.notna(ask):
            order_id = self._next_order_id()
            order = Order(
                order_id=order_id,
                side=Side.ASK,
                price=float(ask),
                qty=self.default_quote_size,
                order_type=OrderType.LIMIT,
            )
            self.engine.process(order)
            self._quote_order_ids.append(order_id)

    def _execute_trade(self, row: pd.Series, side_col: str, size_col: str) -> None:
        side_value = row.get(side_col)
        size_value = row.get(size_col)
        if pd.isna(size_value):
            return
        size = int(size_value)
        if size <= 0:
            return
        sign = _sign_from_value(side_value)
        if sign == 0:
            return
        side = Side.BID if sign > 0 else Side.ASK
        order = Order(
            order_id=self._next_order_id(),
            side=side,
            price=None,
            qty=size,
            order_type=OrderType.MARKET,
        )
        self.engine.process(order)
        self._trade_count += 1
        self._trade_volume += size

    def _record(self, timestamp: Any) -> None:
        self._timestamps.append(pd.Timestamp(timestamp))
        bid = self.book.best_bid
        ask = self.book.best_ask
        self._best_bids.append(bid)
        self._best_asks.append(ask)
        if bid is not None and ask is not None:
            self._spreads.append(float(ask - bid))
        else:
            self._spreads.append(np.nan)

    def _next_order_id(self) -> int:
        self._order_id += 1
        return self._order_id
Methods:
replay(events, event_col='event_type', bid_col='bid_px', ask_col='ask_px', side_col='side', size_col='size')

Process events in row order and return a metrics DataFrame.

Parameters

events: DataFrame with at least timestamp and event_type columns. event_type values must be "quote" or "trade". Quote rows must contain bid_col and ask_col. Trade rows must contain side_col and size_col. event_col: Name of the column that distinguishes quote and trade events. bid_col, ask_col: Column names for quoted bid/ask prices. side_col, size_col: Column names for trade side and trade size.

Returns

pandas.DataFrame One row per event with columns timestamp, spread, best_bid, and best_ask.

Source code in lumina_lob/data/replay.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def replay(
    self,
    events: pd.DataFrame,
    event_col: str = "event_type",
    bid_col: str = "bid_px",
    ask_col: str = "ask_px",
    side_col: str = "side",
    size_col: str = "size",
) -> pd.DataFrame:
    """Process ``events`` in row order and return a metrics DataFrame.

    Parameters
    ----------
    events:
        DataFrame with at least ``timestamp`` and ``event_type`` columns.
        ``event_type`` values must be ``"quote"`` or ``"trade"``.  Quote rows
        must contain ``bid_col`` and ``ask_col``.  Trade rows must contain
        ``side_col`` and ``size_col``.
    event_col:
        Name of the column that distinguishes quote and trade events.
    bid_col, ask_col:
        Column names for quoted bid/ask prices.
    side_col, size_col:
        Column names for trade side and trade size.

    Returns
    -------
    pandas.DataFrame
        One row per event with columns ``timestamp``, ``spread``,
        ``best_bid``, and ``best_ask``.
    """
    if events.empty:
        raise ValueError("events DataFrame is empty")
    if "timestamp" not in events.columns or event_col not in events.columns:
        raise ValueError(f"events DataFrame must contain 'timestamp' and '{event_col}' columns")

    for _, row in events.iterrows():
        kind = str(row[event_col]).lower()
        self._clear_quotes()
        if kind == "quote":
            self._post_quote(row, bid_col, ask_col)
        elif kind == "trade":
            self._execute_trade(row, side_col, size_col)
        else:
            raise ValueError(f"unsupported event_type: {row[event_col]}")
        self._record(row["timestamp"])

    return pd.DataFrame(
        {
            "timestamp": self._timestamps,
            "spread": self._spreads,
            "best_bid": self._best_bids,
            "best_ask": self._best_asks,
        }
    )

Functions:

validate_spread_distribution(real_spreads, simulated_spreads, bins=None)

Return a histogram-overlap similarity score between two spread distributions.

The score is in [0, 1] where 1 means identical normalized histograms and 0 means no overlap.

Source code in lumina_lob/data/replay.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
def validate_spread_distribution(real_spreads: pd.Series, simulated_spreads: pd.Series, bins: int | None = None) -> float:
    """Return a histogram-overlap similarity score between two spread distributions.

    The score is in ``[0, 1]`` where ``1`` means identical normalized histograms and
    ``0`` means no overlap.
    """
    real = np.asarray(real_spreads.dropna(), dtype=float)
    sim = np.asarray(simulated_spreads.dropna(), dtype=float)
    if real.size == 0 or sim.size == 0:
        return 0.0

    if bins is None:
        bins = max(5, int(min(real.size, sim.size) ** 0.5))
    lo = min(real.min(), sim.min())
    hi = max(real.max(), sim.max())
    if hi <= lo:
        return 1.0 if np.allclose(real, sim) else 0.0

    edges = np.linspace(lo, hi, bins + 1)
    real_hist, _ = np.histogram(real, bins=edges)
    sim_hist, _ = np.histogram(sim, bins=edges)
    real_density = real_hist / real_hist.sum() if real_hist.sum() else np.zeros_like(real_hist)
    sim_density = sim_hist / sim_hist.sum() if sim_hist.sum() else np.zeros_like(sim_hist)
    overlap = np.minimum(real_density, sim_density).sum()
    return float(overlap)