use std::hash::Hash;
use ahash::AHashSet;
use nautilus_core::Params;
use nautilus_model::{
data::BarType,
identifiers::{ClientId, InstrumentId},
instruments::InstrumentAny,
};
use serde::{Deserialize, de::Error};
use crate::cache::Cache;
#[derive(Debug, thiserror::Error)]
#[error("{context} requires one {kind} on this adapter path, received {count}")]
pub struct IdentifierCountError {
context: &'static str,
kind: &'static str,
count: usize,
}
pub trait IntoInstrumentIds: sealed::Sealed {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>>;
}
impl IntoInstrumentIds for InstrumentId {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
Ok(vec![self])
}
}
impl IntoInstrumentIds for &InstrumentId {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
(*self).into_instrument_ids()
}
}
impl IntoInstrumentIds for Vec<InstrumentId> {
fn into_instrument_ids(mut self) -> anyhow::Result<Vec<InstrumentId>> {
anyhow::ensure!(!self.is_empty(), "instrument IDs cannot be empty");
deduplicate(&mut self);
Ok(self)
}
}
impl IntoInstrumentIds for &Vec<InstrumentId> {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
self.as_slice().into_instrument_ids()
}
}
impl IntoInstrumentIds for &[InstrumentId] {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
self.to_vec().into_instrument_ids()
}
}
impl<const N: usize> IntoInstrumentIds for [InstrumentId; N] {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
Vec::from(self).into_instrument_ids()
}
}
impl<const N: usize> IntoInstrumentIds for &[InstrumentId; N] {
fn into_instrument_ids(self) -> anyhow::Result<Vec<InstrumentId>> {
self.as_slice().into_instrument_ids()
}
}
pub(super) fn deserialize_instrument_ids<'de, D>(
deserializer: D,
) -> Result<Vec<InstrumentId>, D::Error>
where
D: serde::Deserializer<'de>,
{
Vec::<InstrumentId>::deserialize(deserializer)?
.into_instrument_ids()
.map_err(D::Error::custom)
}
pub fn ensure_routable_instrument_ids(
instrument_ids: &[InstrumentId],
has_client_id: bool,
) -> anyhow::Result<()> {
let first = instrument_ids
.first()
.ok_or_else(|| anyhow::anyhow!("instrument IDs cannot be empty"))?;
let mut seen = AHashSet::new();
anyhow::ensure!(
instrument_ids.iter().all(|id| seen.insert(*id)),
"instrument IDs must be unique"
);
anyhow::ensure!(
has_client_id || instrument_ids.iter().all(|id| id.venue == first.venue),
"multiple instrument venues require an explicit client ID"
);
Ok(())
}
pub fn ensure_subscription_routes(
ids: &[InstrumentId],
client_id: Option<&ClientId>,
cache: &Cache,
) -> anyhow::Result<()> {
anyhow::ensure!(
client_id.is_some_and(|id| cache.is_external_data_client(id))
|| ids.len() <= 1
|| !ids.iter().any(InstrumentId::is_synthetic),
"Cannot subscribe synthetic instruments for this data type"
);
Ok(())
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QuoteRouting {
Client,
Synthetic,
Spread,
}
pub fn quote_routing(
ids: &[InstrumentId],
client_id: Option<&ClientId>,
params: Option<&Params>,
cache: &Cache,
) -> anyhow::Result<QuoteRouting> {
let first = ids
.first()
.ok_or_else(|| anyhow::anyhow!("instrument IDs cannot be empty"))?;
if client_id.is_some_and(|id| cache.is_external_data_client(id)) {
return Ok(QuoteRouting::Client);
}
let aggregate_spreads = params
.and_then(|params| params.get_bool("aggregate_spread_quotes"))
.unwrap_or(false);
let classify = |id: &InstrumentId| {
if id.is_synthetic() {
QuoteRouting::Synthetic
} else if aggregate_spreads && cache.instrument(id).is_some_and(InstrumentAny::is_spread) {
QuoteRouting::Spread
} else {
QuoteRouting::Client
}
};
let routing = classify(first);
anyhow::ensure!(
ids.iter().all(|id| classify(id) == routing),
"quote identifier lists cannot mix client, synthetic, and spread routes"
);
Ok(routing)
}
pub(super) fn single_instrument_id(
instrument_ids: &[InstrumentId],
context: &'static str,
) -> anyhow::Result<InstrumentId> {
match instrument_ids {
[instrument_id] => Ok(*instrument_id),
_ => Err(IdentifierCountError {
context,
kind: "instrument ID",
count: instrument_ids.len(),
}
.into()),
}
}
pub trait IntoBarTypes: sealed::Sealed {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>>;
}
impl IntoBarTypes for BarType {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
Ok(vec![self])
}
}
impl IntoBarTypes for &BarType {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
(*self).into_bar_types()
}
}
impl IntoBarTypes for Vec<BarType> {
fn into_bar_types(mut self) -> anyhow::Result<Vec<BarType>> {
anyhow::ensure!(!self.is_empty(), "bar types cannot be empty");
deduplicate(&mut self);
Ok(self)
}
}
impl IntoBarTypes for &Vec<BarType> {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
self.as_slice().into_bar_types()
}
}
impl IntoBarTypes for &[BarType] {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
self.to_vec().into_bar_types()
}
}
impl<const N: usize> IntoBarTypes for [BarType; N] {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
Vec::from(self).into_bar_types()
}
}
impl<const N: usize> IntoBarTypes for &[BarType; N] {
fn into_bar_types(self) -> anyhow::Result<Vec<BarType>> {
self.as_slice().into_bar_types()
}
}
pub(super) fn deserialize_bar_types<'de, D>(deserializer: D) -> Result<Vec<BarType>, D::Error>
where
D: serde::Deserializer<'de>,
{
Vec::<BarType>::deserialize(deserializer)?
.into_bar_types()
.map_err(D::Error::custom)
}
pub fn ensure_routable_bar_types(bar_types: &[BarType], has_client_id: bool) -> anyhow::Result<()> {
let first = bar_types
.first()
.ok_or_else(|| anyhow::anyhow!("bar types cannot be empty"))?;
let mut seen = AHashSet::new();
anyhow::ensure!(
bar_types.iter().all(|bar_type| seen.insert(*bar_type)),
"bar types must be unique"
);
anyhow::ensure!(
has_client_id
|| bar_types
.iter()
.all(|bar_type| bar_type.instrument_id().venue == first.instrument_id().venue),
"multiple bar type venues require an explicit client ID"
);
Ok(())
}
pub(super) fn single_bar_type(
bar_types: &[BarType],
context: &'static str,
) -> anyhow::Result<BarType> {
match bar_types {
[bar_type] => Ok(*bar_type),
_ => Err(IdentifierCountError {
context,
kind: "bar type",
count: bar_types.len(),
}
.into()),
}
}
fn deduplicate<T: Copy + Eq + Hash>(values: &mut Vec<T>) {
if values.len() > 1 {
let mut seen = AHashSet::new();
values.retain(|value| seen.insert(*value));
}
}
mod sealed {
use nautilus_model::{data::BarType, identifiers::InstrumentId};
pub trait Sealed {}
impl Sealed for InstrumentId {}
impl Sealed for &InstrumentId {}
impl Sealed for Vec<InstrumentId> {}
impl Sealed for &Vec<InstrumentId> {}
impl Sealed for &[InstrumentId] {}
impl<const N: usize> Sealed for [InstrumentId; N] {}
impl<const N: usize> Sealed for &[InstrumentId; N] {}
impl Sealed for BarType {}
impl Sealed for &BarType {}
impl Sealed for Vec<BarType> {}
impl Sealed for &Vec<BarType> {}
impl Sealed for &[BarType] {}
impl<const N: usize> Sealed for [BarType; N] {}
impl<const N: usize> Sealed for &[BarType; N] {}
}
#[cfg(test)]
mod tests {
use super::*;
#[rstest::rstest]
fn converts_single_inputs() {
let instrument_id = InstrumentId::from("BTC/USD.SIM");
assert_eq!(
instrument_id.into_instrument_ids().unwrap(),
vec![instrument_id]
);
assert_eq!(
(&instrument_id).into_instrument_ids().unwrap(),
vec![instrument_id]
);
}
#[rstest::rstest]
#[case::singleton(["BTC/USD.SIM"], &["BTC/USD.SIM"])]
#[case::repeated_single_id(
["ETH/USD.SIM", "ETH/USD.SIM", "ETH/USD.SIM"],
&["ETH/USD.SIM"]
)]
#[case::distinct_ids(
["SOL/USD.SIM", "BTC/USD.SIM", "ETH/USD.SIM"],
&["SOL/USD.SIM", "BTC/USD.SIM", "ETH/USD.SIM"]
)]
#[case::interleaved_duplicates_two_ids(
["ETH/USD.SIM", "BTC/USD.SIM", "ETH/USD.SIM"],
&["ETH/USD.SIM", "BTC/USD.SIM"]
)]
#[case::interleaved_duplicates_three_ids(
[
"SOL/USD.SIM", "ETH/USD.SIM", "SOL/USD.SIM", "BTC/USD.SIM",
"ETH/USD.SIM", "BTC/USD.SIM", "SOL/USD.SIM"
],
&["SOL/USD.SIM", "ETH/USD.SIM", "BTC/USD.SIM"]
)]
#[case::same_symbol_different_venues(
["BTC/USD.OTHER", "BTC/USD.SIM", "BTC/USD.OTHER"],
&["BTC/USD.OTHER", "BTC/USD.SIM"]
)]
fn converts_collection_inputs_in_first_occurrence_order<const N: usize>(
#[case] input: [&str; N],
#[case] expected: &[&str],
) {
let input = input.map(InstrumentId::from);
let expected: Vec<_> = expected.iter().copied().map(InstrumentId::from).collect();
let vector = input.to_vec();
assert_eq!(input.into_instrument_ids().unwrap(), expected);
assert_eq!((&input).into_instrument_ids().unwrap(), expected);
assert_eq!(input.as_slice().into_instrument_ids().unwrap(), expected);
assert_eq!((&vector).into_instrument_ids().unwrap(), expected);
assert_eq!(vector.into_instrument_ids().unwrap(), expected);
}
#[rstest::rstest]
fn rejects_empty_collection_inputs() {
let input: [InstrumentId; 0] = [];
let vector = Vec::<InstrumentId>::new();
assert_eq!(
input.into_instrument_ids().unwrap_err().to_string(),
"instrument IDs cannot be empty"
);
assert_eq!(
(&input).into_instrument_ids().unwrap_err().to_string(),
"instrument IDs cannot be empty"
);
assert_eq!(
input
.as_slice()
.into_instrument_ids()
.unwrap_err()
.to_string(),
"instrument IDs cannot be empty"
);
assert_eq!(
(&vector).into_instrument_ids().unwrap_err().to_string(),
"instrument IDs cannot be empty"
);
assert_eq!(
vector.into_instrument_ids().unwrap_err().to_string(),
"instrument IDs cannot be empty"
);
}
#[rstest::rstest]
fn normalizes_full_bar_types_in_first_occurrence_order() {
let first = BarType::from("BTC/USD.SIM-1-MINUTE-LAST-EXTERNAL");
let second = BarType::from("BTC/USD.SIM-5-MINUTE-LAST-EXTERNAL");
let composite = BarType::from("BTC/USD.SIM-5-MINUTE-LAST-INTERNAL@1-MINUTE-EXTERNAL");
let input = [second, first, composite, first, second];
let expected = vec![second, first, composite];
assert_eq!(first.into_bar_types().unwrap(), vec![first]);
assert_eq!((&first).into_bar_types().unwrap(), vec![first]);
assert_eq!(input.into_bar_types().unwrap(), expected);
assert_eq!((&input).into_bar_types().unwrap(), expected);
assert_eq!(input.as_slice().into_bar_types().unwrap(), expected);
assert_eq!(input.to_vec().into_bar_types().unwrap(), expected);
assert_eq!((&input.to_vec()).into_bar_types().unwrap(), expected);
}
#[rstest::rstest]
fn rejects_empty_bar_type_inputs() {
let empty: [BarType; 0] = [];
let vector = Vec::<BarType>::new();
assert_eq!(
empty.into_bar_types().unwrap_err().to_string(),
"bar types cannot be empty"
);
assert!((&empty).into_bar_types().is_err());
assert!(empty.as_slice().into_bar_types().is_err());
assert!((&vector).into_bar_types().is_err());
assert!(vector.into_bar_types().is_err());
}
}