Skip to main content

rustrade_execution/
indexer.rs

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/// Why [`AccountEventIndexer::index_account_event`] did not index an event.
41///
42/// [`AccountEventIndexer::account_event`] folds both causes into one [`IndexError`]; this tells
43/// them apart for a caller that treats them differently.
44#[derive(Debug, Clone, PartialEq)]
45pub enum AccountIndexError {
46    /// A balance snapshot or update named an asset the [`ExecutionInstrumentMap`] does not
47    /// register.
48    ///
49    /// Expected rather than a fault: a venue such as Binance pushes balances for every asset on
50    /// the account, including ones the caller never asked to track (BNB held for fees, say). The
51    /// balance is not tracked. An unregistered asset on any other event is still an
52    /// [`Self::Rejected`] error, because that drops ledger data.
53    UntrackedBalance(AssetNameExchange),
54    /// The event named an exchange, asset or instrument the map does not register, and is lost.
55    Rejected(IndexError),
56}
57
58/// An indexed [`AccountSnapshot`] and the assets whose balances it left out. See
59/// [`AccountEventIndexer::snapshot_partitioned`], whose caller must report `untracked`.
60#[derive(Debug, Clone, PartialEq)]
61#[non_exhaustive]
62pub struct PartitionedSnapshot {
63    /// The snapshot, without the balances of the untracked assets.
64    pub snapshot: AccountSnapshot,
65    /// The assets of the skipped balances, in the order the venue reported them.
66    pub untracked: Vec<AssetNameExchange>,
67}
68
69impl AccountEventIndexer {
70    /// Index `event`, telling a balance for an unregistered asset apart from a lost event.
71    ///
72    /// See [`AccountIndexError`]. [`Self::account_event`] is the same indexing with one error.
73    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                // A balance event reaches no other lookup that can fail on an asset.
86                (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            // Termination reason carries no exchange/asset/instrument keys — pass through verbatim.
126            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    /// Index `snapshot`, failing on any asset the [`ExecutionInstrumentMap`] does not register.
151    ///
152    /// Use [`Self::snapshot_partitioned`] to skip an unregistered balance instead. This is also
153    /// the path of an [`AccountEventKind::Snapshot`] event on an account stream, which stays
154    /// strict: it replaces the engine's whole view of the account, so losing part of it is not
155    /// tolerated.
156    ///
157    /// # Errors
158    ///
159    /// [`IndexError`] for an exchange, asset, instrument or order the map does not register.
160    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    /// Index `snapshot`, skipping the top-level balance of an asset the
179    /// [`ExecutionInstrumentMap`] does not register, and returning those assets beside it.
180    ///
181    /// A venue may report more assets than the caller registered (see
182    /// [`AccountIndexError::UntrackedBalance`]), so such a balance is not a fault. Only top-level
183    /// balances are skipped: an unregistered exchange, instrument, order, or isolated-margin
184    /// base or quote asset still fails the snapshot, because each registered instrument brings
185    /// its own assets and the loss would be a mapping defect.
186    ///
187    /// The caller must report [`PartitionedSnapshot::untracked`], or the drop is silent. An empty
188    /// [`PartitionedSnapshot::snapshot`] balance list beside a non-empty `untracked` usually means
189    /// the registered asset names do not match the venue's; it can also be an account that holds
190    /// only unregistered assets.
191    ///
192    /// # Errors
193    ///
194    /// [`IndexError`] for an exchange, instrument, order or isolated-margin asset the map does
195    /// not register.
196    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                // Per-pair isolated balances are generic over `AssetKey`, so (unlike `position`)
241                // their base/quote asset names must be mapped to indices. A pair whose asset is
242                // unregistered fails the snapshot index, matching top-level-balance behaviour.
243                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    /// Index the per-pair isolated state's `base`/`quote` asset names to indices.
304    ///
305    /// # Errors
306    /// Returns `IndexError` if either the base or quote asset is not registered in the map —
307    /// matching the fail-fast behaviour of top-level [`Self::asset_balance`].
308    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    /// Index a [`FillRecoveryGap`]'s instruments.
322    ///
323    /// # Errors
324    /// Returns `IndexError` if any instrument is not registered in the map. The whole gap fails
325    /// rather than dropping that instrument, which would understate what is missing. A venue names
326    /// only the instruments its stream was opened with, so this fails only for a stream opened
327    /// with an instrument the map does not hold.
328    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    /// Index a [`CashFlow`]'s asset and, when it names one, instrument.
360    ///
361    /// # Errors
362    /// Returns `IndexError` if the asset, or a named instrument, is not registered in the map.
363    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    /// Index a [`MarginDebtUpdate`]'s asset and, when it names one, instrument.
391    ///
392    /// Unlike a notice's, an unknown instrument is an error rather than dropped: an update without
393    /// one is a debt of the account as a whole, which the engine applies to its asset balances, so
394    /// an isolated pair's debt delivered unattributed would overwrite the account's.
395    ///
396    /// # Errors
397    /// Returns `IndexError` if the asset, or a named instrument, is not registered in the map.
398    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    /// Index an [`AccountNotice`]'s instrument, when it names one.
424    ///
425    /// Never fails. A notice naming an instrument that is not in the map is kept, with its
426    /// instrument set to `None` and a `warn`: a notice changes no state, so dropping one about a
427    /// margin call or liquidation would cost more than losing its attribution.
428    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    /// Index an [`InstrumentBalanceUpdate`]'s instrument and asset keys.
468    ///
469    /// # Errors
470    /// Returns `IndexError` if the instrument or the asset is not registered in the map.
471    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    /// Index an [`UnindexedOrderState`] to an [`OrderState`].
561    ///
562    /// Used by `ExecutionManager` to index `open_order` responses. Never fails: an order state
563    /// carries no key, and the error of a failed open is indexed by [`Self::order_error`].
564    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    /// Index an [`UnindexedApiError`] returned for an order request (open or cancel).
579    ///
580    /// Never fails. An error carries the venue's answer to a request, and the caller must
581    /// receive that answer, so a name the [`ExecutionInstrumentMap`] does not hold degrades the
582    /// error instead of discarding it (and the order response or client error carrying it). Each
583    /// degrade logs a `warn!` naming what was dropped:
584    /// - [`ApiError::BalanceInsufficient`] keeps its variant, with the asset `None`.
585    /// - [`ApiError::AssetInvalid`] and [`ApiError::InstrumentInvalid`] have no index to carry,
586    ///   so they become [`ApiError::OrderRejected`] holding their own message, which keeps the
587    ///   name and the venue's text.
588    ///
589    /// For an error from any other request, use [`Self::client_error`], which degrades to
590    /// [`ApiError::RequestRejected`] instead.
591    pub fn api_error(&self, error: UnindexedApiError) -> ApiError {
592        self.index_api_error(error, ApiError::OrderRejected)
593    }
594
595    /// Index an [`UnindexedApiError`], degrading a name the map does not hold as
596    /// [`Self::api_error`] describes. `rejected` builds the rejection that an unresolvable
597    /// [`ApiError::AssetInvalid`] or [`ApiError::InstrumentInvalid`] becomes, which depends on
598    /// whether the request was an order.
599    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    /// Degrade an error naming an asset or instrument the map does not hold to the rejection
656    /// `rejected` builds, carrying the error's own message so neither the name nor the venue's
657    /// text is lost.
658    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    /// Index an order request for this exchange, resolving its instrument to the name this
674    /// exchange knows it by.
675    ///
676    /// # Errors
677    /// Returns [`KeyError`] when the order's exchange or instrument is not in the map, and when
678    /// the instrument is indexed but listed on a different exchange: a request keyed to another
679    /// exchange's instrument is not this exchange's to send.
680    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    /// Index an [`UnindexedOrderError`]. Never fails: see [`Self::api_error`].
721    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    /// Index an [`UnindexedClientError`]. Never fails: like [`Self::api_error`], but an
738    /// unresolvable [`ApiError::AssetInvalid`] or [`ApiError::InstrumentInvalid`] becomes
739    /// [`ApiError::RequestRejected`], since the request need not have been an order.
740    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    /// Index an [`UnindexedAccountStreamInitError`]. Never fails, as [`Self::client_error`].
756    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    /// Index an [`AccountReinitFailure`]. Never fails, as [`Self::client_error`].
769    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    /// Index a [`TradeAmendment`]. A replacement trade is indexed as [`trade`](Self::trade) indexes
781    /// one, so an amendment fails where its trade would.
782    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    /// Index a trade, converting fee asset and computing `fees_quote`.
821    ///
822    /// Computes `fees_quote` based on fee asset relationship to instrument:
823    /// - Fee in quote asset: `fees_quote = Some(fees)`
824    /// - Fee in base asset: `fees_quote = Some(fees * price)`
825    /// - Fee in third-party asset (e.g., BNB): `fees_quote = None`, since the indexer holds no
826    ///   prices. The `rustrade` engine values it when it prices that asset.
827    ///
828    /// # A fee asset that is not in the map
829    /// The trade is kept, with its fee zeroed: `fees` is `0` in the instrument's quote asset and
830    /// `fees_quote` is `None`. A `warn` names the fee's asset and amount. Dropping the trade would
831    /// leave the engine's position different from the venue's, which costs far more than one
832    /// unknown fee. The `warn` is per trade, so an account that pays every fee in an unregistered
833    /// asset logs one per fill until the asset is registered. Two cases reach this:
834    /// - An asset the account pays fees in but no instrument implies, such as BNB on Binance spot
835    ///   with "pay fees with BNB" turned on. Register it with
836    ///   [`IndexedInstrumentsBuilder::add_asset`](rustrade_instrument::index::builder::IndexedInstrumentsBuilder::add_asset)
837    ///   to keep its fee.
838    /// - The `"UNKNOWN"` placeholder some integrations use when the venue gave no fee asset: IBKR
839    ///   `fetch_trades` for an execution IB sent no commission report for, and Binance when a
840    ///   report omits `commissionAsset`.
841    ///
842    /// # Errors
843    /// Returns `IndexError` if the trade's instrument is not in the map.
844    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        // Invariant check: `find_instrument_index` resolves only names built from `instruments`,
867        // so the index is always present. Failing loudly beats indexing a fee against nothing.
868        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                    // Fee is in quote asset — no conversion needed
883                    Some(fees.fees)
884                } else if fee_asset == underlying.base {
885                    // Fee is in base asset — convert using trade price
886                    Some(fees.fees * trade_price)
887                } else {
888                    // Fee is in third-party asset (e.g., BNB) — needs external price
889                    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)] // Test code: panics on bad input are acceptable
929mod 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        // A skipped balance does not hide a failure elsewhere in the same snapshot.
991        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    /// An order naming an instrument listed on another exchange does not resolve, though the plain
1012    /// instrument lookup would find it.
1013    #[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        // A balance on an exchange the map does not hold fails on the exchange, not the asset.
1130        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    /// An isolated pair's balance snapshot indexes both its instrument and its asset; either one
1139    /// unknown rejects it, an unknown asset included, since the balance is the pair's.
1140    #[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    /// `StreamTerminated` carries no asset/instrument keys, so it must index by mapping only the
1190    /// exchange and passing the reason through verbatim.
1191    #[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    /// A failed re-init never fails to index: its client error is indexed as
1214    /// [`AccountEventIndexer::client_error`] indexes one, and an index error passes through.
1215    #[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    /// A fill recovery give-up indexes each instrument it names, and fails as a whole when one
1253    /// is not in the map rather than understating what is missing. `AllInstruments` names none.
1254    #[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    /// A trade amendment indexes its instrument, and a corrected one its replacement as a trade
1313    /// is indexed. An unmapped instrument fails it.
1314    #[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    /// A fill is never dropped for its fee: an unmapped fee asset, or the `"UNKNOWN"` placeholder,
1410    /// keeps the trade with its fee zeroed in quote and `fees_quote: None`, with one `warn`. A fee
1411    /// asset registered with `add_asset` indexes as a third-party fee.
1412    #[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    /// Run `index`, returning its value and the number of `WARN` events it emitted on this
1505    /// thread. A degrade is reported only by its log line, so asserting on the value alone could
1506    /// not tell a logged degrade from a silent one.
1507    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            // An instrument name, as Binance's error mapping once put there.
1536            (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    /// An invalid asset or instrument the map does not hold has no index to carry, so it becomes
1555    /// a rejection keeping its own message: `OrderRejected` for an order, `RequestRejected` for
1556    /// any other request.
1557    #[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        // A name the map holds still indexes, and nothing is logged.
1587        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    /// A response whose rejection names something the map does not hold still indexes, so the
1595    /// order is settled rather than left in flight.
1596    #[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    /// A cash flow indexes its asset and, when it names one, its instrument, keeps its period,
1628    /// and fails when either key is not in the map.
1629    #[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    /// A notice indexes its instrument when it names a mapped one, and is kept unattributed,
1703    /// rather than dropped, when it names one that is not mapped.
1704    #[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    /// A margin debt update indexes its asset and instrument, and one naming an unmapped
1757    /// instrument fails rather than becoming the account's debt.
1758    #[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}