nautilus_common/serialization/
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::{borrow::Cow, sync::Arc};
17
18use bytes::Bytes;
19use nautilus_core::{Params, UUID4, UnixNanos};
20use nautilus_model::{
21    data::{BookOrder, CustomData, Data, DataType},
22    enums::{BookType, OrderSide, RecordFlag},
23    identifiers::{ClientId, InstrumentId, Venue},
24    orderbook::OrderBook,
25};
26use serde::{Deserialize, Deserializer, Serialize, Serializer};
27
28use crate::messages::data::{
29    BarsResponse, BookDeltasResponse, BookDepthResponse, BookResponse, CustomDataResponse,
30    DataResponse, FundingRatesResponse, InstrumentResponse, InstrumentsResponse,
31    OptionChainReferencePriceResponse, Payload, QuotesResponse, TradesResponse,
32};
33
34impl Serialize for DataResponse {
35    fn serialize<S: Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
36        WireResponse::from_native(self)
37            .map_err(serde::ser::Error::custom)?
38            .serialize(serializer)
39    }
40}
41
42impl<'de> Deserialize<'de> for DataResponse {
43    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
44        WireResponse::deserialize(deserializer)?
45            .into_native()
46            .map_err(serde::de::Error::custom)
47    }
48}
49
50#[derive(Serialize, Deserialize)]
51#[allow(
52    clippy::large_enum_variant,
53    reason = "model features vary enum sizes; transient serialization views avoid an extra allocation"
54)]
55enum WireResponse<'a> {
56    Data(WireCustomResponse<'a>),
57    Book(WireBookResponse<'a>),
58    Instrument(Cow<'a, InstrumentResponse>),
59    Instruments(Cow<'a, InstrumentsResponse>),
60    BookDeltas(Cow<'a, BookDeltasResponse>),
61    BookDepth(Cow<'a, BookDepthResponse>),
62    Quotes(Cow<'a, QuotesResponse>),
63    Trades(Cow<'a, TradesResponse>),
64    FundingRates(Cow<'a, FundingRatesResponse>),
65    OptionChainReferencePrice(Cow<'a, OptionChainReferencePriceResponse>),
66    Bars(Cow<'a, BarsResponse>),
67}
68
69impl<'a> WireResponse<'a> {
70    fn from_native(value: &'a DataResponse) -> anyhow::Result<Self> {
71        Ok(match value {
72            DataResponse::Data(value) => Self::Data(WireCustomResponse::from_native(value)?),
73            DataResponse::Book(value) => Self::Book(WireBookResponse::from_native(value)),
74            DataResponse::Instrument(value) => Self::Instrument(Cow::Borrowed(value)),
75            DataResponse::Instruments(value) => Self::Instruments(Cow::Borrowed(value)),
76            DataResponse::BookDeltas(value) => Self::BookDeltas(Cow::Borrowed(value)),
77            DataResponse::BookDepth(value) => Self::BookDepth(Cow::Borrowed(value)),
78            DataResponse::Quotes(value) => Self::Quotes(Cow::Borrowed(value)),
79            DataResponse::Trades(value) => Self::Trades(Cow::Borrowed(value)),
80            DataResponse::FundingRates(value) => Self::FundingRates(Cow::Borrowed(value)),
81            DataResponse::OptionChainReferencePrice(value) => {
82                Self::OptionChainReferencePrice(Cow::Borrowed(value))
83            }
84            DataResponse::Bars(value) => Self::Bars(Cow::Borrowed(value)),
85        })
86    }
87
88    fn into_native(self) -> anyhow::Result<DataResponse> {
89        Ok(match self {
90            Self::Data(value) => DataResponse::Data(value.into_native()?),
91            Self::Book(value) => DataResponse::Book(value.into_native()?),
92            Self::Instrument(value) => DataResponse::Instrument(Box::new(value.into_owned())),
93            Self::Instruments(value) => DataResponse::Instruments(value.into_owned()),
94            Self::BookDeltas(value) => DataResponse::BookDeltas(value.into_owned()),
95            Self::BookDepth(value) => DataResponse::BookDepth(value.into_owned()),
96            Self::Quotes(value) => DataResponse::Quotes(value.into_owned()),
97            Self::Trades(value) => DataResponse::Trades(value.into_owned()),
98            Self::FundingRates(value) => DataResponse::FundingRates(value.into_owned()),
99            Self::OptionChainReferencePrice(value) => {
100                DataResponse::OptionChainReferencePrice(value.into_owned())
101            }
102            Self::Bars(value) => DataResponse::Bars(value.into_owned()),
103        })
104    }
105}
106
107#[derive(Serialize, Deserialize)]
108struct WireCustomResponse<'a> {
109    correlation_id: UUID4,
110    client_id: ClientId,
111    venue: Option<Venue>,
112    data_type: Cow<'a, DataType>,
113    data: CustomPayload<'a>,
114    start: Option<UnixNanos>,
115    end: Option<UnixNanos>,
116    ts_init: UnixNanos,
117    params: Cow<'a, Option<Params>>,
118}
119
120impl<'a> WireCustomResponse<'a> {
121    fn from_native(value: &'a CustomDataResponse) -> anyhow::Result<Self> {
122        Ok(Self {
123            correlation_id: value.correlation_id,
124            client_id: value.client_id,
125            venue: value.venue,
126            data_type: Cow::Borrowed(&value.data_type),
127            data: CustomPayload::from_native(value.data.as_ref())?,
128            start: value.start,
129            end: value.end,
130            ts_init: value.ts_init,
131            params: Cow::Borrowed(&value.params),
132        })
133    }
134
135    fn into_native(self) -> anyhow::Result<CustomDataResponse> {
136        Ok(CustomDataResponse {
137            correlation_id: self.correlation_id,
138            client_id: self.client_id,
139            venue: self.venue,
140            data_type: self.data_type.into_owned(),
141            data: self.data.into_native()?,
142            start: self.start,
143            end: self.end,
144            ts_init: self.ts_init,
145            params: self.params.into_owned(),
146        })
147    }
148}
149
150#[derive(Serialize, Deserialize)]
151#[expect(
152    clippy::large_enum_variant,
153    reason = "transient serialization views avoid an extra allocation for scalar payloads"
154)]
155pub(super) enum CustomPayload<'a> {
156    Scalar(CustomItem<'a>),
157    Batch(Vec<CustomItem<'a>>),
158    Bytes(Cow<'a, [u8]>),
159    Buffer(Cow<'a, Bytes>),
160    Text(Cow<'a, str>),
161}
162
163impl<'a> CustomPayload<'a> {
164    pub(super) fn from_native(value: &'a dyn std::any::Any) -> anyhow::Result<Self> {
165        if let Some(value) = value.downcast_ref::<CustomData>() {
166            return Ok(Self::Scalar(CustomItem::from_native(value)));
167        }
168
169        if let Some(value) = value.downcast_ref::<Vec<CustomData>>() {
170            return Ok(Self::Batch(
171                value.iter().map(CustomItem::from_native).collect(),
172            ));
173        }
174
175        if let Some(value) = value.downcast_ref::<Vec<u8>>() {
176            return Ok(Self::Bytes(Cow::Borrowed(value)));
177        }
178
179        if let Some(value) = value.downcast_ref::<Bytes>() {
180            return Ok(Self::Buffer(Cow::Borrowed(value)));
181        }
182
183        if let Some(value) = value.downcast_ref::<String>() {
184            return Ok(Self::Text(Cow::Borrowed(value)));
185        }
186
187        anyhow::bail!(
188            "custom response payload requires CustomData, Vec<CustomData>, Vec<u8>, Bytes, or String"
189        )
190    }
191
192    pub(super) fn into_native(self) -> anyhow::Result<Payload> {
193        Ok(match self {
194            Self::Scalar(value) => Arc::new(value.into_native()?),
195            Self::Batch(values) => Arc::new(
196                values
197                    .into_iter()
198                    .map(CustomItem::into_native)
199                    .collect::<anyhow::Result<Vec<_>>>()?,
200            ),
201            Self::Bytes(value) => Arc::new(value.into_owned()),
202            Self::Buffer(value) => Arc::new(value.into_owned()),
203            Self::Text(value) => Arc::new(value.into_owned()),
204        })
205    }
206}
207
208#[derive(Serialize, Deserialize)]
209pub(super) struct CustomItem<'a> {
210    pub(super) data_type: Cow<'a, DataType>,
211    pub(super) value: CustomValue<'a>,
212}
213
214impl<'a> CustomItem<'a> {
215    pub(super) fn from_native(value: &'a CustomData) -> Self {
216        Self {
217            data_type: Cow::Borrowed(&value.data_type),
218            value: CustomValue::Borrowed(value),
219        }
220    }
221
222    fn into_native(self) -> anyhow::Result<CustomData> {
223        let mut value = match self.value {
224            CustomValue::Borrowed(value) => value.clone(),
225            CustomValue::Owned(Data::Custom(value)) => value,
226            CustomValue::Owned(_) => {
227                anyhow::bail!("custom response contains a non-custom payload")
228            }
229        };
230
231        value.data_type = self.data_type.into_owned();
232        Ok(value)
233    }
234}
235
236#[derive(Serialize)]
237#[serde(untagged)]
238#[allow(
239    clippy::large_enum_variant,
240    reason = "model features vary enum sizes; owned decoding avoids an extra allocation"
241)]
242pub(super) enum CustomValue<'a> {
243    Borrowed(&'a CustomData),
244    Owned(Data),
245}
246
247impl<'de> Deserialize<'de> for CustomValue<'_> {
248    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
249        Data::deserialize(deserializer).map(Self::Owned)
250    }
251}
252
253#[derive(Serialize, Deserialize)]
254struct WireBookResponse<'a> {
255    correlation_id: UUID4,
256    client_id: ClientId,
257    instrument_id: InstrumentId,
258    data: BookState,
259    start: Option<UnixNanos>,
260    end: Option<UnixNanos>,
261    ts_init: UnixNanos,
262    params: Cow<'a, Option<Params>>,
263}
264
265impl<'a> WireBookResponse<'a> {
266    fn from_native(value: &'a BookResponse) -> Self {
267        let book = &value.data;
268
269        Self {
270            correlation_id: value.correlation_id,
271            client_id: value.client_id,
272            instrument_id: value.instrument_id,
273            data: BookState {
274                instrument_id: book.instrument_id,
275                book_type: book.book_type,
276                sequence: book.sequence,
277                ts_last: book.ts_last,
278                update_count: book.update_count,
279                orders: book
280                    .bids(None)
281                    .chain(book.asks(None))
282                    .flat_map(|level| level.iter().copied())
283                    .collect(),
284                bid_batch_state: book.batch_state_code(OrderSide::Buy),
285                ask_batch_state: book.batch_state_code(OrderSide::Sell),
286            },
287            start: value.start,
288            end: value.end,
289            ts_init: value.ts_init,
290            params: Cow::Borrowed(&value.params),
291        }
292    }
293
294    fn into_native(self) -> anyhow::Result<BookResponse> {
295        let mut book = OrderBook::new(self.data.instrument_id, self.data.book_type);
296
297        for order in self.data.orders {
298            anyhow::ensure!(order.side.is_some(), "book snapshot order has no side");
299            book.add(
300                order,
301                RecordFlag::F_SNAPSHOT as u8 | RecordFlag::F_LAST as u8,
302                self.data.sequence,
303                self.data.ts_last,
304            );
305        }
306
307        book.set_batch_state_code(OrderSide::Buy, self.data.bid_batch_state)?;
308        book.set_batch_state_code(OrderSide::Sell, self.data.ask_batch_state)?;
309        book.sequence = self.data.sequence;
310        book.ts_last = self.data.ts_last;
311        book.update_count = self.data.update_count;
312        Ok(BookResponse {
313            correlation_id: self.correlation_id,
314            client_id: self.client_id,
315            instrument_id: self.instrument_id,
316            data: book,
317            start: self.start,
318            end: self.end,
319            ts_init: self.ts_init,
320            params: self.params.into_owned(),
321        })
322    }
323}
324
325#[derive(Serialize, Deserialize)]
326struct BookState {
327    instrument_id: InstrumentId,
328    book_type: BookType,
329    sequence: u64,
330    ts_last: UnixNanos,
331    update_count: u64,
332    orders: Vec<BookOrder>,
333    #[serde(default)]
334    bid_batch_state: u8,
335    #[serde(default)]
336    ask_batch_state: u8,
337}
nautilus_common/serialization/response.rs

nautilus_common/serialization/
response.rs
