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