nautilus_common/messages/data/
response.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 std::{any::Any, ops::Range, sync::Arc};
17
18use nautilus_core::{Params, UUID4, UnixNanos};
19use nautilus_model::{
20    data::{
21        Bar, BarType, BatchView, DataType, FundingRateUpdate, HasTsInit, OrderBookDelta,
22        OrderBookDepth, QuoteTick, TradeTick,
23    },
24    identifiers::{ClientId, InstrumentId, OptionSeriesId, Venue},
25    instruments::InstrumentAny,
26    orderbook::OrderBook,
27    types::Price,
28};
29use serde::{Deserialize, Serialize};
30
31use super::{
32    IntoInstrumentIds, Payload,
33    identifiers::{deserialize_instrument_ids, single_instrument_id},
34};
35
36// Bounds are inclusive on ts_init; trimming leaves interior rows unchanged
37pub(crate) fn trim_data_to_bounds<T: HasTsInit>(
38    data: &mut Vec<T>,
39    start: Option<UnixNanos>,
40    end: Option<UnixNanos>,
41) {
42    let range = range_for_bounds(data, start, end);
43    data.drain(..range.start);
44    data.truncate(range.len());
45}
46
47pub(crate) fn trim_batch_to_bounds<T: HasTsInit>(
48    data: &mut BatchView<T>,
49    start: Option<UnixNanos>,
50    end: Option<UnixNanos>,
51) {
52    let range = range_for_bounds(data, start, end);
53    if range != (0..data.len()) {
54        *data = data.slice(range.start, range.end);
55    }
56}
57
58fn range_for_bounds<T: HasTsInit>(
59    data: &[T],
60    start: Option<UnixNanos>,
61    end: Option<UnixNanos>,
62) -> Range<usize> {
63    let first = start.map_or(0, |start| {
64        data.iter()
65            .position(|item| item.ts_init() >= start)
66            .unwrap_or(data.len())
67    });
68
69    let end = end.map_or(data.len(), |end| {
70        data.iter()
71            .rposition(|item| item.ts_init() <= end)
72            .map_or(0, |index| index + 1)
73    });
74
75    first..end.max(first)
76}
77
78#[derive(Clone, Debug)]
79pub struct CustomDataResponse {
80    pub correlation_id: UUID4,
81    pub client_id: ClientId,
82    pub venue: Option<Venue>,
83    pub data_type: DataType,
84    pub data: Payload,
85    pub start: Option<UnixNanos>,
86    pub end: Option<UnixNanos>,
87    pub ts_init: UnixNanos,
88    pub params: Option<Params>,
89}
90
91impl CustomDataResponse {
92    /// Creates a new [`CustomDataResponse`] instance.
93    #[expect(clippy::too_many_arguments)]
94    pub fn new<T: Any + Send + Sync>(
95        correlation_id: UUID4,
96        client_id: ClientId,
97        venue: Option<Venue>,
98        data_type: DataType,
99        data: T,
100        start: Option<UnixNanos>,
101        end: Option<UnixNanos>,
102        ts_init: UnixNanos,
103        params: Option<Params>,
104    ) -> Self {
105        Self {
106            correlation_id,
107            client_id,
108            venue,
109            data_type,
110            data: Arc::new(data),
111            start,
112            end,
113            ts_init,
114            params,
115        }
116    }
117
118    /// Converts the response to a dyn Any trait object for messaging.
119    pub fn as_any(&self) -> &dyn Any {
120        self
121    }
122}
123
124#[derive(Clone, Debug, Serialize, Deserialize)]
125pub struct InstrumentResponse {
126    pub correlation_id: UUID4,
127    pub client_id: ClientId,
128    pub instrument_id: InstrumentId,
129    pub data: InstrumentAny,
130    pub start: Option<UnixNanos>,
131    pub end: Option<UnixNanos>,
132    pub ts_init: UnixNanos,
133    pub params: Option<Params>,
134}
135
136impl InstrumentResponse {
137    /// Converts to a dyn Any trait object for messaging.
138    pub fn as_any(&self) -> &dyn Any {
139        self
140    }
141
142    /// Creates a new [`InstrumentResponse`] instance.
143    #[expect(clippy::too_many_arguments)]
144    pub fn new(
145        correlation_id: UUID4,
146        client_id: ClientId,
147        instrument_id: InstrumentId,
148        data: InstrumentAny,
149        start: Option<UnixNanos>,
150        end: Option<UnixNanos>,
151        ts_init: UnixNanos,
152        params: Option<Params>,
153    ) -> Self {
154        Self {
155            correlation_id,
156            client_id,
157            instrument_id,
158            data,
159            start,
160            end,
161            ts_init,
162            params,
163        }
164    }
165}
166
167#[derive(Clone, Debug, Serialize, Deserialize)]
168pub struct InstrumentsResponse {
169    pub correlation_id: UUID4,
170    pub client_id: ClientId,
171    pub venue: Venue,
172    pub data: Vec<InstrumentAny>,
173    pub start: Option<UnixNanos>,
174    pub end: Option<UnixNanos>,
175    pub ts_init: UnixNanos,
176    pub params: Option<Params>,
177}
178
179impl InstrumentsResponse {
180    /// Converts to a dyn Any trait object for messaging.
181    pub fn as_any(&self) -> &dyn Any {
182        self
183    }
184
185    /// Creates a new [`InstrumentsResponse`] instance.
186    #[expect(clippy::too_many_arguments)]
187    pub fn new(
188        correlation_id: UUID4,
189        client_id: ClientId,
190        venue: Venue,
191        data: Vec<InstrumentAny>,
192        start: Option<UnixNanos>,
193        end: Option<UnixNanos>,
194        ts_init: UnixNanos,
195        params: Option<Params>,
196    ) -> Self {
197        Self {
198            correlation_id,
199            client_id,
200            venue,
201            data,
202            start,
203            end,
204            ts_init,
205            params,
206        }
207    }
208}
209
210#[derive(Clone, Debug)]
211pub struct BookResponse {
212    pub correlation_id: UUID4,
213    pub client_id: ClientId,
214    pub instrument_id: InstrumentId,
215    pub data: OrderBook,
216    pub start: Option<UnixNanos>,
217    pub end: Option<UnixNanos>,
218    pub ts_init: UnixNanos,
219    pub params: Option<Params>,
220}
221
222impl BookResponse {
223    /// Converts to a dyn Any trait object for messaging.
224    pub fn as_any(&self) -> &dyn Any {
225        self
226    }
227
228    /// Creates a new [`BookResponse`] instance.
229    #[expect(clippy::too_many_arguments)]
230    pub fn new(
231        correlation_id: UUID4,
232        client_id: ClientId,
233        instrument_id: InstrumentId,
234        data: OrderBook,
235        start: Option<UnixNanos>,
236        end: Option<UnixNanos>,
237        ts_init: UnixNanos,
238        params: Option<Params>,
239    ) -> Self {
240        Self {
241            correlation_id,
242            client_id,
243            instrument_id,
244            data,
245            start,
246            end,
247            ts_init,
248            params,
249        }
250    }
251}
252
253#[derive(Clone, Debug, Serialize, Deserialize)]
254pub struct BookDeltasResponse {
255    pub correlation_id: UUID4,
256    pub client_id: ClientId,
257    pub instrument_id: InstrumentId,
258    pub data: BatchView<OrderBookDelta>,
259    pub start: Option<UnixNanos>,
260    pub end: Option<UnixNanos>,
261    pub ts_init: UnixNanos,
262    pub params: Option<Params>,
263}
264
265impl BookDeltasResponse {
266    /// Converts to a dyn Any trait object for messaging.
267    pub fn as_any(&self) -> &dyn Any {
268        self
269    }
270
271    /// Creates a new [`BookDeltasResponse`] instance.
272    ///
273    /// Accepts an owned vector, a shared buffer, or a batch view.
274    #[expect(clippy::too_many_arguments)]
275    pub fn new(
276        correlation_id: UUID4,
277        client_id: ClientId,
278        instrument_id: InstrumentId,
279        data: impl Into<BatchView<OrderBookDelta>>,
280        start: Option<UnixNanos>,
281        end: Option<UnixNanos>,
282        ts_init: UnixNanos,
283        params: Option<Params>,
284    ) -> Self {
285        Self {
286            correlation_id,
287            client_id,
288            instrument_id,
289            data: data.into(),
290            start,
291            end,
292            ts_init,
293            params,
294        }
295    }
296}
297
298#[derive(Clone, Debug, Serialize, Deserialize)]
299pub struct BookDepthResponse {
300    pub correlation_id: UUID4,
301    pub client_id: ClientId,
302    pub instrument_id: InstrumentId,
303    pub data: BatchView<OrderBookDepth>,
304    pub start: Option<UnixNanos>,
305    pub end: Option<UnixNanos>,
306    pub ts_init: UnixNanos,
307    pub params: Option<Params>,
308}
309
310impl BookDepthResponse {
311    /// Converts to a dyn Any trait object for messaging.
312    pub fn as_any(&self) -> &dyn Any {
313        self
314    }
315
316    /// Creates a new [`BookDepthResponse`] instance.
317    ///
318    /// Accepts an owned vector, a shared buffer, or a batch view.
319    #[expect(clippy::too_many_arguments)]
320    pub fn new(
321        correlation_id: UUID4,
322        client_id: ClientId,
323        instrument_id: InstrumentId,
324        data: impl Into<BatchView<OrderBookDepth>>,
325        start: Option<UnixNanos>,
326        end: Option<UnixNanos>,
327        ts_init: UnixNanos,
328        params: Option<Params>,
329    ) -> Self {
330        Self {
331            correlation_id,
332            client_id,
333            instrument_id,
334            data: data.into(),
335            start,
336            end,
337            ts_init,
338            params,
339        }
340    }
341}
342
343#[derive(Clone, Debug, Serialize, Deserialize)]
344pub struct QuotesResponse {
345    pub correlation_id: UUID4,
346    pub client_id: ClientId,
347    #[serde(deserialize_with = "deserialize_instrument_ids")]
348    pub instrument_ids: Vec<InstrumentId>,
349    pub data: BatchView<QuoteTick>,
350    pub start: Option<UnixNanos>,
351    pub end: Option<UnixNanos>,
352    pub ts_init: UnixNanos,
353    pub params: Option<Params>,
354}
355
356impl QuotesResponse {
357    /// Converts to a dyn Any trait object for messaging.
358    pub fn as_any(&self) -> &dyn Any {
359        self
360    }
361
362    /// Creates a new [`QuotesResponse`] instance.
363    ///
364    /// Accepts an owned vector, a shared buffer, or a batch view.
365    ///
366    /// # Panics
367    ///
368    /// Panics if the identifier list is empty.
369    #[expect(clippy::too_many_arguments)]
370    pub fn new(
371        correlation_id: UUID4,
372        client_id: ClientId,
373        instrument_ids: impl IntoInstrumentIds,
374        data: impl Into<BatchView<QuoteTick>>,
375        start: Option<UnixNanos>,
376        end: Option<UnixNanos>,
377        ts_init: UnixNanos,
378        params: Option<Params>,
379    ) -> Self {
380        let instrument_ids = instrument_ids
381            .into_instrument_ids()
382            .expect("invalid instrument ID list");
383
384        Self {
385            correlation_id,
386            client_id,
387            instrument_ids,
388            data: data.into(),
389            start,
390            end,
391            ts_init,
392            params,
393        }
394    }
395
396    /// Returns the instrument ID for a single-member response path.
397    ///
398    /// # Errors
399    ///
400    /// Returns an error if the response does not contain exactly one member.
401    pub fn single_instrument_id(&self) -> anyhow::Result<InstrumentId> {
402        single_instrument_id(&self.instrument_ids, "response")
403    }
404}
405
406#[derive(Clone, Debug, Serialize, Deserialize)]
407pub struct TradesResponse {
408    pub correlation_id: UUID4,
409    pub client_id: ClientId,
410    #[serde(deserialize_with = "deserialize_instrument_ids")]
411    pub instrument_ids: Vec<InstrumentId>,
412    pub data: BatchView<TradeTick>,
413    pub start: Option<UnixNanos>,
414    pub end: Option<UnixNanos>,
415    pub ts_init: UnixNanos,
416    pub params: Option<Params>,
417}
418
419impl TradesResponse {
420    /// Converts to a dyn Any trait object for messaging.
421    pub fn as_any(&self) -> &dyn Any {
422        self
423    }
424
425    /// Creates a new [`TradesResponse`] instance.
426    ///
427    /// Accepts an owned vector, a shared buffer, or a batch view.
428    ///
429    /// # Panics
430    ///
431    /// Panics if the identifier list is empty.
432    #[expect(clippy::too_many_arguments)]
433    pub fn new(
434        correlation_id: UUID4,
435        client_id: ClientId,
436        instrument_ids: impl IntoInstrumentIds,
437        data: impl Into<BatchView<TradeTick>>,
438        start: Option<UnixNanos>,
439        end: Option<UnixNanos>,
440        ts_init: UnixNanos,
441        params: Option<Params>,
442    ) -> Self {
443        let instrument_ids = instrument_ids
444            .into_instrument_ids()
445            .expect("invalid instrument ID list");
446
447        Self {
448            correlation_id,
449            client_id,
450            instrument_ids,
451            data: data.into(),
452            start,
453            end,
454            ts_init,
455            params,
456        }
457    }
458
459    /// Returns the instrument ID for a single-member response path.
460    ///
461    /// # Errors
462    ///
463    /// Returns an error if the response does not contain exactly one member.
464    pub fn single_instrument_id(&self) -> anyhow::Result<InstrumentId> {
465        single_instrument_id(&self.instrument_ids, "response")
466    }
467}
468
469#[derive(Clone, Debug, Serialize, Deserialize)]
470pub struct FundingRatesResponse {
471    pub correlation_id: UUID4,
472    pub client_id: ClientId,
473    pub instrument_id: InstrumentId,
474    pub data: BatchView<FundingRateUpdate>,
475    pub start: Option<UnixNanos>,
476    pub end: Option<UnixNanos>,
477    pub ts_init: UnixNanos,
478    pub params: Option<Params>,
479}
480
481impl FundingRatesResponse {
482    /// Converts to a dyn Any trait object for messaging.
483    pub fn as_any(&self) -> &dyn Any {
484        self
485    }
486
487    /// Creates a new [`FundingRatesResponse`] instance.
488    ///
489    /// Accepts an owned vector, a shared buffer, or a batch view.
490    #[expect(clippy::too_many_arguments)]
491    pub fn new(
492        correlation_id: UUID4,
493        client_id: ClientId,
494        instrument_id: InstrumentId,
495        data: impl Into<BatchView<FundingRateUpdate>>,
496        start: Option<UnixNanos>,
497        end: Option<UnixNanos>,
498        ts_init: UnixNanos,
499        params: Option<Params>,
500    ) -> Self {
501        Self {
502            correlation_id,
503            client_id,
504            instrument_id,
505            data: data.into(),
506            start,
507            end,
508            ts_init,
509            params,
510        }
511    }
512}
513
514#[derive(Clone, Debug, Serialize, Deserialize)]
515pub struct OptionChainReferencePriceResponse {
516    pub correlation_id: UUID4,
517    pub client_id: ClientId,
518    pub series_id: OptionSeriesId,
519    pub price: Option<Price>,
520    pub ts_init: UnixNanos,
521    pub params: Option<Params>,
522}
523
524impl OptionChainReferencePriceResponse {
525    /// Creates a new [`OptionChainReferencePriceResponse`] instance.
526    pub fn new(
527        correlation_id: UUID4,
528        client_id: ClientId,
529        series_id: OptionSeriesId,
530        price: Option<Price>,
531        ts_init: UnixNanos,
532        params: Option<Params>,
533    ) -> Self {
534        Self {
535            correlation_id,
536            client_id,
537            series_id,
538            price,
539            ts_init,
540            params,
541        }
542    }
543}
544
545#[derive(Clone, Debug, Serialize, Deserialize)]
546pub struct BarsResponse {
547    pub correlation_id: UUID4,
548    pub client_id: ClientId,
549    pub bar_type: BarType,
550    pub data: BatchView<Bar>,
551    pub ts_init: UnixNanos,
552    pub start: Option<UnixNanos>,
553    pub end: Option<UnixNanos>,
554    pub params: Option<Params>,
555}
556
557impl BarsResponse {
558    /// Converts to a dyn Any trait object for messaging.
559    pub fn as_any(&self) -> &dyn Any {
560        self
561    }
562
563    /// Creates a new [`BarsResponse`] instance.
564    ///
565    /// Accepts an owned vector, a shared buffer, or a batch view.
566    #[expect(clippy::too_many_arguments)]
567    pub fn new(
568        correlation_id: UUID4,
569        client_id: ClientId,
570        bar_type: BarType,
571        data: impl Into<BatchView<Bar>>,
572        start: Option<UnixNanos>,
573        end: Option<UnixNanos>,
574        ts_init: UnixNanos,
575        params: Option<Params>,
576    ) -> Self {
577        Self {
578            correlation_id,
579            client_id,
580            bar_type,
581            data: data.into(),
582            ts_init,
583            start,
584            end,
585            params,
586        }
587    }
588}
589
590#[cfg(test)]
591mod tests {
592    use std::sync::Arc;
593
594    use nautilus_model::data::stubs::quote_audusd;
595    use rstest::{fixture, rstest};
596
597    use super::*;
598    use crate::messages::data::DataResponse;
599
600    #[fixture]
601    fn quote_rows() -> Vec<QuoteTick> {
602        let mut first = quote_audusd();
603        first.ts_event = UnixNanos::from(7);
604        first.ts_init = UnixNanos::from(11);
605        let mut second = first;
606        second.ts_event = UnixNanos::from(17);
607        second.ts_init = UnixNanos::from(23);
608        vec![first, second]
609    }
610
611    #[rstest]
612    fn test_quote_response_shares_supplied_buffer(quote_rows: Vec<QuoteTick>) {
613        let instrument_id = quote_rows[0].instrument_id;
614        let rows = Arc::new(quote_rows);
615        let request_id = UUID4::new();
616        let client_id = ClientId::from("HISTORY");
617        let start = Some(UnixNanos::from(5));
618        let end = Some(UnixNanos::from(29));
619        let ts_init = UnixNanos::from(31);
620        let params = Some(Params::new([("source".into(), "history".into())].into()));
621
622        let response = QuotesResponse::new(
623            request_id,
624            client_id,
625            instrument_id,
626            Arc::clone(&rows),
627            start,
628            end,
629            ts_init,
630            params.clone(),
631        );
632        let serialized = serde_json::to_value(&response).unwrap();
633
634        assert_eq!(response.correlation_id, request_id);
635        assert_eq!(response.client_id, client_id);
636        assert_eq!(response.instrument_ids, vec![instrument_id]);
637        assert_eq!(response.data.as_ref(), rows.as_slice());
638        assert!(Arc::ptr_eq(response.data.arc(), &rows));
639        assert_eq!(response.start, start);
640        assert_eq!(response.end, end);
641        assert_eq!(response.ts_init, ts_init);
642        assert_eq!(response.params, params);
643        assert_eq!(
644            serialized["data"],
645            serde_json::to_value(rows.as_ref()).unwrap()
646        );
647    }
648
649    #[rstest]
650    fn test_quote_response_vector_constructor_keeps_row_allocation(quote_rows: Vec<QuoteTick>) {
651        let instrument_id = quote_rows[0].instrument_id;
652        let rows = quote_rows.clone();
653        let row_ptr = rows.as_ptr();
654
655        let response = QuotesResponse::new(
656            UUID4::new(),
657            ClientId::from("HISTORY"),
658            instrument_id,
659            rows,
660            None,
661            None,
662            UnixNanos::from(31),
663            None,
664        );
665
666        assert_eq!(response.data.as_ref(), &quote_rows);
667        assert_eq!(response.data.as_ptr(), row_ptr);
668    }
669
670    #[rstest]
671    #[case::unbounded(None, None)]
672    #[case::exact(Some(11), Some(23))]
673    #[case::wider(Some(5), Some(29))]
674    fn test_shared_response_trim_keeps_existing_allocation(
675        quote_rows: Vec<QuoteTick>,
676        #[case] start: Option<u64>,
677        #[case] end: Option<u64>,
678    ) {
679        let instrument_id = quote_rows[0].instrument_id;
680        let rows = Arc::new(quote_rows);
681        let mut response = DataResponse::Quotes(QuotesResponse::new(
682            UUID4::new(),
683            ClientId::from("HISTORY"),
684            instrument_id,
685            Arc::clone(&rows),
686            start.map(UnixNanos::from),
687            end.map(UnixNanos::from),
688            UnixNanos::from(31),
689            None,
690        ));
691        response.trim_to_bounds();
692
693        let DataResponse::Quotes(response) = response else {
694            unreachable!()
695        };
696
697        assert_eq!(response.data.as_ref(), rows.as_slice());
698        assert!(Arc::ptr_eq(response.data.arc(), &rows));
699    }
700
701    #[rstest]
702    fn test_shared_response_trim_preserves_other_consumers(quote_rows: Vec<QuoteTick>) {
703        let instrument_id = quote_rows[0].instrument_id;
704        let rows = Arc::new(quote_rows.clone());
705        let mut response = DataResponse::Quotes(QuotesResponse::new(
706            UUID4::new(),
707            ClientId::from("HISTORY"),
708            instrument_id,
709            Arc::clone(&rows),
710            Some(UnixNanos::from(23)),
711            Some(UnixNanos::from(23)),
712            UnixNanos::from(31),
713            None,
714        ));
715        response.trim_to_bounds();
716
717        let DataResponse::Quotes(response) = response else {
718            unreachable!()
719        };
720
721        assert_eq!(rows.as_ref(), &quote_rows);
722        assert_eq!(response.data.as_ref(), &vec![quote_rows[1]]);
723        assert!(Arc::ptr_eq(response.data.arc(), &rows));
724        assert_eq!(response.data.range(), 1..2);
725    }
726
727    #[rstest]
728    #[case::unbounded(vec![11, 23], None, None, 0..2)]
729    #[case::inclusive(vec![11, 23], Some(11), Some(23), 0..2)]
730    #[case::first(vec![11, 23], None, Some(11), 0..1)]
731    #[case::last(vec![11, 23], Some(23), None, 1..2)]
732    #[case::after(vec![11, 23], Some(29), None, 2..2)]
733    #[case::before(vec![11, 23], None, Some(5), 0..0)]
734    #[case::reversed(vec![11, 23], Some(23), Some(11), 1..1)]
735    #[case::unsorted(vec![11, 37, 23], Some(11), Some(23), 0..3)]
736    #[case::empty(Vec::new(), Some(11), Some(23), 0..0)]
737    fn test_response_trim_preserves_narrowed_backing(
738        #[case] timestamps: Vec<u64>,
739        #[case] start: Option<u64>,
740        #[case] end: Option<u64>,
741        #[case] range: Range<usize>,
742    ) {
743        let visible: Vec<_> = timestamps
744            .into_iter()
745            .map(|ts| QuoteTick {
746                ts_init: UnixNanos::from(ts),
747                ..Default::default()
748            })
749            .collect();
750
751        let mut rows = vec![QuoteTick::default()];
752        rows.extend_from_slice(&visible);
753        rows.push(QuoteTick {
754            ts_init: UnixNanos::from(9999),
755            ..Default::default()
756        });
757
758        let rows = Arc::new(rows);
759        let mut view = BatchView::new(Arc::clone(&rows), 1..visible.len() + 1);
760
761        trim_batch_to_bounds(
762            &mut view,
763            start.map(UnixNanos::from),
764            end.map(UnixNanos::from),
765        );
766
767        assert_eq!(view.as_ref(), &visible[range.clone()]);
768        assert_eq!(view.range(), (1 + range.start)..(1 + range.end));
769        assert!(Arc::ptr_eq(view.arc(), &rows));
770        assert_eq!(&rows[1..rows.len() - 1], visible.as_slice());
771    }
772}
nautilus_common/messages/data/response.rs

nautilus_common/messages/data/
response.rs
