RL API

The lumina_lob.rl package provides a Gymnasium environment and training helpers for reinforcement-learning market makers.

lumina_lob.rl.env

Gymnasium environment for training an RL market maker.

Classes

MarketMakerEnv

Bases: Env[ndarray, ndarray]

A Gymnasium environment where an agent acts as a single market maker.

The observation combines the current limit-order-book state, the agent's inventory, and its mark-to-market P&L. The action is a continuous vector that controls bid/ask quote offsets and sizes.

Parameters

max_steps: Maximum number of steps per episode. warmup_steps: Number of steps to run before the agent starts observing, so the book has initial liquidity. tick_size: Price tick size used for rounding and normalisation. max_quote_offset_ticks: Maximum number of ticks away from the mid price the agent may quote. min_quote_size: Minimum quantity per quote. max_quote_size: Maximum quantity per quote. inventory_penalty: Coefficient for the quadratic inventory penalty applied each step. seed: Optional global RNG seed.

Source code in lumina_lob/rl/env.py

class MarketMakerEnv(gym.Env[np.ndarray, np.ndarray]):
    """A Gymnasium environment where an agent acts as a single market maker.

    The observation combines the current limit-order-book state, the agent's
    inventory, and its mark-to-market P&L.  The action is a continuous vector
    that controls bid/ask quote offsets and sizes.

    Parameters
    ----------
    max_steps:
        Maximum number of steps per episode.
    warmup_steps:
        Number of steps to run before the agent starts observing, so the book
        has initial liquidity.
    tick_size:
        Price tick size used for rounding and normalisation.
    max_quote_offset_ticks:
        Maximum number of ticks away from the mid price the agent may quote.
    min_quote_size:
        Minimum quantity per quote.
    max_quote_size:
        Maximum quantity per quote.
    inventory_penalty:
        Coefficient for the quadratic inventory penalty applied each step.
    seed:
        Optional global RNG seed.
    """

    def __init__(
        self,
        max_steps: int = 200,
        warmup_steps: int = 10,
        tick_size: float = 0.01,
        max_quote_offset_ticks: int = 5,
        min_quote_size: int = 10,
        max_quote_size: int = 100,
        inventory_penalty: float = 0.0,
        seed: int | None = None,
    ) -> None:
        super().__init__()
        self.max_steps = max(1, int(max_steps))
        self.warmup_steps = max(0, int(warmup_steps))
        self.tick_size = float(tick_size)
        self.max_quote_offset_ticks = max(0, int(max_quote_offset_ticks))
        self.min_quote_size = max(1, int(min_quote_size))
        self.max_quote_size = max(self.min_quote_size, int(max_quote_size))
        self.inventory_penalty = float(inventory_penalty)
        self._seed = seed

        self.observation_features: list[str] = [
            "best_bid_norm",
            "best_ask_norm",
            "mid_price_norm",
            "spread_norm",
            "bid_depth_norm",
            "ask_depth_norm",
            "inventory_norm",
            "cash_norm",
            "unrealised_pnl_norm",
            "time_fraction",
        ]
        n_features = len(self.observation_features)
        self.observation_space = gym.spaces.Box(
            low=-np.inf,
            high=np.inf,
            shape=(n_features,),
            dtype=np.float32,
        )
        # Continuous action: [bid_offset, ask_offset, bid_size, ask_size] in [-1, 1].
        self.action_space = gym.spaces.Box(
            low=-1.0,
            high=1.0,
            shape=(4,),
            dtype=np.float32,
        )

        self.simulation: Simulation | None = None
        self._inventory = 0.0
        self._cash = 0.0
        self._avg_fill_price = 0.0
        self._current_step = 0
        self._reference_price = 100.0
        self._previous_total_pnl = 0.0
        self._pending_action: np.ndarray | None = None
        self._proxy = self._AgentProxy(self)

    def reset(
        self,
        *,
        seed: int | None = None,
        options: dict[str, Any] | None = None,
    ) -> tuple[np.ndarray, dict[str, Any]]:
        super().reset(seed=seed)
        effective_seed = seed if seed is not None else self._seed

        book = OrderBook()
        engine = MatchingEngine(book)
        reference_price = ReferencePriceProcess(seed=effective_seed)
        self._reference_price = float(reference_price.price)

        noise = NoiseTrader(
            arrival_rate=0.5,
            tick_size=self.tick_size,
            seed=effective_seed,
        )
        informed = InformedTrader(
            signal="bullish",
            trade_size=50,
            order_type="market",
            participation_rate=0.1,
            tick_size=self.tick_size,
            seed=effective_seed,
        )

        self.simulation = Simulation(
            book=book,
            engine=engine,
            reference_price=reference_price,
            agents=[noise, informed, self._proxy],
            seed=effective_seed,
        )

        self._inventory = 0.0
        self._cash = 0.0
        self._avg_fill_price = 0.0
        self._current_step = 0
        self._pending_action = None

        for _ in range(self.warmup_steps):
            self.simulation.step()
            self._current_step += 1

        self._previous_total_pnl = self._total_pnl()

        observation = self._get_observation()
        info: dict[str, Any] = {"reference_price": self._reference_price}
        return observation, info

    def step(self, action: np.ndarray) -> tuple[np.ndarray, float, bool, bool, dict[str, Any]]:
        """Advance the market by one step and return the new observation."""
        if self.simulation is None:
            raise RuntimeError("Environment must be reset before calling step()")

        action_arr = np.asarray(action, dtype=np.float32)
        if action_arr.shape != self.action_space.shape:
            raise ValueError(
                f"action shape {action_arr.shape} does not match {self.action_space.shape}"
            )
        self._pending_action = action_arr

        metrics = self.simulation.step()
        self._current_step += 1
        self._reference_price = cast(float, metrics["reference_price"])

        observation = self._get_observation()
        reward = self._compute_reward()
        terminated = False
        truncated = self._current_step >= self.max_steps
        info: dict[str, Any] = {
            "reference_price": self._reference_price,
            "inventory": self._inventory,
            "cash": self._cash,
        }
        return observation, reward, terminated, truncated, info

    def _get_observation(self) -> np.ndarray:
        """Build the normalised observation vector."""
        if self.simulation is None:
            shape = self.observation_space.shape
            if shape is None:
                shape = (len(self.observation_features),)
            return np.zeros(shape, dtype=np.float32)

        book = self.simulation.book
        best_bid = book.best_bid if book.best_bid is not None else self._reference_price
        best_ask = book.best_ask if book.best_ask is not None else self._reference_price
        mid = book.mid_price if book.mid_price is not None else self._reference_price
        spread = best_ask - best_bid

        bid_depth = sum(level.total_qty for level in book.bids.values())
        ask_depth = sum(level.total_qty for level in book.asks.values())

        unrealised_pnl = self._inventory * (mid - self._avg_fill_price) if self._avg_fill_price else 0.0

        max_position = 1_000.0
        norm = max(self._reference_price, 1e-9)
        obs = np.array(
            [
                best_bid / norm,
                best_ask / norm,
                mid / norm,
                spread / norm,
                bid_depth / max_position,
                ask_depth / max_position,
                self._inventory / max_position,
                self._cash / norm,
                unrealised_pnl / norm,
                self._current_step / self.max_steps,
            ],
            dtype=np.float32,
        )
        return obs

    def _update_inventory(
        self, side: Side, qty: int, price: float | None = None
    ) -> None:
        """Record a fill against the agent's inventory."""
        fill_qty = float(qty)
        fill_price = self._fill_price(price)
        if side == Side.BID:  # BID: agent bought
            cost = fill_qty * fill_price
            total_value = self._inventory * self._avg_fill_price + cost
            self._inventory += fill_qty
            self._cash -= cost
            if self._inventory > 0:
                self._avg_fill_price = total_value / self._inventory
        else:  # ASK: agent sold
            proceeds = fill_qty * fill_price
            self._inventory -= fill_qty
            self._cash += proceeds
            if self._inventory == 0:
                self._avg_fill_price = 0.0

    def _fill_price(self, price: float | None) -> float:
        """Resolve a fill price, falling back to mid or reference price."""
        if price is not None:
            return float(price)
        if self.simulation is not None:
            mid = self.simulation.book.mid_price
            if mid is not None:
                return float(mid)
        return float(self._reference_price)

    def _total_pnl(self) -> float:
        """Mark-to-market value of cash plus inventory at the current mid price."""
        mid = self._reference_price
        if self.simulation is not None:
            mid = self.simulation.book.mid_price if self.simulation.book.mid_price is not None else mid
        return float(self._cash + self._inventory * mid)

    def _compute_reward(self) -> float:
        """Return the change in mark-to-market P&L minus inventory penalty."""
        total_pnl = self._total_pnl()
        pnl_delta = total_pnl - self._previous_total_pnl
        self._previous_total_pnl = total_pnl
        penalty = self.inventory_penalty * (self._inventory ** 2)
        return float(pnl_delta - penalty)

    def _action_to_quotes(
        self, action: np.ndarray, reference_price: float
    ) -> list[Order]:
        """Translate a continuous action into bid/ask limit orders."""
        if self.simulation is None:
            return []

        book = self.simulation.book
        mid = book.mid_price if book.mid_price is not None else reference_price

        bid_offset = int(round(((float(action[0]) + 1.0) / 2.0) * self.max_quote_offset_ticks))
        ask_offset = int(round(((float(action[1]) + 1.0) / 2.0) * self.max_quote_offset_ticks))

        bid_size = int(round(((float(action[2]) + 1.0) / 2.0) * (self.max_quote_size - self.min_quote_size) + self.min_quote_size))
        ask_size = int(round(((float(action[3]) + 1.0) / 2.0) * (self.max_quote_size - self.min_quote_size) + self.min_quote_size))

        bid_tick = max(1, round((mid - bid_offset * self.tick_size) / self.tick_size))
        ask_tick = max(bid_tick + 1, round((mid + ask_offset * self.tick_size) / self.tick_size))
        bid_price = bid_tick * self.tick_size
        ask_price = ask_tick * self.tick_size

        orders: list[Order] = []
        if bid_size > 0:
            order = Order(
                order_id=0,
                side=Side.BID,
                price=bid_price,
                qty=bid_size,
            )
            order._agent_quote = True  # type: ignore[attr-defined]
            orders.append(order)
        if ask_size > 0:
            order = Order(
                order_id=0,
                side=Side.ASK,
                price=ask_price,
                qty=ask_size,
            )
            order._agent_quote = True  # type: ignore[attr-defined]
            orders.append(order)
        return orders

    def _cancel_agent_quotes(self) -> None:
        """Cancel any resting orders placed by the RL agent."""
        if self.simulation is None:
            return
        book = self.simulation.book
        for order_id in list(book.orders.keys()):
            if getattr(book.orders[order_id], "_agent_quote", False):
                book.cancel(order_id)

    class _AgentProxy(Agent):
        """Thin agent wrapper that turns the env action into limit orders."""

        def __init__(self, env: MarketMakerEnv) -> None:
            self.env = env

        def act(self, reference_price: float, book: OrderBook) -> list[Order]:
            """Cancel old quotes and submit new ones from the pending action."""
            self.env._cancel_agent_quotes()
            action = self.env._pending_action
            if action is None:
                return []
            return self.env._action_to_quotes(action, reference_price)

        def on_fill(self, side: Side, qty: int) -> None:
            """Forward fill notifications to the environment inventory tracker."""
            self.env._update_inventory(side, qty)

Methods:

step(action)

Advance the market by one step and return the new observation.

Source code in lumina_lob/rl/env.py

def step(self, action: np.ndarray) -> tuple[np.ndarray, float, bool, bool, dict[str, Any]]:
    """Advance the market by one step and return the new observation."""
    if self.simulation is None:
        raise RuntimeError("Environment must be reset before calling step()")

    action_arr = np.asarray(action, dtype=np.float32)
    if action_arr.shape != self.action_space.shape:
        raise ValueError(
            f"action shape {action_arr.shape} does not match {self.action_space.shape}"
        )
    self._pending_action = action_arr

    metrics = self.simulation.step()
    self._current_step += 1
    self._reference_price = cast(float, metrics["reference_price"])

    observation = self._get_observation()
    reward = self._compute_reward()
    terminated = False
    truncated = self._current_step >= self.max_steps
    info: dict[str, Any] = {
        "reference_price": self._reference_price,
        "inventory": self._inventory,
        "cash": self._cash,
    }
    return observation, reward, terminated, truncated, info

lumina_lob.rl.train

Training and evaluation helpers for RL market makers.

Classes

Functions:

evaluate_agent(model, env, n_eval_episodes=5)

Evaluate a trained model and return mean and std episode reward.

Source code in lumina_lob/rl/train.py

def evaluate_agent(
    model: Any,
    env: gym.Env[np.ndarray, np.ndarray],
    n_eval_episodes: int = 5,
) -> tuple[float, float]:
    """Evaluate a trained model and return mean and std episode reward."""
    mean_reward, std_reward = evaluate_policy(
        model,
        env,
        n_eval_episodes=n_eval_episodes,
        deterministic=True,
    )
    return float(cast(float, mean_reward)), float(cast(float, std_reward))

make_env(seed=0)

Return a no-argument factory that creates a fresh MarketMakerEnv.

The environment is wrapped with SB3's Monitor so training/evaluation logs episode statistics without warnings.

Source code in lumina_lob/rl/train.py

def make_env(seed: int = 0) -> Callable[[], gym.Env[np.ndarray, np.ndarray]]:
    """Return a no-argument factory that creates a fresh ``MarketMakerEnv``.

    The environment is wrapped with SB3's ``Monitor`` so training/evaluation
    logs episode statistics without warnings.
    """

    def _init() -> gym.Env[np.ndarray, np.ndarray]:
        return Monitor(MarketMakerEnv(seed=seed))

    return _init

save_model(model, path)

Save a Stable-Baselines3 model to disk.

Source code in lumina_lob/rl/train.py

def save_model(model: Any, path: Path) -> None:
    """Save a Stable-Baselines3 model to disk."""
    path.parent.mkdir(parents=True, exist_ok=True)
    model.save(path)

train_ppo(env, total_timesteps=10000, verbose=0, **kwargs)

Train a PPO agent on MarketMakerEnv.

Parameters

env: A Gymnasium environment (typically MarketMakerEnv). total_timesteps: Number of timesteps to train for. verbose: SB3 verbosity level. kwargs: Extra keyword arguments forwarded to PPO.

Returns

A trained PPO model.

Source code in lumina_lob/rl/train.py

def train_ppo(
    env: gym.Env[np.ndarray, np.ndarray],
    total_timesteps: int = 10_000,
    verbose: int = 0,
    **kwargs: Any,
) -> PPO:
    """Train a PPO agent on ``MarketMakerEnv``.

    Parameters
    ----------
    env:
        A Gymnasium environment (typically ``MarketMakerEnv``).
    total_timesteps:
        Number of timesteps to train for.
    verbose:
        SB3 verbosity level.
    kwargs:
        Extra keyword arguments forwarded to ``PPO``.

    Returns
    -------
    A trained ``PPO`` model.
    """
    device = kwargs.pop("device", "cpu")
    model = PPO("MlpPolicy", env, verbose=verbose, device=device, **kwargs)
    model.learn(total_timesteps=total_timesteps)
    return model

train_sac(env, total_timesteps=10000, verbose=0, **kwargs)

Train an SAC agent on MarketMakerEnv.

Parameters

env: A Gymnasium environment (typically MarketMakerEnv). total_timesteps: Number of timesteps to train for. verbose: SB3 verbosity level. kwargs: Extra keyword arguments forwarded to SAC.

Returns

A trained SAC model.

Source code in lumina_lob/rl/train.py

def train_sac(
    env: gym.Env[np.ndarray, np.ndarray],
    total_timesteps: int = 10_000,
    verbose: int = 0,
    **kwargs: Any,
) -> SAC:
    """Train an SAC agent on ``MarketMakerEnv``.

    Parameters
    ----------
    env:
        A Gymnasium environment (typically ``MarketMakerEnv``).
    total_timesteps:
        Number of timesteps to train for.
    verbose:
        SB3 verbosity level.
    kwargs:
        Extra keyword arguments forwarded to ``SAC``.

    Returns
    -------
    A trained ``SAC`` model.
    """
    device = kwargs.pop("device", "cpu")
    model = SAC("MlpPolicy", env, verbose=verbose, device=device, **kwargs)
    model.learn(total_timesteps=total_timesteps)
    return model

lumina_lob.rl.compare

Compare PPO and SAC market-maker agents on MarketMakerEnv.

Functions:

compare_ppo_sac(env_factory, total_timesteps=10000, n_eval_episodes=5)

Train PPO and SAC agents and return their evaluation statistics.

Parameters

env_factory: Callable that returns a fresh Gymnasium environment. total_timesteps: Number of timesteps to train each agent for. n_eval_episodes: Number of evaluation episodes per agent.

Returns

A dictionary with ppo and sac keys, each mapping to {"mean": float, "std": float} reward statistics.

Source code in lumina_lob/rl/compare.py

def compare_ppo_sac(
    env_factory: Callable[[], gym.Env[np.ndarray, np.ndarray]],
    total_timesteps: int = 10_000,
    n_eval_episodes: int = 5,
) -> dict[str, dict[str, float]]:
    """Train PPO and SAC agents and return their evaluation statistics.

    Parameters
    ----------
    env_factory:
        Callable that returns a fresh Gymnasium environment.
    total_timesteps:
        Number of timesteps to train each agent for.
    n_eval_episodes:
        Number of evaluation episodes per agent.

    Returns
    -------
    A dictionary with ``ppo`` and ``sac`` keys, each mapping to
    ``{"mean": float, "std": float}`` reward statistics.
    """
    ppo_model = train_ppo(env_factory(), total_timesteps=total_timesteps)
    sac_model = train_sac(env_factory(), total_timesteps=total_timesteps)

    ppo_mean, ppo_std = evaluate_agent(ppo_model, env_factory(), n_eval_episodes)
    sac_mean, sac_std = evaluate_agent(sac_model, env_factory(), n_eval_episodes)

    return {
        "ppo": {"mean": ppo_mean, "std": ppo_std},
        "sac": {"mean": sac_mean, "std": sac_std},
    }

lumina_lob.rl.evaluate

Evaluate heuristic and trained RL market-making policies.

Classes

EpisodeResult dataclass

Summary statistics for one episode.

Source code in lumina_lob/rl/evaluate.py

@dataclass
class EpisodeResult:
    """Summary statistics for one episode."""

    total_reward: float
    total_pnl: float
    final_inventory: float
    max_inventory: float
    min_inventory: float
    n_steps: int

SimpleMarketMakerPolicy

Deterministic heuristic that skews quotes based on inventory.

Long inventory widens the bid offset and tightens the ask offset to encourage selling; short inventory does the opposite. Quote sizes are always at the configured minimum.

Source code in lumina_lob/rl/evaluate.py

class SimpleMarketMakerPolicy:
    """Deterministic heuristic that skews quotes based on inventory.

    Long inventory widens the bid offset and tightens the ask offset to
    encourage selling; short inventory does the opposite.  Quote sizes are
    always at the configured minimum.
    """

    def __init__(
        self,
        base_offset_ticks: int = 2,
        max_inventory_skew: float = 1.0,
    ) -> None:
        self.base_offset_ticks = max(0, int(base_offset_ticks))
        self.max_inventory_skew = float(max_inventory_skew)

    def __call__(self, env: MarketMakerEnv) -> np.ndarray:
        """Return an action vector for the current env state."""
        base_env = getattr(env, "unwrapped", env)
        if not isinstance(base_env, MarketMakerEnv):
            base_env = env

        if base_env.simulation is None:
            shape = env.action_space.shape
            if shape is None:
                shape = (4,)
            return np.zeros(shape, dtype=np.float32)

        max_position = 1_000.0
        inventory_ratio = base_env._inventory / max_position

        bid_skew = self.base_offset_ticks + max(0.0, inventory_ratio * self.max_inventory_skew)
        ask_skew = self.base_offset_ticks + max(0.0, -inventory_ratio * self.max_inventory_skew)

        # Map offsets to [-1, 1] given the env's maximum quote offset.
        max_offset = max(1, base_env.max_quote_offset_ticks)
        bid_action = bid_skew / max_offset * 2.0 - 1.0
        ask_action = ask_skew / max_offset * 2.0 - 1.0

        bid_action = max(-1.0, min(1.0, bid_action))
        ask_action = max(-1.0, min(1.0, ask_action))

        return np.array(
            [bid_action, ask_action, -1.0, -1.0],
            dtype=np.float32,
        )

Methods:

__call__(env)

Return an action vector for the current env state.

Source code in lumina_lob/rl/evaluate.py

def __call__(self, env: MarketMakerEnv) -> np.ndarray:
    """Return an action vector for the current env state."""
    base_env = getattr(env, "unwrapped", env)
    if not isinstance(base_env, MarketMakerEnv):
        base_env = env

    if base_env.simulation is None:
        shape = env.action_space.shape
        if shape is None:
            shape = (4,)
        return np.zeros(shape, dtype=np.float32)

    max_position = 1_000.0
    inventory_ratio = base_env._inventory / max_position

    bid_skew = self.base_offset_ticks + max(0.0, inventory_ratio * self.max_inventory_skew)
    ask_skew = self.base_offset_ticks + max(0.0, -inventory_ratio * self.max_inventory_skew)

    # Map offsets to [-1, 1] given the env's maximum quote offset.
    max_offset = max(1, base_env.max_quote_offset_ticks)
    bid_action = bid_skew / max_offset * 2.0 - 1.0
    ask_action = ask_skew / max_offset * 2.0 - 1.0

    bid_action = max(-1.0, min(1.0, bid_action))
    ask_action = max(-1.0, min(1.0, ask_action))

    return np.array(
        [bid_action, ask_action, -1.0, -1.0],
        dtype=np.float32,
    )

Functions:

evaluate_heuristic_policy(env_factory, policy, n_episodes=5)

Run a heuristic policy for several episodes and return summaries.

Source code in lumina_lob/rl/evaluate.py

def evaluate_heuristic_policy(
    env_factory: Callable[[], gym.Env],
    policy: Callable[[gym.Env], np.ndarray],
    n_episodes: int = 5,
) -> list[EpisodeResult]:
    """Run a heuristic policy for several episodes and return summaries."""
    results: list[EpisodeResult] = []
    for _ in range(n_episodes):
        env = env_factory()
        base_env = _unwrap_env(env)
        obs, _ = env.reset()
        total_reward = 0.0
        max_inv = 0.0
        min_inv = 0.0
        truncated = False
        while not truncated:
            action = policy(env)
            obs, reward, terminated, truncated, _ = env.step(action)
            total_reward += reward
            max_inv = max(max_inv, base_env._inventory)
            min_inv = min(min_inv, base_env._inventory)

        total_pnl = base_env._cash + base_env._inventory * base_env._reference_price
        results.append(
            EpisodeResult(
                total_reward=total_reward,
                total_pnl=total_pnl,
                final_inventory=base_env._inventory,
                max_inventory=max_inv,
                min_inventory=min_inv,
                n_steps=base_env._current_step,
            )
        )
    return results

summarize_results(results)

Return mean and std of key metrics across episodes.

Source code in lumina_lob/rl/evaluate.py

def summarize_results(results: list[EpisodeResult]) -> dict[str, float]:
    """Return mean and std of key metrics across episodes."""
    if not results:
        return {}
    return {
        "mean_reward": float(np.mean([r.total_reward for r in results])),
        "mean_pnl": float(np.mean([r.total_pnl for r in results])),
        "mean_final_inventory": float(np.mean([r.final_inventory for r in results])),
        "max_abs_inventory": float(
            np.max([max(abs(r.max_inventory), abs(r.min_inventory)) for r in results])
        ),
    }