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

nautilus_serialization/arrow/instrument/
mod.rs
