apple_quant_algorithmic/backends/

simulated.rs

use std::{
	marker::PhantomData,
	sync::{
		Arc,
		atomic::{self, AtomicBool},
	},
	time::Instant,
};

use rand::rngs::SmallRng;
use rand_distr::Normal;
use smallvec::SmallVec;
use time::Duration;
use tokio::{sync::mpsc, task::JoinHandle};

use crate::{
	aggregation::TradeTradeTimestamp,
	backend::{
		LocalOrderId, MarketDataStream, OrdersBackend, OrdersBackendSubmitRemoteOrderError,
		RemoteOrderId,
	},
	instrument::InstrumentSpec,
	order::{OrderStateUpdate, RemoteOrder},
	order_manager::CancelRemoteOrderError,
	timestamp::Timestamp,
	volume::DirectionlessVolume,
};

mod orders;
mod remote;
mod requests;
mod specs;

use orders::*;
use remote::*;
use requests::*;

pub use specs::*;

pub struct SimulatedOrdersBackend<
	IS: InstrumentSpec,
	CapacitySpec: specs::CapacitySpec = CapacitySpecLowFrequencyTrades,
	PhysicalLocationLD: specs::PhysicalLocationLD = PhysicalLocationLDProximityBareMetal,
	OrderRouterLD: specs::OrderRouterLD = OrderRounterLDRithmicFull,
	ExchangeLD: specs::ExchangeLD = ExchangeLD_CME,
> {
	rng: SmallRng,
	normal_ns: Normal<f64>,

	min_ns: u32,
	max_ns: u32,

	link_request_sender: mpsc::Sender<(Instant, RemoteRequest<IS>)>,

	tick_trade_trade_timestamps: SmallVec<[TradeTradeTimestamp<IS>; 10]>,
	tick_previous_send_instant: Instant,
	send_trades_flag: Arc<AtomicBool>,

	tick_trades_sender: mpsc::Sender<SmallVec<[TradeTradeTimestamp<IS>; 10]>>,

	remote_join_handle: JoinHandle<()>,

	_capacity_spec: PhantomData<CapacitySpec>,
	_physical_location_ld: PhantomData<PhysicalLocationLD>,
	_order_router_ld: PhantomData<OrderRouterLD>,
	_exchange_ld: PhantomData<ExchangeLD>,
}

impl<
	IS: InstrumentSpec + Send + 'static,
	CapacitySpec: specs::CapacitySpec,
	PhysicalLocationLD: specs::PhysicalLocationLD,
	OrderRouterLD: specs::OrderRouterLD,
	ExchangeLD: specs::ExchangeLD,
> OrdersBackend<IS>
	for SimulatedOrdersBackend<IS, CapacitySpec, PhysicalLocationLD, OrderRouterLD, ExchangeLD>
{
	async fn new() -> (
		mpsc::Receiver<OrderStateUpdate<IS>>,
		Self,
	) {
		let mean_ns = PhysicalLocationLD::LATENCY_DISTRIBUTION.mean_ns
			+ OrderRouterLD::LATENCY_DISTRIBUTION.mean_ns
			+ ExchangeLD::LATENCY_DISTRIBUTION.mean_ns;

		let std_dev_ns = PhysicalLocationLD::LATENCY_DISTRIBUTION.std_dev_ns
			+ OrderRouterLD::LATENCY_DISTRIBUTION.std_dev_ns
			+ ExchangeLD::LATENCY_DISTRIBUTION.std_dev_ns;

		let min_ns = PhysicalLocationLD::LATENCY_DISTRIBUTION.min_ns
			+ OrderRouterLD::LATENCY_DISTRIBUTION.min_ns
			+ ExchangeLD::LATENCY_DISTRIBUTION.min_ns;

		let max_ns = PhysicalLocationLD::LATENCY_DISTRIBUTION.max_ns
			+ OrderRouterLD::LATENCY_DISTRIBUTION.max_ns
			+ ExchangeLD::LATENCY_DISTRIBUTION.max_ns;

		let (link_request_sender, link_request_receiver) = mpsc::channel(8);

		let (order_state_update_sender, order_state_update_receiver) = mpsc::channel(8);

		let (tick_trades_sender, tick_trades_receiver) = mpsc::channel(8);

		let send_trades_flag = Arc::new(AtomicBool::new(false));

		let remote_send_trades_flag = send_trades_flag.clone();

		let remote_join_handle = tokio::spawn(async move {
			remote_link::<IS, CapacitySpec>(
				link_request_receiver,
				order_state_update_sender,
				tick_trades_receiver,
				remote_send_trades_flag,
			)
			.await;
		});

		(
			order_state_update_receiver,
			Self {
				rng: rand::make_rng(),

				normal_ns: Normal::new(
					mean_ns as f64,
					std_dev_ns as f64,
				)
				.unwrap(),
				min_ns,
				max_ns,

				link_request_sender,

				tick_trade_trade_timestamps: SmallVec::default(),
				tick_previous_send_instant: Instant::now(),
				send_trades_flag,

				tick_trades_sender,

				remote_join_handle,

				_capacity_spec: PhantomData::default(),
				_physical_location_ld: PhantomData::default(),
				_order_router_ld: PhantomData::default(),
				_exchange_ld: PhantomData::default(),
			},
		)
	}

	async fn submit_order(
		&mut self,
		local_order_id: &LocalOrderId,
		client_submission_timestamp: &Timestamp,
		remote_order: RemoteOrder<IS>,
	) -> Result<(), OrdersBackendSubmitRemoteOrderError> {
		// let mut sample_ns: f64 = self
		// 	.normal_ns
		// 	.sample(&mut self.rng)
		// 	.max(self.min_ns as f64)
		// 	.min(self.max_ns as f64);

		// if !sample_ns.is_finite() {
		// 	sample_ns = 0.0;
		// }
		// let sample_ns = 100_000_000.0;

		let trigger_instant = Instant::now() + Duration::milliseconds(1);
		// let trigger_timestamp = *client_submission_timestamp + Duration::milliseconds(1);

		let remote_request = RemoteRequest::SubmitOrder {
			local_order_id: *local_order_id,
			remote_order,
		};

		self.link_request_sender
			.send((
				trigger_instant,
				remote_request,
			))
			.await;

		Ok(())
	}

	async fn modify_order_volume(
		&mut self,
		remote_order_id: &RemoteOrderId,
		volume: &DirectionlessVolume<IS>,
	) {
		let instant = Instant::now() + Duration::milliseconds(1);

		let remote_request = RemoteRequest::ModifyVolume {
			remote_order_id: *remote_order_id,
			volume: *volume,
		};

		self.link_request_sender
			.send((instant, remote_request))
			.await
			.unwrap();
	}

	async fn initiate_cancel_order(
		&mut self,
		remote_order_id: &RemoteOrderId,
	) -> Result<(), CancelRemoteOrderError> {
		// let trigger_timestamp = *client_submission_timestamp + Duration::milliseconds(1);
		let instant = Instant::now() + Duration::milliseconds(1);
		let remote_request = RemoteRequest::CancelOrder(*remote_order_id);

		self.link_request_sender
			.send((instant, remote_request))
			.await
			.unwrap();

		Ok(())
	}

	async fn initiate_cancel_all(&mut self) {
		// let _ = client_submission_timestamp;
		// let trigger_timestamp = *client_submission_timestamp + Duration::milliseconds(1);
		let instant = Instant::now() + Duration::milliseconds(1);
		let remote_request = RemoteRequest::CancelAll;

		self.link_request_sender
			.send((instant, remote_request))
			.await
			.unwrap();
	}

	async fn initiate_cancel_flatten_all(&mut self) {
		// let _ = client_submission_timestamp;
		// let trigger_timestamp = *client_submission_timestamp + Duration::milliseconds(1);
		let instant = Instant::now() + Duration::milliseconds(1);
		let remote_request = RemoteRequest::CancelFlattenAll;

		self.link_request_sender
			.send((instant, remote_request))
			.await
			.unwrap();
	}
}

impl<
	IS: InstrumentSpec,
	CapacitySpec: specs::CapacitySpec,
	PhysicalLocationLD: specs::PhysicalLocationLD,
	OrderRouterLD: specs::OrderRouterLD,
	ExchangeLD: specs::ExchangeLD,
> MarketDataStream<IS>
	for SimulatedOrdersBackend<IS, CapacitySpec, PhysicalLocationLD, OrderRouterLD, ExchangeLD>
{
	async fn trades_stream<'a>(
		&mut self,
		mut just_added: impl ExactSizeIterator<Item = &'a TradeTradeTimestamp<IS>> + Clone,
	) where
		IS: 'a,
	{
		if just_added.is_empty() {
			return;
		}

		if !self
			.send_trades_flag
			.load(atomic::Ordering::Relaxed)
		{
			return;
		}

		let now = Instant::now();

		'just_added: loop {
			for _ in (self
				.tick_trade_trade_timestamps
				.len())
				..(self
					.tick_trade_trade_timestamps
					.capacity())
			{
				let Some(trade_trade_timestamp) = just_added.next() else {
					break 'just_added;
				};

				self.tick_trade_trade_timestamps
					.push(*trade_trade_timestamp);
			}

			debug_assert!(
				!self
					.tick_trade_trade_timestamps
					.spilled()
			);

			if self
				.tick_trade_trade_timestamps
				.len() == self
				.tick_trade_trade_timestamps
				.capacity()
			{
				self.tick_trades_sender
					.send(
						self.tick_trade_trade_timestamps
							.clone(),
					)
					.await
					.unwrap();

				self.link_request_sender
					.send((
						Instant::now(),
						RemoteRequest::MarketDataUpdate,
					))
					.await
					.unwrap();

				self.tick_trade_trade_timestamps
					.clear();

				self.tick_previous_send_instant = now;
			}
		}

		if self
			.tick_trade_trade_timestamps
			.is_empty()
		{
			return;
		}

		if self
			.tick_previous_send_instant
			.elapsed()
			.as_micros() as u64
			>= 100
		{
			self.tick_trades_sender
				.send(
					self.tick_trade_trade_timestamps
						.clone(),
				)
				.await
				.unwrap();

			self.link_request_sender
				.send((
					Instant::now(),
					RemoteRequest::MarketDataUpdate,
				))
				.await
				.unwrap();

			self.tick_trade_trade_timestamps
				.clear();

			self.tick_previous_send_instant = now;
		}
	}
}