feat: Add Raydium AMM V4 support and middleware system

- Add Raydium AMM V4 trading protocol support - Implement instruction middleware system for dynamic processing - Refactor trade executors to support middleware parameters - Add fee calculation and slippage protection mechanisms - Maintain backward compatibility with optional middleware
2025-08-19 18:07:21 +08:00
parent 4b74d3cfb4
commit d27a44735a
23 changed files with 1534 additions and 168 deletions
@@ -9,7 +9,10 @@ use super::{
 };
 use crate::{
    swqos::TradeType,
-    trading::common::{build_rpc_transaction, build_sell_transaction},
+    trading::{
+        common::{build_rpc_transaction, build_sell_transaction},
+        middleware::MiddlewareManager,
+    },
 };

 const MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT: u32 = 256 * 1024;
@@ -25,16 +28,17 @@ impl GenericTradeExecutor {
        instruction_builder: Arc<dyn InstructionBuilder>,
        protocol_name: &'static str,
    ) -> Self {
-        Self {
-            instruction_builder,
-            protocol_name,
-        }
+        Self { instruction_builder, protocol_name }
    }
 }

 #[async_trait::async_trait]
 impl TradeExecutor for GenericTradeExecutor {
-    async fn buy(&self, mut params: BuyParams) -> Result<()> {
+    async fn buy(
+        &self,
+        mut params: BuyParams,
+        middleware_manager: Option<Arc<MiddlewareManager>>,
+    ) -> Result<()> {
        if params.data_size_limit == 0 {
            params.data_size_limit = MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT;
        }
@@ -44,20 +48,29 @@ impl TradeExecutor for GenericTradeExecutor {
        let rpc = params.rpc.as_ref().unwrap().clone();
        let mut timer = TradeTimer::new("构建买入交易指令");
        // 构建指令
-        let instructions = self
-            .instruction_builder
-            .build_buy_instructions(&params)
-            .await?;
+        let instructions = self.instruction_builder.build_buy_instructions(&params).await?;
+        let final_instructions = match middleware_manager.clone() {
+            Some(middleware_manager) => middleware_manager
+                .apply_middlewares_process_protocol_instructions(
+                    instructions,
+                    self.protocol_name.to_string(),
+                    true,
+                )?,
+            None => instructions,
+        };
        timer.stage("构建rpc交易指令");

        // 构建交易
        let transaction = build_rpc_transaction(
            params.payer.clone(),
            &params.priority_fee,
-            instructions,
+            final_instructions,
            params.lookup_table_key,
            params.recent_blockhash,
            params.data_size_limit,
+            middleware_manager,
+            self.protocol_name.to_string(),
+            true,
        )
        .await?;
        timer.stage("rpc提交确认");
@@ -69,7 +82,11 @@ impl TradeExecutor for GenericTradeExecutor {
        Ok(())
    }

-    async fn buy_with_tip(&self, mut params: BuyWithTipParams) -> Result<()> {
+    async fn buy_with_tip(
+        &self,
+        mut params: BuyWithTipParams,
+        middleware_manager: Option<Arc<MiddlewareManager>>,
+    ) -> Result<()> {
        if params.data_size_limit == 0 {
            params.data_size_limit = MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT;
        }
@@ -91,10 +108,16 @@ impl TradeExecutor for GenericTradeExecutor {
        };

        // 构建指令
-        let instructions = self
-            .instruction_builder
-            .build_buy_instructions(&buy_params)
-            .await?;
+        let instructions = self.instruction_builder.build_buy_instructions(&buy_params).await?;
+        let final_instructions = match middleware_manager.clone() {
+            Some(middleware_manager) => middleware_manager
+                .apply_middlewares_process_protocol_instructions(
+                    instructions,
+                    self.protocol_name.to_string(),
+                    true,
+                )?,
+            None => instructions,
+        };

        timer.finish();

@@ -102,19 +125,26 @@ impl TradeExecutor for GenericTradeExecutor {
        parallel_execute_with_tips(
            params.swqos_clients,
            params.payer,
-            instructions,
+            final_instructions,
            params.priority_fee,
            params.lookup_table_key,
            params.recent_blockhash,
            params.data_size_limit,
            TradeType::Buy,
+            middleware_manager,
+            self.protocol_name.to_string(),
+            true,
        )
        .await?;

        Ok(())
    }

-    async fn sell(&self, params: SellParams) -> Result<()> {
+    async fn sell(
+        &self,
+        params: SellParams,
+        middleware_manager: Option<Arc<MiddlewareManager>>,
+    ) -> Result<()> {
        if params.rpc.is_none() {
            return Err(anyhow!("RPC is not set"));
        }
@@ -122,19 +152,28 @@ impl TradeExecutor for GenericTradeExecutor {
        let mut timer = TradeTimer::new("构建卖出交易指令");

        // 构建指令
-        let instructions = self
-            .instruction_builder
-            .build_sell_instructions(&params)
-            .await?;
+        let instructions = self.instruction_builder.build_sell_instructions(&params).await?;
+        let final_instructions = match middleware_manager.clone() {
+            Some(middleware_manager) => middleware_manager
+                .apply_middlewares_process_protocol_instructions(
+                    instructions,
+                    self.protocol_name.to_string(),
+                    false,
+                )?,
+            None => instructions,
+        };
        timer.stage("卖出交易指令");

        // 构建交易
        let transaction = build_sell_transaction(
            params.payer.clone(),
            &params.priority_fee,
-            instructions,
+            final_instructions,
            params.lookup_table_key,
            params.recent_blockhash,
+            middleware_manager,
+            self.protocol_name.to_string(),
+            false,
        )
        .await?;
        timer.stage("卖出交易签名");
@@ -146,7 +185,11 @@ impl TradeExecutor for GenericTradeExecutor {
        Ok(())
    }

-    async fn sell_with_tip(&self, params: SellWithTipParams) -> Result<()> {
+    async fn sell_with_tip(
+        &self,
+        params: SellWithTipParams,
+        middleware_manager: Option<Arc<MiddlewareManager>>,
+    ) -> Result<()> {
        let timer = TradeTimer::new("构建卖出交易指令");

        // 转换为SellParams进行指令构建
@@ -164,10 +207,16 @@ impl TradeExecutor for GenericTradeExecutor {
        };

        // 构建指令
-        let instructions = self
-            .instruction_builder
-            .build_sell_instructions(&sell_params)
-            .await?;
+        let instructions = self.instruction_builder.build_sell_instructions(&sell_params).await?;
+        let final_instructions = match middleware_manager.clone() {
+            Some(middleware_manager) => middleware_manager
+                .apply_middlewares_process_protocol_instructions(
+                    instructions,
+                    self.protocol_name.to_string(),
+                    false,
+                )?,
+            None => instructions,
+        };

        timer.finish();

@@ -175,12 +224,15 @@ impl TradeExecutor for GenericTradeExecutor {
        parallel_execute_with_tips(
            params.swqos_clients,
            params.payer,
-            instructions,
+            final_instructions,
            params.priority_fee,
            params.lookup_table_key,
            params.recent_blockhash,
            0,
            TradeType::Sell,
+            middleware_manager,
+            self.protocol_name.to_string(),
+            false,
        )
        .await?;

@@ -6,11 +6,14 @@ use tokio::task::JoinHandle;

 use crate::{
    common::PriorityFee,
-    swqos::{SwqosType, SwqosClient, TradeType},
-    trading::core::timer::TradeTimer,
-    trading::common::{
-        build_rpc_transaction, build_sell_tip_transaction_with_priority_fee,
-        build_sell_transaction, build_tip_transaction_with_priority_fee,
+    swqos::{SwqosClient, SwqosType, TradeType},
+    trading::{
+        common::{
+            build_rpc_transaction, build_sell_tip_transaction_with_priority_fee,
+            build_sell_transaction, build_tip_transaction_with_priority_fee,
+        },
+        core::timer::TradeTimer,
+        MiddlewareManager,
    },
 };

@@ -24,6 +27,9 @@ pub async fn parallel_execute_with_tips(
    recent_blockhash: Hash,
    data_size_limit: u32,
    trade_type: TradeType,
+    middleware_manager: Option<Arc<MiddlewareManager>>,
+    protocol_name: String,
+    is_buy: bool,
 ) -> Result<()> {
    let cores = core_affinity::get_core_ids().unwrap();
    let mut handles: Vec<JoinHandle<Result<()>>> = vec![];
@@ -35,10 +41,14 @@ pub async fn parallel_execute_with_tips(
        let mut priority_fee = priority_fee.clone();
        let core_id = cores[i % cores.len()];

+        let middleware_manager = middleware_manager.clone();
+        let protocol_name = protocol_name.clone();
+
        let handle = tokio::spawn(async move {
            core_affinity::set_for_current(core_id);

-            let mut timer = TradeTimer::new(format!("构建交易指令: {:?}", swqos_client.get_swqos_type()));
+            let mut timer =
+                TradeTimer::new(format!("构建交易指令: {:?}", swqos_client.get_swqos_type()));

            let transaction = if matches!(trade_type, TradeType::Sell)
                && swqos_client.get_swqos_type() == SwqosType::Default
@@ -49,6 +59,9 @@ pub async fn parallel_execute_with_tips(
                    instructions,
                    lookup_table_key,
                    recent_blockhash,
+                    middleware_manager,
+                    protocol_name,
+                    is_buy,
                )
                .await?
            } else if matches!(trade_type, TradeType::Sell)
@@ -63,6 +76,9 @@ pub async fn parallel_execute_with_tips(
                    &tip_account,
                    lookup_table_key,
                    recent_blockhash,
+                    middleware_manager,
+                    protocol_name,
+                    is_buy,
                )
                .await?
            } else if swqos_client.get_swqos_type() == SwqosType::Default {
@@ -73,6 +89,9 @@ pub async fn parallel_execute_with_tips(
                    lookup_table_key,
                    recent_blockhash,
                    data_size_limit,
+                    middleware_manager,
+                    protocol_name,
+                    is_buy,
                )
                .await?
            } else {
@@ -88,15 +107,16 @@ pub async fn parallel_execute_with_tips(
                    lookup_table_key,
                    recent_blockhash,
                    data_size_limit,
+                    middleware_manager,
+                    protocol_name,
+                    is_buy,
                )
                .await?
            };

            timer.stage(format!("提交交易指令: {:?}", swqos_client.get_swqos_type()));

-            swqos_client
-                .send_transaction(trade_type, &transaction)
-                .await?;
+            swqos_client.send_transaction(trade_type, &transaction).await?;

            timer.finish();
            Ok::<(), anyhow::Error>(())
@@ -4,6 +4,8 @@ use solana_streamer_sdk::streaming::event_parser::protocols::pumpfun::PumpFunTra
 use solana_streamer_sdk::streaming::event_parser::protocols::pumpswap::{
    PumpSwapBuyEvent, PumpSwapSellEvent,
 };
+use solana_streamer_sdk::streaming::event_parser::protocols::raydium_amm_v4::types::AmmInfo;
+use solana_streamer_sdk::streaming::event_parser::protocols::raydium_amm_v4::RaydiumAmmV4SwapEvent;
 use std::sync::Arc;

 use super::traits::ProtocolParams;
@@ -16,6 +18,7 @@ use crate::solana_streamer_sdk::streaming::event_parser::common::EventType;
 use crate::solana_streamer_sdk::streaming::event_parser::protocols::bonk::BonkTradeEvent;
 use crate::swqos::SwqosClient;
 use crate::trading::bonk::common::{get_amount_in, get_amount_in_net, get_amount_out};
+use crate::trading::common::get_multi_token_balances;
 use crate::trading::pumpswap::common::get_token_balances;
 use crate::trading::raydium_cpmm::common::get_pool_token_balances;

@@ -367,6 +370,76 @@ impl ProtocolParams for RaydiumCpmmParams {
    }
 }

+/// RaydiumCpmm protocol specific parameters
+/// Configuration parameters specific to Raydium CPMM trading protocol
+#[derive(Clone)]
+pub struct RaydiumAmmV4Params {
+    /// AMM pool address
+    pub amm: Pubkey,
+    /// Base token (coin) mint address
+    pub coin_mint: Pubkey,
+    /// Quote token (pc) mint address  
+    pub pc_mint: Pubkey,
+    /// Pool's coin token account address
+    pub token_coin: Pubkey,
+    /// Pool's pc token account address
+    pub token_pc: Pubkey,
+    /// Current coin reserve amount in the pool
+    pub coin_reserve: u64,
+    /// Current pc reserve amount in the pool
+    pub pc_reserve: u64,
+    /// Whether to automatically handle wSOL wrapping and unwrapping
+    pub auto_handle_wsol: bool,
+}
+
+impl RaydiumAmmV4Params {
+    pub fn from_amm_info_and_reserves(
+        amm: Pubkey,
+        amm_info: AmmInfo,
+        coin_reserve: u64,
+        pc_reserve: u64,
+    ) -> Self {
+        Self {
+            amm,
+            coin_mint: amm_info.coin_mint,
+            pc_mint: amm_info.pc_mint,
+            token_coin: amm_info.token_coin,
+            token_pc: amm_info.token_pc,
+            coin_reserve,
+            pc_reserve,
+            auto_handle_wsol: true,
+        }
+    }
+    pub async fn from_amm_address_by_rpc(
+        rpc: &SolanaRpcClient,
+        amm: Pubkey,
+    ) -> Result<Self, anyhow::Error> {
+        let amm_info = crate::trading::raydium_amm_v4::common::fetch_amm_info(rpc, amm).await?;
+        let (coin_reserve, pc_reserve) =
+            get_multi_token_balances(rpc, &amm_info.token_coin, &amm_info.token_pc).await?;
+        Ok(Self {
+            amm,
+            coin_mint: amm_info.coin_mint,
+            pc_mint: amm_info.pc_mint,
+            token_coin: amm_info.token_coin,
+            token_pc: amm_info.token_pc,
+            coin_reserve,
+            pc_reserve,
+            auto_handle_wsol: true,
+        })
+    }
+}
+
+impl ProtocolParams for RaydiumAmmV4Params {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn clone_box(&self) -> Box<dyn ProtocolParams> {
+        Box::new(self.clone())
+    }
+}
+
 impl BuyParams {
    /// Convert to BuyWithTipParams
    /// Transforms basic buy parameters into MEV-enabled parameters
@@ -1,21 +1,25 @@
+use std::sync::Arc;
+
 use anyhow::Result;
 use solana_sdk::instruction::Instruction;
+use crate::trading::MiddlewareManager;
+
 use super::params::{BuyParams, BuyWithTipParams, SellParams, SellWithTipParams};

 /// 交易执行器trait - 定义了所有交易协议都需要实现的核心方法
 #[async_trait::async_trait]
 pub trait TradeExecutor: Send + Sync {
    /// 执行买入交易
-    async fn buy(&self, params: BuyParams) -> Result<()>;
+    async fn buy(&self, params: BuyParams, middleware_manager: Option<Arc<MiddlewareManager>>) -> Result<()>;

    /// 使用MEV服务执行买入交易
-    async fn buy_with_tip(&self, params: BuyWithTipParams) -> Result<()>;
+    async fn buy_with_tip(&self, params: BuyWithTipParams, middleware_manager: Option<Arc<MiddlewareManager>>) -> Result<()>;

    /// 执行卖出交易
-    async fn sell(&self, params: SellParams) -> Result<()>;
+    async fn sell(&self, params: SellParams, middleware_manager: Option<Arc<MiddlewareManager>>) -> Result<()>;

    /// 使用MEV服务执行卖出交易
-    async fn sell_with_tip(&self, params: SellWithTipParams) -> Result<()>;
+    async fn sell_with_tip(&self, params: SellWithTipParams, middleware_manager: Option<Arc<MiddlewareManager>>) -> Result<()>;

    /// 获取协议名称
    fn protocol_name(&self) -> &'static str;