use crate::margin::MarginEffect;
use derive_more::Constructor;
use rust_decimal::Decimal;
use rustrade_instrument::{
asset::{AssetIndex, name::AssetNameExchange},
exchange::ExchangeId,
index::error::IndexError,
instrument::{InstrumentIndex, name::InstrumentNameExchange},
};
use rustrade_integration::error::SocketError;
use serde::{Deserialize, Serialize};
use std::{fmt, time::Duration};
use thiserror::Error;
pub type UnindexedClientError = ClientError<AssetNameExchange, InstrumentNameExchange>;
pub type UnindexedApiError = ApiError<AssetNameExchange, InstrumentNameExchange>;
pub type UnindexedOrderError = OrderError<AssetNameExchange, InstrumentNameExchange>;
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum ClientError<AssetKey = AssetIndex, InstrumentKey = InstrumentIndex> {
#[error("Connectivity: {0}")]
Connectivity(#[from] ConnectivityError),
#[error("API: {0}")]
Api(#[from] ApiError<AssetKey, InstrumentKey>),
#[error("task failed: {0}")]
TaskFailed(String),
#[error("internal error: {0}")]
Internal(String),
#[error("trade read stopped after {fills_read} fills without advancing past its start")]
Truncated {
fills_read: usize,
},
#[error("open orders snapshot truncated at {limit} results — data may be incomplete")]
TruncatedSnapshot {
limit: usize,
},
}
impl<AssetKey, InstrumentKey> ClientError<AssetKey, InstrumentKey> {
pub fn is_transient(&self) -> bool {
match self {
Self::Connectivity(e) => e.is_transient(),
Self::Api(ApiError::RateLimit { .. }) => true,
Self::Api(_) => false,
Self::TaskFailed(_) => false,
Self::Internal(_) => false,
Self::Truncated { .. } => false,
Self::TruncatedSnapshot { .. } => false,
}
}
pub fn retry_after(&self) -> Option<Duration> {
match self {
Self::Api(error) => error.retry_after(),
_ => None,
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum ConnectivityError {
#[error("Exchange offline: {0}")]
ExchangeOffline(ExchangeId),
#[error("request timed out")]
Timeout,
#[error("{0}")]
Socket(String),
}
impl From<SocketError> for ConnectivityError {
fn from(value: SocketError) -> Self {
Self::Socket(value.to_string())
}
}
impl ConnectivityError {
pub fn is_transient(&self) -> bool {
match self {
Self::ExchangeOffline(_) => true,
Self::Timeout => true,
Self::Socket(_) => true,
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum ApiError<AssetKey = AssetIndex, InstrumentKey = InstrumentIndex> {
#[error("asset {0} invalid: {1}")]
AssetInvalid(AssetKey, String),
#[error("instrument {0} invalid: {1}")]
InstrumentInvalid(InstrumentKey, String),
#[error("rate limit exceeded{}", RetryAfter(*retry_after))]
#[non_exhaustive]
RateLimit { retry_after: Option<Duration> },
#[error("venue capacity exceeded: {0}")]
CapacityExceeded(String),
#[error("authentication failed: {0}")]
Unauthenticated(String),
#[error(
"balance insufficient{asset}: {1}",
asset = .0.as_ref().map(|asset| format!(" for asset {asset}")).unwrap_or_default()
)]
BalanceInsufficient(Option<AssetKey>, String),
#[error("order rejected: {0}")]
OrderRejected(String),
#[error("order already cancelled")]
OrderAlreadyCancelled,
#[error("order already fully filled")]
OrderAlreadyFullyFilled,
#[error("order already expired")]
OrderAlreadyExpired,
#[error("client order id in use: {0}")]
DuplicateClientOrderId(String),
#[error("request rejected: {0}")]
RequestRejected(String),
#[error("borrow rejected: {0}")]
BorrowRejected(BorrowReject),
}
impl<AssetKey, InstrumentKey> ApiError<AssetKey, InstrumentKey> {
pub fn rate_limited(retry_after: Option<Duration>) -> Self {
Self::RateLimit { retry_after }
}
pub fn retry_after(&self) -> Option<Duration> {
match self {
Self::RateLimit { retry_after } => *retry_after,
_ => None,
}
}
}
struct RetryAfter(Option<Duration>);
impl fmt::Display for RetryAfter {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self.0 {
Some(retry_after) => write!(f, "; retry after {retry_after:?}"),
None => Ok(()),
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub struct BorrowReject {
pub reason: BorrowRejectReason,
pub venue_code: Option<String>,
pub message: String,
pub available: Option<Decimal>,
}
impl BorrowReject {
pub fn new(reason: BorrowRejectReason, venue_code: Option<String>, message: String) -> Self {
Self {
reason,
venue_code,
message,
available: None,
}
}
#[must_use]
pub fn with_available(mut self, available: Decimal) -> Self {
self.available = Some(available);
self
}
}
impl fmt::Display for BorrowReject {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let Self {
reason,
venue_code,
message,
available,
} = self;
write!(f, "{reason}")?;
if let Some(code) = venue_code {
write!(f, " (code {code})")?;
}
write!(f, ": {message}")?;
if let Some(available) = available {
write!(f, " ({available} available)")?;
}
Ok(())
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub enum BorrowRejectReason {
NotShortable,
BorrowDisabled,
InventoryUnavailable,
BorrowLimitExceeded,
ShortSaleRestricted,
Other,
}
impl fmt::Display for BorrowRejectReason {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::NotShortable => "not shortable",
Self::BorrowDisabled => "borrow disabled",
Self::InventoryUnavailable => "inventory unavailable",
Self::BorrowLimitExceeded => "borrow limit exceeded",
Self::ShortSaleRestricted => "short sale restricted",
Self::Other => "other",
})
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum OrderError<AssetKey = AssetIndex, InstrumentKey = InstrumentIndex> {
#[error("connectivity: {0}")]
Connectivity(#[from] ConnectivityError),
#[error("order rejected: {0}")]
Rejected(#[from] ApiError<AssetKey, InstrumentKey>),
#[error("unsupported order type: {0}")]
UnsupportedOrderType(String),
#[error("invalid precision: {0}")]
InvalidPrecision(PrecisionViolation),
#[error("out of bounds: {0}")]
OutOfBounds(BoundsViolation),
#[error("unsupported margin effect: {0:?}")]
UnsupportedMarginEffect(MarginEffect),
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub struct PrecisionViolation {
pub field: OrderField,
pub value: Decimal,
pub limit: PrecisionLimit,
}
impl PrecisionViolation {
pub fn new(field: OrderField, value: Decimal, limit: PrecisionLimit) -> Self {
Self {
field,
value,
limit,
}
}
}
impl fmt::Display for PrecisionViolation {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let Self {
field,
value,
limit,
} = self;
write!(f, "{field} {value} ")?;
match limit {
PrecisionLimit::DecimalPlaces { max } => {
write!(f, "has more than {max} decimal places")
}
PrecisionLimit::SignificantFigures { max } => {
write!(
f,
"is not an integer and has more than {max} significant figures"
)
}
PrecisionLimit::NotRepresentable => {
write!(f, "cannot be sent exactly in the venue's number format")
}
PrecisionLimit::NotPositive => write!(f, "is not positive"),
PrecisionLimit::NotOnGrid { step } => {
write!(f, "is not a multiple of {step}")
}
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub struct BoundsViolation {
pub field: OrderField,
pub bound: Bound,
pub limit: Decimal,
pub value: Decimal,
}
impl BoundsViolation {
pub fn new(field: OrderField, bound: Bound, limit: Decimal, value: Decimal) -> Self {
Self {
field,
bound,
limit,
value,
}
}
}
impl fmt::Display for BoundsViolation {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let Self {
field,
bound,
limit,
value,
} = self;
let (relation, name) = match bound {
Bound::Min => ("below", "minimum"),
Bound::Max => ("above", "maximum"),
};
write!(f, "{field} {value} is {relation} the {name} {limit}")
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub enum Bound {
Min,
Max,
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub enum OrderField {
Quantity,
Price,
TriggerPrice,
TrailingOffset,
Notional,
}
impl fmt::Display for OrderField {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::Quantity => "quantity",
Self::Price => "price",
Self::TriggerPrice => "trigger price",
Self::TrailingOffset => "trailing offset",
Self::Notional => "notional",
})
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Copy, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize)]
pub enum PrecisionLimit {
DecimalPlaces {
max: u32,
},
SignificantFigures {
max: u32,
},
NotRepresentable,
NotPositive,
NotOnGrid {
step: Decimal,
},
}
impl<AssetKey, InstrumentKey> OrderError<AssetKey, InstrumentKey> {
pub fn is_transient(&self) -> bool {
match self {
Self::Connectivity(e) => e.is_transient(),
Self::Rejected(ApiError::RateLimit { .. }) => true,
Self::Rejected(_) => false,
Self::UnsupportedOrderType(_)
| Self::InvalidPrecision(_)
| Self::OutOfBounds(_)
| Self::UnsupportedMarginEffect(_) => false,
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum StreamTerminationReason {
#[error("reconnect budget exhausted after {attempts} attempts: {last_error}")]
ReconnectBudgetExhausted {
attempts: u32,
last_error: String,
},
#[error("unrecoverable stream error: {0}")]
Error(String),
#[error("rate-limited by the venue; retry after {retry_after:?}")]
RateLimited {
retry_after: Duration,
},
}
pub type UnindexedAccountStreamInitError =
AccountStreamInitError<AssetNameExchange, InstrumentNameExchange>;
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum AccountStreamInitError<AssetKey = AssetIndex, InstrumentKey = InstrumentIndex> {
#[error("{0}")]
Client(#[from] ClientError<AssetKey, InstrumentKey>),
#[error("IndexError: {0}")]
Index(#[from] IndexError),
}
impl<AssetKey, InstrumentKey> AccountStreamInitError<AssetKey, InstrumentKey> {
pub fn retry_after(&self) -> Option<Duration> {
match self {
Self::Client(error) => error.retry_after(),
Self::Index(_) => None,
}
}
}
#[non_exhaustive]
#[derive(Debug, Clone, Eq, PartialEq, Hash, Deserialize, Serialize, Constructor)]
pub struct AccountReinitFailure<AssetKey = AssetIndex, InstrumentKey = InstrumentIndex> {
pub attempt: u32,
pub error: AccountStreamInitError<AssetKey, InstrumentKey>,
}
#[derive(Debug, Clone, Eq, PartialEq, Ord, PartialOrd, Hash, Deserialize, Serialize, Error)]
pub enum KeyError {
#[error("ExchangeId: {0}")]
ExchangeId(String),
#[error("AssetKey: {0}")]
AssetKey(String),
#[error("InstrumentKey: {0}")]
InstrumentKey(String),
}
#[cfg(test)]
#[allow(clippy::unwrap_used)] mod tests {
use super::*;
use rust_decimal_macros::dec;
#[test]
fn a_rate_limit_carries_and_shows_the_venues_wait() {
let hinted: UnindexedApiError = ApiError::rate_limited(Some(Duration::from_secs(7)));
assert_eq!(hinted.retry_after(), Some(Duration::from_secs(7)));
assert_eq!(hinted.to_string(), "rate limit exceeded; retry after 7s");
assert!(UnindexedClientError::Api(hinted.clone()).is_transient());
assert!(UnindexedOrderError::Rejected(hinted).is_transient());
let unhinted: UnindexedApiError = ApiError::rate_limited(None);
assert_eq!(unhinted.retry_after(), None);
assert_eq!(unhinted.to_string(), "rate limit exceeded");
let other: UnindexedApiError = ApiError::Unauthenticated("bad key".to_string());
assert_eq!(other.retry_after(), None);
}
#[test]
fn a_client_or_init_error_reads_the_wait_of_its_rate_limit() {
let wait = Some(Duration::from_secs(7));
let hinted = UnindexedClientError::Api(ApiError::rate_limited(wait));
assert_eq!(hinted.retry_after(), wait);
assert_eq!(
UnindexedAccountStreamInitError::Client(hinted).retry_after(),
wait
);
let unhinted = UnindexedClientError::Api(ApiError::rate_limited(None));
assert_eq!(unhinted.retry_after(), None);
let socket = UnindexedClientError::Connectivity(ConnectivityError::Timeout);
assert_eq!(socket.retry_after(), None);
assert_eq!(
UnindexedAccountStreamInitError::Client(socket).retry_after(),
None
);
}
#[test]
fn a_rate_limited_termination_shows_and_round_trips_its_wait() {
let reason = StreamTerminationReason::RateLimited {
retry_after: Duration::from_secs(120),
};
assert_eq!(
reason.to_string(),
"rate-limited by the venue; retry after 120s"
);
let json = serde_json::to_string(&reason).unwrap();
assert_eq!(
serde_json::from_str::<StreamTerminationReason>(&json).unwrap(),
reason
);
}
#[test]
fn a_rate_limit_round_trips_through_serde() {
for retry_after in [None, Some(Duration::from_millis(7_250))] {
let error: UnindexedApiError = ApiError::rate_limited(retry_after);
let json = serde_json::to_string(&error).unwrap();
assert_eq!(
serde_json::from_str::<UnindexedApiError>(&json).unwrap(),
error,
"{json}"
);
}
}
#[test]
fn test_connectivity_error_is_transient() {
assert!(ConnectivityError::Timeout.is_transient());
assert!(ConnectivityError::Socket("connection refused".into()).is_transient());
assert!(ConnectivityError::ExchangeOffline(ExchangeId::BinanceSpot).is_transient());
}
#[test]
fn test_client_error_is_transient_connectivity() {
let err: ClientError = ClientError::Connectivity(ConnectivityError::Timeout);
assert!(err.is_transient());
let err: ClientError = ClientError::Connectivity(ConnectivityError::Socket("err".into()));
assert!(err.is_transient());
}
#[test]
fn test_client_error_is_transient_rate_limit() {
let err: ClientError = ClientError::Api(ApiError::rate_limited(None));
assert!(err.is_transient());
}
#[test]
fn test_client_error_not_transient_api_errors() {
let err: ClientError =
ClientError::Api(ApiError::AssetInvalid(AssetIndex(0), "bad".into()));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::BalanceInsufficient(
Some(AssetIndex(0)),
"low".into(),
));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::InstrumentInvalid(
InstrumentIndex(0),
"bad".into(),
));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::OrderRejected("rejected".into()));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::OrderAlreadyCancelled);
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::OrderAlreadyFullyFilled);
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError =
ClientError::Api(ApiError::Unauthenticated("invalid signature".into()));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError =
ClientError::Api(ApiError::RequestRejected("-1127 More than 24 hours".into()));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::Api(ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::InventoryUnavailable,
Some("-3045".into()),
"-3045 The system doesn't have enough asset now.".into(),
)));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
}
#[test]
fn borrow_rejected_shows_its_reason_and_code() {
let coded: UnindexedApiError = ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::BorrowDisabled,
Some("40310000".into()),
"account is not allowed to short".into(),
));
assert_eq!(
coded.to_string(),
"borrow rejected: borrow disabled (code 40310000): account is not allowed to short"
);
let uncoded: UnindexedApiError = ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::Other,
None,
"no borrow".into(),
));
assert_eq!(uncoded.to_string(), "borrow rejected: other: no borrow");
let short: UnindexedApiError = ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::ShortSaleRestricted,
None,
"priced at the bid".into(),
));
assert_eq!(
short.to_string(),
"borrow rejected: short sale restricted: priced at the bid"
);
}
#[test]
fn borrow_rejected_shows_what_is_available_only_when_known() {
let reject = BorrowReject::new(
BorrowRejectReason::InventoryUnavailable,
None,
"too little to lend".into(),
);
assert_eq!(reject.available, None);
let reject = reject.with_available(dec!(2.5));
assert_eq!(reject.available, Some(dec!(2.5)));
assert_eq!(
reject.to_string(),
"inventory unavailable: too little to lend (2.5 available)"
);
}
#[test]
fn balance_insufficient_names_its_asset_only_when_known() {
let named: UnindexedApiError =
ApiError::BalanceInsufficient(Some(AssetNameExchange::new("BTC")), "low".to_string());
assert_eq!(named.to_string(), "balance insufficient for asset BTC: low");
let unnamed: UnindexedApiError = ApiError::BalanceInsufficient(None, "low".to_string());
assert_eq!(unnamed.to_string(), "balance insufficient: low");
}
#[test]
fn test_client_error_not_transient_task_failed() {
let err: ClientError = ClientError::TaskFailed("task panicked".into());
assert!(!err.is_transient());
}
#[test]
fn test_client_error_not_transient_internal() {
let err: ClientError = ClientError::Internal("unknown error".into());
assert!(!err.is_transient());
}
#[test]
fn test_client_error_not_transient_truncated() {
let err: ClientError = ClientError::Truncated { fills_read: 5_000 };
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: ClientError = ClientError::TruncatedSnapshot { limit: 500 };
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
}
#[test]
fn test_client_error_is_transient_exchange_offline() {
let err: ClientError =
ClientError::Connectivity(ConnectivityError::ExchangeOffline(ExchangeId::BinanceSpot));
assert!(err.is_transient(), "expected transient for {:?}", err);
}
#[test]
fn test_order_error_is_transient_connectivity() {
let err: UnindexedOrderError = OrderError::Connectivity(ConnectivityError::Timeout);
assert!(err.is_transient(), "expected transient for {:?}", err);
let err: UnindexedOrderError =
OrderError::Connectivity(ConnectivityError::Socket("connection reset".into()));
assert!(err.is_transient(), "expected transient for {:?}", err);
let err: UnindexedOrderError =
OrderError::Connectivity(ConnectivityError::ExchangeOffline(ExchangeId::BinanceSpot));
assert!(err.is_transient(), "expected transient for {:?}", err);
}
#[test]
fn test_order_error_is_transient_rate_limit() {
let err: UnindexedOrderError = OrderError::Rejected(ApiError::rate_limited(None));
assert!(err.is_transient(), "expected transient for {:?}", err);
}
#[test]
fn test_order_error_not_transient_api_errors() {
let err: UnindexedOrderError =
OrderError::Rejected(ApiError::OrderRejected("price out of range".into()));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: UnindexedOrderError = OrderError::Rejected(ApiError::OrderAlreadyCancelled);
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: UnindexedOrderError =
OrderError::Rejected(ApiError::BorrowRejected(BorrowReject::new(
BorrowRejectReason::NotShortable,
None,
"asset XYZ cannot be sold short".into(),
)));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
let err: UnindexedOrderError = OrderError::Rejected(ApiError::BalanceInsufficient(
Some(AssetNameExchange::from("BTC")),
"insufficient".into(),
));
assert!(!err.is_transient(), "expected non-transient for {:?}", err);
}
}