apple_quant_algorithmic/order_manager/

remote_pending.rs

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

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

use crate::{
	backend::{
		LocalOrderId, OrderIdError, OrdersBackend, OrdersBackendSubmitRemoteOrderError,
		RemoteOrderId,
	},
	order::{
		DeferredOrderActions, DesiredVolumeOrder, DesiredVolumeOrderError, OrderAction,
		RemoteOrder, RemoteOrderTracker, RemoteTimingCondition, TriggerError,
	},
	timestamp::{TickTimestamp, Timestamp, Timestamped},
	volume::{DirectionalExposure, DirectionalIntentVolume, ZeroableVolume},
	instrument::InstrumentSpec, price::AbsolutePrice,
};

use super::{
	OrdersCapacitySpec, PendingClientCancelError, WorkingRemoteOrderSubmitError,
	WorkingRemoteOrders,
};

#[derive(Debug, Error)]
pub enum PendingRemoteSubmitError {
	#[error("{0:?}")]
	DesiredVolumeOrder(#[from] DesiredVolumeOrderError),

	#[error("{0:?}")]
	OrdersBackendSubmitRemoteOrder(#[from] OrdersBackendSubmitRemoteOrderError),
}

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

	#[error("{0:?}")]
	PendingClientCancel(#[from] PendingClientCancelError),

	#[error("{0:?}")]
	WorkingRemoteOrderSubmit(#[from] WorkingRemoteOrderSubmitError),

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

pub struct PendingRemoteOrder<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> {
	pub(crate) submission_timestamp: Timestamp,
	pub(crate) local_order_id: LocalOrderId,

	pub(crate) dependencies: SmallVec<[(
		RemoteTimingCondition,
		OrderAction<IS>,
	); core::direct_const_arg!(CS::DEPENDENCY)]>,

	pub(crate) desired_directional_intent_volume: DirectionalIntentVolume<IS>,
	pub(crate) rest_at: Option<AbsolutePrice<IS>>,
	pub(crate) cancel_on_process: bool,
}

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> PendingRemoteOrder<IS, CS> {
	pub(crate) fn new(
		submission_timestamp: Timestamp,
		local_order_id: LocalOrderId,
		dependencies: impl IntoIterator<Item = (
			RemoteTimingCondition,
			OrderAction<IS>,
		)>,
		desired_directional_intent_volume: DirectionalIntentVolume<IS>,
		rest_at: Option<AbsolutePrice<IS>>,
	) -> Self {
		Self {
			submission_timestamp,
			local_order_id,
			dependencies: dependencies.into_iter().collect(),
			desired_directional_intent_volume,
			rest_at,
			cancel_on_process: false,
		}
	}
}

/// Unordered list of pending remote orders. The order will be booked into a fill, partial fill, or resting state before this client knows. Once the client recognizes the booking, the order will be moved to [`working_remote_orders`].
pub struct PendingRemoteOrders<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
>(SmallVec<[PendingRemoteOrder<IS, CS>; core::direct_const_arg!(CS::PENDING_REMOTE)]>);

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> PendingRemoteOrders<IS, CS> {
	#[instrument(skip_all)]
	pub async fn submit<OB: OrdersBackend<IS>>(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		orders_backend: &mut OB,
		tick_timestamp: &TickTimestamp,
		remote_order_tracker: RemoteOrderTracker,
		remote_order: RemoteOrder<IS>,
		dependencies: impl IntoIterator<Item = (
			RemoteTimingCondition,
			OrderAction<IS>,
		)>,
	) -> Result<(), PendingRemoteSubmitError>
	where
		IS: Send,
	{
		let submission_timestamp = tick_timestamp.timestamp();
		let local_order_id = *remote_order_tracker.as_local_order_id();

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

		let desired_directional_intent_volume = remote_order.desired_directional_intent_volume()?;

		let rest_at = if let RemoteOrder::Limit(
			limit_order,
		) = &remote_order {
			Some(limit_order.price.into_absolute().unwrap())
		} else {
			None
		};

		orders_backend.submit_order(
			&local_order_id,
			&submission_timestamp,
			remote_order,
		).await?;

		let pending_remote_order = PendingRemoteOrder::new(
			submission_timestamp,
			local_order_id,
			dependencies,
			desired_directional_intent_volume,
			rest_at,
		);

		self.0.push(pending_remote_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<&PendingRemoteOrder<IS, CS>, OrderIdError> {
		let Some(
			pending_remote_order,
		) = self.0.iter().find(|
			pending_remote_order,
		| &pending_remote_order.local_order_id == local_order_id) else {
			return local_order_id.err_invalid();
		};

		Ok(pending_remote_order)
	}

	#[instrument(skip_all)]
	pub(crate) fn get_mut(
		&mut self,
		local_order_id: &LocalOrderId,
	) -> Result<&mut PendingRemoteOrder<IS, CS>, OrderIdError> {
		let Some(
			pending_remote_order,
		) = self.0.iter_mut().find(|
			pending_remote_order,
		| &pending_remote_order.local_order_id == local_order_id) else {
			return local_order_id.err_invalid();
		};

		Ok(pending_remote_order)
	}

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

		let Some(
			idx,
		) = self.0.iter().position(|
			pending_remote_order,
		| &pending_remote_order.local_order_id == local_order_id) else {
			return local_order_id.err_invalid();
		};

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

		let desired_directional_intent_volume = &pending_remote_order.desired_directional_intent_volume;

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

		Ok(pending_remote_order)
	}

	#[instrument(skip_all)]
	pub(crate) async fn acknowledged<OB: OrdersBackend<IS>>(
		&mut self,
		tick_timestamp: &TickTimestamp,
		working_remote_orders: &mut WorkingRemoteOrders<IS, CS>,
		directional_exposure: &mut DirectionalExposure<IS>,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		local_order_id: &LocalOrderId,
		remote_order_id: RemoteOrderId,
	) -> Result<(), PendingRemoteProcessedError>
	where
		IS: Send,
	{
		#[cfg(feature = "log-trace-order-manager")]
		trace!(
			"Acknowledging `{:?}` with `{:?}`.",
			local_order_id,
			remote_order_id,
		);

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

		for (
			remote_timing_condition,
			remote_order_action,
		) in pending_remote_order.dependencies.iter_mut() {
			if remote_timing_condition != &RemoteTimingCondition::Acknowledged {
				continue;
			}

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

		working_remote_orders.submit::<OB>(
			directional_exposure,
			deferred_order_actions,
			tick_timestamp,
			pending_remote_order,
			remote_order_id,
		)?;

		Ok(())
	}
}

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