nautilus_tardis/csv/
convert.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
16use std::{path::PathBuf, time::Duration};
17
18use ahash::AHashMap;
19use anyhow::Context;
20use csv::StringRecord;
21use nautilus_core::UnixNanos;
22use nautilus_model::{
23    data::{Data, HasTsInit},
24    enums::OptionKind,
25    identifiers::{InstrumentId, Symbol},
26    instruments::{CryptoOption, InstrumentAny},
27    types::{Currency, Price, Quantity, fixed::FIXED_PRECISION},
28};
29use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
30use rust_decimal::Decimal;
31
32use crate::{
33    common::{
34        enums::{TardisExchange, TardisOptionType},
35        parse::{parse_instrument_id, parse_option_kind, parse_timestamp},
36    },
37    csv::{
38        create_csv_reader, infer_precision, load::OptionsChainPrecision, matches_underlying_filter,
39        normalize_underlying_filters, parse_options_chain_record,
40        parse_options_chain_record_as_quote, record::TardisOptionsChainRecord,
41    },
42};
43
44const DATA_FLUSH_ROWS: usize = 100_000;
45
46/// Configuration for converting Tardis `options_chain` CSV files into a Nautilus catalog.
47#[derive(Debug, Clone, bon::Builder)]
48pub struct TardisOptionsChainCSVConverterConfig {
49    /// Tardis daily `options_chain` CSV file paths, processed in order.
50    pub filepaths: Vec<PathBuf>,
51    /// Nautilus catalog path.
52    pub catalog_path: PathBuf,
53    /// Optional underlying prefixes, such as `BTC-`.
54    pub underlyings: Option<Vec<String>>,
55    /// Optional thinning interval. Keeps the last row per instrument per bucket.
56    pub snapshot_interval: Option<Duration>,
57    /// Whether to emit quotes from best bid/offer fields.
58    #[builder(default = true)]
59    pub extract_bbo_as_quotes: bool,
60    /// Whether to derive and write instrument definitions from the CSV rows.
61    #[builder(default = true)]
62    pub write_instruments: bool,
63    /// Optional explicit price precision.
64    pub price_precision: Option<u8>,
65    /// Optional explicit size precision.
66    pub size_precision: Option<u8>,
67}
68
69/// Converts Tardis `options_chain` CSV files into `QuoteTick`, `OptionGreeks`, and instruments.
70///
71/// # Errors
72///
73/// Returns an error if the catalog cannot be opened, a CSV file cannot be read, a row cannot be
74/// parsed, a complete best bid/offer row contains invalid values, instrument derivation fails, or
75/// catalog writes fail.
76pub fn convert_options_chain_csv(
77    config: &TardisOptionsChainCSVConverterConfig,
78) -> anyhow::Result<()> {
79    let underlyings = normalize_underlying_filters(config.underlyings.clone());
80    let catalog = ParquetDataCatalog::from_uri(
81        &config.catalog_path.to_string_lossy(),
82        None,
83        None,
84        None,
85        None,
86    )?;
87    let mut precision_by_instrument: AHashMap<InstrumentId, OptionsChainPrecision> =
88        AHashMap::new();
89    let mut instrument_states: AHashMap<InstrumentId, InstrumentBuildState> = AHashMap::new();
90    let mut data_buffers: AHashMap<InstrumentId, DataBuffer> = AHashMap::new();
91    let mut pending_records: AHashMap<(InstrumentId, u64), TardisOptionsChainRecord> =
92        AHashMap::new();
93    let mut current_bucket = None;
94
95    for filepath in &config.filepaths {
96        let mut reader = create_csv_reader(filepath)
97            .with_context(|| format!("failed to open CSV file {}", filepath.display()))?;
98        let mut csv_record = StringRecord::new();
99
100        while reader
101            .read_record(&mut csv_record)
102            .with_context(|| format!("failed to read CSV file {}", filepath.display()))?
103        {
104            if let Some(underlyings) = underlyings.as_deref() {
105                let Some(symbol) = csv_record.get(1) else {
106                    continue;
107                };
108                let symbol = symbol.to_uppercase();
109                if !matches_underlying_filter(&symbol, Some(underlyings)) {
110                    continue;
111                }
112            }
113
114            let record: TardisOptionsChainRecord = csv_record
115                .deserialize(None)
116                .with_context(|| format!("failed to parse CSV file {}", filepath.display()))?;
117            let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
118            precision_by_instrument
119                .entry(instrument_id)
120                .or_insert_with(|| {
121                    OptionsChainPrecision::new(config.price_precision, config.size_precision)
122                })
123                .update(&record, config.price_precision, config.size_precision);
124            instrument_states
125                .entry(instrument_id)
126                .and_modify(|state| state.update_activation(record.local_timestamp))
127                .or_insert_with(|| InstrumentBuildState::new(record.clone()));
128
129            if let Some(interval) = config.snapshot_interval {
130                let interval_us = u64::try_from(interval.as_micros())
131                    .context("snapshot interval exceeds u64 microseconds")?;
132                anyhow::ensure!(interval_us > 0, "snapshot interval must be positive");
133                let bucket = record.local_timestamp / interval_us;
134
135                if let Some(current_bucket) = current_bucket {
136                    anyhow::ensure!(
137                        bucket >= current_bucket,
138                        "options_chain CSV rows must be ordered by local_timestamp when thinning"
139                    );
140                }
141
142                if current_bucket.is_none_or(|current| bucket > current) {
143                    flush_pending_records_before(
144                        &catalog,
145                        &mut pending_records,
146                        &mut data_buffers,
147                        &precision_by_instrument,
148                        bucket,
149                        config.extract_bbo_as_quotes,
150                    )?;
151                    current_bucket = Some(bucket);
152                }
153                pending_records
154                    .entry((instrument_id, bucket))
155                    .and_modify(|pending| {
156                        if record.local_timestamp >= pending.local_timestamp {
157                            *pending = record.clone();
158                        }
159                    })
160                    .or_insert(record);
161            } else {
162                flush_data_buffer_if_ready(
163                    &catalog,
164                    &mut data_buffers,
165                    instrument_id,
166                    parse_timestamp(record.local_timestamp),
167                )?;
168                let data = options_chain_record_to_data(
169                    &record,
170                    &precision_by_instrument,
171                    config.extract_bbo_as_quotes,
172                )?;
173                data_buffers
174                    .entry(instrument_id)
175                    .or_default()
176                    .extend(data, parse_timestamp(record.local_timestamp));
177            }
178        }
179
180        if config.snapshot_interval.is_some() {
181            flush_pending_records_before(
182                &catalog,
183                &mut pending_records,
184                &mut data_buffers,
185                &precision_by_instrument,
186                u64::MAX,
187                config.extract_bbo_as_quotes,
188            )?;
189            flush_data_buffers(&catalog, &mut data_buffers)?;
190            current_bucket = None;
191        }
192    }
193
194    flush_data_buffers(&catalog, &mut data_buffers)?;
195
196    if config.write_instruments {
197        let instruments = build_instruments(instrument_states, &precision_by_instrument)?;
198        catalog.write_instruments(instruments)?;
199    }
200
201    Ok(())
202}
203
204#[derive(Debug, Clone)]
205struct InstrumentBuildState {
206    record: TardisOptionsChainRecord,
207    activation: UnixNanos,
208}
209
210impl InstrumentBuildState {
211    fn new(record: TardisOptionsChainRecord) -> Self {
212        Self {
213            activation: parse_timestamp(record.local_timestamp),
214            record,
215        }
216    }
217
218    fn update_activation(&mut self, local_timestamp: u64) {
219        self.activation = self.activation.min(parse_timestamp(local_timestamp));
220    }
221}
222
223#[derive(Debug, Default)]
224struct DataBuffer {
225    data: Vec<Data>,
226    last_ts_init: Option<UnixNanos>,
227}
228
229impl DataBuffer {
230    fn extend(&mut self, mut data: Vec<Data>, ts_init: UnixNanos) {
231        self.data.append(&mut data);
232        self.last_ts_init = Some(ts_init);
233    }
234}
235
236fn flush_data_buffer_if_ready(
237    catalog: &ParquetDataCatalog,
238    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
239    instrument_id: InstrumentId,
240    ts_init: UnixNanos,
241) -> anyhow::Result<()> {
242    let Some(buffer) = data_buffers.get_mut(&instrument_id) else {
243        return Ok(());
244    };
245
246    if buffer.data.len() >= DATA_FLUSH_ROWS
247        && buffer
248            .last_ts_init
249            .is_some_and(|last_ts_init| last_ts_init < ts_init)
250    {
251        write_data_buffer(catalog, &mut buffer.data)?;
252    }
253
254    Ok(())
255}
256
257fn flush_data_buffers(
258    catalog: &ParquetDataCatalog,
259    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
260) -> anyhow::Result<()> {
261    for buffer in data_buffers.values_mut() {
262        write_data_buffer(catalog, &mut buffer.data)?;
263    }
264    data_buffers.clear();
265    Ok(())
266}
267
268fn flush_pending_records_before(
269    catalog: &ParquetDataCatalog,
270    pending_records: &mut AHashMap<(InstrumentId, u64), TardisOptionsChainRecord>,
271    data_buffers: &mut AHashMap<InstrumentId, DataBuffer>,
272    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
273    next_bucket: u64,
274    extract_bbo_as_quotes: bool,
275) -> anyhow::Result<()> {
276    let mut ready = Vec::new();
277    pending_records.retain(|(_, bucket), record| {
278        if *bucket < next_bucket {
279            ready.push(record.clone());
280            false
281        } else {
282            true
283        }
284    });
285    ready.sort_by_key(|record| (record.local_timestamp, record.symbol));
286
287    for record in ready {
288        let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
289        let ts_init = parse_timestamp(record.local_timestamp);
290        flush_data_buffer_if_ready(catalog, data_buffers, instrument_id, ts_init)?;
291        let data =
292            options_chain_record_to_data(&record, precision_by_instrument, extract_bbo_as_quotes)?;
293        data_buffers
294            .entry(instrument_id)
295            .or_default()
296            .extend(data, ts_init);
297    }
298    Ok(())
299}
300
301fn options_chain_record_to_data(
302    record: &TardisOptionsChainRecord,
303    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
304    extract_bbo_as_quotes: bool,
305) -> anyhow::Result<Vec<Data>> {
306    let instrument_id = parse_instrument_id(&record.exchange, record.symbol);
307    let precision = precision_by_instrument
308        .get(&instrument_id)
309        .copied()
310        .unwrap_or_else(|| OptionsChainPrecision::new(None, None));
311    let mut data = Vec::with_capacity(2);
312
313    if extract_bbo_as_quotes
314        && let Some(quote) = parse_options_chain_record_as_quote(
315            record,
316            precision.price,
317            precision.size,
318            instrument_id,
319        )?
320    {
321        data.push(Data::Quote(quote));
322    }
323
324    data.push(Data::OptionGreeks(parse_options_chain_record(
325        record,
326        instrument_id,
327    )));
328    Ok(data)
329}
330
331fn write_data_buffer(catalog: &ParquetDataCatalog, data: &mut Vec<Data>) -> anyhow::Result<()> {
332    if data.is_empty() {
333        return Ok(());
334    }
335
336    let mut data_by_instrument: AHashMap<InstrumentId, Vec<Data>> = AHashMap::new();
337    for item in data.drain(..) {
338        data_by_instrument
339            .entry(item.instrument_id())
340            .or_default()
341            .push(item);
342    }
343
344    for instrument_data in data_by_instrument.values_mut() {
345        instrument_data.sort_by_key(HasTsInit::ts_init);
346        catalog.write_data_enum(instrument_data, None, None, None)?;
347    }
348
349    data.clear();
350    Ok(())
351}
352
353fn build_instruments(
354    instrument_states: AHashMap<InstrumentId, InstrumentBuildState>,
355    precision_by_instrument: &AHashMap<InstrumentId, OptionsChainPrecision>,
356) -> anyhow::Result<Vec<InstrumentAny>> {
357    let mut states = instrument_states.into_iter().collect::<Vec<_>>();
358    states.sort_by_key(|(instrument_id, _)| instrument_id.to_string());
359
360    states
361        .into_iter()
362        .map(|(instrument_id, state)| {
363            let precision = precision_by_instrument
364                .get(&instrument_id)
365                .copied()
366                .unwrap_or_else(|| OptionsChainPrecision::new(None, None));
367            create_crypto_option_from_options_chain_record(
368                &state.record,
369                instrument_id,
370                state.activation,
371                precision,
372            )
373        })
374        .collect()
375}
376
377fn create_crypto_option_from_options_chain_record(
378    record: &TardisOptionsChainRecord,
379    instrument_id: InstrumentId,
380    activation: UnixNanos,
381    precision: OptionsChainPrecision,
382) -> anyhow::Result<InstrumentAny> {
383    let underlying = record
384        .symbol
385        .as_str()
386        .split_once('-')
387        .map(|(underlying, _)| underlying)
388        .context("options_chain symbol missing underlying prefix")?;
389    let underlying_currency = Currency::get_or_create_crypto(underlying);
390    let (quote_currency, settlement_currency, is_inverse) =
391        option_currency_mapping(record.exchange, underlying_currency)?;
392    let instrument_price_precision = precision
393        .price
394        .max(infer_precision(record.strike_price).min(FIXED_PRECISION));
395    let price_increment = decimal_increment_price(instrument_price_precision)?;
396    let size_increment = decimal_increment_quantity(precision.size)?;
397    let strike_price = Price::from_decimal_dp(
398        Decimal::try_from(record.strike_price)?,
399        instrument_price_precision,
400    )?;
401    let expiration = parse_timestamp(record.expiration);
402    let option_kind = option_kind(record.option_type);
403
404    Ok(InstrumentAny::CryptoOption(
405        CryptoOption::builder()
406            .instrument_id(instrument_id)
407            .raw_symbol(Symbol::from_ustr_unchecked(record.symbol))
408            .underlying(underlying_currency)
409            .quote_currency(quote_currency)
410            .settlement_currency(settlement_currency)
411            .is_inverse(is_inverse)
412            .option_kind(option_kind)
413            .strike_price(strike_price)
414            .activation_ns(activation)
415            .expiration_ns(expiration)
416            .price_precision(instrument_price_precision)
417            .size_precision(precision.size)
418            .price_increment(price_increment)
419            .size_increment(size_increment)
420            .lot_size(size_increment)
421            .min_quantity(size_increment)
422            .ts_event(parse_timestamp(record.timestamp))
423            .ts_init(parse_timestamp(record.local_timestamp))
424            .build()?,
425    ))
426}
427
428fn option_currency_mapping(
429    exchange: TardisExchange,
430    underlying_currency: Currency,
431) -> anyhow::Result<(Currency, Currency, bool)> {
432    match exchange {
433        TardisExchange::Deribit => Ok((underlying_currency, underlying_currency, true)),
434        exchange => anyhow::bail!(
435            "options_chain instrument derivation supports Deribit only, received {exchange}"
436        ),
437    }
438}
439
440fn decimal_increment_price(precision: u8) -> anyhow::Result<Price> {
441    Ok(Price::from_decimal_dp(
442        Decimal::new(1, u32::from(precision)),
443        precision,
444    )?)
445}
446
447fn decimal_increment_quantity(precision: u8) -> anyhow::Result<Quantity> {
448    Ok(Quantity::from_decimal_dp(
449        Decimal::new(1, u32::from(precision)),
450        precision,
451    )?)
452}
453
454const fn option_kind(value: TardisOptionType) -> OptionKind {
455    parse_option_kind(value)
456}
457
458#[cfg(test)]
459mod tests {
460    use std::{
461        fs,
462        path::{Path, PathBuf},
463        time::Duration,
464    };
465
466    use nautilus_model::{
467        data::{OptionGreeks, QuoteTick},
468        enums::OptionKind,
469        instruments::Instrument,
470    };
471    use nautilus_persistence::backend::parquet::catalog::ParquetDataCatalog;
472    use rstest::rstest;
473    use tempfile::TempDir;
474    use ustr::Ustr;
475
476    use super::*;
477    use crate::common::testing::get_test_data_path;
478
479    #[rstest]
480    fn test_options_chain_converter_config_defaults_extract_bbo_as_quotes() {
481        let config = TardisOptionsChainCSVConverterConfig::builder()
482            .filepaths(Vec::<PathBuf>::new())
483            .catalog_path(PathBuf::from("/tmp/options-chain-catalog"))
484            .build();
485
486        assert!(config.extract_bbo_as_quotes);
487        assert!(config.write_instruments);
488    }
489
490    #[rstest]
491    fn test_convert_options_chain_csv_returns_error_for_missing_catalog_directory() {
492        let temp_dir = TempDir::new().unwrap();
493        let catalog_path = temp_dir.path().join("missing");
494        let config = TardisOptionsChainCSVConverterConfig::builder()
495            .filepaths(vec![get_test_data_path("options_chain.csv")])
496            .catalog_path(catalog_path.clone())
497            .build();
498
499        let error = convert_options_chain_csv(&config).unwrap_err();
500
501        assert_eq!(
502            error.to_string(),
503            format!(
504                "failed to open local storage directory '{}'; create it if it does not exist \
505                 and check access permissions",
506                catalog_path.display()
507            )
508        );
509        assert!(!catalog_path.exists());
510    }
511
512    #[rstest]
513    fn test_convert_options_chain_csv_thins_and_round_trips_catalog() {
514        let temp_dir = TempDir::new().unwrap();
515        let filepath = get_test_data_path("options_chain.csv");
516        let config = TardisOptionsChainCSVConverterConfig {
517            filepaths: vec![filepath],
518            catalog_path: temp_dir.path().to_path_buf(),
519            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
520            snapshot_interval: Some(Duration::from_secs(60)),
521            extract_bbo_as_quotes: true,
522            write_instruments: true,
523            price_precision: None,
524            size_precision: None,
525        };
526
527        convert_options_chain_csv(&config).unwrap();
528
529        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
530        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
531        let quotes = catalog
532            .query::<QuoteTick>(
533                Some(vec![instrument_id.clone()]),
534                None,
535                None,
536                None,
537                None,
538                true,
539            )
540            .unwrap();
541        let greeks = catalog
542            .query::<OptionGreeks>(
543                Some(vec![instrument_id.clone()]),
544                None,
545                None,
546                None,
547                None,
548                true,
549            )
550            .unwrap();
551        let instruments = catalog
552            .query_instruments(Some(std::slice::from_ref(&instrument_id)))
553            .unwrap();
554
555        assert_eq!(quotes.len(), 2);
556        assert_eq!(greeks.len(), 2);
557        assert_eq!(instruments.len(), 1);
558
559        assert_eq!(quotes[0].bid_price, Price::from("0.0206"));
560        assert_eq!(quotes[0].ask_price, Price::from("0.0236"));
561        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_473_112_000);
562        assert_eq!(quotes[1].bid_price, Price::from("0.0207"));
563        assert_eq!(quotes[1].ts_init.as_u64(), 1_591_574_460_473_112_000);
564
565        let InstrumentAny::CryptoOption(option) = &instruments[0] else {
566            panic!("Expected CryptoOption");
567        };
568        assert_eq!(option.id().to_string(), instrument_id);
569        assert_eq!(option.raw_symbol(), Symbol::from("BTC-9JUN20-9875-P"));
570        assert_eq!(option.option_kind, OptionKind::Put);
571        assert_eq!(option.strike_price, Price::from("9875.0000"));
572        assert_eq!(
573            option.expiration_ns,
574            UnixNanos::from(1_591_689_600_000_000_000)
575        );
576        assert_eq!(
577            option.activation_ns,
578            UnixNanos::from(1_591_574_400_196_008_000)
579        );
580        assert_eq!(option.price_increment, Price::from("0.0001"));
581        assert_eq!(option.size_increment, Quantity::from("0.1"));
582        assert!(option.is_inverse);
583        assert_eq!(option.quote_currency, Currency::from("BTC"));
584        assert_eq!(option.settlement_currency, Currency::from("BTC"));
585    }
586
587    #[rstest]
588    fn test_convert_options_chain_csv_thinned_multi_instrument_writes_separate_catalogs() {
589        let temp_dir = TempDir::new().unwrap();
590        let filepath = get_test_data_path("options_chain.csv");
591        let config = TardisOptionsChainCSVConverterConfig {
592            filepaths: vec![filepath],
593            catalog_path: temp_dir.path().to_path_buf(),
594            underlyings: Some(vec!["BTC-".to_string()]),
595            snapshot_interval: Some(Duration::from_secs(60)),
596            extract_bbo_as_quotes: true,
597            write_instruments: false,
598            price_precision: None,
599            size_precision: None,
600        };
601
602        convert_options_chain_csv(&config).unwrap();
603
604        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
605        let call_id = "BTC-9JUN20-10000-C.DERIBIT".to_string();
606        let next_expiry_id = "BTC-10JUN20-10000-C.DERIBIT".to_string();
607        let call_quotes = catalog
608            .query::<QuoteTick>(Some(vec![call_id.clone()]), None, None, None, None, true)
609            .unwrap();
610        let next_expiry_greeks = catalog
611            .query::<OptionGreeks>(
612                Some(vec![next_expiry_id.clone()]),
613                None,
614                None,
615                None,
616                None,
617                true,
618            )
619            .unwrap();
620
621        assert_eq!(call_quotes.len(), 1);
622        assert_eq!(call_quotes[0].instrument_id.to_string(), call_id);
623        assert_eq!(call_quotes[0].bid_price, Price::from("0.0305"));
624        assert_eq!(next_expiry_greeks.len(), 1);
625        assert_eq!(
626            next_expiry_greeks[0].instrument_id.to_string(),
627            next_expiry_id
628        );
629        assert_eq!(next_expiry_greeks[0].greeks.delta, 0.0);
630    }
631
632    #[rstest]
633    fn test_convert_options_chain_csv_unthinned_round_trips_catalog() {
634        let temp_dir = TempDir::new().unwrap();
635        let filepath = get_test_data_path("options_chain.csv");
636        let config = TardisOptionsChainCSVConverterConfig {
637            filepaths: vec![filepath],
638            catalog_path: temp_dir.path().to_path_buf(),
639            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
640            snapshot_interval: None,
641            extract_bbo_as_quotes: true,
642            write_instruments: false,
643            price_precision: None,
644            size_precision: None,
645        };
646
647        convert_options_chain_csv(&config).unwrap();
648
649        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
650        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
651        let quotes = catalog
652            .query::<QuoteTick>(
653                Some(vec![instrument_id.clone()]),
654                None,
655                None,
656                None,
657                None,
658                true,
659            )
660            .unwrap();
661        let greeks = catalog
662            .query::<OptionGreeks>(
663                Some(vec![instrument_id.clone()]),
664                None,
665                None,
666                None,
667                None,
668                true,
669            )
670            .unwrap();
671
672        assert_eq!(quotes.len(), 3);
673        assert_eq!(greeks.len(), 3);
674        assert_eq!(quotes[0].bid_price, Price::from("0.0205"));
675        assert_eq!(quotes[1].bid_price, Price::from("0.0206"));
676        assert_eq!(quotes[2].bid_price, Price::from("0.0207"));
677        assert_eq!(quotes[0].ts_init.as_u64(), 1_591_574_400_196_008_000);
678        assert_eq!(quotes[2].ts_init.as_u64(), 1_591_574_460_473_112_000);
679        assert!(
680            greeks
681                .iter()
682                .all(|greek| greek.instrument_id.to_string() == instrument_id)
683        );
684    }
685
686    #[rstest]
687    fn test_convert_options_chain_csv_can_suppress_bbo_quotes() {
688        let temp_dir = TempDir::new().unwrap();
689        let filepath = get_test_data_path("options_chain.csv");
690        let config = TardisOptionsChainCSVConverterConfig {
691            filepaths: vec![filepath],
692            catalog_path: temp_dir.path().to_path_buf(),
693            underlyings: Some(vec!["BTC-9JUN20-9875".to_string()]),
694            snapshot_interval: None,
695            extract_bbo_as_quotes: false,
696            write_instruments: false,
697            price_precision: None,
698            size_precision: None,
699        };
700
701        convert_options_chain_csv(&config).unwrap();
702
703        let mut catalog = ParquetDataCatalog::new(temp_dir.path(), None, None, None, None);
704        let instrument_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
705        let quotes = catalog
706            .query::<QuoteTick>(
707                Some(vec![instrument_id.clone()]),
708                None,
709                None,
710                None,
711                None,
712                true,
713            )
714            .unwrap();
715        let greeks = catalog
716            .query::<OptionGreeks>(
717                Some(vec![instrument_id.clone()]),
718                None,
719                None,
720                None,
721                None,
722                true,
723            )
724            .unwrap();
725
726        assert!(quotes.is_empty());
727        assert_eq!(greeks.len(), 3);
728        assert!(
729            greeks
730                .iter()
731                .all(|greek| greek.instrument_id.to_string() == instrument_id)
732        );
733    }
734
735    #[rstest]
736    fn test_convert_options_chain_csv_thinned_multi_file_resets_bucket_state() {
737        let temp_dir = TempDir::new().unwrap();
738        let fixture = fs::read_to_string(get_test_data_path("options_chain.csv")).unwrap();
739        let lines = fixture.lines().collect::<Vec<_>>();
740        let first_filepath = temp_dir.path().join("late_btc_options_chain.csv");
741        let second_filepath = temp_dir.path().join("early_eth_options_chain.csv");
742        let catalog_path = temp_dir.path().join("catalog");
743        fs::create_dir(&catalog_path).unwrap();
744        write_options_chain_rows(&first_filepath, lines[0], &[lines[7]]);
745        write_options_chain_rows(&second_filepath, lines[0], &[lines[5]]);
746
747        let config = TardisOptionsChainCSVConverterConfig {
748            filepaths: vec![first_filepath, second_filepath],
749            catalog_path: catalog_path.clone(),
750            underlyings: None,
751            snapshot_interval: Some(Duration::from_secs(60)),
752            extract_bbo_as_quotes: true,
753            write_instruments: false,
754            price_precision: None,
755            size_precision: None,
756        };
757
758        convert_options_chain_csv(&config).unwrap();
759
760        let mut catalog = ParquetDataCatalog::new(&catalog_path, None, None, None, None);
761        let btc_id = "BTC-9JUN20-9875-P.DERIBIT".to_string();
762        let eth_id = "ETH-9JUN20-250-P.DERIBIT".to_string();
763        let btc_quotes = catalog
764            .query::<QuoteTick>(Some(vec![btc_id.clone()]), None, None, None, None, true)
765            .unwrap();
766        let eth_quotes = catalog
767            .query::<QuoteTick>(Some(vec![eth_id.clone()]), None, None, None, None, true)
768            .unwrap();
769        let eth_greeks = catalog
770            .query::<OptionGreeks>(Some(vec![eth_id.clone()]), None, None, None, None, true)
771            .unwrap();
772
773        assert_eq!(btc_quotes.len(), 1);
774        assert_eq!(btc_quotes[0].instrument_id.to_string(), btc_id);
775        assert_eq!(btc_quotes[0].bid_price, Price::from("0.0207"));
776        assert_eq!(eth_quotes.len(), 1);
777        assert_eq!(eth_quotes[0].instrument_id.to_string(), eth_id);
778        assert_eq!(eth_quotes[0].bid_price, Price::from("0.12345"));
779        assert_eq!(eth_greeks.len(), 1);
780        assert_eq!(eth_greeks[0].instrument_id.to_string(), eth_id);
781    }
782
783    #[rstest]
784    fn test_create_crypto_option_preserves_fractional_strike_and_new_underlying() {
785        let record = TardisOptionsChainRecord {
786            exchange: TardisExchange::Deribit,
787            symbol: Ustr::from("NEW-9JUN20-2.05-C"),
788            timestamp: 1,
789            local_timestamp: 2,
790            option_type: TardisOptionType::Call,
791            strike_price: 2.05,
792            expiration: 1_591_689_600_000_000,
793            open_interest: None,
794            last_price: Some(0.1),
795            bid_price: Some(0.1),
796            bid_amount: Some(1.0),
797            bid_iv: None,
798            ask_price: Some(0.2),
799            ask_amount: Some(1.0),
800            ask_iv: None,
801            mark_price: None,
802            mark_iv: None,
803            underlying_index: "SYN.NEW-9JUN20".to_string(),
804            underlying_price: Some(2.0),
805            delta: None,
806            gamma: None,
807            vega: None,
808            theta: None,
809            rho: None,
810        };
811        let instrument_id = InstrumentId::from("NEW-9JUN20-2.05-C.DERIBIT");
812        let instrument = create_crypto_option_from_options_chain_record(
813            &record,
814            instrument_id,
815            UnixNanos::from(2_000),
816            OptionsChainPrecision { price: 1, size: 0 },
817        )
818        .unwrap();
819
820        let InstrumentAny::CryptoOption(option) = instrument else {
821            panic!("Expected CryptoOption");
822        };
823
824        assert_eq!(option.underlying, Currency::get_or_create_crypto("NEW"));
825        assert_eq!(option.strike_price, Price::from("2.05"));
826        assert_eq!(option.price_precision, 2);
827        assert_eq!(option.price_increment, Price::from("0.01"));
828    }
829
830    fn write_options_chain_rows(path: &Path, header: &str, rows: &[&str]) {
831        let mut contents = String::from(header);
832        contents.push('\n');
833        for row in rows {
834            contents.push_str(row);
835            contents.push('\n');
836        }
837        fs::write(path, contents).unwrap();
838    }
839}
nautilus_tardis/csv/convert.rs

nautilus_tardis/csv/
convert.rs
