Skip to content

Simulation API

The lumina_lob.simulation module orchestrates the engine, agents, and market model.

lumina_lob.simulation

Simulation orchestrator: runs agents and matching engine through time.

Classes

Simulation

Orchestrate a multi-agent limit order book simulation.

A Simulation owns one OrderBook, one MatchingEngine, one ReferencePriceProcess, and a list of Agent instances. At each discrete time step the reference price is advanced, every agent is asked for orders, and those orders are submitted to the matching engine. Fill notifications are routed back to agents that implement on_fill (e.g. market makers updating inventory).

Parameters

book: OrderBook to use. A fresh book is created if omitted. engine: MatchingEngine to use. Created from book if omitted. reference_price: ReferencePriceProcess to use. Created with default parameters if omitted. If seed is also provided it is passed to the default process. agents: List of Agent instances participating in the simulation. seed: Optional RNG seed for the default ReferencePriceProcess.

Source code in lumina_lob/simulation.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
 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
class Simulation:
    """Orchestrate a multi-agent limit order book simulation.

    A ``Simulation`` owns one ``OrderBook``, one ``MatchingEngine``, one
    ``ReferencePriceProcess``, and a list of ``Agent`` instances. At each
    discrete time step the reference price is advanced, every agent is asked
    for orders, and those orders are submitted to the matching engine.  Fill
    notifications are routed back to agents that implement ``on_fill`` (e.g.
    market makers updating inventory).

    Parameters
    ----------
    book:
        ``OrderBook`` to use. A fresh book is created if omitted.
    engine:
        ``MatchingEngine`` to use. Created from ``book`` if omitted.
    reference_price:
        ``ReferencePriceProcess`` to use. Created with default parameters if
        omitted.  If ``seed`` is also provided it is passed to the default
        process.
    agents:
        List of ``Agent`` instances participating in the simulation.
    seed:
        Optional RNG seed for the default ``ReferencePriceProcess``.
    """

    def __init__(
        self,
        book: OrderBook | None = None,
        engine: MatchingEngine | None = None,
        reference_price: ReferencePriceProcess | None = None,
        agents: list[Agent] | None = None,
        seed: int | None = None,
    ) -> None:
        self.book = book if book is not None else OrderBook()
        self.engine = engine if engine is not None else MatchingEngine(self.book)
        if reference_price is not None:
            self.reference_price = reference_price
        else:
            self.reference_price = ReferencePriceProcess(seed=seed)
        self.agents: list[Agent] = list(agents) if agents is not None else []
        self.history: list[dict[str, object]] = []
        self._next_order_id = 1
        self._order_owner: dict[int, Agent] = {}
        self._order_side: dict[int, Side] = {}

    def step(self) -> dict[str, object]:
        """Advance the simulation by one time step.

        Returns
        -------
        A dictionary of metrics for the step:
        ``step``, ``reference_price``, ``best_bid``, ``best_ask``,
        ``mid_price``, ``spread``, ``trade_count``, ``trade_volume``,
        ``book_size``.
        """
        reference_price = self.reference_price.step()
        pre_trade_count = len(self.book.trades)

        for agent in self.agents:
            orders = agent.act(reference_price, self.book)
            for order in orders:
                self._submit(order, agent)

        self._notify_fills(pre_trade_count)

        new_trades = self.book.trades[pre_trade_count:]
        trade_count = len(new_trades)
        trade_volume = sum(qty for _, _, qty in new_trades)
        signed_volume = 0
        for aggressor_id, _, qty in new_trades:
            side = self._order_side.get(aggressor_id)
            if side == Side.BID:
                signed_volume += qty
            elif side == Side.ASK:
                signed_volume -= qty

        record: dict[str, object] = {
            "step": len(self.history) + 1,
            "reference_price": reference_price,
            "best_bid": self.book.best_bid,
            "best_ask": self.book.best_ask,
            "mid_price": self.book.mid_price,
            "spread": self.book.spread,
            "trade_count": trade_count,
            "trade_volume": trade_volume,
            "signed_volume": signed_volume,
            "book_size": len(self.book),
        }
        self.history.append(record)
        return record

    def run(self, n_steps: int) -> list[dict[str, object]]:
        """Run ``n_steps`` and return the full history list."""
        if n_steps <= 0:
            raise ValueError("n_steps must be positive")
        for _ in range(n_steps):
            self.step()
        return list(self.history)

    def to_dataframe(self) -> Any:
        """Return the simulation history as a pandas ``DataFrame``."""
        import pandas as pd

        return pd.DataFrame(self.history)

    def _submit(self, order: Order, agent: Agent) -> None:
        """Assign a globally unique order id and route the order to the engine."""
        new_id = self._next_order_id
        self._next_order_id += 1
        order.order_id = new_id
        self._order_owner[new_id] = agent
        self._order_side[new_id] = order.side
        self.engine.process(order)

    def _notify_fills(self, pre_trade_count: int) -> None:
        """Call ``on_fill`` on agents whose orders traded this step.

        Self-trades (where the aggressor and resting order belong to the same
        agent) are skipped so that an agent does not artificially move its own
        inventory.
        """
        for aggressor_id, resting_id, qty in self.book.trades[pre_trade_count:]:
            owner_aggr = self._order_owner.get(aggressor_id)
            owner_rest = self._order_owner.get(resting_id)
            if owner_aggr is owner_rest:
                continue

            side_aggr = self._order_side.get(aggressor_id)
            side_rest = self._order_side.get(resting_id)

            if owner_aggr is not None:
                on_fill = getattr(owner_aggr, "on_fill", None)
                if callable(on_fill):
                    on_fill(side_aggr, qty)

            if owner_rest is not None:
                on_fill = getattr(owner_rest, "on_fill", None)
                if callable(on_fill):
                    on_fill(side_rest, qty)
Methods:
run(n_steps)

Run n_steps and return the full history list.

Source code in lumina_lob/simulation.py 
103
104
105
106
107
108
109
def run(self, n_steps: int) -> list[dict[str, object]]:
    """Run ``n_steps`` and return the full history list."""
    if n_steps <= 0:
        raise ValueError("n_steps must be positive")
    for _ in range(n_steps):
        self.step()
    return list(self.history)
step()

Advance the simulation by one time step.

Returns

A dictionary of metrics for the step: step, reference_price, best_bid, best_ask, mid_price, spread, trade_count, trade_volume, book_size.

Source code in lumina_lob/simulation.py 
 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
def step(self) -> dict[str, object]:
    """Advance the simulation by one time step.

    Returns
    -------
    A dictionary of metrics for the step:
    ``step``, ``reference_price``, ``best_bid``, ``best_ask``,
    ``mid_price``, ``spread``, ``trade_count``, ``trade_volume``,
    ``book_size``.
    """
    reference_price = self.reference_price.step()
    pre_trade_count = len(self.book.trades)

    for agent in self.agents:
        orders = agent.act(reference_price, self.book)
        for order in orders:
            self._submit(order, agent)

    self._notify_fills(pre_trade_count)

    new_trades = self.book.trades[pre_trade_count:]
    trade_count = len(new_trades)
    trade_volume = sum(qty for _, _, qty in new_trades)
    signed_volume = 0
    for aggressor_id, _, qty in new_trades:
        side = self._order_side.get(aggressor_id)
        if side == Side.BID:
            signed_volume += qty
        elif side == Side.ASK:
            signed_volume -= qty

    record: dict[str, object] = {
        "step": len(self.history) + 1,
        "reference_price": reference_price,
        "best_bid": self.book.best_bid,
        "best_ask": self.book.best_ask,
        "mid_price": self.book.mid_price,
        "spread": self.book.spread,
        "trade_count": trade_count,
        "trade_volume": trade_volume,
        "signed_volume": signed_volume,
        "book_size": len(self.book),
    }
    self.history.append(record)
    return record
to_dataframe()

Return the simulation history as a pandas DataFrame.

Source code in lumina_lob/simulation.py 
111
112
113
114
115
def to_dataframe(self) -> Any:
    """Return the simulation history as a pandas ``DataFrame``."""
    import pandas as pd

    return pd.DataFrame(self.history)