nautilus_serialization/arrow/instrument/
mod.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//! Arrow serialization for instruments.
17//!
18//! `InstrumentAny` acts as a dispatcher that routes to the appropriate concrete instrument type's
19//! Arrow serialization implementation. Each concrete instrument type implements its own schema
20//! with all fields as columns (wide schema approach), matching the Python implementation.
21
22use std::{any::type_name, borrow::Borrow, collections::HashMap, fmt, str::FromStr};
23
24use arrow::{
25    array::{Array, StringArray},
26    datatypes::Schema,
27    error::ArrowError,
28    record_batch::RecordBatch,
29};
30use nautilus_model::{
31    instruments::{
32        Instrument, InstrumentAny, betting::BettingInstrument, binary_option::BinaryOption,
33        cfd::Cfd, commodity::Commodity, crypto_future::CryptoFuture,
34        crypto_futures_spread::CryptoFuturesSpread, crypto_option::CryptoOption,
35        crypto_option_spread::CryptoOptionSpread, crypto_perpetual::CryptoPerpetual,
36        currency_pair::CurrencyPair, equity::Equity, futures_contract::FuturesContract,
37        futures_spread::FuturesSpread, index_instrument::IndexInstrument,
38        option_contract::OptionContract, option_spread::OptionSpread,
39        perpetual_contract::PerpetualContract, tokenized_asset::TokenizedAsset,
40    },
41    types::{Currency, Price, Quantity},
42};
43
44#[allow(unused)]
45use crate::arrow::{
46    ArrowSchemaProvider, Data, DecodeDataFromRecordBatch, DecodeFromRecordBatch,
47    EncodeToRecordBatch, EncodingError, KEY_INSTRUMENT_ID, KEY_TYPE_NAME, first_logical_null,
48};
49
50pub mod betting;
51pub mod binary_option;
52pub mod cfd;
53pub mod commodity;
54pub mod crypto_future;
55pub mod crypto_futures_spread;
56pub mod crypto_option;
57pub mod crypto_option_spread;
58pub mod crypto_perpetual;
59pub mod currency_pair;
60pub mod equity;
61pub mod futures_contract;
62pub mod futures_spread;
63pub mod index_instrument;
64pub mod option_contract;
65pub mod option_spread;
66pub mod perpetual_contract;
67pub mod tokenized_asset;
68
69// Columns added after the original schemas are read by name and yield `None` when absent,
70// so fragments written before the column existed decode exactly as they did previously.
71pub(crate) fn optional_quantity_value(
72    values: Option<&StringArray>,
73    field: &'static str,
74    row: usize,
75) -> Result<Option<Quantity>, EncodingError> {
76    let Some(column) = values else {
77        return Ok(None);
78    };
79
80    if column.is_null(row) {
81        return Ok(None);
82    }
83
84    Quantity::from_str(column.value(row))
85        .map(Some)
86        .map_err(|e| EncodingError::ParseError(field, format!("row {row}: {e}")))
87}
88
89pub(crate) fn optional_price_value(
90    values: Option<&StringArray>,
91    field: &'static str,
92    row: usize,
93) -> Result<Option<Price>, EncodingError> {
94    let Some(column) = values else {
95        return Ok(None);
96    };
97
98    if column.is_null(row) {
99        return Ok(None);
100    }
101
102    Price::from_str(column.value(row))
103        .map(Some)
104        .map_err(|e| EncodingError::ParseError(field, format!("row {row}: {e}")))
105}
106
107// Errors on empty/whitespace codes so corrupted rows surface as ParseError,
108// instead of silently registering as a fallback currency. Known codes resolve
109// from CURRENCY_MAP with original metadata; unknown non-empty codes fall back
110// to a new crypto currency to support newly listed exchange assets.
111pub(crate) fn decode_currency(
112    value: &str,
113    field: &'static str,
114    context: &'static str,
115    row: usize,
116) -> Result<Currency, EncodingError> {
117    let trimmed = value.trim();
118    if trimmed.is_empty() {
119        return Err(EncodingError::ParseError(
120            field,
121            format!("row {row}: empty currency code"),
122        ));
123    }
124
125    Ok(Currency::get_or_create_crypto_with_context(
126        trimmed,
127        Some(context),
128    ))
129}
130
131const INSTRUMENT_VALIDATION_FIELD: &str = "instrument";
132
133pub(crate) fn instrument_validation_error<T>(
134    row: usize,
135    error: impl fmt::Display,
136) -> EncodingError {
137    let type_name = type_name::<T>();
138    let instrument_type = type_name.rsplit("::").next().unwrap_or(type_name);
139
140    EncodingError::ParseError(
141        INSTRUMENT_VALIDATION_FIELD,
142        format!("row {row}: invalid {instrument_type}: {error}"),
143    )
144}
145
146pub(crate) fn validate_required_instrument_columns<T: ArrowSchemaProvider>(
147    metadata: &HashMap<String, String>,
148    record_batch: &RecordBatch,
149) -> Result<(), EncodingError> {
150    let expected = T::get_schema(Some(metadata.clone()));
151    let actual = record_batch.schema();
152    for field in expected
153        .fields()
154        .iter()
155        .filter(|field| !field.is_nullable())
156    {
157        let Ok(index) = actual.index_of(field.name()) else {
158            continue;
159        };
160
161        if let Some(row) = first_logical_null(record_batch.column(index).as_ref()) {
162            return Err(EncodingError::ParseError(
163                INSTRUMENT_VALIDATION_FIELD,
164                format!("required field '{}' is null at row {row}", field.name()),
165            ));
166        }
167    }
168
169    Ok(())
170}
171
172impl ArrowSchemaProvider for InstrumentAny {
173    fn get_schema(metadata: Option<HashMap<String, String>>) -> Schema {
174        let instrument_type = metadata
175            .as_ref()
176            .and_then(|m| m.get(KEY_TYPE_NAME))
177            .map_or("CurrencyPair", |s| s.as_str());
178
179        match instrument_type {
180            "BettingInstrument" => BettingInstrument::get_schema(metadata),
181            "BinaryOption" => BinaryOption::get_schema(metadata),
182            "Cfd" => Cfd::get_schema(metadata),
183            "Commodity" => Commodity::get_schema(metadata),
184            "CryptoFuture" => CryptoFuture::get_schema(metadata),
185            "CryptoFuturesSpread" => CryptoFuturesSpread::get_schema(metadata),
186            "CryptoOption" => CryptoOption::get_schema(metadata),
187            "CryptoOptionSpread" => CryptoOptionSpread::get_schema(metadata),
188            "CryptoPerpetual" => CryptoPerpetual::get_schema(metadata),
189            "CurrencyPair" => CurrencyPair::get_schema(metadata),
190            "Equity" => Equity::get_schema(metadata),
191            "FuturesContract" => FuturesContract::get_schema(metadata),
192            "FuturesSpread" => FuturesSpread::get_schema(metadata),
193            "IndexInstrument" => IndexInstrument::get_schema(metadata),
194            "OptionContract" => OptionContract::get_schema(metadata),
195            "OptionSpread" => OptionSpread::get_schema(metadata),
196            "PerpetualContract" => PerpetualContract::get_schema(metadata),
197            "TokenizedAsset" => TokenizedAsset::get_schema(metadata),
198            _ => {
199                // Fallback to CurrencyPair schema if type is unknown
200                CurrencyPair::get_schema(metadata)
201            }
202        }
203    }
204}
205
206impl EncodeToRecordBatch for InstrumentAny {
207    fn encode_batch<T>(
208        #[allow(unused)] metadata: &HashMap<String, String>,
209        data: &[T],
210    ) -> Result<RecordBatch, ArrowError>
211    where
212        T: std::borrow::Borrow<Self>,
213    {
214        if data.is_empty() {
215            return Err(ArrowError::InvalidArgumentError(
216                "Cannot encode empty instrument batch".to_string(),
217            ));
218        }
219
220        let mut by_type: HashMap<String, Vec<&Self>> = HashMap::new();
221
222        for instrument in data.iter().map(Borrow::borrow) {
223            let type_name = match instrument {
224                Self::Cfd(_) => "Cfd",
225                Self::Commodity(_) => "Commodity",
226                Self::CurrencyPair(_) => "CurrencyPair",
227                Self::Equity(_) => "Equity",
228                Self::CryptoFuture(_) => "CryptoFuture",
229                Self::CryptoFuturesSpread(_) => "CryptoFuturesSpread",
230                Self::CryptoPerpetual(_) => "CryptoPerpetual",
231                Self::CryptoOption(_) => "CryptoOption",
232                Self::CryptoOptionSpread(_) => "CryptoOptionSpread",
233                Self::FuturesContract(_) => "FuturesContract",
234                Self::FuturesSpread(_) => "FuturesSpread",
235                Self::IndexInstrument(_) => "IndexInstrument",
236                Self::OptionContract(_) => "OptionContract",
237                Self::OptionSpread(_) => "OptionSpread",
238                Self::BinaryOption(_) => "BinaryOption",
239                Self::Betting(_) => "BettingInstrument",
240                Self::PerpetualContract(_) => "PerpetualContract",
241                Self::TokenizedAsset(_) => "TokenizedAsset",
242            };
243            by_type
244                .entry(type_name.to_string())
245                .or_default()
246                .push(instrument);
247        }
248
249        if by_type.len() > 1 {
250            return Err(ArrowError::InvalidArgumentError(
251                "Cannot encode mixed instrument types in a single batch. Use separate batches for each type.".to_string(),
252            ));
253        }
254
255        let (type_name, instruments) = by_type.iter().next().unwrap();
256        match type_name.as_str() {
257            "Cfd" => {
258                let cfds: Vec<_> = instruments
259                    .iter()
260                    .map(|i| {
261                        if let Self::Cfd(c) = i {
262                            c
263                        } else {
264                            unreachable!()
265                        }
266                    })
267                    .cloned()
268                    .collect();
269                Cfd::encode_batch(metadata, &cfds)
270            }
271            "Commodity" => {
272                let commodities: Vec<_> = instruments
273                    .iter()
274                    .map(|i| {
275                        if let Self::Commodity(c) = i {
276                            c
277                        } else {
278                            unreachable!()
279                        }
280                    })
281                    .cloned()
282                    .collect();
283                Commodity::encode_batch(metadata, &commodities)
284            }
285            "BettingInstrument" => {
286                let betting: Vec<_> = instruments
287                    .iter()
288                    .map(|i| {
289                        if let Self::Betting(b) = i {
290                            b
291                        } else {
292                            unreachable!()
293                        }
294                    })
295                    .cloned()
296                    .collect();
297                BettingInstrument::encode_batch(metadata, &betting)
298            }
299            "BinaryOption" => {
300                let binary_options: Vec<_> = instruments
301                    .iter()
302                    .map(|i| {
303                        if let Self::BinaryOption(bo) = i {
304                            bo
305                        } else {
306                            unreachable!()
307                        }
308                    })
309                    .cloned()
310                    .collect();
311                BinaryOption::encode_batch(metadata, &binary_options)
312            }
313            "CryptoFuture" => {
314                let crypto_futures: Vec<_> = instruments
315                    .iter()
316                    .map(|i| {
317                        if let Self::CryptoFuture(cf) = i {
318                            cf
319                        } else {
320                            unreachable!()
321                        }
322                    })
323                    .cloned()
324                    .collect();
325                CryptoFuture::encode_batch(metadata, &crypto_futures)
326            }
327            "CryptoFuturesSpread" => {
328                let spreads: Vec<_> = instruments
329                    .iter()
330                    .map(|i| {
331                        if let Self::CryptoFuturesSpread(cfs) = i {
332                            cfs
333                        } else {
334                            unreachable!()
335                        }
336                    })
337                    .cloned()
338                    .collect();
339                CryptoFuturesSpread::encode_batch(metadata, &spreads)
340            }
341            "CryptoOption" => {
342                let crypto_options: Vec<_> = instruments
343                    .iter()
344                    .map(|i| {
345                        if let Self::CryptoOption(co) = i {
346                            co
347                        } else {
348                            unreachable!()
349                        }
350                    })
351                    .cloned()
352                    .collect();
353                CryptoOption::encode_batch(metadata, &crypto_options)
354            }
355            "CryptoOptionSpread" => {
356                let spreads: Vec<_> = instruments
357                    .iter()
358                    .map(|i| {
359                        if let Self::CryptoOptionSpread(cos) = i {
360                            cos
361                        } else {
362                            unreachable!()
363                        }
364                    })
365                    .cloned()
366                    .collect();
367                CryptoOptionSpread::encode_batch(metadata, &spreads)
368            }
369            "CryptoPerpetual" => {
370                let crypto_perps: Vec<_> = instruments
371                    .iter()
372                    .map(|i| {
373                        if let Self::CryptoPerpetual(cp) = i {
374                            cp
375                        } else {
376                            unreachable!()
377                        }
378                    })
379                    .cloned()
380                    .collect();
381                CryptoPerpetual::encode_batch(metadata, &crypto_perps)
382            }
383            "CurrencyPair" => {
384                let currency_pairs: Vec<_> = instruments
385                    .iter()
386                    .map(|i| {
387                        if let Self::CurrencyPair(cp) = i {
388                            cp
389                        } else {
390                            unreachable!()
391                        }
392                    })
393                    .cloned()
394                    .collect();
395                CurrencyPair::encode_batch(metadata, &currency_pairs)
396            }
397            "Equity" => {
398                let equities: Vec<_> = instruments
399                    .iter()
400                    .map(|i| {
401                        if let Self::Equity(e) = i {
402                            e
403                        } else {
404                            unreachable!()
405                        }
406                    })
407                    .cloned()
408                    .collect();
409                Equity::encode_batch(metadata, &equities)
410            }
411            "FuturesContract" => {
412                let futures_contracts: Vec<_> = instruments
413                    .iter()
414                    .map(|i| {
415                        if let Self::FuturesContract(fc) = i {
416                            fc
417                        } else {
418                            unreachable!()
419                        }
420                    })
421                    .cloned()
422                    .collect();
423                FuturesContract::encode_batch(metadata, &futures_contracts)
424            }
425            "FuturesSpread" => {
426                let futures_spreads: Vec<_> = instruments
427                    .iter()
428                    .map(|i| {
429                        if let Self::FuturesSpread(fs) = i {
430                            fs
431                        } else {
432                            unreachable!()
433                        }
434                    })
435                    .cloned()
436                    .collect();
437                FuturesSpread::encode_batch(metadata, &futures_spreads)
438            }
439            "IndexInstrument" => {
440                let index_instruments: Vec<_> = instruments
441                    .iter()
442                    .map(|i| {
443                        if let Self::IndexInstrument(ii) = i {
444                            ii
445                        } else {
446                            unreachable!()
447                        }
448                    })
449                    .cloned()
450                    .collect();
451                IndexInstrument::encode_batch(metadata, &index_instruments)
452            }
453            "OptionContract" => {
454                let option_contracts: Vec<_> = instruments
455                    .iter()
456                    .map(|i| {
457                        if let Self::OptionContract(oc) = i {
458                            oc
459                        } else {
460                            unreachable!()
461                        }
462                    })
463                    .cloned()
464                    .collect();
465                OptionContract::encode_batch(metadata, &option_contracts)
466            }
467            "OptionSpread" => {
468                let option_spreads: Vec<_> = instruments
469                    .iter()
470                    .map(|i| {
471                        if let Self::OptionSpread(os) = i {
472                            os
473                        } else {
474                            unreachable!()
475                        }
476                    })
477                    .cloned()
478                    .collect();
479                OptionSpread::encode_batch(metadata, &option_spreads)
480            }
481            "PerpetualContract" => {
482                let perpetual_contracts: Vec<_> = instruments
483                    .iter()
484                    .map(|i| {
485                        if let Self::PerpetualContract(pc) = i {
486                            pc
487                        } else {
488                            unreachable!()
489                        }
490                    })
491                    .cloned()
492                    .collect();
493                PerpetualContract::encode_batch(metadata, &perpetual_contracts)
494            }
495            "TokenizedAsset" => {
496                let tokenized_assets: Vec<_> = instruments
497                    .iter()
498                    .map(|i| {
499                        if let Self::TokenizedAsset(ta) = i {
500                            ta
501                        } else {
502                            unreachable!()
503                        }
504                    })
505                    .cloned()
506                    .collect();
507                TokenizedAsset::encode_batch(metadata, &tokenized_assets)
508            }
509            _ => Err(ArrowError::InvalidArgumentError(format!(
510                "Instrument type {type_name} serialization not yet implemented"
511            ))),
512        }
513    }
514
515    fn metadata(&self) -> HashMap<String, String> {
516        let mut metadata = HashMap::new();
517        metadata.insert(
518            KEY_INSTRUMENT_ID.to_string(),
519            Instrument::id(self).to_string(),
520        );
521
522        let type_name = match self {
523            Self::Cfd(_) => "Cfd",
524            Self::Commodity(_) => "Commodity",
525            Self::CurrencyPair(_) => "CurrencyPair",
526            Self::Equity(_) => "Equity",
527            Self::CryptoFuture(_) => "CryptoFuture",
528            Self::CryptoFuturesSpread(_) => "CryptoFuturesSpread",
529            Self::CryptoPerpetual(_) => "CryptoPerpetual",
530            Self::CryptoOption(_) => "CryptoOption",
531            Self::CryptoOptionSpread(_) => "CryptoOptionSpread",
532            Self::FuturesContract(_) => "FuturesContract",
533            Self::FuturesSpread(_) => "FuturesSpread",
534            Self::IndexInstrument(_) => "IndexInstrument",
535            Self::OptionContract(_) => "OptionContract",
536            Self::OptionSpread(_) => "OptionSpread",
537            Self::BinaryOption(_) => "BinaryOption",
538            Self::Betting(_) => "BettingInstrument",
539            Self::PerpetualContract(_) => "PerpetualContract",
540            Self::TokenizedAsset(_) => "TokenizedAsset",
541        };
542
543        metadata.insert(KEY_TYPE_NAME.to_string(), type_name.to_string());
544        metadata
545    }
546}
547
548/// Decode InstrumentAny from RecordBatch
549/// (Cannot implement DecodeFromRecordBatch trait due to `Into<Data>` bound)
550///
551/// # Errors
552///
553/// Returns an error if the RecordBatch cannot be decoded.
554pub fn decode_instrument_any_batch(
555    #[allow(unused)] metadata: &HashMap<String, String>,
556    record_batch: &RecordBatch,
557) -> Result<Vec<InstrumentAny>, EncodingError> {
558    let type_name = metadata
559        .get(KEY_TYPE_NAME)
560        .map(String::as_str)
561        .ok_or_else(|| EncodingError::MissingMetadata(KEY_TYPE_NAME))?;
562
563    match type_name {
564        "Cfd" => {
565            let cfds = cfd::decode_cfd_batch(metadata, record_batch)?;
566            Ok(cfds.into_iter().map(InstrumentAny::Cfd).collect())
567        }
568        "Commodity" => {
569            let commodities = commodity::decode_commodity_batch(metadata, record_batch)?;
570            Ok(commodities
571                .into_iter()
572                .map(InstrumentAny::Commodity)
573                .collect())
574        }
575        "BettingInstrument" => {
576            let betting = betting::decode_betting_instrument_batch(metadata, record_batch)?;
577            Ok(betting.into_iter().map(InstrumentAny::Betting).collect())
578        }
579        "BinaryOption" => {
580            let binary_options = binary_option::decode_binary_option_batch(metadata, record_batch)?;
581            Ok(binary_options
582                .into_iter()
583                .map(InstrumentAny::BinaryOption)
584                .collect())
585        }
586        "CryptoFuture" => {
587            let crypto_futures = crypto_future::decode_crypto_future_batch(metadata, record_batch)?;
588            Ok(crypto_futures
589                .into_iter()
590                .map(InstrumentAny::CryptoFuture)
591                .collect())
592        }
593        "CryptoFuturesSpread" => {
594            let spreads =
595                crypto_futures_spread::decode_crypto_futures_spread_batch(metadata, record_batch)?;
596            Ok(spreads
597                .into_iter()
598                .map(InstrumentAny::CryptoFuturesSpread)
599                .collect())
600        }
601        "CryptoOption" => {
602            let crypto_options = crypto_option::decode_crypto_option_batch(metadata, record_batch)?;
603            Ok(crypto_options
604                .into_iter()
605                .map(InstrumentAny::CryptoOption)
606                .collect())
607        }
608        "CryptoOptionSpread" => {
609            let spreads =
610                crypto_option_spread::decode_crypto_option_spread_batch(metadata, record_batch)?;
611            Ok(spreads
612                .into_iter()
613                .map(InstrumentAny::CryptoOptionSpread)
614                .collect())
615        }
616        "CryptoPerpetual" => {
617            let crypto_perps =
618                crypto_perpetual::decode_crypto_perpetual_batch(metadata, record_batch)?;
619            Ok(crypto_perps
620                .into_iter()
621                .map(InstrumentAny::CryptoPerpetual)
622                .collect())
623        }
624        "CurrencyPair" => {
625            let currency_pairs = currency_pair::decode_currency_pair_batch(metadata, record_batch)?;
626            Ok(currency_pairs
627                .into_iter()
628                .map(InstrumentAny::CurrencyPair)
629                .collect())
630        }
631        "Equity" => {
632            let equities = equity::decode_equity_batch(metadata, record_batch)?;
633            Ok(equities.into_iter().map(InstrumentAny::Equity).collect())
634        }
635        "FuturesContract" => {
636            let futures_contracts =
637                futures_contract::decode_futures_contract_batch(metadata, record_batch)?;
638            Ok(futures_contracts
639                .into_iter()
640                .map(InstrumentAny::FuturesContract)
641                .collect())
642        }
643        "FuturesSpread" => {
644            let futures_spreads =
645                futures_spread::decode_futures_spread_batch(metadata, record_batch)?;
646            Ok(futures_spreads
647                .into_iter()
648                .map(InstrumentAny::FuturesSpread)
649                .collect())
650        }
651        "IndexInstrument" => {
652            let index_instruments =
653                index_instrument::decode_index_instrument_batch(metadata, record_batch)?;
654            Ok(index_instruments
655                .into_iter()
656                .map(InstrumentAny::IndexInstrument)
657                .collect())
658        }
659        "OptionContract" => {
660            let option_contracts =
661                option_contract::decode_option_contract_batch(metadata, record_batch)?;
662            Ok(option_contracts
663                .into_iter()
664                .map(InstrumentAny::OptionContract)
665                .collect())
666        }
667        "OptionSpread" => {
668            let option_spreads = option_spread::decode_option_spread_batch(metadata, record_batch)?;
669            Ok(option_spreads
670                .into_iter()
671                .map(InstrumentAny::OptionSpread)
672                .collect())
673        }
674        "PerpetualContract" => {
675            let perpetual_contracts =
676                perpetual_contract::decode_perpetual_contract_batch(metadata, record_batch)?;
677            Ok(perpetual_contracts
678                .into_iter()
679                .map(InstrumentAny::PerpetualContract)
680                .collect())
681        }
682        "TokenizedAsset" => {
683            let tokenized_assets =
684                tokenized_asset::decode_tokenized_asset_batch(metadata, record_batch)?;
685            Ok(tokenized_assets
686                .into_iter()
687                .map(InstrumentAny::TokenizedAsset)
688                .collect())
689        }
690        _ => Err(EncodingError::ParseError(
691            KEY_TYPE_NAME,
692            format!("Unknown instrument type: {type_name}"),
693        )),
694    }
695}
696
697#[cfg(test)]
698mod tests {
699    use std::sync::Arc;
700
701    use arrow::{
702        array::{Array, ArrayRef, StringArray, UInt8Array},
703        datatypes::DataType,
704    };
705    use nautilus_core::{Params, UnixNanos};
706    use nautilus_model::{
707        enums::{AssetClass, CurrencyType, OptionKind},
708        identifiers::{InstrumentId, Symbol},
709        instruments::{
710            Instrument, InstrumentAny,
711            currency_pair::CurrencyPair,
712            stubs::{
713                betting, cfd_gold, commodity_gold, crypto_future_btcusdt,
714                crypto_futures_spread_btc_deribit, crypto_option_btc_deribit,
715                crypto_option_spread_btc_deribit, crypto_perpetual_ethusdt, currency_pair_ethusdt,
716                equity_aapl, perpetual_contract_eurusd, tokenized_asset_aaplx,
717            },
718        },
719        types::{Currency, Money, Price, Quantity},
720    };
721    use rstest::rstest;
722    use rust_decimal_macros::dec;
723    use ustr::Ustr;
724
725    use super::*;
726
727    #[rstest]
728    fn test_get_schema() {
729        let mut metadata = HashMap::new();
730        metadata.insert(KEY_TYPE_NAME.to_string(), "CurrencyPair".to_string());
731        let schema = InstrumentAny::get_schema(Some(metadata));
732        assert!(schema.fields().len() >= 20);
733        assert_eq!(schema.field(0).name(), "id");
734    }
735
736    #[rstest]
737    #[case("")]
738    #[case("   ")]
739    #[case("\t\n")]
740    fn test_decode_currency_empty_or_whitespace_errors(#[case] value: &str) {
741        let result = decode_currency(value, "currency", "test.currency", 7);
742        let err = result.expect_err("empty code must surface EncodingError");
743        match err {
744            EncodingError::ParseError(field, msg) => {
745                assert_eq!(field, "currency");
746                assert!(
747                    msg.contains("row 7"),
748                    "message should include row index, found: {msg}",
749                );
750                assert!(
751                    msg.contains("empty currency code"),
752                    "message should describe empty code, found: {msg}",
753                );
754            }
755            other => panic!("unexpected error variant: {other:?}"),
756        }
757        // Ensure the fallback did not register a phantom currency under the empty key.
758        assert!(Currency::try_from_str(value.trim()).is_none());
759    }
760
761    #[rstest]
762    #[case("USD", CurrencyType::Fiat, 2)]
763    #[case("BTC", CurrencyType::Crypto, 8)]
764    #[case("XAU", CurrencyType::CommodityBacked, 2)]
765    fn test_decode_currency_known_code_preserves_metadata(
766        #[case] code: &str,
767        #[case] expected_type: CurrencyType,
768        #[case] expected_precision: u8,
769    ) {
770        let currency = decode_currency(code, "currency", "test.currency", 0).unwrap();
771        assert_eq!(currency.code.as_str(), code);
772        assert_eq!(currency.currency_type, expected_type);
773        assert_eq!(currency.precision, expected_precision);
774    }
775
776    #[rstest]
777    fn test_decode_currency_unknown_code_registers_as_crypto() {
778        let code = "XDECTEST";
779        assert!(
780            Currency::try_from_str(code).is_none(),
781            "test precondition: '{code}' must not be pre-registered",
782        );
783
784        let currency = decode_currency(code, "base_currency", "test.base_currency", 0).unwrap();
785        assert_eq!(currency.code.as_str(), code);
786        assert_eq!(currency.currency_type, CurrencyType::Crypto);
787        assert_eq!(currency.precision, 8);
788        assert_eq!(currency.iso4217, 0);
789
790        let registered = Currency::try_from_str(code).expect("unknown code must be registered");
791        assert_eq!(registered, currency);
792    }
793
794    #[rstest]
795    fn test_encode_decode_round_trip() {
796        let instrument_id = InstrumentId::from("EUR/USD.SIM");
797        let currency_pair = CurrencyPair::builder()
798            .instrument_id(instrument_id)
799            .raw_symbol(Symbol::from("EUR/USD"))
800            .base_currency(Currency::from("EUR"))
801            .quote_currency(Currency::from("USD"))
802            .price_precision(5)
803            // size_precision must match size_increment precision (0)
804            .size_precision(0)
805            .price_increment(Price::new(0.00001, 5))
806            // precision 0
807            .size_increment(Quantity::new(1.0, 0))
808            .tick_scheme(Ustr::from("FOREX_5DECIMAL"))
809            .ts_event(UnixNanos::default())
810            .ts_init(UnixNanos::default())
811            .build()
812            .unwrap();
813        let instrument = InstrumentAny::CurrencyPair(currency_pair);
814
815        let metadata = instrument.metadata();
816        let record_batch =
817            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
818        let decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
819
820        assert_eq!(decoded.len(), 1);
821        assert_eq!(Instrument::id(&decoded[0]), Instrument::id(&instrument));
822        assert_eq!(
823            Instrument::raw_symbol(&decoded[0]),
824            Instrument::raw_symbol(&instrument)
825        );
826        assert_eq!(
827            Instrument::asset_class(&decoded[0]),
828            Instrument::asset_class(&instrument)
829        );
830
831        match (&decoded[0], &instrument) {
832            (InstrumentAny::CurrencyPair(decoded_cp), InstrumentAny::CurrencyPair(original_cp)) => {
833                assert_eq!(decoded_cp.id, original_cp.id);
834                assert_eq!(decoded_cp.base_currency, original_cp.base_currency);
835                assert_eq!(decoded_cp.quote_currency, original_cp.quote_currency);
836                assert_eq!(decoded_cp.price_precision, original_cp.price_precision);
837                assert_eq!(decoded_cp.size_precision, original_cp.size_precision);
838                assert_eq!(decoded_cp.tick_scheme, original_cp.tick_scheme);
839            }
840            _ => panic!("Decoded instrument type mismatch"),
841        }
842    }
843
844    #[rstest]
845    #[case::binary_option(nautilus_model::instruments::stubs::binary_option().into_any())]
846    #[case::betting(betting().into_any())]
847    #[case::cfd(cfd_gold().into_any())]
848    #[case::commodity(commodity_gold().into_any())]
849    #[case::crypto_future(crypto_future_btcusdt(2, 6, Price::from("0.01"), Quantity::from("0.000001")).into_any())]
850    #[case::crypto_futures_spread(crypto_futures_spread_btc_deribit().into_any())]
851    #[case::crypto_option(crypto_option_btc_deribit(3, 1, Price::from("0.001"), Quantity::from("0.1")).into_any())]
852    #[case::crypto_option_spread(crypto_option_spread_btc_deribit().into_any())]
853    #[case::crypto_perpetual(crypto_perpetual_ethusdt().into_any())]
854    #[case::currency_pair(currency_pair_ethusdt().into_any())]
855    #[case::perpetual_contract(perpetual_contract_eurusd().into_any())]
856    #[case::tokenized_asset(tokenized_asset_aaplx().into_any())]
857    fn test_instrument_exact_money_notional_round_trip(#[case] instrument: InstrumentAny) {
858        let currency = instrument.quote_currency();
859        let maximum = Money::from_decimal_exact(dec!(1.000000001), currency).unwrap();
860        let minimum = Money::from_decimal_exact(dec!(0.000000003), currency).unwrap();
861        let mut encoded = serde_json::to_value(instrument).unwrap();
862        let fields = encoded
863            .as_object_mut()
864            .unwrap()
865            .values_mut()
866            .next()
867            .unwrap();
868        fields["max_notional"] = serde_json::to_value(maximum).unwrap();
869        fields["min_notional"] = serde_json::to_value(minimum).unwrap();
870        let instrument: InstrumentAny = serde_json::from_value(encoded).unwrap();
871        let metadata = instrument.metadata();
872        let batch =
873            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
874        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
875
876        assert_eq!(decoded.len(), 1);
877        assert_eq!(decoded[0].id(), instrument.id());
878        assert_eq!(decoded[0].max_notional(), Some(maximum));
879        assert_eq!(decoded[0].min_notional(), Some(minimum));
880    }
881
882    #[rstest]
883    fn test_decode_currency_pair_without_tick_scheme_column_defaults_none() {
884        let instrument_id = InstrumentId::from("EUR/USD.SIM");
885        let currency_pair = CurrencyPair::builder()
886            .instrument_id(instrument_id)
887            .raw_symbol(Symbol::from("EUR/USD"))
888            .base_currency(Currency::from("EUR"))
889            .quote_currency(Currency::from("USD"))
890            .price_precision(5)
891            .size_precision(0)
892            .price_increment(Price::new(0.00001, 5))
893            .size_increment(Quantity::new(1.0, 0))
894            .tick_scheme(Ustr::from("FOREX_5DECIMAL"))
895            .ts_event(UnixNanos::default())
896            .ts_init(UnixNanos::default())
897            .build()
898            .unwrap();
899        let instrument = InstrumentAny::CurrencyPair(currency_pair);
900
901        let metadata = instrument.metadata();
902        let record_batch =
903            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
904        let record_batch = batch_without_column(&record_batch, "tick_scheme");
905        let decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
906
907        assert_eq!(decoded.len(), 1);
908        match &decoded[0] {
909            InstrumentAny::CurrencyPair(decoded_cp) => {
910                assert_eq!(decoded_cp.id, instrument.id());
911                assert_eq!(decoded_cp.tick_scheme, None);
912            }
913            _ => panic!("Decoded instrument type mismatch"),
914        }
915    }
916
917    #[rstest]
918    fn test_encode_decode_round_trip_equity() {
919        use nautilus_model::instruments::{Instrument, equity::Equity};
920
921        let instrument_id = InstrumentId::from("AAPL.NASDAQ");
922        let equity = Equity::builder()
923            .instrument_id(instrument_id)
924            .raw_symbol(Symbol::from("AAPL"))
925            .currency(Currency::from("USD"))
926            .price_precision(2)
927            .price_increment(Price::new(0.01, 2))
928            .ts_event(UnixNanos::default())
929            .ts_init(UnixNanos::default())
930            .build()
931            .unwrap();
932        let instrument = InstrumentAny::Equity(equity);
933
934        let metadata = instrument.metadata();
935        let record_batch =
936            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
937        let decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
938        assert_eq!(decoded.len(), 1);
939        assert_eq!(Instrument::id(&decoded[0]), Instrument::id(&instrument));
940        assert_eq!(
941            Instrument::raw_symbol(&decoded[0]),
942            Instrument::raw_symbol(&instrument)
943        );
944        assert_eq!(
945            Instrument::asset_class(&decoded[0]),
946            Instrument::asset_class(&instrument)
947        );
948
949        match (&decoded[0], &instrument) {
950            (InstrumentAny::Equity(decoded_eq), InstrumentAny::Equity(original_eq)) => {
951                assert_eq!(decoded_eq.id, original_eq.id);
952                assert_eq!(decoded_eq.currency, original_eq.currency);
953                assert_eq!(decoded_eq.price_precision, original_eq.price_precision);
954            }
955            _ => panic!("Decoded instrument type mismatch"),
956        }
957    }
958
959    #[rstest]
960    fn test_encode_decode_round_trip_equity_all_fields() {
961        use nautilus_core::Params;
962
963        let info = Params::new([("sector".to_string(), serde_json::json!("technology"))].into());
964
965        let equity = Equity::builder()
966            .instrument_id(InstrumentId::from("AAPL.NASDAQ"))
967            .raw_symbol(Symbol::from("AAPL"))
968            .isin(Ustr::from("US0378331005"))
969            .currency(Currency::from("USD"))
970            .price_precision(2)
971            .price_increment(Price::from("0.01"))
972            .lot_size(Quantity::from("100"))
973            .max_quantity(Quantity::from("10000"))
974            .min_quantity(Quantity::from("1"))
975            .max_price(Price::from("9999.99"))
976            .min_price(Price::from("0.01"))
977            .margin_init(dec!(0.01))
978            .margin_maint(dec!(0.02))
979            .tick_scheme(Ustr::from("TOPIX100"))
980            .info(info)
981            .ts_event(1.into())
982            .ts_init(2.into())
983            .build()
984            .unwrap();
985        let instrument = InstrumentAny::Equity(equity.clone());
986
987        let metadata = instrument.metadata();
988        let record_batch =
989            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
990        let decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
991
992        assert_eq!(decoded.len(), 1);
993        let InstrumentAny::Equity(decoded_equity) = &decoded[0] else {
994            panic!("Decoded instrument type mismatch");
995        };
996
997        // The v1 `from_dict` dropped these quantity constraints (#4461), so check them here
998        assert_eq!(decoded_equity.max_quantity, equity.max_quantity);
999        assert_eq!(decoded_equity.min_quantity, equity.min_quantity);
1000
1001        // `PartialEq` compares only `id`, so compare every field via its serialized form
1002        assert_eq!(
1003            serde_json::to_value(decoded_equity).unwrap(),
1004            serde_json::to_value(&equity).unwrap(),
1005        );
1006    }
1007
1008    #[rstest]
1009    fn test_instrument_info_uses_nullable_utf8_json() {
1010        let absent = equity_aapl();
1011        let mut populated = absent.clone();
1012
1013        let info = nautilus_core::Params::new(
1014            [
1015                ("enabled".to_string(), serde_json::json!(true)),
1016                ("source".to_string(), serde_json::json!("unit-test")),
1017            ]
1018            .into(),
1019        );
1020        populated.info = Some(info);
1021        let instruments = [
1022            InstrumentAny::Equity(populated.clone()),
1023            InstrumentAny::Equity(absent.clone()),
1024        ];
1025        let metadata = instruments[0].metadata();
1026
1027        let batch = InstrumentAny::encode_batch(&metadata, &instruments).unwrap();
1028        let schema = batch.schema();
1029        let field = schema.field_with_name("info").unwrap();
1030        let info = batch
1031            .column_by_name("info")
1032            .unwrap()
1033            .as_any()
1034            .downcast_ref::<StringArray>()
1035            .unwrap();
1036        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1037
1038        assert_eq!(field.data_type(), &DataType::Utf8);
1039        assert!(field.is_nullable());
1040        assert_eq!(field.extension_type_name(), Some("arrow.json"));
1041        assert_eq!(
1042            serde_json::from_str::<nautilus_core::Params>(info.value(0)).unwrap(),
1043            populated.info.clone().unwrap(),
1044        );
1045        assert!(info.is_null(1));
1046        let InstrumentAny::Equity(decoded_populated) = &decoded[0] else {
1047            panic!("Decoded instrument type mismatch");
1048        };
1049        let InstrumentAny::Equity(decoded_absent) = &decoded[1] else {
1050            panic!("Decoded instrument type mismatch");
1051        };
1052        assert_eq!(decoded_populated.info, populated.info);
1053        assert_eq!(decoded_absent.info, absent.info);
1054    }
1055
1056    #[rstest]
1057    fn test_encode_decode_round_trip_futures_contract_all_fields() {
1058        let contract = FuturesContract::builder()
1059            .instrument_id(InstrumentId::from("ESZ4.XCME"))
1060            .raw_symbol(Symbol::from("ESZ4"))
1061            .asset_class(AssetClass::Index)
1062            .exchange(Ustr::from("XCME"))
1063            .underlying(Ustr::from("ES"))
1064            .activation_ns(1.into())
1065            .expiration_ns(2.into())
1066            .currency(Currency::from("USD"))
1067            .price_precision(2)
1068            .price_increment(Price::from("0.01"))
1069            .multiplier(Quantity::from("1"))
1070            .lot_size(Quantity::from("1"))
1071            .max_quantity(Quantity::from("10000"))
1072            .min_quantity(Quantity::from("5"))
1073            .max_price(Price::from("9999.99"))
1074            .min_price(Price::from("0.01"))
1075            .margin_init(dec!(0.01))
1076            .margin_maint(dec!(0.02))
1077            .ts_event(1.into())
1078            .ts_init(2.into())
1079            .build()
1080            .unwrap();
1081
1082        let decoded = encode_decode_instrument(&InstrumentAny::FuturesContract(contract.clone()));
1083        let InstrumentAny::FuturesContract(decoded) = decoded else {
1084            panic!("Decoded instrument type mismatch");
1085        };
1086
1087        assert_eq!(
1088            serde_json::to_value(&decoded).unwrap(),
1089            serde_json::to_value(&contract).unwrap(),
1090        );
1091    }
1092
1093    #[rstest]
1094    fn test_encode_decode_round_trip_option_contract_all_fields() {
1095        let contract = OptionContract::builder()
1096            .instrument_id(InstrumentId::from("AAPL_C100.OPRA"))
1097            .raw_symbol(Symbol::from("AAPL_C100"))
1098            .asset_class(AssetClass::Equity)
1099            .exchange(Ustr::from("OPRA"))
1100            .underlying(Ustr::from("AAPL"))
1101            .option_kind(OptionKind::Call)
1102            .strike_price(Price::from("100.00"))
1103            .currency(Currency::from("USD"))
1104            .activation_ns(1.into())
1105            .expiration_ns(2.into())
1106            .price_precision(2)
1107            .price_increment(Price::from("0.01"))
1108            .multiplier(Quantity::from("100"))
1109            .lot_size(Quantity::from("1"))
1110            .max_quantity(Quantity::from("10000"))
1111            .min_quantity(Quantity::from("5"))
1112            .max_price(Price::from("9999.99"))
1113            .min_price(Price::from("0.01"))
1114            .margin_init(dec!(0.01))
1115            .margin_maint(dec!(0.02))
1116            .ts_event(1.into())
1117            .ts_init(2.into())
1118            .build()
1119            .unwrap();
1120
1121        let decoded = encode_decode_instrument(&InstrumentAny::OptionContract(contract.clone()));
1122        let InstrumentAny::OptionContract(decoded) = decoded else {
1123            panic!("Decoded instrument type mismatch");
1124        };
1125
1126        assert_eq!(
1127            serde_json::to_value(&decoded).unwrap(),
1128            serde_json::to_value(&contract).unwrap(),
1129        );
1130    }
1131
1132    #[rstest]
1133    fn test_encode_decode_round_trip_binary_option_all_fields() {
1134        let raw = "0.1234567890123456789012345678";
1135        let info = Params::new([("gamma_market".to_string(), serde_json::json!(raw))].into());
1136        let option = BinaryOption::builder()
1137            .instrument_id(InstrumentId::from("ELECTION.POLYMARKET"))
1138            .raw_symbol(Symbol::from("ELECTION"))
1139            .asset_class(AssetClass::Alternative)
1140            .currency(Currency::from("USDC"))
1141            .activation_ns(1.into())
1142            .expiration_ns(2.into())
1143            .price_precision(2)
1144            .size_precision(0)
1145            .price_increment(Price::from("0.01"))
1146            .size_increment(Quantity::from("1"))
1147            .event_id(Ustr::from("event-123"))
1148            .info(info)
1149            .outcome(Ustr::from("YES"))
1150            .description(Ustr::from("Election outcome"))
1151            .max_quantity(Quantity::from("10000"))
1152            .min_quantity(Quantity::from("5"))
1153            .max_notional(Money::from("50000 USDC"))
1154            .min_notional(Money::from("5 USDC"))
1155            .max_price(Price::from("0.99"))
1156            .min_price(Price::from("0.01"))
1157            .margin_init(dec!(0.01))
1158            .margin_maint(dec!(0.02))
1159            .ts_event(1.into())
1160            .ts_init(2.into())
1161            .build()
1162            .unwrap();
1163
1164        let decoded = encode_decode_instrument(&InstrumentAny::BinaryOption(option.clone()));
1165        let InstrumentAny::BinaryOption(decoded) = decoded else {
1166            panic!("Decoded instrument type mismatch");
1167        };
1168
1169        assert_eq!(
1170            serde_json::to_value(&decoded).unwrap(),
1171            serde_json::to_value(&option).unwrap(),
1172        );
1173    }
1174
1175    #[rstest]
1176    #[case::missing(false)]
1177    #[case::null(true)]
1178    fn test_binary_option_event_id_legacy_default(#[case] null: bool) {
1179        let mut option = nautilus_model::instruments::stubs::binary_option();
1180        option.event_id = Some(Ustr::from("event-123"));
1181        let metadata = option.metadata();
1182        let batch = BinaryOption::encode_batch(&metadata, &[option]).unwrap();
1183
1184        let batch = if null {
1185            batch_with_null_string_column(&batch, "event_id")
1186        } else {
1187            batch_without_column(&batch, "event_id")
1188        };
1189
1190        let decoded = binary_option::decode_binary_option_batch(&metadata, &batch).unwrap();
1191        assert_eq!(decoded.len(), 1);
1192        assert_eq!(decoded[0].event_id, None);
1193    }
1194
1195    // The `betting` stub populates every bound, margin, and fee, so this covers the whole struct
1196    #[rstest]
1197    fn test_encode_decode_round_trip_betting_all_fields() {
1198        let instrument = betting();
1199
1200        let decoded = encode_decode_instrument(&InstrumentAny::Betting(instrument.clone()));
1201        let InstrumentAny::Betting(decoded) = decoded else {
1202            panic!("Decoded instrument type mismatch");
1203        };
1204
1205        assert_eq!(
1206            serde_json::to_value(&decoded).unwrap(),
1207            serde_json::to_value(&instrument).unwrap(),
1208        );
1209    }
1210
1211    fn encode_decode_instrument(instrument: &InstrumentAny) -> InstrumentAny {
1212        let metadata = instrument.metadata();
1213        let record_batch =
1214            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(instrument)).unwrap();
1215        let mut decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
1216
1217        assert_eq!(decoded.len(), 1);
1218        decoded.remove(0)
1219    }
1220
1221    fn roundtrip_case(instrument: &InstrumentAny) {
1222        let metadata = instrument.metadata();
1223        let record_batch =
1224            InstrumentAny::encode_batch(&metadata, std::slice::from_ref(instrument)).unwrap();
1225        let decoded = decode_instrument_any_batch(&metadata, &record_batch).unwrap();
1226
1227        assert_eq!(decoded.len(), 1);
1228        assert_eq!(Instrument::id(&decoded[0]), Instrument::id(instrument));
1229        assert_eq!(
1230            Instrument::raw_symbol(&decoded[0]),
1231            Instrument::raw_symbol(instrument)
1232        );
1233        assert_eq!(
1234            Instrument::asset_class(&decoded[0]),
1235            Instrument::asset_class(instrument)
1236        );
1237        assert_eq!(
1238            Instrument::instrument_class(&decoded[0]),
1239            Instrument::instrument_class(instrument)
1240        );
1241        assert_eq!(
1242            Instrument::price_precision(&decoded[0]),
1243            Instrument::price_precision(instrument)
1244        );
1245        assert_eq!(
1246            Instrument::size_precision(&decoded[0]),
1247            Instrument::size_precision(instrument)
1248        );
1249        assert_eq!(
1250            Instrument::quote_currency(&decoded[0]),
1251            Instrument::quote_currency(instrument)
1252        );
1253        assert_eq!(
1254            std::mem::discriminant(&decoded[0]),
1255            std::mem::discriminant(instrument),
1256            "decoded variant must match encoded variant"
1257        );
1258    }
1259
1260    fn batch_without_column(record_batch: &RecordBatch, column_name: &str) -> RecordBatch {
1261        let schema = record_batch.schema();
1262        let column_index = schema.index_of(column_name).unwrap();
1263        let fields: Vec<_> = schema
1264            .fields()
1265            .iter()
1266            .enumerate()
1267            .filter(|(index, _)| *index != column_index)
1268            .map(|(_, field)| field.as_ref().clone())
1269            .collect();
1270        let columns = record_batch
1271            .columns()
1272            .iter()
1273            .enumerate()
1274            .filter(|(index, _)| *index != column_index)
1275            .map(|(_, column)| Arc::clone(column))
1276            .collect();
1277        let new_schema = Schema::new_with_metadata(fields, schema.metadata().clone());
1278
1279        RecordBatch::try_new(Arc::new(new_schema), columns).unwrap()
1280    }
1281
1282    fn batch_with_null_string_column(record_batch: &RecordBatch, column_name: &str) -> RecordBatch {
1283        let schema = record_batch.schema();
1284        let column_index = schema.index_of(column_name).unwrap();
1285        let mut columns = record_batch.columns().to_vec();
1286        let null_column: ArrayRef = Arc::new(StringArray::from(vec![None::<&str>]));
1287        columns[column_index] = null_column;
1288
1289        RecordBatch::try_new(schema, columns).unwrap()
1290    }
1291
1292    fn batch_with_uint8_column(
1293        record_batch: &RecordBatch,
1294        column_name: &str,
1295        values: Vec<u8>,
1296    ) -> RecordBatch {
1297        let schema = record_batch.schema();
1298        let column_index = schema.index_of(column_name).unwrap();
1299        let mut columns = record_batch.columns().to_vec();
1300        columns[column_index] = Arc::new(UInt8Array::from(values));
1301
1302        RecordBatch::try_new(schema, columns).unwrap()
1303    }
1304
1305    fn batch_with_null_uint8_column(record_batch: &RecordBatch, column_name: &str) -> RecordBatch {
1306        let schema = record_batch.schema();
1307        let column_index = schema.index_of(column_name).unwrap();
1308        let fields = schema
1309            .fields()
1310            .iter()
1311            .enumerate()
1312            .map(|(index, field)| {
1313                if index == column_index {
1314                    field.as_ref().clone().with_nullable(true)
1315                } else {
1316                    field.as_ref().clone()
1317                }
1318            })
1319            .collect::<Vec<_>>();
1320
1321        let mut columns = record_batch.columns().to_vec();
1322        columns[column_index] = Arc::new(UInt8Array::from(vec![None]));
1323        let schema = Schema::new_with_metadata(fields, schema.metadata().clone());
1324
1325        RecordBatch::try_new(Arc::new(schema), columns).unwrap()
1326    }
1327
1328    #[rstest]
1329    fn test_public_instrument_decoders_reject_null_required_columns() {
1330        let instruments = vec![
1331            InstrumentAny::Betting(nautilus_model::instruments::stubs::betting()),
1332            InstrumentAny::BinaryOption(nautilus_model::instruments::stubs::binary_option()),
1333            InstrumentAny::Cfd(nautilus_model::instruments::stubs::cfd_gold()),
1334            InstrumentAny::Commodity(nautilus_model::instruments::stubs::commodity_gold()),
1335            InstrumentAny::CryptoFuture(nautilus_model::instruments::stubs::crypto_future_btcusdt(
1336                2,
1337                6,
1338                Price::from("0.01"),
1339                Quantity::from("0.000001"),
1340            )),
1341            InstrumentAny::CryptoFuturesSpread(
1342                nautilus_model::instruments::stubs::crypto_futures_spread_btc_deribit(),
1343            ),
1344            InstrumentAny::CryptoOption(
1345                nautilus_model::instruments::stubs::crypto_option_btc_deribit(
1346                    3,
1347                    1,
1348                    Price::from("0.001"),
1349                    Quantity::from("0.1"),
1350                ),
1351            ),
1352            InstrumentAny::CryptoOptionSpread(
1353                nautilus_model::instruments::stubs::crypto_option_spread_btc_deribit(),
1354            ),
1355            InstrumentAny::CryptoPerpetual(
1356                nautilus_model::instruments::stubs::crypto_perpetual_ethusdt(),
1357            ),
1358            InstrumentAny::CurrencyPair(nautilus_model::instruments::stubs::currency_pair_btcusdt()),
1359            InstrumentAny::Equity(nautilus_model::instruments::stubs::equity_aapl()),
1360            InstrumentAny::FuturesContract(
1361                nautilus_model::instruments::stubs::futures_contract_es(None, None),
1362            ),
1363            InstrumentAny::FuturesSpread(nautilus_model::instruments::stubs::futures_spread_es()),
1364            InstrumentAny::IndexInstrument(
1365                nautilus_model::instruments::stubs::index_instrument_spx(),
1366            ),
1367            InstrumentAny::OptionContract(
1368                nautilus_model::instruments::stubs::option_contract_appl(),
1369            ),
1370            InstrumentAny::OptionSpread(nautilus_model::instruments::stubs::option_spread()),
1371            InstrumentAny::PerpetualContract(
1372                nautilus_model::instruments::stubs::perpetual_contract_eurusd(),
1373            ),
1374            InstrumentAny::TokenizedAsset(
1375                nautilus_model::instruments::stubs::tokenized_asset_aaplx(),
1376            ),
1377        ];
1378
1379        for instrument in instruments {
1380            let metadata = instrument.metadata();
1381            let record_batch =
1382                InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&instrument)).unwrap();
1383            let record_batch = batch_with_null_uint8_column(&record_batch, "price_precision");
1384
1385            let result = decode_instrument_any_batch(&metadata, &record_batch);
1386
1387            assert!(matches!(
1388                result,
1389                Err(EncodingError::ParseError(INSTRUMENT_VALIDATION_FIELD, ref message))
1390                    if message == "required field 'price_precision' is null at row 0"
1391            ));
1392        }
1393    }
1394
1395    #[rstest]
1396    #[case::binary_option(InstrumentAny::BinaryOption(
1397        nautilus_model::instruments::stubs::binary_option()
1398    ))]
1399    #[case::cfd(InstrumentAny::Cfd(nautilus_model::instruments::stubs::cfd_gold()))]
1400    #[case::commodity(InstrumentAny::Commodity(
1401        nautilus_model::instruments::stubs::commodity_gold()
1402    ))]
1403    #[case::crypto_future(InstrumentAny::CryptoFuture(
1404        nautilus_model::instruments::stubs::crypto_future_btcusdt(
1405            2,
1406            6,
1407            Price::from("0.01"),
1408            Quantity::from("0.000001"),
1409        )
1410    ))]
1411    #[case::crypto_futures_spread(InstrumentAny::CryptoFuturesSpread(
1412        nautilus_model::instruments::stubs::crypto_futures_spread_btc_deribit()
1413    ))]
1414    #[case::crypto_option(InstrumentAny::CryptoOption(
1415        nautilus_model::instruments::stubs::crypto_option_btc_deribit(
1416            3,
1417            1,
1418            Price::from("0.001"),
1419            Quantity::from("0.1"),
1420        )
1421    ))]
1422    #[case::crypto_option_spread(InstrumentAny::CryptoOptionSpread(
1423        nautilus_model::instruments::stubs::crypto_option_spread_btc_deribit()
1424    ))]
1425    #[case::crypto_perpetual(InstrumentAny::CryptoPerpetual(
1426        nautilus_model::instruments::stubs::crypto_perpetual_ethusdt()
1427    ))]
1428    #[case::currency_pair(InstrumentAny::CurrencyPair(
1429        nautilus_model::instruments::stubs::currency_pair_btcusdt()
1430    ))]
1431    #[case::equity(InstrumentAny::Equity(nautilus_model::instruments::stubs::equity_aapl()))]
1432    #[case::futures_contract(InstrumentAny::FuturesContract(
1433        nautilus_model::instruments::stubs::futures_contract_es(None, None,)
1434    ))]
1435    #[case::futures_spread(InstrumentAny::FuturesSpread(
1436        nautilus_model::instruments::stubs::futures_spread_es()
1437    ))]
1438    #[case::index_instrument(InstrumentAny::IndexInstrument(
1439        nautilus_model::instruments::stubs::index_instrument_spx()
1440    ))]
1441    #[case::option_contract(InstrumentAny::OptionContract(
1442        nautilus_model::instruments::stubs::option_contract_appl()
1443    ))]
1444    #[case::option_spread(InstrumentAny::OptionSpread(
1445        nautilus_model::instruments::stubs::option_spread()
1446    ))]
1447    #[case::perpetual_contract(InstrumentAny::PerpetualContract(
1448        nautilus_model::instruments::stubs::perpetual_contract_eurusd()
1449    ))]
1450    #[case::tokenized_asset(InstrumentAny::TokenizedAsset(
1451        nautilus_model::instruments::stubs::tokenized_asset_aaplx()
1452    ))]
1453    fn test_decode_instrument_checked_constructor_error(#[case] instrument: InstrumentAny) {
1454        let metadata = instrument.metadata();
1455        let class = metadata.get(KEY_TYPE_NAME).unwrap();
1456        let first_row_price_precision = Instrument::price_precision(&instrument);
1457        let instruments = vec![instrument.clone(), instrument];
1458        let record_batch = InstrumentAny::encode_batch(&metadata, &instruments).unwrap();
1459        let record_batch = batch_with_uint8_column(
1460            &record_batch,
1461            "price_precision",
1462            vec![first_row_price_precision, u8::MAX],
1463        );
1464
1465        let error = decode_instrument_any_batch(&metadata, &record_batch)
1466            .expect_err("invalid precision must return EncodingError");
1467
1468        match error {
1469            EncodingError::ParseError(field, message) => {
1470                assert_eq!(field, INSTRUMENT_VALIDATION_FIELD);
1471                assert!(
1472                    message.contains(class),
1473                    "message should include instrument class, found: {message}",
1474                );
1475                assert!(
1476                    message.starts_with("row 1:"),
1477                    "message should include row index, found: {message}",
1478                );
1479                assert!(
1480                    message.contains("price_precision"),
1481                    "message should include failed precision, found: {message}",
1482                );
1483            }
1484            other => panic!("unexpected error variant: {other:?}"),
1485        }
1486    }
1487
1488    #[rstest]
1489    fn test_roundtrip_betting() {
1490        use nautilus_model::instruments::stubs::betting;
1491        roundtrip_case(&InstrumentAny::Betting(betting()));
1492    }
1493
1494    #[rstest]
1495    fn test_roundtrip_binary_option() {
1496        use nautilus_model::instruments::stubs::binary_option;
1497        roundtrip_case(&InstrumentAny::BinaryOption(binary_option()));
1498    }
1499
1500    #[rstest]
1501    fn test_roundtrip_cfd() {
1502        use nautilus_model::instruments::stubs::cfd_gold;
1503        roundtrip_case(&InstrumentAny::Cfd(cfd_gold()));
1504    }
1505
1506    #[rstest]
1507    fn test_roundtrip_commodity() {
1508        use nautilus_model::instruments::stubs::commodity_gold;
1509        roundtrip_case(&InstrumentAny::Commodity(commodity_gold()));
1510    }
1511
1512    #[rstest]
1513    fn test_roundtrip_crypto_future() {
1514        use nautilus_model::instruments::stubs::crypto_future_btcusdt;
1515
1516        let mut inst = crypto_future_btcusdt(2, 6, Price::from("0.01"), Quantity::from("0.000001"));
1517        inst.lot_size = Quantity::from("0.25");
1518        let any = InstrumentAny::CryptoFuture(inst.clone());
1519        roundtrip_case(&any);
1520        let metadata = any.metadata();
1521        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1522        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1523        let InstrumentAny::CryptoFuture(decoded_inst) = &decoded[0] else {
1524            panic!("decoded variant is not CryptoFuture");
1525        };
1526        assert_eq!(decoded_inst.lot_size, inst.lot_size);
1527    }
1528
1529    #[rstest]
1530    fn test_decode_crypto_future_without_lot_size_column_defaults_to_one() {
1531        use nautilus_model::instruments::stubs::crypto_future_btcusdt;
1532
1533        let inst = crypto_future_btcusdt(2, 6, Price::from("0.01"), Quantity::from("0.000001"));
1534        let any = InstrumentAny::CryptoFuture(inst);
1535        let metadata = any.metadata();
1536        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1537        let batch = batch_without_column(&batch, "lot_size");
1538
1539        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1540
1541        let InstrumentAny::CryptoFuture(decoded_inst) = &decoded[0] else {
1542            panic!("decoded variant is not CryptoFuture");
1543        };
1544        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1545    }
1546
1547    #[rstest]
1548    fn test_decode_crypto_future_null_lot_size_defaults_to_one() {
1549        use nautilus_model::instruments::stubs::crypto_future_btcusdt;
1550
1551        let inst = crypto_future_btcusdt(2, 6, Price::from("0.01"), Quantity::from("0.000001"));
1552        let any = InstrumentAny::CryptoFuture(inst);
1553        let metadata = any.metadata();
1554        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1555        let batch = batch_with_null_string_column(&batch, "lot_size");
1556
1557        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1558
1559        let InstrumentAny::CryptoFuture(decoded_inst) = &decoded[0] else {
1560            panic!("decoded variant is not CryptoFuture");
1561        };
1562        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1563    }
1564
1565    #[rstest]
1566    fn test_roundtrip_crypto_option() {
1567        use nautilus_model::instruments::stubs::crypto_option_btc_deribit;
1568
1569        let mut inst = crypto_option_btc_deribit(3, 1, Price::from("0.001"), Quantity::from("0.1"));
1570        inst.lot_size = Quantity::from("0.5");
1571        let any = InstrumentAny::CryptoOption(inst.clone());
1572        roundtrip_case(&any);
1573        let metadata = any.metadata();
1574        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1575        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1576        let InstrumentAny::CryptoOption(decoded_inst) = &decoded[0] else {
1577            panic!("decoded variant is not CryptoOption");
1578        };
1579        assert_eq!(decoded_inst.lot_size, inst.lot_size);
1580    }
1581
1582    #[rstest]
1583    fn test_decode_crypto_option_without_lot_size_column_defaults_to_one() {
1584        use nautilus_model::instruments::stubs::crypto_option_btc_deribit;
1585
1586        let inst = crypto_option_btc_deribit(3, 1, Price::from("0.001"), Quantity::from("0.1"));
1587        let any = InstrumentAny::CryptoOption(inst);
1588        let metadata = any.metadata();
1589        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1590        let batch = batch_without_column(&batch, "lot_size");
1591
1592        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1593
1594        let InstrumentAny::CryptoOption(decoded_inst) = &decoded[0] else {
1595            panic!("decoded variant is not CryptoOption");
1596        };
1597        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1598    }
1599
1600    #[rstest]
1601    fn test_decode_crypto_option_null_lot_size_defaults_to_one() {
1602        use nautilus_model::instruments::stubs::crypto_option_btc_deribit;
1603
1604        let inst = crypto_option_btc_deribit(3, 1, Price::from("0.001"), Quantity::from("0.1"));
1605        let any = InstrumentAny::CryptoOption(inst);
1606        let metadata = any.metadata();
1607        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1608        let batch = batch_with_null_string_column(&batch, "lot_size");
1609
1610        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1611
1612        let InstrumentAny::CryptoOption(decoded_inst) = &decoded[0] else {
1613            panic!("decoded variant is not CryptoOption");
1614        };
1615        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1616    }
1617
1618    #[rstest]
1619    fn test_roundtrip_crypto_futures_spread() {
1620        use nautilus_model::instruments::{Instrument, stubs::crypto_futures_spread_btc_deribit};
1621        let inst = crypto_futures_spread_btc_deribit();
1622        let any = InstrumentAny::CryptoFuturesSpread(inst.clone());
1623        roundtrip_case(&any);
1624        let metadata = any.metadata();
1625        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1626        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1627        let InstrumentAny::CryptoFuturesSpread(decoded_inst) = &decoded[0] else {
1628            panic!("decoded variant is not CryptoFuturesSpread");
1629        };
1630        assert_eq!(decoded_inst.lot_size, inst.lot_size);
1631        assert_eq!(decoded_inst.is_inverse, inst.is_inverse);
1632        assert_eq!(decoded_inst.strategy_type, inst.strategy_type);
1633        assert_eq!(decoded_inst.settlement_currency, inst.settlement_currency);
1634        assert_eq!(Instrument::id(decoded_inst), Instrument::id(&inst));
1635    }
1636
1637    #[rstest]
1638    fn test_roundtrip_crypto_option_spread() {
1639        use nautilus_model::instruments::{Instrument, stubs::crypto_option_spread_btc_deribit};
1640        let inst = crypto_option_spread_btc_deribit();
1641        let any = InstrumentAny::CryptoOptionSpread(inst.clone());
1642        roundtrip_case(&any);
1643        let metadata = any.metadata();
1644        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1645        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1646        let InstrumentAny::CryptoOptionSpread(decoded_inst) = &decoded[0] else {
1647            panic!("decoded variant is not CryptoOptionSpread");
1648        };
1649        // Deribit BTC option combos carry min_trade_amount=0.1, which sets
1650        // lot_size=0.1; dropping the lot_size Arrow column would silently
1651        // default it back to 1
1652        assert_eq!(decoded_inst.lot_size, inst.lot_size);
1653        assert_eq!(decoded_inst.size_precision, inst.size_precision);
1654        assert_eq!(decoded_inst.size_increment, inst.size_increment);
1655        assert_eq!(decoded_inst.is_inverse, inst.is_inverse);
1656        assert_eq!(decoded_inst.strategy_type, inst.strategy_type);
1657        assert_eq!(decoded_inst.settlement_currency, inst.settlement_currency);
1658        assert_eq!(Instrument::id(decoded_inst), Instrument::id(&inst));
1659    }
1660
1661    #[rstest]
1662    fn test_roundtrip_crypto_perpetual_inverse() {
1663        use nautilus_model::instruments::stubs::btcusd_bybit;
1664        roundtrip_case(&InstrumentAny::CryptoPerpetual(btcusd_bybit()));
1665    }
1666
1667    #[rstest]
1668    fn test_roundtrip_crypto_perpetual_linear() {
1669        use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
1670
1671        let mut inst = crypto_perpetual_ethusdt();
1672        inst.lot_size = Quantity::from("0.005");
1673        let any = InstrumentAny::CryptoPerpetual(inst.clone());
1674        roundtrip_case(&any);
1675        let metadata = any.metadata();
1676        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1677        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1678        let InstrumentAny::CryptoPerpetual(decoded_inst) = &decoded[0] else {
1679            panic!("decoded variant is not CryptoPerpetual");
1680        };
1681        assert_eq!(decoded_inst.lot_size, inst.lot_size);
1682    }
1683
1684    #[rstest]
1685    fn test_decode_crypto_perpetual_without_lot_size_column_defaults_to_one() {
1686        use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
1687
1688        let inst = crypto_perpetual_ethusdt();
1689        let any = InstrumentAny::CryptoPerpetual(inst);
1690        let metadata = any.metadata();
1691        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1692        let batch = batch_without_column(&batch, "lot_size");
1693
1694        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1695
1696        let InstrumentAny::CryptoPerpetual(decoded_inst) = &decoded[0] else {
1697            panic!("decoded variant is not CryptoPerpetual");
1698        };
1699        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1700    }
1701
1702    #[rstest]
1703    fn test_decode_crypto_perpetual_null_lot_size_defaults_to_one() {
1704        use nautilus_model::instruments::stubs::crypto_perpetual_ethusdt;
1705
1706        let inst = crypto_perpetual_ethusdt();
1707        let any = InstrumentAny::CryptoPerpetual(inst);
1708        let metadata = any.metadata();
1709        let batch = InstrumentAny::encode_batch(&metadata, std::slice::from_ref(&any)).unwrap();
1710        let batch = batch_with_null_string_column(&batch, "lot_size");
1711
1712        let decoded = decode_instrument_any_batch(&metadata, &batch).unwrap();
1713
1714        let InstrumentAny::CryptoPerpetual(decoded_inst) = &decoded[0] else {
1715            panic!("decoded variant is not CryptoPerpetual");
1716        };
1717        assert_eq!(decoded_inst.lot_size, Quantity::from(1));
1718    }
1719
1720    #[rstest]
1721    fn test_roundtrip_futures_contract() {
1722        use nautilus_model::instruments::stubs::futures_contract_es;
1723        roundtrip_case(&InstrumentAny::FuturesContract(futures_contract_es(
1724            None, None,
1725        )));
1726    }
1727
1728    #[rstest]
1729    fn test_roundtrip_futures_spread() {
1730        use nautilus_model::instruments::stubs::futures_spread_es;
1731        roundtrip_case(&InstrumentAny::FuturesSpread(futures_spread_es()));
1732    }
1733
1734    #[rstest]
1735    fn test_roundtrip_index_instrument() {
1736        use nautilus_model::instruments::stubs::index_instrument_spx;
1737        roundtrip_case(&InstrumentAny::IndexInstrument(index_instrument_spx()));
1738    }
1739
1740    #[rstest]
1741    fn test_roundtrip_option_contract() {
1742        use nautilus_model::instruments::stubs::option_contract_appl;
1743        roundtrip_case(&InstrumentAny::OptionContract(option_contract_appl()));
1744    }
1745
1746    #[rstest]
1747    fn test_roundtrip_option_spread() {
1748        use nautilus_model::instruments::stubs::option_spread;
1749        roundtrip_case(&InstrumentAny::OptionSpread(option_spread()));
1750    }
1751
1752    #[rstest]
1753    fn test_roundtrip_perpetual_contract() {
1754        use nautilus_model::instruments::stubs::perpetual_contract_eurusd;
1755        roundtrip_case(&InstrumentAny::PerpetualContract(
1756            perpetual_contract_eurusd(),
1757        ));
1758    }
1759
1760    #[rstest]
1761    fn test_roundtrip_tokenized_asset() {
1762        use nautilus_model::instruments::stubs::tokenized_asset_aaplx;
1763        roundtrip_case(&InstrumentAny::TokenizedAsset(tokenized_asset_aaplx()));
1764    }
1765}
nautilus_serialization/arrow/instrument/mod.rs

nautilus_serialization/arrow/instrument/
mod.rs
