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