use super::error::DatabentoErrorKind;
use crate::books::Level;
use crate::error::DataError;
use crate::subscription::book::OrderBookL1;
use crate::subscription::candle::{Candle, CandleInterval, close_time_from_open};
use crate::subscription::{quote::Quote, trade::PublicTrade};
use chrono::{DateTime, TimeZone, Utc};
use databento::dbn::{Mbp1Msg, OhlcvMsg, Schema, TradeMsg};
use rust_decimal::Decimal;
use rustrade_instrument::Side;
use rustrade_instrument::exchange::ExchangeId;
use smol_str::format_smolstr;
const UNDEF_PRICE: i64 = i64::MAX;
const UNDEF_SIZE: u32 = u32::MAX;
pub fn dbn_trade_to_public_trade(
trade: &TradeMsg,
) -> Result<(DateTime<Utc>, PublicTrade), &'static str> {
if trade.price == UNDEF_PRICE {
return Err("undefined price");
}
let price = Decimal::new(trade.price, 9);
let amount = Decimal::from(trade.size);
let time_exchange = nanos_to_datetime(trade.hd.ts_event)?;
#[allow(clippy::cast_sign_loss)]
let side = match trade.side as u8 {
b'A' => Some(Side::Sell),
b'B' => Some(Side::Buy),
b'N' => None,
_ => return Err("unknown trade side value"),
};
Ok((
time_exchange,
PublicTrade {
id: format_smolstr!("{}", trade.sequence),
price,
amount,
side,
},
))
}
pub fn dbn_mbp1_to_quote(msg: &Mbp1Msg) -> Result<(DateTime<Utc>, Quote), &'static str> {
let [level] = &msg.levels;
if level.bid_px == UNDEF_PRICE || level.ask_px == UNDEF_PRICE {
return Err("undefined bid or ask price");
}
let bid_price = Decimal::new(level.bid_px, 9);
let ask_price = Decimal::new(level.ask_px, 9);
let bid_amount = if level.bid_sz == UNDEF_SIZE {
Decimal::ZERO
} else {
Decimal::from(level.bid_sz)
};
let ask_amount = if level.ask_sz == UNDEF_SIZE {
Decimal::ZERO
} else {
Decimal::from(level.ask_sz)
};
let time_exchange = nanos_to_datetime(msg.hd.ts_event)?;
Ok((
time_exchange,
Quote {
bid_price,
bid_amount,
ask_price,
ask_amount,
},
))
}
pub fn dbn_mbp1_to_orderbook_l1(
msg: &Mbp1Msg,
) -> Result<(DateTime<Utc>, OrderBookL1), &'static str> {
let [level] = &msg.levels;
let time_exchange = nanos_to_datetime(msg.hd.ts_event)?;
let best_bid = if level.bid_px != UNDEF_PRICE {
let price = Decimal::new(level.bid_px, 9);
let amount = if level.bid_sz == UNDEF_SIZE {
Decimal::ZERO
} else {
Decimal::from(level.bid_sz)
};
Some(Level { price, amount })
} else {
None
};
let best_ask = if level.ask_px != UNDEF_PRICE {
let price = Decimal::new(level.ask_px, 9);
let amount = if level.ask_sz == UNDEF_SIZE {
Decimal::ZERO
} else {
Decimal::from(level.ask_sz)
};
Some(Level { price, amount })
} else {
None
};
Ok((
time_exchange,
OrderBookL1 {
last_update_time: time_exchange,
best_bid,
best_ask,
},
))
}
fn nanos_to_datetime(nanos: u64) -> Result<DateTime<Utc>, &'static str> {
let secs = i64::try_from(nanos / 1_000_000_000).map_err(|_| "timestamp out of i64 range")?;
let nsecs = (nanos % 1_000_000_000) as u32;
Utc.timestamp_opt(secs, nsecs)
.single()
.ok_or("invalid timestamp")
}
fn ohlcv_conversion_error(message: String) -> DataError {
DataError::Databento {
kind: DatabentoErrorKind::Decode,
context: "converting OHLCV record".to_string(),
message,
}
}
pub fn dbn_ohlcv_to_candle(
msg: &OhlcvMsg,
interval: CandleInterval,
) -> Result<(DateTime<Utc>, Candle), DataError> {
if msg.open == UNDEF_PRICE
|| msg.high == UNDEF_PRICE
|| msg.low == UNDEF_PRICE
|| msg.close == UNDEF_PRICE
{
return Err(ohlcv_conversion_error(
"undefined price (UNDEF_PRICE sentinel) in OHLCV record".to_string(),
));
}
let open_time = nanos_to_datetime(msg.hd.ts_event)
.map_err(|e| ohlcv_conversion_error(format!("OHLCV ts_event: {e}")))?;
let close_time = close_time_from_open(open_time, interval.to_step()).ok_or_else(|| {
ohlcv_conversion_error(format!(
"OHLCV close_time overflow: open={open_time}, interval={}",
interval.as_str()
))
})?;
Ok((
close_time,
Candle {
close_time,
open: Decimal::new(msg.open, 9),
high: Decimal::new(msg.high, 9),
low: Decimal::new(msg.low, 9),
close: Decimal::new(msg.close, 9),
volume: Decimal::from(msg.volume),
trade_count: 0,
},
))
}
pub fn ensure_databento_ohlcv_supports(
exchange: ExchangeId,
interval: CandleInterval,
) -> Result<Schema, DataError> {
match interval {
CandleInterval::Sec1 => Ok(Schema::Ohlcv1S),
CandleInterval::Min1 => Ok(Schema::Ohlcv1M),
CandleInterval::Hour1 => Ok(Schema::Ohlcv1H),
CandleInterval::Day1 => Ok(Schema::Ohlcv1D),
CandleInterval::Min3
| CandleInterval::Min5
| CandleInterval::Min15
| CandleInterval::Min30
| CandleInterval::Hour2
| CandleInterval::Hour4
| CandleInterval::Hour6
| CandleInterval::Hour8
| CandleInterval::Hour12
| CandleInterval::Day3
| CandleInterval::Week1
| CandleInterval::Month1 => Err(DataError::UnsupportedInterval { exchange, interval }),
}
}
#[must_use]
pub fn rtype_to_candle_interval(rtype: u8) -> Option<CandleInterval> {
use databento::dbn::enums::rtype;
match rtype {
rtype::OHLCV_1S => Some(CandleInterval::Sec1),
rtype::OHLCV_1M => Some(CandleInterval::Min1),
rtype::OHLCV_1H => Some(CandleInterval::Hour1),
rtype::OHLCV_1D => Some(CandleInterval::Day1),
_ => None,
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
use super::*;
use std::str::FromStr;
#[test]
fn test_trade_conversion() {
use rust_decimal_macros::dec;
let mut trade = TradeMsg::default();
trade.hd.ts_event = 1_700_000_000_000_000_000;
trade.price = 150_250_000_000;
trade.size = 100;
trade.side = b'B' as i8;
trade.sequence = 12345;
let (time, public_trade) = dbn_trade_to_public_trade(&trade).unwrap();
assert_eq!(public_trade.price, dec!(150.25));
assert_eq!(public_trade.amount, dec!(100));
assert_eq!(public_trade.side, Some(Side::Buy));
assert_eq!(public_trade.id.as_str(), "12345");
assert_eq!(time.timestamp(), 1_700_000_000);
}
#[test]
fn test_trade_sell_side() {
let mut trade = TradeMsg::default();
trade.hd.ts_event = 1_700_000_000_000_000_000;
trade.price = 100_000_000_000;
trade.size = 50;
trade.side = b'A' as i8;
trade.sequence = 1;
let (_, public_trade) = dbn_trade_to_public_trade(&trade).unwrap();
assert_eq!(public_trade.side, Some(Side::Sell));
}
#[test]
fn test_trade_no_side() {
let mut trade = TradeMsg::default();
trade.hd.ts_event = 1_700_000_000_000_000_000;
trade.price = 100_000_000_000;
trade.size = 10;
trade.side = b'N' as i8;
trade.sequence = 1;
let (_, public_trade) = dbn_trade_to_public_trade(&trade).unwrap();
assert!(public_trade.side.is_none());
}
#[test]
fn test_trade_invalid_side_rejected() {
let mut trade = TradeMsg::default();
trade.hd.ts_event = 1_700_000_000_000_000_000;
trade.price = 100_000_000_000;
trade.size = 10;
trade.side = b'X' as i8;
assert!(dbn_trade_to_public_trade(&trade).is_err());
}
#[test]
fn test_trade_undefined_price() {
let mut trade = TradeMsg::default();
trade.hd.ts_event = 1_700_000_000_000_000_000;
trade.price = i64::MAX;
assert!(dbn_trade_to_public_trade(&trade).is_err());
}
#[test]
fn test_quote_conversion() {
use rust_decimal_macros::dec;
let mut msg = Mbp1Msg::default();
msg.hd.ts_event = 1_700_000_000_000_000_000;
msg.levels[0].bid_px = 100_000_000_000;
msg.levels[0].ask_px = 100_500_000_000;
msg.levels[0].bid_sz = 1000;
msg.levels[0].ask_sz = 500;
let (time, quote) = dbn_mbp1_to_quote(&msg).unwrap();
assert_eq!(quote.bid_price, dec!(100));
assert_eq!(quote.ask_price, dec!(100.5));
assert_eq!(quote.bid_amount, dec!(1000));
assert_eq!(quote.ask_amount, dec!(500));
assert_eq!(time.timestamp(), 1_700_000_000);
}
#[test]
fn test_orderbook_l1_conversion() {
let mut msg = Mbp1Msg::default();
msg.hd.ts_event = 1_700_000_000_000_000_000;
msg.levels[0].bid_px = 100_000_000_000;
msg.levels[0].ask_px = 100_500_000_000;
msg.levels[0].bid_sz = 1000;
msg.levels[0].ask_sz = 500;
let (time, l1) = dbn_mbp1_to_orderbook_l1(&msg).unwrap();
let best_bid = l1.best_bid.unwrap();
let best_ask = l1.best_ask.unwrap();
assert_eq!(best_bid.price, Decimal::from(100));
assert_eq!(best_ask.price, Decimal::from_str("100.5").unwrap());
assert_eq!(best_bid.amount, Decimal::from(1000));
assert_eq!(best_ask.amount, Decimal::from(500));
assert_eq!(time.timestamp(), 1_700_000_000);
assert_eq!(l1.last_update_time.timestamp(), 1_700_000_000);
}
#[test]
fn test_orderbook_l1_undefined_prices() {
let mut msg = Mbp1Msg::default();
msg.hd.ts_event = 1_700_000_000_000_000_000;
msg.levels[0].bid_px = i64::MAX; msg.levels[0].ask_px = i64::MAX; msg.levels[0].bid_sz = 1000;
msg.levels[0].ask_sz = 500;
let (_, l1) = dbn_mbp1_to_orderbook_l1(&msg).unwrap();
assert!(l1.best_bid.is_none());
assert!(l1.best_ask.is_none());
}
fn ohlcv(ts_event: u64, open: i64, high: i64, low: i64, close: i64, volume: u64) -> OhlcvMsg {
let mut msg = OhlcvMsg::default_for_schema(Schema::Ohlcv1M);
msg.hd.ts_event = ts_event;
msg.open = open;
msg.high = high;
msg.low = low;
msg.close = close;
msg.volume = volume;
msg
}
#[test]
fn ohlcv_to_candle_normalises_close_time_from_open() {
use rust_decimal_macros::dec;
let msg = ohlcv(
1_704_067_200_000_000_000,
45_000_500_000_000, 45_500_000_000_000, 44_800_000_000_000, 45_250_000_000_000, 1234,
);
let (time_exchange, candle) = dbn_ohlcv_to_candle(&msg, CandleInterval::Min1).unwrap();
assert_eq!(candle.close_time.timestamp(), 1_704_067_260);
assert_eq!(time_exchange, candle.close_time);
assert_eq!(candle.open, dec!(45000.5));
assert_eq!(candle.high, dec!(45500.0));
assert_eq!(candle.low, dec!(44800.0));
assert_eq!(candle.close, dec!(45250.0));
assert_eq!(candle.volume, dec!(1234));
assert_eq!(candle.trade_count, 0);
}
#[test]
fn ohlcv_to_candle_close_time_per_interval() {
let open_ns = 1_704_067_200_000_000_000; let cases = [
(CandleInterval::Sec1, 1_704_067_201), (CandleInterval::Min1, 1_704_067_260), (CandleInterval::Hour1, 1_704_070_800), (CandleInterval::Day1, 1_704_153_600), ];
for (interval, expected_close_secs) in cases {
let msg = ohlcv(
open_ns,
1_000_000_000,
1_000_000_000,
1_000_000_000,
1_000_000_000,
1,
);
let (_, candle) = dbn_ohlcv_to_candle(&msg, interval).unwrap();
assert_eq!(
candle.close_time.timestamp(),
expected_close_secs,
"interval {interval}"
);
}
}
#[test]
fn ohlcv_to_candle_never_overflows_for_max_u64_ts_event() {
let msg = ohlcv(u64::MAX, 1, 1, 1, 1, 1);
let (time_exchange, candle) = dbn_ohlcv_to_candle(&msg, CandleInterval::Day1).unwrap();
assert_eq!(time_exchange, candle.close_time);
}
#[test]
fn ohlcv_to_candle_rejects_undef_price_in_any_field() {
let open_ns = 1_704_067_200_000_000_000; for field in 0..4 {
let mut vals = [1_000_000_000_i64; 4];
vals[field] = UNDEF_PRICE;
let msg = ohlcv(open_ns, vals[0], vals[1], vals[2], vals[3], 1);
assert!(
dbn_ohlcv_to_candle(&msg, CandleInterval::Min1).is_err(),
"expected UNDEF_PRICE in field {field} to be rejected"
);
}
let msg = ohlcv(open_ns, 1, 1, 1, 1, 1);
assert!(dbn_ohlcv_to_candle(&msg, CandleInterval::Min1).is_ok());
}
#[test]
fn ensure_databento_ohlcv_supports_maps_native_intervals() {
let ex = ExchangeId::DatabentoGlbx;
assert_eq!(
ensure_databento_ohlcv_supports(ex, CandleInterval::Sec1).unwrap(),
Schema::Ohlcv1S
);
assert_eq!(
ensure_databento_ohlcv_supports(ex, CandleInterval::Min1).unwrap(),
Schema::Ohlcv1M
);
assert_eq!(
ensure_databento_ohlcv_supports(ex, CandleInterval::Hour1).unwrap(),
Schema::Ohlcv1H
);
assert_eq!(
ensure_databento_ohlcv_supports(ex, CandleInterval::Day1).unwrap(),
Schema::Ohlcv1D
);
}
#[test]
fn ensure_databento_ohlcv_supports_rejects_non_native_intervals() {
let ex = ExchangeId::DatabentoGlbx;
for interval in CandleInterval::ALL {
let native = matches!(
interval,
CandleInterval::Sec1
| CandleInterval::Min1
| CandleInterval::Hour1
| CandleInterval::Day1
);
let result = ensure_databento_ohlcv_supports(ex, interval);
assert_eq!(result.is_ok(), native, "interval {interval}");
if !native {
assert!(
matches!(
result,
Err(DataError::UnsupportedInterval { interval: i, .. }) if i == interval
),
"expected UnsupportedInterval for {interval}"
);
}
}
}
#[test]
fn rtype_to_candle_interval_maps_native_and_skips_others() {
use databento::dbn::enums::rtype;
assert_eq!(
rtype_to_candle_interval(rtype::OHLCV_1S),
Some(CandleInterval::Sec1)
);
assert_eq!(
rtype_to_candle_interval(rtype::OHLCV_1M),
Some(CandleInterval::Min1)
);
assert_eq!(
rtype_to_candle_interval(rtype::OHLCV_1H),
Some(CandleInterval::Hour1)
);
assert_eq!(
rtype_to_candle_interval(rtype::OHLCV_1D),
Some(CandleInterval::Day1)
);
assert_eq!(rtype_to_candle_interval(rtype::OHLCV_EOD), None);
assert_eq!(rtype_to_candle_interval(0x11), None);
assert_eq!(rtype_to_candle_interval(rtype::MBP_1), None);
}
}