use crate::order::{
Order, TimeInForce, UnindexedOrder,
id::ClientOrderId,
state::{ActiveOrderState, Open},
};
use chrono::{DateTime, Utc};
use fnv::FnvHashMap;
use rust_decimal::Decimal;
use rustrade_instrument::{
Side, asset::name::AssetNameExchange, exchange::ExchangeId,
instrument::name::InstrumentNameExchange,
};
use std::collections::{BTreeMap, BTreeSet, hash_map::Entry};
use thiserror::Error;
pub type OpenOrder = Order<ExchangeId, InstrumentNameExchange, Open>;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Reservation {
pub asset: AssetNameExchange,
pub amount: Decimal,
}
#[derive(Debug, Clone)]
pub struct RestingOrder {
pub order: OpenOrder,
pub reservation: Option<Reservation>,
}
#[derive(Debug, Clone, Error)]
#[error("an order is already resting under {}", .0.order.key.cid)]
pub struct AlreadyResting(pub Box<RestingOrder>);
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
struct QueuePosition {
price_rank: Decimal,
time_exchange: DateTime<Utc>,
seq: u64,
}
impl QueuePosition {
fn new(side: Side, price: Decimal, time_exchange: DateTime<Utc>, seq: u64) -> Self {
Self {
price_rank: match side {
Side::Buy => -price,
Side::Sell => price,
},
time_exchange,
seq,
}
}
}
#[derive(Debug, Default)]
pub struct OpenOrders {
by_id: FnvHashMap<ClientOrderId, RestingOrder>,
queues: FnvHashMap<(InstrumentNameExchange, Side), BTreeMap<QueuePosition, ClientOrderId>>,
expiries: BTreeSet<(DateTime<Utc>, ClientOrderId)>,
seq: u64,
}
fn deadline_of(order: &OpenOrder) -> Option<DateTime<Utc>> {
match order.time_in_force {
TimeInForce::GoodTillDate { expiry } => Some(expiry),
_ => None,
}
}
impl OpenOrders {
pub fn insert(
&mut self,
order: OpenOrder,
reservation: Option<Reservation>,
) -> Result<(), AlreadyResting> {
let cid = order.key.cid.clone();
let vacant = match self.by_id.entry(cid.clone()) {
Entry::Occupied(_) => {
return Err(AlreadyResting(Box::new(RestingOrder {
order,
reservation,
})));
}
Entry::Vacant(vacant) => vacant,
};
if let Some(expiry) = deadline_of(&order) {
self.expiries.insert((expiry, cid.clone()));
}
if let Some(price) = order.price {
self.seq += 1;
self.queues
.entry((order.key.instrument.clone(), order.side))
.or_default()
.insert(
QueuePosition::new(order.side, price, order.state.time_exchange, self.seq),
cid,
);
}
vacant.insert(RestingOrder { order, reservation });
Ok(())
}
pub fn contains(&self, cid: &ClientOrderId) -> bool {
self.by_id.contains_key(cid)
}
pub fn remove(&mut self, cid: &ClientOrderId) -> Option<RestingOrder> {
let resting = self.by_id.remove(cid)?;
self.unindex(&resting.order);
Some(resting)
}
pub fn expired_as_of(&self, now: DateTime<Utc>) -> Vec<ClientOrderId> {
self.expiries
.iter()
.take_while(|(expiry, _)| *expiry <= now)
.map(|(_, cid)| cid.clone())
.collect()
}
fn unindex(&mut self, order: &OpenOrder) {
self.dequeue(order);
if let Some(expiry) = deadline_of(order) {
self.expiries.remove(&(expiry, order.key.cid.clone()));
}
}
pub fn get(&self, cid: &ClientOrderId) -> Option<&OpenOrder> {
self.by_id.get(cid).map(|resting| &resting.order)
}
pub fn len(&self) -> usize {
self.by_id.len()
}
pub fn is_empty(&self) -> bool {
self.by_id.is_empty()
}
pub fn iter(&self) -> impl Iterator<Item = &OpenOrder> + '_ {
self.by_id.values().map(|resting| &resting.order)
}
pub fn resting(
&self,
instrument: &InstrumentNameExchange,
side: Side,
) -> impl Iterator<Item = &OpenOrder> + '_ {
self.queues
.get(&(instrument.clone(), side))
.into_iter()
.flat_map(|queue| queue.values())
.filter_map(|cid| self.by_id.get(cid))
.map(|resting| &resting.order)
}
fn dequeue(&mut self, order: &OpenOrder) {
let Some(price) = order.price else {
return;
};
let key = (order.key.instrument.clone(), order.side);
let Some(queue) = self.queues.get_mut(&key) else {
return;
};
let position = queue
.range(QueuePosition::new(order.side, price, order.state.time_exchange, 0)..)
.find(|(_, queued)| *queued == &order.key.cid)
.map(|(position, _)| *position);
debug_assert!(
position.is_some(),
"OpenOrders lost the queue position of {} ({} {price} at {}): its ranking fields were \
mutated while it was on the book",
order.key.cid,
order.side,
order.state.time_exchange
);
if let Some(position) = position {
queue.remove(&position);
}
if queue.is_empty() {
self.queues.remove(&key);
}
}
}
pub(super) fn as_open(order: UnindexedOrder) -> Option<OpenOrder> {
match order.state {
crate::order::state::OrderState::Active(ActiveOrderState::Open(open)) => Some(Order {
key: order.key,
side: order.side,
price: order.price,
quantity: order.quantity,
kind: order.kind,
time_in_force: order.time_in_force,
state: open,
}),
_ => None,
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)] mod tests {
use super::*;
use crate::order::{
OrderKey, OrderKind, TimeInForce,
id::{OrderId, StrategyId, VenueOrderId},
};
use rust_decimal_macros::dec;
fn at(millis: i64) -> DateTime<Utc> {
DateTime::from_timestamp_millis(millis).unwrap()
}
fn instrument() -> InstrumentNameExchange {
InstrumentNameExchange::new("btc_usdt")
}
fn gtd(cid: &str, expiry_millis: i64) -> OpenOrder {
let mut order = order(cid, Side::Buy, Some(dec!(100)), 1_000);
order.time_in_force = TimeInForce::GoodTillDate {
expiry: at(expiry_millis),
};
order
}
fn order(cid: &str, side: Side, price: Option<Decimal>, millis: i64) -> OpenOrder {
Order {
key: OrderKey {
exchange: ExchangeId::BinanceSpot,
instrument: instrument(),
strategy: StrategyId::new("test"),
cid: ClientOrderId::new(cid),
},
side,
price,
quantity: dec!(1),
kind: OrderKind::Limit,
time_in_force: TimeInForce::GoodUntilCancelled { post_only: false },
state: Open {
id: VenueOrderId::Assigned(OrderId::new(cid)),
time_exchange: at(millis),
filled_quantity: Decimal::ZERO,
},
}
}
fn book(orders: impl IntoIterator<Item = OpenOrder>) -> OpenOrders {
let mut open = OpenOrders::default();
for order in orders {
open.insert(order, None).unwrap();
}
open
}
fn cids(orders: impl Iterator<Item = &'static OpenOrder>) -> Vec<String> {
orders.map(|order| order.key.cid.0.to_string()).collect()
}
fn resting_cids(open: &OpenOrders, side: Side) -> Vec<String> {
open.resting(&instrument(), side)
.map(|order| order.key.cid.0.to_string())
.collect()
}
#[test]
fn buys_rest_highest_price_then_earliest_first() {
let open = book([
order("low", Side::Buy, Some(dec!(99)), 1_000),
order("high_late", Side::Buy, Some(dec!(101)), 3_000),
order("high_early", Side::Buy, Some(dec!(101)), 2_000),
order("mid", Side::Buy, Some(dec!(100)), 500),
]);
assert_eq!(
resting_cids(&open, Side::Buy),
["high_early", "high_late", "mid", "low"],
"price first, then arrival — not insertion order, and not the map's order"
);
}
#[test]
fn sells_rest_lowest_price_then_earliest_first() {
let open = book([
order("high", Side::Sell, Some(dec!(101)), 1_000),
order("low_late", Side::Sell, Some(dec!(99)), 3_000),
order("low_early", Side::Sell, Some(dec!(99)), 2_000),
]);
assert_eq!(
resting_cids(&open, Side::Sell),
["low_early", "low_late", "high"]
);
}
#[test]
fn orders_at_the_same_price_and_instant_both_rest() {
let open = book([
order("first", Side::Buy, Some(dec!(100)), 1_000),
order("second", Side::Buy, Some(dec!(100)), 1_000),
]);
assert_eq!(open.len(), 2);
assert_eq!(
resting_cids(&open, Side::Buy),
["first", "second"],
"the tie is broken by insertion, which is deterministic"
);
}
#[test]
fn the_two_sides_are_separate_queues() {
let open = book([
order("bid", Side::Buy, Some(dec!(99)), 1_000),
order("ask", Side::Sell, Some(dec!(101)), 1_000),
]);
assert_eq!(resting_cids(&open, Side::Buy), ["bid"]);
assert_eq!(resting_cids(&open, Side::Sell), ["ask"]);
}
#[test]
fn a_removed_order_leaves_the_queue() {
let mut open = book([
order("a", Side::Buy, Some(dec!(100)), 1_000),
order("b", Side::Buy, Some(dec!(99)), 1_000),
]);
let removed = open.remove(&ClientOrderId::new("a")).expect("a is open");
assert_eq!(removed.order.key.cid, ClientOrderId::new("a"));
assert_eq!(resting_cids(&open, Side::Buy), ["b"]);
assert_eq!(open.len(), 1);
assert!(open.get(&ClientOrderId::new("a")).is_none());
assert!(open.remove(&ClientOrderId::new("a")).is_none());
}
#[test]
fn emptying_the_book_empties_the_queues() {
let mut open = book([
order("a", Side::Buy, Some(dec!(100)), 1_000),
order("b", Side::Buy, Some(dec!(99)), 2_000),
]);
open.remove(&ClientOrderId::new("a"));
open.remove(&ClientOrderId::new("b"));
assert!(open.is_empty());
assert!(resting_cids(&open, Side::Buy).is_empty());
open.insert(order("c", Side::Buy, Some(dec!(98)), 3_000), None)
.unwrap();
assert_eq!(resting_cids(&open, Side::Buy), ["c"]);
}
#[test]
fn a_second_order_under_a_resting_id_is_refused_and_the_first_keeps_its_place() {
let mut open = book([
order("taken", Side::Buy, Some(dec!(100)), 1_000),
order("other", Side::Buy, Some(dec!(99)), 1_000),
]);
let held = Reservation {
asset: AssetNameExchange::new("usdt"),
amount: dec!(98),
};
let Err(AlreadyResting(refused)) = open.insert(
order("taken", Side::Buy, Some(dec!(98)), 2_000),
Some(held.clone()),
) else {
panic!("an id already resting is refused");
};
assert_eq!(
refused.order.price,
Some(dec!(98)),
"the refused order comes back"
);
assert_eq!(
refused.reservation,
Some(held),
"and so does what the caller may already hold against it"
);
assert_eq!(open.len(), 2, "nothing was added");
assert_eq!(
open.get(&ClientOrderId::new("taken"))
.map(|order| order.price),
Some(Some(dec!(100))),
"the resting order is the one still held"
);
assert_eq!(
resting_cids(&open, Side::Buy),
["taken", "other"],
"at its own price and place, appearing exactly once"
);
}
#[test]
fn an_order_without_a_price_is_held_but_never_rests() {
let open = book([
order("priceless", Side::Buy, None, 1_000),
order("priced", Side::Buy, Some(dec!(100)), 1_000),
]);
assert_eq!(open.len(), 2, "a snapshot must still report it");
assert!(open.get(&ClientOrderId::new("priceless")).is_some());
assert_eq!(
resting_cids(&open, Side::Buy),
["priced"],
"an order with no price to wait at has nothing to wait for"
);
}
#[test]
fn only_orders_past_their_deadline_are_swept() {
let open = book([
order("no_deadline", Side::Buy, Some(dec!(100)), 1_000),
gtd("due", 5_000),
gtd("later", 9_000),
]);
assert!(
open.expired_as_of(at(4_999)).is_empty(),
"nothing is due before the earliest deadline"
);
assert_eq!(
open.expired_as_of(at(5_000)),
[ClientOrderId::new("due")],
"a deadline is reached at its instant, not after it"
);
assert_eq!(
open.expired_as_of(at(9_999)),
[ClientOrderId::new("due"), ClientOrderId::new("later")],
"earliest deadline first, and an order without one is never swept"
);
}
#[test]
fn orders_sharing_a_deadline_are_both_swept_in_id_order() {
let open = book([gtd("b", 5_000), gtd("a", 5_000)]);
assert_eq!(
open.expired_as_of(at(5_000)),
[ClientOrderId::new("a"), ClientOrderId::new("b")],
"the tie is broken by id, which does not depend on map iteration"
);
}
#[test]
fn a_removed_order_is_no_longer_swept() {
let mut open = book([gtd("gone", 5_000), gtd("stays", 5_000)]);
open.remove(&ClientOrderId::new("gone"));
assert_eq!(
open.expired_as_of(at(5_000)),
[ClientOrderId::new("stays")],
"a deadline that outlived its order would retire an order twice"
);
}
#[test]
fn a_refused_order_leaves_the_resting_ones_deadline() {
let mut open = book([gtd("taken", 5_000)]);
assert!(open.insert(gtd("taken", 9_000), None).is_err());
assert_eq!(
open.expired_as_of(at(9_000)),
[ClientOrderId::new("taken")],
"the resting order's deadline, once: the refused one's never reached the index"
);
assert_eq!(open.expired_as_of(at(5_000)), [ClientOrderId::new("taken")]);
}
#[test]
fn an_id_is_free_again_once_its_order_leaves_the_book() {
let mut open = book([gtd("reused", 5_000)]);
open.remove(&ClientOrderId::new("reused")).unwrap();
open.insert(order("reused", Side::Buy, Some(dec!(100)), 6_000), None)
.unwrap();
assert!(
open.expired_as_of(at(9_000)).is_empty(),
"the first order's deadline left with it"
);
assert_eq!(resting_cids(&open, Side::Buy), ["reused"]);
}
#[test]
fn iter_reports_every_open_order() {
let open = book([
order("a", Side::Buy, Some(dec!(100)), 1_000),
order("b", Side::Sell, Some(dec!(101)), 1_000),
order("c", Side::Buy, None, 1_000),
]);
let mut all = cids(Vec::leak(open.iter().cloned().collect::<Vec<_>>()).iter());
all.sort();
assert_eq!(all, ["a", "b", "c"]);
}
}