apple-quant-algorithmic 0.3.0

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

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

use smallvec::SmallVec;

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

use crate::{
	aggregation::{Trade, TradeTradeTimestamp, TradeTradeTimestampListInline},
	instrument::{InstrumentSpec, IsFloatingPoint, XSpec},
	points::{
		Subpoints, SubpointsType, VolumeFromWholePoints, WholePoints, WholePointsType,
	},
	timestamp::{
		Timestamp, TimestampType, Timestamped, TradeTimestamp, TradeTimestamped, UtcNs,
	},
	volume::{AggressiveVolume, AggressorSide, DirectionlessVolume, VolumeConstruct},
	backend::MarketDataDecoder, price::AbsolutePrice,
};

pub struct DatabentoMarketDataDecoder<IS: InstrumentSpec> {
	dbn_decoder: AsyncDbnDecoder<BufReader<File>>,
	leading_trade: TradeTradeTimestamp<IS>,
	buf_readers: Vec<BufReader<File>>,
}

impl<IS: InstrumentSpec> DatabentoMarketDataDecoder<IS> {
	pub(super) async fn new(
		buf_readers: impl ExactSizeIterator<Item = BufReader<File>>,
	) -> Result<Self, ()> {
		let mut buf_readers: Vec<BufReader<File>> = buf_readers.collect();

		let Some(
			first_buf_reader,
		) = buf_readers.try_remove(0) else {
			return Err(());
		};

		let (
			dbn_decoder,
			leading_trade,
		) = Self::create_dbn_decoder(first_buf_reader).await?;

		Ok(Self {
			dbn_decoder,
			leading_trade,
			buf_readers,
		})
	}

	async fn create_dbn_decoder(
		buf_reader: BufReader<File>,
	) -> Result<
		(
			AsyncDbnDecoder<BufReader<File>>,
			TradeTradeTimestamp<IS>,
		),
		(),
	> {
		let mut dbn_decoder = match AsyncDbnDecoder::new(buf_reader).await {
			Ok(
				dbn_decoder,
			) => dbn_decoder,
			Err(e) => {
				error!("Failed to construct the databento market data decoder: {e}");
				return Err(());
			},
		};

		let record_ref = match dbn_decoder.decode_record_ref().await {
			Ok(Some(
				record_ref,
			)) => record_ref,
			_ => {
				error!("Failed to retrieve first dbn record ref.");
				return Err(());
			},
		};

		let Some(
			leading_trade,
		) = process_record_trades(record_ref) else {
			error!("Failed to process first dbn record ref.");
			return Err(());
		};

		Ok((dbn_decoder, leading_trade))
	}
}

impl<IS: InstrumentSpec> MarketDataDecoder<IS> for DatabentoMarketDataDecoder<IS> {
	async fn decode_trade_list(
		&mut self,
	) -> Option<(
		TradeTradeTimestampListInline<IS, 8>,
		TradeTimestamp,
	)> {
		const GROUP_NS: i64 = 100_000;

		let mut previous_trade = self.leading_trade;

		let mut grouped_trades: SmallVec<[TradeTradeTimestamp<IS>; 8]> = SmallVec::default();

		loop {
			yield_now().await;

			let record_ref = match self.dbn_decoder.decode_record_ref().await {
				Ok(Some(
					record_ref,
				)) => record_ref,
				Ok(None) => {
					grouped_trades.push(previous_trade);

					let Some(
						buf_reader,
					) = self.buf_readers.try_remove(0) else {
						info!("Decoder out of readers.");
						return None;
					};

					let Ok((
						dbn_decoder,
						leading_trade,
					)) = Self::create_dbn_decoder(
						buf_reader,
					).await else {
						return None;
					};

					previous_trade = leading_trade;

					self.dbn_decoder = dbn_decoder;
					self.leading_trade = leading_trade;

					continue;
				},
				Err(_) => {
					error!("Failed to retrieve dbn record ref.");
					return None;
				},
			};

			let Some(
				leading_trade,
			) = process_record_trades(record_ref) else {
				continue;
			};

			let time_delta = leading_trade.timestamp() - self.leading_trade.timestamp();

			let Some(
				delta_nanoseconds,
			) = time_delta.num_nanoseconds() else {
				error!("Failed to process delta time between trades.");
				return None;
			};

			if delta_nanoseconds < 0 {
				error!("Unexpected negative delta time between trades.");
				return None;
			}

			grouped_trades.push(previous_trade);
			previous_trade = leading_trade;

			if delta_nanoseconds > GROUP_NS {
				break;
			}
		}

		self.leading_trade = previous_trade;
		let newest_trade_timestamp = grouped_trades.last().unwrap().trade_timestamp();
		Some((grouped_trades, newest_trade_timestamp))
	}
}

pub(super) fn process_record_trades<IS: InstrumentSpec>(
	record_ref: RecordRef,
) -> Option<TradeTradeTimestamp<IS>> {
	match record_ref.as_enum().unwrap() {
		RecordRefEnum::Mbo(
			mbo_msg,
		) => unimplemented!(),
		RecordRefEnum::Mbp10(msg) => unimplemented!(),
		RecordRefEnum::Mbp1(msg) => unimplemented!(),
		RecordRefEnum::Bbo(msg) => unimplemented!(),
		RecordRefEnum::Cbbo(msg) => unimplemented!(),
		RecordRefEnum::Trade(
			trade_msg,
		) => {
			let Ok(
				side,
			) = trade_msg.side() else {
				return None;
			};

			let aggressor_side = match side {
				Side::Ask => AggressorSide::Ask,
				Side::Bid => AggressorSide::Bid,
				Side::None => return None,
			};

			let trade_trade_timestamp = create_trade_trade_timestamp::<IS>(
				trade_msg.ts_recv,
				trade_msg.price,
				trade_msg.size,
				aggressor_side,
			);

			Some(trade_trade_timestamp)
		},
		RecordRefEnum::Ohlcv(msg) => unimplemented!(),
		_ => None,
	}
}

pub(super) fn create_trade_trade_timestamp<IS: InstrumentSpec>(
	ts_recv: u64,
	price: i64,
	volume: u32,
	aggressor_side: AggressorSide,
) -> TradeTradeTimestamp<IS> {
	let trade_timestamp = unsafe {
		TradeTimestamp::from_timestamp(
			Timestamp::from(UtcNs::new_unchecked(ts_recv as TimestampType)),
		)
	};

	let price = if IS::PriceType::IS_FLOATING_POINT {
		unimplemented!()
	} else {
		price_integer_ticks::<IS>(price)
	};

	let volume: IS::VolumeType = unsafe {
		VolumeFromWholePoints::volume_from_whole_points(
			WholePoints::from_whole_points_type(volume as WholePointsType),
		)
	};

	let directionless_volume = DirectionlessVolume::<IS>::new_unchecked_release(volume);
	let aggressive_volume = AggressiveVolume::new(directionless_volume, aggressor_side);

	let trade = Trade { price, aggressive_volume };

	TradeTradeTimestamp::new(trade, trade_timestamp)
}

#[inline]
pub(super) fn price_integer_ticks<IS: InstrumentSpec>(
	value: i64,
) -> AbsolutePrice<IS> {
	let ticks_per_value = FIXED_PRICE_SCALE / IS::PriceSpec::TICKS_PER_POINT as i64;
	let ticks = value / ticks_per_value;
	let price: IS::PriceType = Subpoints::new(ticks as SubpointsType).into();

	AbsolutePrice::new(price)
}