Skip to content

Visualization API

The lumina_lob.viz package provides Matplotlib-based plotting and animation export.

lumina_lob.viz.depth_ladder

Depth-ladder visualization for an order book.

Functions:

plot_depth_ladder(book, top_n=10)

Render a horizontal depth-ladder plot for book.

Parameters

book: An OrderBook instance (Python lumina_lob.core.OrderBook or the C++ lumina_lob._core.OrderBook). top_n: Number of price levels to show on each side.

Returns

(fig, ax) from matplotlib.

Source code in lumina_lob/viz/depth_ladder.py 
 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
def plot_depth_ladder(book: Any, top_n: int = 10) -> tuple["Figure", Any]:
    """Render a horizontal depth-ladder plot for *book*.

    Parameters
    ----------
    book:
        An ``OrderBook`` instance (Python ``lumina_lob.core.OrderBook`` or the
        C++ ``lumina_lob._core.OrderBook``).
    top_n:
        Number of price levels to show on each side.

    Returns
    -------
    ``(fig, ax)`` from matplotlib.
    """
    _ensure_matplotlib()

    if top_n <= 0:
        raise ValueError("top_n must be positive")

    bid_side, ask_side = _detect_side_enums(book)
    bid_items = _to_level_items(book.depth(bid_side, top_n), reverse=True)
    ask_items = _to_level_items(book.depth(ask_side, top_n), reverse=False)

    all_prices = sorted({price for price, _ in bid_items + ask_items})
    if not all_prices:
        raise ValueError("book has no price levels to plot")

    y_positions = {price: idx for idx, price in enumerate(all_prices)}

    fig, ax = plt.subplots(figsize=(8, max(4, len(all_prices) * 0.4)))

    for price, qty in bid_items:
        ax.barh(
            y_positions[price],
            -qty,
            color="#2ecc71",
            edgecolor="black",
            height=0.6,
            label="Bid" if price == bid_items[0][0] else "",
        )

    for price, qty in ask_items:
        ax.barh(
            y_positions[price],
            qty,
            color="#e74c3c",
            edgecolor="black",
            height=0.6,
            label="Ask" if price == ask_items[0][0] else "",
        )

    ax.axvline(0, color="black", linewidth=0.8)
    ax.set_yticks(range(len(all_prices)))
    ax.set_yticklabels(str(p) for p in all_prices)
    ax.set_xlabel("Quantity")
    ax.set_ylabel("Price")
    ax.set_title("Limit Order Book Depth Ladder")
    ax.legend(loc="lower right")
    fig.tight_layout()

    return fig, ax

lumina_lob.viz.history

Time-series visualisation for simulation history.

Functions:

plot_simulation_history(history)

Plot mid price, spread, and trades from a simulation history.

Parameters

history: A list of step records (as returned by Simulation.run) or a pandas.DataFrame produced by Simulation.to_dataframe().

Returns

(fig, axes) from matplotlib. axes is a 3-element array with: - mid price + trade markers - bid-ask spread - trade volume per step

Source code in lumina_lob/viz/history.py 
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
def plot_simulation_history(history: Any) -> tuple["Figure", Any]:
    """Plot mid price, spread, and trades from a simulation history.

    Parameters
    ----------
    history:
        A list of step records (as returned by ``Simulation.run``) or a
        ``pandas.DataFrame`` produced by ``Simulation.to_dataframe()``.

    Returns
    -------
    ``(fig, axes)`` from matplotlib. ``axes`` is a 3-element array with:
    - mid price + trade markers
    - bid-ask spread
    - trade volume per step
    """
    _ensure_matplotlib()

    df = _to_dataframe(history)

    required = {"step", "mid_price", "spread", "trade_count", "trade_volume"}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"history is missing required columns: {sorted(missing)}")

    fig, axes = plt.subplots(3, 1, figsize=(10, 9), sharex=True)
    assert len(axes) == 3  # appease type checkers / maintainers

    # Mid price with trade markers.
    ax_mid = axes[0]
    ax_mid.plot(df["step"], df["mid_price"], color="#2980b9", label="Mid price")
    trade_steps = df.loc[df["trade_count"] > 0, "step"]
    trade_prices = df.loc[df["trade_count"] > 0, "mid_price"]
    if not trade_steps.empty:
        ax_mid.scatter(trade_steps, trade_prices, color="#e67e22", marker="x", label="Trade")
    ax_mid.set_ylabel("Mid price")
    ax_mid.set_title("Simulation time series")
    ax_mid.legend(loc="best")
    ax_mid.grid(axis="y", linestyle="--", alpha=0.5)

    # Spread.
    ax_spread = axes[1]
    ax_spread.plot(df["step"], df["spread"], color="#8e44ad", label="Spread")
    ax_spread.set_ylabel("Spread")
    ax_spread.legend(loc="best")
    ax_spread.grid(axis="y", linestyle="--", alpha=0.5)

    # Trade volume.
    ax_vol = axes[2]
    ax_vol.bar(df["step"], df["trade_volume"], color="#c0392b", label="Trade volume")
    ax_vol.set_xlabel("Step")
    ax_vol.set_ylabel("Volume")
    ax_vol.legend(loc="best")
    ax_vol.grid(axis="y", linestyle="--", alpha=0.5)

    fig.tight_layout()
    return fig, axes

lumina_lob.viz.realtime

Real-time simulation visualisation using matplotlib.

Classes

SimulationAnimator

Animate a running Simulation with depth ladder + mid-price trace.

Source code in lumina_lob/viz/realtime.py 
 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
class SimulationAnimator:
    """Animate a running ``Simulation`` with depth ladder + mid-price trace."""

    def __init__(
        self,
        simulation: Simulation,
        top_n: int = 10,
        history_window: int = 50,
        interval_ms: int = 200,
    ) -> None:
        _ensure_matplotlib()

        self.simulation = simulation
        self.top_n = top_n
        self.history_window = history_window
        self.interval_ms = interval_ms

        self.fig = plt.figure(figsize=(12, 5))
        self.ax_depth = self.fig.add_subplot(1, 2, 1)
        self.ax_price = self.fig.add_subplot(1, 2, 2)

        self._price_line, = self.ax_price.plot([], [], color="#2980b9", label="Mid price")
        self._trade_scatter = self.ax_price.scatter([], [], color="#e67e22", marker="x", label="Trade")
        self.ax_price.set_xlabel("Step")
        self.ax_price.set_ylabel("Mid price")
        self.ax_price.set_title("Mid-price trace")
        self.ax_price.legend(loc="best")
        self.ax_price.grid(axis="y", linestyle="--", alpha=0.5)

    def _update_price_axis(self) -> None:
        """Rescale the price trace to the available history."""
        history = self.simulation.history[-self.history_window :]
        if not history:
            return
        valid = [(r["step"], r["mid_price"]) for r in history if r["mid_price"] is not None]
        if not valid:
            return
        steps, mids = zip(*valid, strict=False)
        self._price_line.set_data(steps, mids)
        self.ax_price.set_xlim(min(steps), max(max(steps), min(steps) + 1))
        y_min, y_max = min(mids), max(mids)
        pad = max((y_max - y_min) * 0.1, 0.01)
        self.ax_price.set_ylim(y_min - pad, y_max + pad)

        trade_steps = [
            cast(int, r["step"]) for r in history
            if cast(int, r["trade_count"]) > 0 and r["mid_price"] is not None
        ]
        trade_mids = [
            cast(float, r["mid_price"]) for r in history
            if cast(int, r["trade_count"]) > 0 and r["mid_price"] is not None
        ]
        if trade_mids:
            self._trade_scatter.set_offsets(np.column_stack([trade_steps, trade_mids]))
        else:
            self._trade_scatter.set_offsets(np.zeros((0, 2)))

    def update(self, _frame: int | None = None) -> tuple[Any, ...]:
        """Advance the simulation by one step and redraw both panels."""
        self.simulation.step()
        _draw_depth_ladder(self.ax_depth, self.simulation.book, self.top_n)
        self._update_price_axis()
        self.fig.tight_layout()
        return (self._price_line, self._trade_scatter)

    def run(self, n_steps: int = 100) -> Any:
        """Return a ``FuncAnimation`` that runs *n_steps* frames."""
        _ensure_matplotlib()
        return FuncAnimation(
            self.fig,
            self.update,
            frames=n_steps,
            interval=self.interval_ms,
            blit=False,
            repeat=False,
        )
Methods:
run(n_steps=100)

Return a FuncAnimation that runs n_steps frames.

Source code in lumina_lob/viz/realtime.py 
141
142
143
144
145
146
147
148
149
150
151
def run(self, n_steps: int = 100) -> Any:
    """Return a ``FuncAnimation`` that runs *n_steps* frames."""
    _ensure_matplotlib()
    return FuncAnimation(
        self.fig,
        self.update,
        frames=n_steps,
        interval=self.interval_ms,
        blit=False,
        repeat=False,
    )
update(_frame=None)

Advance the simulation by one step and redraw both panels.

Source code in lumina_lob/viz/realtime.py 
133
134
135
136
137
138
139
def update(self, _frame: int | None = None) -> tuple[Any, ...]:
    """Advance the simulation by one step and redraw both panels."""
    self.simulation.step()
    _draw_depth_ladder(self.ax_depth, self.simulation.book, self.top_n)
    self._update_price_axis()
    self.fig.tight_layout()
    return (self._price_line, self._trade_scatter)

Functions:

run_animation(simulation, n_steps=100, top_n=10, history_window=50, interval_ms=200)

Create and return a running animation for simulation.

Parameters

simulation: The Simulation to animate. n_steps: Number of animation frames. top_n: Number of price levels to show on each side of the depth ladder. history_window: How many recent steps to show in the mid-price trace. interval_ms: Delay between frames in milliseconds.

Returns

A matplotlib.animation.FuncAnimation instance. Call plt.show() or save it to display/record.

Source code in lumina_lob/viz/realtime.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def run_animation(
    simulation: Simulation,
    n_steps: int = 100,
    top_n: int = 10,
    history_window: int = 50,
    interval_ms: int = 200,
) -> Any:
    """Create and return a running animation for *simulation*.

    Parameters
    ----------
    simulation:
        The ``Simulation`` to animate.
    n_steps:
        Number of animation frames.
    top_n:
        Number of price levels to show on each side of the depth ladder.
    history_window:
        How many recent steps to show in the mid-price trace.
    interval_ms:
        Delay between frames in milliseconds.

    Returns
    -------
    A ``matplotlib.animation.FuncAnimation`` instance. Call ``plt.show()`` or
    save it to display/record.
    """
    animator = SimulationAnimator(
        simulation,
        top_n=top_n,
        history_window=history_window,
        interval_ms=interval_ms,
    )
    return animator.run(n_steps)

save_animation(animation, path, fps=5)

Save a simulation animation to path as GIF or MP4.

Parameters

animation: A matplotlib.animation.FuncAnimation instance (e.g. from run_animation). path: Destination file path. Supported extensions: .gif (requires Pillow) and .mp4 (requires ffmpeg). fps: Frames per second for the output file.

Raises

ValueError If the file extension is unsupported or the required writer is not available.

Source code in lumina_lob/viz/realtime.py 
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
def save_animation(
    animation: Any,
    path: str | Path,
    fps: int = 5,
) -> None:
    """Save a simulation animation to *path* as GIF or MP4.

    Parameters
    ----------
    animation:
        A ``matplotlib.animation.FuncAnimation`` instance (e.g. from
        ``run_animation``).
    path:
        Destination file path. Supported extensions: ``.gif`` (requires
        Pillow) and ``.mp4`` (requires ffmpeg).
    fps:
        Frames per second for the output file.

    Raises
    ------
    ValueError
        If the file extension is unsupported or the required writer is not
        available.
    """
    _ensure_matplotlib()

    out = Path(path)
    writer_map = {".gif": "pillow", ".mp4": "ffmpeg"}
    ext = out.suffix.lower()
    if ext not in writer_map:
        raise ValueError(f"Unsupported animation format: {ext!r}. Use .gif or .mp4.")

    writer = writer_map[ext]
    if not _animation.writers.is_available(writer):
        raise ValueError(
            f"{writer!r} writer is not available. "
            f"Install it to save {ext} animations (e.g. `pip install pillow` for .gif)."
        )

    out.parent.mkdir(parents=True, exist_ok=True)
    animation.save(str(out), writer=writer, fps=fps)