Skip to content

Agents API

The lumina_lob.agents package contains the built-in trading agents and the base class for custom agents.

lumina_lob.agents.base

Abstract base class for market agents.

Classes

Agent

Bases: ABC

Base class for all market agents.

An agent observes the current reference price and order book state, then emits zero or more orders to be processed by the matching engine.

Source code in lumina_lob/agents/base.py
class Agent(ABC):
    """Base class for all market agents.

    An agent observes the current reference price and order book state, then
    emits zero or more orders to be processed by the matching engine.
    """

    @abstractmethod
    def act(self, reference_price: float, book: OrderBook) -> list[Order]:
        """Return a list of orders to submit at this simulation step."""
        ...
Methods:
act(reference_price, book) abstractmethod

Return a list of orders to submit at this simulation step.

Source code in lumina_lob/agents/base.py
@abstractmethod
def act(self, reference_price: float, book: OrderBook) -> list[Order]:
    """Return a list of orders to submit at this simulation step."""
    ...

lumina_lob.agents.noise_trader

Noise trader: random Poisson arrivals with randomized size and side.

Classes

NoiseTrader dataclass

Bases: Agent

Liquidity-demanding agent that submits random limit orders.

Parameters

arrival_rate: Expected number of orders emitted per act() call (Poisson). size_dist: Distribution for order quantity: uniform or lognormal. size_min: Minimum order quantity for uniform distribution. Default 1. size_max: Maximum order quantity for uniform distribution. Default 10. size_mu: Mean of log quantity for log-normal distribution. Default 1.0. size_sigma: Standard deviation of log quantity for log-normal distribution. Default 0.5. side_bias: Probability of generating a bid (0.5 = neutral). Default 0.5. price_offset_max: Maximum number of ticks away from the rounded reference price. Default 5. tick_size: Price tick size. Default 1.0. seed: Optional RNG seed for reproducibility.

Source code in lumina_lob/agents/noise_trader.py
@dataclass
class NoiseTrader(Agent):
    """Liquidity-demanding agent that submits random limit orders.

    Parameters
    ----------
    arrival_rate:
        Expected number of orders emitted per `act()` call (Poisson).
    size_dist:
        Distribution for order quantity: ``uniform`` or ``lognormal``.
    size_min:
        Minimum order quantity for uniform distribution. Default 1.
    size_max:
        Maximum order quantity for uniform distribution. Default 10.
    size_mu:
        Mean of log quantity for log-normal distribution. Default 1.0.
    size_sigma:
        Standard deviation of log quantity for log-normal distribution. Default 0.5.
    side_bias:
        Probability of generating a bid (0.5 = neutral). Default 0.5.
    price_offset_max:
        Maximum number of ticks away from the rounded reference price. Default 5.
    tick_size:
        Price tick size. Default 1.0.
    seed:
        Optional RNG seed for reproducibility.
    """

    arrival_rate: float = 1.0
    size_dist: Literal["uniform", "lognormal"] = "uniform"
    size_min: int = 1
    size_max: int = 10
    size_mu: float = 1.0
    size_sigma: float = 0.5
    side_bias: float = 0.5
    price_offset_max: int = 5
    tick_size: float = 1.0
    seed: int | None = None

    _rng: np.random.Generator = field(init=False, repr=False)
    _next_order_id: int = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.arrival_rate < 0:
            raise ValueError("arrival_rate must be non-negative")
        if self.size_min <= 0:
            raise ValueError("size_min must be positive")
        if self.size_max < self.size_min:
            raise ValueError("size_max must be >= size_min")
        if self.size_sigma < 0:
            raise ValueError("size_sigma must be non-negative")
        if not 0.0 <= self.side_bias <= 1.0:
            raise ValueError("side_bias must be in [0, 1]")
        if self.price_offset_max < 0:
            raise ValueError("price_offset_max must be non-negative")
        if self.tick_size <= 0:
            raise ValueError("tick_size must be positive")

        self._rng = np.random.default_rng(self.seed)
        self._next_order_id = 1

    def act(self, reference_price: float, book: OrderBook) -> list[Order]:
        """Generate a batch of random limit orders."""
        n_orders = self._rng.poisson(lam=self.arrival_rate)
        orders: list[Order] = []
        for _ in range(n_orders):
            side = Side.BID if self._rng.random() < self.side_bias else Side.ASK
            qty = self._draw_qty()
            price = self._draw_price(reference_price, side)
            orders.append(
                Order(
                    order_id=self._next_order_id,
                    side=side,
                    price=price,
                    qty=qty,
                )
            )
            self._next_order_id += 1
        return orders

    def _draw_qty(self) -> int:
        if self.size_dist == "uniform":
            return int(self._rng.integers(self.size_min, self.size_max + 1))
        # lognormal
        return max(self.size_min, int(self._rng.lognormal(self.size_mu, self.size_sigma)))

    def _draw_price(self, reference_price: float, side: Side) -> int:
        ticks_away = int(self._rng.integers(0, self.price_offset_max + 1))
        mid_tick = round(reference_price / self.tick_size)
        if side == Side.BID:
            tick = mid_tick - ticks_away
        else:
            tick = mid_tick + ticks_away
        price = tick * self.tick_size
        return max(1, int(round(price)))
Methods:
act(reference_price, book)

Generate a batch of random limit orders.

Source code in lumina_lob/agents/noise_trader.py
def act(self, reference_price: float, book: OrderBook) -> list[Order]:
    """Generate a batch of random limit orders."""
    n_orders = self._rng.poisson(lam=self.arrival_rate)
    orders: list[Order] = []
    for _ in range(n_orders):
        side = Side.BID if self._rng.random() < self.side_bias else Side.ASK
        qty = self._draw_qty()
        price = self._draw_price(reference_price, side)
        orders.append(
            Order(
                order_id=self._next_order_id,
                side=side,
                price=price,
                qty=qty,
            )
        )
        self._next_order_id += 1
    return orders

lumina_lob.agents.informed_trader

Informed trader: directional signal with temporary/permanent impact tracking.

Classes

InformedTrader dataclass

Bases: Agent

Trader that trades in the direction of a private signal.

The informed trader submits aggressive market orders or large crossing limit orders to move the fair value. It tracks its own traded volume so downstream impact models can estimate temporary and permanent price impact.

Parameters

signal: Direction of private information: bullish or bearish. trade_size: Base quantity per order. Must be positive. participation_rate: Probability of submitting an order on any given act() call. Must be in [0, 1]. order_type: market for aggressive fills, limit for large crossing orders. price_offset: For limit orders, how many ticks past the best price to cross. Default 1. tick_size: Minimum price increment. Default 1.0. seed: Optional RNG seed.

Source code in lumina_lob/agents/informed_trader.py
@dataclass
class InformedTrader(Agent):
    """Trader that trades in the direction of a private signal.

    The informed trader submits aggressive market orders or large crossing
    limit orders to move the fair value. It tracks its own traded volume so
    downstream impact models can estimate temporary and permanent price impact.

    Parameters
    ----------
    signal:
        Direction of private information: ``bullish`` or ``bearish``.
    trade_size:
        Base quantity per order. Must be positive.
    participation_rate:
        Probability of submitting an order on any given ``act()`` call.
        Must be in [0, 1].
    order_type:
        ``market`` for aggressive fills, ``limit`` for large crossing orders.
    price_offset:
        For limit orders, how many ticks past the best price to cross. Default 1.
    tick_size:
        Minimum price increment. Default 1.0.
    seed:
        Optional RNG seed.
    """

    signal: Literal["bullish", "bearish"] = "bullish"
    trade_size: int = 100
    participation_rate: float = 0.5
    order_type: Literal["market", "limit"] = "market"
    price_offset: int = 1
    tick_size: float = 1.0
    seed: int | None = None

    _rng: np.random.Generator = field(init=False, repr=False)
    _next_order_id: int = field(init=False, repr=False)
    _total_traded: int = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.signal not in ("bullish", "bearish"):
            raise ValueError("signal must be 'bullish' or 'bearish'")
        if self.trade_size <= 0:
            raise ValueError("trade_size must be positive")
        if not 0.0 <= self.participation_rate <= 1.0:
            raise ValueError("participation_rate must be in [0, 1]")
        if self.price_offset < 0:
            raise ValueError("price_offset must be non-negative")
        if self.tick_size <= 0:
            raise ValueError("tick_size must be positive")

        self._rng = np.random.default_rng(self.seed)
        self._next_order_id = 1
        self._total_traded = 0

    @property
    def side(self) -> Side:
        """Trading side implied by the signal."""
        return Side.BID if self.signal == "bullish" else Side.ASK

    @property
    def total_traded(self) -> int:
        """Cumulative quantity traded by this agent."""
        return self._total_traded

    def act(self, reference_price: float, book: OrderBook) -> list[Order]:
        """Generate an order if the trader participates this step."""
        if self._rng.random() >= self.participation_rate:
            return []

        side = self.side
        qty = self.trade_size

        if self.order_type == "market":
            order = Order(
                order_id=self._next_order_id,
                side=side,
                price=None,
                qty=qty,
                order_type=OrderType.MARKET,
            )
        else:
            price = self._crossing_price(reference_price, side, book)
            order = Order(
                order_id=self._next_order_id,
                side=side,
                price=price,
                qty=qty,
                order_type=OrderType.LIMIT,
            )

        self._next_order_id += 1
        self._total_traded += qty
        return [order]

    def _crossing_price(self, reference_price: float, side: Side, book: OrderBook) -> int:
        """Price that guarantees immediate execution in the signal direction."""
        if side == Side.BID:
            best_ask = book.best_ask
            if best_ask is not None:
                tick = round((best_ask + self.price_offset * self.tick_size) / self.tick_size)
            else:
                tick = round(reference_price / self.tick_size) + self.price_offset
        else:
            best_bid = book.best_bid
            if best_bid is not None:
                tick = round((best_bid - self.price_offset * self.tick_size) / self.tick_size)
            else:
                tick = round(reference_price / self.tick_size) - self.price_offset
        return max(1, int(round(tick * self.tick_size)))
Attributes
side property

Trading side implied by the signal.

total_traded property

Cumulative quantity traded by this agent.

Methods:
act(reference_price, book)

Generate an order if the trader participates this step.

Source code in lumina_lob/agents/informed_trader.py
def act(self, reference_price: float, book: OrderBook) -> list[Order]:
    """Generate an order if the trader participates this step."""
    if self._rng.random() >= self.participation_rate:
        return []

    side = self.side
    qty = self.trade_size

    if self.order_type == "market":
        order = Order(
            order_id=self._next_order_id,
            side=side,
            price=None,
            qty=qty,
            order_type=OrderType.MARKET,
        )
    else:
        price = self._crossing_price(reference_price, side, book)
        order = Order(
            order_id=self._next_order_id,
            side=side,
            price=price,
            qty=qty,
            order_type=OrderType.LIMIT,
        )

    self._next_order_id += 1
    self._total_traded += qty
    return [order]

lumina_lob.agents.market_maker

Market maker: symmetric quotes around reference price with inventory limits.

Classes

MarketMaker dataclass

Bases: Agent

Simple market maker that quotes symmetrically around the reference price.

The agent maintains a target half-spread and a maximum inventory (long/short). If inventory reaches the limit on one side, it stops quoting that side until the position comes back within bounds.

Parameters

spread_half_width: Half-spread in price ticks. Must be positive. quote_size: Quantity to quote on each side. Must be positive. max_inventory: Maximum absolute inventory allowed. Must be non-negative. tick_size: Minimum price increment. Default 1.0.

Source code in lumina_lob/agents/market_maker.py
@dataclass
class MarketMaker(Agent):
    """Simple market maker that quotes symmetrically around the reference price.

    The agent maintains a target half-spread and a maximum inventory (long/short).
    If inventory reaches the limit on one side, it stops quoting that side until
    the position comes back within bounds.

    Parameters
    ----------
    spread_half_width:
        Half-spread in price ticks. Must be positive.
    quote_size:
        Quantity to quote on each side. Must be positive.
    max_inventory:
        Maximum absolute inventory allowed. Must be non-negative.
    tick_size:
        Minimum price increment. Default 1.0.
    """

    spread_half_width: float = 2.0
    quote_size: int = 10
    max_inventory: int = 100
    tick_size: float = 1.0

    _inventory: int = field(init=False, repr=False)
    _next_order_id: int = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.spread_half_width <= 0:
            raise ValueError("spread_half_width must be positive")
        if self.quote_size <= 0:
            raise ValueError("quote_size must be positive")
        if self.max_inventory < 0:
            raise ValueError("max_inventory must be non-negative")
        if self.tick_size <= 0:
            raise ValueError("tick_size must be positive")

        self._inventory = 0
        self._next_order_id = 1

    @property
    def inventory(self) -> int:
        """Current signed inventory (positive = long, negative = short)."""
        return self._inventory

    def act(self, reference_price: float, book: OrderBook) -> list[Order]:
        """Submit bid/ask quotes around reference price, respecting inventory limits."""
        bid_price, ask_price = self._quote_prices(reference_price)
        orders: list[Order] = []

        if self._can_quote_bid():
            orders.append(
                Order(
                    order_id=self._next_order_id,
                    side=Side.BID,
                    price=bid_price,
                    qty=self.quote_size,
                )
            )
            self._next_order_id += 1

        if self._can_quote_ask():
            orders.append(
                Order(
                    order_id=self._next_order_id,
                    side=Side.ASK,
                    price=ask_price,
                    qty=self.quote_size,
                )
            )
            self._next_order_id += 1

        return orders

    def _quote_prices(self, reference_price: float) -> tuple[float, float]:
        """Compute bid and ask quote prices rounded to ticks."""
        half_spread = self.spread_half_width * self.tick_size
        bid_tick = round((reference_price - half_spread) / self.tick_size)
        ask_tick = round((reference_price + half_spread) / self.tick_size)
        # Ensure positive ask tick and non-overlapping spread
        bid_tick = max(1, bid_tick)
        ask_tick = max(bid_tick + 1, ask_tick)
        return float(bid_tick * self.tick_size), float(ask_tick * self.tick_size)

    def _can_quote_bid(self) -> bool:
        """Can quote a bid if not at the maximum long position."""
        return self._inventory < self.max_inventory

    def _can_quote_ask(self) -> bool:
        """Can quote an ask if not at the maximum short position."""
        return self._inventory > -self.max_inventory

    def on_fill(self, side: Side, qty: int) -> None:
        """Update inventory when one of the market maker's orders is filled.

        Call this from the simulation loop after processing the agent's orders.
        """
        if side == Side.BID:
            # Bought qty -> inventory increases
            self._inventory += qty
        elif side == Side.ASK:
            # Sold qty -> inventory decreases
            self._inventory -= qty

    def reset_inventory(self, value: int = 0) -> None:
        """Reset inventory to a target value."""
        self._inventory = value
Attributes
inventory property

Current signed inventory (positive = long, negative = short).

Methods:
act(reference_price, book)

Submit bid/ask quotes around reference price, respecting inventory limits.

Source code in lumina_lob/agents/market_maker.py
def act(self, reference_price: float, book: OrderBook) -> list[Order]:
    """Submit bid/ask quotes around reference price, respecting inventory limits."""
    bid_price, ask_price = self._quote_prices(reference_price)
    orders: list[Order] = []

    if self._can_quote_bid():
        orders.append(
            Order(
                order_id=self._next_order_id,
                side=Side.BID,
                price=bid_price,
                qty=self.quote_size,
            )
        )
        self._next_order_id += 1

    if self._can_quote_ask():
        orders.append(
            Order(
                order_id=self._next_order_id,
                side=Side.ASK,
                price=ask_price,
                qty=self.quote_size,
            )
        )
        self._next_order_id += 1

    return orders
on_fill(side, qty)

Update inventory when one of the market maker's orders is filled.

Call this from the simulation loop after processing the agent's orders.

Source code in lumina_lob/agents/market_maker.py
def on_fill(self, side: Side, qty: int) -> None:
    """Update inventory when one of the market maker's orders is filled.

    Call this from the simulation loop after processing the agent's orders.
    """
    if side == Side.BID:
        # Bought qty -> inventory increases
        self._inventory += qty
    elif side == Side.ASK:
        # Sold qty -> inventory decreases
        self._inventory -= qty
reset_inventory(value=0)

Reset inventory to a target value.

Source code in lumina_lob/agents/market_maker.py
def reset_inventory(self, value: int = 0) -> None:
    """Reset inventory to a target value."""
    self._inventory = value

lumina_lob.agents.skewed_market_maker

Skewed market maker: inventory-sensitive quoting.

Classes

SkewedMarketMaker dataclass

Bases: Agent

Market maker that skews quotes based on signed inventory.

As inventory grows long, the market maker lowers bids and offers to attract sells. As inventory grows short, it raises bids and offers to attract buys. The skew is linear in the signed inventory ratio.

Parameters

base_half_spread: Base half-spread in ticks before skew. Must be positive. quote_size: Quantity to quote on each side. Must be positive. max_inventory: Maximum absolute inventory allowed. Must be non-negative. skew_factor: How aggressively to skew quotes per unit of inventory ratio. inventory_ratio = inventory / max_inventory. Default 2.0. tick_size: Minimum price increment. Default 1.0.

Source code in lumina_lob/agents/skewed_market_maker.py
@dataclass
class SkewedMarketMaker(Agent):
    """Market maker that skews quotes based on signed inventory.

    As inventory grows long, the market maker lowers bids and offers to attract
    sells. As inventory grows short, it raises bids and offers to attract buys.
    The skew is linear in the signed inventory ratio.

    Parameters
    ----------
    base_half_spread:
        Base half-spread in ticks before skew. Must be positive.
    quote_size:
        Quantity to quote on each side. Must be positive.
    max_inventory:
        Maximum absolute inventory allowed. Must be non-negative.
    skew_factor:
        How aggressively to skew quotes per unit of inventory ratio.
        inventory_ratio = inventory / max_inventory. Default 2.0.
    tick_size:
        Minimum price increment. Default 1.0.
    """

    base_half_spread: float = 2.0
    quote_size: int = 10
    max_inventory: int = 100
    skew_factor: float = 2.0
    tick_size: float = 1.0

    _inventory: int = field(init=False, repr=False)
    _next_order_id: int = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.base_half_spread <= 0:
            raise ValueError("base_half_spread must be positive")
        if self.quote_size <= 0:
            raise ValueError("quote_size must be positive")
        if self.max_inventory < 0:
            raise ValueError("max_inventory must be non-negative")
        if self.skew_factor < 0:
            raise ValueError("skew_factor must be non-negative")
        if self.tick_size <= 0:
            raise ValueError("tick_size must be positive")

        self._inventory = 0
        self._next_order_id = 1

    @property
    def inventory(self) -> int:
        """Current signed inventory (positive = long, negative = short)."""
        return self._inventory

    def act(self, reference_price: float, book: OrderBook) -> list[Order]:
        """Submit inventory-skewed bid/ask quotes."""
        bid_price, ask_price = self._quote_prices(reference_price)
        orders: list[Order] = []

        if self._can_quote_bid():
            orders.append(
                Order(
                    order_id=self._next_order_id,
                    side=Side.BID,
                    price=bid_price,
                    qty=self.quote_size,
                )
            )
            self._next_order_id += 1

        if self._can_quote_ask():
            orders.append(
                Order(
                    order_id=self._next_order_id,
                    side=Side.ASK,
                    price=ask_price,
                    qty=self.quote_size,
                )
            )
            self._next_order_id += 1

        return orders

    def _quote_prices(self, reference_price: float) -> tuple[float, float]:
        """Compute bid and ask quote prices with inventory skew."""
        if self.max_inventory == 0:
            ratio = 0.0
        else:
            ratio = self._inventory / self.max_inventory

        # Inventory skew:
        #   long (ratio > 0) -> raise bid and lower ask to encourage selling
        #   short (ratio < 0) -> lower bid and raise ask to encourage buying
        skew_ticks = self.skew_factor * ratio

        bid_half = self.base_half_spread + skew_ticks
        ask_half = self.base_half_spread - skew_ticks

        # Keep a minimum one-tick spread on each side of mid
        min_half = 0.5
        bid_half = max(min_half, bid_half)
        ask_half = max(min_half, ask_half)

        bid_tick = round((reference_price - bid_half * self.tick_size) / self.tick_size)
        ask_tick = round((reference_price + ask_half * self.tick_size) / self.tick_size)

        bid_tick = max(1, bid_tick)
        ask_tick = max(bid_tick + 1, ask_tick)
        return float(bid_tick * self.tick_size), float(ask_tick * self.tick_size)

    def _can_quote_bid(self) -> bool:
        """Can quote a bid if not at the maximum long position."""
        return self._inventory < self.max_inventory

    def _can_quote_ask(self) -> bool:
        """Can quote an ask if not at the maximum short position."""
        return self._inventory > -self.max_inventory

    def on_fill(self, side: Side, qty: int) -> None:
        """Update inventory when one of the market maker's orders is filled."""
        if side == Side.BID:
            self._inventory += qty
        elif side == Side.ASK:
            self._inventory -= qty

    def reset_inventory(self, value: int = 0) -> None:
        """Reset inventory to a target value."""
        self._inventory = value
Attributes
inventory property

Current signed inventory (positive = long, negative = short).

Methods:
act(reference_price, book)

Submit inventory-skewed bid/ask quotes.

Source code in lumina_lob/agents/skewed_market_maker.py
def act(self, reference_price: float, book: OrderBook) -> list[Order]:
    """Submit inventory-skewed bid/ask quotes."""
    bid_price, ask_price = self._quote_prices(reference_price)
    orders: list[Order] = []

    if self._can_quote_bid():
        orders.append(
            Order(
                order_id=self._next_order_id,
                side=Side.BID,
                price=bid_price,
                qty=self.quote_size,
            )
        )
        self._next_order_id += 1

    if self._can_quote_ask():
        orders.append(
            Order(
                order_id=self._next_order_id,
                side=Side.ASK,
                price=ask_price,
                qty=self.quote_size,
            )
        )
        self._next_order_id += 1

    return orders
on_fill(side, qty)

Update inventory when one of the market maker's orders is filled.

Source code in lumina_lob/agents/skewed_market_maker.py
def on_fill(self, side: Side, qty: int) -> None:
    """Update inventory when one of the market maker's orders is filled."""
    if side == Side.BID:
        self._inventory += qty
    elif side == Side.ASK:
        self._inventory -= qty
reset_inventory(value=0)

Reset inventory to a target value.

Source code in lumina_lob/agents/skewed_market_maker.py
def reset_inventory(self, value: int = 0) -> None:
    """Reset inventory to a target value."""
    self._inventory = value