Skip to content

Market Model API

The lumina_lob.market_model package separates the fundamental reference price from market impact.

lumina_lob.market_model.reference_price

Reference price process for simulated markets.

Models the latent fair price as a discrete-time jump-diffusion:

log(S_{t+1}) = log(S_t) + (mu - 0.5*sigma^2)*dt + sigma*sqrt(dt)*Z + J

where Z ~ N(0, 1) and J is a compound Poisson jump with log-normal jump size. The price is floored at a small positive epsilon to avoid zero/negative quotes.

Classes

ReferencePriceProcess dataclass

Discrete-time jump-diffusion reference price simulator.

Parameters

initial_price: Starting reference price. Must be positive. drift: Instantaneous drift (per unit time). Default 0.0. volatility: Instantaneous volatility (per unit time). Must be non-negative. Default 0.2. dt: Time step size. Must be positive. Default 1.0. jump_intensity: Expected number of jumps per unit time. Must be non-negative. Default 0.0. jump_mean: Mean of the log jump size. Default 0.0. jump_std: Standard deviation of the log jump size. Must be non-negative. Default 0.05. seed: Optional RNG seed for reproducibility. min_price: Floor for the price path. Default 0.0001.

Source code in lumina_lob/market_model/reference_price.py 
 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
@dataclass
class ReferencePriceProcess:
    """Discrete-time jump-diffusion reference price simulator.

    Parameters
    ----------
    initial_price:
        Starting reference price. Must be positive.
    drift:
        Instantaneous drift (per unit time). Default 0.0.
    volatility:
        Instantaneous volatility (per unit time). Must be non-negative. Default 0.2.
    dt:
        Time step size. Must be positive. Default 1.0.
    jump_intensity:
        Expected number of jumps per unit time. Must be non-negative. Default 0.0.
    jump_mean:
        Mean of the log jump size. Default 0.0.
    jump_std:
        Standard deviation of the log jump size. Must be non-negative. Default 0.05.
    seed:
        Optional RNG seed for reproducibility.
    min_price:
        Floor for the price path. Default 0.0001.
    """

    initial_price: float = 100.0
    drift: float = 0.0
    volatility: float = 0.2
    dt: float = 1.0
    jump_intensity: float = 0.0
    jump_mean: float = 0.0
    jump_std: float = 0.05
    seed: int | None = None
    min_price: float = 0.0001

    _rng: np.random.Generator = field(init=False, repr=False)
    _path: list[float] = field(init=False, repr=False)
    _steps: int = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.initial_price <= 0:
            raise ValueError("initial_price must be positive")
        if self.volatility < 0:
            raise ValueError("volatility must be non-negative")
        if self.dt <= 0:
            raise ValueError("dt must be positive")
        if self.jump_intensity < 0:
            raise ValueError("jump_intensity must be non-negative")
        if self.jump_std < 0:
            raise ValueError("jump_std must be non-negative")
        if self.min_price <= 0:
            raise ValueError("min_price must be positive")

        self._rng = np.random.default_rng(self.seed)
        self._path = [float(self.initial_price)]
        self._steps = 0

    @property
    def price(self) -> float:
        """Current reference price."""
        return self._path[-1]

    @property
    def path(self) -> list[float]:
        """Full price history including the initial price."""
        return list(self._path)

    def _draw_jump(self, n: int) -> np.ndarray:
        """Draw compound Poisson jumps for n steps."""
        if self.jump_intensity == 0.0:
            return np.zeros(n, dtype=float)
        arrivals = self._rng.poisson(lam=self.jump_intensity * self.dt, size=n)
        if self.jump_std == 0.0:
            return self.jump_mean * arrivals
        log_jumps = self._rng.normal(
            loc=self.jump_mean * arrivals,
            scale=self.jump_std * np.sqrt(arrivals),
            size=n,
        )
        return log_jumps

    def step(self) -> float:
        """Advance one time step and return the new price."""
        prev = self._path[-1]
        diffusion = (self.drift - 0.5 * self.volatility**2) * self.dt
        diffusion += self.volatility * np.sqrt(self.dt) * self._rng.standard_normal()
        jump = self._draw_jump(1)[0]
        new_price = max(prev * np.exp(diffusion + jump), self.min_price)
        self._path.append(float(new_price))
        self._steps += 1
        return float(new_price)

    def simulate(self, n_steps: int) -> list[float]:
        """Simulate n_steps ahead and return the full path."""
        if n_steps <= 0:
            raise ValueError("n_steps must be positive")
        prev = self._path[-1]
        diffusion_drift = (self.drift - 0.5 * self.volatility**2) * self.dt
        shocks = self._rng.standard_normal(n_steps)
        jumps = self._draw_jump(n_steps)
        log_returns = diffusion_drift + self.volatility * np.sqrt(self.dt) * shocks + jumps
        cum_log_returns = np.cumsum(log_returns)
        prices = prev * np.exp(cum_log_returns)
        prices = np.maximum(prices, self.min_price)
        self._path.extend(float(p) for p in prices)
        self._steps += n_steps
        return self.path

    def reset(self, price: float | None = None) -> None:
        """Reset the process to the initial price or a new price."""
        start = self.initial_price if price is None else price
        if start <= 0:
            raise ValueError("reset price must be positive")
        self._path = [float(start)]
        self._steps = 0
Attributes
path property

Full price history including the initial price.

price property

Current reference price.

Methods:
reset(price=None)

Reset the process to the initial price or a new price.

Source code in lumina_lob/market_model/reference_price.py 
126
127
128
129
130
131
132
def reset(self, price: float | None = None) -> None:
    """Reset the process to the initial price or a new price."""
    start = self.initial_price if price is None else price
    if start <= 0:
        raise ValueError("reset price must be positive")
    self._path = [float(start)]
    self._steps = 0
simulate(n_steps)

Simulate n_steps ahead and return the full path.

Source code in lumina_lob/market_model/reference_price.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def simulate(self, n_steps: int) -> list[float]:
    """Simulate n_steps ahead and return the full path."""
    if n_steps <= 0:
        raise ValueError("n_steps must be positive")
    prev = self._path[-1]
    diffusion_drift = (self.drift - 0.5 * self.volatility**2) * self.dt
    shocks = self._rng.standard_normal(n_steps)
    jumps = self._draw_jump(n_steps)
    log_returns = diffusion_drift + self.volatility * np.sqrt(self.dt) * shocks + jumps
    cum_log_returns = np.cumsum(log_returns)
    prices = prev * np.exp(cum_log_returns)
    prices = np.maximum(prices, self.min_price)
    self._path.extend(float(p) for p in prices)
    self._steps += n_steps
    return self.path
step()

Advance one time step and return the new price.

Source code in lumina_lob/market_model/reference_price.py 
 99
100
101
102
103
104
105
106
107
108
def step(self) -> float:
    """Advance one time step and return the new price."""
    prev = self._path[-1]
    diffusion = (self.drift - 0.5 * self.volatility**2) * self.dt
    diffusion += self.volatility * np.sqrt(self.dt) * self._rng.standard_normal()
    jump = self._draw_jump(1)[0]
    new_price = max(prev * np.exp(diffusion + jump), self.min_price)
    self._path.append(float(new_price))
    self._steps += 1
    return float(new_price)

lumina_lob.market_model.impact

Market impact models: propagator and Almgren-Chriss style impact.

Classes

AlmgrenChrissImpact dataclass

Almgren-Chriss permanent and temporary market impact model.

Parameters

gamma: Permanent impact coefficient. Default 0.0. eta: Temporary impact coefficient. Default 0.0. sigma: Volatility of the asset. Default 0.0. dt: Time step. Default 1.0.

Source code in lumina_lob/market_model/impact.py 
 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
@dataclass
class AlmgrenChrissImpact:
    """Almgren-Chriss permanent and temporary market impact model.

    Parameters
    ----------
    gamma:
        Permanent impact coefficient. Default 0.0.
    eta:
        Temporary impact coefficient. Default 0.0.
    sigma:
        Volatility of the asset. Default 0.0.
    dt:
        Time step. Default 1.0.
    """

    gamma: float = 0.0
    eta: float = 0.0
    sigma: float = 0.0
    dt: float = 1.0

    _permanent_drift: float = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.gamma < 0:
            raise ValueError("gamma must be non-negative")
        if self.eta < 0:
            raise ValueError("eta must be non-negative")
        if self.sigma < 0:
            raise ValueError("sigma must be non-negative")
        if self.dt <= 0:
            raise ValueError("dt must be positive")
        self._permanent_drift = 0.0

    def apply(self, signed_volume: float, reference_price: float) -> float:
        """Return reference price adjusted by permanent and temporary impact."""
        permanent = self.gamma * signed_volume
        temporary = self.eta * signed_volume / self.dt
        self._permanent_drift += permanent
        return reference_price + permanent + temporary

    def drift(self) -> float:
        """Cumulative permanent price displacement."""
        return self._permanent_drift

    def reset(self) -> None:
        """Clear cumulative permanent drift."""
        self._permanent_drift = 0.0
Methods:
apply(signed_volume, reference_price)

Return reference price adjusted by permanent and temporary impact.

Source code in lumina_lob/market_model/impact.py 
 95
 96
 97
 98
 99
100
def apply(self, signed_volume: float, reference_price: float) -> float:
    """Return reference price adjusted by permanent and temporary impact."""
    permanent = self.gamma * signed_volume
    temporary = self.eta * signed_volume / self.dt
    self._permanent_drift += permanent
    return reference_price + permanent + temporary
drift()

Cumulative permanent price displacement.

Source code in lumina_lob/market_model/impact.py 
102
103
104
def drift(self) -> float:
    """Cumulative permanent price displacement."""
    return self._permanent_drift
reset()

Clear cumulative permanent drift.

Source code in lumina_lob/market_model/impact.py 
106
107
108
def reset(self) -> None:
    """Clear cumulative permanent drift."""
    self._permanent_drift = 0.0

PropagatorImpact dataclass

Propagator-style temporary and permanent market impact.

Trade at time t moves the reference price by a permanent component plus a temporary component that decays geometrically over subsequent steps.

Parameters

permanent_impact: Permanent impact coefficient per unit volume. Default 0.0. temporary_impact: Immediate temporary impact coefficient per unit volume. Default 0.0. decay: Decay factor for temporary impact in (0, 1]. Default 0.5.

Source code in lumina_lob/market_model/impact.py 
 7
 8
 9
10
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
@dataclass
class PropagatorImpact:
    """Propagator-style temporary and permanent market impact.

    Trade at time t moves the reference price by a permanent component plus a
    temporary component that decays geometrically over subsequent steps.

    Parameters
    ----------
    permanent_impact:
        Permanent impact coefficient per unit volume. Default 0.0.
    temporary_impact:
        Immediate temporary impact coefficient per unit volume. Default 0.0.
    decay:
        Decay factor for temporary impact in (0, 1]. Default 0.5.
    """

    permanent_impact: float = 0.0
    temporary_impact: float = 0.0
    decay: float = 0.5

    _residual: float = field(init=False, repr=False)

    def __post_init__(self) -> None:
        if self.permanent_impact < 0:
            raise ValueError("permanent_impact must be non-negative")
        if self.temporary_impact < 0:
            raise ValueError("temporary_impact must be non-negative")
        if not 0.0 < self.decay <= 1.0:
            raise ValueError("decay must be in (0, 1]")
        self._residual = 0.0

    def apply(self, signed_volume: float, reference_price: float) -> float:
        """Return the total price impact for a trade at the current step.

        The impact is expressed as a signed price displacement. Positive signed
        volume (buying) pushes the price up; negative volume (selling) pushes it
        down.
        """
        permanent = self.permanent_impact * signed_volume
        temporary = self.temporary_impact * signed_volume
        total = permanent + temporary + self._residual
        self._residual = (temporary + self._residual) * self.decay
        return reference_price + total

    def step(self) -> None:
        """Decay residual temporary impact one time step."""
        self._residual *= self.decay

    def reset(self) -> None:
        """Clear residual temporary impact."""
        self._residual = 0.0
Methods:
apply(signed_volume, reference_price)

Return the total price impact for a trade at the current step.

The impact is expressed as a signed price displacement. Positive signed volume (buying) pushes the price up; negative volume (selling) pushes it down.

Source code in lumina_lob/market_model/impact.py 
39
40
41
42
43
44
45
46
47
48
49
50
def apply(self, signed_volume: float, reference_price: float) -> float:
    """Return the total price impact for a trade at the current step.

    The impact is expressed as a signed price displacement. Positive signed
    volume (buying) pushes the price up; negative volume (selling) pushes it
    down.
    """
    permanent = self.permanent_impact * signed_volume
    temporary = self.temporary_impact * signed_volume
    total = permanent + temporary + self._residual
    self._residual = (temporary + self._residual) * self.decay
    return reference_price + total
reset()

Clear residual temporary impact.

Source code in lumina_lob/market_model/impact.py 
56
57
58
def reset(self) -> None:
    """Clear residual temporary impact."""
    self._residual = 0.0
step()

Decay residual temporary impact one time step.

Source code in lumina_lob/market_model/impact.py 
52
53
54
def step(self) -> None:
    """Decay residual temporary impact one time step."""
    self._residual *= self.decay