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

nautilus_serialization/arrow/instrument/
mod.rs
