liquidator: when rebalancing allow multi-hop jupiter routes

and if that doesn't fit into a transaction, try a direct route with USDC
or SOL.

(cherry picked from commit db8f5ae30d)

bin/liquidator/src/liquidate.rs

@@ -1,6 +1,7 @@
 use std::collections::HashSet;
 use std::time::Duration;
 
+use itertools::Itertools;
 use mango_v4::accounts_zerocopy::KeyedAccountSharedData;
 use mango_v4::health::{HealthCache, HealthType};
 use mango_v4::state::{
@@ -91,17 +92,15 @@ struct LiquidateHelper<'a> {
 impl<'a> LiquidateHelper<'a> {
     async fn serum3_close_orders(&self) -> anyhow::Result<Option<Signature>> {
         // look for any open serum orders or settleable balances
-        let serum_oos: anyhow::Result<Vec<_>> = stream::iter(self.liqee.active_serum3_orders())
-            .then(|orders| async {
-                let open_orders_account = self
-                    .account_fetcher
-                    .fetch_raw_account(&orders.open_orders)
-                    .await?;
+        let serum_oos: anyhow::Result<Vec<_>> = self
+            .liqee
+            .active_serum3_orders()
+            .map(|orders| {
+                let open_orders_account = self.account_fetcher.fetch_raw(&orders.open_orders)?;
                 let open_orders = mango_v4::serum3_cpi::load_open_orders(&open_orders_account)?;
                 Ok((*orders, *open_orders))
             })
-            .try_collect()
-            .await;
+            .try_collect();
         let serum_force_cancels = serum_oos?
             .into_iter()
             .filter_map(|(orders, open_orders)| {

bin/liquidator/src/rebalance.rs

@@ -1,15 +1,17 @@
+use itertools::Itertools;
 use mango_v4::accounts_zerocopy::KeyedAccountSharedData;
 use mango_v4::state::{
     Bank, BookSide, PlaceOrderType, Side, TokenIndex, TokenPosition, QUOTE_TOKEN_INDEX,
 };
 use mango_v4_client::{
-    chain_data, perp_pnl, AccountFetcher, AnyhowWrap, JupiterSwapMode, MangoClient, TokenContext,
+    chain_data, jupiter::QueryRoute, perp_pnl, AnyhowWrap, JupiterSwapMode, MangoClient,
+    TokenContext, TransactionBuilder,
 };
 
 use {fixed::types::I80F48, solana_sdk::pubkey::Pubkey};
 
-use futures::{stream, StreamExt, TryStreamExt};
 use solana_sdk::signature::Signature;
+use std::str::FromStr;
 use std::sync::Arc;
 use std::{collections::HashMap, time::Duration};
 
@@ -32,14 +34,14 @@ struct TokenState {
 }
 
 impl TokenState {
-    async fn new_position(
+    fn new_position(
         token: &TokenContext,
         position: &TokenPosition,
         account_fetcher: &chain_data::AccountFetcher,
     ) -> anyhow::Result<Self> {
         let bank = Self::bank(token, account_fetcher)?;
         Ok(Self {
-            price: Self::fetch_price(token, &bank, account_fetcher).await?,
+            price: Self::fetch_price(token, &bank, account_fetcher)?,
             native_position: position.native(&bank),
             in_use: position.is_in_use(),
         })
@@ -52,14 +54,12 @@ impl TokenState {
         account_fetcher.fetch::<Bank>(&token.mint_info.first_bank())
     }
 
-    async fn fetch_price(
+    fn fetch_price(
         token: &TokenContext,
         bank: &Bank,
         account_fetcher: &chain_data::AccountFetcher,
     ) -> anyhow::Result<I80F48> {
-        let oracle = account_fetcher
-            .fetch_raw_account(&token.mint_info.oracle)
-            .await?;
+        let oracle = account_fetcher.fetch_raw(&token.mint_info.oracle)?;
         bank.oracle_price(
             &KeyedAccountSharedData::new(token.mint_info.oracle, oracle.into()),
             None,
@@ -68,6 +68,13 @@ impl TokenState {
     }
 }
 
+#[derive(Clone)]
+struct WrappedJupRoute {
+    input_mint: Pubkey,
+    output_mint: Pubkey,
+    route: QueryRoute,
+}
+
 pub struct Rebalancer {
     pub mango_client: Arc<MangoClient>,
     pub account_fetcher: Arc<chain_data::AccountFetcher>,
@@ -105,6 +112,188 @@ impl Rebalancer {
         Ok(true)
     }
 
+    /// Wrapping client.jupiter_route() in a way that preserves the in/out mints
+    async fn jupiter_route(
+        &self,
+        input_mint: Pubkey,
+        output_mint: Pubkey,
+        amount: u64,
+        only_direct_routes: bool,
+    ) -> anyhow::Result<WrappedJupRoute> {
+        let route = self
+            .mango_client
+            .jupiter_route(
+                input_mint,
+                output_mint,
+                amount,
+                self.config.slippage_bps,
+                JupiterSwapMode::ExactIn,
+                only_direct_routes,
+            )
+            .await?;
+        Ok(WrappedJupRoute {
+            input_mint,
+            output_mint,
+            route,
+        })
+    }
+
+    /// Grab three possible routes:
+    /// 1. USDC -> output (complex routes)
+    /// 2. USDC -> output (direct route only)
+    /// 3. SOL -> output (direct route only)
+    /// Use 1. if it fits into a tx. Otherwise use the better of 2./3.
+    async fn token_swap_buy(
+        &self,
+        output_mint: Pubkey,
+        in_amount_quote: u64,
+    ) -> anyhow::Result<(Signature, WrappedJupRoute)> {
+        let quote_token = self.mango_client.context.token(QUOTE_TOKEN_INDEX);
+        let sol_token = self.mango_client.context.token(
+            *self
+                .mango_client
+                .context
+                .token_indexes_by_name
+                .get("SOL") // TODO: better use mint
+                .unwrap(),
+        );
+
+        let full_route_job = self.jupiter_route(
+            quote_token.mint_info.mint,
+            output_mint,
+            in_amount_quote,
+            false,
+        );
+        let direct_quote_route_job = self.jupiter_route(
+            quote_token.mint_info.mint,
+            output_mint,
+            in_amount_quote,
+            true,
+        );
+
+        // For the SOL -> output route we need to adjust the in amount by the SOL price
+        let sol_bank = TokenState::bank(sol_token, &self.account_fetcher)?;
+        let sol_price = TokenState::fetch_price(sol_token, &sol_bank, &self.account_fetcher)?;
+        let in_amount_sol = (I80F48::from(in_amount_quote) / sol_price)
+            .ceil()
+            .to_num::<u64>();
+        let direct_sol_route_job =
+            self.jupiter_route(sol_token.mint_info.mint, output_mint, in_amount_sol, true);
+
+        let (full_route, direct_quote_route, direct_sol_route) =
+            tokio::join!(full_route_job, direct_quote_route_job, direct_sol_route_job);
+        let alternatives = [direct_quote_route, direct_sol_route]
+            .into_iter()
+            .filter_map(|v| v.ok())
+            .collect_vec();
+
+        let (tx_builder, route) = self
+            .determine_best_jupiter_tx(
+                // If the best_route couldn't be fetched, something is wrong
+                &full_route?,
+                &alternatives,
+            )
+            .await?;
+        let sig = tx_builder
+            .send_and_confirm(&self.mango_client.client)
+            .await?;
+        Ok((sig, route))
+    }
+
+    /// Grab three possible routes:
+    /// 1. input -> USDC (complex routes)
+    /// 2. input -> USDC (direct route only)
+    /// 3. input -> SOL (direct route only)
+    /// Use 1. if it fits into a tx. Otherwise use the better of 2./3.
+    async fn token_swap_sell(
+        &self,
+        input_mint: Pubkey,
+        in_amount: u64,
+    ) -> anyhow::Result<(Signature, WrappedJupRoute)> {
+        let quote_token = self.mango_client.context.token(QUOTE_TOKEN_INDEX);
+        let sol_token = self.mango_client.context.token(
+            *self
+                .mango_client
+                .context
+                .token_indexes_by_name
+                .get("SOL") // TODO: better use mint
+                .unwrap(),
+        );
+
+        let full_route_job =
+            self.jupiter_route(input_mint, quote_token.mint_info.mint, in_amount, false);
+        let direct_quote_route_job =
+            self.jupiter_route(input_mint, quote_token.mint_info.mint, in_amount, true);
+        let direct_sol_route_job =
+            self.jupiter_route(input_mint, sol_token.mint_info.mint, in_amount, true);
+        let (full_route, direct_quote_route, direct_sol_route) =
+            tokio::join!(full_route_job, direct_quote_route_job, direct_sol_route_job);
+        let alternatives = [direct_quote_route, direct_sol_route]
+            .into_iter()
+            .filter_map(|v| v.ok())
+            .collect_vec();
+
+        let (tx_builder, route) = self
+            .determine_best_jupiter_tx(
+                // If the best_route couldn't be fetched, something is wrong
+                &full_route?,
+                &alternatives,
+            )
+            .await?;
+
+        let sig = tx_builder
+            .send_and_confirm(&self.mango_client.client)
+            .await?;
+        Ok((sig, route))
+    }
+
+    async fn determine_best_jupiter_tx(
+        &self,
+        full: &WrappedJupRoute,
+        alternatives: &[WrappedJupRoute],
+    ) -> anyhow::Result<(TransactionBuilder, WrappedJupRoute)> {
+        let builder = self
+            .mango_client
+            .prepare_jupiter_swap_transaction(full.input_mint, full.output_mint, &full.route)
+            .await?;
+        if builder.transaction_size_ok()? {
+            return Ok((builder, full.clone()));
+        }
+        log::trace!(
+            "full route from {} to {} does not fit in a tx, market_info.label {}",
+            full.input_mint,
+            full.output_mint,
+            full.route
+                .market_infos
+                .first()
+                .map(|v| v.label.clone())
+                .unwrap_or_else(|| "no market_info".into())
+        );
+
+        if alternatives.is_empty() {
+            anyhow::bail!(
+                "no alternative routes from {} to {}",
+                full.input_mint,
+                full.output_mint
+            );
+        }
+
+        let best = alternatives
+            .iter()
+            .min_by(|a, b| {
+                a.route
+                    .price_impact_pct
+                    .partial_cmp(&b.route.price_impact_pct)
+                    .unwrap()
+            })
+            .unwrap();
+        let builder = self
+            .mango_client
+            .prepare_jupiter_swap_transaction(best.input_mint, best.output_mint, &best.route)
+            .await?;
+        Ok((builder, best.clone()))
+    }
+
     async fn rebalance_tokens(&self) -> anyhow::Result<()> {
         let account = self
             .account_fetcher
@@ -113,18 +302,16 @@ impl Rebalancer {
         // TODO: configurable?
         let quote_token = self.mango_client.context.token(QUOTE_TOKEN_INDEX);
 
-        let tokens: anyhow::Result<HashMap<TokenIndex, TokenState>> =
-            stream::iter(account.active_token_positions())
-                .then(|token_position| async {
-                    let token = self.mango_client.context.token(token_position.token_index);
-                    Ok((
-                        token.token_index,
-                        TokenState::new_position(token, token_position, &self.account_fetcher)
-                            .await?,
-                    ))
-                })
-                .try_collect()
-                .await;
+        let tokens: anyhow::Result<HashMap<TokenIndex, TokenState>> = account
+            .active_token_positions()
+            .map(|token_position| {
+                let token = self.mango_client.context.token(token_position.token_index);
+                Ok((
+                    token.token_index,
+                    TokenState::new_position(token, token_position, &self.account_fetcher)?,
+                ))
+            })
+            .try_collect();
         let tokens = tokens?;
         log::trace!("account tokens: {:?}", tokens);
 
@@ -134,7 +321,6 @@ impl Rebalancer {
                 continue;
             }
             let token_mint = token.mint_info.mint;
-            let quote_mint = quote_token.mint_info.mint;
 
             // It's not always possible to bring the native balance to 0 through swaps:
             // Consider a price <1. You need to sell a bunch of tokens to get 1 USDC native and
@@ -157,23 +343,21 @@ impl Rebalancer {
                 let input_amount = buy_amount
                     * token_state.price
                     * I80F48::from_num(self.config.borrow_settle_excess);
-                let txsig = self
-                    .mango_client
-                    .jupiter_swap(
-                        quote_mint,
-                        token_mint,
-                        input_amount.to_num::<u64>(),
-                        self.config.slippage_bps,
-                        JupiterSwapMode::ExactIn,
-                        false,
-                    )
+                let (txsig, route) = self
+                    .token_swap_buy(token_mint, input_amount.to_num())
                     .await?;
+                let in_token = self
+                    .mango_client
+                    .context
+                    .token_by_mint(&route.input_mint)
+                    .unwrap();
                 log::info!(
-                    "bought {} {} for {} in tx {}",
-                    token.native_to_ui(buy_amount),
+                    "bought {} {} for {} {} in tx {}",
+                    token.native_to_ui(I80F48::from_str(&route.route.out_amount).unwrap()),
                     token.name,
-                    quote_token.name,
-                    txsig
+                    in_token.native_to_ui(I80F48::from_str(&route.route.in_amount).unwrap()),
+                    in_token.name,
+                    txsig,
                 );
                 if !self.refresh_mango_account_after_tx(txsig).await? {
                     return Ok(());
@@ -190,23 +374,21 @@ impl Rebalancer {
 
             if amount > dust_threshold {
                 // Sell
-                let txsig = self
-                    .mango_client
-                    .jupiter_swap(
-                        token_mint,
-                        quote_mint,
-                        amount.to_num::<u64>(),
-                        self.config.slippage_bps,
-                        JupiterSwapMode::ExactIn,
-                        false,
-                    )
+                let (txsig, route) = self
+                    .token_swap_sell(token_mint, amount.to_num::<u64>())
                     .await?;
+                let out_token = self
+                    .mango_client
+                    .context
+                    .token_by_mint(&route.output_mint)
+                    .unwrap();
                 log::info!(
-                    "sold {} {} for {} in tx {}",
-                    token.native_to_ui(amount),
+                    "sold {} {} for {} {} in tx {}",
+                    token.native_to_ui(I80F48::from_str(&route.route.in_amount).unwrap()),
                     token.name,
-                    quote_token.name,
-                    txsig
+                    out_token.native_to_ui(I80F48::from_str(&route.route.out_amount).unwrap()),
+                    out_token.name,
+                    txsig,
                 );
                 if !self.refresh_mango_account_after_tx(txsig).await? {
                     return Ok(());

lib/client/src/client.rs

@@ -17,6 +17,7 @@ use futures::{stream, StreamExt, TryStreamExt};
 use itertools::Itertools;
 
 use mango_v4::accounts_ix::{Serum3OrderType, Serum3SelfTradeBehavior, Serum3Side};
+use mango_v4::accounts_zerocopy::KeyedAccountSharedData;
 use mango_v4::state::{
     Bank, Group, MangoAccountValue, PerpMarketIndex, PlaceOrderType, SelfTradeBehavior,
     Serum3MarketIndex, Side, TokenIndex, INSURANCE_TOKEN_INDEX,
@@ -419,6 +420,20 @@ impl MangoClient {
         self.send_and_confirm_owner_tx(ixs).await
     }
 
+    pub async fn bank_oracle_price(&self, token_index: TokenIndex) -> anyhow::Result<I80F48> {
+        let bank = self.first_bank(token_index).await?;
+        let mint_info = self.context.mint_info(token_index);
+        let oracle = self
+            .account_fetcher
+            .fetch_raw_account(&mint_info.oracle)
+            .await?;
+        let price = bank.oracle_price(
+            &KeyedAccountSharedData::new(mint_info.oracle, oracle.into()),
+            None,
+        )?;
+        Ok(price)
+    }
+
     pub async fn get_oracle_price(
         &self,
         token_name: &str,

lib/client/src/lib.rs

@@ -11,7 +11,7 @@ mod client;
 mod context;
 mod gpa;
 pub mod health_cache;
-mod jupiter;
+pub mod jupiter;
 pub mod perp_pnl;
 pub mod snapshot_source;
 mod util;