apple_quant_algorithmic/order_manager/

working.rs

use apple_quant_core::{AddUnchecked, SubChecked, SubUnchecked};

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

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

use crate::backend::LocalOrderId;
use crate::order::OrderAction;
use crate::order::dependency::remote::CancelRemoteOrder;
use crate::volume::ZeroableExt;

use crate::{
	backend::{OrderId, OrderIdError, OrdersBackend, RemoteOrderId},
	order::{
		DeferredOrderActions, DesiredVolumeOrderError, PartialOrderFill,
		RemoteTimingCondition,
	},
	price::{AbsolutePrice, BidAskPriceSpread},
	timestamp::{TickTimestamp, Timestamp, Timestamped},
	volume::{
		DirectionalExposure, DirectionalIntentVolume, DirectionlessVolume, Zeroable,
		ZeroableVolume,
	},
	instrument::InstrumentSpec,
};

use super::{CompletedRemoteOrders, OrdersCapacitySpec, PendingRemoteOrder};

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

	#[error("{0:?}")]
	OrderId(#[from] OrderIdError),
}

pub struct WorkingRemoteOrder<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> {
	pub(crate) local_order_id: LocalOrderId,
	pub(crate) remote_order_id: RemoteOrderId,

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

	pub(crate) desired_directional_intent_volume: DirectionalIntentVolume<IS>,
	pub(crate) booking_timestamp: Timestamp,

	pub(crate) partial_order_fills: SmallVec<[PartialOrderFill<IS>; core::direct_const_arg!(
		CS::PARTIAL_FILL
	)]>,

	pub(crate) armed_directional_exposure: Zeroable<DirectionalIntentVolume<IS>>,
	pub(crate) effective_directional_exposure: Zeroable<DirectionalIntentVolume<IS>>,
	pub(crate) rest_at: Option<AbsolutePrice<IS>>,
}

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> WorkingRemoteOrder<IS, CS> {
	#[instrument(skip_all)]
	pub(crate) fn get_fully_filled(
		&self,
	) -> Option<BidAskPriceSpread<IS>> {
		let mut filled_volume = Zeroable::ZERO;
		let mut iter_partial_order_fills = self.partial_order_fills.iter();

		let Some(
			mut bid_ask_price_spread,
		) = iter_partial_order_fills
			.next()
			.map(|
				partial_order_fill,
			| {
				filled_volume = filled_volume.add_unchecked(
					partial_order_fill.directional_intent_volume.directionless_volume.as_zeroable(),
				);

				BidAskPriceSpread {
					ask_price: partial_order_fill.price,
					bid_price: partial_order_fill.price,
				}
			}) else {
			return None;
		};

		for partial_order_fill in iter_partial_order_fills {
			filled_volume = filled_volume.add_unchecked(
				partial_order_fill.directional_intent_volume.directionless_volume.as_zeroable(),
			);

			if partial_order_fill.price > bid_ask_price_spread.ask_price {
				bid_ask_price_spread.ask_price = partial_order_fill.price;
			}

			if partial_order_fill.price < bid_ask_price_spread.bid_price {
				bid_ask_price_spread.bid_price = partial_order_fill.price;
			}
		}

		if (
			filled_volume <
			self.desired_directional_intent_volume.directionless_volume.as_zeroable()
		) {
			return None;
		}

		Some(bid_ask_price_spread)
	}

	#[instrument(skip_all)]
	pub(crate) fn modify_volume(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		volume: &DirectionlessVolume<IS>,
	) {
		let Ok(
			desired_volume_delta,
		) = self.desired_directional_intent_volume.directionless_volume.sub_checked(
			*volume,
		) else {
			return;
		};

		let desired_volume_delta = desired_volume_delta.with_directional_intent(
			self.desired_directional_intent_volume.directional_intent,
		).as_zeroable();

		directional_exposure.change_armed_directional_exposure(&desired_volume_delta);

		self.desired_directional_intent_volume.directionless_volume = *volume;

		self.armed_directional_exposure = self.armed_directional_exposure.add_unchecked(
			desired_volume_delta,
		);

		#[cfg(feature = "log-trace-order-manager")]
		trace!(
			"Modifing working order volume. New desired: `{:?}` armed: `{:?}`",
			self.desired_directional_intent_volume,
			self.armed_directional_exposure
		);

		if let Some(
			directional_intent_volume,
		) = self.armed_directional_exposure.as_optional_nonzero_ref() {
			if (
				directional_intent_volume.directional_intent !=
				self.desired_directional_intent_volume.directional_intent
			) {
				unimplemented!()
			}
		} else {
			unimplemented!()
		}
	}
}

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> WorkingRemoteOrder<IS, CS> {
	pub(crate) fn new(
		local_order_id: LocalOrderId,
		remote_order_id: RemoteOrderId,
		dependencies: impl IntoIterator<Item = (
			RemoteTimingCondition,
			OrderAction<IS>,
		)>,
		desired_directional_intent_volume: DirectionalIntentVolume<IS>,
		booking_timestamp: Timestamp,
		rest_at: Option<AbsolutePrice<IS>>,
	) -> Self {
		Self {
			local_order_id,
			remote_order_id,
			dependencies: dependencies.into_iter().collect(),
			desired_directional_intent_volume,
			booking_timestamp,
			partial_order_fills: SmallVec::default(),
			armed_directional_exposure: desired_directional_intent_volume.as_zeroable(),
			effective_directional_exposure: Zeroable::ZERO,
			rest_at,
		}
	}
}

/// Unordered list of working remote orders. These include all unfilled resting orders, partially filled resting orders, and market orders that have not been matched. All orders in `working_remote_orders` have been recognized by the matching engine and the client has received that the matching engine knows these orders. Only when the client is aware that the order is 100% filled will it state transition to [`filled_remote_orders`].
pub struct WorkingRemoteOrders<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
>(SmallVec<[WorkingRemoteOrder<IS, CS>; core::direct_const_arg!(CS::WORKING_REMOTE)]>);

impl<
	IS: InstrumentSpec,
	CS: OrdersCapacitySpec,
> WorkingRemoteOrders<IS, CS> {
	#[instrument(skip_all)]
	pub(crate) fn submit<OB: OrdersBackend<IS>>(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		tick_timestamp: &TickTimestamp,
		pending_remote_order: PendingRemoteOrder<IS, CS>,
		remote_order_id: RemoteOrderId,
	) -> Result<(), WorkingRemoteOrderSubmitError>
	where
		IS: Send,
	{
		#[cfg(feature = "log-trace-order-manager")]
		trace!("Adding id: `{:?}`.", remote_order_id);

		let PendingRemoteOrder {
			submission_timestamp: _,
			local_order_id,
			dependencies,
			desired_directional_intent_volume,
			rest_at,
			cancel_on_process,
		} = pending_remote_order;

		let booking_timestamp = tick_timestamp.timestamp();

		let working_remote_order = WorkingRemoteOrder::new(
			local_order_id,
			remote_order_id,
			dependencies,
			desired_directional_intent_volume,
			booking_timestamp,
			rest_at,
		);

		directional_exposure.change_armed_directional_exposure(
			&working_remote_order.armed_directional_exposure,
		);

		self.0.push(working_remote_order);

		if cancel_on_process {
			let order_action = OrderAction::CancelRemote(
				CancelRemoteOrder::new(local_order_id),
			);

			deferred_order_actions.push(order_action);
		}

		Ok(())
	}

	pub fn get(
		&self,
		order_id: impl Into<OrderId>,
	) -> Result<&WorkingRemoteOrder<IS, CS>, OrderIdError> {
		match &order_id.into() {
			OrderId::Local(
				local_order_id,
			) => self.get_with_local(local_order_id),
			OrderId::Remote(
				remote_order_id,
			) => self.get_with_remote(remote_order_id),
		}
	}

	#[instrument(skip_all)]
	pub fn get_with_local(
		&self,
		local_order_id: &LocalOrderId,
	) -> Result<&WorkingRemoteOrder<IS, CS>, OrderIdError> {
		let Some(
			working_remote_order,
		) = self.0.iter().find(|
			working_remote_order,
		| &working_remote_order.local_order_id == local_order_id) else {
			return local_order_id.err_invalid();
		};

		Ok(working_remote_order)
	}

	#[instrument(skip_all)]
	pub fn get_with_remote(
		&self,
		remote_order_id: &RemoteOrderId,
	) -> Result<&WorkingRemoteOrder<IS, CS>, OrderIdError> {
		let Some(
			working_remote_order,
		) = self.0.iter().find(|
			working_remote_order,
		| &working_remote_order.remote_order_id == remote_order_id) else {
			return remote_order_id.err_invalid();
		};

		Ok(working_remote_order)
	}

	pub(crate) fn get_mut(
		&mut self,
		order_id: impl Into<OrderId>,
	) -> Result<&mut WorkingRemoteOrder<IS, CS>, OrderIdError> {
		match &order_id.into() {
			OrderId::Local(
				local_order_id,
			) => self.get_mut_with_local(local_order_id),
			OrderId::Remote(
				remote_order_id,
			) => self.get_mut_with_remote(remote_order_id),
		}
	}

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

		Ok(working_remote_order)
	}

	#[instrument(skip_all)]
	pub(crate) fn get_mut_with_remote(
		&mut self,
		remote_order_id: &RemoteOrderId,
	) -> Result<&mut WorkingRemoteOrder<IS, CS>, OrderIdError> {
		let Some(
			working_remote_order,
		) = self.0.iter_mut().find(|
			working_remote_order,
		| &working_remote_order.remote_order_id == remote_order_id) else {
			return remote_order_id.err_invalid();
		};

		Ok(working_remote_order)
	}

	#[instrument(skip_all)]
	pub(crate) fn remove(
		&mut self,
		directional_exposure: &mut DirectionalExposure<IS>,
		order_id: impl Into<OrderId>,
	) -> Result<WorkingRemoteOrder<IS, CS>, OrderIdError>
	where
		IS: Send,
	{
		let order_id = order_id.into();

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

		let idx = match order_id {
			OrderId::Local(
				local_order_id,
			) => {
				self.0.iter().position(|
					working_remote_order,
				| working_remote_order.local_order_id == local_order_id)
			},
			OrderId::Remote(
				remote_order_id,
			) => {
				self.0.iter().position(|
					working_remote_order,
				| working_remote_order.remote_order_id == remote_order_id)
			},
		};

		let Some(
			idx,
		) = idx else {
			return order_id.err_invalid();
		};

		let mut working_remote_order = self.0.swap_remove(idx);

		if let Some(
			armed_directional_exposure,
		) = working_remote_order.armed_directional_exposure.as_optional_nonzero_ref() {
			#[cfg(feature = "log-trace-directional-exposure")]
			trace!(
				"Reverting armed directional exposure `{:?}` during removing order from working state.",
				armed_directional_exposure
			);

			directional_exposure.change_armed_directional_exposure(
				&armed_directional_exposure.as_flipped().as_zeroable(),
			);
		}

		working_remote_order.armed_directional_exposure = Zeroable::ZERO;
		Ok(working_remote_order)
	}

	#[instrument(skip_all)]
	pub(crate) fn processed<OB: OrdersBackend<IS>>(
		&mut self,
		remote_order_id: &RemoteOrderId,
	) -> Result<(), OrderIdError>
	where
		IS: Send,
	{
		#[cfg(feature = "log-trace-order-manager")]
		trace!("Processing id: `{:?}`.", remote_order_id);

		let working_remote_order = self.get_mut_with_remote(remote_order_id)?;

		let Some(
			rest_at,
		) = &working_remote_order.rest_at else {
			return Ok(());
		};

		let bid_ask_price_spread = BidAskPriceSpread {
			ask_price: *rest_at,
			bid_price: *rest_at,
		};

		for (
			_,
			remote_order_action,
		) in working_remote_order.dependencies.iter_mut() {
			remote_order_action.parent_processed(&bid_ask_price_spread);
		}

		Ok(())
	}

	#[instrument(skip_all)]
	pub(crate) async fn partial_fill<OB: OrdersBackend<IS>>(
		&mut self,
		tick_timestamp: &TickTimestamp,
		completed_remote_orders: &mut CompletedRemoteOrders<IS, CS>,
		deferred_order_actions: &mut DeferredOrderActions<IS>,
		directional_exposure: &mut DirectionalExposure<IS>,
		partial_order_fill: PartialOrderFill<IS>,
		remote_order_id: &RemoteOrderId,
	) -> Result<(), OrderIdError>
	where
		IS: Send,
	{
		#[cfg(feature = "log-trace-order-manager")]
		trace!(
			"Partial filling `{partial_order_fill:?}` for id: `{:?}`.",
			remote_order_id,
		);

		let working_remote_order = self.get_mut_with_remote(remote_order_id)?;

		working_remote_order.armed_directional_exposure = working_remote_order.armed_directional_exposure.sub_unchecked(
			partial_order_fill.directional_intent_volume.as_zeroable(),
		);

		directional_exposure.change_armed_directional_exposure(
			&partial_order_fill
				.directional_intent_volume
				.as_flipped().as_zeroable(),
		);

		working_remote_order.effective_directional_exposure = working_remote_order.effective_directional_exposure.add_unchecked(
			partial_order_fill.directional_intent_volume.as_zeroable(),
		);

		directional_exposure.apply_partial_order_fill(
			&partial_order_fill,
		).await;

		working_remote_order.partial_order_fills.push(partial_order_fill);

		for (
			remote_timing_condition,
			remote_order_action,
		) in working_remote_order.dependencies.iter() {
			if remote_timing_condition != &RemoteTimingCondition::PartialFill {
				continue;
			}

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

		let Some(
			bid_ask_price_spread,
		) = working_remote_order.get_fully_filled() else {
			return Ok(());
		};

		#[cfg(feature = "log-trace-order-manager")]
		trace!("Partial fill fully filled working order with spread: `{bid_ask_price_spread:?}`.");

		let mut working_remote_order = self.remove(
			directional_exposure,
			remote_order_id,
		)?;

		if working_remote_order.rest_at.is_none() {
			for (
				_,
				remote_order_action,
			) in working_remote_order.dependencies.iter_mut() {
				remote_order_action.parent_processed(&bid_ask_price_spread);
			}
		}

		for (
			remote_timing_condition,
			remote_order_action,
		) in working_remote_order.dependencies.iter() {
			if remote_timing_condition != &RemoteTimingCondition::FullyFilled {
				continue;
			}

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

		completed_remote_orders.submit::<OB>(tick_timestamp, working_remote_order);
		Ok(())
	}

	#[instrument(skip_all)]
	pub(crate) async fn modify_volume<OB: OrdersBackend<IS>>(
		&mut self,
		orders_backend: &mut OB,
		directional_exposure: &mut DirectionalExposure<IS>,
		local_order_id: &LocalOrderId,
		volume: &DirectionlessVolume<IS>,
	) -> Result<(), OrderIdError>
	where
		IS: Send,
	{
		let working_remote_order = self.get_mut_with_local(local_order_id)?;

		working_remote_order.modify_volume(directional_exposure, volume);

		orders_backend.modify_order_volume(
			&working_remote_order.remote_order_id,
			volume,
		).await;

		Ok(())
	}
}

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