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