nautilus_serialization/arrow/display/
position.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 [`Position`].
17
18use std::sync::Arc;
19
20use arrow::{
21    array::{
22        BooleanBuilder, Float64Builder, StringBuilder, TimestampNanosecondBuilder, UInt8Builder,
23        UInt32Builder, UInt64Builder,
24    },
25    datatypes::Schema,
26    error::ArrowError,
27    record_batch::RecordBatch,
28};
29use nautilus_model::position::Position;
30
31use super::{
32    bool_field, float64_field, money_to_f64, quantity_to_f64, timestamp_field, uint8_field,
33    uint32_field, uint64_field, unix_nanos_to_i64, utf8_field,
34};
35use crate::arrow::timestamp_data_type;
36
37/// Returns the display-mode Arrow schema for [`Position`].
38#[must_use]
39pub fn position_schema() -> Schema {
40    Schema::new(vec![
41        utf8_field("trader_id", false),
42        utf8_field("strategy_id", false),
43        utf8_field("instrument_id", false),
44        utf8_field("position_id", false),
45        utf8_field("account_id", false),
46        utf8_field("opening_order_id", false),
47        utf8_field("closing_order_id", true),
48        utf8_field("entry", false),
49        utf8_field("side", false),
50        float64_field("signed_qty", false),
51        float64_field("quantity", false),
52        float64_field("peak_qty", false),
53        uint8_field("price_precision", false),
54        uint8_field("size_precision", false),
55        float64_field("multiplier", false),
56        bool_field("is_inverse", false),
57        bool_field("is_currency_pair", false),
58        utf8_field("instrument_class", false),
59        utf8_field("base_currency", true),
60        utf8_field("quote_currency", false),
61        utf8_field("settlement_currency", false),
62        timestamp_field("ts_init", false),
63        timestamp_field("ts_opened", false),
64        timestamp_field("ts_last", false),
65        timestamp_field("ts_closed", true),
66        uint64_field("duration_ns", false),
67        float64_field("avg_px_open", false),
68        float64_field("avg_px_close", true),
69        float64_field("realized_return", false),
70        float64_field("realized_pnl_amount", true),
71        utf8_field("realized_pnl_currency", true),
72        utf8_field("trade_ids", false),
73        float64_field("buy_qty", false),
74        float64_field("sell_qty", false),
75        utf8_field("commissions", false),
76        uint32_field("event_count", false),
77        uint32_field("adjustment_count", false),
78    ])
79}
80
81fn trade_ids_to_json(position: &Position) -> String {
82    let mut trade_ids: Vec<String> = position.trade_ids.iter().map(ToString::to_string).collect();
83    trade_ids.sort();
84    serde_json::to_string(&trade_ids).unwrap_or_default()
85}
86
87fn commissions_to_json(position: &Position) -> String {
88    let mut commissions: Vec<(String, f64)> = position
89        .commissions
90        .iter()
91        .map(|(currency, money)| (currency.to_string(), money_to_f64(money)))
92        .collect();
93    commissions.sort_by(|lhs, rhs| lhs.0.cmp(&rhs.0));
94    serde_json::to_string(&commissions).unwrap_or_default()
95}
96
97/// Encodes positions as a display-friendly Arrow [`RecordBatch`].
98///
99/// Emits `Utf8` columns for identifiers and enums, `Float64` columns for
100/// quantities and PnL amounts, `Timestamp(Nanosecond)` columns for all time
101/// fields, and `Boolean` columns for `is_inverse` and `is_currency_pair`.
102/// The `trade_ids` and `commissions` columns carry deterministic JSON payloads
103/// (sorted by trade id and currency respectively) so that repeated encodings
104/// of the same position produce identical bytes.
105///
106/// Returns an empty [`RecordBatch`] with the correct schema when `data` is empty.
107///
108/// # Errors
109///
110/// Returns an [`ArrowError`] if the Arrow `RecordBatch` cannot be constructed.
111pub fn encode_positions(data: &[Position]) -> Result<RecordBatch, ArrowError> {
112    let mut trader_id = StringBuilder::new();
113    let mut strategy_id = StringBuilder::new();
114    let mut instrument_id = StringBuilder::new();
115    let mut position_id = StringBuilder::new();
116    let mut account_id = StringBuilder::new();
117    let mut opening_order_id = StringBuilder::new();
118    let mut closing_order_id = StringBuilder::new();
119    let mut entry = StringBuilder::new();
120    let mut side = StringBuilder::new();
121    let mut signed_qty = Float64Builder::with_capacity(data.len());
122    let mut quantity = Float64Builder::with_capacity(data.len());
123    let mut peak_qty = Float64Builder::with_capacity(data.len());
124    let mut price_precision = UInt8Builder::with_capacity(data.len());
125    let mut size_precision = UInt8Builder::with_capacity(data.len());
126    let mut multiplier = Float64Builder::with_capacity(data.len());
127    let mut is_inverse = BooleanBuilder::with_capacity(data.len());
128    let mut is_currency_pair = BooleanBuilder::with_capacity(data.len());
129    let mut instrument_class = StringBuilder::new();
130    let mut base_currency = StringBuilder::new();
131    let mut quote_currency = StringBuilder::new();
132    let mut settlement_currency = StringBuilder::new();
133    let mut ts_init =
134        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
135    let mut ts_opened =
136        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
137    let mut ts_last =
138        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
139    let mut ts_closed =
140        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
141    let mut duration_ns = UInt64Builder::with_capacity(data.len());
142    let mut avg_px_open = Float64Builder::with_capacity(data.len());
143    let mut avg_px_close = Float64Builder::with_capacity(data.len());
144    let mut realized_return = Float64Builder::with_capacity(data.len());
145    let mut realized_pnl_amount = Float64Builder::with_capacity(data.len());
146    let mut realized_pnl_currency = StringBuilder::new();
147    let mut trade_ids = StringBuilder::new();
148    let mut buy_qty = Float64Builder::with_capacity(data.len());
149    let mut sell_qty = Float64Builder::with_capacity(data.len());
150    let mut commissions = StringBuilder::new();
151    let mut event_count = UInt32Builder::with_capacity(data.len());
152    let mut adjustment_count = UInt32Builder::with_capacity(data.len());
153
154    for position in data {
155        trader_id.append_value(position.trader_id);
156        strategy_id.append_value(position.strategy_id);
157        instrument_id.append_value(position.instrument_id.to_string());
158        position_id.append_value(position.id);
159        account_id.append_value(position.account_id);
160        opening_order_id.append_value(position.opening_order_id);
161        closing_order_id.append_option(position.closing_order_id.map(|v| v.to_string()));
162        entry.append_value(format!("{}", position.entry));
163        side.append_value(format!("{}", position.side));
164        signed_qty.append_value(position.signed_qty);
165        quantity.append_value(quantity_to_f64(&position.quantity));
166        peak_qty.append_value(quantity_to_f64(&position.peak_qty));
167        price_precision.append_value(position.price_precision);
168        size_precision.append_value(position.size_precision);
169        multiplier.append_value(quantity_to_f64(&position.multiplier));
170        is_inverse.append_value(position.is_inverse);
171        is_currency_pair.append_value(position.is_currency_pair);
172        instrument_class.append_value(format!("{}", position.instrument_class));
173        base_currency.append_option(position.base_currency.map(|v| v.to_string()));
174        quote_currency.append_value(position.quote_currency.to_string());
175        settlement_currency.append_value(position.settlement_currency.to_string());
176        ts_init.append_value(unix_nanos_to_i64(position.ts_init.as_u64()));
177        ts_opened.append_value(unix_nanos_to_i64(position.ts_opened.as_u64()));
178        ts_last.append_value(unix_nanos_to_i64(position.ts_last.as_u64()));
179        ts_closed.append_option(position.ts_closed.map(|v| unix_nanos_to_i64(v.as_u64())));
180        duration_ns.append_value(position.duration_ns.as_u64());
181        avg_px_open.append_value(position.avg_px_open);
182        avg_px_close.append_option(position.avg_px_close);
183        realized_return.append_value(position.realized_return);
184        realized_pnl_amount.append_option(position.realized_pnl.map(|v| money_to_f64(&v)));
185        realized_pnl_currency.append_option(position.realized_pnl.map(|v| v.currency.to_string()));
186        trade_ids.append_value(trade_ids_to_json(position));
187        buy_qty.append_value(quantity_to_f64(&position.buy_qty));
188        sell_qty.append_value(quantity_to_f64(&position.sell_qty));
189        commissions.append_value(commissions_to_json(position));
190        event_count.append_value(position.events.len() as u32);
191        adjustment_count.append_value(position.adjustments.len() as u32);
192    }
193
194    RecordBatch::try_new(
195        Arc::new(position_schema()),
196        vec![
197            Arc::new(trader_id.finish()),
198            Arc::new(strategy_id.finish()),
199            Arc::new(instrument_id.finish()),
200            Arc::new(position_id.finish()),
201            Arc::new(account_id.finish()),
202            Arc::new(opening_order_id.finish()),
203            Arc::new(closing_order_id.finish()),
204            Arc::new(entry.finish()),
205            Arc::new(side.finish()),
206            Arc::new(signed_qty.finish()),
207            Arc::new(quantity.finish()),
208            Arc::new(peak_qty.finish()),
209            Arc::new(price_precision.finish()),
210            Arc::new(size_precision.finish()),
211            Arc::new(multiplier.finish()),
212            Arc::new(is_inverse.finish()),
213            Arc::new(is_currency_pair.finish()),
214            Arc::new(instrument_class.finish()),
215            Arc::new(base_currency.finish()),
216            Arc::new(quote_currency.finish()),
217            Arc::new(settlement_currency.finish()),
218            Arc::new(ts_init.finish()),
219            Arc::new(ts_opened.finish()),
220            Arc::new(ts_last.finish()),
221            Arc::new(ts_closed.finish()),
222            Arc::new(duration_ns.finish()),
223            Arc::new(avg_px_open.finish()),
224            Arc::new(avg_px_close.finish()),
225            Arc::new(realized_return.finish()),
226            Arc::new(realized_pnl_amount.finish()),
227            Arc::new(realized_pnl_currency.finish()),
228            Arc::new(trade_ids.finish()),
229            Arc::new(buy_qty.finish()),
230            Arc::new(sell_qty.finish()),
231            Arc::new(commissions.finish()),
232            Arc::new(event_count.finish()),
233            Arc::new(adjustment_count.finish()),
234        ],
235    )
236}
237
238#[cfg(test)]
239mod tests {
240    use arrow::{
241        array::{
242            Array, BooleanArray, Float64Array, StringArray, TimestampNanosecondArray, UInt8Array,
243            UInt32Array, UInt64Array,
244        },
245        datatypes::{DataType, TimeUnit},
246    };
247    use nautilus_model::{
248        enums::OrderSide,
249        events::{OrderFilled, order::spec::OrderFilledSpec},
250        identifiers::{ClientOrderId, PositionId, TradeId, VenueOrderId},
251        instruments::{CurrencyPair, InstrumentAny, stubs::currency_pair_btcusdt},
252        types::{Money, Price, Quantity},
253    };
254    use rstest::rstest;
255
256    use super::*;
257
258    #[expect(clippy::too_many_arguments)]
259    fn make_fill(
260        instrument: &CurrencyPair,
261        side: OrderSide,
262        qty: &str,
263        price: &str,
264        trade_id: &str,
265        order_id: &str,
266        ts: u64,
267        commission: Option<Money>,
268    ) -> OrderFilled {
269        OrderFilledSpec::builder()
270            .instrument_id(instrument.id)
271            .client_order_id(ClientOrderId::from(order_id))
272            .venue_order_id(VenueOrderId::from(order_id))
273            .trade_id(TradeId::from(trade_id))
274            .order_side(side)
275            .last_qty(Quantity::from(qty))
276            .last_px(Price::from(price))
277            .currency(instrument.quote_currency)
278            .ts_event(ts.into())
279            .ts_init((ts + 1).into())
280            .position_id(PositionId::from("P-001"))
281            .maybe_commission(commission)
282            .build()
283    }
284
285    fn make_position(ts: u64) -> Position {
286        let instrument = currency_pair_btcusdt();
287        let fill = make_fill(
288            &instrument,
289            OrderSide::Buy,
290            "1.0",
291            "50000.0",
292            "T-1",
293            "O-1",
294            ts,
295            None,
296        );
297        let any = InstrumentAny::CurrencyPair(instrument);
298        Position::new(&any, fill)
299    }
300
301    #[rstest]
302    fn test_encode_positions_schema() {
303        let batch = encode_positions(&[]).unwrap();
304        let schema = batch.schema();
305        let fields = schema.fields();
306        assert_eq!(fields.len(), 37);
307        assert_eq!(fields[0].name(), "trader_id");
308        assert_eq!(fields[0].data_type(), &DataType::Utf8);
309        assert_eq!(fields[9].name(), "signed_qty");
310        assert_eq!(fields[9].data_type(), &DataType::Float64);
311        assert_eq!(fields[12].name(), "price_precision");
312        assert_eq!(fields[12].data_type(), &DataType::UInt8);
313        assert_eq!(fields[15].name(), "is_inverse");
314        assert_eq!(fields[15].data_type(), &DataType::Boolean);
315        assert_eq!(fields[21].name(), "ts_init");
316        assert_eq!(
317            fields[21].data_type(),
318            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
319        );
320        assert_eq!(fields[25].name(), "duration_ns");
321        assert_eq!(fields[25].data_type(), &DataType::UInt64);
322        assert_eq!(fields[35].name(), "event_count");
323        assert_eq!(fields[35].data_type(), &DataType::UInt32);
324    }
325
326    #[rstest]
327    fn test_encode_positions_empty() {
328        let batch = encode_positions(&[]).unwrap();
329        assert_eq!(batch.num_rows(), 0);
330        assert_eq!(batch.schema().fields().len(), 37);
331    }
332
333    #[rstest]
334    fn test_encode_positions_values() {
335        let positions = vec![make_position(1_000_000)];
336        let batch = encode_positions(&positions).unwrap();
337
338        assert_eq!(batch.num_rows(), 1);
339
340        let trader_id_col = batch
341            .column(0)
342            .as_any()
343            .downcast_ref::<StringArray>()
344            .unwrap();
345        let quantity_col = batch
346            .column(10)
347            .as_any()
348            .downcast_ref::<Float64Array>()
349            .unwrap();
350        let price_precision_col = batch
351            .column(12)
352            .as_any()
353            .downcast_ref::<UInt8Array>()
354            .unwrap();
355        let is_currency_pair_col = batch
356            .column(16)
357            .as_any()
358            .downcast_ref::<BooleanArray>()
359            .unwrap();
360        let ts_opened_col = batch
361            .column(22)
362            .as_any()
363            .downcast_ref::<TimestampNanosecondArray>()
364            .unwrap();
365        let duration_col = batch
366            .column(25)
367            .as_any()
368            .downcast_ref::<UInt64Array>()
369            .unwrap();
370        let event_count_col = batch
371            .column(35)
372            .as_any()
373            .downcast_ref::<UInt32Array>()
374            .unwrap();
375
376        assert_eq!(trader_id_col.value(0), "TRADER-001");
377        assert!((quantity_col.value(0) - 1.0).abs() < 1e-9);
378        assert_eq!(price_precision_col.value(0), 2);
379        assert!(is_currency_pair_col.value(0));
380        assert_eq!(ts_opened_col.value(0), 1_000_000);
381        assert_eq!(duration_col.value(0), 0);
382        assert_eq!(event_count_col.value(0), 1);
383    }
384
385    #[rstest]
386    fn test_encode_positions_nullable_fields() {
387        let positions = vec![make_position(1_000)];
388        let batch = encode_positions(&positions).unwrap();
389
390        let closing_order_id_col = batch
391            .column(6)
392            .as_any()
393            .downcast_ref::<StringArray>()
394            .unwrap();
395        let ts_closed_col = batch
396            .column(24)
397            .as_any()
398            .downcast_ref::<TimestampNanosecondArray>()
399            .unwrap();
400        let avg_px_close_col = batch
401            .column(27)
402            .as_any()
403            .downcast_ref::<Float64Array>()
404            .unwrap();
405
406        assert!(closing_order_id_col.is_null(0));
407        assert!(ts_closed_col.is_null(0));
408        assert!(avg_px_close_col.is_null(0));
409    }
410
411    #[rstest]
412    fn test_encode_positions_trade_ids_sorted() {
413        let instrument = currency_pair_btcusdt();
414        let any = InstrumentAny::CurrencyPair(instrument.clone());
415        let open = make_fill(
416            &instrument,
417            OrderSide::Buy,
418            "1.0",
419            "50000.0",
420            "T-Z",
421            "O-1",
422            1_000,
423            None,
424        );
425        let add = make_fill(
426            &instrument,
427            OrderSide::Buy,
428            "1.0",
429            "50000.0",
430            "T-A",
431            "O-2",
432            2_000,
433            None,
434        );
435        let mut position = Position::new(&any, open);
436        position.apply(&add);
437
438        let batch = encode_positions(&[position]).unwrap();
439        let trade_ids_col = batch
440            .column(31)
441            .as_any()
442            .downcast_ref::<StringArray>()
443            .unwrap();
444
445        let parsed: Vec<String> = serde_json::from_str(trade_ids_col.value(0)).unwrap();
446        assert_eq!(parsed, vec!["T-A".to_string(), "T-Z".to_string()]);
447    }
448
449    #[rstest]
450    fn test_encode_positions_closed() {
451        let instrument = currency_pair_btcusdt();
452        let any = InstrumentAny::CurrencyPair(instrument.clone());
453        let open = make_fill(
454            &instrument,
455            OrderSide::Buy,
456            "1.0",
457            "50000.0",
458            "T-1",
459            "O-1",
460            1_000,
461            None,
462        );
463        let close = make_fill(
464            &instrument,
465            OrderSide::Sell,
466            "1.0",
467            "50500.0",
468            "T-2",
469            "O-2",
470            5_000,
471            None,
472        );
473        let mut position = Position::new(&any, open);
474        position.apply(&close);
475
476        let batch = encode_positions(&[position]).unwrap();
477        let closing_order_id_col = batch
478            .column(6)
479            .as_any()
480            .downcast_ref::<StringArray>()
481            .unwrap();
482        let ts_closed_col = batch
483            .column(24)
484            .as_any()
485            .downcast_ref::<TimestampNanosecondArray>()
486            .unwrap();
487        let duration_col = batch
488            .column(25)
489            .as_any()
490            .downcast_ref::<UInt64Array>()
491            .unwrap();
492        let avg_px_close_col = batch
493            .column(27)
494            .as_any()
495            .downcast_ref::<Float64Array>()
496            .unwrap();
497        let realized_pnl_amount_col = batch
498            .column(29)
499            .as_any()
500            .downcast_ref::<Float64Array>()
501            .unwrap();
502        let realized_pnl_currency_col = batch
503            .column(30)
504            .as_any()
505            .downcast_ref::<StringArray>()
506            .unwrap();
507        let event_count_col = batch
508            .column(35)
509            .as_any()
510            .downcast_ref::<UInt32Array>()
511            .unwrap();
512
513        assert_eq!(closing_order_id_col.value(0), "O-2");
514        assert!(!ts_closed_col.is_null(0));
515        assert_eq!(ts_closed_col.value(0), 5_000);
516        assert_eq!(duration_col.value(0), 4_000);
517        assert!((avg_px_close_col.value(0) - 50_500.0).abs() < 1e-9);
518        assert!(!realized_pnl_amount_col.is_null(0));
519        assert_eq!(realized_pnl_currency_col.value(0), "USDT");
520        assert_eq!(event_count_col.value(0), 2);
521    }
522
523    #[rstest]
524    fn test_encode_positions_commissions_sorted() {
525        let instrument = currency_pair_btcusdt();
526        let any = InstrumentAny::CurrencyPair(instrument.clone());
527        let usdt_fill = make_fill(
528            &instrument,
529            OrderSide::Buy,
530            "1.0",
531            "50000.0",
532            "T-1",
533            "O-1",
534            1_000,
535            Some(Money::from("0.50 USDT")),
536        );
537        let btc_fill = make_fill(
538            &instrument,
539            OrderSide::Buy,
540            "1.0",
541            "50000.0",
542            "T-2",
543            "O-2",
544            2_000,
545            Some(Money::from("0.00001 BTC")),
546        );
547        let mut position = Position::new(&any, usdt_fill);
548        position.apply(&btc_fill);
549
550        let batch = encode_positions(&[position]).unwrap();
551        let commissions_col = batch
552            .column(34)
553            .as_any()
554            .downcast_ref::<StringArray>()
555            .unwrap();
556
557        let parsed: Vec<(String, f64)> = serde_json::from_str(commissions_col.value(0)).unwrap();
558        let currencies: Vec<&str> = parsed.iter().map(|(c, _)| c.as_str()).collect();
559        assert_eq!(currencies, vec!["BTC", "USDT"]);
560    }
561}
nautilus_serialization/arrow/display/position.rs

nautilus_serialization/arrow/display/
position.rs
