Improved Priority bidding strategy

crates/uniswapx-rs/Cargo.toml

@@ -10,4 +10,5 @@ alloy-primitives = "0.8.20"
 alloy-sol-types = "0.8.20"
 alloy-dyn-abi = "0.8.20"
 anyhow = "1.0.70"
-hex = "0.4.3"
+hex = "0.4.3"
+serde = "1.0.168"

crates/uniswapx-rs/src/order.rs

@@ -6,6 +6,7 @@ use alloy_primitives::I256;
 use alloy_primitives::U256;
 use alloy_sol_types::sol;
 use anyhow::Result;
+use serde::Serialize;
 
 use crate::sol_math::MulDiv;
 
@@ -151,6 +152,14 @@ pub enum Order {
     V3DutchOrder(V3DutchOrder),
 }
 
+#[derive(Serialize, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum TradeType {
+    #[serde(rename = "exactIn")]
+    ExactIn,
+    #[serde(rename = "exactOut")]
+    ExactOut,
+}
+
 impl Order {
     pub fn encode(&self) -> Vec<u8> {
         match self {
@@ -159,6 +168,39 @@ impl Order {
             Order::V3DutchOrder(order) => order.encode_inner(),
         }
     }
+
+    pub fn trade_type(&self) -> TradeType {
+        match self {
+            Order::V2DutchOrder(order) => {
+                if order.baseOutputs.iter().any(|o| o.startAmount == o.endAmount) {
+                    TradeType::ExactOut
+                } else {
+                    TradeType::ExactIn
+                }
+            }
+            Order::PriorityOrder(order) => {
+                if order.outputs.iter().any(|o| o.mpsPerPriorityFeeWei == U256::from(0)) {
+                    TradeType::ExactOut
+                } else {
+                    TradeType::ExactIn
+                }
+            }
+            Order::V3DutchOrder(order) => {
+                if order.baseOutputs.iter().any(
+                    |o| o.curve.relativeAmounts.len() == 0 ||
+                    o.curve.relativeAmounts.iter().all(|&x| x.eq(&I256::ZERO))
+                ) {
+                    TradeType::ExactOut
+                } else {
+                    TradeType::ExactIn
+                }
+            }
+        }
+    }
+
+    pub fn is_exact_output(&self) -> bool {
+        matches!(self.trade_type(), TradeType::ExactOut)
+    }
 }
 
 #[derive(Debug, Clone)]
@@ -268,7 +310,7 @@ impl PriorityOrder {
         PriorityOrder::abi_encode(self)
     }
 
-    pub fn resolve(&self, block_number: u64, block_timestamp: u64, block_time_ms: u64, priority_fee: U256, ) -> OrderResolution {
+    pub fn resolve(&self, block_number: u64, block_timestamp: u64, block_time_ms: u64, priority_fee: U256, min_block_percentage_buffer: u64) -> OrderResolution {
         let block_time = block_time_ms / 1000;
         let next_block_timestamp = U256::from(block_timestamp) + U256::from(block_time);
 
@@ -294,7 +336,7 @@ impl PriorityOrder {
             block_timestamp,
             block_time_ms
         );
-        let time_buffer_ms = block_time_ms * 1300 / 1000; // TODO: fine tune
+        let time_buffer_ms = block_time_ms * min_block_percentage_buffer / 100;
         if U256::from(current_timestamp_ms() + time_buffer_ms).lt(&target_block_ms) {
             return OrderResolution::NotFillableYet(ResolvedOrder { input, outputs });
         }

src/aws_utils/cloudwatch_utils.rs

@@ -26,6 +26,9 @@ pub const EXECUTION_ATTEMPTED_METRIC: &str = "ExecutionAttempted";
 pub const EXECUTION_SKIPPED_ALREADY_FILLED_METRIC: &str = "ExecutionSkippedAlreadyFilled";
 pub const EXECUTION_SKIPPED_PAST_DEADLINE_METRIC: &str = "ExecutionSkippedPastDeadline";
 pub const UNPROFITABLE_METRIC: &str = "Unprofitable";
+pub const TARGET_BLOCK_DELTA: &str = "TargetBlockDelta";
+pub const REVERT_CODE_METRIC: &str = "RevertCode";
+
 pub enum DimensionName {
     Service,
 }
@@ -85,9 +88,12 @@ pub enum CwMetrics {
     TxSubmitted(u64),
     TxStatusUnknown(u64),
     LatestBlock(u64),
+    RevertCode(u64, String), // chain_id and revert code string
 
     /// Balance for individual address
     Balance(String),
+    // negative is too early, positive is too late
+    TargetBlockDelta(u64),
 }
 impl From<CwMetrics> for String {
     fn from(metric: CwMetrics) -> Self {
@@ -111,6 +117,8 @@ impl From<CwMetrics> for String {
             }
             CwMetrics::Balance(val) => format!("Bal-{}", val),
             CwMetrics::LatestBlock(chain_id) => format!("{}-{}", chain_id, LATEST_BLOCK),
+            CwMetrics::TargetBlockDelta(chain_id) => format!("{}-{}", chain_id, TARGET_BLOCK_DELTA),
+            CwMetrics::RevertCode(chain_id, code) => format!("{}-{}-{}", chain_id, REVERT_CODE_METRIC, code),
         }
     }
 }
@@ -167,6 +175,10 @@ pub fn receipt_status_to_metric(status: bool, chain_id: u64) -> CwMetrics {
     }
 }
 
+pub fn revert_code_to_metric(chain_id: u64, revert_code: String) -> CwMetrics {
+    CwMetrics::RevertCode(chain_id, revert_code)
+}
+
 pub fn build_metric_future(
     cloudwatch_client: Option<Arc<CloudWatchClient>>,
     dimension_value: DimensionValue,

src/collectors/uniswapx_order_collector.rs

@@ -122,7 +122,7 @@ impl Collector<UniswapXOrder> for UniswapXOrderCollector {
             "Starting UniswapX order collector stream"
         );
 
-        // stream that polls the UniswapX API every 5 seconds
+        // stream that polls the UniswapX API
         let stream = IntervalStream::new(tokio::time::interval(Duration::from_millis(
             POLL_INTERVAL_MS,
         )))
@@ -133,31 +133,45 @@ impl Collector<UniswapXOrder> for UniswapXOrderCollector {
             async move {
                 tracing::debug!("Polling UniswapX API for new orders");
 
-                let response = match client.get(url.clone())
-                    .header("x-api-key", api_key)
-                    .send()
-                    .await {
-                    Ok(resp) => resp,
-                    Err(e) => {
-                        tracing::error!(error = %e, "Failed to fetch orders from UniswapX API");
-                        return Err(anyhow::anyhow!("Failed to fetch orders: {}", e));
+                #[allow(unused_assignments)]
+                let mut last_error = None;
+                loop {
+                    match client.get(url.clone())
+                        .header("x-api-key", api_key.clone())
+                        .send()
+                        .await {
+                        Ok(resp) => {
+                            match resp.json::<UniswapXOrderResponse>().await {
+                                Ok(data) => {
+                                    tracing::debug!(
+                                        num_orders = data.orders.len(),
+                                        "Successfully fetched orders from UniswapX API"
+                                    );
+                                    return Ok(data.orders);
+                                },
+                                Err(e) => {
+                                    last_error = Some(e.to_string());
+                                    tracing::warn!(
+                                        error = %e,
+                                        "Failed to parse UniswapX API response, retrying..."
+                                    );
+                                }
+                            }
+                        },
+                        Err(e) => {
+                            last_error = Some(e.to_string());
+                            tracing::warn!(
+                                error = %e,
+                                "Failed to fetch orders from UniswapX API, retrying..."
+                            );
+                        }
                     }
-                };
-
-                let data = match response.json::<UniswapXOrderResponse>().await {
-                    Ok(data) => data,
-                    Err(e) => {
-                        tracing::error!(error = %e, "Failed to parse UniswapX API response");
-                        return Err(anyhow::anyhow!("Failed to parse response: {}", e));
+
+                    if let Some(err) = last_error {
+                        tracing::warn!(error = %err, "Error in order stream, retrying...");
+                        tokio::time::sleep(Duration::from_millis(1000)).await;
                     }
-                };
-
-                tracing::debug!(
-                    num_orders = data.orders.len(),
-                    "Successfully fetched orders from UniswapX API"
-                );
-
-                Ok(data.orders)
+                }
             }
         })
         .flat_map(
@@ -268,6 +282,7 @@ mod tests {
         } else {
             encoded_order
         };
+        tracing::info!("encoded_order: {:?}", encoded_order);
         let order_hex: Vec<u8> = hex::decode(encoded_order).unwrap();
 
         let result = V2DutchOrder::decode_inner(&order_hex, false);

src/collectors/uniswapx_route_collector.rs

@@ -7,7 +7,7 @@ use reqwest::header::ORIGIN;
 use serde::{Deserialize, Serialize};
 use tokio::sync::mpsc::{Receiver, Sender};
 use tracing::{error, info};
-use uniswapx_rs::order::{Order, ResolvedOrder};
+use uniswapx_rs::order::{Order, ResolvedOrder, TradeType};
 
 use artemis_core::types::{Collector, CollectorStream};
 use async_trait::async_trait;
@@ -40,20 +40,12 @@ pub struct OrderBatchData {
     pub orders: Vec<OrderData>,
     pub chain_id: u64,
     pub amount_in: Uint<256, 4>,
-    pub amount_out_required: Uint<256, 4>,
+    pub amount_out: Uint<256, 4>,
+    pub amount_required: Uint<256, 4>,
     pub token_in: String,
     pub token_out: String,
 }
 
-#[derive(Serialize, Debug)]
-#[allow(dead_code)]
-enum TradeType {
-    #[serde(rename = "exactIn")]
-    ExactIn,
-    #[serde(rename = "exactOut")]
-    ExactOut,
-}
-
 #[derive(Serialize, Debug)]
 #[serde(rename_all = "camelCase")]
 struct RoutingApiQuery {
@@ -82,6 +74,16 @@ pub struct TokenInRoute {
     decimals: String,
 }
 
+#[derive(Clone, Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+#[allow(dead_code)]
+pub struct V4Route {
+    address: String,
+    token_in: TokenInRoute,
+    token_out: TokenInRoute,
+    fee: String,
+}
+
 #[derive(Clone, Debug, Deserialize)]
 #[serde(rename_all = "camelCase")]
 #[allow(dead_code)]
@@ -104,6 +106,8 @@ pub struct V2Route {
 #[derive(Clone, Debug, Deserialize)]
 #[serde(tag = "type")]
 pub enum Route {
+    #[serde(rename = "v4-pool")]
+    V4(V4Route),
     #[serde(rename = "v3-pool")]
     V3(V3Route),
     #[serde(rename = "v2-pool")]
@@ -128,6 +132,7 @@ pub struct RouteOrderParams {
     pub token_out: String,
     pub amount: String,
     pub recipient: String,
+    pub trade_type: TradeType,
 }
 
 #[derive(Clone, Debug)]
@@ -177,13 +182,12 @@ impl UniswapXRouteCollector {
         params: RouteOrderParams,
         order_hash: String,
     ) -> Result<OrderRoute> {
-        // TODO: support exactOutput
         let query = RoutingApiQuery {
             token_in_address: resolve_address(params.token_in),
             token_out_address: resolve_address(params.token_out),
             token_in_chain_id: params.chain_id,
             token_out_chain_id: params.chain_id,
-            trade_type: TradeType::ExactIn,
+            trade_type: params.trade_type,
             amount: params.amount,
             recipient: params.recipient,
             slippage_tolerance: SLIPPAGE_TOLERANCE.to_string(),
@@ -192,7 +196,7 @@ impl UniswapXRouteCollector {
             protocols: "v2,v3,v4,mixed".to_string(),
         };
 
-        let query_string = serde_qs::to_string(&query).unwrap();
+        let query_string = serde_qs::to_string(&query)?;
         let full_query = format!("{}?{}", ROUTING_API, query_string);
         info!("{} - full query: {}", order_hash, full_query);
         let client = reqwest::Client::new();
@@ -202,6 +206,7 @@ impl UniswapXRouteCollector {
             .get(format!("{}?{}", ROUTING_API, query_string))
             .header(ORIGIN, "https://app.uniswap.org")
             .header("x-request-source", "uniswap-web")
+            .header("x-universal-router-version", "2.0")
             .send()
             .await
             .map_err(|e| anyhow!("Quote request failed with error: {}", e))?;
@@ -218,10 +223,14 @@ impl UniswapXRouteCollector {
         }
 
         match response.status() {
-            StatusCode::OK => Ok(response
-                .json::<OrderRoute>()
-                .await
-                .map_err(|e| anyhow!("{} - Failed to parse response: {}", order_hash, e))?),
+            StatusCode::OK => {
+                let order_route = response
+                    .json::<OrderRoute>()
+                    .await
+                    .map_err(|e| anyhow!("{} - Failed to parse response: {}", order_hash, e))?;
+                info!("{} - Received route: {:?}", order_hash, order_route);
+                Ok(order_route)
+            }
             StatusCode::BAD_REQUEST => Err(anyhow!(
                 "{} - Bad request: {}",
                 order_hash,
@@ -292,20 +301,24 @@ impl Collector<RoutedOrder> for UniswapXRouteCollector {
 
                 for batch in all_requests {
                     let order_hash = batch.orders[0].hash.clone();
-                    let OrderBatchData { token_in, token_out, amount_in, .. } = batch.clone();
+                    let OrderBatchData { token_in, token_out, amount_in, amount_out, .. } = batch.clone();
                     info!(
-                        "{} - Routing order, token in: {}, token out: {}",
+                        "{} - Routing order, token in: {}, token out: {}, amount in: {}, amount out: {}",
                         order_hash,
-                        token_in, token_out
+                        token_in, token_out, amount_in, amount_out
                     );
-
                     let future = async move {
                         let route_result = self.route_order(RouteOrderParams {
                             chain_id: self.chain_id,
                             token_in: token_in.clone(),
                             token_out: token_out.clone(),
-                            amount: amount_in.to_string(),
+                            amount: if batch.orders[0].order.is_exact_output() {
+                                amount_out.to_string()
+                            } else {
+                                amount_in.to_string()
+                            },
                             recipient: self.executor_address.clone(),
+                            trade_type: batch.orders[0].order.trade_type(),
                         }, order_hash).await;
                         (batch, route_result)
                     };