nautilus_data/engine/
streaming.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 ahash::AHashMap;
17use jiff::Timestamp;
18use nautilus_common::messages::data::{
19    BarsResponse, BookDeltasResponse, BookDepthResponse, CustomDataResponse, DataResponse,
20    FundingRatesResponse, InstrumentResponse, InstrumentsResponse, QuotesResponse, RequestBars,
21    RequestBookDeltas, RequestBookDepth, RequestCommand, RequestCustomData, RequestFundingRates,
22    RequestInstrument, RequestInstruments, RequestQuotes, RequestTrades, SubscribeBars,
23    SubscribeCommand, SubscribeCustomData, SubscribeQuotes, SubscribeTrades, TradesResponse,
24};
25use nautilus_core::{
26    Params, UUID4, UnixNanos,
27    correctness::{FAILED, check_key_not_in_map},
28};
29use nautilus_model::{
30    data::{
31        Bar, CustomData, DataBatch, FromDataBatch, FundingRateUpdate, NautilusDataType,
32        OrderBookDelta, OrderBookDepth, QuoteTick, TradeTick,
33    },
34    identifiers::{ClientId, Venue},
35    instruments::{Instrument, InstrumentAny},
36};
37use nautilus_persistence::catalog::traits::{CatalogInstrumentQuery, CatalogQuery, DataCatalog};
38use serde_json::Value;
39use ustr::Ustr;
40
41use super::{DataEngine, requests::request_params};
42
43const PARAM_SKIP_CATALOG_DATA: &str = "skip_catalog_data";
44const PARAM_UPDATE_CATALOG: &str = "update_catalog";
45const PARAM_FORCE_INSTRUMENT_UPDATE: &str = "force_instrument_update";
46const PARAM_SUBSCRIPTION_NAME: &str = "subscription_name";
47const PARAM_FROM_DAY_START: &str = "from_day_start";
48const CATALOG_CLIENT_ID: &str = "CATALOG";
49
50pub(crate) type CatalogMap = AHashMap<Ustr, DataCatalog>;
51
52impl DataEngine {
53    /// Registers the `catalog` with the engine with an optional specific `name`.
54    ///
55    /// # Panics
56    ///
57    /// Panics if a catalog with the same `name` has already been registered.
58    pub fn register_catalog(&mut self, catalog: DataCatalog, name: Option<&str>) {
59        let name = Ustr::from(name.unwrap_or("catalog_0"));
60
61        check_key_not_in_map(&name, &self.catalogs, "name", "catalogs").expect(FAILED);
62
63        self.catalogs.insert(name, catalog);
64        log::info!("Registered catalog <{name}>");
65    }
66
67    pub(super) fn subscribe_command_with_prefilled_start_ns(
68        &mut self,
69        cmd: SubscribeCommand,
70    ) -> anyhow::Result<SubscribeCommand> {
71        match cmd {
72            SubscribeCommand::Quotes(cmd) if Self::is_start_ns_missing(cmd.params.as_ref()) => {
73                let identifier = cmd.instrument_id.to_string();
74                let params = self.params_with_prefilled_start_ns(
75                    cmd.params.as_ref(),
76                    &NautilusDataType::QuoteTick,
77                    &identifier,
78                )?;
79                Ok(SubscribeCommand::Quotes(SubscribeQuotes { params, ..cmd }))
80            }
81            SubscribeCommand::Trades(cmd) if Self::is_start_ns_missing(cmd.params.as_ref()) => {
82                let identifier = cmd.instrument_id.to_string();
83                let params = self.params_with_prefilled_start_ns(
84                    cmd.params.as_ref(),
85                    &NautilusDataType::TradeTick,
86                    &identifier,
87                )?;
88                Ok(SubscribeCommand::Trades(SubscribeTrades { params, ..cmd }))
89            }
90            SubscribeCommand::Bars(cmd)
91                if cmd.bar_type.is_externally_aggregated()
92                    && Self::is_start_ns_missing(cmd.params.as_ref()) =>
93            {
94                let identifier = cmd.bar_type.to_string();
95                let params = self.params_with_prefilled_start_ns(
96                    cmd.params.as_ref(),
97                    &NautilusDataType::Bar,
98                    &identifier,
99                )?;
100                Ok(SubscribeCommand::Bars(SubscribeBars { params, ..cmd }))
101            }
102            SubscribeCommand::Data(cmd) if Self::is_start_ns_missing(cmd.params.as_ref()) => {
103                let type_name = cmd.data_type.type_name().to_string();
104                let identifier = cmd.data_type.identifier().map(String::from);
105                let params = self.params_with_custom_data_prefilled_start_ns(
106                    cmd.params.as_ref(),
107                    &type_name,
108                    identifier.as_deref(),
109                )?;
110                Ok(SubscribeCommand::Data(SubscribeCustomData {
111                    params,
112                    ..cmd
113                }))
114            }
115            _ => Ok(cmd),
116        }
117    }
118
119    fn is_start_ns_missing(params: Option<&Params>) -> bool {
120        params.is_none_or(|params| !params.contains_key("start_ns"))
121    }
122
123    fn params_with_prefilled_start_ns(
124        &mut self,
125        params: Option<&Params>,
126        data_type: &NautilusDataType,
127        identifier: &str,
128    ) -> anyhow::Result<Option<Params>> {
129        let last_timestamp = self.catalog_last_timestamp(data_type, identifier)?;
130
131        Ok(Some(Self::params_with_start_ns(params, last_timestamp)))
132    }
133
134    fn params_with_custom_data_prefilled_start_ns(
135        &mut self,
136        params: Option<&Params>,
137        type_name: &str,
138        identifier: Option<&str>,
139    ) -> anyhow::Result<Option<Params>> {
140        let last_timestamp = self.catalog_custom_data_last_timestamp(type_name, identifier)?;
141
142        Ok(Some(Self::params_with_start_ns(params, last_timestamp)))
143    }
144
145    fn params_with_start_ns(params: Option<&Params>, last_timestamp: Option<u64>) -> Params {
146        let start_ns = last_timestamp.map_or(Value::Null, |last_timestamp| {
147            Value::from(last_timestamp.saturating_add(1))
148        });
149        let mut params = params.cloned().unwrap_or_else(Params::new);
150
151        params.insert("start_ns".to_string(), start_ns);
152
153        params
154    }
155
156    fn catalog_last_timestamp(
157        &mut self,
158        data_type: &NautilusDataType,
159        identifier: &str,
160    ) -> anyhow::Result<Option<u64>> {
161        for catalog in self.catalogs.values_mut() {
162            if let Some(last_timestamp) =
163                catalog.query_last_timestamp(data_type.clone(), Some(identifier))?
164            {
165                return Ok(Some(last_timestamp));
166            }
167        }
168
169        Ok(None)
170    }
171
172    fn catalog_custom_data_last_timestamp(
173        &mut self,
174        type_name: &str,
175        identifier: Option<&str>,
176    ) -> anyhow::Result<Option<u64>> {
177        // `make_path_custom_data` / `get_directory_intervals` are inherent on
178        // `ParquetDataCatalog` and not exposed through `Catalog`. Use the
179        // trait-level `query_last_timestamp` with `NautilusDataType::Custom` so the
180        // path works for any backend that implements the trait.
181        let data_type = NautilusDataType::Custom {
182            type_name: type_name.to_string(),
183        };
184
185        for catalog in self.catalogs.values_mut() {
186            if let Some(last_timestamp) =
187                catalog.query_last_timestamp(data_type.clone(), identifier)?
188            {
189                return Ok(Some(last_timestamp));
190            }
191        }
192
193        Ok(None)
194    }
195
196    pub(super) fn catalogs_registered(&self) -> bool {
197        !self.catalogs.is_empty()
198    }
199
200    // Bounds the request window, walks the catalogs to find one whose missing
201    // intervals differ from the full requested range, then fans the parent out via
202    // the pipeline with one catalog leg plus one client leg per missing interval.
203    // With no catalog match and no resolvable client, the engine emits an empty
204    // response keyed by the parent request ID.
205    pub(super) fn dispatch_date_range_request(
206        &mut self,
207        req: RequestCommand,
208    ) -> anyhow::Result<()> {
209        if matches!(
210            req,
211            RequestCommand::Instrument(_) | RequestCommand::Instruments(_)
212        ) {
213            return self.dispatch_instrument_catalog_request(req);
214        }
215
216        let Some(key) = request_identifier(&req) else {
217            return self.dispatch_request_to_client(req).map(|_| ());
218        };
219
220        let now_ns = self.clock.borrow().timestamp_ns();
221        let now_dt = now_ns.to_datetime_utc();
222        let query_past_data = request_params(&req)
223            .and_then(|p| p.get(PARAM_SUBSCRIPTION_NAME))
224            .is_none();
225
226        let (start_dt, end_dt) = bound_request_dates(
227            request_start(&req),
228            request_end(&req),
229            now_dt,
230            query_past_data,
231        );
232        let start_ns = datetime_to_unix_nanos_or_zero(start_dt);
233        let end_ns = datetime_to_unix_nanos_or_zero(end_dt);
234
235        if start_ns > end_ns {
236            anyhow::bail!(
237                "Cannot dispatch request, start {start_ns} was greater than end {end_ns}"
238            );
239        }
240
241        let client_id = req.client_id().copied();
242        let venue = req.venue().copied();
243        let used_client_id = self
244            .get_client(client_id.as_ref(), venue.as_ref())
245            .map(|client| client.client_id());
246
247        // Floor the catalog window to the UTC day boundary so the day-start F_SNAPSHOT frame is
248        // selected and read for the snapshot replay; client gaps keep the original window.
249        // The parent request keeps its original start, so the merged response trims back to it.
250        let (catalog_start_dt, catalog_start_ns) = if matches!(req, RequestCommand::BookDeltas(_))
251            && request_params(&req)
252                .and_then(|p| p.get_bool(PARAM_FROM_DAY_START))
253                .unwrap_or(true)
254        {
255            let floored = floor_to_utc_day(start_dt);
256            (floored, datetime_to_unix_nanos_or_zero(floored))
257        } else {
258            (start_dt, start_ns)
259        };
260
261        let query_interval = vec![(start_ns.as_u64(), end_ns.as_u64())];
262        let catalog_query_interval = vec![(catalog_start_ns.as_u64(), end_ns.as_u64())];
263        let mut missing_intervals = query_interval.clone();
264        let mut has_catalog_data = false;
265        let mut winning_catalog: Option<Ustr> = None;
266
267        for (name, catalog) in &mut self.catalogs {
268            let catalog_intervals = catalog_missing_intervals(
269                catalog,
270                catalog_start_ns.as_u64(),
271                end_ns.as_u64(),
272                &key,
273            )?;
274
275            if catalog_intervals != catalog_query_interval {
276                has_catalog_data = true;
277                winning_catalog = Some(*name);
278                // Client legs fill only the requested window, not the pre-start range
279                missing_intervals = if catalog_start_ns == start_ns {
280                    catalog_intervals
281                } else {
282                    catalog_missing_intervals(catalog, start_ns.as_u64(), end_ns.as_u64(), &key)?
283                };
284                break;
285            }
286        }
287
288        let skip_catalog_data = request_params(&req)
289            .and_then(|p| p.get_bool(PARAM_SKIP_CATALOG_DATA))
290            .unwrap_or(false);
291
292        // When `skip_catalog_data` is set the client must serve the full parent window;
293        // dropping the catalog leg without resetting the missing intervals would leave
294        // the catalog-covered range unanswered.
295        if skip_catalog_data {
296            missing_intervals = query_interval;
297        }
298
299        let n_client_requests = if used_client_id.is_some() {
300            missing_intervals.len()
301        } else {
302            0
303        };
304        let n_catalog_requests = usize::from(has_catalog_data && !skip_catalog_data);
305        let n_requests = n_client_requests + n_catalog_requests;
306
307        if n_requests == 0 {
308            let empty = build_empty_response(&req, start_ns, end_ns, used_client_id, now_ns)?;
309            self.response(empty);
310            return Ok(());
311        }
312
313        let parent_id = *req.request_id();
314        self.new_request_pipeline(req.clone(), n_requests);
315
316        if n_catalog_requests == 1
317            && let Some(catalog_name) = winning_catalog
318        {
319            let leg = with_dates_for_pipeline(&req, Some(catalog_start_dt), Some(end_dt), now_ns);
320            let leg_id = *leg.request_id();
321            self.register_request_pipeline_leg(leg_id, parent_id);
322
323            match self.query_catalog_leg(
324                &leg,
325                catalog_name,
326                catalog_start_ns,
327                end_ns,
328                used_client_id,
329                now_ns,
330            ) {
331                Ok(resp) => self.response(resp),
332                Err(e) => {
333                    log::error!(
334                        "Catalog leg query failed for parent {parent_id} (catalog {catalog_name}): {e}"
335                    );
336                    let empty = match build_empty_response(
337                        &leg,
338                        start_ns,
339                        end_ns,
340                        used_client_id,
341                        now_ns,
342                    ) {
343                        Ok(empty) => empty,
344                        Err(e) => {
345                            self.abort_request_pipeline(parent_id);
346                            return Err(e);
347                        }
348                    };
349                    self.response(empty);
350                }
351            }
352        }
353
354        if n_client_requests > 0 {
355            for (leg_start_ns, leg_end_ns) in &missing_intervals {
356                let leg_start_dt = UnixNanos::from(*leg_start_ns).to_datetime_utc();
357                let leg_end_dt = UnixNanos::from(*leg_end_ns).to_datetime_utc();
358                let leg =
359                    with_dates_for_pipeline(&req, Some(leg_start_dt), Some(leg_end_dt), now_ns);
360                let leg_id = *leg.request_id();
361                self.register_request_pipeline_leg(leg_id, parent_id);
362
363                if let Err(e) = self.dispatch_request_to_client(leg) {
364                    // Abort the whole pipeline so the parent does not stay half-registered
365                    // waiting on a leg the client never accepted. Any catalog leg already
366                    // buffered for this parent is discarded with the pipeline state.
367                    log::error!("Client leg dispatch failed for parent {parent_id}: {e}");
368                    self.abort_request_pipeline(parent_id);
369                    return Err(e);
370                }
371            }
372        }
373
374        Ok(())
375    }
376
377    fn abort_request_pipeline(&mut self, parent_id: UUID4) {
378        self.request_pipeline_n_components.remove(&parent_id);
379        self.request_pipeline_parent_request.remove(&parent_id);
380        self.request_pipeline_responses.remove(&parent_id);
381        self.request_pipeline_parent_request_id
382            .retain(|_, p_id| *p_id != parent_id);
383    }
384
385    fn query_catalog_leg(
386        &mut self,
387        leg: &RequestCommand,
388        catalog_name: Ustr,
389        start_ns: UnixNanos,
390        end_ns: UnixNanos,
391        used_client_id: Option<ClientId>,
392        ts_init: UnixNanos,
393    ) -> anyhow::Result<DataResponse> {
394        let catalog = self.catalogs.get_mut(&catalog_name).ok_or_else(|| {
395            anyhow::anyhow!("Catalog {catalog_name} disappeared between intervals query and read")
396        })?;
397
398        match leg {
399            RequestCommand::Quotes(cmd) => {
400                let data = quote_ticks_from_query_result(
401                    catalog.query_batch(
402                        &CatalogQuery::new(NautilusDataType::QuoteTick)
403                            .with_identifiers(Some(vec![cmd.instrument_id.to_string()]))
404                            .with_range(Some(start_ns), Some(end_ns)),
405                    )?,
406                )?;
407                Ok(build_quotes_catalog_response(
408                    cmd,
409                    data,
410                    start_ns,
411                    end_ns,
412                    used_client_id,
413                    ts_init,
414                ))
415            }
416            RequestCommand::Trades(cmd) => {
417                let data = trade_ticks_from_query_result(
418                    catalog.query_batch(
419                        &CatalogQuery::new(NautilusDataType::TradeTick)
420                            .with_identifiers(Some(vec![cmd.instrument_id.to_string()]))
421                            .with_range(Some(start_ns), Some(end_ns)),
422                    )?,
423                )?;
424                Ok(build_trades_catalog_response(
425                    cmd,
426                    data,
427                    start_ns,
428                    end_ns,
429                    used_client_id,
430                    ts_init,
431                ))
432            }
433            RequestCommand::FundingRates(cmd) => {
434                let data = funding_rates_from_query_result(
435                    catalog.query_batch(
436                        &CatalogQuery::new(NautilusDataType::FundingRateUpdate)
437                            .with_identifiers(Some(vec![cmd.instrument_id.to_string()]))
438                            .with_range(Some(start_ns), Some(end_ns)),
439                    )?,
440                )?;
441                Ok(build_funding_rates_catalog_response(
442                    cmd,
443                    data,
444                    start_ns,
445                    end_ns,
446                    used_client_id,
447                    ts_init,
448                ))
449            }
450            RequestCommand::Bars(cmd) => {
451                let data = bars_from_query_result(
452                    catalog.query_batch(
453                        &CatalogQuery::new(NautilusDataType::Bar)
454                            .with_identifiers(Some(vec![cmd.bar_type.to_string()]))
455                            .with_range(Some(start_ns), Some(end_ns)),
456                    )?,
457                )?;
458                Ok(build_bars_catalog_response(
459                    cmd,
460                    data,
461                    start_ns,
462                    end_ns,
463                    used_client_id,
464                    ts_init,
465                ))
466            }
467            RequestCommand::Data(cmd) => {
468                let identifiers = cmd
469                    .data_type
470                    .identifier()
471                    .map(|identifier| vec![identifier.to_string()]);
472                let where_clause = request_filter_expr(cmd.params.as_ref());
473                let data = catalog.query_batch(
474                    &CatalogQuery::new(NautilusDataType::Custom {
475                        type_name: cmd.data_type.type_name().to_string(),
476                    })
477                    .with_identifiers(identifiers)
478                    .with_range(Some(start_ns), Some(end_ns))
479                    .with_where_clause(where_clause),
480                )?;
481                Ok(build_custom_data_catalog_response(
482                    cmd,
483                    custom_data_from_query_result(data)?,
484                    start_ns,
485                    end_ns,
486                    ts_init,
487                ))
488            }
489            RequestCommand::BookDeltas(cmd) => {
490                let data = order_book_deltas_from_query_result(
491                    catalog.query_batch(
492                        &CatalogQuery::new(NautilusDataType::OrderBookDelta)
493                            .with_identifiers(Some(vec![cmd.instrument_id.to_string()]))
494                            .with_range(Some(start_ns), Some(end_ns)),
495                    )?,
496                )?;
497                Ok(build_book_deltas_catalog_response(
498                    cmd,
499                    data,
500                    start_ns,
501                    end_ns,
502                    used_client_id,
503                    ts_init,
504                ))
505            }
506            RequestCommand::BookDepth(cmd) => {
507                let data = order_book_depths_from_query_result(
508                    catalog.query_batch(
509                        &CatalogQuery::new(NautilusDataType::OrderBookDepth)
510                            .with_identifiers(Some(vec![cmd.instrument_id.to_string()]))
511                            .with_range(Some(start_ns), Some(end_ns)),
512                    )?,
513                )?;
514                Ok(build_book_depth_catalog_response(
515                    cmd,
516                    data,
517                    start_ns,
518                    end_ns,
519                    used_client_id,
520                    ts_init,
521                ))
522            }
523            _ => {
524                anyhow::bail!("query_catalog_leg called with non-catalog-eligible variant {leg:?}")
525            }
526        }
527    }
528
529    fn dispatch_instrument_catalog_request(&mut self, req: RequestCommand) -> anyhow::Result<()> {
530        match req {
531            RequestCommand::Instrument(cmd) => self.dispatch_instrument_request(cmd),
532            RequestCommand::Instruments(cmd) => self.dispatch_instruments_request(cmd),
533            _ => self.dispatch_request_to_client(req).map(|_| ()),
534        }
535    }
536
537    fn dispatch_instrument_request(&mut self, cmd: RequestInstrument) -> anyhow::Result<()> {
538        let force_instrument_update = cmd
539            .params
540            .as_ref()
541            .and_then(|params| params.get_bool(PARAM_FORCE_INSTRUMENT_UPDATE))
542            .unwrap_or(false);
543
544        if force_instrument_update {
545            return self
546                .dispatch_request_to_client(RequestCommand::Instrument(cmd))
547                .map(|_| ());
548        }
549
550        let identifier = cmd.instrument_id.to_string();
551        let now_ns = self.clock.borrow().timestamp_ns();
552        let used_client_id = self
553            .get_client(cmd.client_id.as_ref(), Some(&cmd.instrument_id.venue))
554            .map(|client| client.client_id());
555        let (start_dt, end_dt) =
556            bound_request_dates(cmd.start, cmd.end, now_ns.to_datetime_utc(), true);
557        let start_ns = datetime_to_unix_nanos_or_zero(start_dt);
558        let end_ns = datetime_to_unix_nanos_or_zero(end_dt);
559        let start = Some(start_ns);
560        let end = cmd.end.map(datetime_to_unix_nanos_or_zero);
561
562        for catalog in self.catalogs.values_mut() {
563            let data = latest_instruments(
564                catalog.instruments(
565                    &CatalogInstrumentQuery::new()
566                        .with_instrument_ids(Some(vec![identifier.clone()]))
567                        .with_range(start, end),
568                )?,
569            );
570
571            if let Some(instrument) = data.into_iter().next() {
572                let response = DataResponse::Instrument(Box::new(InstrumentResponse::new(
573                    cmd.request_id,
574                    resolve_response_client_id(cmd.client_id, used_client_id),
575                    cmd.instrument_id,
576                    instrument,
577                    Some(start_ns),
578                    Some(end_ns),
579                    now_ns,
580                    Some(catalog_response_params(cmd.params.as_ref())),
581                )));
582                self.response(response);
583                return Ok(());
584            }
585        }
586
587        self.dispatch_request_to_client(RequestCommand::Instrument(cmd))
588            .map(|_| ())
589    }
590
591    fn dispatch_instruments_request(&mut self, cmd: RequestInstruments) -> anyhow::Result<()> {
592        let update_catalog = cmd
593            .params
594            .as_ref()
595            .and_then(|params| params.get_bool(PARAM_UPDATE_CATALOG))
596            .unwrap_or(false);
597        let force_instrument_update = cmd
598            .params
599            .as_ref()
600            .and_then(|params| params.get_bool(PARAM_FORCE_INSTRUMENT_UPDATE))
601            .unwrap_or(false);
602
603        if update_catalog || force_instrument_update {
604            return self
605                .dispatch_request_to_client(RequestCommand::Instruments(cmd))
606                .map(|_| ());
607        }
608
609        let now_ns = self.clock.borrow().timestamp_ns();
610        let used_client_id = self
611            .get_client(cmd.client_id.as_ref(), cmd.venue.as_ref())
612            .map(|client| client.client_id());
613        let (start_dt, end_dt) =
614            bound_request_dates(cmd.start, cmd.end, now_ns.to_datetime_utc(), true);
615        let start_ns = datetime_to_unix_nanos_or_zero(start_dt);
616        let end_ns = datetime_to_unix_nanos_or_zero(end_dt);
617        let start = Some(start_ns);
618        let end = cmd.end.map(datetime_to_unix_nanos_or_zero);
619        let mut data = Vec::new();
620
621        for catalog in self.catalogs.values_mut() {
622            data.extend(
623                catalog.instruments(
624                    &CatalogInstrumentQuery::new()
625                        .with_range(start, end)
626                        .with_where_clause(request_filter_expr(cmd.params.as_ref())),
627                )?,
628            );
629        }
630
631        if let Some(venue) = cmd.venue {
632            data.retain(|instrument| instrument.venue() == venue);
633        }
634
635        if instrument_only_last(cmd.params.as_ref()) {
636            data = latest_instruments(data);
637        }
638
639        let response = DataResponse::Instruments(InstrumentsResponse::new(
640            cmd.request_id,
641            resolve_response_client_id(cmd.client_id, used_client_id),
642            instrument_response_venue(cmd.venue, &data),
643            data,
644            Some(start_ns),
645            Some(end_ns),
646            now_ns,
647            Some(catalog_response_params(cmd.params.as_ref())),
648        ));
649        self.response(response);
650        Ok(())
651    }
652}
653
654struct RequestCatalogKey {
655    data_type: NautilusDataType,
656    identifier: Option<String>,
657}
658
659pub(super) fn is_date_range_variant(req: &RequestCommand) -> bool {
660    matches!(
661        req,
662        RequestCommand::Data(_)
663            | RequestCommand::Instrument(_)
664            | RequestCommand::Instruments(_)
665            | RequestCommand::Quotes(_)
666            | RequestCommand::Trades(_)
667            | RequestCommand::FundingRates(_)
668            | RequestCommand::Bars(_)
669            | RequestCommand::BookDeltas(_)
670            | RequestCommand::BookDepth(_)
671    )
672}
673
674fn request_identifier(req: &RequestCommand) -> Option<RequestCatalogKey> {
675    match req {
676        RequestCommand::Data(cmd) => Some(RequestCatalogKey {
677            data_type: NautilusDataType::Custom {
678                type_name: cmd.data_type.type_name().to_string(),
679            },
680            identifier: cmd.data_type.identifier().map(String::from),
681        }),
682        RequestCommand::Quotes(cmd) => Some(RequestCatalogKey::new(
683            NautilusDataType::QuoteTick,
684            Some(cmd.instrument_id.to_string()),
685        )),
686        RequestCommand::Trades(cmd) => Some(RequestCatalogKey::new(
687            NautilusDataType::TradeTick,
688            Some(cmd.instrument_id.to_string()),
689        )),
690        RequestCommand::FundingRates(cmd) => Some(RequestCatalogKey::new(
691            NautilusDataType::FundingRateUpdate,
692            Some(cmd.instrument_id.to_string()),
693        )),
694        RequestCommand::Bars(cmd) => Some(RequestCatalogKey::new(
695            NautilusDataType::Bar,
696            Some(cmd.bar_type.to_string()),
697        )),
698        RequestCommand::BookDeltas(cmd) => Some(RequestCatalogKey::new(
699            NautilusDataType::OrderBookDelta,
700            Some(cmd.instrument_id.to_string()),
701        )),
702        RequestCommand::BookDepth(cmd) => Some(RequestCatalogKey::new(
703            NautilusDataType::OrderBookDepth,
704            Some(cmd.instrument_id.to_string()),
705        )),
706        _ => None,
707    }
708}
709
710impl RequestCatalogKey {
711    fn new(data_type: NautilusDataType, identifier: Option<String>) -> Self {
712        Self {
713            data_type,
714            identifier,
715        }
716    }
717
718    fn catalog_data_type(&self) -> NautilusDataType {
719        self.data_type.clone()
720    }
721}
722
723fn catalog_missing_intervals(
724    catalog: &mut DataCatalog,
725    start: u64,
726    end: u64,
727    key: &RequestCatalogKey,
728) -> anyhow::Result<Vec<(u64, u64)>> {
729    catalog.get_missing_intervals_for_request(
730        UnixNanos::from(start),
731        UnixNanos::from(end),
732        key.catalog_data_type(),
733        key.identifier.as_deref(),
734    )
735}
736
737fn quote_ticks_from_query_result(data: DataBatch) -> anyhow::Result<Vec<QuoteTick>> {
738    QuoteTick::from_batch(data)
739}
740
741fn trade_ticks_from_query_result(data: DataBatch) -> anyhow::Result<Vec<TradeTick>> {
742    TradeTick::from_batch(data)
743}
744
745fn funding_rates_from_query_result(data: DataBatch) -> anyhow::Result<Vec<FundingRateUpdate>> {
746    FundingRateUpdate::from_batch(data)
747}
748
749fn bars_from_query_result(data: DataBatch) -> anyhow::Result<Vec<Bar>> {
750    Bar::from_batch(data)
751}
752
753fn order_book_deltas_from_query_result(data: DataBatch) -> anyhow::Result<Vec<OrderBookDelta>> {
754    OrderBookDelta::from_batch(data)
755}
756
757fn order_book_depths_from_query_result(data: DataBatch) -> anyhow::Result<Vec<OrderBookDepth>> {
758    OrderBookDepth::from_batch(data)
759}
760
761fn request_start(req: &RequestCommand) -> Option<Timestamp> {
762    match req {
763        RequestCommand::Data(cmd) => cmd.start,
764        RequestCommand::Instrument(cmd) => cmd.start,
765        RequestCommand::Instruments(cmd) => cmd.start,
766        RequestCommand::Quotes(cmd) => cmd.start,
767        RequestCommand::Trades(cmd) => cmd.start,
768        RequestCommand::FundingRates(cmd) => cmd.start,
769        RequestCommand::Bars(cmd) => cmd.start,
770        RequestCommand::BookDeltas(cmd) => cmd.start,
771        RequestCommand::BookDepth(cmd) => cmd.start,
772        _ => None,
773    }
774}
775
776fn request_end(req: &RequestCommand) -> Option<Timestamp> {
777    match req {
778        RequestCommand::Data(cmd) => cmd.end,
779        RequestCommand::Instrument(cmd) => cmd.end,
780        RequestCommand::Instruments(cmd) => cmd.end,
781        RequestCommand::Quotes(cmd) => cmd.end,
782        RequestCommand::Trades(cmd) => cmd.end,
783        RequestCommand::FundingRates(cmd) => cmd.end,
784        RequestCommand::Bars(cmd) => cmd.end,
785        RequestCommand::BookDeltas(cmd) => cmd.end,
786        RequestCommand::BookDepth(cmd) => cmd.end,
787        _ => None,
788    }
789}
790
791fn request_filter_expr(params: Option<&Params>) -> Option<String> {
792    params
793        .and_then(|params| params.get_str("filter_expr"))
794        .filter(|filter_expr| !filter_expr.is_empty())
795        .map(ToString::to_string)
796}
797
798fn bound_request_dates(
799    start: Option<Timestamp>,
800    end: Option<Timestamp>,
801    now: Timestamp,
802    query_past_data: bool,
803) -> (Timestamp, Timestamp) {
804    let zero = Timestamp::UNIX_EPOCH;
805    let mut start = start.unwrap_or(zero);
806    let mut end = end.unwrap_or(now);
807
808    if query_past_data {
809        if start > now {
810            start = now;
811        }
812
813        if end > now {
814            end = now;
815        }
816    }
817
818    (start, end)
819}
820
821fn datetime_to_unix_nanos_or_zero(dt: Timestamp) -> UnixNanos {
822    UnixNanos::from(u64::try_from(dt.as_nanosecond().max(0)).unwrap_or(0))
823}
824
825fn floor_to_utc_day(dt: Timestamp) -> Timestamp {
826    let midnight = jiff::tz::Offset::UTC.to_datetime(dt).date().at(0, 0, 0, 0);
827    jiff::tz::Offset::UTC
828        .to_timestamp(midnight)
829        .expect("midnight UTC is always valid")
830}
831
832fn with_dates_for_pipeline(
833    req: &RequestCommand,
834    start: Option<Timestamp>,
835    end: Option<Timestamp>,
836    ts_init: UnixNanos,
837) -> RequestCommand {
838    let new_id = UUID4::new();
839
840    match req {
841        RequestCommand::Quotes(cmd) => RequestCommand::Quotes(RequestQuotes {
842            instrument_id: cmd.instrument_id,
843            start,
844            end,
845            limit: cmd.limit,
846            client_id: cmd.client_id,
847            request_id: new_id,
848            ts_init,
849            params: cmd.params.clone(),
850        }),
851        RequestCommand::Trades(cmd) => RequestCommand::Trades(RequestTrades {
852            instrument_id: cmd.instrument_id,
853            start,
854            end,
855            limit: cmd.limit,
856            client_id: cmd.client_id,
857            request_id: new_id,
858            ts_init,
859            params: cmd.params.clone(),
860        }),
861        RequestCommand::FundingRates(cmd) => RequestCommand::FundingRates(RequestFundingRates {
862            instrument_id: cmd.instrument_id,
863            start,
864            end,
865            limit: cmd.limit,
866            client_id: cmd.client_id,
867            request_id: new_id,
868            ts_init,
869            params: cmd.params.clone(),
870        }),
871        RequestCommand::BookDeltas(cmd) => RequestCommand::BookDeltas(RequestBookDeltas {
872            instrument_id: cmd.instrument_id,
873            start,
874            end,
875            limit: cmd.limit,
876            client_id: cmd.client_id,
877            request_id: new_id,
878            ts_init,
879            params: cmd.params.clone(),
880        }),
881        RequestCommand::BookDepth(cmd) => RequestCommand::BookDepth(RequestBookDepth {
882            instrument_id: cmd.instrument_id,
883            start,
884            end,
885            limit: cmd.limit,
886            depth: cmd.depth,
887            client_id: cmd.client_id,
888            request_id: new_id,
889            ts_init,
890            params: cmd.params.clone(),
891        }),
892        RequestCommand::Data(cmd) => RequestCommand::Data(RequestCustomData {
893            client_id: cmd.client_id,
894            data_type: cmd.data_type.clone(),
895            start,
896            end,
897            limit: cmd.limit,
898            request_id: new_id,
899            ts_init,
900            params: cmd.params.clone(),
901        }),
902        RequestCommand::Bars(cmd) => RequestCommand::Bars(RequestBars {
903            bar_type: cmd.bar_type,
904            start,
905            end,
906            limit: cmd.limit,
907            client_id: cmd.client_id,
908            request_id: new_id,
909            ts_init,
910            params: cmd.params.clone(),
911        }),
912        // `Join` and the non-date-range variants should never reach this path; the dispatcher
913        // gates on `is_date_range_variant` first. Cloning preserves behavior if a caller
914        // reaches this arm.
915        _ => req.clone(),
916    }
917}
918
919fn build_empty_response(
920    req: &RequestCommand,
921    start: UnixNanos,
922    end: UnixNanos,
923    used_client_id: Option<ClientId>,
924    ts_init: UnixNanos,
925) -> anyhow::Result<DataResponse> {
926    let response = match req {
927        RequestCommand::Data(cmd) => DataResponse::Data(CustomDataResponse::new(
928            cmd.request_id,
929            cmd.client_id,
930            None,
931            cmd.data_type.clone(),
932            Vec::<CustomData>::new(),
933            Some(start),
934            Some(end),
935            ts_init,
936            cmd.params.clone(),
937        )),
938        RequestCommand::Quotes(cmd) => DataResponse::Quotes(QuotesResponse::new(
939            cmd.request_id,
940            resolve_response_client_id(cmd.client_id, used_client_id),
941            cmd.instrument_id,
942            Vec::new(),
943            Some(start),
944            Some(end),
945            ts_init,
946            cmd.params.clone(),
947        )),
948        RequestCommand::Trades(cmd) => DataResponse::Trades(TradesResponse::new(
949            cmd.request_id,
950            resolve_response_client_id(cmd.client_id, used_client_id),
951            cmd.instrument_id,
952            Vec::new(),
953            Some(start),
954            Some(end),
955            ts_init,
956            cmd.params.clone(),
957        )),
958        RequestCommand::FundingRates(cmd) => DataResponse::FundingRates(FundingRatesResponse::new(
959            cmd.request_id,
960            resolve_response_client_id(cmd.client_id, used_client_id),
961            cmd.instrument_id,
962            Vec::new(),
963            Some(start),
964            Some(end),
965            ts_init,
966            cmd.params.clone(),
967        )),
968        RequestCommand::Bars(cmd) => DataResponse::Bars(BarsResponse::new(
969            cmd.request_id,
970            resolve_response_client_id(cmd.client_id, used_client_id),
971            cmd.bar_type,
972            Vec::new(),
973            Some(start),
974            Some(end),
975            ts_init,
976            cmd.params.clone(),
977        )),
978        RequestCommand::BookDeltas(cmd) => DataResponse::BookDeltas(BookDeltasResponse::new(
979            cmd.request_id,
980            resolve_response_client_id(cmd.client_id, used_client_id),
981            cmd.instrument_id,
982            Vec::new(),
983            Some(start),
984            Some(end),
985            ts_init,
986            cmd.params.clone(),
987        )),
988        RequestCommand::BookDepth(cmd) => DataResponse::BookDepth(BookDepthResponse::new(
989            cmd.request_id,
990            resolve_response_client_id(cmd.client_id, used_client_id),
991            cmd.instrument_id,
992            Vec::new(),
993            Some(start),
994            Some(end),
995            ts_init,
996            cmd.params.clone(),
997        )),
998        _ => {
999            anyhow::bail!("Cannot build empty catalog response for non-catalog-eligible request")
1000        }
1001    };
1002
1003    Ok(response)
1004}
1005
1006fn build_quotes_catalog_response(
1007    cmd: &RequestQuotes,
1008    data: Vec<QuoteTick>,
1009    start: UnixNanos,
1010    end: UnixNanos,
1011    used_client_id: Option<ClientId>,
1012    ts_init: UnixNanos,
1013) -> DataResponse {
1014    let params = catalog_response_params(cmd.params.as_ref());
1015    DataResponse::Quotes(QuotesResponse::new(
1016        cmd.request_id,
1017        resolve_response_client_id(cmd.client_id, used_client_id),
1018        cmd.instrument_id,
1019        data,
1020        Some(start),
1021        Some(end),
1022        ts_init,
1023        Some(params),
1024    ))
1025}
1026
1027fn build_trades_catalog_response(
1028    cmd: &RequestTrades,
1029    data: Vec<TradeTick>,
1030    start: UnixNanos,
1031    end: UnixNanos,
1032    used_client_id: Option<ClientId>,
1033    ts_init: UnixNanos,
1034) -> DataResponse {
1035    let params = catalog_response_params(cmd.params.as_ref());
1036    DataResponse::Trades(TradesResponse::new(
1037        cmd.request_id,
1038        resolve_response_client_id(cmd.client_id, used_client_id),
1039        cmd.instrument_id,
1040        data,
1041        Some(start),
1042        Some(end),
1043        ts_init,
1044        Some(params),
1045    ))
1046}
1047
1048fn build_funding_rates_catalog_response(
1049    cmd: &RequestFundingRates,
1050    data: Vec<FundingRateUpdate>,
1051    start: UnixNanos,
1052    end: UnixNanos,
1053    used_client_id: Option<ClientId>,
1054    ts_init: UnixNanos,
1055) -> DataResponse {
1056    let params = catalog_response_params(cmd.params.as_ref());
1057    DataResponse::FundingRates(FundingRatesResponse::new(
1058        cmd.request_id,
1059        resolve_response_client_id(cmd.client_id, used_client_id),
1060        cmd.instrument_id,
1061        data,
1062        Some(start),
1063        Some(end),
1064        ts_init,
1065        Some(params),
1066    ))
1067}
1068
1069fn build_bars_catalog_response(
1070    cmd: &RequestBars,
1071    data: Vec<Bar>,
1072    start: UnixNanos,
1073    end: UnixNanos,
1074    used_client_id: Option<ClientId>,
1075    ts_init: UnixNanos,
1076) -> DataResponse {
1077    let params = catalog_response_params(cmd.params.as_ref());
1078    DataResponse::Bars(BarsResponse::new(
1079        cmd.request_id,
1080        resolve_response_client_id(cmd.client_id, used_client_id),
1081        cmd.bar_type,
1082        data,
1083        Some(start),
1084        Some(end),
1085        ts_init,
1086        Some(params),
1087    ))
1088}
1089
1090fn build_custom_data_catalog_response(
1091    cmd: &RequestCustomData,
1092    data: Vec<CustomData>,
1093    start: UnixNanos,
1094    end: UnixNanos,
1095    ts_init: UnixNanos,
1096) -> DataResponse {
1097    let params = catalog_response_params(cmd.params.as_ref());
1098    DataResponse::Data(CustomDataResponse::new(
1099        cmd.request_id,
1100        cmd.client_id,
1101        None,
1102        cmd.data_type.clone(),
1103        data,
1104        Some(start),
1105        Some(end),
1106        ts_init,
1107        Some(params),
1108    ))
1109}
1110
1111fn build_book_deltas_catalog_response(
1112    cmd: &RequestBookDeltas,
1113    data: Vec<OrderBookDelta>,
1114    start: UnixNanos,
1115    end: UnixNanos,
1116    used_client_id: Option<ClientId>,
1117    ts_init: UnixNanos,
1118) -> DataResponse {
1119    let params = catalog_response_params(cmd.params.as_ref());
1120    DataResponse::BookDeltas(BookDeltasResponse::new(
1121        cmd.request_id,
1122        resolve_response_client_id(cmd.client_id, used_client_id),
1123        cmd.instrument_id,
1124        data,
1125        Some(start),
1126        Some(end),
1127        ts_init,
1128        Some(params),
1129    ))
1130}
1131
1132fn build_book_depth_catalog_response(
1133    cmd: &RequestBookDepth,
1134    data: Vec<OrderBookDepth>,
1135    start: UnixNanos,
1136    end: UnixNanos,
1137    used_client_id: Option<ClientId>,
1138    ts_init: UnixNanos,
1139) -> DataResponse {
1140    let params = catalog_response_params(cmd.params.as_ref());
1141    DataResponse::BookDepth(BookDepthResponse::new(
1142        cmd.request_id,
1143        resolve_response_client_id(cmd.client_id, used_client_id),
1144        cmd.instrument_id,
1145        data,
1146        Some(start),
1147        Some(end),
1148        ts_init,
1149        Some(params),
1150    ))
1151}
1152
1153fn catalog_response_params(existing: Option<&Params>) -> Params {
1154    let mut params = existing.cloned().unwrap_or_else(Params::new);
1155    params.insert(PARAM_UPDATE_CATALOG.to_string(), Value::Bool(false));
1156    params
1157}
1158
1159fn custom_data_from_query_result(data: DataBatch) -> anyhow::Result<Vec<CustomData>> {
1160    CustomData::from_batch(data)
1161}
1162
1163fn instrument_only_last(params: Option<&Params>) -> bool {
1164    params
1165        .and_then(|params| params.get_bool("only_last"))
1166        .unwrap_or(true)
1167}
1168
1169fn latest_instruments(data: Vec<InstrumentAny>) -> Vec<InstrumentAny> {
1170    let mut instruments: AHashMap<_, InstrumentAny> = AHashMap::new();
1171
1172    for instrument in data {
1173        let id = instrument.id();
1174        match instruments.get(&id) {
1175            Some(existing) if existing.ts_init() >= instrument.ts_init() => {}
1176            _ => {
1177                instruments.insert(id, instrument);
1178            }
1179        }
1180    }
1181
1182    let mut data: Vec<_> = instruments.into_values().collect();
1183    data.sort_by_key(|instrument| instrument.id().to_string());
1184    data
1185}
1186
1187fn instrument_response_venue(request_venue: Option<Venue>, data: &[InstrumentAny]) -> Venue {
1188    request_venue.unwrap_or_else(|| {
1189        data.iter()
1190            .map(Instrument::venue)
1191            .min_by_key(std::string::ToString::to_string)
1192            .unwrap_or_else(|| Venue::from(CATALOG_CLIENT_ID))
1193    })
1194}
1195
1196fn resolve_response_client_id(
1197    request_client_id: Option<ClientId>,
1198    used_client_id: Option<ClientId>,
1199) -> ClientId {
1200    request_client_id
1201        .or(used_client_id)
1202        .unwrap_or_else(|| ClientId::new(CATALOG_CLIENT_ID))
1203}
1204
1205#[cfg(test)]
1206mod tests {
1207    use nautilus_common::messages::data::RequestJoin;
1208    use rstest::rstest;
1209
1210    use super::*;
1211
1212    #[rstest]
1213    fn test_build_empty_response_rejects_non_catalog_variant() {
1214        let request = RequestCommand::Join(RequestJoin::new(
1215            vec![UUID4::new()],
1216            None,
1217            None,
1218            UUID4::new(),
1219            UnixNanos::default(),
1220            None,
1221            None,
1222        ));
1223
1224        let result = build_empty_response(
1225            &request,
1226            UnixNanos::from(1u64),
1227            UnixNanos::from(2u64),
1228            None,
1229            UnixNanos::from(3u64),
1230        );
1231
1232        assert_eq!(
1233            result.unwrap_err().to_string(),
1234            "Cannot build empty catalog response for non-catalog-eligible request"
1235        );
1236    }
1237}
nautilus_data/engine/streaming.rs

nautilus_data/engine/
streaming.rs
