Skip to main content

nautilus_databento/decode/
instruments.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 databento::dbn;
17use nautilus_core::UnixNanos;
18use nautilus_model::{
19    enums::AssetClass,
20    identifiers::{InstrumentId, Symbol},
21    instruments::{
22        CurrencyPair, Equity, FuturesContract, FuturesSpread, InstrumentAny, OptionContract,
23        OptionSpread,
24    },
25    types::{Currency, Quantity},
26};
27use ustr::Ustr;
28
29use super::{
30    expiration::{DatabentoDecodeConfig, corrected_option_expiration},
31    primitives::{
32        decode_lot_size, decode_multiplier, decode_optional_timestamp, decode_price,
33        decode_price_increment, decode_timestamp, decode_underlying, parse_cfi_iso10926,
34        parse_currency_or_usd_default, parse_option_kind,
35    },
36};
37
38/// # Errors
39///
40/// Returns an error if decoding the `InstrumentDefMsg` fails.
41///
42/// Returns `Ok(None)` for instrument classes with no Nautilus equivalent (`'I'` Index,
43/// `'B'` Bond, or any future class) and FX spots that cannot be mapped to known currencies.
44pub fn decode_instrument_def_msg(
45    msg: &dbn::InstrumentDefMsg,
46    instrument_id: InstrumentId,
47    ts_init: Option<UnixNanos>,
48    decode_config: Option<&DatabentoDecodeConfig>,
49) -> anyhow::Result<Option<InstrumentAny>> {
50    match msg.instrument_class as u8 as char {
51        'K' => Ok(Some(InstrumentAny::Equity(decode_equity(
52            msg,
53            instrument_id,
54            ts_init,
55        )?))),
56        'X' => {
57            Ok(decode_currency_pair(msg, instrument_id, ts_init)?.map(InstrumentAny::CurrencyPair))
58        }
59        'F' => Ok(Some(InstrumentAny::FuturesContract(
60            decode_futures_contract(msg, instrument_id, ts_init)?,
61        ))),
62        'S' => Ok(Some(InstrumentAny::FuturesSpread(decode_futures_spread(
63            msg,
64            instrument_id,
65            ts_init,
66        )?))),
67        'C' | 'P' => Ok(Some(InstrumentAny::OptionContract(decode_option_contract(
68            msg,
69            instrument_id,
70            ts_init,
71            decode_config,
72        )?))),
73        'T' | 'M' => Ok(Some(InstrumentAny::OptionSpread(decode_option_spread(
74            msg,
75            instrument_id,
76            ts_init,
77            decode_config,
78        )?))),
79        other => {
80            let label = match other {
81                'I' => "'I' (Index)".to_string(),
82                'B' => "'B' (Bond)".to_string(),
83                _ => format!("'{other}'"),
84            };
85            log::warn!("Skipping unsupported `instrument_class` {label} for {instrument_id}",);
86            Ok(None)
87        }
88    }
89}
90
91fn decode_currency_pair(
92    msg: &dbn::InstrumentDefMsg,
93    instrument_id: InstrumentId,
94    ts_init: Option<UnixNanos>,
95) -> anyhow::Result<Option<CurrencyPair>> {
96    let raw_symbol_str = msg.raw_symbol()?;
97    let raw_symbol = Symbol::from(raw_symbol_str);
98    let Some((base_currency, quote_currency)) = parse_fx_pair(
99        raw_symbol_str,
100        msg.asset().unwrap_or_default(),
101        msg.currency().unwrap_or_default(),
102    ) else {
103        log::warn!(
104            "Skipping FX spot {instrument_id}: could not parse currencies from raw_symbol='{raw_symbol_str}'"
105        );
106        return Ok(None);
107    };
108    let price_increment = decode_price_increment(msg.min_price_increment, quote_currency.precision);
109    let size_increment = decode_lot_size(msg.min_lot_size_round_lot);
110    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
111    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
112    let ts_event = UnixNanos::from(msg.ts_recv);
113    let ts_init = ts_init.unwrap_or(ts_event);
114
115    Ok(Some(
116        CurrencyPair::builder()
117            .instrument_id(instrument_id)
118            .raw_symbol(raw_symbol)
119            .base_currency(base_currency)
120            .quote_currency(quote_currency)
121            .price_precision(price_increment.precision)
122            .size_precision(size_increment.precision)
123            .price_increment(price_increment)
124            .size_increment(size_increment)
125            .multiplier(multiplier)
126            .lot_size(lot_size)
127            .ts_event(ts_event)
128            .ts_init(ts_init)
129            .build()?,
130    ))
131}
132
133fn parse_fx_pair(raw_symbol: &str, asset: &str, currency: &str) -> Option<(Currency, Currency)> {
134    parse_fx_pair_from_symbol(raw_symbol)
135        .or_else(|| parse_fx_pair_from_symbol(asset))
136        .or_else(|| parse_fx_pair_from_asset_currency(asset, currency))
137}
138
139fn parse_fx_pair_from_symbol(value: &str) -> Option<(Currency, Currency)> {
140    let normalized = value
141        .chars()
142        .filter(|ch| ch.is_ascii_alphabetic())
143        .collect::<String>()
144        .to_ascii_uppercase();
145
146    if normalized.len() != 6 {
147        return None;
148    }
149
150    let base = Currency::try_from_str(&normalized[..3])?;
151    let quote = Currency::try_from_str(&normalized[3..])?;
152    Some((base, quote))
153}
154
155fn parse_fx_pair_from_asset_currency(asset: &str, currency: &str) -> Option<(Currency, Currency)> {
156    let base = Currency::try_from_str(asset.trim().to_ascii_uppercase().as_str())?;
157    let quote = Currency::try_from_str(currency.trim().to_ascii_uppercase().as_str())?;
158    Some((base, quote))
159}
160
161/// Decodes a Databento instrument definition message into an `Equity` instrument.
162///
163/// # Errors
164///
165/// Returns an error if parsing or constructing `Equity` fails.
166///
167/// # Panics
168///
169/// Panics if the constructed instrument fails validation.
170pub fn decode_equity(
171    msg: &dbn::InstrumentDefMsg,
172    instrument_id: InstrumentId,
173    ts_init: Option<UnixNanos>,
174) -> anyhow::Result<Equity> {
175    let currency = parse_currency_or_usd_default(msg.currency());
176    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
177    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
178    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
179    let ts_init = ts_init.unwrap_or(ts_event);
180
181    Ok(Equity::builder()
182        .instrument_id(instrument_id)
183        .raw_symbol(instrument_id.symbol)
184        // No ISIN available yet
185        .currency(currency)
186        .price_precision(price_increment.precision)
187        .price_increment(price_increment)
188        .lot_size(lot_size)
189        .ts_event(ts_event)
190        .ts_init(ts_init)
191        .build()
192        .unwrap())
193}
194
195/// Decodes a Databento instrument definition message into a `FuturesContract` instrument.
196///
197/// # Errors
198///
199/// Returns an error if parsing or constructing `FuturesContract` fails.
200pub fn decode_futures_contract(
201    msg: &dbn::InstrumentDefMsg,
202    instrument_id: InstrumentId,
203    ts_init: Option<UnixNanos>,
204) -> anyhow::Result<FuturesContract> {
205    let currency = parse_currency_or_usd_default(msg.currency());
206    let exchange = Ustr::from(msg.exchange()?);
207    let underlying = decode_underlying(msg.asset()?, &instrument_id.symbol);
208    let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
209    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
210    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
211    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
212    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
213    let ts_init = ts_init.unwrap_or(ts_event);
214
215    Ok(FuturesContract::builder()
216        .instrument_id(instrument_id)
217        .raw_symbol(instrument_id.symbol)
218        .asset_class(asset_class.unwrap_or(AssetClass::Commodity))
219        .exchange(exchange)
220        .underlying(underlying)
221        .activation_ns(decode_optional_timestamp(msg.activation).unwrap_or_default())
222        .expiration_ns(decode_timestamp(msg.expiration, "expiration")?)
223        .currency(currency)
224        .price_precision(price_increment.precision)
225        .price_increment(price_increment)
226        .multiplier(multiplier)
227        .lot_size(lot_size)
228        .ts_event(ts_event)
229        .ts_init(ts_init)
230        .build()?)
231}
232
233/// Decodes a Databento instrument definition message into a `FuturesSpread` instrument.
234///
235/// # Errors
236///
237/// Returns an error if parsing or constructing `FuturesSpread` fails.
238pub fn decode_futures_spread(
239    msg: &dbn::InstrumentDefMsg,
240    instrument_id: InstrumentId,
241    ts_init: Option<UnixNanos>,
242) -> anyhow::Result<FuturesSpread> {
243    let exchange = Ustr::from(msg.exchange()?);
244    let underlying = decode_underlying(msg.asset()?, &instrument_id.symbol);
245    let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
246    let strategy_type = Ustr::from(msg.secsubtype()?);
247    let currency = parse_currency_or_usd_default(msg.currency());
248    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
249    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
250    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
251    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
252    let ts_init = ts_init.unwrap_or(ts_event);
253
254    Ok(FuturesSpread::builder()
255        .instrument_id(instrument_id)
256        .raw_symbol(instrument_id.symbol)
257        .asset_class(asset_class.unwrap_or(AssetClass::Commodity))
258        .exchange(exchange)
259        .underlying(underlying)
260        .strategy_type(strategy_type)
261        .activation_ns(decode_optional_timestamp(msg.activation).unwrap_or_default())
262        .expiration_ns(decode_timestamp(msg.expiration, "expiration")?)
263        .currency(currency)
264        .price_precision(price_increment.precision)
265        .price_increment(price_increment)
266        .multiplier(multiplier)
267        .lot_size(lot_size)
268        .ts_event(ts_event)
269        .ts_init(ts_init)
270        .build()?)
271}
272
273/// Decodes a Databento instrument definition message into an `OptionContract` instrument.
274///
275/// # Errors
276///
277/// Returns an error if parsing or constructing `OptionContract` fails.
278pub fn decode_option_contract(
279    msg: &dbn::InstrumentDefMsg,
280    instrument_id: InstrumentId,
281    ts_init: Option<UnixNanos>,
282    decode_config: Option<&DatabentoDecodeConfig>,
283) -> anyhow::Result<OptionContract> {
284    let currency = parse_currency_or_usd_default(msg.currency());
285    let strike_price_currency = parse_currency_or_usd_default(msg.strike_price_currency());
286    let exchange = Ustr::from(msg.exchange()?);
287    let underlying = decode_underlying(msg.underlying()?, &instrument_id.symbol);
288    let asset_class_opt = if instrument_id.venue.as_str() == "OPRA" {
289        Some(AssetClass::Equity)
290    } else {
291        let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
292        asset_class
293    };
294    let option_kind = parse_option_kind(msg.instrument_class)?;
295    let strike_price = decode_price(
296        msg.strike_price,
297        strike_price_currency.precision,
298        "strike_price",
299    )?;
300    let dataset = msg.hd.publisher().ok().map(|p| p.dataset());
301    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
302    let multiplier =
303        decode_option_multiplier(dataset, msg.contract_multiplier, msg.unit_of_measure_qty)?;
304    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
305    let expiration = corrected_option_expiration(
306        decode_timestamp(msg.expiration, "expiration")?,
307        underlying,
308        dataset,
309        decode_config,
310    );
311    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
312    let ts_init = ts_init.unwrap_or(ts_event);
313
314    Ok(OptionContract::builder()
315        .instrument_id(instrument_id)
316        .raw_symbol(instrument_id.symbol)
317        .asset_class(asset_class_opt.unwrap_or(AssetClass::Commodity))
318        .exchange(exchange)
319        .underlying(underlying)
320        .option_kind(option_kind)
321        .strike_price(strike_price)
322        .currency(currency)
323        .activation_ns(decode_optional_timestamp(msg.activation).unwrap_or_default())
324        .expiration_ns(expiration)
325        .price_precision(price_increment.precision)
326        .price_increment(price_increment)
327        .multiplier(multiplier)
328        .lot_size(lot_size)
329        .ts_event(ts_event)
330        .ts_init(ts_init)
331        .build()?)
332}
333
334fn decode_option_multiplier(
335    dataset: Option<dbn::Dataset>,
336    contract_multiplier: i32,
337    unit_of_measure_qty: i64,
338) -> anyhow::Result<Quantity> {
339    match (dataset, contract_multiplier) {
340        (Some(dbn::Dataset::OpraPillar), 0 | i32::MAX) => decode_multiplier(unit_of_measure_qty),
341        (Some(dbn::Dataset::OpraPillar), value) if value < 0 => {
342            anyhow::bail!("Invalid negative `contract_multiplier`: {value}")
343        }
344        (Some(dbn::Dataset::OpraPillar), value) => {
345            Ok(Quantity::from_mantissa_exponent(value as u64, 0, 0))
346        }
347        _ => decode_multiplier(unit_of_measure_qty),
348    }
349}
350
351/// Decodes a Databento instrument definition message into an `OptionSpread` instrument.
352///
353/// # Errors
354///
355/// Returns an error if parsing or constructing `OptionSpread` fails.
356pub fn decode_option_spread(
357    msg: &dbn::InstrumentDefMsg,
358    instrument_id: InstrumentId,
359    ts_init: Option<UnixNanos>,
360    decode_config: Option<&DatabentoDecodeConfig>,
361) -> anyhow::Result<OptionSpread> {
362    let exchange = Ustr::from(msg.exchange()?);
363    let underlying = decode_underlying(msg.underlying()?, &instrument_id.symbol);
364    let asset_class_opt = if instrument_id.venue.as_str() == "OPRA" {
365        Some(AssetClass::Equity)
366    } else {
367        let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
368        asset_class
369    };
370    let strategy_type = Ustr::from(msg.secsubtype()?);
371    let currency = parse_currency_or_usd_default(msg.currency());
372    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
373    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
374    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
375    let expiration = corrected_option_expiration(
376        decode_timestamp(msg.expiration, "expiration")?,
377        underlying,
378        msg.hd.publisher().ok().map(|p| p.dataset()),
379        decode_config,
380    );
381    let ts_event = msg.ts_recv.into(); // More accurate and reliable timestamp
382    let ts_init = ts_init.unwrap_or(ts_event);
383
384    Ok(OptionSpread::builder()
385        .instrument_id(instrument_id)
386        .raw_symbol(instrument_id.symbol)
387        .asset_class(asset_class_opt.unwrap_or(AssetClass::Commodity))
388        .exchange(exchange)
389        .underlying(underlying)
390        .strategy_type(strategy_type)
391        .activation_ns(decode_optional_timestamp(msg.activation).unwrap_or_default())
392        .expiration_ns(expiration)
393        .currency(currency)
394        .price_precision(price_increment.precision)
395        .price_increment(price_increment)
396        .multiplier(multiplier)
397        .lot_size(lot_size)
398        .ts_event(ts_event)
399        .ts_init(ts_init)
400        .build()?)
401}