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,
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(CurrencyPair::new_checked(
116        instrument_id,
117        raw_symbol,
118        base_currency,
119        quote_currency,
120        price_increment.precision,
121        size_increment.precision,
122        price_increment,
123        size_increment,
124        Some(multiplier),
125        Some(lot_size),
126        None,
127        None,
128        None,
129        None,
130        None,
131        None,
132        None,
133        None,
134        None,
135        None,
136        None,
137        None,
138        ts_event,
139        ts_init,
140    )?))
141}
142
143fn parse_fx_pair(raw_symbol: &str, asset: &str, currency: &str) -> Option<(Currency, Currency)> {
144    parse_fx_pair_from_symbol(raw_symbol)
145        .or_else(|| parse_fx_pair_from_symbol(asset))
146        .or_else(|| parse_fx_pair_from_asset_currency(asset, currency))
147}
148
149fn parse_fx_pair_from_symbol(value: &str) -> Option<(Currency, Currency)> {
150    let normalized = value
151        .chars()
152        .filter(|ch| ch.is_ascii_alphabetic())
153        .collect::<String>()
154        .to_ascii_uppercase();
155
156    if normalized.len() != 6 {
157        return None;
158    }
159
160    let base = Currency::try_from_str(&normalized[..3])?;
161    let quote = Currency::try_from_str(&normalized[3..])?;
162    Some((base, quote))
163}
164
165fn parse_fx_pair_from_asset_currency(asset: &str, currency: &str) -> Option<(Currency, Currency)> {
166    let base = Currency::try_from_str(asset.trim().to_ascii_uppercase().as_str())?;
167    let quote = Currency::try_from_str(currency.trim().to_ascii_uppercase().as_str())?;
168    Some((base, quote))
169}
170
171/// Decodes a Databento instrument definition message into an `Equity` instrument.
172///
173/// # Errors
174///
175/// Returns an error if parsing or constructing `Equity` fails.
176pub fn decode_equity(
177    msg: &dbn::InstrumentDefMsg,
178    instrument_id: InstrumentId,
179    ts_init: Option<UnixNanos>,
180) -> anyhow::Result<Equity> {
181    let currency = parse_currency_or_usd_default(msg.currency());
182    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
183    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
184    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
185    let ts_init = ts_init.unwrap_or(ts_event);
186
187    Ok(Equity::new(
188        instrument_id,
189        instrument_id.symbol,
190        None, // No ISIN available yet
191        currency,
192        price_increment.precision,
193        price_increment,
194        Some(lot_size),
195        None, // max_quantity
196        None, // min_quantity
197        None, // max_price
198        None, // min_price
199        None, // margin_init
200        None, // margin_maint
201        None, // maker_fee
202        None, // taker_fee
203        None, // tick_scheme
204        None, // info
205        ts_event,
206        ts_init,
207    ))
208}
209
210/// Decodes a Databento instrument definition message into a `FuturesContract` instrument.
211///
212/// # Errors
213///
214/// Returns an error if parsing or constructing `FuturesContract` fails.
215pub fn decode_futures_contract(
216    msg: &dbn::InstrumentDefMsg,
217    instrument_id: InstrumentId,
218    ts_init: Option<UnixNanos>,
219) -> anyhow::Result<FuturesContract> {
220    let currency = parse_currency_or_usd_default(msg.currency());
221    let exchange = Ustr::from(msg.exchange()?);
222    let underlying = decode_underlying(msg.asset()?, &instrument_id.symbol);
223    let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
224    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
225    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
226    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
227    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
228    let ts_init = ts_init.unwrap_or(ts_event);
229
230    Ok(FuturesContract::new_checked(
231        instrument_id,
232        instrument_id.symbol,
233        asset_class.unwrap_or(AssetClass::Commodity),
234        Some(exchange),
235        underlying,
236        decode_optional_timestamp(msg.activation).unwrap_or_default(),
237        decode_timestamp(msg.expiration, "expiration")?,
238        currency,
239        price_increment.precision,
240        price_increment,
241        multiplier,
242        lot_size,
243        None, // max_quantity
244        None, // min_quantity
245        None, // max_price
246        None, // min_price
247        None, // margin_init
248        None, // margin_maint
249        None, // maker_fee
250        None, // taker_fee
251        None, // tick_scheme
252        None, // info
253        ts_event,
254        ts_init,
255    )?)
256}
257
258/// Decodes a Databento instrument definition message into a `FuturesSpread` instrument.
259///
260/// # Errors
261///
262/// Returns an error if parsing or constructing `FuturesSpread` fails.
263pub fn decode_futures_spread(
264    msg: &dbn::InstrumentDefMsg,
265    instrument_id: InstrumentId,
266    ts_init: Option<UnixNanos>,
267) -> anyhow::Result<FuturesSpread> {
268    let exchange = Ustr::from(msg.exchange()?);
269    let underlying = decode_underlying(msg.asset()?, &instrument_id.symbol);
270    let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
271    let strategy_type = Ustr::from(msg.secsubtype()?);
272    let currency = parse_currency_or_usd_default(msg.currency());
273    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
274    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
275    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
276    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
277    let ts_init = ts_init.unwrap_or(ts_event);
278
279    Ok(FuturesSpread::new_checked(
280        instrument_id,
281        instrument_id.symbol,
282        asset_class.unwrap_or(AssetClass::Commodity),
283        Some(exchange),
284        underlying,
285        strategy_type,
286        decode_optional_timestamp(msg.activation).unwrap_or_default(),
287        decode_timestamp(msg.expiration, "expiration")?,
288        currency,
289        price_increment.precision,
290        price_increment,
291        multiplier,
292        lot_size,
293        None, // max_quantity
294        None, // min_quantity
295        None, // max_price
296        None, // min_price
297        None, // margin_init
298        None, // margin_maint
299        None, // maker_fee
300        None, // taker_fee
301        None, // tick_scheme
302        None, // info
303        ts_event,
304        ts_init,
305    )?)
306}
307
308/// Decodes a Databento instrument definition message into an `OptionContract` instrument.
309///
310/// # Errors
311///
312/// Returns an error if parsing or constructing `OptionContract` fails.
313pub fn decode_option_contract(
314    msg: &dbn::InstrumentDefMsg,
315    instrument_id: InstrumentId,
316    ts_init: Option<UnixNanos>,
317    decode_config: Option<&DatabentoDecodeConfig>,
318) -> anyhow::Result<OptionContract> {
319    let currency = parse_currency_or_usd_default(msg.currency());
320    let strike_price_currency = parse_currency_or_usd_default(msg.strike_price_currency());
321    let exchange = Ustr::from(msg.exchange()?);
322    let underlying = decode_underlying(msg.underlying()?, &instrument_id.symbol);
323    let asset_class_opt = if instrument_id.venue.as_str() == "OPRA" {
324        Some(AssetClass::Equity)
325    } else {
326        let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
327        asset_class
328    };
329    let option_kind = parse_option_kind(msg.instrument_class)?;
330    let strike_price = decode_price(
331        msg.strike_price,
332        strike_price_currency.precision,
333        "strike_price",
334    )?;
335    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
336    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
337    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
338    let expiration = corrected_option_expiration(
339        decode_timestamp(msg.expiration, "expiration")?,
340        underlying,
341        msg.hd.publisher().ok().map(|p| p.dataset()),
342        decode_config,
343    );
344    let ts_event = UnixNanos::from(msg.ts_recv); // More accurate and reliable timestamp
345    let ts_init = ts_init.unwrap_or(ts_event);
346
347    Ok(OptionContract::new_checked(
348        instrument_id,
349        instrument_id.symbol,
350        asset_class_opt.unwrap_or(AssetClass::Commodity),
351        Some(exchange),
352        underlying,
353        option_kind,
354        strike_price,
355        currency,
356        decode_optional_timestamp(msg.activation).unwrap_or_default(),
357        expiration,
358        price_increment.precision,
359        price_increment,
360        multiplier,
361        lot_size,
362        None, // max_quantity
363        None, // min_quantity
364        None, // max_price
365        None, // min_price
366        None, // margin_init
367        None, // margin_maint
368        None, // maker_fee
369        None, // taker_fee
370        None, // tick_scheme
371        None, // info
372        ts_event,
373        ts_init,
374    )?)
375}
376
377/// Decodes a Databento instrument definition message into an `OptionSpread` instrument.
378///
379/// # Errors
380///
381/// Returns an error if parsing or constructing `OptionSpread` fails.
382pub fn decode_option_spread(
383    msg: &dbn::InstrumentDefMsg,
384    instrument_id: InstrumentId,
385    ts_init: Option<UnixNanos>,
386    decode_config: Option<&DatabentoDecodeConfig>,
387) -> anyhow::Result<OptionSpread> {
388    let exchange = Ustr::from(msg.exchange()?);
389    let underlying = decode_underlying(msg.underlying()?, &instrument_id.symbol);
390    let asset_class_opt = if instrument_id.venue.as_str() == "OPRA" {
391        Some(AssetClass::Equity)
392    } else {
393        let (asset_class, _) = parse_cfi_iso10926(msg.cfi()?);
394        asset_class
395    };
396    let strategy_type = Ustr::from(msg.secsubtype()?);
397    let currency = parse_currency_or_usd_default(msg.currency());
398    let price_increment = decode_price_increment(msg.min_price_increment, currency.precision);
399    let multiplier = decode_multiplier(msg.unit_of_measure_qty)?;
400    let lot_size = decode_lot_size(msg.min_lot_size_round_lot);
401    let expiration = corrected_option_expiration(
402        decode_timestamp(msg.expiration, "expiration")?,
403        underlying,
404        msg.hd.publisher().ok().map(|p| p.dataset()),
405        decode_config,
406    );
407    let ts_event = msg.ts_recv.into(); // More accurate and reliable timestamp
408    let ts_init = ts_init.unwrap_or(ts_event);
409
410    Ok(OptionSpread::new_checked(
411        instrument_id,
412        instrument_id.symbol,
413        asset_class_opt.unwrap_or(AssetClass::Commodity),
414        Some(exchange),
415        underlying,
416        strategy_type,
417        decode_optional_timestamp(msg.activation).unwrap_or_default(),
418        expiration,
419        currency,
420        price_increment.precision,
421        price_increment,
422        multiplier,
423        lot_size,
424        None, // max_quantity
425        None, // min_quantity
426        None, // max_price
427        None, // min_price
428        None, // margin_init
429        None, // margin_maint
430        None, // maker_fee
431        None, // taker_fee
432        None, // tick_scheme
433        None, // info
434        ts_event,
435        ts_init,
436    )?)
437}
nautilus_databento/decode/instruments.rs

nautilus_databento/decode/
instruments.rs
