nautilus_serialization/arrow/display/
order_filled.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 [`OrderFilled`].
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::events::OrderFilled;
27
28use super::{
29    bool_field, float64_field, price_to_f64, quantity_to_f64, timestamp_field, unix_nanos_to_i64,
30    utf8_field,
31};
32use crate::arrow::timestamp_data_type;
33
34/// Returns the display-mode Arrow schema for [`OrderFilled`].
35#[must_use]
36pub fn order_filled_schema() -> Schema {
37    Schema::new(vec![
38        utf8_field("trader_id", false),
39        utf8_field("strategy_id", false),
40        utf8_field("instrument_id", false),
41        utf8_field("client_order_id", false),
42        utf8_field("venue_order_id", false),
43        utf8_field("account_id", false),
44        utf8_field("trade_id", false),
45        utf8_field("order_side", false),
46        utf8_field("order_type", false),
47        float64_field("last_qty", false),
48        float64_field("last_px", false),
49        utf8_field("currency", false),
50        utf8_field("liquidity_side", false),
51        utf8_field("event_id", false),
52        timestamp_field("ts_event", false),
53        timestamp_field("ts_init", false),
54        bool_field("reconciliation", false),
55        utf8_field("position_id", true),
56        utf8_field("commission", true),
57    ])
58}
59
60/// Encodes order fills as a display-friendly Arrow [`RecordBatch`].
61///
62/// Emits `Float64` columns for `last_qty` and `last_px`, `Timestamp(Nanosecond)`
63/// columns for event and init times, and `Utf8` columns for identifiers and enums.
64/// Commission renders as its `Display` representation (e.g. `"100.50 USD"`).
65/// Mixed-instrument batches are supported.
66///
67/// Returns an empty [`RecordBatch`] with the correct schema when `data` is empty.
68///
69/// # Errors
70///
71/// Returns an [`ArrowError`] if the Arrow `RecordBatch` cannot be constructed.
72pub fn encode_order_fills(data: &[OrderFilled]) -> Result<RecordBatch, ArrowError> {
73    let mut trader_id = StringBuilder::new();
74    let mut strategy_id = StringBuilder::new();
75    let mut instrument_id = StringBuilder::new();
76    let mut client_order_id = StringBuilder::new();
77    let mut venue_order_id = StringBuilder::new();
78    let mut account_id = StringBuilder::new();
79    let mut trade_id = StringBuilder::new();
80    let mut order_side = StringBuilder::new();
81    let mut order_type = StringBuilder::new();
82    let mut last_qty = Float64Builder::with_capacity(data.len());
83    let mut last_px = Float64Builder::with_capacity(data.len());
84    let mut currency = StringBuilder::new();
85    let mut liquidity_side = StringBuilder::new();
86    let mut event_id = StringBuilder::new();
87    let mut ts_event =
88        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
89    let mut ts_init =
90        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
91    let mut reconciliation = BooleanBuilder::with_capacity(data.len());
92    let mut position_id = StringBuilder::new();
93    let mut commission = StringBuilder::new();
94
95    for fill in data {
96        trader_id.append_value(fill.trader_id);
97        strategy_id.append_value(fill.strategy_id);
98        instrument_id.append_value(fill.instrument_id.to_string());
99        client_order_id.append_value(fill.client_order_id);
100        venue_order_id.append_value(fill.venue_order_id);
101        account_id.append_value(fill.account_id);
102        trade_id.append_value(fill.trade_id.to_string());
103        order_side.append_value(format!("{}", fill.order_side));
104        order_type.append_value(format!("{}", fill.order_type));
105        last_qty.append_value(quantity_to_f64(&fill.last_qty));
106        last_px.append_value(price_to_f64(&fill.last_px));
107        currency.append_value(fill.currency.to_string());
108        liquidity_side.append_value(format!("{}", fill.liquidity_side));
109        event_id.append_value(fill.event_id.to_string());
110        ts_event.append_value(unix_nanos_to_i64(fill.ts_event.as_u64()));
111        ts_init.append_value(unix_nanos_to_i64(fill.ts_init.as_u64()));
112        reconciliation.append_value(fill.reconciliation);
113        position_id.append_option(fill.position_id.map(|v| v.to_string()));
114        commission.append_option(fill.commission.map(|v| format!("{v}")));
115    }
116
117    RecordBatch::try_new(
118        Arc::new(order_filled_schema()),
119        vec![
120            Arc::new(trader_id.finish()),
121            Arc::new(strategy_id.finish()),
122            Arc::new(instrument_id.finish()),
123            Arc::new(client_order_id.finish()),
124            Arc::new(venue_order_id.finish()),
125            Arc::new(account_id.finish()),
126            Arc::new(trade_id.finish()),
127            Arc::new(order_side.finish()),
128            Arc::new(order_type.finish()),
129            Arc::new(last_qty.finish()),
130            Arc::new(last_px.finish()),
131            Arc::new(currency.finish()),
132            Arc::new(liquidity_side.finish()),
133            Arc::new(event_id.finish()),
134            Arc::new(ts_event.finish()),
135            Arc::new(ts_init.finish()),
136            Arc::new(reconciliation.finish()),
137            Arc::new(position_id.finish()),
138            Arc::new(commission.finish()),
139        ],
140    )
141}
142
143#[cfg(test)]
144mod tests {
145    use arrow::{
146        array::{Array, BooleanArray, Float64Array, StringArray, TimestampNanosecondArray},
147        datatypes::{DataType, TimeUnit},
148    };
149    use nautilus_model::{
150        enums::{LiquiditySide, OrderType},
151        events::order::spec::OrderFilledSpec,
152        identifiers::{ClientOrderId, InstrumentId, PositionId, TradeId, VenueOrderId},
153        types::{Currency, Money, Price, Quantity},
154    };
155    use rstest::rstest;
156
157    use super::*;
158
159    fn make_fill(instrument_id: &str, commission: Option<Money>, ts: u64) -> OrderFilled {
160        OrderFilledSpec::builder()
161            .instrument_id(InstrumentId::from(instrument_id))
162            .client_order_id(ClientOrderId::from("O-001"))
163            .venue_order_id(VenueOrderId::from("V-001"))
164            .trade_id(TradeId::new("T-001"))
165            .order_type(OrderType::Limit)
166            .last_qty(Quantity::from(100))
167            .last_px(Price::from("50.25"))
168            .liquidity_side(LiquiditySide::Maker)
169            .ts_event(ts.into())
170            .ts_init((ts + 1).into())
171            .position_id(PositionId::from("P-001"))
172            .maybe_commission(commission)
173            .build()
174    }
175
176    #[rstest]
177    fn test_encode_order_fills_schema() {
178        let batch = encode_order_fills(&[]).unwrap();
179        let schema = batch.schema();
180        let fields = schema.fields();
181        assert_eq!(fields.len(), 19);
182        assert_eq!(fields[0].name(), "trader_id");
183        assert_eq!(fields[0].data_type(), &DataType::Utf8);
184        assert_eq!(fields[9].name(), "last_qty");
185        assert_eq!(fields[9].data_type(), &DataType::Float64);
186        assert_eq!(fields[10].name(), "last_px");
187        assert_eq!(fields[10].data_type(), &DataType::Float64);
188        assert_eq!(fields[14].name(), "ts_event");
189        assert_eq!(
190            fields[14].data_type(),
191            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
192        );
193        assert_eq!(fields[16].name(), "reconciliation");
194        assert_eq!(fields[16].data_type(), &DataType::Boolean);
195        assert_eq!(fields[18].name(), "commission");
196        assert!(fields[18].is_nullable());
197    }
198
199    #[rstest]
200    fn test_encode_order_fills_values() {
201        let commission = Money::new(10.50, Currency::USD());
202        let fills = vec![make_fill("AAPL.XNAS", Some(commission), 1_000)];
203        let batch = encode_order_fills(&fills).unwrap();
204
205        assert_eq!(batch.num_rows(), 1);
206
207        let last_qty_col = batch
208            .column(9)
209            .as_any()
210            .downcast_ref::<Float64Array>()
211            .unwrap();
212        let last_px_col = batch
213            .column(10)
214            .as_any()
215            .downcast_ref::<Float64Array>()
216            .unwrap();
217        let ts_event_col = batch
218            .column(14)
219            .as_any()
220            .downcast_ref::<TimestampNanosecondArray>()
221            .unwrap();
222        let reconciliation_col = batch
223            .column(16)
224            .as_any()
225            .downcast_ref::<BooleanArray>()
226            .unwrap();
227        let commission_col = batch
228            .column(18)
229            .as_any()
230            .downcast_ref::<StringArray>()
231            .unwrap();
232
233        assert!((last_qty_col.value(0) - 100.0).abs() < 1e-9);
234        assert!((last_px_col.value(0) - 50.25).abs() < 1e-9);
235        assert_eq!(ts_event_col.value(0), 1_000);
236        assert!(!reconciliation_col.value(0));
237        assert_eq!(commission_col.value(0), "10.50 USD");
238    }
239
240    #[rstest]
241    fn test_encode_order_fills_null_commission() {
242        let fills = vec![make_fill("AAPL.XNAS", None, 1_000)];
243        let batch = encode_order_fills(&fills).unwrap();
244
245        let commission_col = batch
246            .column(18)
247            .as_any()
248            .downcast_ref::<StringArray>()
249            .unwrap();
250        assert!(commission_col.is_null(0));
251    }
252
253    #[rstest]
254    fn test_encode_order_fills_empty() {
255        let batch = encode_order_fills(&[]).unwrap();
256        assert_eq!(batch.num_rows(), 0);
257        assert_eq!(batch.schema().fields().len(), 19);
258    }
259
260    #[rstest]
261    fn test_encode_order_fills_mixed_instruments() {
262        let fills = vec![
263            make_fill("AAPL.XNAS", None, 1),
264            make_fill("MSFT.XNAS", None, 2),
265        ];
266        let batch = encode_order_fills(&fills).unwrap();
267
268        let instrument_id_col = batch
269            .column(2)
270            .as_any()
271            .downcast_ref::<StringArray>()
272            .unwrap();
273        assert_eq!(instrument_id_col.value(0), "AAPL.XNAS");
274        assert_eq!(instrument_id_col.value(1), "MSFT.XNAS");
275    }
276}
nautilus_serialization/arrow/display/order_filled.rs

nautilus_serialization/arrow/display/
order_filled.rs
