1use crate::{
2 AccountEvent, AccountEventKind, AccountSnapshot, InstrumentAccountSnapshot,
3 InstrumentBalanceSnapshot, InstrumentBalanceUpdate, IsolatedInstrumentState,
4 UnindexedAccountEvent, UnindexedAccountSnapshot,
5 balance::{AssetBalance, AssetBalanceUpdate},
6 cash_flow::CashFlow,
7 error::{
8 AccountReinitFailure, AccountStreamInitError, ApiError, ClientError, KeyError, OrderError,
9 UnindexedAccountStreamInitError, UnindexedApiError, UnindexedClientError,
10 UnindexedOrderError,
11 },
12 fill_recovery::{FillRecoveryGap, FillRecoveryScope},
13 map::ExecutionInstrumentMap,
14 margin_debt::MarginDebtUpdate,
15 notice::AccountNotice,
16 order::{
17 Order, OrderEvent, OrderKey, OrderSnapshot, UnindexedOrderKey, UnindexedOrderSnapshot,
18 request::OrderResponseCancel,
19 state::{InactiveOrderState, OrderState, UnindexedOrderState},
20 },
21 trade::{AssetFees, Trade, TradeAmendment, TradeAmendmentKind},
22};
23use derive_more::Constructor;
24use rust_decimal::Decimal;
25use rustrade_instrument::{
26 asset::{AssetIndex, name::AssetNameExchange},
27 exchange::{ExchangeId, ExchangeIndex},
28 index::error::IndexError,
29 instrument::{InstrumentIndex, name::InstrumentNameExchange},
30};
31use rustrade_integration::collection::snapshot::Snapshot;
32use std::sync::Arc;
33use tracing::warn;
34
35#[derive(Debug, Clone, Constructor)]
36pub struct AccountEventIndexer {
37 pub map: Arc<ExecutionInstrumentMap>,
38}
39
40#[derive(Debug, Clone, PartialEq)]
45pub enum AccountIndexError {
46 UntrackedBalance(AssetNameExchange),
54 Rejected(IndexError),
56}
57
58#[derive(Debug, Clone, PartialEq)]
61#[non_exhaustive]
62pub struct PartitionedSnapshot {
63 pub snapshot: AccountSnapshot,
65 pub untracked: Vec<AssetNameExchange>,
67}
68
69impl AccountEventIndexer {
70 pub fn index_account_event(
74 &self,
75 event: UnindexedAccountEvent,
76 ) -> Result<AccountEvent, AccountIndexError> {
77 let balance_asset = match &event.kind {
78 AccountEventKind::BalanceSnapshot(snapshot) => Some(snapshot.0.asset.clone()),
79 AccountEventKind::BalanceStreamUpdate(update) => Some(update.0.asset.clone()),
80 _ => None,
81 };
82
83 self.account_event(event)
84 .map_err(|error| match (error, balance_asset) {
85 (IndexError::AssetIndex(_), Some(asset)) => {
87 AccountIndexError::UntrackedBalance(asset)
88 }
89 (error, _) => AccountIndexError::Rejected(error),
90 })
91 }
92
93 pub fn account_event(&self, event: UnindexedAccountEvent) -> Result<AccountEvent, IndexError> {
94 let UnindexedAccountEvent { exchange, kind } = event;
95
96 let exchange = self.map.find_exchange_index(exchange)?;
97
98 let kind = match kind {
99 AccountEventKind::Snapshot(snapshot) => {
100 AccountEventKind::Snapshot(self.snapshot(snapshot)?)
101 }
102 AccountEventKind::BalanceSnapshot(snapshot) => {
103 AccountEventKind::BalanceSnapshot(self.asset_balance(snapshot.0).map(Snapshot)?)
104 }
105 AccountEventKind::BalanceStreamUpdate(snapshot) => {
106 AccountEventKind::BalanceStreamUpdate(
107 self.asset_balance_update(snapshot.0).map(Snapshot)?,
108 )
109 }
110 AccountEventKind::InstrumentBalanceUpdate(update) => {
111 AccountEventKind::InstrumentBalanceUpdate(self.instrument_balance_update(update)?)
112 }
113 AccountEventKind::InstrumentBalanceSnapshot(snapshot) => {
114 AccountEventKind::InstrumentBalanceSnapshot(
115 self.instrument_balance_snapshot(snapshot)?,
116 )
117 }
118 AccountEventKind::OrderSnapshot(snapshot) => {
119 AccountEventKind::OrderSnapshot(self.order_snapshot(snapshot.0).map(Snapshot)?)
120 }
121 AccountEventKind::OrderCancelled(response) => {
122 AccountEventKind::OrderCancelled(self.order_response_cancel(response)?)
123 }
124 AccountEventKind::Trade(trade) => AccountEventKind::Trade(self.trade(trade)?),
125 AccountEventKind::StreamTerminated(reason) => {
127 AccountEventKind::StreamTerminated(reason)
128 }
129 AccountEventKind::FillRecoveryGaveUp(gap) => {
130 AccountEventKind::FillRecoveryGaveUp(self.fill_recovery_gap(gap)?)
131 }
132 AccountEventKind::TradeAmended(amendment) => {
133 AccountEventKind::TradeAmended(self.trade_amendment(amendment)?)
134 }
135 AccountEventKind::ReinitFailed(failure) => {
136 AccountEventKind::ReinitFailed(self.reinit_failure(failure))
137 }
138 AccountEventKind::CashFlow(flow) => AccountEventKind::CashFlow(self.cash_flow(flow)?),
139 AccountEventKind::Notice(notice) => {
140 AccountEventKind::Notice(self.notice(exchange, notice))
141 }
142 AccountEventKind::MarginDebt(update) => {
143 AccountEventKind::MarginDebt(self.margin_debt(update)?)
144 }
145 };
146
147 Ok(AccountEvent { exchange, kind })
148 }
149
150 pub fn snapshot(
161 &self,
162 snapshot: UnindexedAccountSnapshot,
163 ) -> Result<AccountSnapshot, IndexError> {
164 let PartitionedSnapshot {
165 snapshot,
166 untracked,
167 } = self.snapshot_partitioned(snapshot)?;
168 match untracked.first() {
169 None => Ok(snapshot),
170 Some(asset) => Err(self
171 .map
172 .find_asset_index(asset)
173 .err()
174 .unwrap_or_else(|| IndexError::AssetIndex(format!("not registered: {asset}")))),
175 }
176 }
177
178 pub fn snapshot_partitioned(
197 &self,
198 snapshot: UnindexedAccountSnapshot,
199 ) -> Result<PartitionedSnapshot, IndexError> {
200 let UnindexedAccountSnapshot {
201 exchange,
202 balances,
203 instruments,
204 } = snapshot;
205
206 let exchange = self.map.find_exchange_index(exchange)?;
207
208 let mut untracked = Vec::new();
209 let mut indexed_balances = Vec::with_capacity(balances.len());
210 for balance in balances {
211 match self.map.find_asset_index(&balance.asset) {
212 Ok(asset) => indexed_balances.push(AssetBalance {
213 asset,
214 balance: balance.balance,
215 time_exchange: balance.time_exchange,
216 }),
217 Err(IndexError::AssetIndex(_)) => untracked.push(balance.asset),
218 Err(error) => return Err(error),
219 }
220 }
221
222 let instruments = instruments
223 .into_iter()
224 .map(|snapshot| {
225 let InstrumentAccountSnapshot {
226 instrument,
227 orders,
228 orders_complete,
229 position,
230 isolated,
231 } = snapshot;
232
233 let instrument = self.map.find_instrument_index(&instrument)?;
234
235 let orders = orders
236 .into_iter()
237 .map(|order| self.order_snapshot(order))
238 .collect::<Result<Vec<_>, _>>()?;
239
240 let isolated = isolated
244 .map(|state| self.isolated_instrument_state(state))
245 .transpose()?;
246
247 Ok(InstrumentAccountSnapshot {
248 instrument,
249 orders,
250 orders_complete,
251 position,
252 isolated,
253 })
254 })
255 .collect::<Result<Vec<_>, _>>()?;
256
257 Ok(PartitionedSnapshot {
258 snapshot: AccountSnapshot {
259 exchange,
260 balances: indexed_balances,
261 instruments,
262 },
263 untracked,
264 })
265 }
266
267 pub fn asset_balance(
268 &self,
269 balance: AssetBalance<AssetNameExchange>,
270 ) -> Result<AssetBalance<AssetIndex>, IndexError> {
271 let AssetBalance {
272 asset,
273 balance,
274 time_exchange,
275 } = balance;
276 let asset = self.map.find_asset_index(&asset)?;
277
278 Ok(AssetBalance {
279 asset,
280 balance,
281 time_exchange,
282 })
283 }
284
285 pub fn asset_balance_update(
286 &self,
287 update: AssetBalanceUpdate<AssetNameExchange>,
288 ) -> Result<AssetBalanceUpdate<AssetIndex>, IndexError> {
289 let AssetBalanceUpdate {
290 asset,
291 update,
292 time_exchange,
293 } = update;
294 let asset = self.map.find_asset_index(&asset)?;
295
296 Ok(AssetBalanceUpdate {
297 asset,
298 update,
299 time_exchange,
300 })
301 }
302
303 pub fn isolated_instrument_state(
309 &self,
310 state: IsolatedInstrumentState<AssetNameExchange>,
311 ) -> Result<IsolatedInstrumentState<AssetIndex>, IndexError> {
312 let IsolatedInstrumentState { base, quote, risk } = state;
313
314 Ok(IsolatedInstrumentState {
315 base: self.asset_balance(base)?,
316 quote: self.asset_balance(quote)?,
317 risk,
318 })
319 }
320
321 pub fn fill_recovery_gap(
329 &self,
330 gap: FillRecoveryGap<InstrumentNameExchange>,
331 ) -> Result<FillRecoveryGap<InstrumentIndex>, IndexError> {
332 let FillRecoveryGap {
333 scope,
334 start,
335 end,
336 attempts,
337 reason,
338 } = gap;
339
340 let scope = match scope {
341 FillRecoveryScope::Instruments(instruments) => FillRecoveryScope::Instruments(
342 instruments
343 .iter()
344 .map(|instrument| self.map.find_instrument_index(instrument))
345 .collect::<Result<_, _>>()?,
346 ),
347 FillRecoveryScope::AllInstruments => FillRecoveryScope::AllInstruments,
348 };
349
350 Ok(FillRecoveryGap {
351 scope,
352 start,
353 end,
354 attempts,
355 reason,
356 })
357 }
358
359 pub fn cash_flow(
364 &self,
365 flow: CashFlow<AssetNameExchange, InstrumentNameExchange>,
366 ) -> Result<CashFlow<AssetIndex, InstrumentIndex>, IndexError> {
367 let CashFlow {
368 kind,
369 asset,
370 amount,
371 instrument,
372 time_exchange,
373 id,
374 } = flow;
375
376 Ok(CashFlow {
377 kind,
378 asset: self.map.find_asset_index(&asset)?,
379 amount,
380 instrument: instrument
381 .map(|instrument| self.map.find_instrument_index(&instrument))
382 .transpose()?,
383 time_exchange,
384 id,
385 })
386 }
387
388 pub fn margin_debt(
397 &self,
398 update: MarginDebtUpdate<AssetNameExchange, InstrumentNameExchange>,
399 ) -> Result<MarginDebtUpdate<AssetIndex, InstrumentIndex>, IndexError> {
400 let MarginDebtUpdate {
401 asset,
402 instrument,
403 debt,
404 kind,
405 venue_kind,
406 time_exchange,
407 } = update;
408
409 Ok(MarginDebtUpdate {
410 asset: self.map.find_asset_index(&asset)?,
411 instrument: instrument
412 .map(|instrument| self.map.find_instrument_index(&instrument))
413 .transpose()?,
414 debt,
415 kind,
416 venue_kind,
417 time_exchange,
418 })
419 }
420
421 pub fn notice(
427 &self,
428 exchange: ExchangeIndex,
429 notice: AccountNotice<InstrumentNameExchange>,
430 ) -> AccountNotice<InstrumentIndex> {
431 let AccountNotice {
432 kind,
433 instrument,
434 time_exchange,
435 margin_level,
436 status,
437 } = notice;
438
439 let instrument = instrument.and_then(|instrument| {
440 self.map
441 .find_instrument_index(&instrument)
442 .inspect_err(|error| {
443 warn!(
444 ?exchange,
445 %instrument,
446 %kind,
447 %status,
448 %error,
449 "account notice names an instrument that is not in the map, \
450 delivering it unattributed",
451 );
452 })
453 .ok()
454 });
455
456 AccountNotice {
457 kind,
458 instrument,
459 time_exchange,
460 margin_level,
461 status,
462 }
463 }
464
465 pub fn instrument_balance_update(
470 &self,
471 update: InstrumentBalanceUpdate<AssetNameExchange, InstrumentNameExchange>,
472 ) -> Result<InstrumentBalanceUpdate, IndexError> {
473 let InstrumentBalanceUpdate {
474 instrument,
475 balance,
476 } = update;
477
478 Ok(InstrumentBalanceUpdate {
479 instrument: self.map.find_instrument_index(&instrument)?,
480 balance: self.asset_balance_update(balance)?,
481 })
482 }
483
484 pub fn instrument_balance_snapshot(
485 &self,
486 snapshot: InstrumentBalanceSnapshot<AssetNameExchange, InstrumentNameExchange>,
487 ) -> Result<InstrumentBalanceSnapshot, IndexError> {
488 let InstrumentBalanceSnapshot {
489 instrument,
490 balance,
491 } = snapshot;
492
493 Ok(InstrumentBalanceSnapshot {
494 instrument: self.map.find_instrument_index(&instrument)?,
495 balance: self.asset_balance(balance)?,
496 })
497 }
498
499 pub fn order_snapshot(
500 &self,
501 order: UnindexedOrderSnapshot,
502 ) -> Result<OrderSnapshot, IndexError> {
503 let Order {
504 key,
505 side,
506 price,
507 quantity,
508 kind,
509 time_in_force,
510 state,
511 } = order;
512
513 let key = self.order_key(key)?;
514 let state = self.order_state(state);
515
516 Ok(Order {
517 key,
518 side,
519 price,
520 quantity,
521 kind,
522 time_in_force,
523 state,
524 })
525 }
526
527 pub fn order_response_cancel(
528 &self,
529 response: OrderResponseCancel<ExchangeId, AssetNameExchange, InstrumentNameExchange>,
530 ) -> Result<OrderResponseCancel, IndexError> {
531 let OrderResponseCancel { key, state } = response;
532
533 Ok(OrderResponseCancel {
534 key: self.order_key(key)?,
535 state: match state {
536 Ok(cancelled) => Ok(cancelled),
537 Err(error) => Err(self.order_error(error)),
538 },
539 })
540 }
541
542 pub fn order_key(&self, key: UnindexedOrderKey) -> Result<OrderKey, IndexError> {
543 let UnindexedOrderKey {
544 exchange,
545 instrument,
546 strategy,
547 cid,
548 } = key;
549
550 Ok(OrderKey {
551 exchange: self.map.find_exchange_index(exchange)?,
552 instrument: self.map.find_instrument_index(&instrument)?,
553 strategy,
554 cid,
555 })
556 }
557
558 pub fn order_state(&self, state: UnindexedOrderState) -> OrderState {
563 match state {
564 UnindexedOrderState::Active(active) => OrderState::Active(active),
565 UnindexedOrderState::Inactive(inactive) => match inactive {
566 InactiveOrderState::OpenFailed(failed) => {
567 OrderState::inactive(self.order_error(failed))
568 }
569 InactiveOrderState::Cancelled(cancelled) => OrderState::inactive(cancelled),
570 InactiveOrderState::FullyFilled(filled) => OrderState::fully_filled(filled),
571 InactiveOrderState::Expired(expired) => OrderState::expired(expired),
572 },
573 }
574 }
575
576 pub fn api_error(&self, error: UnindexedApiError) -> ApiError {
590 self.index_api_error(error, ApiError::OrderRejected)
591 }
592
593 fn index_api_error(
598 &self,
599 error: UnindexedApiError,
600 rejected: fn(String) -> ApiError,
601 ) -> ApiError {
602 match error {
603 UnindexedApiError::RateLimit { retry_after } => ApiError::rate_limited(retry_after),
604 UnindexedApiError::CapacityExceeded(msg) => ApiError::CapacityExceeded(msg),
605 UnindexedApiError::Unauthenticated(msg) => ApiError::Unauthenticated(msg),
606 UnindexedApiError::AssetInvalid(asset, value) => {
607 match self.map.find_asset_index(&asset) {
608 Ok(asset) => ApiError::AssetInvalid(asset, value),
609 Err(_) => self.degrade_unresolvable(
610 UnindexedApiError::AssetInvalid(asset, value),
611 rejected,
612 ),
613 }
614 }
615 UnindexedApiError::InstrumentInvalid(instrument, value) => {
616 match self.map.find_instrument_index(&instrument) {
617 Ok(instrument) => ApiError::InstrumentInvalid(instrument, value),
618 Err(_) => self.degrade_unresolvable(
619 UnindexedApiError::InstrumentInvalid(instrument, value),
620 rejected,
621 ),
622 }
623 }
624 UnindexedApiError::BalanceInsufficient(asset, value) => {
625 let asset = asset.and_then(|asset| {
626 self.map
627 .find_asset_index(&asset)
628 .inspect_err(|_| {
629 warn!(
630 exchange = %self.map.exchange.value,
631 %asset,
632 message = %value,
633 "AccountEventIndexer dropping the asset of a BalanceInsufficient \
634 error: the instrument map does not hold it"
635 )
636 })
637 .ok()
638 });
639 ApiError::BalanceInsufficient(asset, value)
640 }
641 UnindexedApiError::OrderRejected(reason) => ApiError::OrderRejected(reason),
642 UnindexedApiError::OrderAlreadyCancelled => ApiError::OrderAlreadyCancelled,
643 UnindexedApiError::OrderAlreadyFullyFilled => ApiError::OrderAlreadyFullyFilled,
644 UnindexedApiError::OrderAlreadyExpired => ApiError::OrderAlreadyExpired,
645 UnindexedApiError::DuplicateClientOrderId(message) => {
646 ApiError::DuplicateClientOrderId(message)
647 }
648 UnindexedApiError::RequestRejected(reason) => ApiError::RequestRejected(reason),
649 UnindexedApiError::BorrowRejected(reject) => ApiError::BorrowRejected(reject),
650 }
651 }
652
653 fn degrade_unresolvable(
657 &self,
658 error: UnindexedApiError,
659 rejected: fn(String) -> ApiError,
660 ) -> ApiError {
661 let error = error.to_string();
662 warn!(
663 exchange = %self.map.exchange.value,
664 %error,
665 "AccountEventIndexer degrading an API error to a rejection: the instrument map does \
666 not hold the asset or instrument it names"
667 );
668 rejected(error)
669 }
670
671 pub fn order_request<Kind>(
679 &self,
680 order: &OrderEvent<Kind, ExchangeIndex, InstrumentIndex>,
681 ) -> Result<OrderEvent<Kind, ExchangeId, &InstrumentNameExchange>, KeyError>
682 where
683 Kind: Clone,
684 {
685 let OrderEvent {
686 key:
687 OrderKey {
688 exchange,
689 instrument,
690 strategy,
691 cid,
692 },
693 state,
694 } = order;
695
696 let exchange = self.map.find_exchange_id(*exchange)?;
697 let instrument = self
698 .map
699 .find_own_instrument_name_exchange(*instrument)
700 .ok_or_else(|| {
701 KeyError::InstrumentKey(format!(
702 "ExecutionInstrumentMap for {} does not contain: {instrument}",
703 self.map.exchange.value
704 ))
705 })?;
706
707 Ok(OrderEvent {
708 key: OrderKey {
709 exchange,
710 instrument,
711 strategy: strategy.clone(),
712 cid: cid.clone(),
713 },
714 state: state.clone(),
715 })
716 }
717
718 pub fn order_error(&self, error: UnindexedOrderError) -> OrderError {
720 match error {
721 UnindexedOrderError::Connectivity(error) => OrderError::Connectivity(error),
722 UnindexedOrderError::NotPlaced(reason) => OrderError::NotPlaced(reason),
723 UnindexedOrderError::Rejected(error) => OrderError::Rejected(self.api_error(error)),
724 UnindexedOrderError::UnsupportedOrderType(msg) => OrderError::UnsupportedOrderType(msg),
725 UnindexedOrderError::InvalidPrecision(violation) => {
726 OrderError::InvalidPrecision(violation)
727 }
728 UnindexedOrderError::OutOfBounds(violation) => OrderError::OutOfBounds(violation),
729 UnindexedOrderError::UnsupportedMarginEffect(effect) => {
730 OrderError::UnsupportedMarginEffect(effect)
731 }
732 }
733 }
734
735 pub fn client_error(&self, error: UnindexedClientError) -> ClientError {
739 match error {
740 UnindexedClientError::Connectivity(error) => ClientError::Connectivity(error),
741 UnindexedClientError::Api(error) => {
742 ClientError::Api(self.index_api_error(error, ApiError::RequestRejected))
743 }
744 UnindexedClientError::TaskFailed(value) => ClientError::TaskFailed(value),
745 UnindexedClientError::Internal(value) => ClientError::Internal(value),
746 UnindexedClientError::Truncated { fills_read } => ClientError::Truncated { fills_read },
747 UnindexedClientError::TruncatedSnapshot { limit } => {
748 ClientError::TruncatedSnapshot { limit }
749 }
750 }
751 }
752
753 pub fn account_stream_init_error(
755 &self,
756 error: UnindexedAccountStreamInitError,
757 ) -> AccountStreamInitError {
758 match error {
759 AccountStreamInitError::Client(error) => {
760 AccountStreamInitError::Client(self.client_error(error))
761 }
762 AccountStreamInitError::Index(error) => AccountStreamInitError::Index(error),
763 }
764 }
765
766 pub fn reinit_failure(
768 &self,
769 failure: AccountReinitFailure<AssetNameExchange, InstrumentNameExchange>,
770 ) -> AccountReinitFailure {
771 let AccountReinitFailure { attempt, error } = failure;
772 AccountReinitFailure {
773 attempt,
774 error: self.account_stream_init_error(error),
775 }
776 }
777
778 pub fn trade_amendment(
781 &self,
782 amendment: TradeAmendment<AssetNameExchange, InstrumentNameExchange>,
783 ) -> Result<TradeAmendment<AssetIndex, InstrumentIndex>, IndexError> {
784 let TradeAmendment {
785 instrument,
786 order_id,
787 time_exchange,
788 original,
789 kind,
790 } = amendment;
791
792 let instrument = self.map.find_instrument_index(&instrument)?;
793 let kind = match kind {
794 TradeAmendmentKind::Busted { quantity } => TradeAmendmentKind::Busted { quantity },
795 TradeAmendmentKind::Corrected { replacement } => TradeAmendmentKind::Corrected {
796 replacement: self.trade(replacement)?,
797 },
798 TradeAmendmentKind::CorrectedUnresolved {
799 id,
800 price,
801 quantity,
802 } => TradeAmendmentKind::CorrectedUnresolved {
803 id,
804 price,
805 quantity,
806 },
807 };
808
809 Ok(TradeAmendment {
810 instrument,
811 order_id,
812 time_exchange,
813 original,
814 kind,
815 })
816 }
817
818 pub fn trade(
843 &self,
844 trade: Trade<AssetNameExchange, InstrumentNameExchange>,
845 ) -> Result<Trade<AssetIndex, InstrumentIndex>, IndexError> {
846 let Trade {
847 id,
848 order_id,
849 instrument,
850 strategy,
851 time_exchange,
852 side,
853 price: trade_price,
854 quantity,
855 order_filled_quantity,
856 fees,
857 origin,
858 position_side,
859 client_order_id,
860 time_source,
861 } = trade;
862
863 let instrument_index = self.map.find_instrument_index(&instrument)?;
864 let underlying = &self
867 .map
868 .instruments
869 .get_index(instrument_index.index())
870 .ok_or_else(|| {
871 IndexError::InstrumentIndex(format!(
872 "ExecutionInstrumentMap has no instrument at {instrument_index} for {instrument}"
873 ))
874 })?
875 .underlying;
876
877 let fees = match self.map.find_asset_index(&fees.asset) {
878 Ok(fee_asset) => {
879 let fees_quote = if fee_asset == underlying.quote {
880 Some(fees.fees)
882 } else if fee_asset == underlying.base {
883 Some(fees.fees * trade_price)
885 } else {
886 None
888 };
889 AssetFees::new(fee_asset, fees.fees, fees_quote)
890 }
891 Err(error) => {
892 warn!(
893 exchange = %self.map.exchange.value,
894 %instrument,
895 trade_id = %id,
896 fee_asset = %fees.asset,
897 fee = %fees.fees,
898 %error,
899 "trade's fee asset is not in the map, keeping the trade with its fee zeroed; \
900 register the asset with IndexedInstrumentsBuilder::add_asset to keep its fee",
901 );
902 AssetFees::new(underlying.quote, Decimal::ZERO, None)
903 }
904 };
905
906 Ok(Trade {
907 id,
908 order_id,
909 instrument: instrument_index,
910 strategy,
911 time_exchange,
912 side,
913 price: trade_price,
914 quantity,
915 order_filled_quantity,
916 fees,
917 origin,
918 position_side,
919 client_order_id,
920 time_source,
921 })
922 }
923}
924
925#[cfg(test)]
926#[allow(clippy::unwrap_used)] mod tests {
928 use super::*;
929 use crate::{
930 error::StreamTerminationReason, fill_recovery::FillRecoveryFailure,
931 map::generate_execution_instrument_map,
932 };
933 use chrono::{DateTime, Utc};
934 use rustrade_instrument::{index::IndexedInstruments, test_utils};
935
936 #[test]
937 fn a_partitioned_snapshot_skips_an_unregistered_balance_and_the_strict_one_fails() {
938 let indexer = binance_indexer();
939 let snapshot = || {
940 UnindexedAccountSnapshot::new(
941 ExchangeId::BinanceSpot,
942 ["usdt", "bnb", "btc", "sol"]
943 .map(|asset| {
944 AssetBalance::new(
945 AssetNameExchange::new(asset),
946 crate::balance::Balance::new(Decimal::ONE, Decimal::ONE),
947 DateTime::<Utc>::MIN_UTC,
948 )
949 })
950 .to_vec(),
951 vec![],
952 )
953 };
954
955 let partitioned = indexer.snapshot_partitioned(snapshot()).unwrap();
956 assert_eq!(partitioned.snapshot.balances.len(), 2);
957 assert_eq!(
958 partitioned.untracked,
959 vec![AssetNameExchange::new("bnb"), AssetNameExchange::new("sol")]
960 );
961 assert!(matches!(
962 indexer.snapshot(snapshot()),
963 Err(IndexError::AssetIndex(_))
964 ));
965 }
966
967 #[test]
968 fn a_partitioned_snapshot_still_fails_on_an_unregistered_instrument_or_exchange() {
969 let indexer = binance_indexer();
970 let instrument = |name: &str| {
971 InstrumentAccountSnapshot::new(
972 InstrumentNameExchange::new(name),
973 vec![],
974 false,
975 crate::position::PositionReport::Unreported,
976 None,
977 )
978 };
979 let unknown_instrument = UnindexedAccountSnapshot::new(
980 ExchangeId::BinanceSpot,
981 vec![],
982 vec![instrument("ETHUSDT")],
983 );
984 assert!(matches!(
985 indexer.snapshot_partitioned(unknown_instrument),
986 Err(IndexError::InstrumentIndex(_))
987 ));
988 let skipped_and_unknown_instrument = UnindexedAccountSnapshot::new(
990 ExchangeId::BinanceSpot,
991 vec![AssetBalance::new(
992 AssetNameExchange::new("bnb"),
993 crate::balance::Balance::new(Decimal::ONE, Decimal::ONE),
994 DateTime::<Utc>::MIN_UTC,
995 )],
996 vec![instrument("ETHUSDT")],
997 );
998 assert!(matches!(
999 indexer.snapshot_partitioned(skipped_and_unknown_instrument),
1000 Err(IndexError::InstrumentIndex(_))
1001 ));
1002 let other_exchange = UnindexedAccountSnapshot::new(ExchangeId::Ibkr, vec![], vec![]);
1003 assert!(matches!(
1004 indexer.snapshot_partitioned(other_exchange),
1005 Err(IndexError::ExchangeIndex(_))
1006 ));
1007 }
1008
1009 #[test]
1012 fn an_order_request_for_another_exchanges_instrument_fails() {
1013 use crate::order::id::{ClientOrderId, StrategyId};
1014 let instruments = IndexedInstruments::new(vec![
1015 test_utils::instrument(ExchangeId::Kraken, "BTC", "USDT"),
1016 test_utils::instrument(ExchangeId::BinanceSpot, "BTC", "USDT"),
1017 ]);
1018 let binance = generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot)
1019 .map(Arc::new)
1020 .map(AccountEventIndexer::new)
1021 .unwrap();
1022 let instrument_on = |exchange| {
1023 instruments
1024 .find_instrument_index(
1025 exchange,
1026 &test_utils::instrument(exchange, "BTC", "USDT").name_internal,
1027 )
1028 .unwrap()
1029 };
1030 let order = |exchange, instrument| OrderEvent {
1031 key: OrderKey {
1032 exchange,
1033 instrument,
1034 strategy: StrategyId::new("test"),
1035 cid: ClientOrderId::random(),
1036 },
1037 state: (),
1038 };
1039 let binance_index = instruments
1040 .find_exchange_index(ExchangeId::BinanceSpot)
1041 .unwrap();
1042
1043 let own = binance
1044 .order_request(&order(
1045 binance_index,
1046 instrument_on(ExchangeId::BinanceSpot),
1047 ))
1048 .unwrap();
1049 assert_eq!(own.key.exchange, ExchangeId::BinanceSpot);
1050 assert_eq!(own.key.instrument.as_ref(), "BTC_USDT");
1051
1052 let foreign =
1053 binance.order_request(&order(binance_index, instrument_on(ExchangeId::Kraken)));
1054 assert!(
1055 matches!(foreign, Err(KeyError::InstrumentKey(_))),
1056 "{foreign:?}"
1057 );
1058 }
1059
1060 fn binance_indexer() -> AccountEventIndexer {
1061 let instruments = IndexedInstruments::new(vec![test_utils::instrument(
1062 ExchangeId::BinanceSpot,
1063 "BTC",
1064 "USDT",
1065 )]);
1066 let map = generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot).unwrap();
1067 AccountEventIndexer::new(Arc::new(map))
1068 }
1069
1070 fn balance_update(asset: &str) -> UnindexedAccountEvent {
1071 UnindexedAccountEvent::new(
1072 ExchangeId::BinanceSpot,
1073 AccountEventKind::BalanceStreamUpdate(Snapshot(AssetBalanceUpdate::new(
1074 AssetNameExchange::new(asset),
1075 crate::balance::BalanceUpdate::new(rust_decimal::Decimal::ONE, Default::default()),
1076 DateTime::<Utc>::MIN_UTC,
1077 ))),
1078 )
1079 }
1080
1081 #[test]
1082 fn a_balance_for_an_unregistered_asset_is_an_untracked_balance_not_a_rejection() {
1083 let indexer = binance_indexer();
1084
1085 assert!(indexer.index_account_event(balance_update("usdt")).is_ok());
1086 assert_eq!(
1087 indexer.index_account_event(balance_update("bnb")),
1088 Err(AccountIndexError::UntrackedBalance(AssetNameExchange::new(
1089 "bnb"
1090 )))
1091 );
1092
1093 let snapshot = UnindexedAccountEvent::new(
1094 ExchangeId::BinanceSpot,
1095 AccountEventKind::BalanceSnapshot(Snapshot(AssetBalance::new(
1096 AssetNameExchange::new("bnb"),
1097 crate::balance::Balance::new(Default::default(), Default::default()),
1098 DateTime::<Utc>::MIN_UTC,
1099 ))),
1100 );
1101 assert!(matches!(
1102 indexer.index_account_event(snapshot),
1103 Err(AccountIndexError::UntrackedBalance(_))
1104 ));
1105 }
1106
1107 #[test]
1108 fn an_unregistered_asset_on_any_other_event_is_still_a_rejection() {
1109 let indexer = binance_indexer();
1110 let cash_flow = UnindexedAccountEvent::new(
1111 ExchangeId::BinanceSpot,
1112 AccountEventKind::CashFlow(CashFlow {
1113 kind: crate::cash_flow::CashFlowKind::Rebate,
1114 asset: AssetNameExchange::new("bnb"),
1115 amount: rust_decimal::Decimal::ONE,
1116 instrument: None,
1117 time_exchange: DateTime::<Utc>::MIN_UTC,
1118 id: None,
1119 }),
1120 );
1121 assert!(matches!(
1122 indexer.index_account_event(cash_flow),
1123 Err(AccountIndexError::Rejected(IndexError::AssetIndex(_)))
1124 ));
1125
1126 let mut wrong_exchange = balance_update("bnb");
1128 wrong_exchange.exchange = ExchangeId::Kraken;
1129 assert!(matches!(
1130 indexer.index_account_event(wrong_exchange),
1131 Err(AccountIndexError::Rejected(IndexError::ExchangeIndex(_)))
1132 ));
1133 }
1134
1135 #[test]
1138 fn an_instrument_balance_snapshot_indexes_its_instrument_and_asset() {
1139 let indexer = binance_indexer();
1140 let snapshot = |instrument: &str, asset: &str| {
1141 UnindexedAccountEvent::new(
1142 ExchangeId::BinanceSpot,
1143 AccountEventKind::InstrumentBalanceSnapshot(InstrumentBalanceSnapshot::new(
1144 InstrumentNameExchange::new(instrument),
1145 AssetBalance::new(
1146 AssetNameExchange::new(asset),
1147 crate::balance::Balance::new(Decimal::ONE, Decimal::ONE),
1148 DateTime::<Utc>::MIN_UTC,
1149 ),
1150 )),
1151 )
1152 };
1153
1154 let indexed = indexer
1155 .index_account_event(snapshot("BTC_USDT", "usdt"))
1156 .unwrap();
1157 let AccountEventKind::InstrumentBalanceSnapshot(indexed) = indexed.kind else {
1158 panic!("{indexed:?}");
1159 };
1160 let map = &indexer.map;
1161 assert_eq!(
1162 indexed.instrument,
1163 map.find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1164 .unwrap()
1165 );
1166 assert_eq!(
1167 indexed.balance.asset,
1168 map.find_asset_index(&AssetNameExchange::new("usdt"))
1169 .unwrap()
1170 );
1171 assert_eq!(
1172 indexed.balance.balance,
1173 crate::balance::Balance::new(Decimal::ONE, Decimal::ONE)
1174 );
1175
1176 assert!(matches!(
1177 indexer.index_account_event(snapshot("ETH_USDT", "usdt")),
1178 Err(AccountIndexError::Rejected(IndexError::InstrumentIndex(_)))
1179 ));
1180 assert!(matches!(
1181 indexer.index_account_event(snapshot("BTC_USDT", "bnb")),
1182 Err(AccountIndexError::Rejected(IndexError::AssetIndex(_)))
1183 ));
1184 }
1185
1186 #[test]
1189 fn account_event_passes_stream_terminated_through_verbatim() {
1190 let indexer = binance_indexer();
1191 let reason = StreamTerminationReason::ReconnectBudgetExhausted {
1192 attempts: 3,
1193 last_error: "socket reset".to_string(),
1194 };
1195
1196 let indexed = indexer
1197 .account_event(UnindexedAccountEvent::new(
1198 ExchangeId::BinanceSpot,
1199 AccountEventKind::StreamTerminated(reason.clone()),
1200 ))
1201 .unwrap();
1202
1203 assert_eq!(indexed.exchange, indexer.map.exchange.key);
1204 assert!(
1205 matches!(indexed.kind, AccountEventKind::StreamTerminated(r) if r == reason),
1206 "expected StreamTerminated to pass through unchanged",
1207 );
1208 }
1209
1210 #[test]
1213 fn account_event_indexes_a_reinit_failure() {
1214 let indexer = binance_indexer();
1215 let btc = indexer
1216 .map
1217 .find_asset_index(&AssetNameExchange::new("BTC"))
1218 .unwrap();
1219 let unindexed_client = AccountStreamInitError::Client(ClientError::Api(
1220 ApiError::BalanceInsufficient(Some(AssetNameExchange::new("BTC")), "low".to_owned()),
1221 ));
1222 let indexed_client = AccountStreamInitError::Client(ClientError::Api(
1223 ApiError::BalanceInsufficient(Some(btc), "low".to_owned()),
1224 ));
1225 let index_error = IndexError::AssetIndex("ETH".to_owned());
1226
1227 for (error, expected) in [
1228 (unindexed_client, indexed_client),
1229 (
1230 AccountStreamInitError::Index(index_error.clone()),
1231 AccountStreamInitError::Index(index_error),
1232 ),
1233 ] {
1234 let indexed = indexer
1235 .account_event(UnindexedAccountEvent::new(
1236 ExchangeId::BinanceSpot,
1237 AccountEventKind::ReinitFailed(AccountReinitFailure::new(2, error)),
1238 ))
1239 .unwrap();
1240
1241 assert_eq!(indexed.exchange, indexer.map.exchange.key);
1242 assert_eq!(
1243 indexed.kind,
1244 AccountEventKind::ReinitFailed(AccountReinitFailure::new(2, expected))
1245 );
1246 }
1247 }
1248
1249 #[test]
1252 fn account_event_indexes_a_fill_recovery_give_up_by_its_instruments() {
1253 let indexer = binance_indexer();
1254 let start = DateTime::<Utc>::MIN_UTC;
1255 let end = DateTime::<Utc>::MAX_UTC;
1256 let give_up = |scope| {
1257 UnindexedAccountEvent::new(
1258 ExchangeId::BinanceSpot,
1259 AccountEventKind::FillRecoveryGaveUp(FillRecoveryGap::new(
1260 scope,
1261 start,
1262 end,
1263 6,
1264 FillRecoveryFailure::TimedOut { timeout_secs: 30 },
1265 )),
1266 )
1267 };
1268 let index_of = |scope| match indexer.account_event(give_up(scope)) {
1269 Ok(AccountEvent {
1270 kind: AccountEventKind::FillRecoveryGaveUp(gap),
1271 ..
1272 }) => Ok(gap),
1273 Ok(other) => panic!("expected FillRecoveryGaveUp, got {other:?}"),
1274 Err(error) => Err(error),
1275 };
1276 let Ok(btc) = indexer
1277 .map
1278 .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1279 else {
1280 panic!("BTC_USDT is mapped");
1281 };
1282
1283 assert_eq!(
1284 index_of(FillRecoveryScope::Instruments(vec![
1285 InstrumentNameExchange::new("BTC_USDT")
1286 ])),
1287 Ok(FillRecoveryGap::new(
1288 FillRecoveryScope::Instruments(vec![btc]),
1289 start,
1290 end,
1291 6,
1292 FillRecoveryFailure::TimedOut { timeout_secs: 30 },
1293 ))
1294 );
1295 assert_eq!(
1296 index_of(FillRecoveryScope::AllInstruments).map(|gap| gap.scope),
1297 Ok(FillRecoveryScope::AllInstruments)
1298 );
1299 assert!(
1300 index_of(FillRecoveryScope::Instruments(vec![
1301 InstrumentNameExchange::new("BTC_USDT"),
1302 InstrumentNameExchange::new("ETHUSDT"),
1303 ]))
1304 .is_err(),
1305 "an unmapped instrument fails the whole give-up"
1306 );
1307 }
1308
1309 #[test]
1312 fn account_event_indexes_a_trade_amendment() {
1313 use crate::{
1314 order::id::{OrderId, StrategyId},
1315 trade::TradeId,
1316 };
1317 use rust_decimal::Decimal;
1318 use rustrade_instrument::Side;
1319
1320 let indexer = binance_indexer();
1321 let time = DateTime::<Utc>::MIN_UTC;
1322 let amended = |instrument: &str, kind| {
1323 UnindexedAccountEvent::new(
1324 ExchangeId::BinanceSpot,
1325 AccountEventKind::TradeAmended(TradeAmendment::new(
1326 InstrumentNameExchange::new(instrument),
1327 OrderId::new("ord-1"),
1328 time,
1329 Some(TradeId::new("t-1")),
1330 kind,
1331 )),
1332 )
1333 };
1334 let index_of = |event| match indexer.account_event(event) {
1335 Ok(AccountEvent {
1336 kind: AccountEventKind::TradeAmended(amendment),
1337 ..
1338 }) => Ok(amendment),
1339 Ok(other) => panic!("expected TradeAmended, got {other:?}"),
1340 Err(error) => Err(error),
1341 };
1342 let Ok(btc) = indexer
1343 .map
1344 .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1345 else {
1346 panic!("BTC_USDT is mapped");
1347 };
1348
1349 assert_eq!(
1350 index_of(amended(
1351 "BTC_USDT",
1352 TradeAmendmentKind::Busted {
1353 quantity: Some(Decimal::ONE)
1354 }
1355 )),
1356 Ok(TradeAmendment::new(
1357 btc,
1358 OrderId::new("ord-1"),
1359 time,
1360 Some(TradeId::new("t-1")),
1361 TradeAmendmentKind::Busted {
1362 quantity: Some(Decimal::ONE)
1363 },
1364 ))
1365 );
1366
1367 let replacement = Trade::new(
1368 TradeId::new("t-2"),
1369 OrderId::new("ord-1"),
1370 InstrumentNameExchange::new("BTC_USDT"),
1371 StrategyId::unknown(),
1372 time,
1373 Side::Buy,
1374 Decimal::TEN,
1375 Decimal::ONE,
1376 None,
1377 AssetFees::new(AssetNameExchange::new("USDT"), Decimal::ONE, None),
1378 );
1379 let Ok(TradeAmendment {
1380 kind: TradeAmendmentKind::Corrected { replacement },
1381 ..
1382 }) = index_of(amended(
1383 "BTC_USDT",
1384 TradeAmendmentKind::Corrected { replacement },
1385 ))
1386 else {
1387 panic!("a correction indexes to a correction");
1388 };
1389 assert_eq!(replacement.instrument, btc);
1390 assert_eq!(
1391 replacement.fees.fees_quote,
1392 Some(Decimal::ONE),
1393 "indexed as a trade is, fees and all"
1394 );
1395
1396 assert!(
1397 index_of(amended(
1398 "ETHUSDT",
1399 TradeAmendmentKind::Busted { quantity: None }
1400 ))
1401 .is_err(),
1402 "an unmapped instrument fails the amendment"
1403 );
1404 }
1405
1406 #[test]
1410 fn trade_keeps_a_fill_whose_fee_asset_is_unmapped() {
1411 use crate::{
1412 order::id::{OrderId, StrategyId},
1413 trade::TradeId,
1414 };
1415 use rust_decimal_macros::dec;
1416 use rustrade_instrument::{Side, index::builder::IndexedInstrumentsBuilder};
1417
1418 let fill = |fee_asset: &str, fee| {
1419 Trade::new(
1420 TradeId::new("t-1"),
1421 OrderId::new("ord-1"),
1422 InstrumentNameExchange::new("BTC_USDT"),
1423 StrategyId::unknown(),
1424 DateTime::<Utc>::MIN_UTC,
1425 Side::Buy,
1426 dec!(50000),
1427 dec!(0.01),
1428 None,
1429 AssetFees::new(AssetNameExchange::new(fee_asset), fee, None),
1430 )
1431 };
1432 let asset = |indexer: &AccountEventIndexer, name: &str| {
1433 indexer
1434 .map
1435 .find_asset_index(&AssetNameExchange::new(name))
1436 .unwrap()
1437 };
1438
1439 let indexer = binance_indexer();
1440 let usdt = asset(&indexer, "USDT");
1441 let btc = asset(&indexer, "BTC");
1442
1443 let (quote, warnings) = count_warnings(|| indexer.trade(fill("USDT", dec!(0.5))));
1444 assert_eq!(
1445 quote.unwrap().fees,
1446 AssetFees::new(usdt, dec!(0.5), Some(dec!(0.5)))
1447 );
1448 assert_eq!(warnings, 0);
1449
1450 let (base, warnings) = count_warnings(|| indexer.trade(fill("BTC", dec!(0.00001))));
1451 assert_eq!(
1452 base.unwrap().fees,
1453 AssetFees::new(btc, dec!(0.00001), Some(dec!(0.5)))
1454 );
1455 assert_eq!(warnings, 0);
1456
1457 for unmapped in ["BNB", "UNKNOWN"] {
1458 let (kept, warnings) = count_warnings(|| indexer.trade(fill(unmapped, dec!(0.001))));
1459 let kept = kept.unwrap_or_else(|error| panic!("{unmapped}: dropped: {error}"));
1460 assert_eq!(
1461 kept.quantity,
1462 dec!(0.01),
1463 "{unmapped}: the fill itself is intact"
1464 );
1465 assert_eq!(
1466 kept.fees,
1467 AssetFees::new(usdt, Decimal::ZERO, None),
1468 "{unmapped}: fee zeroed in quote, with no quote equivalent"
1469 );
1470 assert_eq!(warnings, 1, "{unmapped}: the zeroed fee is reported");
1471 }
1472
1473 let instruments = IndexedInstrumentsBuilder::new()
1474 .add_instrument(test_utils::instrument(
1475 ExchangeId::BinanceSpot,
1476 "BTC",
1477 "USDT",
1478 ))
1479 .add_asset(test_utils::exchange_asset(ExchangeId::BinanceSpot, "BNB"))
1480 .build();
1481 let with_bnb = AccountEventIndexer::new(Arc::new(
1482 generate_execution_instrument_map(&instruments, ExchangeId::BinanceSpot).unwrap(),
1483 ));
1484 let bnb = asset(&with_bnb, "BNB");
1485 let (third_party, warnings) = count_warnings(|| with_bnb.trade(fill("BNB", dec!(0.001))));
1486 assert_eq!(
1487 third_party.unwrap().fees,
1488 AssetFees::new(bnb, dec!(0.001), None),
1489 "a registered fee asset keeps its fee, with no quote equivalent"
1490 );
1491 assert_eq!(warnings, 0);
1492
1493 let mut unmapped_instrument = fill("USDT", dec!(0.5));
1494 unmapped_instrument.instrument = InstrumentNameExchange::new("ETH_USDT");
1495 assert!(
1496 indexer.trade(unmapped_instrument).is_err(),
1497 "an unmapped instrument still fails"
1498 );
1499 }
1500
1501 fn count_warnings<T>(index: impl FnOnce() -> T) -> (T, usize) {
1505 let (value, events) = rustrade_instrument::test_utils::logs::capture(index);
1506 let warnings = events
1507 .iter()
1508 .filter(|event| event.level == tracing::Level::WARN)
1509 .count();
1510 (value, warnings)
1511 }
1512
1513 fn btc_usdt() -> InstrumentNameExchange {
1514 binance_indexer()
1515 .map
1516 .exchange_instruments()
1517 .next()
1518 .cloned()
1519 .unwrap_or_else(|| panic!("the map holds one instrument"))
1520 }
1521
1522 #[test]
1523 fn balance_insufficient_keeps_a_held_asset_and_drops_an_unheld_one() {
1524 let indexer = binance_indexer();
1525 let btc = indexer
1526 .map
1527 .find_asset_index(&AssetNameExchange::new("BTC"))
1528 .unwrap();
1529
1530 for (asset, expected, warnings) in [
1531 (Some("BTC"), Some(btc), 0),
1532 (Some("ETH_USDT"), None, 1),
1534 (None, None, 0),
1535 ] {
1536 let (indexed, warned) = count_warnings(|| {
1537 indexer.api_error(ApiError::BalanceInsufficient(
1538 asset.map(AssetNameExchange::new),
1539 "low".to_string(),
1540 ))
1541 });
1542 assert_eq!(
1543 indexed,
1544 ApiError::BalanceInsufficient(expected, "low".to_string()),
1545 "{asset:?}"
1546 );
1547 assert_eq!(warned, warnings, "{asset:?}: a dropped asset is logged");
1548 }
1549 }
1550
1551 #[test]
1555 fn an_unresolvable_invalid_name_degrades_to_a_rejection_keeping_its_text() {
1556 let indexer = binance_indexer();
1557 for (error, text) in [
1558 (
1559 ApiError::AssetInvalid(AssetNameExchange::new("XYZ"), "bad".to_string()),
1560 "asset XYZ invalid: bad",
1561 ),
1562 (
1563 ApiError::InstrumentInvalid(
1564 InstrumentNameExchange::new("XYZUSDT"),
1565 "bad".to_string(),
1566 ),
1567 "instrument XYZUSDT invalid: bad",
1568 ),
1569 ] {
1570 let (indexed, warned) = count_warnings(|| indexer.api_error(error.clone()));
1571 assert_eq!(indexed, ApiError::OrderRejected(text.to_string()));
1572 assert_eq!(warned, 1, "{text}: a degrade is logged");
1573
1574 let (indexed, warned) =
1575 count_warnings(|| indexer.client_error(ClientError::Api(error)));
1576 assert_eq!(
1577 indexed,
1578 ClientError::Api(ApiError::RequestRejected(text.to_string()))
1579 );
1580 assert_eq!(warned, 1, "{text}: a degrade is logged");
1581 }
1582
1583 let (indexed, warned) = count_warnings(|| {
1585 indexer.api_error(ApiError::InstrumentInvalid(btc_usdt(), "bad".to_string()))
1586 });
1587 assert!(matches!(indexed, ApiError::InstrumentInvalid(_, _)));
1588 assert_eq!(warned, 0);
1589 }
1590
1591 #[test]
1594 fn a_response_whose_rejection_names_an_unresolvable_asset_still_indexes() {
1595 let indexer = binance_indexer();
1596 let rejection = || {
1597 OrderError::Rejected(ApiError::BalanceInsufficient(
1598 Some(AssetNameExchange::new("ETH_USDT")),
1599 "low".to_string(),
1600 ))
1601 };
1602 let indexed = OrderError::Rejected(ApiError::BalanceInsufficient(None, "low".to_string()));
1603
1604 assert_eq!(
1605 indexer.order_state(OrderState::inactive(rejection())),
1606 OrderState::inactive(indexed.clone())
1607 );
1608
1609 let key = OrderKey {
1610 exchange: ExchangeId::BinanceSpot,
1611 instrument: btc_usdt(),
1612 strategy: crate::order::id::StrategyId::new("test"),
1613 cid: crate::order::id::ClientOrderId::random(),
1614 };
1615 let cancel = indexer
1616 .order_response_cancel(OrderResponseCancel {
1617 key,
1618 state: Err(rejection()),
1619 })
1620 .unwrap();
1621 assert_eq!(cancel.state, Err(indexed));
1622 }
1623
1624 #[test]
1627 fn account_event_indexes_a_cash_flow() {
1628 use crate::cash_flow::{CashFlowId, CashFlowKind};
1629 use rust_decimal_macros::dec;
1630
1631 let indexer = binance_indexer();
1632 let time = DateTime::<Utc>::MIN_UTC;
1633 let kind = CashFlowKind::Funding {
1634 rate: Some(dec!(0.0001)),
1635 position_quantity: Some(dec!(-2)),
1636 };
1637 let flow = |asset: &str, instrument: Option<&str>| {
1638 UnindexedAccountEvent::new(
1639 ExchangeId::BinanceSpot,
1640 AccountEventKind::CashFlow(CashFlow::new(
1641 kind.clone(),
1642 AssetNameExchange::new(asset),
1643 dec!(1.5),
1644 instrument.map(InstrumentNameExchange::new),
1645 time,
1646 Some(CashFlowId::new("cf-1")),
1647 )),
1648 )
1649 };
1650 let index_of = |event| match indexer.account_event(event) {
1651 Ok(AccountEvent {
1652 kind: AccountEventKind::CashFlow(flow),
1653 ..
1654 }) => Ok(flow),
1655 Ok(other) => panic!("expected CashFlow, got {other:?}"),
1656 Err(error) => Err(error),
1657 };
1658 let (Ok(usdt), Ok(btc_usdt)) = (
1659 indexer
1660 .map
1661 .find_asset_index(&AssetNameExchange::new("USDT")),
1662 indexer
1663 .map
1664 .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT")),
1665 ) else {
1666 panic!("USDT and BTC_USDT are mapped");
1667 };
1668
1669 assert_eq!(
1670 index_of(flow("USDT", Some("BTC_USDT"))),
1671 Ok(CashFlow::new(
1672 kind.clone(),
1673 usdt,
1674 dec!(1.5),
1675 Some(btc_usdt),
1676 time,
1677 Some(CashFlowId::new("cf-1")),
1678 ))
1679 );
1680 assert_eq!(
1681 index_of(flow("USDT", None)).map(|flow| flow.instrument),
1682 Ok(None)
1683 );
1684 assert!(
1685 index_of(flow("ETH", None)).is_err(),
1686 "an unmapped asset fails"
1687 );
1688 assert!(
1689 index_of(flow("USDT", Some("ETHUSDT"))).is_err(),
1690 "an unmapped instrument fails"
1691 );
1692 }
1693
1694 #[test]
1697 fn account_event_indexes_a_notice() {
1698 use crate::notice::{AccountNotice, NoticeKind};
1699 use rust_decimal_macros::dec;
1700 use smol_str::SmolStr;
1701
1702 let indexer = binance_indexer();
1703 let time = DateTime::<Utc>::MIN_UTC;
1704 let notice = |instrument: Option<&str>| {
1705 UnindexedAccountEvent::new(
1706 ExchangeId::BinanceSpot,
1707 AccountEventKind::Notice(AccountNotice::new(
1708 NoticeKind::MarginCall,
1709 instrument.map(InstrumentNameExchange::new),
1710 time,
1711 Some(dec!(1.1)),
1712 SmolStr::new("MARGIN_CALL"),
1713 )),
1714 )
1715 };
1716 let index_of = |event| match indexer.account_event(event) {
1717 Ok(AccountEvent {
1718 kind: AccountEventKind::Notice(notice),
1719 ..
1720 }) => notice,
1721 other => panic!("expected Notice, got {other:?}"),
1722 };
1723 let Ok(btc_usdt) = indexer
1724 .map
1725 .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT"))
1726 else {
1727 panic!("BTC_USDT is mapped");
1728 };
1729
1730 assert_eq!(
1731 index_of(notice(Some("BTC_USDT"))),
1732 AccountNotice::new(
1733 NoticeKind::MarginCall,
1734 Some(btc_usdt),
1735 time,
1736 Some(dec!(1.1)),
1737 SmolStr::new("MARGIN_CALL"),
1738 )
1739 );
1740 assert_eq!(index_of(notice(None)).instrument, None);
1741 assert_eq!(
1742 index_of(notice(Some("ETHUSDT"))).instrument,
1743 None,
1744 "an unmapped instrument is dropped from the notice, not the notice itself"
1745 );
1746 }
1747
1748 #[test]
1751 fn account_event_indexes_a_margin_debt_update() {
1752 use crate::{
1753 balance::MarginDetails,
1754 margin_debt::{MarginDebtKind, MarginDebtUpdate},
1755 };
1756 use rust_decimal_macros::dec;
1757 use smol_str::SmolStr;
1758
1759 let indexer = binance_indexer();
1760 let time = DateTime::<Utc>::MIN_UTC;
1761 let debt = MarginDetails::new(dec!(0.5), dec!(0.01));
1762 let update = |asset: &str, instrument: Option<&str>| {
1763 UnindexedAccountEvent::new(
1764 ExchangeId::BinanceSpot,
1765 AccountEventKind::MarginDebt(MarginDebtUpdate::new(
1766 AssetNameExchange::new(asset),
1767 instrument.map(InstrumentNameExchange::new),
1768 debt,
1769 MarginDebtKind::Borrow,
1770 SmolStr::new("BORROW"),
1771 time,
1772 )),
1773 )
1774 };
1775 let index_of = |event| match indexer.account_event(event) {
1776 Ok(AccountEvent {
1777 kind: AccountEventKind::MarginDebt(update),
1778 ..
1779 }) => Ok(update),
1780 Ok(other) => panic!("expected MarginDebt, got {other:?}"),
1781 Err(error) => Err(error),
1782 };
1783 let (Ok(btc), Ok(btc_usdt)) = (
1784 indexer.map.find_asset_index(&AssetNameExchange::new("BTC")),
1785 indexer
1786 .map
1787 .find_instrument_index(&InstrumentNameExchange::new("BTC_USDT")),
1788 ) else {
1789 panic!("BTC and BTC_USDT are mapped");
1790 };
1791
1792 assert_eq!(
1793 index_of(update("BTC", Some("BTC_USDT"))),
1794 Ok(MarginDebtUpdate::new(
1795 btc,
1796 Some(btc_usdt),
1797 debt,
1798 MarginDebtKind::Borrow,
1799 SmolStr::new("BORROW"),
1800 time,
1801 ))
1802 );
1803 assert_eq!(
1804 index_of(update("BTC", None)).map(|update| update.instrument),
1805 Ok(None)
1806 );
1807 assert!(
1808 index_of(update("ETH", None)).is_err(),
1809 "an unmapped asset fails"
1810 );
1811 assert!(
1812 index_of(update("BTC", Some("ETHUSDT"))).is_err(),
1813 "an unmapped instrument fails"
1814 );
1815 }
1816}