nautilus_serialization/arrow/display/
quote.rs

1// -------------------------------------------------------------------------------------------------
2//  Copyright (C) 2015-2026 Nautech Systems Pty Ltd. All rights reserved.
3//  https://nautechsystems.io
4//
5//  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
6//  You may not use this file except in compliance with the License.
7//  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
8//
9//  Unless required by applicable law or agreed to in writing, software
10//  distributed under the License is distributed on an "AS IS" BASIS,
11//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12//  See the License for the specific language governing permissions and
13//  limitations under the License.
14// -------------------------------------------------------------------------------------------------
15
16//! Display-mode Arrow encoder for [`QuoteTick`].
17
18use std::sync::Arc;
19
20use arrow::{
21    array::{Float64Builder, StringBuilder, TimestampNanosecondBuilder},
22    datatypes::Schema,
23    error::ArrowError,
24    record_batch::RecordBatch,
25};
26use nautilus_model::data::QuoteTick;
27
28use super::{
29    float64_field, price_to_f64, quantity_to_f64, timestamp_field, unix_nanos_to_i64, utf8_field,
30};
31use crate::arrow::timestamp_data_type;
32
33/// Returns the display-mode Arrow schema for [`QuoteTick`].
34#[must_use]
35pub fn quotes_schema() -> Schema {
36    Schema::new(vec![
37        utf8_field("instrument_id", false),
38        float64_field("bid_price", false),
39        float64_field("ask_price", false),
40        float64_field("bid_size", false),
41        float64_field("ask_size", false),
42        timestamp_field("ts_event", false),
43        timestamp_field("ts_init", false),
44    ])
45}
46
47/// Encodes quotes as a display-friendly Arrow [`RecordBatch`].
48///
49/// Emits `Float64` columns for prices and sizes, a `Utf8` `instrument_id`
50/// column, and `Timestamp(Nanosecond)` columns for event and init times.
51/// Mixed-instrument batches are supported. Precision is lost in the
52/// conversion to `f64`; use [`crate::arrow::quotes_to_arrow_record_batch_bytes`]
53/// for catalog storage.
54///
55/// Returns an empty [`RecordBatch`] with the correct schema when `data` is empty.
56///
57/// # Errors
58///
59/// Returns an [`ArrowError`] if the Arrow `RecordBatch` cannot be constructed.
60pub fn encode_quotes(data: &[QuoteTick]) -> Result<RecordBatch, ArrowError> {
61    let mut instrument_id_builder = StringBuilder::new();
62    let mut bid_price_builder = Float64Builder::with_capacity(data.len());
63    let mut ask_price_builder = Float64Builder::with_capacity(data.len());
64    let mut bid_size_builder = Float64Builder::with_capacity(data.len());
65    let mut ask_size_builder = Float64Builder::with_capacity(data.len());
66    let mut ts_event_builder =
67        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
68    let mut ts_init_builder =
69        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
70
71    for quote in data {
72        instrument_id_builder.append_value(quote.instrument_id.to_string());
73        bid_price_builder.append_value(price_to_f64(&quote.bid_price));
74        ask_price_builder.append_value(price_to_f64(&quote.ask_price));
75        bid_size_builder.append_value(quantity_to_f64(&quote.bid_size));
76        ask_size_builder.append_value(quantity_to_f64(&quote.ask_size));
77        ts_event_builder.append_value(unix_nanos_to_i64(quote.ts_event.as_u64()));
78        ts_init_builder.append_value(unix_nanos_to_i64(quote.ts_init.as_u64()));
79    }
80
81    RecordBatch::try_new(
82        Arc::new(quotes_schema()),
83        vec![
84            Arc::new(instrument_id_builder.finish()),
85            Arc::new(bid_price_builder.finish()),
86            Arc::new(ask_price_builder.finish()),
87            Arc::new(bid_size_builder.finish()),
88            Arc::new(ask_size_builder.finish()),
89            Arc::new(ts_event_builder.finish()),
90            Arc::new(ts_init_builder.finish()),
91        ],
92    )
93}
94
95#[cfg(test)]
96mod tests {
97    use arrow::{
98        array::{Array, Float64Array, StringArray, TimestampNanosecondArray},
99        datatypes::{DataType, TimeUnit},
100    };
101    use nautilus_model::{
102        identifiers::InstrumentId,
103        types::{Price, Quantity},
104    };
105    use rstest::rstest;
106
107    use super::*;
108
109    fn make_quote(instrument_id: &str, bid: &str, ask: &str, ts: u64) -> QuoteTick {
110        QuoteTick {
111            instrument_id: InstrumentId::from(instrument_id),
112            bid_price: Price::from(bid),
113            ask_price: Price::from(ask),
114            bid_size: Quantity::from(1_000),
115            ask_size: Quantity::from(500),
116            ts_event: ts.into(),
117            ts_init: (ts + 1).into(),
118        }
119    }
120
121    #[rstest]
122    fn test_encode_quotes_schema() {
123        let quotes = vec![make_quote("AAPL.XNAS", "100.10", "100.20", 1)];
124        let batch = encode_quotes(&quotes).unwrap();
125
126        let schema = batch.schema();
127        let fields = schema.fields();
128        assert_eq!(fields.len(), 7);
129        assert_eq!(fields[0].name(), "instrument_id");
130        assert_eq!(fields[0].data_type(), &DataType::Utf8);
131        assert_eq!(fields[1].name(), "bid_price");
132        assert_eq!(fields[1].data_type(), &DataType::Float64);
133        assert_eq!(fields[2].name(), "ask_price");
134        assert_eq!(fields[2].data_type(), &DataType::Float64);
135        assert_eq!(fields[3].name(), "bid_size");
136        assert_eq!(fields[3].data_type(), &DataType::Float64);
137        assert_eq!(fields[4].name(), "ask_size");
138        assert_eq!(fields[4].data_type(), &DataType::Float64);
139        assert_eq!(fields[5].name(), "ts_event");
140        assert_eq!(
141            fields[5].data_type(),
142            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
143        );
144        assert_eq!(fields[6].name(), "ts_init");
145        assert_eq!(
146            fields[6].data_type(),
147            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
148        );
149    }
150
151    #[rstest]
152    fn test_encode_quotes_values() {
153        let quotes = vec![
154            make_quote("AAPL.XNAS", "100.10", "100.20", 1_000_000_000),
155            make_quote("AAPL.XNAS", "100.15", "100.25", 2_000_000_000),
156        ];
157        let batch = encode_quotes(&quotes).unwrap();
158
159        assert_eq!(batch.num_rows(), 2);
160
161        let instrument_id_col = batch
162            .column(0)
163            .as_any()
164            .downcast_ref::<StringArray>()
165            .unwrap();
166        let bid_price_col = batch
167            .column(1)
168            .as_any()
169            .downcast_ref::<Float64Array>()
170            .unwrap();
171        let ask_price_col = batch
172            .column(2)
173            .as_any()
174            .downcast_ref::<Float64Array>()
175            .unwrap();
176        let bid_size_col = batch
177            .column(3)
178            .as_any()
179            .downcast_ref::<Float64Array>()
180            .unwrap();
181        let ask_size_col = batch
182            .column(4)
183            .as_any()
184            .downcast_ref::<Float64Array>()
185            .unwrap();
186        let ts_event_col = batch
187            .column(5)
188            .as_any()
189            .downcast_ref::<TimestampNanosecondArray>()
190            .unwrap();
191        let ts_init_col = batch
192            .column(6)
193            .as_any()
194            .downcast_ref::<TimestampNanosecondArray>()
195            .unwrap();
196
197        assert_eq!(instrument_id_col.value(0), "AAPL.XNAS");
198        assert_eq!(instrument_id_col.value(1), "AAPL.XNAS");
199        assert!((bid_price_col.value(0) - 100.10).abs() < 1e-9);
200        assert!((bid_price_col.value(1) - 100.15).abs() < 1e-9);
201        assert!((ask_price_col.value(0) - 100.20).abs() < 1e-9);
202        assert!((ask_price_col.value(1) - 100.25).abs() < 1e-9);
203        assert!((bid_size_col.value(0) - 1_000.0).abs() < 1e-9);
204        assert!((ask_size_col.value(0) - 500.0).abs() < 1e-9);
205        assert_eq!(ts_event_col.value(0), 1_000_000_000);
206        assert_eq!(ts_event_col.value(1), 2_000_000_000);
207        assert_eq!(ts_init_col.value(0), 1_000_000_001);
208        assert_eq!(ts_init_col.value(1), 2_000_000_001);
209    }
210
211    #[rstest]
212    fn test_encode_quotes_empty() {
213        let batch = encode_quotes(&[]).unwrap();
214        assert_eq!(batch.num_rows(), 0);
215        assert_eq!(batch.schema().fields().len(), 7);
216    }
217
218    #[rstest]
219    fn test_encode_quotes_mixed_instruments() {
220        let quotes = vec![
221            make_quote("AAPL.XNAS", "100.10", "100.20", 1),
222            make_quote("MSFT.XNAS", "250.00", "250.05", 2),
223        ];
224        let batch = encode_quotes(&quotes).unwrap();
225
226        let instrument_id_col = batch
227            .column(0)
228            .as_any()
229            .downcast_ref::<StringArray>()
230            .unwrap();
231        assert_eq!(instrument_id_col.value(0), "AAPL.XNAS");
232        assert_eq!(instrument_id_col.value(1), "MSFT.XNAS");
233    }
234}
nautilus_serialization/arrow/display/quote.rs

nautilus_serialization/arrow/display/
quote.rs
