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