apple_quant_algorithmic/backends/

databento.rs

mod decoder;
mod encoder;
mod historical;
mod live;
mod symbology;

use std::{marker::PhantomData, path::PathBuf, range::Range};

#[allow(unused_imports)]
use apple_quant_core::log::trace;

use apple_quant_core::{
	log::{error, info},
	UnwrapResultExt,
};

use databento::{
	dbn::{
		decode::AsyncDbnDecoder, RecordRef, RecordRefEnum, Schema, Side,
		FIXED_PRICE_SCALE,
	},
	DateTimeLike,
};

use smallstr::SmallString;

use time::{Duration, OffsetDateTime, Time, UtcDateTime, UtcOffset};

use tokio::{
	io::{BufReader, BufWriter},
	fs::File, task::yield_now, time::sleep,
};

use crate::{
	aggregation::{Trade, TradeTradeTimestamp},
	backend::{
		DataBackend, HistoricalDataBackend, MarketDataDecoder, MarketDataDecoderProvider,
		OrderIdGenerator, OrdersBackend, OrdersBackendUpdate, OrdersBackendUpdateRecycle,
		RealtimeDataBackend,
	},
	instrument::{
		InstrumentData, InstrumentSpec, InstrumentTicker, IsFloatingPoint, XSpec,
	},
	order::{DeferredOrderActions, StateGoal},
	order_manager::{OrderManager, OrdersCapacitySpec},
	points::{
		Subpoints, SubpointsType, VolumeFromWholePoints, WholePoints, WholePointsType,
	},
	timestamp::{
		TickTimestamp, Timestamp, TimestampRangeExcluded, TimestampRangeExclusive,
		Timestamped, TradeTimestamp, TradeTimestamped, UtcNs,
	},
	volume::{
		AggressiveVolume, AggressorSide, DirectionalExposure, DirectionlessVolume,
		VolumeConstruct,
	},
	iso_string_date, price::AbsolutePrice, schema::SchemaFlags, strategy::Strategy,
	Frontend,
};

use decoder::*;
use encoder::*;
use historical::*;
use live::*;
use symbology::*;

impl From<&Schema> for SchemaFlags {
	fn from(
		value: &Schema,
	) -> Self {
		match value {
			// Self::Mbo => SchemaFlags::AddCancelModifyFillClear | SchemaFlags::Trades,
			// Self::Mbp10 => SchemaFlags::LimitedAddCancelModifyFillClear | SchemaFlags::Trades,
			// Self::Mbp1 => SchemaFlags::LimitedAddCancelModifyFillClear | SchemaFlags::Trades,
			// Self::Tbbo => SchemaFlags::Trades,
			Schema::Trades => SchemaFlags::Trades,
			// Self::Ohlcv1S => SchemaFlags::OHLC1s,
			// Self::Ohlcv1M => SchemaFlags::OHLC1m,
			_ => SchemaFlags::empty(),
		}
	}
}

impl From<Schema> for SchemaFlags {
	fn from(
		value: Schema,
	) -> Self {
		Self::from(&value)
	}
}

pub(crate) fn schema_value(
	schema: &Schema,
) -> u8 {
	match schema {
		Schema::Mbo => 8,
		Schema::Mbp10 => 7,
		Schema::Mbp1 => 6,
		Schema::Tbbo => 5,
		Schema::Tcbbo => 4,
		Schema::Trades => 3,
		Schema::Ohlcv1S => 2,
		Schema::Ohlcv1M => 1,
		_ => 0,
	}
}

impl DateTimeLike for Timestamp {
	fn to_date_time(
		self,
	) -> OffsetDateTime {
		OffsetDateTime::from_unix_timestamp_nanos(
			*self.as_timestamp_type(),
		).unwrap()
	}
}

pub struct Databento<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS>,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS>,
> {
	frontend: Frontend<'instrument_data, 'aggregated_data, IS, OB, OrdersCS, S>,

	orders_backend_update_receiver: thingbuf::mpsc::Receiver<
		OrdersBackendUpdate<IS>,
		OrdersBackendUpdateRecycle,
	>,

	key: Option<SmallString<[u8; 64]>>,
	realtime: Option<DatabentoLive>,
	historical: Option<DatabentoHistorical>,
}

impl<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS>,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS>,
> Databento<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> {
	const HISTORICAL_LIVE_OVERLAP_SHORT: Duration = Duration::seconds(5);
	const HISTORICAL_LIVE_OVERLAP_LONG: Duration = Duration::minutes(10);

	pub fn new(
		key: Option<&str>,
		frontend: Frontend<'instrument_data, 'aggregated_data, IS, OB, OrdersCS, S>,
		orders_backend_update_receiver: thingbuf::mpsc::Receiver<
			OrdersBackendUpdate<IS>,
			OrdersBackendUpdateRecycle,
		>,
	) -> Self {
		Databento::<'instrument_data, 'aggregated_data> {
			frontend,
			orders_backend_update_receiver,
			key: key.map(|key| SmallString::from_str(key)),
			realtime: None,
			historical: None,
		}
	}

	async fn new_realtime(
		&mut self,
		instrument_ticker: &InstrumentTicker,
		start_timestamp: impl DateTimeLike,
	) -> Result<&mut DatabentoLive, ()> {
		if let Some(
			mut realtime_databento,
		) = self.realtime.take() {
			let _ = realtime_databento.client.close().await;
		}

		let Some(
			key,
		) = &self.key else {
			return Err(());
		};

		let Ok((
			realtime_databento,
			_schema_flags,
		)) = DatabentoLive::new(
			instrument_ticker,
			start_timestamp,
			&[Schema::Trades],
			key.as_str(),
		).await else {
			return Err(());
		};

		unsafe {
			self.realtime = Some(realtime_databento);
			Ok(self.realtime.as_mut().unwrap_unchecked())
		}
	}

	async fn current_realtime(
		&mut self,
	) -> Option<&mut DatabentoLive> {
		self.realtime.as_mut()
	}

	async fn initialize_historical(
		&mut self,
	) -> Result<&mut DatabentoHistorical, ()> {
		let Some(
			key,
		) = &self.key else {
			return Err(());
		};

		self.historical = Some(DatabentoHistorical::new(key.as_str()));

		unsafe {
			Ok(self.historical.as_mut().unwrap_unchecked())
		}
	}

	async fn databento_historical(
		&mut self,
	) -> Result<&mut DatabentoHistorical, ()> {
		if self.historical.is_none() {
			return self.initialize_historical().await;
		}

		let Some(
			databento_historical,
		) = self.historical.as_mut() else {
			return Err(());
		};

		Ok(databento_historical)
	}
}

impl<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS>,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS>,
> HistoricalDataBackend<IS> for Databento<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> {
	async fn fetch_once(
		&mut self,
		instrument_ticker: &InstrumentTicker,
		utc_date_time_range: Range<UtcDateTime>,
	) -> impl IntoIterator<Item = TradeTradeTimestamp<IS>> {
		debug_assert!(utc_date_time_range.start <= utc_date_time_range.end);

		let now = UtcDateTime::now();
		let exact_cutoff = now - Duration::hours(24);
		let live_safe_cutoff = exact_cutoff + Self::HISTORICAL_LIVE_OVERLAP_SHORT;
		let historical_safe_cutoff = exact_cutoff - Self::HISTORICAL_LIVE_OVERLAP_SHORT;

		debug_assert!(utc_date_time_range.end < (now + Duration::seconds(5)));

		let mut market_data = Vec::with_capacity(1_000_000);

		// Historical client only.
		if (
			utc_date_time_range.start <= historical_safe_cutoff &&
			utc_date_time_range.end <= historical_safe_cutoff
		) {
			let historical_databento = self.databento_historical().await.unwrap();

			let mut stream = historical_databento
				.stream(
					instrument_ticker,
					utc_date_time_range,
				).await.unwrap();

			while let Ok(Some(
				record_ref,
			)) = stream.decode_record_ref().await {
				let Some(
					trade_trade_timestamp,
				) = process_record_trades(record_ref) else {
					continue;
				};

				market_data.push(trade_trade_timestamp);
			}

			return market_data;
		}

		// Live client only.
		if (
			utc_date_time_range.start >= live_safe_cutoff &&
			utc_date_time_range.end >= live_safe_cutoff
		) {
			let realtime_databento = self.new_realtime(
				instrument_ticker,
				utc_date_time_range.start.to_offset(UtcOffset::UTC),
			).await.unwrap();

			while let Ok(Some(
				record_ref,
			)) = realtime_databento.client.next_record().await {
				let Some(
					trade_trade_timestamp,
				) = process_record_trades(record_ref) else {
					continue;
				};

				let trade_date_time = trade_trade_timestamp
					.trade_timestamp()
					.to_date_time().to_utc();

				if trade_date_time >= utc_date_time_range.end {
					break;
				}

				market_data.push(trade_trade_timestamp);
			}

			let _ = realtime_databento.client.close().await;
			self.realtime = None;
			return market_data;
		}

		let historical_start = if utc_date_time_range.start < historical_safe_cutoff {
			utc_date_time_range.start
		} else {
			historical_safe_cutoff
		};

		let realtime_end = if utc_date_time_range.end > live_safe_cutoff {
			utc_date_time_range.end
		} else {
			live_safe_cutoff
		};

		let realtime_databento = self.new_realtime(
			instrument_ticker,
			OffsetDateTime::UNIX_EPOCH,
		).await.unwrap();

		let mut realtime_satisfied_start = false;

		while let Ok(Some(
			record_ref,
		)) = realtime_databento.client.next_record().await {
			let Some(
				trade_trade_timestamp,
			) = process_record_trades(record_ref) else {
				continue;
			};

			let trade_date_time = trade_trade_timestamp
				.trade_timestamp()
				.to_date_time().to_utc();

			if trade_date_time >= realtime_end {
				break;
			}

			if trade_date_time < historical_start {
				realtime_satisfied_start = true;
				continue;
			}

			market_data.push(trade_trade_timestamp);
		}

		let _ = realtime_databento.client.close().await;
		self.realtime = None;

		if realtime_satisfied_start {
			return market_data;
		}

		let first_live_timestamp = market_data
			.first()
			.unwrap()
			.trade_timestamp()
			.to_date_time().to_utc();

		debug_assert!(first_live_timestamp < live_safe_cutoff);

		let historical_databento = self.databento_historical().await.unwrap();
		let start_utc = UtcDateTime::now();

		while UtcDateTime::now() - start_utc < Duration::seconds(30) {
			let available_end = historical_databento.available_end().await.unwrap();

			if available_end > first_live_timestamp {
				break;
			}

			sleep(std::time::Duration::from_millis(100)).await;
		}

		let mut historical_market_data = Vec::with_capacity(10_000);

		let mut stream = historical_databento
			.stream(
				instrument_ticker,
				Range {
					start: historical_start,
					end: (first_live_timestamp + Duration::nanoseconds(1)),
				},
			).await.unwrap();

		while let Ok(Some(
			record_ref,
		)) = stream.decode_record_ref().await {
			let Some(
				trade_trade_timestamp,
			) = process_record_trades(record_ref) else {
				continue;
			};

			historical_market_data.push(trade_trade_timestamp);
		}

		let inclusive_start_live_idx = market_data.iter().position(|
			trade_trade_timestamp,
		| {
			trade_trade_timestamp
				.trade_timestamp()
				.to_date_time().to_utc() > first_live_timestamp
		}).unwrap();

		historical_market_data.extend(
			market_data.into_iter().skip(inclusive_start_live_idx),
		);

		historical_market_data
	}

	async fn fetch_save_day(
		&mut self,
		instrument_ticker: &InstrumentTicker,
		date: time::Date,
	) -> impl IntoIterator<Item = TradeTradeTimestamp<IS>> {
		let start = UtcDateTime::new(date, Time::MIDNIGHT);
		let end = UtcDateTime::new(date.next_day().unwrap(), Time::MIDNIGHT);

		let utc_date_time_range = Range { start, end };

		debug_assert!(utc_date_time_range.start <= utc_date_time_range.end);

		let mut iso_string_date = iso_string_date(date).unwrap();

		iso_string_date.push_str(".dbn");

		let mut path_buf = PathBuf::new();

		path_buf.push("market-data");
		path_buf.push(IS::INSTRUMENT_KIND.as_str());
		path_buf.push(IS::BARE_SYMBOL);
		path_buf.push("trades");
		path_buf.push(iso_string_date);

		let first_buf_writer = BufWriter::new(File::create(path_buf).await.unwrap());

		let mut encoder = DatabentoMarketDataEncoder::new(
			first_buf_writer,
			&utc_date_time_range,
			instrument_ticker,
		).await;

		let now = UtcDateTime::now();
		let exact_cutoff = now - Duration::hours(24);
		let live_safe_cutoff = exact_cutoff + Self::HISTORICAL_LIVE_OVERLAP_SHORT;
		let historical_safe_cutoff = exact_cutoff - Self::HISTORICAL_LIVE_OVERLAP_SHORT;

		debug_assert!(utc_date_time_range.end < (now + Duration::seconds(5)));

		let mut market_data = Vec::with_capacity(1_000_000);

		if (
			utc_date_time_range.start > historical_safe_cutoff ||
			utc_date_time_range.end > historical_safe_cutoff
		) {
			unimplemented!();
		}

		let historical_databento = self.databento_historical().await.unwrap();

		let mut stream = historical_databento
			.stream(
				instrument_ticker,
				utc_date_time_range,
			).await.unwrap();

		while let Ok(Some(
			record_ref,
		)) = stream.decode_record_ref().await {
			let Some(
				trade_trade_timestamp,
			) = process_record_trades(record_ref) else {
				continue;
			};

			encoder.encode_record_ref(record_ref).await;
			market_data.push(trade_trade_timestamp);
		}

		encoder.shutdown().await;
		market_data
	}
}

impl<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS>,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS> + Send,
> RealtimeDataBackend<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> for Databento<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> {
	async fn initialize_realtime(
		&mut self,
		instrument_ticker: &InstrumentTicker,
		recent_trade_timestamp: Option<&TradeTimestamp>,
	) {
		let start_timestamp = recent_trade_timestamp
			.map(|
				trade_timestamp,
			| trade_timestamp.to_date_time())
			.unwrap_or_else(|| OffsetDateTime::now_utc());

		let _ = self.new_realtime(
			instrument_ticker,
			start_timestamp,
		).await;
	}

	async fn poll_realtime(
		&mut self,
		instrument_data: &'instrument_data mut InstrumentData<
			'instrument_data,
			'aggregated_data,
			IS,
		>,
		strategy: &mut S,
		order_manager: &OrderManager<IS, OrdersCS>,
		directional_exposure: &DirectionalExposure<IS>,
		state_goal: &mut StateGoal,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		order_id_generator: &mut OrderIdGenerator,
		recent_trade_timestamp: Option<&TradeTimestamp>,
	) {
		let start_recent_trade_timestamp = recent_trade_timestamp.cloned();
		let mut last_recent_trade_timestamp = recent_trade_timestamp.cloned();

		#[rustfmt::skip]
		let Some(
			realtime_databento,
		) = &mut self.realtime else {
			return;
		};

		loop {
			let next_record = realtime_databento.client.try_next_record();

			let record_ref = match next_record {
				Err(
					error,
				) => {
					error!("{error}");
					break;
				},
				Ok(
					record_ref,
				) => record_ref,
			};

			if let Some(
				record_ref,
			) = record_ref {
				let Some(
					trade_trade_timestamp,
				) = process_record_trades::<IS>(record_ref) else {
					continue;
				};

				last_recent_trade_timestamp = Some(
					trade_trade_timestamp.trade_timestamp(),
				);

				instrument_data.new_trades_binned([trade_trade_timestamp].into_iter());
				continue;
			}

			let size = match realtime_databento.client.fill_buf().await {
				Err(
					error,
				) => {
					error!("{error}");
					break;
				},
				Ok(size) => size,
			};

			if size == 0 {
				break;
			}

			if realtime_databento.client.is_closed() {
				unimplemented!();
			}
		}

		let Some(
			last_recent_trade_timestamp,
		) = last_recent_trade_timestamp else {
			return;
		};

		let start_recent_trade_timestamp = start_recent_trade_timestamp.unwrap_or(
			last_recent_trade_timestamp,
		);

		let tick_timestamp = TickTimestamp::from_timestamp(
			start_recent_trade_timestamp.timestamp(),
		);

		strategy.data_update(
			&tick_timestamp,
			instrument_data,
			order_manager,
			directional_exposure,
			state_goal,
			deferred_order_actions,
			order_id_generator,
		).await;
	}
}

impl<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS> + Send,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS> + Send,
> DataBackend<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> for Databento<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> {
	fn new(
		data_key: Option<&str>,
		frontend: Frontend<'instrument_data, 'aggregated_data, IS, OB, OrdersCS, S>,
		orders_backend_update_receiver: thingbuf::mpsc::Receiver<
			OrdersBackendUpdate<IS>,
			OrdersBackendUpdateRecycle,
		>,
	) -> Self {
		Databento::new(data_key, frontend, orders_backend_update_receiver)
	}

	fn file_name_postpend() -> &'static str {
		".dbn"
	}

	async fn backtest(
		&mut self,
		walk_range: TimestampRangeExclusive,
	) {
		let mut tick_timestamp = TickTimestamp::from_timestamp(
			walk_range.start.as_timestamp(),
		);

		self.frontend.initialize(&tick_timestamp).await;

		loop {
			yield_now().await;

			unsafe {
				self.frontend
					.tick(
						&mut self.orders_backend_update_receiver,
						&mut tick_timestamp,
					).await.unwrap_unchecked_release()
			};
		}

		#[cfg(feature = "log-trace-frontend")]
		trace!("Walk loop ended.");
	}

	async fn decode_market_data(
		buf_readers: impl ExactSizeIterator<Item = BufReader<File>>,
	) -> Self::MarketDataDecoder {
		DatabentoMarketDataDecoder::new(buf_readers).await.unwrap()
	}

	fn historical_mut(
		&mut self,
	) -> Option<Result<&mut impl HistoricalDataBackend<IS>, ()>> {
		Some(Ok(self))
	}
}

impl<
	'instrument_data,
	'aggregated_data,
	IS: InstrumentSpec,
	OB: OrdersBackend<IS>,
	OrdersCS: OrdersCapacitySpec,
	S: Strategy<IS, OB, OrdersCS>,
> MarketDataDecoderProvider<IS> for Databento<
	'instrument_data,
	'aggregated_data,
	IS,
	OB,
	OrdersCS,
	S,
> {
	type MarketDataDecoder = DatabentoMarketDataDecoder<IS>;
}