nautilus_trading/examples/strategies/composite_market_maker/
strategy.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Composite market making strategy implementation.
17
18use std::fmt::Debug;
19
20use ahash::AHashSet;
21use nautilus_common::actor::DataActor;
22use nautilus_core::DurationNanos;
23use nautilus_model::{
24    data::QuoteTick,
25    enums::{OrderSide, TimeInForce},
26    events::{OrderCanceled, OrderExpired, OrderFilled, OrderRejected},
27    identifiers::{ClientOrderId, StrategyId},
28    instruments::{Instrument, InstrumentAny},
29    orders::Order,
30    types::{Price, Quantity},
31};
32use rust_decimal::Decimal;
33
34use super::config::CompositeMarketMakerConfig;
35use crate::{
36    nautilus_strategy,
37    strategy::{Strategy, StrategyCore},
38};
39
40/// Composite market making strategy with book-mid quoting and signal-driven skew.
41///
42/// Quotes a single bid and a single ask around the target instrument's book mid.
43/// A second instrument (typically a `SyntheticInstrument`) supplies a signal
44/// whose residual against a baseline shifts both sides up or down. Inventory
45/// skew shifts both sides in the opposite direction of the current position.
46/// With nonzero signal skew, quoting waits for a valid signal quote and baseline.
47/// Orders persist across ticks and are only replaced when either the anchor
48/// or the signal residual's price impact (`signal_skew_factor * residual`)
49/// moves by at least `requote_threshold_bps` of the anchor.
50pub struct CompositeMarketMaker {
51    pub(super) core: StrategyCore,
52    pub(super) config: CompositeMarketMakerConfig,
53    pub(super) instrument: Option<InstrumentAny>,
54    pub(super) trade_size: Option<Quantity>,
55    pub(super) price_precision: Option<u8>,
56    pub(super) last_quoted_anchor: Option<Price>,
57    pub(super) last_quoted_residual: Option<f64>,
58    pub(super) signal_baseline: Option<f64>,
59    pub(super) last_signal: Option<f64>,
60    pub(super) pending_self_cancels: AHashSet<ClientOrderId>,
61}
62
63impl CompositeMarketMaker {
64    /// Creates a new [`CompositeMarketMaker`] instance from config.
65    #[must_use]
66    pub fn new(config: CompositeMarketMakerConfig) -> Self {
67        let signal_baseline = config.signal_baseline;
68        Self {
69            core: StrategyCore::new(config.base.clone()),
70            instrument: None,
71            trade_size: config.trade_size,
72            config,
73            price_precision: None,
74            last_quoted_anchor: None,
75            last_quoted_residual: None,
76            signal_baseline,
77            last_signal: None,
78            pending_self_cancels: AHashSet::new(),
79        }
80    }
81
82    pub(super) fn should_requote_on_anchor(&self, anchor: Price) -> bool {
83        match self.last_quoted_anchor {
84            Some(last_anchor) => {
85                let last_f64 = last_anchor.as_f64();
86                if last_f64 == 0.0 {
87                    return true;
88                }
89                let threshold = self.config.requote_threshold_bps as f64 / 10_000.0;
90                (anchor.as_f64() - last_f64).abs() / last_f64 >= threshold
91            }
92            None => true,
93        }
94    }
95
96    pub(super) fn should_requote_on_residual(&self, residual: f64, anchor: Price) -> bool {
97        if self.config.signal_skew_factor == 0.0 {
98            return false;
99        }
100        let anchor_f64 = anchor.as_f64();
101        if anchor_f64 == 0.0 {
102            return false;
103        }
104
105        match self.last_quoted_residual {
106            Some(last) => {
107                // The signal's contribution to bid/ask price is signal_skew_factor * residual,
108                // so a residual delta translates to that price units. Compare against the same
109                // bps threshold expressed in price units of the anchor.
110                let price_delta = (residual - last).abs() * self.config.signal_skew_factor.abs();
111                let threshold = self.config.requote_threshold_bps as f64 / 10_000.0;
112                price_delta / anchor_f64 >= threshold
113            }
114            None => true,
115        }
116    }
117
118    pub(super) fn should_requote(&self, anchor: Price, residual: f64) -> bool {
119        self.should_requote_on_anchor(anchor) || self.should_requote_on_residual(residual, anchor)
120    }
121
122    pub(super) fn signal_residual(&self) -> Option<f64> {
123        if self.config.signal_skew_factor == 0.0 {
124            return Some(0.0);
125        }
126
127        let signal = self.last_signal?;
128        let baseline = self.signal_baseline?;
129
130        if !signal.is_finite() || signal <= 0.0 || !baseline.is_finite() || baseline <= 0.0 {
131            return None;
132        }
133
134        let residual = signal / baseline - 1.0;
135        residual.is_finite().then_some(residual)
136    }
137
138    pub(super) fn compute_quotes(
139        &self,
140        anchor: Price,
141        signal_residual: f64,
142        net_position: f64,
143        worst_long: Decimal,
144        worst_short: Decimal,
145    ) -> anyhow::Result<Vec<(OrderSide, Price)>> {
146        let Some(instrument) = self.instrument.as_ref() else {
147            anyhow::bail!("Cannot compute quotes: instrument is not resolved");
148        };
149        let Some(trade_size) = self.trade_size else {
150            anyhow::bail!("Cannot compute quotes: trade_size is not resolved");
151        };
152        let trade_size = trade_size.as_decimal();
153        let max_pos = self.config.max_position.as_decimal();
154
155        let anchor_f64 = anchor.as_f64();
156        let half_spread = anchor_f64 * (self.config.half_spread_bps as f64 / 10_000.0);
157        let inventory_shift = self.config.inventory_skew_factor * net_position;
158        let signal_shift = self.config.signal_skew_factor * signal_residual;
159        // Positive signal residual lifts both sides; positive inventory lowers both sides.
160        let total_shift = signal_shift - inventory_shift;
161
162        let bid_f64 = anchor_f64 - half_spread + total_shift;
163        let ask_f64 = anchor_f64 + half_spread + total_shift;
164        // next_bid_price floors to the nearest valid bid tick (<=bid_f64),
165        // next_ask_price ceils to the nearest valid ask tick (>=ask_f64),
166        // so a non-crossing pair stays non-crossing after rounding.
167        let bid_price = instrument.next_bid_price(bid_f64, 0);
168        let ask_price = instrument.next_ask_price(ask_f64, 0);
169
170        // Drop both sides if rounding has collapsed or crossed the spread,
171        // which can happen when skew exceeds the half-spread on coarse-tick instruments.
172        let crossed = match (bid_price, ask_price) {
173            (Some(bp), Some(ap)) => bp >= ap,
174            _ => false,
175        };
176
177        if crossed {
178            return Ok(Vec::new());
179        }
180
181        let mut orders = Vec::new();
182
183        if let Some(price) = bid_price
184            && worst_long + trade_size <= max_pos
185        {
186            orders.push((OrderSide::Buy, price));
187        }
188
189        if let Some(price) = ask_price
190            && worst_short - trade_size >= -max_pos
191        {
192            orders.push((OrderSide::Sell, price));
193        }
194
195        Ok(orders)
196    }
197}
198
199nautilus_strategy!(CompositeMarketMaker, {
200    fn on_order_rejected(&mut self, event: OrderRejected) {
201        self.pending_self_cancels.remove(&event.client_order_id);
202        self.last_quoted_anchor = None;
203        self.last_quoted_residual = None;
204    }
205
206    fn on_order_expired(&mut self, event: OrderExpired) {
207        self.pending_self_cancels.remove(&event.client_order_id);
208        self.last_quoted_anchor = None;
209        self.last_quoted_residual = None;
210    }
211
212    fn on_order_filled(&mut self, event: &OrderFilled) {
213        let closed = {
214            let cache = self.cache();
215            cache
216                .order(&event.client_order_id)
217                .is_some_and(|o| o.is_closed())
218        };
219
220        if closed {
221            self.pending_self_cancels.remove(&event.client_order_id);
222        }
223    }
224
225    fn on_order_canceled(&mut self, event: &OrderCanceled) {
226        if self.pending_self_cancels.remove(&event.client_order_id) {
227            return;
228        }
229
230        if self.config.on_cancel_resubmit {
231            self.last_quoted_anchor = None;
232            self.last_quoted_residual = None;
233        }
234    }
235});
236
237impl Debug for CompositeMarketMaker {
238    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
239        f.debug_struct(stringify!(CompositeMarketMaker))
240            .field("config", &self.config)
241            .field("trade_size", &self.trade_size)
242            .field("signal_baseline", &self.signal_baseline)
243            .field("last_signal", &self.last_signal)
244            .finish()
245    }
246}
247
248impl DataActor for CompositeMarketMaker {
249    fn on_start(&mut self) -> anyhow::Result<()> {
250        let instrument_id = self.config.instrument_id;
251        let signal_instrument_id = self.config.signal_instrument_id;
252
253        let (instrument, size_precision, min_quantity) = {
254            let cache = self.cache();
255            let instrument = cache.try_instrument(&instrument_id)?;
256            let size_precision = instrument.size_precision();
257            let min_quantity = instrument.min_quantity();
258            (instrument, size_precision, min_quantity)
259        };
260        self.price_precision = Some(instrument.price_precision());
261        self.instrument = Some(instrument);
262
263        if self.trade_size.is_none() {
264            self.trade_size =
265                Some(min_quantity.unwrap_or_else(|| Quantity::new(1.0, size_precision)));
266        }
267
268        self.subscribe_quotes(instrument_id, None, None);
269        self.subscribe_quotes(signal_instrument_id, None, None);
270        Ok(())
271    }
272
273    fn on_stop(&mut self) -> anyhow::Result<()> {
274        let instrument_id = self.config.instrument_id;
275        let signal_instrument_id = self.config.signal_instrument_id;
276        self.cancel_all_orders(instrument_id, None, None, true, None)?;
277        self.close_all_positions(instrument_id, None, None, None, None, None, None, None)?;
278        self.unsubscribe_quotes(instrument_id, None, None);
279        self.unsubscribe_quotes(signal_instrument_id, None, None);
280        Ok(())
281    }
282
283    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
284        if quote.instrument_id == self.config.signal_instrument_id {
285            let bid = quote.bid_price.as_f64();
286            let ask = quote.ask_price.as_f64();
287
288            if bid <= 0.0 || ask <= 0.0 || bid > ask {
289                return Ok(());
290            }
291
292            let signal_mid = f64::midpoint(bid, ask);
293            self.last_signal = Some(signal_mid);
294            if self.signal_baseline.is_none() {
295                self.signal_baseline = Some(signal_mid);
296            }
297            return Ok(());
298        }
299
300        if quote.instrument_id != self.config.instrument_id {
301            return Ok(());
302        }
303
304        let Some(signal_residual) = self.signal_residual() else {
305            return Ok(());
306        };
307
308        let anchor_f64 = f64::midpoint(quote.bid_price.as_f64(), quote.ask_price.as_f64());
309        let price_precision = self.price_precision.ok_or_else(|| {
310            anyhow::anyhow!("Cannot handle quote: price_precision is not resolved")
311        })?;
312        let anchor = Price::new(anchor_f64, price_precision);
313
314        let instrument_id = self.config.instrument_id;
315        let strategy_id = self.strategy_id().expect("Strategy must be registered");
316
317        let has_resting = {
318            let cache = self.cache();
319            let inst = Some(&instrument_id);
320            let sid = Some(&strategy_id);
321            cache.orders_open_count(None, inst, sid, None, None) > 0
322                || cache.orders_inflight_count(None, inst, sid, None, None) > 0
323        };
324
325        if !self.should_requote(anchor, signal_residual) && has_resting {
326            return Ok(());
327        }
328
329        log::info!(
330            "Requoting: anchor={anchor}, last_anchor={:?}, residual={signal_residual:.6}, last_residual={:?}, instrument={instrument_id}",
331            self.last_quoted_anchor,
332            self.last_quoted_residual,
333        );
334
335        if self.config.on_cancel_resubmit {
336            let inst = Some(&instrument_id);
337            let strategy = Some(&strategy_id);
338            let ids: Vec<ClientOrderId> = {
339                let cache = self.cache();
340                let open = cache.orders_open(None, inst, strategy, None, None);
341                let inflight = cache.orders_inflight(None, inst, strategy, None, None);
342                open.iter()
343                    .chain(inflight.iter())
344                    .map(Order::client_order_id)
345                    .collect()
346            };
347            self.pending_self_cancels.extend(ids);
348        }
349
350        self.cancel_all_orders(instrument_id, None, None, true, None)?;
351
352        let (net_position, worst_long, worst_short) = self.position_exposure(strategy_id);
353
354        let quotes = self.compute_quotes(
355            anchor,
356            signal_residual,
357            net_position,
358            worst_long,
359            worst_short,
360        )?;
361
362        if quotes.is_empty() {
363            return Ok(());
364        }
365
366        let trade_size = self
367            .trade_size
368            .ok_or_else(|| anyhow::anyhow!("Cannot handle quote: trade_size is not resolved"))?;
369
370        let (tif, expire_time) = match self.config.expire_time_secs {
371            Some(secs) => {
372                let expire_ns = self.clock().timestamp_ns() + DurationNanos::try_from_secs(secs)?;
373                (Some(TimeInForce::Gtd), Some(expire_ns))
374            }
375            None => (None, None),
376        };
377
378        for (side, price) in quotes {
379            let order = self.order().limit(
380                instrument_id,
381                side,
382                trade_size,
383                price,
384                tif,
385                expire_time,
386                Some(true), // post_only
387                None,
388                None,
389                None,
390                None,
391                None,
392                None,
393                None,
394                None,
395                None,
396            );
397            self.submit_order(order, None, None, None)?;
398        }
399
400        self.last_quoted_anchor = Some(anchor);
401        self.last_quoted_residual = Some(signal_residual);
402        Ok(())
403    }
404
405    fn on_reset(&mut self) -> anyhow::Result<()> {
406        self.instrument = None;
407        self.trade_size = self.config.trade_size;
408        self.price_precision = None;
409        self.last_quoted_anchor = None;
410        self.last_quoted_residual = None;
411        self.signal_baseline = self.config.signal_baseline;
412        self.last_signal = None;
413        self.pending_self_cancels.clear();
414        Ok(())
415    }
416}
417
418impl CompositeMarketMaker {
419    fn position_exposure(&self, strategy_id: StrategyId) -> (f64, Decimal, Decimal) {
420        let instrument_id = Some(&self.config.instrument_id);
421        let strategy = Some(&strategy_id);
422        let cache = self.cache();
423
424        let mut position_qty = 0.0_f64;
425        let mut position_dec = Decimal::ZERO;
426
427        for p in cache.positions_open(None, instrument_id, strategy, None, None) {
428            position_qty += p.signed_qty;
429            position_dec += p.quantity.as_decimal()
430                * if p.signed_qty < 0.0 {
431                    Decimal::NEGATIVE_ONE
432                } else {
433                    Decimal::ONE
434                };
435        }
436
437        let mut pending_buy_dec = Decimal::ZERO;
438        let mut pending_sell_dec = Decimal::ZERO;
439        let mut seen = AHashSet::new();
440
441        let open = cache.orders_open(None, instrument_id, strategy, None, None);
442        let inflight = cache.orders_inflight(None, instrument_id, strategy, None, None);
443        for order in open.iter().chain(inflight.iter()) {
444            if !seen.insert(order.client_order_id()) {
445                continue;
446            }
447
448            let qty = order.leaves_qty().as_decimal();
449            match order.order_side() {
450                OrderSide::Buy => pending_buy_dec += qty,
451                _ => pending_sell_dec += qty,
452            }
453        }
454
455        (
456            position_qty,
457            position_dec + pending_buy_dec,
458            position_dec - pending_sell_dec,
459        )
460    }
461}
nautilus_trading/examples/strategies/composite_market_maker/strategy.rs

nautilus_trading/examples/strategies/composite_market_maker/
strategy.rs
