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

nautilus_serialization/arrow/display/
position.rs
