nautilus_serialization/arrow/
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
16use nautilus_model::reports::{
17    ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport,
18};
19
20use super::json::{JsonFieldSpec, impl_json_arrow};
21
22const ORDER_STATUS_REPORT_FIELDS: &[JsonFieldSpec] = &[
23    JsonFieldSpec::utf8("account_id", false),
24    JsonFieldSpec::utf8("instrument_id", false),
25    JsonFieldSpec::utf8("client_order_id", true),
26    JsonFieldSpec::utf8("venue_order_id", false),
27    JsonFieldSpec::utf8("order_side", false),
28    JsonFieldSpec::utf8("order_type", false),
29    JsonFieldSpec::utf8("time_in_force", false),
30    JsonFieldSpec::utf8("order_status", false),
31    JsonFieldSpec::utf8("quantity", false),
32    JsonFieldSpec::utf8("filled_qty", false),
33    JsonFieldSpec::utf8("report_id", false),
34    JsonFieldSpec::u64("ts_accepted", false),
35    JsonFieldSpec::u64("ts_last", false),
36    JsonFieldSpec::u64("ts_init", false),
37    JsonFieldSpec::utf8("order_list_id", true),
38    JsonFieldSpec::utf8("venue_position_id", true),
39    JsonFieldSpec::utf8_json("linked_order_ids", true),
40    JsonFieldSpec::utf8("parent_order_id", true),
41    JsonFieldSpec::utf8("contingency_type", false),
42    JsonFieldSpec::u64("expire_time", true),
43    JsonFieldSpec::utf8("price", true),
44    JsonFieldSpec::utf8("activation_price", true),
45    JsonFieldSpec::utf8("trigger_price", true),
46    JsonFieldSpec::utf8("trigger_type", true),
47    JsonFieldSpec::utf8("limit_offset", true),
48    JsonFieldSpec::utf8("trailing_offset", true),
49    JsonFieldSpec::utf8("trailing_offset_type", false),
50    JsonFieldSpec::utf8("avg_px", true),
51    JsonFieldSpec::utf8("display_qty", true),
52    JsonFieldSpec::boolean("post_only", false),
53    JsonFieldSpec::boolean("reduce_only", false),
54    JsonFieldSpec::utf8("cancel_reason", true),
55    JsonFieldSpec::u64("ts_triggered", true),
56];
57
58const FILL_REPORT_FIELDS: &[JsonFieldSpec] = &[
59    JsonFieldSpec::utf8("account_id", false),
60    JsonFieldSpec::utf8("instrument_id", false),
61    JsonFieldSpec::utf8("venue_order_id", false),
62    JsonFieldSpec::utf8("trade_id", false),
63    JsonFieldSpec::utf8("order_side", false),
64    JsonFieldSpec::utf8("last_qty", false),
65    JsonFieldSpec::utf8("last_px", false),
66    JsonFieldSpec::utf8("commission", false),
67    JsonFieldSpec::utf8("liquidity_side", false),
68    JsonFieldSpec::utf8("report_id", false),
69    JsonFieldSpec::u64("ts_event", false),
70    JsonFieldSpec::u64("ts_init", false),
71    JsonFieldSpec::utf8("client_order_id", true),
72    JsonFieldSpec::utf8("venue_position_id", true),
73    // Appended (not inserted) so a batch written here still decodes positionally in builds
74    // that predate name-based column resolution.
75    JsonFieldSpec::decimal_str("avg_px", true),
76];
77
78const POSITION_STATUS_REPORT_FIELDS: &[JsonFieldSpec] = &[
79    JsonFieldSpec::utf8("account_id", false),
80    JsonFieldSpec::utf8("instrument_id", false),
81    JsonFieldSpec::utf8("position_side", false),
82    JsonFieldSpec::utf8("quantity", false),
83    JsonFieldSpec::utf8("signed_decimal_qty", false),
84    JsonFieldSpec::utf8("report_id", false),
85    JsonFieldSpec::u64("ts_last", false),
86    JsonFieldSpec::u64("ts_init", false),
87    JsonFieldSpec::utf8("venue_position_id", true),
88    JsonFieldSpec::utf8("avg_px_open", true),
89];
90
91const EXECUTION_MASS_STATUS_FIELDS: &[JsonFieldSpec] = &[
92    JsonFieldSpec::utf8("client_id", false),
93    JsonFieldSpec::utf8("account_id", false),
94    JsonFieldSpec::utf8("venue", false),
95    JsonFieldSpec::utf8("report_id", false),
96    JsonFieldSpec::u64("ts_init", false),
97    JsonFieldSpec::utf8_json("order_reports", false),
98    JsonFieldSpec::utf8_json("fill_reports", false),
99    JsonFieldSpec::utf8_json("position_reports", false),
100    // Appended (not inserted) so a batch written here still decodes positionally in builds
101    // that predate name-based column resolution.
102    JsonFieldSpec::u64("lookback_start", true),
103    JsonFieldSpec::boolean_default_true("reports_complete"),
104];
105
106impl_json_arrow!(instrument OrderStatusReport, "OrderStatusReport", ORDER_STATUS_REPORT_FIELDS);
107impl_json_arrow!(instrument FillReport, "FillReport", FILL_REPORT_FIELDS, &["avg_px"]);
108impl_json_arrow!(instrument PositionStatusReport, "PositionStatusReport", POSITION_STATUS_REPORT_FIELDS);
109impl_json_arrow!(
110    typed ExecutionMassStatus,
111    "ExecutionMassStatus",
112    EXECUTION_MASS_STATUS_FIELDS,
113    &["lookback_start", "reports_complete"]
114);
115
116#[cfg(test)]
117mod tests {
118    use std::{str::FromStr, sync::Arc};
119
120    use arrow::{
121        array::{ArrayRef, BooleanArray},
122        datatypes::{DataType, Field, Schema},
123        record_batch::RecordBatch,
124    };
125    use nautilus_core::{UUID4, UnixNanos};
126    use nautilus_model::{
127        enums::{LiquiditySide, OrderSide, OrderStatus, OrderType, PositionSide, TimeInForce},
128        identifiers::{
129            AccountId, ClientId, ClientOrderId, InstrumentId, PositionId, TradeId, Venue,
130            VenueOrderId,
131        },
132        reports::{ExecutionMassStatus, FillReport, OrderStatusReport, PositionStatusReport},
133        types::{Currency, Money, Price, Quantity},
134    };
135    use rstest::rstest;
136    use rust_decimal::Decimal;
137
138    use crate::arrow::{
139        ArrowSchemaProvider, DecodeTypedFromRecordBatch, EncodeToRecordBatch, EncodingError,
140    };
141
142    // Builds that predate name-based column resolution decode these batches by position, so a
143    // new column must be appended rather than inserted. Pinning the order fails a mid-schema
144    // insertion here instead of silently shifting every later column for those readers.
145    #[rstest]
146    #[case::fill_report(
147        FillReport::get_schema(None),
148        &[
149            "account_id",
150            "instrument_id",
151            "venue_order_id",
152            "trade_id",
153            "order_side",
154            "last_qty",
155            "last_px",
156            "commission",
157            "liquidity_side",
158            "report_id",
159            "ts_event",
160            "ts_init",
161            "client_order_id",
162            "venue_position_id",
163            "avg_px",
164        ]
165    )]
166    #[case::execution_mass_status(
167        ExecutionMassStatus::get_schema(None),
168        &[
169            "client_id",
170            "account_id",
171            "venue",
172            "report_id",
173            "ts_init",
174            "order_reports",
175            "fill_reports",
176            "position_reports",
177            "lookback_start",
178            "reports_complete",
179        ]
180    )]
181    fn test_schema_column_order(#[case] schema: Schema, #[case] expected: &[&str]) {
182        let names: Vec<&str> = schema
183            .fields()
184            .iter()
185            .map(|field| field.name().as_str())
186            .collect();
187
188        assert_eq!(names, expected);
189    }
190
191    #[rstest]
192    fn test_order_status_report_round_trip() {
193        let report = OrderStatusReport::new(
194            AccountId::from("SIM-001"),
195            InstrumentId::from("AUDUSD.SIM"),
196            Some(ClientOrderId::from("O-19700101-000000-001-001-1")),
197            VenueOrderId::from("1"),
198            OrderSide::Buy.into(),
199            OrderType::Limit,
200            TimeInForce::Gtc,
201            OrderStatus::Accepted,
202            Quantity::from("100"),
203            Quantity::from("25"),
204            UnixNanos::from(1_000_000_000),
205            UnixNanos::from(2_000_000_000),
206            UnixNanos::from(3_000_000_000),
207            None,
208        )
209        .with_linked_order_ids([ClientOrderId::from("O-19700101-000000-001-001-2")]);
210        let report = OrderStatusReport {
211            activation_price: Some(Price::from("1.05000")),
212            limit_offset: Some(Decimal::from_str("0.123456789123456789").unwrap()),
213            trailing_offset: Some(Decimal::from_str("0.987654321987654321").unwrap()),
214            avg_px: Some(Decimal::from_str("1.23456789123456789").unwrap()),
215            ..report
216        };
217
218        let metadata = report.metadata();
219        let batch =
220            OrderStatusReport::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
221        let decoded =
222            OrderStatusReport::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
223
224        assert_eq!(decoded, vec![report]);
225    }
226
227    #[rstest]
228    fn test_position_status_report_round_trip_preserves_decimal_precision() {
229        let report = PositionStatusReport {
230            account_id: AccountId::from("SIM-001"),
231            instrument_id: InstrumentId::from("AUDUSD.SIM"),
232            position_side: PositionSide::Long,
233            quantity: Quantity::from("100.25"),
234            signed_decimal_qty: Decimal::from_str("100.250000000123456789").unwrap(),
235            report_id: UUID4::default(),
236            ts_last: UnixNanos::from(1_000_000_000),
237            ts_init: UnixNanos::from(2_000_000_000),
238            venue_position_id: Some(PositionId::from("P-001")),
239            avg_px_open: Some(Decimal::from_str("1.23456789123456789").unwrap()),
240        };
241        let metadata = report.metadata();
242        let batch =
243            PositionStatusReport::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
244        let decoded =
245            PositionStatusReport::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
246
247        assert_eq!(decoded, vec![report]);
248    }
249
250    #[rstest]
251    fn test_fill_report_round_trip_preserves_average_price() {
252        let report = sample_fill_report(Some(Decimal::from_str("1.23456789123456789").unwrap()));
253
254        let metadata = report.metadata();
255        let batch = FillReport::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
256        let decoded = FillReport::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
257
258        assert_eq!(decoded, vec![report]);
259    }
260
261    #[rstest]
262    fn test_fill_report_decodes_merged_schema_order() {
263        let report = sample_fill_report(Some(Decimal::from_str("1.23456789123456789").unwrap()));
264        let metadata = report.metadata();
265        let batch = FillReport::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
266        let merged_batch = batch_with_columns_at_end(&batch, &["account_id", "avg_px"]);
267
268        let decoded =
269            FillReport::decode_typed_batch(merged_batch.schema().metadata(), merged_batch).unwrap();
270
271        assert_eq!(decoded, vec![report]);
272    }
273
274    #[rstest]
275    fn test_fill_report_decodes_legacy_batch_without_average_price() {
276        let report = sample_fill_report(None);
277        let metadata = report.metadata();
278        let batch = FillReport::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
279        let legacy_batch = batch_without_columns(&batch, &["avg_px"]);
280
281        let decoded =
282            FillReport::decode_typed_batch(legacy_batch.schema().metadata(), legacy_batch).unwrap();
283
284        assert_eq!(decoded, vec![report]);
285    }
286
287    #[rstest]
288    fn test_execution_mass_status_round_trip_preserves_report_window_and_reports() {
289        let order_report = OrderStatusReport::new(
290            AccountId::from("SIM-001"),
291            InstrumentId::from("AUDUSD.SIM"),
292            Some(ClientOrderId::from("O-19700101-000000-001-001-4")),
293            VenueOrderId::from("3"),
294            OrderSide::Buy.into(),
295            OrderType::Market,
296            TimeInForce::Ioc,
297            OrderStatus::Filled,
298            Quantity::from("20"),
299            Quantity::from("20"),
300            UnixNanos::from(6_000_000_000),
301            UnixNanos::from(7_000_000_000),
302            UnixNanos::from(8_000_000_000),
303            None,
304        );
305        let fill_report = sample_fill_report(Some(Decimal::from_str("1.25001").unwrap()));
306        let position_report = PositionStatusReport::new(
307            AccountId::from("SIM-001"),
308            InstrumentId::from("AUDUSD.SIM"),
309            PositionSide::Long,
310            Quantity::from("20"),
311            UnixNanos::from(7_000_000_000),
312            UnixNanos::from(8_000_000_000),
313            None,
314            Some(PositionId::from("P-003")),
315            Some(Decimal::from_str("1.25001").unwrap()),
316        );
317        let mut report = ExecutionMassStatus::new(
318            ClientId::from("SIM"),
319            AccountId::from("SIM-001"),
320            Venue::from("SIM"),
321            UnixNanos::from(9_000_000_000),
322            None,
323        );
324        report.set_report_window(Some(UnixNanos::from(5_000_000_000)), false);
325        report.add_order_reports(vec![order_report]);
326        report.add_fill_reports(vec![fill_report]);
327        report.add_position_reports(vec![position_report]);
328
329        let metadata = report.metadata();
330        let batch =
331            ExecutionMassStatus::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
332        let decoded =
333            ExecutionMassStatus::decode_typed_batch(batch.schema().metadata(), batch).unwrap();
334
335        assert_eq!(decoded, vec![report]);
336    }
337
338    #[rstest]
339    fn test_execution_mass_status_decodes_legacy_batch_with_report_window_defaults() {
340        let report = ExecutionMassStatus::new(
341            ClientId::from("SIM"),
342            AccountId::from("SIM-001"),
343            Venue::from("SIM"),
344            UnixNanos::from(10_000_000_000),
345            None,
346        );
347        let metadata = report.metadata();
348        let batch =
349            ExecutionMassStatus::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
350        let legacy_batch = batch_without_columns(&batch, &["lookback_start", "reports_complete"]);
351
352        let decoded =
353            ExecutionMassStatus::decode_typed_batch(legacy_batch.schema().metadata(), legacy_batch)
354                .unwrap();
355
356        assert_eq!(decoded, vec![report]);
357    }
358
359    #[rstest]
360    fn test_execution_mass_status_decodes_merged_legacy_rows_with_report_window_defaults() {
361        let report = ExecutionMassStatus::new(
362            ClientId::from("SIM"),
363            AccountId::from("SIM-001"),
364            Venue::from("SIM"),
365            UnixNanos::from(11_000_000_000),
366            None,
367        );
368        let metadata = report.metadata();
369        let batch =
370            ExecutionMassStatus::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
371        let batch = batch_with_null_boolean_column(&batch, "reports_complete");
372        let merged_batch = batch_with_columns_at_end(&batch, &["client_id", "lookback_start"]);
373
374        let decoded =
375            ExecutionMassStatus::decode_typed_batch(merged_batch.schema().metadata(), merged_batch)
376                .unwrap();
377
378        assert_eq!(decoded, vec![report]);
379    }
380
381    #[rstest]
382    #[case("lookback_start", 8)]
383    #[case("reports_complete", 9)]
384    fn test_execution_mass_status_rejects_partial_report_window_schema(
385        #[case] missing: &'static str,
386        #[case] expected_index: usize,
387    ) {
388        let report = ExecutionMassStatus::new(
389            ClientId::from("SIM"),
390            AccountId::from("SIM-001"),
391            Venue::from("SIM"),
392            UnixNanos::from(12_000_000_000),
393            None,
394        );
395        let metadata = report.metadata();
396        let batch =
397            ExecutionMassStatus::encode_batch(&metadata, std::slice::from_ref(&report)).unwrap();
398        let partial_batch = batch_without_columns(&batch, &[missing]);
399
400        let error = ExecutionMassStatus::decode_typed_batch(
401            partial_batch.schema().metadata(),
402            partial_batch,
403        )
404        .expect_err("partial report window schema must be rejected");
405
406        assert!(matches!(
407            error,
408            EncodingError::MissingColumn(field, index)
409                if field == missing && index == expected_index
410        ));
411    }
412
413    fn sample_fill_report(avg_px: Option<Decimal>) -> FillReport {
414        let report = FillReport::new(
415            AccountId::from("SIM-001"),
416            InstrumentId::from("AUDUSD.SIM"),
417            VenueOrderId::from("2"),
418            TradeId::from("T-002"),
419            OrderSide::Sell,
420            Quantity::from("17.25"),
421            Price::from("1.23456"),
422            Money::new(2.75, Currency::USD()),
423            LiquiditySide::Maker,
424            Some(ClientOrderId::from("O-19700101-000000-001-001-3")),
425            Some(PositionId::from("P-002")),
426            UnixNanos::from(4_000_000_000),
427            UnixNanos::from(5_000_000_000),
428            None,
429        );
430
431        FillReport { avg_px, ..report }
432    }
433
434    fn batch_without_columns(batch: &RecordBatch, names: &[&str]) -> RecordBatch {
435        batch_with_column_order(batch, &column_indices(batch, names, false))
436    }
437
438    fn batch_with_columns_at_end(batch: &RecordBatch, names: &[&str]) -> RecordBatch {
439        let mut indices = column_indices(batch, names, false);
440        indices.extend(column_indices(batch, names, true));
441
442        batch_with_column_order(batch, &indices)
443    }
444
445    fn column_indices(batch: &RecordBatch, names: &[&str], named: bool) -> Vec<usize> {
446        batch
447            .schema()
448            .fields()
449            .iter()
450            .enumerate()
451            .filter(|(_, field)| names.contains(&field.name().as_str()) == named)
452            .map(|(index, _)| index)
453            .collect()
454    }
455
456    fn batch_with_column_order(batch: &RecordBatch, indices: &[usize]) -> RecordBatch {
457        let schema = batch.schema();
458        let fields: Vec<_> = indices
459            .iter()
460            .map(|index| schema.field(*index).clone())
461            .collect();
462        let columns: Vec<ArrayRef> = indices
463            .iter()
464            .map(|index| Arc::clone(batch.column(*index)))
465            .collect();
466        let reordered = Schema::new_with_metadata(fields, schema.metadata().clone());
467
468        RecordBatch::try_new(Arc::new(reordered), columns).unwrap()
469    }
470
471    fn batch_with_null_boolean_column(batch: &RecordBatch, name: &str) -> RecordBatch {
472        let schema = batch.schema();
473        let column_index = schema.index_of(name).unwrap();
474        let mut fields = schema.fields().to_vec();
475        fields[column_index] = Arc::new(Field::new(name, DataType::Boolean, true));
476        let mut columns = batch.columns().to_vec();
477        columns[column_index] = Arc::new(BooleanArray::from(vec![None; batch.num_rows()]));
478        let schema = Schema::new_with_metadata(fields, schema.metadata().clone());
479
480        RecordBatch::try_new(Arc::new(schema), columns).unwrap()
481    }
482}
nautilus_serialization/arrow/report.rs

nautilus_serialization/arrow/
report.rs
