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

nautilus_serialization/arrow/instrument/
mod.rs
