apple_quant_algorithmic/order_manager/

client_pending.rs

use smallvec::SmallVec;
use thiserror::Error;
use tracing::instrument;

#[allow(unused_imports)]
use tracing::trace;

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

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>,

	pub(crate) dependencies: Vec<(
		ClientTimingCondition,
		OrderAction<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>,
		dependencies: Vec<(
			ClientTimingCondition,
			OrderAction<IS>,
		)>,
	) -> Self {
		Self {
			submission_timestamp,
			local_order_id,
			client_deployed_remote_dependency,
			client_order,
			liquidity_estimation: LiquidityEstimation::default(),
			dependencies,
		}
	}
}

/// 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.
pub struct PendingClientOrders<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
>(SmallVec<[PendingClientOrder<IS>; core::direct_const_arg!(CS::PENDING_CLIENT)]>);

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> PendingClientOrders<IS, CS> {
	#[instrument(skip_all)]
	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>,
		dependencies: Vec<(
			ClientTimingCondition,
			OrderAction<IS>,
		)>,
	) -> Result<(), DesiredVolumeOrderError>
	where
		IS: Send,
	{
		let submission_timestamp = tick_timestamp.timestamp();
		let local_order_id = *client_order_tracker.as_local_order_id();

		#[cfg(feature = "log-trace-order-manager")]
		trace!(
			"Adding id: `{:?}` order: `{:?}`.",
			local_order_id,
			client_order
		);

		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,
			dependencies,
		);

		self.0.push(pending_client_order);

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

		Ok(())
	}

	#[instrument(skip_all)]
	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)
	}

	#[instrument(skip_all)]
	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)
	}

	#[instrument(skip_all)]
	pub(crate) fn remove(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		local_order_id: &LocalOrderId,
	) -> Result<PendingClientOrder<IS>, PendingClientCancelError> {
		#[cfg(feature = "log-trace-order-manager")]
		trace!("Removing id: `{:?}`.", local_order_id);

		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_flipped().as_zeroable(),
		);

		Ok(pending_client_order)
	}

	#[instrument(skip_all)]
	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,
	{
		#[cfg(feature = "log-trace-order-manager")]
		trace!("Activating id: `{:?}`.", local_order_id);

		let pending_client_order = self.remove(directional_exposure, local_order_id)?;

		for (
			client_timing_condition,
			order_action,
		) in pending_client_order.dependencies.iter() {
			if client_timing_condition != &ClientTimingCondition::Activated {
				continue;
			}

			deferred_order_actions.push(order_action.clone());
		}

		deferred_order_actions.push(
			pending_client_order.client_deployed_remote_dependency.into_order_action(),
		);

		Ok(())
	}

	#[instrument(skip_all)]
	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 {
			for (
				client_timing_condition,
				order_action,
			) in pending_client_order.dependencies.iter() {
				if client_timing_condition != &ClientTimingCondition::Activated {
					continue;
				}

				deferred_order_actions.push(order_action.clone());
			}

			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,
> std::fmt::Debug for PendingClientOrders<IS, CS> {
	fn fmt(
		&self,
		f: &mut std::fmt::Formatter<'_>,
	) -> std::fmt::Result {
		self.0.fmt(f)
	}
}

#[instrument(skip_all)]
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 {
		#[cfg(feature = "log-trace-order-manager")]
		trace!(
			"Activating client order `{:?}`",
			pending_client_order.local_order_id
		);

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

		directional_exposure.change_armed_directional_exposure(
			&delta_directional_exposure,
		);
	}

	is_liquidable
}