nautilus_serialization/arrow/display/
report.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 [`OrderStatusReport`].
17
18use std::sync::Arc;
19
20use arrow::{
21    array::{BooleanBuilder, Float64Builder, StringBuilder, TimestampNanosecondBuilder},
22    datatypes::Schema,
23    error::ArrowError,
24    record_batch::RecordBatch,
25};
26use nautilus_model::reports::OrderStatusReport;
27use rust_decimal::prelude::ToPrimitive;
28
29use super::{
30    bool_field, float64_field, quantity_to_f64, timestamp_field, unix_nanos_to_i64, utf8_field,
31};
32use crate::arrow::timestamp_data_type;
33
34/// Returns the display-mode Arrow schema for [`OrderStatusReport`].
35#[must_use]
36pub fn order_status_report_schema() -> Schema {
37    Schema::new(vec![
38        utf8_field("account_id", false),
39        utf8_field("instrument_id", false),
40        utf8_field("client_order_id", true),
41        utf8_field("venue_order_id", false),
42        utf8_field("order_side", false),
43        utf8_field("order_type", false),
44        utf8_field("time_in_force", false),
45        utf8_field("order_status", false),
46        float64_field("quantity", false),
47        float64_field("filled_qty", false),
48        utf8_field("report_id", false),
49        timestamp_field("ts_accepted", false),
50        timestamp_field("ts_last", false),
51        timestamp_field("ts_init", false),
52        utf8_field("order_list_id", true),
53        utf8_field("venue_position_id", true),
54        utf8_field("linked_order_ids", true),
55        utf8_field("parent_order_id", true),
56        utf8_field("contingency_type", false),
57        timestamp_field("expire_time", true),
58        float64_field("price", true),
59        float64_field("activation_price", true),
60        float64_field("trigger_price", true),
61        utf8_field("trigger_type", true),
62        float64_field("limit_offset", true),
63        float64_field("trailing_offset", true),
64        utf8_field("trailing_offset_type", false),
65        float64_field("avg_px", true),
66        float64_field("display_qty", true),
67        bool_field("post_only", false),
68        bool_field("reduce_only", false),
69        utf8_field("cancel_reason", true),
70        timestamp_field("ts_triggered", true),
71    ])
72}
73
74/// Encodes order status reports as a display-friendly Arrow [`RecordBatch`].
75///
76/// Emits `Float64` columns for quantities, prices, and offsets,
77/// `Timestamp(Nanosecond)` columns for all time fields, and `Utf8` columns for
78/// identifiers and enums. Mixed-instrument batches are supported.
79///
80/// Returns an empty [`RecordBatch`] with the correct schema when `data` is empty.
81///
82/// # Errors
83///
84/// Returns an [`ArrowError`] if the Arrow `RecordBatch` cannot be constructed.
85pub fn encode_order_status_reports(data: &[OrderStatusReport]) -> Result<RecordBatch, ArrowError> {
86    let mut account_id = StringBuilder::new();
87    let mut instrument_id = StringBuilder::new();
88    let mut client_order_id = StringBuilder::new();
89    let mut venue_order_id = StringBuilder::new();
90    let mut order_side = StringBuilder::new();
91    let mut order_type = StringBuilder::new();
92    let mut time_in_force = StringBuilder::new();
93    let mut order_status = StringBuilder::new();
94    let mut quantity = Float64Builder::with_capacity(data.len());
95    let mut filled_qty = Float64Builder::with_capacity(data.len());
96    let mut report_id = StringBuilder::new();
97    let mut ts_accepted =
98        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
99    let mut ts_last =
100        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
101    let mut ts_init =
102        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
103    let mut order_list_id = StringBuilder::new();
104    let mut venue_position_id = StringBuilder::new();
105    let mut linked_order_ids = StringBuilder::new();
106    let mut parent_order_id = StringBuilder::new();
107    let mut contingency_type = StringBuilder::new();
108    let mut expire_time =
109        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
110    let mut price = Float64Builder::with_capacity(data.len());
111    let mut activation_price = Float64Builder::with_capacity(data.len());
112    let mut trigger_price = Float64Builder::with_capacity(data.len());
113    let mut trigger_type = StringBuilder::new();
114    let mut limit_offset = Float64Builder::with_capacity(data.len());
115    let mut trailing_offset = Float64Builder::with_capacity(data.len());
116    let mut trailing_offset_type = StringBuilder::new();
117    let mut avg_px = Float64Builder::with_capacity(data.len());
118    let mut display_qty = Float64Builder::with_capacity(data.len());
119    let mut post_only = BooleanBuilder::with_capacity(data.len());
120    let mut reduce_only = BooleanBuilder::with_capacity(data.len());
121    let mut cancel_reason = StringBuilder::new();
122    let mut ts_triggered =
123        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
124
125    for report in data {
126        account_id.append_value(report.account_id);
127        instrument_id.append_value(report.instrument_id.to_string());
128        client_order_id.append_option(report.client_order_id.map(|v| v.to_string()));
129        venue_order_id.append_value(report.venue_order_id);
130        order_side.append_value(
131            report
132                .order_side
133                .as_ref()
134                .map_or("NO_ORDER_SIDE", AsRef::as_ref),
135        );
136        order_type.append_value(format!("{}", report.order_type));
137        time_in_force.append_value(format!("{}", report.time_in_force));
138        order_status.append_value(format!("{}", report.order_status));
139        quantity.append_value(quantity_to_f64(&report.quantity));
140        filled_qty.append_value(quantity_to_f64(&report.filled_qty));
141        report_id.append_value(report.report_id.to_string());
142        ts_accepted.append_value(unix_nanos_to_i64(report.ts_accepted.as_u64()));
143        ts_last.append_value(unix_nanos_to_i64(report.ts_last.as_u64()));
144        ts_init.append_value(unix_nanos_to_i64(report.ts_init.as_u64()));
145        order_list_id.append_option(report.order_list_id.map(|v| v.to_string()));
146        venue_position_id.append_option(report.venue_position_id.map(|v| v.to_string()));
147        linked_order_ids.append_option(report.linked_order_ids.as_ref().map(|ids| {
148            let values: Vec<String> = ids.iter().map(ToString::to_string).collect();
149            serde_json::to_string(&values).unwrap_or_default()
150        }));
151        parent_order_id.append_option(report.parent_order_id.map(|v| v.to_string()));
152        contingency_type.append_value(
153            report
154                .contingency_type
155                .as_ref()
156                .map_or("NO_CONTINGENCY", AsRef::as_ref),
157        );
158        expire_time.append_option(report.expire_time.map(|v| unix_nanos_to_i64(v.as_u64())));
159        price.append_option(report.price.map(|v| v.as_f64()));
160        activation_price.append_option(report.activation_price.map(|v| v.as_f64()));
161        trigger_price.append_option(report.trigger_price.map(|v| v.as_f64()));
162        trigger_type.append_option(report.trigger_type.map(|v| format!("{v}")));
163        limit_offset.append_option(report.limit_offset.and_then(|v| v.to_f64()));
164        trailing_offset.append_option(report.trailing_offset.and_then(|v| v.to_f64()));
165        trailing_offset_type.append_value(
166            report
167                .trailing_offset_type
168                .as_ref()
169                .map_or("NO_TRAILING_OFFSET", AsRef::as_ref),
170        );
171        avg_px.append_option(report.avg_px.and_then(|v| v.to_f64()));
172        display_qty.append_option(report.display_qty.map(|v| quantity_to_f64(&v)));
173        post_only.append_value(report.post_only);
174        reduce_only.append_value(report.reduce_only);
175        cancel_reason.append_option(report.cancel_reason.clone());
176        ts_triggered.append_option(report.ts_triggered.map(|v| unix_nanos_to_i64(v.as_u64())));
177    }
178
179    RecordBatch::try_new(
180        Arc::new(order_status_report_schema()),
181        vec![
182            Arc::new(account_id.finish()),
183            Arc::new(instrument_id.finish()),
184            Arc::new(client_order_id.finish()),
185            Arc::new(venue_order_id.finish()),
186            Arc::new(order_side.finish()),
187            Arc::new(order_type.finish()),
188            Arc::new(time_in_force.finish()),
189            Arc::new(order_status.finish()),
190            Arc::new(quantity.finish()),
191            Arc::new(filled_qty.finish()),
192            Arc::new(report_id.finish()),
193            Arc::new(ts_accepted.finish()),
194            Arc::new(ts_last.finish()),
195            Arc::new(ts_init.finish()),
196            Arc::new(order_list_id.finish()),
197            Arc::new(venue_position_id.finish()),
198            Arc::new(linked_order_ids.finish()),
199            Arc::new(parent_order_id.finish()),
200            Arc::new(contingency_type.finish()),
201            Arc::new(expire_time.finish()),
202            Arc::new(price.finish()),
203            Arc::new(activation_price.finish()),
204            Arc::new(trigger_price.finish()),
205            Arc::new(trigger_type.finish()),
206            Arc::new(limit_offset.finish()),
207            Arc::new(trailing_offset.finish()),
208            Arc::new(trailing_offset_type.finish()),
209            Arc::new(avg_px.finish()),
210            Arc::new(display_qty.finish()),
211            Arc::new(post_only.finish()),
212            Arc::new(reduce_only.finish()),
213            Arc::new(cancel_reason.finish()),
214            Arc::new(ts_triggered.finish()),
215        ],
216    )
217}
218
219#[cfg(test)]
220mod tests {
221    use arrow::{
222        array::{Array, BooleanArray, Float64Array, StringArray, TimestampNanosecondArray},
223        datatypes::{DataType, TimeUnit},
224    };
225    use nautilus_core::UUID4;
226    use nautilus_model::{
227        enums::{OrderSide, OrderStatus, OrderType, TimeInForce},
228        identifiers::{AccountId, ClientOrderId, InstrumentId, VenueOrderId},
229        types::{Price, Quantity},
230    };
231    use rstest::rstest;
232
233    use super::*;
234
235    fn make_report(instrument_id: &str, ts: u64) -> OrderStatusReport {
236        OrderStatusReport {
237            account_id: AccountId::from("SIM-001"),
238            instrument_id: InstrumentId::from(instrument_id),
239            client_order_id: Some(ClientOrderId::from("O-001")),
240            venue_order_id: VenueOrderId::from("V-001"),
241            order_side: Some(OrderSide::Buy),
242            order_type: OrderType::Limit,
243            time_in_force: TimeInForce::Gtc,
244            order_status: OrderStatus::Accepted,
245            quantity: Quantity::from(100),
246            filled_qty: Quantity::from(50),
247            report_id: UUID4::default(),
248            ts_accepted: ts.into(),
249            ts_last: (ts + 1_000).into(),
250            ts_init: (ts + 1).into(),
251            order_list_id: None,
252            venue_position_id: None,
253            linked_order_ids: None,
254            parent_order_id: None,
255            contingency_type: None,
256            expire_time: None,
257            price: Some(Price::from("100.50")),
258            activation_price: None,
259            trigger_price: None,
260            trigger_type: None,
261            limit_offset: None,
262            trailing_offset: None,
263            trailing_offset_type: None,
264            avg_px: None,
265            display_qty: None,
266            post_only: true,
267            reduce_only: false,
268            cancel_reason: None,
269            ts_triggered: None,
270        }
271    }
272
273    #[rstest]
274    fn test_encode_order_status_reports_schema() {
275        let batch = encode_order_status_reports(&[]).unwrap();
276        let schema = batch.schema();
277        let fields = schema.fields();
278        assert_eq!(fields.len(), 33);
279        assert_eq!(fields[0].name(), "account_id");
280        assert_eq!(fields[0].data_type(), &DataType::Utf8);
281        assert_eq!(fields[8].name(), "quantity");
282        assert_eq!(fields[8].data_type(), &DataType::Float64);
283        assert_eq!(fields[11].name(), "ts_accepted");
284        assert_eq!(
285            fields[11].data_type(),
286            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
287        );
288        assert_eq!(fields[21].name(), "activation_price");
289        assert_eq!(fields[21].data_type(), &DataType::Float64);
290        assert_eq!(fields[29].name(), "post_only");
291        assert_eq!(fields[29].data_type(), &DataType::Boolean);
292    }
293
294    #[rstest]
295    fn test_encode_order_status_reports_values() {
296        let reports = vec![make_report("AAPL.XNAS", 1_000_000)];
297        let batch = encode_order_status_reports(&reports).unwrap();
298
299        assert_eq!(batch.num_rows(), 1);
300
301        let quantity_col = batch
302            .column(8)
303            .as_any()
304            .downcast_ref::<Float64Array>()
305            .unwrap();
306        let filled_qty_col = batch
307            .column(9)
308            .as_any()
309            .downcast_ref::<Float64Array>()
310            .unwrap();
311        let price_col = batch
312            .column(20)
313            .as_any()
314            .downcast_ref::<Float64Array>()
315            .unwrap();
316        let post_only_col = batch
317            .column(29)
318            .as_any()
319            .downcast_ref::<BooleanArray>()
320            .unwrap();
321        let ts_accepted_col = batch
322            .column(11)
323            .as_any()
324            .downcast_ref::<TimestampNanosecondArray>()
325            .unwrap();
326
327        assert!((quantity_col.value(0) - 100.0).abs() < 1e-9);
328        assert!((filled_qty_col.value(0) - 50.0).abs() < 1e-9);
329        assert!((price_col.value(0) - 100.50).abs() < 1e-9);
330        assert!(post_only_col.value(0));
331        assert_eq!(ts_accepted_col.value(0), 1_000_000);
332    }
333
334    #[rstest]
335    fn test_encode_order_status_reports_linked_order_ids_round_trip() {
336        let mut report = make_report("AAPL.XNAS", 1_000);
337        report.linked_order_ids = Some(vec![
338            ClientOrderId::from("O-Z"),
339            ClientOrderId::from("O-A"),
340            ClientOrderId::from("O-M"),
341        ]);
342        let batch = encode_order_status_reports(&[report]).unwrap();
343
344        let linked_col = batch
345            .column(16)
346            .as_any()
347            .downcast_ref::<StringArray>()
348            .unwrap();
349        assert!(!linked_col.is_null(0));
350
351        let parsed: Vec<String> = serde_json::from_str(linked_col.value(0)).unwrap();
352        assert_eq!(parsed, vec!["O-Z", "O-A", "O-M"]);
353    }
354
355    #[rstest]
356    fn test_encode_order_status_reports_linked_order_ids_null_when_absent() {
357        let batch = encode_order_status_reports(&[make_report("AAPL.XNAS", 1_000)]).unwrap();
358        let linked_col = batch
359            .column(16)
360            .as_any()
361            .downcast_ref::<StringArray>()
362            .unwrap();
363        assert!(linked_col.is_null(0));
364    }
365
366    #[rstest]
367    fn test_encode_order_status_reports_nullable_fields() {
368        let reports = vec![make_report("AAPL.XNAS", 1_000)];
369        let batch = encode_order_status_reports(&reports).unwrap();
370
371        let trigger_price_col = batch
372            .column(22)
373            .as_any()
374            .downcast_ref::<Float64Array>()
375            .unwrap();
376        let expire_time_col = batch
377            .column(19)
378            .as_any()
379            .downcast_ref::<TimestampNanosecondArray>()
380            .unwrap();
381
382        assert!(trigger_price_col.is_null(0));
383        assert!(expire_time_col.is_null(0));
384    }
385
386    #[rstest]
387    fn test_encode_order_status_reports_no_order_side() {
388        let mut report = make_report("AAPL.XNAS", 1_000);
389        report.order_side = None;
390
391        let batch = encode_order_status_reports(&[report]).unwrap();
392        let order_side_col = batch
393            .column(4)
394            .as_any()
395            .downcast_ref::<StringArray>()
396            .unwrap();
397
398        assert_eq!(order_side_col.value(0), "NO_ORDER_SIDE");
399    }
400
401    #[rstest]
402    fn test_encode_order_status_reports_empty() {
403        let batch = encode_order_status_reports(&[]).unwrap();
404        assert_eq!(batch.num_rows(), 0);
405        assert_eq!(batch.schema().fields().len(), 33);
406    }
407
408    #[rstest]
409    fn test_encode_order_status_reports_mixed_instruments() {
410        let reports = vec![make_report("AAPL.XNAS", 1), make_report("MSFT.XNAS", 2)];
411        let batch = encode_order_status_reports(&reports).unwrap();
412
413        let instrument_id_col = batch
414            .column(1)
415            .as_any()
416            .downcast_ref::<StringArray>()
417            .unwrap();
418        assert_eq!(instrument_id_col.value(0), "AAPL.XNAS");
419        assert_eq!(instrument_id_col.value(1), "MSFT.XNAS");
420    }
421}
nautilus_serialization/arrow/display/report.rs

nautilus_serialization/arrow/display/
report.rs
