nautilus_data/engine/
synthetic.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
16use super::{DataEngine, InstrumentId, Quantity, QuoteTick, SyntheticInstrument, TradeTick};
17
18impl DataEngine {
19    /// Returns all synthetic instrument IDs for which quote subscriptions exist.
20    #[must_use]
21    pub fn subscribed_synthetic_quotes(&self) -> Vec<InstrumentId> {
22        self.subscribed_synthetic_quotes.keys().copied().collect()
23    }
24
25    /// Returns all synthetic instrument IDs for which trade subscriptions exist.
26    #[must_use]
27    pub fn subscribed_synthetic_trades(&self) -> Vec<InstrumentId> {
28        self.subscribed_synthetic_trades.keys().copied().collect()
29    }
30
31    pub(super) fn synthetic_quotes_from_quote(&self, update: QuoteTick) -> Vec<QuoteTick> {
32        let Some(synthetics) = self.synthetic_quote_feeds.get(&update.instrument_id) else {
33            return Vec::new();
34        };
35
36        synthetics
37            .iter()
38            .filter_map(|synthetic| self.synthetic_quote_from_update(synthetic, update))
39            .collect()
40    }
41
42    fn synthetic_quote_from_update(
43        &self,
44        synthetic: &SyntheticInstrument,
45        update: QuoteTick,
46    ) -> Option<QuoteTick> {
47        let cache = self.cache.borrow();
48        let mut bid_inputs = Vec::with_capacity(synthetic.components.len());
49        let mut ask_inputs = Vec::with_capacity(synthetic.components.len());
50
51        for instrument_id in &synthetic.components {
52            let (bid_price, ask_price) = if *instrument_id == update.instrument_id {
53                (update.bid_price, update.ask_price)
54            } else {
55                let Some(component_quote) = cache.quote(instrument_id) else {
56                    log::warn!(
57                        "Cannot calculate synthetic instrument {} price, no quotes for {} yet",
58                        synthetic.id,
59                        instrument_id,
60                    );
61                    return None;
62                };
63
64                (component_quote.bid_price, component_quote.ask_price)
65            };
66
67            bid_inputs.push(bid_price.as_f64());
68            ask_inputs.push(ask_price.as_f64());
69        }
70
71        drop(cache);
72
73        let bid_price = match synthetic.calculate(&bid_inputs) {
74            Ok(price) => price,
75            Err(e) => {
76                log::error!(
77                    "Cannot calculate synthetic instrument {} bid price: {e}",
78                    synthetic.id
79                );
80                return None;
81            }
82        };
83
84        let ask_price = match synthetic.calculate(&ask_inputs) {
85            Ok(price) => price,
86            Err(e) => {
87                log::error!(
88                    "Cannot calculate synthetic instrument {} ask price: {e}",
89                    synthetic.id
90                );
91                return None;
92            }
93        };
94
95        let size_one = Quantity::from(1);
96
97        Some(QuoteTick::new(
98            synthetic.id,
99            bid_price,
100            ask_price,
101            size_one,
102            size_one,
103            update.ts_event,
104            self.clock.borrow().timestamp_ns(),
105        ))
106    }
107
108    pub(super) fn synthetic_trades_from_trade(&self, update: TradeTick) -> Vec<TradeTick> {
109        let Some(synthetics) = self.synthetic_trade_feeds.get(&update.instrument_id) else {
110            return Vec::new();
111        };
112
113        synthetics
114            .iter()
115            .filter_map(|synthetic| self.synthetic_trade_from_update(synthetic, update))
116            .collect()
117    }
118
119    fn synthetic_trade_from_update(
120        &self,
121        synthetic: &SyntheticInstrument,
122        update: TradeTick,
123    ) -> Option<TradeTick> {
124        let cache = self.cache.borrow();
125        let mut inputs = Vec::with_capacity(synthetic.components.len());
126
127        for instrument_id in &synthetic.components {
128            let price = if *instrument_id == update.instrument_id {
129                update.price
130            } else {
131                let Some(component_trade) = cache.trade(instrument_id) else {
132                    log::warn!(
133                        "Cannot calculate synthetic instrument {} price, no trades for {} yet",
134                        synthetic.id,
135                        instrument_id,
136                    );
137                    return None;
138                };
139
140                component_trade.price
141            };
142
143            inputs.push(price.as_f64());
144        }
145
146        drop(cache);
147
148        let price = match synthetic.calculate(&inputs) {
149            Ok(price) => price,
150            Err(e) => {
151                log::error!(
152                    "Cannot calculate synthetic instrument {} trade price: {e}",
153                    synthetic.id
154                );
155                return None;
156            }
157        };
158
159        Some(TradeTick::new(
160            synthetic.id,
161            price,
162            Quantity::from(1),
163            update.aggressor_side,
164            update.trade_id,
165            update.ts_event,
166            self.clock.borrow().timestamp_ns(),
167        ))
168    }
169
170    pub(super) fn subscribe_synthetic_quotes(&mut self, instrument_id: InstrumentId) {
171        let synthetic = match self.cache.borrow().try_synthetic(&instrument_id).cloned() {
172            Ok(synthetic) => synthetic,
173            Err(e) => {
174                log::error!("Cannot subscribe to `QuoteTick` data for synthetic instrument: {e}");
175                return;
176            }
177        };
178
179        if let Some(owners) = self.subscribed_synthetic_quotes.get_mut(&instrument_id) {
180            *owners += 1;
181            return;
182        }
183
184        self.subscribed_synthetic_quotes.insert(instrument_id, 1);
185
186        for component_id in &synthetic.components {
187            let synthetics = self.synthetic_quote_feeds.entry(*component_id).or_default();
188            if !synthetics
189                .iter()
190                .any(|registered| registered.id == synthetic.id)
191            {
192                synthetics.push(synthetic.clone());
193            }
194        }
195    }
196
197    pub(super) fn subscribe_synthetic_trades(&mut self, instrument_id: InstrumentId) {
198        let synthetic = match self.cache.borrow().try_synthetic(&instrument_id).cloned() {
199            Ok(synthetic) => synthetic,
200            Err(e) => {
201                log::error!("Cannot subscribe to `TradeTick` data for synthetic instrument: {e}");
202                return;
203            }
204        };
205
206        if let Some(owners) = self.subscribed_synthetic_trades.get_mut(&instrument_id) {
207            *owners += 1;
208            return;
209        }
210
211        self.subscribed_synthetic_trades.insert(instrument_id, 1);
212
213        for component_id in &synthetic.components {
214            let synthetics = self.synthetic_trade_feeds.entry(*component_id).or_default();
215            if !synthetics
216                .iter()
217                .any(|registered| registered.id == synthetic.id)
218            {
219                synthetics.push(synthetic.clone());
220            }
221        }
222    }
223
224    pub(super) fn unsubscribe_synthetic_quotes(&mut self, instrument_id: InstrumentId) {
225        let Some(owners) = self.subscribed_synthetic_quotes.get_mut(&instrument_id) else {
226            log::warn!("Cannot unsubscribe from synthetic `QuoteTick` data: not subscribed");
227            return;
228        };
229
230        if *owners > 1 {
231            *owners -= 1;
232            return;
233        }
234
235        self.subscribed_synthetic_quotes.remove(&instrument_id);
236
237        self.synthetic_quote_feeds.retain(|_, synthetics| {
238            synthetics.retain(|synthetic| synthetic.id != instrument_id);
239            !synthetics.is_empty()
240        });
241    }
242
243    pub(super) fn unsubscribe_synthetic_trades(&mut self, instrument_id: InstrumentId) {
244        let Some(owners) = self.subscribed_synthetic_trades.get_mut(&instrument_id) else {
245            log::warn!("Cannot unsubscribe from synthetic `TradeTick` data: not subscribed");
246            return;
247        };
248
249        if *owners > 1 {
250            *owners -= 1;
251            return;
252        }
253
254        self.subscribed_synthetic_trades.remove(&instrument_id);
255
256        self.synthetic_trade_feeds.retain(|_, synthetics| {
257            synthetics.retain(|synthetic| synthetic.id != instrument_id);
258            !synthetics.is_empty()
259        });
260    }
261}
nautilus_data/engine/synthetic.rs

nautilus_data/engine/
synthetic.rs
