feat: refactor trading architecture with unified framework

- Add unified TradeExecutor interface and protocol abstraction - Refactor PumpFun/PumpSwap into adapter pattern - Introduce TradeFactory for multi-protocol support - Add parallel execution and unified parameter system - Include Raydium protocol support and log parsing - Simplify codebase structure and improve maintainability
2025-06-17 23:32:20 +08:00
parent 768dc92156
commit 428ece5d6a
23 changed files with 1908 additions and 1505 deletions
@@ -0,0 +1 @@
+pub const DEFAULT_SLIPPAGE_BASIS_POINTS: u64 = 100;
@@ -0,0 +1,201 @@
+use anyhow::{anyhow, Result};
+use solana_sdk::signer::Signer;
+use std::sync::Arc;
+
+use super::{
+    parallel::parallel_execute_with_tips,
+    params::{BuyParams, BuyWithTipParams, SellParams, SellWithTipParams},
+    timer::TradeTimer,
+    traits::{InstructionBuilder, TradeExecutor},
+};
+use crate::{
+    swqos::TradeType,
+    trading::common::{build_rpc_transaction, build_sell_transaction},
+};
+
+/// 通用交易执行器实现
+pub struct GenericTradeExecutor {
+    instruction_builder: Arc<dyn InstructionBuilder>,
+    protocol_name: &'static str,
+}
+
+impl GenericTradeExecutor {
+    pub fn new(
+        instruction_builder: Arc<dyn InstructionBuilder>,
+        protocol_name: &'static str,
+    ) -> Self {
+        Self {
+            instruction_builder,
+            protocol_name,
+        }
+    }
+
+    /// 获取代币余额
+    async fn get_token_balance(
+        &self,
+        rpc: Arc<crate::common::SolanaRpcClient>,
+        payer: &solana_sdk::signature::Keypair,
+        mint: &solana_sdk::pubkey::Pubkey,
+    ) -> Result<u64> {
+        let ata = spl_associated_token_account::get_associated_token_address(&payer.pubkey(), mint);
+        let balance = rpc.get_token_account_balance(&ata).await?;
+        balance
+            .amount
+            .parse::<u64>()
+            .map_err(|_| anyhow!("Failed to parse token balance"))
+    }
+}
+
+#[async_trait::async_trait]
+impl TradeExecutor for GenericTradeExecutor {
+    async fn buy(&self, params: BuyParams) -> Result<()> {
+        if params.rpc.is_none() {
+            return Err(anyhow!("RPC is not set"));
+        }
+        let rpc = params.rpc.as_ref().unwrap().clone();
+        let mut timer = TradeTimer::new("构建买入交易指令");
+        // 构建指令
+        let instructions = self
+            .instruction_builder
+            .build_buy_instructions(&params)
+            .await?;
+        timer.stage("买入交易指令");
+
+        // 构建交易
+        let transaction = build_rpc_transaction(
+            params.payer.clone(),
+            &params.priority_fee,
+            instructions,
+            params.lookup_table_key,
+            params.recent_blockhash,
+            params.data_size_limit,
+        )
+        .await?;
+        timer.stage("买入交易签名");
+
+        // 发送交易
+        rpc.send_and_confirm_transaction(&transaction).await?;
+        timer.finish();
+
+        Ok(())
+    }
+
+    async fn buy_with_tip(&self, params: BuyWithTipParams) -> Result<()> {
+        let mut timer = TradeTimer::new("构建买入交易指令");
+
+        // 验证参数 - 转换为BuyParams进行验证
+        let buy_params = BuyParams {
+            rpc: params.rpc,
+            payer: params.payer.clone(),
+            mint: params.mint,
+            creator: params.creator,
+            amount_sol: params.amount_sol,
+            slippage_basis_points: params.slippage_basis_points,
+            priority_fee: params.priority_fee.clone(),
+            lookup_table_key: params.lookup_table_key,
+            recent_blockhash: params.recent_blockhash,
+            data_size_limit: params.data_size_limit,
+            protocol_params: params.protocol_params.clone(),
+        };
+
+        // 构建指令
+        let instructions = self
+            .instruction_builder
+            .build_buy_instructions(&buy_params)
+            .await?;
+        timer.stage("买入交易指令");
+
+        // 并行执行交易
+        parallel_execute_with_tips(
+            params.fee_clients,
+            params.payer,
+            instructions,
+            params.priority_fee,
+            params.lookup_table_key,
+            params.recent_blockhash,
+            params.data_size_limit,
+            TradeType::Buy,
+        )
+        .await?;
+
+        timer.finish();
+        Ok(())
+    }
+
+    async fn sell(&self, params: SellParams) -> Result<()> {
+        if params.rpc.is_none() {
+            return Err(anyhow!("RPC is not set"));
+        }
+        let rpc = params.rpc.as_ref().unwrap().clone();
+        let mut timer = TradeTimer::new("构建卖出交易指令");
+
+        // 构建指令
+        let instructions = self
+            .instruction_builder
+            .build_sell_instructions(&params)
+            .await?;
+        timer.stage("卖出交易指令");
+
+        // 构建交易
+        let transaction = build_sell_transaction(
+            params.payer.clone(),
+            &params.priority_fee,
+            instructions,
+            params.lookup_table_key,
+            params.recent_blockhash,
+        )
+        .await?;
+        timer.stage("卖出交易签名");
+
+        // 发送交易
+        rpc.send_and_confirm_transaction(&transaction).await?;
+        timer.finish();
+
+        Ok(())
+    }
+
+    async fn sell_with_tip(&self, params: SellWithTipParams) -> Result<()> {
+        let mut timer = TradeTimer::new("构建卖出交易指令");
+
+        // 转换为SellParams进行指令构建
+        let sell_params = SellParams {
+            rpc: params.rpc,
+            payer: params.payer.clone(),
+            mint: params.mint,
+            creator: params.creator,
+            amount_token: params.amount_token,
+            slippage_basis_points: params.slippage_basis_points,
+            priority_fee: params.priority_fee.clone(),
+            lookup_table_key: params.lookup_table_key,
+            recent_blockhash: params.recent_blockhash,
+            protocol_params: params.protocol_params.clone(),
+        };
+
+        // 构建指令
+        let instructions = self
+            .instruction_builder
+            .build_sell_instructions(&sell_params)
+            .await?;
+        timer.stage("卖出交易指令");
+
+        // 并行执行交易
+        parallel_execute_with_tips(
+            params.fee_clients,
+            params.payer,
+            instructions,
+            params.priority_fee,
+            params.lookup_table_key,
+            params.recent_blockhash,
+            0,
+            TradeType::Sell,
+        )
+        .await?;
+
+        timer.finish();
+        Ok(())
+    }
+
+    fn protocol_name(&self) -> &'static str {
+        self.protocol_name
+    }
+}
@@ -0,0 +1,6 @@
+pub mod constants;
+pub mod params;
+pub mod traits;
+pub mod executor;
+pub mod parallel;
+pub mod timer; 
@@ -0,0 +1,118 @@
+use anyhow::{anyhow, Result};
+use solana_hash::Hash;
+use solana_sdk::{instruction::Instruction, pubkey::Pubkey, signature::Keypair};
+use std::{str::FromStr, sync::Arc};
+use tokio::task::JoinHandle;
+
+use crate::{
+    common::PriorityFee,
+    swqos::{ClientType, FeeClient, TradeType},
+    trading::common::{
+        build_rpc_transaction, build_sell_tip_transaction_with_priority_fee,
+        build_sell_transaction, build_tip_transaction_with_priority_fee,
+    },
+};
+
+/// 并行执行交易的通用函数
+pub async fn parallel_execute_with_tips(
+    fee_clients: Vec<Arc<FeeClient>>,
+    payer: Arc<Keypair>,
+    instructions: Vec<Instruction>,
+    priority_fee: PriorityFee,
+    lookup_table_key: Option<Pubkey>,
+    recent_blockhash: Hash,
+    data_size_limit: u32,
+    trade_type: TradeType,
+) -> Result<()> {
+    let cores = core_affinity::get_core_ids().unwrap();
+    let mut handles: Vec<JoinHandle<Result<()>>> = vec![];
+
+    for i in 0..fee_clients.len() {
+        let fee_client = fee_clients[i].clone();
+        let payer = payer.clone();
+        let instructions = instructions.clone();
+        let mut priority_fee = priority_fee.clone();
+        let core_id = cores[i % cores.len()];
+
+        let handle = tokio::spawn(async move {
+            core_affinity::set_for_current(core_id);
+            let transaction = if matches!(trade_type, TradeType::Sell)
+                && fee_client.get_client_type() == ClientType::Rpc
+            {
+                build_sell_transaction(
+                    payer,
+                    &priority_fee,
+                    instructions,
+                    lookup_table_key,
+                    recent_blockhash,
+                )
+                .await?
+            } else if matches!(trade_type, TradeType::Sell)
+                && fee_client.get_client_type() != ClientType::Rpc
+            {
+                let tip_account = fee_client.get_tip_account()?;
+                let tip_account = Arc::new(Pubkey::from_str(&tip_account).map_err(|e| anyhow!(e))?);
+                build_sell_tip_transaction_with_priority_fee(
+                    payer,
+                    &priority_fee,
+                    instructions,
+                    &tip_account,
+                    lookup_table_key,
+                    recent_blockhash,
+                )
+                .await?
+            } else if fee_client.get_client_type() == ClientType::Rpc {
+                build_rpc_transaction(
+                    payer,
+                    &priority_fee,
+                    instructions,
+                    lookup_table_key,
+                    recent_blockhash,
+                    data_size_limit,
+                )
+                .await?
+            } else {
+                let tip_account = fee_client.get_tip_account()?;
+                let tip_account = Arc::new(Pubkey::from_str(&tip_account).map_err(|e| anyhow!(e))?);
+                priority_fee.buy_tip_fee = priority_fee.buy_tip_fees[i];
+
+                build_tip_transaction_with_priority_fee(
+                    payer,
+                    &priority_fee,
+                    instructions,
+                    &tip_account,
+                    lookup_table_key,
+                    recent_blockhash,
+                    data_size_limit,
+                )
+                .await?
+            };
+
+            fee_client
+                .send_transaction(trade_type, &transaction)
+                .await?;
+            Ok::<(), anyhow::Error>(())
+        });
+
+        handles.push(handle);
+    }
+
+    // 等待所有任务完成
+    let mut errors = Vec::new();
+    for handle in handles {
+        match handle.await {
+            Ok(Ok(_)) => (),
+            Ok(Err(e)) => errors.push(format!("Task error: {}", e)),
+            Err(e) => errors.push(format!("Join error: {}", e)),
+        }
+    }
+
+    if !errors.is_empty() {
+        for error in &errors {
+            println!("{}", error);
+        }
+        return Err(anyhow!("Some tasks failed: {:?}", errors));
+    }
+
+    Ok(())
+}
@@ -0,0 +1,161 @@
+use solana_hash::Hash;
+use solana_sdk::{pubkey::Pubkey, signature::Keypair};
+use std::sync::Arc;
+
+use super::traits::ProtocolParams;
+use crate::common::{PriorityFee, SolanaRpcClient};
+use crate::swqos::FeeClient;
+
+/// 通用买入参数
+#[derive(Clone)]
+pub struct BuyParams {
+    pub rpc: Option<Arc<SolanaRpcClient>>,
+    pub payer: Arc<Keypair>,
+    pub mint: Pubkey,
+    pub creator: Pubkey,
+    pub amount_sol: u64,
+    pub slippage_basis_points: Option<u64>,
+    pub priority_fee: PriorityFee,
+    pub lookup_table_key: Option<Pubkey>,
+    pub recent_blockhash: Hash,
+    pub data_size_limit: u32,
+    pub protocol_params: Box<dyn ProtocolParams>,
+}
+
+/// 带MEV服务的买入参数
+#[derive(Clone)]
+pub struct BuyWithTipParams {
+    pub rpc: Option<Arc<SolanaRpcClient>>,
+    pub fee_clients: Vec<Arc<FeeClient>>,
+    pub payer: Arc<Keypair>,
+    pub mint: Pubkey,
+    pub creator: Pubkey,
+    pub amount_sol: u64,
+    pub slippage_basis_points: Option<u64>,
+    pub priority_fee: PriorityFee,
+    pub lookup_table_key: Option<Pubkey>,
+    pub recent_blockhash: Hash,
+    pub data_size_limit: u32,
+    pub protocol_params: Box<dyn ProtocolParams>,
+}
+
+/// 通用卖出参数
+#[derive(Clone)]
+pub struct SellParams {
+    pub rpc: Option<Arc<SolanaRpcClient>>,
+    pub payer: Arc<Keypair>,
+    pub mint: Pubkey,
+    pub creator: Pubkey,
+    pub amount_token: Option<u64>,
+    pub slippage_basis_points: Option<u64>,
+    pub priority_fee: PriorityFee,
+    pub lookup_table_key: Option<Pubkey>,
+    pub recent_blockhash: Hash,
+    pub protocol_params: Box<dyn ProtocolParams>,
+}
+
+/// 带MEV服务的卖出参数
+#[derive(Clone)]
+pub struct SellWithTipParams {
+    pub rpc: Option<Arc<SolanaRpcClient>>,
+    pub fee_clients: Vec<Arc<FeeClient>>,
+    pub payer: Arc<Keypair>,
+    pub mint: Pubkey,
+    pub creator: Pubkey,
+    pub amount_token: Option<u64>,
+    pub slippage_basis_points: Option<u64>,
+    pub priority_fee: PriorityFee,
+    pub lookup_table_key: Option<Pubkey>,
+    pub recent_blockhash: Hash,
+    pub protocol_params: Box<dyn ProtocolParams>,
+}
+
+/// PumpFun协议特定参数
+#[derive(Clone)]
+pub struct PumpFunParams {
+    pub dev_buy_token: u64,
+    pub dev_sol_cost: u64,
+    pub trade_type: String,
+}
+
+impl ProtocolParams for PumpFunParams {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn clone_box(&self) -> Box<dyn ProtocolParams> {
+        Box::new(self.clone())
+    }
+}
+
+#[derive(Clone)]
+pub struct PumpFunSellParams {}
+
+impl ProtocolParams for PumpFunSellParams {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn clone_box(&self) -> Box<dyn ProtocolParams> {
+        Box::new(self.clone())
+    }
+}
+
+/// PumpSwap协议特定参数
+#[derive(Clone)]
+pub struct PumpSwapParams {
+    pub pool: Option<Pubkey>,
+    pub pool_base_token_account: Option<Pubkey>,
+    pub pool_quote_token_account: Option<Pubkey>,
+    pub user_base_token_account: Option<Pubkey>,
+    pub user_quote_token_account: Option<Pubkey>,
+}
+
+impl ProtocolParams for PumpSwapParams {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn clone_box(&self) -> Box<dyn ProtocolParams> {
+        Box::new(self.clone())
+    }
+}
+
+impl BuyParams {
+    /// 转换为BuyWithTipParams
+    pub fn with_tip(self, fee_clients: Vec<Arc<FeeClient>>) -> BuyWithTipParams {
+        BuyWithTipParams {
+            rpc: self.rpc,
+            fee_clients,
+            payer: self.payer,
+            mint: self.mint,
+            creator: self.creator,
+            amount_sol: self.amount_sol,
+            slippage_basis_points: self.slippage_basis_points,
+            priority_fee: self.priority_fee,
+            lookup_table_key: self.lookup_table_key,
+            recent_blockhash: self.recent_blockhash,
+            data_size_limit: self.data_size_limit,
+            protocol_params: self.protocol_params,
+        }
+    }
+}
+
+impl SellParams {
+    /// 转换为SellWithTipParams
+    pub fn with_tip(self, fee_clients: Vec<Arc<FeeClient>>) -> SellWithTipParams {
+        SellWithTipParams {
+            rpc: self.rpc,
+            fee_clients,
+            payer: self.payer,
+            mint: self.mint,
+            creator: self.creator,
+            amount_token: self.amount_token,
+            slippage_basis_points: self.slippage_basis_points,
+            priority_fee: self.priority_fee,
+            lookup_table_key: self.lookup_table_key,
+            recent_blockhash: self.recent_blockhash,
+            protocol_params: self.protocol_params,
+        }
+    }
+}
@@ -0,0 +1,46 @@
+use std::time::Instant;
+
+/// 交易时间测量器
+pub struct TradeTimer {
+    start_time: Instant,
+    stage: String,
+}
+
+impl TradeTimer {
+    /// 创建新的计时器
+    pub fn new(stage: impl Into<String>) -> Self {
+        Self {
+            start_time: Instant::now(),
+            stage: stage.into(),
+        }
+    }
+    
+    /// 记录当前阶段耗时并开始新阶段
+    pub fn stage(&mut self, new_stage: impl Into<String>) {
+        let elapsed = self.start_time.elapsed();
+        println!(" {} 耗时: {:?}", self.stage, elapsed);
+        
+        self.start_time = Instant::now();
+        self.stage = new_stage.into();
+    }
+    
+    /// 完成计时并输出最终耗时
+    pub fn finish(self) {
+        let elapsed = self.start_time.elapsed();
+        println!(" {} 耗时: {:?}", self.stage, elapsed);
+    }
+    
+    /// 获取当前阶段的耗时（不重置计时器）
+    pub fn elapsed(&self) -> std::time::Duration {
+        self.start_time.elapsed()
+    }
+}
+
+impl Drop for TradeTimer {
+    fn drop(&mut self) {
+        if !self.stage.is_empty() {
+            let elapsed = self.start_time.elapsed();
+            println!(" {} 耗时: {:?}", self.stage, elapsed);
+        }
+    }
+} 
@@ -0,0 +1,49 @@
+use anyhow::Result;
+use solana_sdk::instruction::Instruction;
+use std::sync::Arc;
+
+use super::params::{BuyParams, BuyWithTipParams, SellParams, SellWithTipParams};
+
+/// 交易执行器trait - 定义了所有交易协议都需要实现的核心方法
+#[async_trait::async_trait]
+pub trait TradeExecutor: Send + Sync {
+    /// 执行买入交易
+    async fn buy(&self, params: BuyParams) -> Result<()>;
+
+    /// 使用MEV服务执行买入交易
+    async fn buy_with_tip(&self, params: BuyWithTipParams) -> Result<()>;
+
+    /// 执行卖出交易
+    async fn sell(&self, params: SellParams) -> Result<()>;
+
+    /// 使用MEV服务执行卖出交易
+    async fn sell_with_tip(&self, params: SellWithTipParams) -> Result<()>;
+
+    /// 获取协议名称
+    fn protocol_name(&self) -> &'static str;
+}
+
+/// 指令构建器trait - 负责构建协议特定的交易指令
+#[async_trait::async_trait]
+pub trait InstructionBuilder: Send + Sync {
+    /// 构建买入指令
+    async fn build_buy_instructions(&self, params: &BuyParams) -> Result<Vec<Instruction>>;
+
+    /// 构建卖出指令
+    async fn build_sell_instructions(&self, params: &SellParams) -> Result<Vec<Instruction>>;
+}
+
+/// 协议特定参数trait - 允许每个协议定义自己的参数
+pub trait ProtocolParams: Send + Sync {
+    /// 将参数转换为Any以便向下转型
+    fn as_any(&self) -> &dyn std::any::Any;
+
+    /// 克隆参数
+    fn clone_box(&self) -> Box<dyn ProtocolParams>;
+}
+
+impl Clone for Box<dyn ProtocolParams> {
+    fn clone(&self) -> Self {
+        self.clone_box()
+    }
+}
				`@@ -0,0 +1 @@`
				`pub const DEFAULT_SLIPPAGE_BASIS_POINTS: u64 = 100;`