fix(funding): resolve USDT conversion route per asset (handles fiat/BRL) (#87)

The saga assumed every non-USDT spot asset X has a direct {X}USDT SELL pair,
so a BRL spot balance produced BRLUSDT (-1121 Invalid symbol) and the quote
500'd. Resolve the route symbol-agnostically (ADR-0023):
- {X}USDT trading  → SELL X (base qty)        [existing]
- else USDT{X} trading → BUY USDT (quote qty)  [new: fiat/quote-asset, e.g. BRL via USDTBRL]
- else skip the asset.

Adds ExchangePort::spot_symbol_is_trading (Binance: /api/v3/exchangeInfo,
-1121 → false), StubExchange BUY balance handling + set_trading_symbols, and
test inverse_pair_brl_to_usdt (R$1000 @ 5.0 BRL/USDT → ~199.8 USDT → REFRESHED).
All money uses Decimal. 6 funding tests pass vs Postgres 16.

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>

Author: ldamasio

robson-connectors/src/binance_rest.rs

@@ -626,6 +626,21 @@ impl BinanceRestClient {
             .map_err(|e| BinanceRestError::ParseError(format!("Invalid spot price: {}", e)))
     }
 
+    pub async fn spot_symbol_is_trading(&self, symbol: &str) -> Result<bool, BinanceRestError> {
+        let body = match self
+            .get_public("/api/v3/exchangeInfo", vec![("symbol", symbol.to_string())])
+            .await
+        {
+            Ok(body) => body,
+            Err(BinanceRestError::ApiError { code: -1121, .. }) => return Ok(false),
+            Err(error) => return Err(error),
+        };
+        let response: BinanceSpotExchangeInfoResponse =
+            serde_json::from_str(&body).map_err(|e| BinanceRestError::ParseError(e.to_string()))?;
+
+        Ok(response.symbols.into_iter().any(|info| info.status == "TRADING"))
+    }
+
     pub async fn place_spot_market_order(
         &self,
         symbol: &str,
@@ -882,6 +897,16 @@ pub struct BinanceSpotBalance {
     pub locked: Decimal,
 }
 
+#[derive(Debug, Clone, Deserialize)]
+struct BinanceSpotExchangeInfoResponse {
+    symbols: Vec<BinanceSpotSymbolInfo>,
+}
+
+#[derive(Debug, Clone, Deserialize)]
+struct BinanceSpotSymbolInfo {
+    status: String,
+}
+
 #[derive(Debug, Clone, Deserialize)]
 #[serde(rename_all = "camelCase")]
 pub struct BinanceSpotOrderResponse {

robson-exec/src/ports.rs

@@ -177,6 +177,11 @@ pub trait ExchangePort: Send + Sync {
 
     async fn get_spot_price(&self, symbol: &str) -> Result<Price, ExecError>;
 
+    async fn spot_symbol_is_trading(&self, symbol: &str) -> Result<bool, ExecError> {
+        let _ = symbol;
+        Err(ExecError::Exchange("spot symbol trading lookup is not implemented".to_string()))
+    }
+
     async fn place_spot_market_order(
         &self,
         request: SpotOrderRequest,

robson-exec/src/stub.rs

@@ -3,7 +3,10 @@
 //! These implementations simulate exchange and market data behavior
 //! without making real API calls.
 
-use std::{collections::HashMap, sync::RwLock};
+use std::{
+    collections::{HashMap, HashSet},
+    sync::RwLock,
+};
 
 use async_trait::async_trait;
 use chrono::{DateTime, Utc};
@@ -52,6 +55,7 @@ pub struct StubExchange {
     spot_balances: RwLock<HashMap<String, SpotBalance>>,
     spot_orders: RwLock<HashMap<String, SpotOrder>>,
     transfers: RwLock<Vec<Transfer>>,
+    trading_symbols: RwLock<Option<HashSet<String>>>,
     spot_order_calls: RwLock<u64>,
     transfer_calls: RwLock<u64>,
 }
@@ -76,14 +80,15 @@ impl StubExchange {
             spot_balances: RwLock::new(HashMap::new()),
             spot_orders: RwLock::new(HashMap::new()),
             transfers: RwLock::new(Vec::new()),
+            trading_symbols: RwLock::new(None),
             spot_order_calls: RwLock::new(0),
             transfer_calls: RwLock::new(0),
         }
     }
 
     /// Create a stub exchange with a specific futures balance.
     pub fn with_balance(default_price: Decimal, balance: Decimal) -> Self {
-        let mut exchange = Self::new(default_price);
+        let exchange = Self::new(default_price);
         *exchange.futures_balance.write().unwrap() = balance;
         exchange
     }
@@ -105,6 +110,11 @@ impl StubExchange {
         prices.insert(symbol.to_string(), price);
     }
 
+    pub fn set_trading_symbols(&self, symbols: &[&str]) {
+        *self.trading_symbols.write().unwrap() =
+            Some(symbols.iter().map(|symbol| (*symbol).to_string()).collect());
+    }
+
     /// Get price for a symbol (or default).
     pub fn get_price_decimal(&self, symbol: &str) -> Decimal {
         let prices = self.prices.read().unwrap();
@@ -322,6 +332,16 @@ impl ExchangePort for StubExchange {
         Ok(Price::new(self.get_price_decimal(symbol)).unwrap())
     }
 
+    async fn spot_symbol_is_trading(&self, symbol: &str) -> Result<bool, ExecError> {
+        if self.should_fail() {
+            return Err(ExecError::Exchange("Simulated symbol lookup failure".to_string()));
+        }
+        if let Some(symbols) = self.trading_symbols.read().unwrap().as_ref() {
+            return Ok(symbols.contains(symbol));
+        }
+        Ok(self.prices.read().unwrap().contains_key(symbol))
+    }
+
     async fn place_spot_market_order(
         &self,
         request: SpotOrderRequest,
@@ -340,7 +360,10 @@ impl ExchangePort for StubExchange {
             SpotOrderSide::Sell => base_qty * price,
             SpotOrderSide::Buy => request.quantity,
         };
-        let fee = quote_qty * self.fee_rate;
+        let fee = match request.side {
+            SpotOrderSide::Sell => quote_qty * self.fee_rate,
+            SpotOrderSide::Buy => base_qty * self.fee_rate,
+        };
         let asset = request.symbol.trim_end_matches("USDT");
 
         if request.side == SpotOrderSide::Sell {
@@ -357,6 +380,28 @@ impl ExchangePort for StubExchange {
                 locked: Decimal::ZERO,
             });
             usdt.free += quote_qty - fee;
+        } else if request.side == SpotOrderSide::Buy
+            && request.quantity_kind == SpotOrderQuantity::Quote
+        {
+            let quote_asset = request.symbol.strip_prefix("USDT").ok_or_else(|| {
+                ExecError::Exchange(format!(
+                    "Stub spot BUY quote order requires an inverse USDT pair, got {}",
+                    request.symbol
+                ))
+            })?;
+            let mut balances = self.spot_balances.write().unwrap();
+            let quote = balances.entry(quote_asset.to_string()).or_insert_with(|| SpotBalance {
+                asset: quote_asset.to_string(),
+                free: Decimal::ZERO,
+                locked: Decimal::ZERO,
+            });
+            quote.free = (quote.free - request.quantity).max(Decimal::ZERO);
+            let usdt = balances.entry("USDT".to_string()).or_insert_with(|| SpotBalance {
+                asset: "USDT".to_string(),
+                free: Decimal::ZERO,
+                locked: Decimal::ZERO,
+            });
+            usdt.free += base_qty - fee;
         }
 
         let order = SpotOrder {

robsond/src/binance_exchange.rs

@@ -337,6 +337,10 @@ impl ExchangePort for BinanceExchangeAdapter {
         self.client.get_spot_price(symbol).await.map_err(Self::map_error)
     }
 
+    async fn spot_symbol_is_trading(&self, symbol: &str) -> Result<bool, ExecError> {
+        self.client.spot_symbol_is_trading(symbol).await.map_err(Self::map_error)
+    }
+
     async fn place_spot_market_order(
         &self,
         request: SpotOrderRequest,

robsond/src/funding/saga.rs

@@ -80,9 +80,15 @@ impl<E: ExchangePort + 'static, S: Store + 'static> FundingService<E, S> {
             if balance.asset == "USDT" || balance.free <= Decimal::ZERO {
                 continue;
             }
-            let symbol = format!("{}USDT", balance.asset);
-            let price = self.exchange.get_spot_price(&symbol).await?;
-            let gross = balance.free * price.as_decimal();
+            let Some(route) = self.resolve_usdt_route(&balance.asset).await? else {
+                tracing::debug!(asset = %balance.asset, "Skipping spot asset without USDT route");
+                continue;
+            };
+            let price = self.exchange.get_spot_price(&route.symbol).await?;
+            let gross = match route.side {
+                SpotOrderSide::Sell => balance.free * price.as_decimal(),
+                SpotOrderSide::Buy => balance.free / price.as_decimal(),
+            };
             if gross < self.config.dust_usdt {
                 continue;
             }
@@ -97,7 +103,7 @@ impl<E: ExchangePort + 'static, S: Store + 'static> FundingService<E, S> {
                 asset: balance.asset,
                 qty: balance.free,
                 est_usdt,
-                symbol,
+                symbol: route.symbol,
             });
         }
 
@@ -152,15 +158,16 @@ impl<E: ExchangePort + 'static, S: Store + 'static> FundingService<E, S> {
         if view.state == FundingState::Converting.as_str() {
             for item in &quote.items {
                 let client_order_id = spot_client_order_id(quote_id, &item.asset);
+                let route = route_from_quote_item(&item.asset, &item.symbol)?;
                 let order =
                     match self.exchange.get_spot_order(&item.symbol, &client_order_id).await? {
                         Some(order) if order.status == "FILLED" => order,
                         _ => {
                             self.exchange
                                 .place_spot_market_order(SpotOrderRequest {
                                     symbol: item.symbol.clone(),
-                                    side: SpotOrderSide::Sell,
-                                    quantity_kind: SpotOrderQuantity::Base,
+                                    side: route.side,
+                                    quantity_kind: route.quantity_kind,
                                     quantity: item.qty,
                                     client_order_id: client_order_id.clone(),
                                 })
@@ -174,14 +181,21 @@ impl<E: ExchangePort + 'static, S: Store + 'static> FundingService<E, S> {
                         item.asset
                     )));
                 }
+                let usdt_out = match route.side {
+                    SpotOrderSide::Sell => order.cummulative_quote_qty,
+                    SpotOrderSide::Buy if order.fee_asset == "USDT" => {
+                        (order.executed_qty - order.fee).max(Decimal::ZERO)
+                    },
+                    SpotOrderSide::Buy => order.executed_qty,
+                };
                 self.append_and_project(
                     quote_id,
                     FundingState::Converting,
                     "ConversionExecuted",
                     json!({
                         "asset": item.asset,
                         "qty": order.executed_qty,
-                        "usdt_out": order.cummulative_quote_qty,
+                        "usdt_out": usdt_out,
                         "client_order_id": client_order_id,
                     }),
                 )
@@ -338,6 +352,28 @@ impl<E: ExchangePort + 'static, S: Store + 'static> FundingService<E, S> {
             .unwrap_or(Decimal::ZERO))
     }
 
+    async fn resolve_usdt_route(&self, asset: &str) -> DaemonResult<Option<SpotConversionRoute>> {
+        let direct = format!("{asset}USDT");
+        if self.exchange.spot_symbol_is_trading(&direct).await? {
+            return Ok(Some(SpotConversionRoute {
+                symbol: direct,
+                side: SpotOrderSide::Sell,
+                quantity_kind: SpotOrderQuantity::Base,
+            }));
+        }
+
+        let inverse = format!("USDT{asset}");
+        if self.exchange.spot_symbol_is_trading(&inverse).await? {
+            return Ok(Some(SpotConversionRoute {
+                symbol: inverse,
+                side: SpotOrderSide::Buy,
+                quantity_kind: SpotOrderQuantity::Quote,
+            }));
+        }
+
+        Ok(None)
+    }
+
     async fn fail(&self, saga_id: Uuid, reason: &str) -> DaemonResult<()> {
         self.append_and_project(
             saga_id,
@@ -480,3 +516,31 @@ fn spot_client_order_id(saga_id: Uuid, asset: &str) -> String {
 fn transfer_client_key(saga_id: Uuid) -> String {
     format!("rbx-fund-transfer-{}", saga_id.simple())
 }
+
+#[cfg(feature = "postgres")]
+struct SpotConversionRoute {
+    symbol: String,
+    side: SpotOrderSide,
+    quantity_kind: SpotOrderQuantity,
+}
+
+#[cfg(feature = "postgres")]
+fn route_from_quote_item(asset: &str, symbol: &str) -> DaemonResult<SpotConversionRoute> {
+    if symbol == format!("{asset}USDT") {
+        return Ok(SpotConversionRoute {
+            symbol: symbol.to_string(),
+            side: SpotOrderSide::Sell,
+            quantity_kind: SpotOrderQuantity::Base,
+        });
+    }
+
+    if symbol == format!("USDT{asset}") {
+        return Ok(SpotConversionRoute {
+            symbol: symbol.to_string(),
+            side: SpotOrderSide::Buy,
+            quantity_kind: SpotOrderQuantity::Quote,
+        });
+    }
+
+    Err(DaemonError::Config(format!("invalid_funding_route:{asset}:{symbol}")))
+}

robsond/src/funding/tests.rs

@@ -152,6 +152,10 @@ impl ExchangePort for PostConvertedCrashExchange {
         self.inner.get_spot_price(symbol).await
     }
 
+    async fn spot_symbol_is_trading(&self, symbol: &str) -> Result<bool, ExecError> {
+        self.inner.spot_symbol_is_trading(symbol).await
+    }
+
     async fn place_spot_market_order(
         &self,
         request: SpotOrderRequest,
@@ -239,6 +243,36 @@ async fn happy_path(pool: PgPool) -> anyhow::Result<()> {
     Ok(())
 }
 
+#[sqlx::test(migrations = "../migrations")]
+#[ignore = "Requires DATABASE_URL"]
+async fn inverse_pair_brl_to_usdt(pool: PgPool) -> anyhow::Result<()> {
+    let exchange = Arc::new(StubExchange::new(dec!(50000)));
+    exchange.set_futures_balance(dec!(10000));
+    exchange.set_spot_balance("BRL", dec!(1000), Decimal::ZERO);
+    exchange.set_price("USDTBRL", dec!(5.0));
+    exchange.set_trading_symbols(&["USDTBRL"]);
+    let harness = test_service(pool, exchange, FundingConfig::default());
+
+    let quote = harness.service.quote().await?;
+
+    assert_eq!(quote.items.len(), 1);
+    assert_eq!(quote.items[0].asset, "BRL");
+    assert_eq!(quote.items[0].symbol, "USDTBRL");
+    assert_eq!(quote.items[0].qty, dec!(1000));
+    assert_eq!(quote.items[0].est_usdt, dec!(199.4000));
+
+    let response = harness.service.execute(quote.quote_id, "inverse-brl").await?;
+
+    assert_eq!(response.state, FundingState::Refreshed.as_str());
+    assert_eq!(harness.exchange.spot_order_call_count(), 1);
+    assert_eq!(harness.exchange.transfer_call_count(), 1);
+
+    let capital = harness.service.refresh_capital().await?;
+    assert_eq!(capital, dec!(10199.800));
+
+    Ok(())
+}
+
 #[sqlx::test(migrations = "../migrations")]
 #[ignore = "Requires DATABASE_URL"]
 async fn idempotent_convert_no_double_execute(pool: PgPool) -> anyhow::Result<()> {