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

nautilus_serialization/arrow/display/
report.rs
