nautilus_serialization/arrow/display/
instrument.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 [`InstrumentAny`].
17//!
18//! Emits a single schema built from the common [`Instrument`] trait surface,
19//! plus the dated/option accessors (`strike_price`, `activation_ns`,
20//! `expiration_ns`, `option_kind`) that are uniformly reachable across
21//! variants. Variants that do not expose a given accessor emit a null for
22//! that row, so mixed-type instrument batches (spot, perp, future, option,
23//! equity, etc.) flow through one Perspective table.
24//!
25//! Variant-only metadata that is not reachable through the trait (e.g.
26//! `BettingInstrument::market_id`, `BinaryOption::outcome`,
27//! `FuturesSpread::strategy_type`) is intentionally not emitted. Consumers
28//! that need those fields should encode the concrete variant through the
29//! open storage encoders in the parent [`crate::arrow`] module.
30
31use std::sync::Arc;
32
33use arrow::{
34    array::{
35        BooleanBuilder, Float64Builder, StringBuilder, TimestampNanosecondBuilder, UInt8Builder,
36    },
37    datatypes::Schema,
38    error::ArrowError,
39    record_batch::RecordBatch,
40};
41use nautilus_model::instruments::{Instrument, InstrumentAny};
42use rust_decimal::prelude::ToPrimitive;
43
44use super::{
45    bool_field, float64_field, money_to_f64, price_to_f64, quantity_to_f64, timestamp_field,
46    uint8_field, unix_nanos_to_i64, utf8_field,
47};
48use crate::arrow::timestamp_data_type;
49
50/// Returns the display-mode Arrow schema for [`InstrumentAny`].
51#[must_use]
52pub fn instrument_schema() -> Schema {
53    Schema::new(vec![
54        utf8_field("instrument_id", false),
55        utf8_field("symbol", false),
56        utf8_field("venue", false),
57        utf8_field("instrument_type", false),
58        utf8_field("raw_symbol", false),
59        utf8_field("asset_class", false),
60        utf8_field("instrument_class", false),
61        utf8_field("underlying", true),
62        utf8_field("base_currency", true),
63        utf8_field("quote_currency", false),
64        utf8_field("settlement_currency", false),
65        utf8_field("isin", true),
66        utf8_field("option_kind", true),
67        utf8_field("exchange", true),
68        float64_field("strike_price", true),
69        timestamp_field("activation_ns", true),
70        timestamp_field("expiration_ns", true),
71        bool_field("is_inverse", false),
72        bool_field("is_quanto", false),
73        uint8_field("price_precision", false),
74        uint8_field("size_precision", false),
75        float64_field("price_increment", false),
76        float64_field("size_increment", false),
77        float64_field("multiplier", false),
78        float64_field("lot_size", true),
79        float64_field("max_quantity", true),
80        float64_field("min_quantity", true),
81        float64_field("max_notional_amount", true),
82        utf8_field("max_notional_currency", true),
83        float64_field("min_notional_amount", true),
84        utf8_field("min_notional_currency", true),
85        float64_field("max_price", true),
86        float64_field("min_price", true),
87        float64_field("margin_init", false),
88        float64_field("margin_maint", false),
89        timestamp_field("ts_event", false),
90        timestamp_field("ts_init", false),
91    ])
92}
93
94/// Returns a stable name for the [`InstrumentAny`] variant.
95fn instrument_type_name(instrument: &InstrumentAny) -> &'static str {
96    match instrument {
97        InstrumentAny::Betting(_) => "BettingInstrument",
98        InstrumentAny::BinaryOption(_) => "BinaryOption",
99        InstrumentAny::Cfd(_) => "Cfd",
100        InstrumentAny::Commodity(_) => "Commodity",
101        InstrumentAny::CryptoFuture(_) => "CryptoFuture",
102        InstrumentAny::CryptoFuturesSpread(_) => "CryptoFuturesSpread",
103        InstrumentAny::CryptoOption(_) => "CryptoOption",
104        InstrumentAny::CryptoOptionSpread(_) => "CryptoOptionSpread",
105        InstrumentAny::CryptoPerpetual(_) => "CryptoPerpetual",
106        InstrumentAny::CurrencyPair(_) => "CurrencyPair",
107        InstrumentAny::Equity(_) => "Equity",
108        InstrumentAny::FuturesContract(_) => "FuturesContract",
109        InstrumentAny::FuturesSpread(_) => "FuturesSpread",
110        InstrumentAny::IndexInstrument(_) => "IndexInstrument",
111        InstrumentAny::OptionContract(_) => "OptionContract",
112        InstrumentAny::OptionSpread(_) => "OptionSpread",
113        InstrumentAny::PerpetualContract(_) => "PerpetualContract",
114        InstrumentAny::TokenizedAsset(_) => "TokenizedAsset",
115    }
116}
117
118/// Encodes instruments as a display-friendly Arrow [`RecordBatch`].
119///
120/// Emits a single schema built from the common [`Instrument`] trait surface.
121/// `Utf8` columns carry identifiers and enum names, `Float64` columns carry
122/// prices and quantities, `Timestamp(Nanosecond)` columns carry activation,
123/// expiration, and bookkeeping timestamps, and `Boolean` columns carry
124/// `is_inverse`/`is_quanto`. Trait accessors that are not applicable to a
125/// row (e.g. `strike_price` on a spot pair) emit as nulls, so mixed-type
126/// batches round-trip cleanly. Variant-only metadata not reachable through
127/// the trait is intentionally omitted; see the module-level comment.
128///
129/// Returns an empty [`RecordBatch`] with the correct schema when `data` is empty.
130///
131/// # Errors
132///
133/// Returns an error if the Arrow `RecordBatch` cannot be constructed.
134pub fn encode_instruments(data: &[InstrumentAny]) -> Result<RecordBatch, ArrowError> {
135    let mut instrument_id = StringBuilder::new();
136    let mut symbol = StringBuilder::new();
137    let mut venue = StringBuilder::new();
138    let mut instrument_type = StringBuilder::new();
139    let mut raw_symbol = StringBuilder::new();
140    let mut asset_class = StringBuilder::new();
141    let mut instrument_class = StringBuilder::new();
142    let mut underlying = StringBuilder::new();
143    let mut base_currency = StringBuilder::new();
144    let mut quote_currency = StringBuilder::new();
145    let mut settlement_currency = StringBuilder::new();
146    let mut isin = StringBuilder::new();
147    let mut option_kind = StringBuilder::new();
148    let mut exchange = StringBuilder::new();
149    let mut strike_price = Float64Builder::with_capacity(data.len());
150    let mut activation_ns =
151        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
152    let mut expiration_ns =
153        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
154    let mut is_inverse = BooleanBuilder::with_capacity(data.len());
155    let mut is_quanto = BooleanBuilder::with_capacity(data.len());
156    let mut price_precision = UInt8Builder::with_capacity(data.len());
157    let mut size_precision = UInt8Builder::with_capacity(data.len());
158    let mut price_increment = Float64Builder::with_capacity(data.len());
159    let mut size_increment = Float64Builder::with_capacity(data.len());
160    let mut multiplier = Float64Builder::with_capacity(data.len());
161    let mut lot_size = Float64Builder::with_capacity(data.len());
162    let mut max_quantity = Float64Builder::with_capacity(data.len());
163    let mut min_quantity = Float64Builder::with_capacity(data.len());
164    let mut max_notional_amount = Float64Builder::with_capacity(data.len());
165    let mut max_notional_currency = StringBuilder::new();
166    let mut min_notional_amount = Float64Builder::with_capacity(data.len());
167    let mut min_notional_currency = StringBuilder::new();
168    let mut max_price = Float64Builder::with_capacity(data.len());
169    let mut min_price = Float64Builder::with_capacity(data.len());
170    let mut margin_init = Float64Builder::with_capacity(data.len());
171    let mut margin_maint = Float64Builder::with_capacity(data.len());
172    let mut ts_event =
173        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
174    let mut ts_init =
175        TimestampNanosecondBuilder::with_capacity(data.len()).with_data_type(timestamp_data_type());
176
177    for instrument in data {
178        instrument_id.append_value(instrument.id().to_string());
179        symbol.append_value(instrument.symbol());
180        venue.append_value(instrument.venue());
181        instrument_type.append_value(instrument_type_name(instrument));
182        raw_symbol.append_value(instrument.raw_symbol());
183        asset_class.append_value(format!("{}", instrument.asset_class()));
184        instrument_class.append_value(format!("{}", instrument.instrument_class()));
185        underlying.append_option(instrument.underlying().map(|v| v.to_string()));
186        base_currency.append_option(instrument.base_currency().map(|v| v.to_string()));
187        quote_currency.append_value(instrument.quote_currency().to_string());
188        settlement_currency.append_value(instrument.settlement_currency().to_string());
189        isin.append_option(instrument.isin().map(|v| v.to_string()));
190        option_kind.append_option(instrument.option_kind().map(|v| format!("{v}")));
191        exchange.append_option(instrument.exchange().map(|v| v.to_string()));
192        strike_price.append_option(instrument.strike_price().map(|v| price_to_f64(&v)));
193        activation_ns.append_option(
194            instrument
195                .activation_ns()
196                .map(|v| unix_nanos_to_i64(v.as_u64())),
197        );
198        expiration_ns.append_option(
199            instrument
200                .expiration_ns()
201                .map(|v| unix_nanos_to_i64(v.as_u64())),
202        );
203        is_inverse.append_value(instrument.is_inverse());
204        is_quanto.append_value(instrument.is_quanto());
205        price_precision.append_value(instrument.price_precision());
206        size_precision.append_value(instrument.size_precision());
207        price_increment.append_value(price_to_f64(&instrument.price_increment()));
208        size_increment.append_value(quantity_to_f64(&instrument.size_increment()));
209        multiplier.append_value(quantity_to_f64(&instrument.multiplier()));
210        lot_size.append_option(instrument.lot_size().map(|v| quantity_to_f64(&v)));
211        max_quantity.append_option(instrument.max_quantity().map(|v| quantity_to_f64(&v)));
212        min_quantity.append_option(instrument.min_quantity().map(|v| quantity_to_f64(&v)));
213        max_notional_amount.append_option(instrument.max_notional().map(|v| money_to_f64(&v)));
214        max_notional_currency
215            .append_option(instrument.max_notional().map(|v| v.currency.to_string()));
216        min_notional_amount.append_option(instrument.min_notional().map(|v| money_to_f64(&v)));
217        min_notional_currency
218            .append_option(instrument.min_notional().map(|v| v.currency.to_string()));
219        max_price.append_option(instrument.max_price().map(|v| price_to_f64(&v)));
220        min_price.append_option(instrument.min_price().map(|v| price_to_f64(&v)));
221        margin_init.append_value(instrument.margin_init().to_f64().unwrap_or(f64::NAN));
222        margin_maint.append_value(instrument.margin_maint().to_f64().unwrap_or(f64::NAN));
223        ts_event.append_value(unix_nanos_to_i64(instrument.ts_event().as_u64()));
224        ts_init.append_value(unix_nanos_to_i64(instrument.ts_init().as_u64()));
225    }
226
227    RecordBatch::try_new(
228        Arc::new(instrument_schema()),
229        vec![
230            Arc::new(instrument_id.finish()),
231            Arc::new(symbol.finish()),
232            Arc::new(venue.finish()),
233            Arc::new(instrument_type.finish()),
234            Arc::new(raw_symbol.finish()),
235            Arc::new(asset_class.finish()),
236            Arc::new(instrument_class.finish()),
237            Arc::new(underlying.finish()),
238            Arc::new(base_currency.finish()),
239            Arc::new(quote_currency.finish()),
240            Arc::new(settlement_currency.finish()),
241            Arc::new(isin.finish()),
242            Arc::new(option_kind.finish()),
243            Arc::new(exchange.finish()),
244            Arc::new(strike_price.finish()),
245            Arc::new(activation_ns.finish()),
246            Arc::new(expiration_ns.finish()),
247            Arc::new(is_inverse.finish()),
248            Arc::new(is_quanto.finish()),
249            Arc::new(price_precision.finish()),
250            Arc::new(size_precision.finish()),
251            Arc::new(price_increment.finish()),
252            Arc::new(size_increment.finish()),
253            Arc::new(multiplier.finish()),
254            Arc::new(lot_size.finish()),
255            Arc::new(max_quantity.finish()),
256            Arc::new(min_quantity.finish()),
257            Arc::new(max_notional_amount.finish()),
258            Arc::new(max_notional_currency.finish()),
259            Arc::new(min_notional_amount.finish()),
260            Arc::new(min_notional_currency.finish()),
261            Arc::new(max_price.finish()),
262            Arc::new(min_price.finish()),
263            Arc::new(margin_init.finish()),
264            Arc::new(margin_maint.finish()),
265            Arc::new(ts_event.finish()),
266            Arc::new(ts_init.finish()),
267        ],
268    )
269}
270
271#[cfg(test)]
272mod tests {
273    use arrow::{
274        array::{Array, BooleanArray, Float64Array, StringArray, TimestampNanosecondArray},
275        datatypes::{DataType, TimeUnit},
276    };
277    use nautilus_model::{
278        instruments::{
279            InstrumentAny,
280            stubs::{
281                betting, binary_option, btcusd_bybit, cfd_gold, commodity_gold,
282                crypto_future_btcusdt, crypto_option_btc_deribit, crypto_perpetual_ethusdt,
283                currency_pair_btcusdt, equity_aapl, futures_contract_es, futures_spread_es,
284                index_instrument_spx, option_contract_appl, option_spread,
285                perpetual_contract_eurusd, tokenized_asset_aaplx,
286            },
287        },
288        types::{Price, Quantity},
289    };
290    use rstest::rstest;
291
292    use super::*;
293
294    fn spot() -> InstrumentAny {
295        InstrumentAny::CurrencyPair(currency_pair_btcusdt())
296    }
297
298    fn all_variants() -> Vec<(InstrumentAny, &'static str)> {
299        vec![
300            (InstrumentAny::Betting(betting()), "BettingInstrument"),
301            (InstrumentAny::BinaryOption(binary_option()), "BinaryOption"),
302            (InstrumentAny::Cfd(cfd_gold()), "Cfd"),
303            (InstrumentAny::Commodity(commodity_gold()), "Commodity"),
304            (
305                InstrumentAny::CryptoFuture(crypto_future_btcusdt(
306                    2,
307                    6,
308                    Price::from("0.01"),
309                    Quantity::from("0.000001"),
310                )),
311                "CryptoFuture",
312            ),
313            (
314                InstrumentAny::CryptoOption(crypto_option_btc_deribit(
315                    3,
316                    1,
317                    Price::from("0.001"),
318                    Quantity::from("0.1"),
319                )),
320                "CryptoOption",
321            ),
322            (
323                InstrumentAny::CryptoPerpetual(crypto_perpetual_ethusdt()),
324                "CryptoPerpetual",
325            ),
326            (
327                InstrumentAny::CurrencyPair(currency_pair_btcusdt()),
328                "CurrencyPair",
329            ),
330            (InstrumentAny::Equity(equity_aapl()), "Equity"),
331            (
332                InstrumentAny::FuturesContract(futures_contract_es(None, None)),
333                "FuturesContract",
334            ),
335            (
336                InstrumentAny::FuturesSpread(futures_spread_es()),
337                "FuturesSpread",
338            ),
339            (
340                InstrumentAny::IndexInstrument(index_instrument_spx()),
341                "IndexInstrument",
342            ),
343            (
344                InstrumentAny::OptionContract(option_contract_appl()),
345                "OptionContract",
346            ),
347            (InstrumentAny::OptionSpread(option_spread()), "OptionSpread"),
348            (
349                InstrumentAny::PerpetualContract(perpetual_contract_eurusd()),
350                "PerpetualContract",
351            ),
352            (
353                InstrumentAny::TokenizedAsset(tokenized_asset_aaplx()),
354                "TokenizedAsset",
355            ),
356        ]
357    }
358
359    #[rstest]
360    fn test_encode_instruments_schema() {
361        let batch = encode_instruments(&[]).unwrap();
362        let schema = batch.schema();
363        let fields = schema.fields();
364        assert_eq!(fields.len(), 37);
365        assert_eq!(fields[0].name(), "instrument_id");
366        assert_eq!(fields[0].data_type(), &DataType::Utf8);
367        assert_eq!(fields[14].name(), "strike_price");
368        assert_eq!(fields[14].data_type(), &DataType::Float64);
369        assert_eq!(fields[17].name(), "is_inverse");
370        assert_eq!(fields[17].data_type(), &DataType::Boolean);
371        assert_eq!(fields[19].name(), "price_precision");
372        assert_eq!(fields[19].data_type(), &DataType::UInt8);
373        assert_eq!(fields[33].name(), "margin_init");
374        assert_eq!(fields[33].data_type(), &DataType::Float64);
375        assert_eq!(fields[35].name(), "ts_event");
376        assert_eq!(
377            fields[35].data_type(),
378            &DataType::Timestamp(TimeUnit::Nanosecond, Some("UTC".into()))
379        );
380    }
381
382    #[rstest]
383    fn test_encode_instruments_empty() {
384        let batch = encode_instruments(&[]).unwrap();
385        assert_eq!(batch.num_rows(), 0);
386        assert_eq!(batch.schema().fields().len(), 37);
387    }
388
389    #[rstest]
390    fn test_encode_instruments_spot_values() {
391        let instruments = vec![spot()];
392        let batch = encode_instruments(&instruments).unwrap();
393
394        assert_eq!(batch.num_rows(), 1);
395
396        let instrument_type_col = batch
397            .column(3)
398            .as_any()
399            .downcast_ref::<StringArray>()
400            .unwrap();
401        let strike_price_col = batch
402            .column(14)
403            .as_any()
404            .downcast_ref::<Float64Array>()
405            .unwrap();
406        let activation_col = batch
407            .column(15)
408            .as_any()
409            .downcast_ref::<TimestampNanosecondArray>()
410            .unwrap();
411        let expiration_col = batch
412            .column(16)
413            .as_any()
414            .downcast_ref::<TimestampNanosecondArray>()
415            .unwrap();
416        let is_inverse_col = batch
417            .column(17)
418            .as_any()
419            .downcast_ref::<BooleanArray>()
420            .unwrap();
421        let price_increment_col = batch
422            .column(21)
423            .as_any()
424            .downcast_ref::<Float64Array>()
425            .unwrap();
426
427        assert_eq!(instrument_type_col.value(0), "CurrencyPair");
428        assert!(strike_price_col.is_null(0));
429        assert!(activation_col.is_null(0));
430        assert!(expiration_col.is_null(0));
431        assert!(!is_inverse_col.value(0));
432        assert!(price_increment_col.value(0) > 0.0);
433    }
434
435    #[rstest]
436    fn test_encode_instruments_mixed_variants_preserves_per_row_nulls() {
437        let instruments = vec![
438            spot(),
439            InstrumentAny::Equity(equity_aapl()),
440            InstrumentAny::OptionContract(option_contract_appl()),
441        ];
442        let batch = encode_instruments(&instruments).unwrap();
443
444        assert_eq!(batch.num_rows(), 3);
445
446        let instrument_type_col = batch
447            .column(3)
448            .as_any()
449            .downcast_ref::<StringArray>()
450            .unwrap();
451        let strike_price_col = batch
452            .column(14)
453            .as_any()
454            .downcast_ref::<Float64Array>()
455            .unwrap();
456        let expiration_col = batch
457            .column(16)
458            .as_any()
459            .downcast_ref::<TimestampNanosecondArray>()
460            .unwrap();
461        let base_currency_col = batch
462            .column(8)
463            .as_any()
464            .downcast_ref::<StringArray>()
465            .unwrap();
466
467        assert_eq!(instrument_type_col.value(0), "CurrencyPair");
468        assert_eq!(instrument_type_col.value(1), "Equity");
469        assert_eq!(instrument_type_col.value(2), "OptionContract");
470
471        // Only the option carries a strike + expiration
472        assert!(strike_price_col.is_null(0));
473        assert!(strike_price_col.is_null(1));
474        assert!(!strike_price_col.is_null(2));
475        assert!(expiration_col.is_null(0));
476        assert!(expiration_col.is_null(1));
477        assert!(!expiration_col.is_null(2));
478
479        // Only the spot pair carries a base currency
480        assert!(!base_currency_col.is_null(0));
481        assert!(base_currency_col.is_null(1));
482    }
483
484    #[rstest]
485    fn test_encode_instruments_shared_schema_across_batches() {
486        let a = encode_instruments(&[spot()]).unwrap();
487        let b = encode_instruments(&[InstrumentAny::Equity(equity_aapl())]).unwrap();
488        assert_eq!(a.schema(), b.schema());
489    }
490
491    #[rstest]
492    fn test_encode_instruments_all_variant_names() {
493        let variants = all_variants();
494        assert_eq!(variants.len(), 16, "all InstrumentAny variants covered");
495
496        let instruments: Vec<InstrumentAny> = variants.iter().map(|(v, _)| v.clone()).collect();
497        let batch = encode_instruments(&instruments).unwrap();
498        let instrument_type_col = batch
499            .column(3)
500            .as_any()
501            .downcast_ref::<StringArray>()
502            .unwrap();
503
504        for (row, (_, expected)) in variants.iter().enumerate() {
505            assert_eq!(instrument_type_col.value(row), *expected);
506        }
507    }
508
509    #[rstest]
510    fn test_encode_instruments_inverse_perpetual() {
511        let instruments = vec![InstrumentAny::CryptoPerpetual(btcusd_bybit())];
512        let batch = encode_instruments(&instruments).unwrap();
513
514        let instrument_type_col = batch
515            .column(3)
516            .as_any()
517            .downcast_ref::<StringArray>()
518            .unwrap();
519        let settlement_currency_col = batch
520            .column(10)
521            .as_any()
522            .downcast_ref::<StringArray>()
523            .unwrap();
524        let is_inverse_col = batch
525            .column(17)
526            .as_any()
527            .downcast_ref::<BooleanArray>()
528            .unwrap();
529        let max_notional_amount_col = batch
530            .column(27)
531            .as_any()
532            .downcast_ref::<Float64Array>()
533            .unwrap();
534        let max_notional_currency_col = batch
535            .column(28)
536            .as_any()
537            .downcast_ref::<StringArray>()
538            .unwrap();
539        let min_notional_amount_col = batch
540            .column(29)
541            .as_any()
542            .downcast_ref::<Float64Array>()
543            .unwrap();
544        let min_notional_currency_col = batch
545            .column(30)
546            .as_any()
547            .downcast_ref::<StringArray>()
548            .unwrap();
549
550        assert_eq!(instrument_type_col.value(0), "CryptoPerpetual");
551        assert_eq!(settlement_currency_col.value(0), "BTC");
552        assert!(is_inverse_col.value(0));
553        assert!((max_notional_amount_col.value(0) - 10_000_000.0).abs() < 1e-9);
554        assert_eq!(max_notional_currency_col.value(0), "USD");
555        assert!((min_notional_amount_col.value(0) - 1.0).abs() < 1e-9);
556        assert_eq!(min_notional_currency_col.value(0), "USD");
557
558        let margin_init_col = batch
559            .column(33)
560            .as_any()
561            .downcast_ref::<Float64Array>()
562            .unwrap();
563        let margin_maint_col = batch
564            .column(34)
565            .as_any()
566            .downcast_ref::<Float64Array>()
567            .unwrap();
568        assert!((margin_init_col.value(0) - 0.01).abs() < 1e-9);
569        assert!((margin_maint_col.value(0) - 0.0035).abs() < 1e-9);
570    }
571}
nautilus_serialization/arrow/display/instrument.rs

nautilus_serialization/arrow/display/
instrument.rs
