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