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