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

nautilus_trading/examples/strategies/composite_market_maker/
strategy.rs
