apple_quant_algorithmic/order_manager/

client_pending.rs

use smallvec::SmallVec;
use thiserror::Error;

use crate::{
	aggregation::TradeTradeTimestamp,
	backend::{LocalOrderId, OrderIdError, OrdersBackend},
	instrument::InstrumentSpec,
	liquidity::LiquidityEstimation,
	order::{
		ClientOrder, ClientOrderTracker, DeferredOrderActions, DesiredVolumeOrder,
		DesiredVolumeOrderError, MatchableOrder, TriggerError,
		dependency::client::ClientDeployedRemoteDependency,
	},
	timestamp::{TickTimestamp, Timestamp},
	volume::DirectionalExposure,
};

use super::{OrdersCapacitySpec, PendingRemoteSubmitError};

#[derive(Debug, Error)]
pub enum PendingClientCancelError {
	#[error("{0:?}")]
	OrderId(#[from] OrderIdError),

	#[error("{0:?}")]
	DesiredVolumeOrder(#[from] DesiredVolumeOrderError),
}

#[derive(Debug, Error)]
pub enum PendingClientActivateError {
	#[error("{0:?}")]
	PendingClientCancel(#[from] PendingClientCancelError),

	#[error("{0:?}")]
	PendingRemoteSubmit(#[from] PendingRemoteSubmitError),

	#[error("{0:?}")]
	Trigger(#[from] TriggerError),
}

#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct PendingClientOrder<IS: InstrumentSpec> {
	pub(crate) submission_timestamp: Timestamp,
	pub(crate) local_order_id: LocalOrderId,
	pub(crate) client_deployed_remote_dependency: ClientDeployedRemoteDependency<IS>,
	pub(crate) client_order: ClientOrder<IS>,
	pub(crate) liquidity_estimation: LiquidityEstimation<IS>,
}

impl<IS: InstrumentSpec> PendingClientOrder<IS> {
	pub(crate) fn new(
		submission_timestamp: Timestamp,
		local_order_id: LocalOrderId,
		client_deployed_remote_dependency: ClientDeployedRemoteDependency<IS>,
		client_order: ClientOrder<IS>,
	) -> Self {
		Self {
			submission_timestamp,
			local_order_id,
			client_deployed_remote_dependency,
			client_order,
			liquidity_estimation: LiquidityEstimation::default(),
		}
	}
}

/// Unordered list of pending client orders (eg. stop orders), where only this client knows the order and its intent. When the condition is tripped, it will be converted into a pending remote order, the state transition will take place, and the [`OrdersBackend`] used will become aware of the order.
#[derive(Debug)]
pub struct PendingClientOrders<IS: InstrumentSpec, CS: OrdersCapacitySpec>(
	SmallVec<[PendingClientOrder<IS>; CS::PENDING_CLIENT]>,
);

impl<IS: InstrumentSpec, CS: OrdersCapacitySpec> PendingClientOrders<IS, CS> {
	pub fn submit<OB: OrdersBackend<IS>>(
		&mut self,
		tick_timestamp: &TickTimestamp,
		directional_exposure: &mut DirectionalExposure<IS>,
		client_order_tracker: &ClientOrderTracker,
		client_order: ClientOrder<IS>,
		client_deployed_remote_dependency: ClientDeployedRemoteDependency<IS>,
	) -> Result<(), DesiredVolumeOrderError>
	where
		IS: Send,
	{
		let submission_timestamp = tick_timestamp.timestamp();
		let local_order_id = *client_order_tracker.as_local_order_id();

		let desired_directional_intent_volume = client_deployed_remote_dependency
			.remote_order
			.desired_directional_intent_volume()?;

		let pending_client_order = PendingClientOrder::new(
			submission_timestamp,
			local_order_id,
			client_deployed_remote_dependency,
			client_order,
		);

		self.0
			.push(pending_client_order);

		directional_exposure
			.change_armed_directional_exposure(&desired_directional_intent_volume.as_zeroable());

		Ok(())
	}

	pub fn get(
		&self,
		local_order_id: &LocalOrderId,
	) -> Result<&PendingClientOrder<IS>, OrderIdError> {
		let Some(pending_client_order) = self
			.0
			.iter()
			.find(|pending_client_order| &pending_client_order.local_order_id == local_order_id)
		else {
			return local_order_id.err_invalid();
		};

		Ok(pending_client_order)
	}

	pub(crate) fn get_mut(
		&mut self,
		local_order_id: &LocalOrderId,
	) -> Result<&mut PendingClientOrder<IS>, OrderIdError> {
		let Some(pending_client_order) = self
			.0
			.iter_mut()
			.find(|pending_client_order| &pending_client_order.local_order_id == local_order_id)
		else {
			return local_order_id.err_invalid();
		};

		Ok(pending_client_order)
	}

	pub(crate) fn remove(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		local_order_id: &LocalOrderId,
	) -> Result<PendingClientOrder<IS>, PendingClientCancelError> {
		let Some(idx) = self
			.0
			.iter()
			.position(|pending_client_order| {
				&pending_client_order.local_order_id == local_order_id
			})
		else {
			return local_order_id.err_invalid();
		};

		let pending_client_order = self.0.swap_remove(idx);

		let desired_directional_intent_volume = pending_client_order
			.client_deployed_remote_dependency
			.remote_order
			.desired_directional_intent_volume()?;

		directional_exposure.change_armed_directional_exposure(
			&desired_directional_intent_volume
				.as_zeroable()
				.as_flipped(),
		);

		Ok(pending_client_order)
	}

	pub(crate) async fn activate<OB: OrdersBackend<IS>>(
		&mut self,
		tick_timestamp: &TickTimestamp,
		directional_exposure: &mut DirectionalExposure<IS>,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		local_order_id: &LocalOrderId,
	) -> Result<(), PendingClientActivateError>
	where
		IS: Send,
	{
		let PendingClientOrder {
			submission_timestamp,
			local_order_id,
			client_deployed_remote_dependency,
			client_order,
			liquidity_estimation,
		} = self.remove(
			directional_exposure,
			local_order_id,
		)?;

		deferred_order_actions.push(client_deployed_remote_dependency.into_order_action());

		Ok(())
	}

	pub fn tick_client_orders<'a>(
		&mut self,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		directional_exposure: &mut DirectionalExposure<IS>,
		just_added: impl ExactSizeIterator<Item = &'a TradeTradeTimestamp<IS>> + Clone,
	) where
		IS: 'a,
	{
		let pending_client_orders = self
			.0
			.drain_filter(|pending_client_order| {
				trades_stream_drain_filter(
					pending_client_order,
					directional_exposure,
					just_added.clone(),
				)
			});

		for pending_client_order in pending_client_orders {
			deferred_order_actions.push(
				pending_client_order
					.client_deployed_remote_dependency
					.into_order_action(),
			);
		}
	}
}

impl<IS: InstrumentSpec, CS: OrdersCapacitySpec> Default for PendingClientOrders<IS, CS> {
	fn default() -> Self {
		Self(SmallVec::default())
	}
}

// impl<IS: InstrumentSpec, CS: OrdersCapacitySpec> MarketDataStream<IS>
// 	for PendingClientOrders<IS, CS>
// {
// }

fn trades_stream_drain_filter<'a, IS: InstrumentSpec>(
	pending_client_order: &mut PendingClientOrder<IS>,
	directional_exposure: &mut DirectionalExposure<IS>,
	just_added: impl ExactSizeIterator<Item = &'a TradeTradeTimestamp<IS>> + Clone,
) -> bool
where
	IS: 'a,
{
	pending_client_order
		.liquidity_estimation
		.walk_trades(just_added);

	let is_liquidable = pending_client_order
		.client_order
		.is_liquidable(&pending_client_order.liquidity_estimation)
		.is_some();

	if is_liquidable {
		let delta_directional_exposure = pending_client_order
			.client_deployed_remote_dependency
			.remote_order
			.desired_directional_intent_volume()
			.unwrap()
			.as_zeroable()
			.as_flipped();

		directional_exposure.change_armed_directional_exposure(&delta_directional_exposure);
	}

	is_liquidable
}