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    fn on_order_filled(&mut self, event: &OrderFilled) {
202        let closed = {
203            let cache = self.cache();
204            cache
205                .order(&event.client_order_id)
206                .is_some_and(|o| o.is_closed())
207        };
208
209        if closed {
210            self.pending_self_cancels.remove(&event.client_order_id);
211        }
212    }
213
214    fn on_order_canceled(&mut self, event: &OrderCanceled) {
215        if self.pending_self_cancels.remove(&event.client_order_id) {
216            return;
217        }
218
219        if self.config.on_cancel_resubmit {
220            self.last_quoted_anchor = None;
221            self.last_quoted_residual = None;
222        }
223    }
224});
225
226impl Debug for CompositeMarketMaker {
227    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
228        f.debug_struct(stringify!(CompositeMarketMaker))
229            .field("config", &self.config)
230            .field("trade_size", &self.trade_size)
231            .field("signal_baseline", &self.signal_baseline)
232            .field("last_signal", &self.last_signal)
233            .finish()
234    }
235}
236
237impl DataActor for CompositeMarketMaker {
238    fn on_start(&mut self) -> anyhow::Result<()> {
239        let instrument_id = self.config.instrument_id;
240        let signal_instrument_id = self.config.signal_instrument_id;
241
242        let (instrument, size_precision, min_quantity) = {
243            let cache = self.cache();
244            let instrument = cache.try_instrument(&instrument_id)?;
245            let size_precision = instrument.size_precision();
246            let min_quantity = instrument.min_quantity();
247            (instrument, size_precision, min_quantity)
248        };
249        self.price_precision = Some(instrument.price_precision());
250        self.instrument = Some(instrument);
251
252        if self.trade_size.is_none() {
253            self.trade_size =
254                Some(min_quantity.unwrap_or_else(|| Quantity::new(1.0, size_precision)));
255        }
256
257        self.subscribe_quotes(instrument_id, None, None);
258        self.subscribe_quotes(signal_instrument_id, None, None);
259        Ok(())
260    }
261
262    fn on_stop(&mut self) -> anyhow::Result<()> {
263        let instrument_id = self.config.instrument_id;
264        let signal_instrument_id = self.config.signal_instrument_id;
265        self.cancel_all_orders(instrument_id, None, None, true, None)?;
266        self.close_all_positions(instrument_id, None, None, None, None, None, None, None)?;
267        self.unsubscribe_quotes(instrument_id, None, None);
268        self.unsubscribe_quotes(signal_instrument_id, None, None);
269        Ok(())
270    }
271
272    fn on_quote(&mut self, quote: &QuoteTick) -> anyhow::Result<()> {
273        if quote.instrument_id == self.config.signal_instrument_id {
274            let signal_mid = f64::midpoint(quote.bid_price.as_f64(), quote.ask_price.as_f64());
275            self.last_signal = Some(signal_mid);
276            if self.signal_baseline.is_none() {
277                self.signal_baseline = Some(signal_mid);
278            }
279            return Ok(());
280        }
281
282        if quote.instrument_id != self.config.instrument_id {
283            return Ok(());
284        }
285
286        let anchor_f64 = f64::midpoint(quote.bid_price.as_f64(), quote.ask_price.as_f64());
287        let price_precision = self.price_precision.ok_or_else(|| {
288            anyhow::anyhow!("Cannot handle quote: price_precision is not resolved")
289        })?;
290        let anchor = Price::new(anchor_f64, price_precision);
291
292        let signal_residual = self.signal_residual();
293        let instrument_id = self.config.instrument_id;
294        let strategy_id = self.strategy_id().expect("Strategy must be registered");
295
296        let has_resting = {
297            let cache = self.cache();
298            let inst = Some(&instrument_id);
299            let sid = Some(&strategy_id);
300            cache.orders_open_count(None, inst, sid, None, None) > 0
301                || cache.orders_inflight_count(None, inst, sid, None, None) > 0
302        };
303
304        if !self.should_requote(anchor, signal_residual) && has_resting {
305            return Ok(());
306        }
307
308        log::info!(
309            "Requoting: anchor={anchor}, last_anchor={:?}, residual={signal_residual:.6}, last_residual={:?}, instrument={instrument_id}",
310            self.last_quoted_anchor,
311            self.last_quoted_residual,
312        );
313
314        if self.config.on_cancel_resubmit {
315            let inst = Some(&instrument_id);
316            let strategy = Some(&strategy_id);
317            let ids: Vec<ClientOrderId> = {
318                let cache = self.cache();
319                let open = cache.orders_open(None, inst, strategy, None, None);
320                let inflight = cache.orders_inflight(None, inst, strategy, None, None);
321                open.iter()
322                    .chain(inflight.iter())
323                    .map(Order::client_order_id)
324                    .collect()
325            };
326            self.pending_self_cancels.extend(ids);
327        }
328
329        self.cancel_all_orders(instrument_id, None, None, true, None)?;
330
331        let (net_position, worst_long, worst_short) = {
332            let instrument_id = Some(&instrument_id);
333            let strategy = Some(&strategy_id);
334            let cache = self.cache();
335
336            let mut position_qty = 0.0_f64;
337            let mut position_dec = Decimal::ZERO;
338
339            for p in cache.positions_open(None, instrument_id, strategy, None, None) {
340                position_qty += p.signed_qty;
341                position_dec += p.quantity.as_decimal()
342                    * if p.signed_qty < 0.0 {
343                        Decimal::NEGATIVE_ONE
344                    } else {
345                        Decimal::ONE
346                    };
347            }
348
349            let mut pending_buy_dec = Decimal::ZERO;
350            let mut pending_sell_dec = Decimal::ZERO;
351            let mut seen = AHashSet::new();
352
353            let open = cache.orders_open(None, instrument_id, strategy, None, None);
354            let inflight = cache.orders_inflight(None, instrument_id, strategy, None, None);
355            for order in open.iter().chain(inflight.iter()) {
356                if !seen.insert(order.client_order_id()) {
357                    continue;
358                }
359                let qty = order.leaves_qty().as_decimal();
360                match order.order_side() {
361                    OrderSide::Buy => pending_buy_dec += qty,
362                    _ => pending_sell_dec += qty,
363                }
364            }
365
366            (
367                position_qty,
368                position_dec + pending_buy_dec,
369                position_dec - pending_sell_dec,
370            )
371        };
372
373        let quotes = self.compute_quotes(
374            anchor,
375            signal_residual,
376            net_position,
377            worst_long,
378            worst_short,
379        )?;
380
381        if quotes.is_empty() {
382            return Ok(());
383        }
384
385        let trade_size = self
386            .trade_size
387            .ok_or_else(|| anyhow::anyhow!("Cannot handle quote: trade_size is not resolved"))?;
388
389        let (tif, expire_time) = match self.config.expire_time_secs {
390            Some(secs) => {
391                let now_ns = self.clock().timestamp_ns();
392                let expire_ns = now_ns + secs * 1_000_000_000;
393                (Some(TimeInForce::Gtd), Some(expire_ns))
394            }
395            None => (None, None),
396        };
397
398        for (side, price) in quotes {
399            let order = self.order().limit(
400                instrument_id,
401                side,
402                trade_size,
403                price,
404                tif,
405                expire_time,
406                Some(true), // post_only
407                None,
408                None,
409                None,
410                None,
411                None,
412                None,
413                None,
414                None,
415                None,
416            );
417            self.submit_order(order, None, None, None)?;
418        }
419
420        self.last_quoted_anchor = Some(anchor);
421        self.last_quoted_residual = Some(signal_residual);
422        Ok(())
423    }
424
425    fn on_reset(&mut self) -> anyhow::Result<()> {
426        self.instrument = None;
427        self.trade_size = self.config.trade_size;
428        self.price_precision = None;
429        self.last_quoted_anchor = None;
430        self.last_quoted_residual = None;
431        self.signal_baseline = self.config.signal_baseline;
432        self.last_signal = None;
433        self.pending_self_cancels.clear();
434        Ok(())
435    }
436}
nautilus_trading/examples/strategies/composite_market_maker/strategy.rs

nautilus_trading/examples/strategies/composite_market_maker/
strategy.rs
