From 602b310cf841be6e695fd01744bc4a510b59f962 Mon Sep 17 00:00:00 2001 From: sgxiang Date: Mon, 16 Jun 2025 18:30:54 +0800 Subject: [PATCH] feat: Add Raydium DEX integration support Add Raydium logs processing, event parsing and GRPC stream handling functionality, remove redundant PumpSwap subscription module, optimize project structure. --- src/common/mod.rs | 1 + src/common/pumpswap/logs_subscribe.rs | 134 --------------------- src/common/pumpswap/mod.rs | 2 - src/common/raydium/logs_data.rs | 161 ++++++++++++++++++++++++++ src/common/raydium/logs_events.rs | 17 +++ src/common/raydium/logs_filters.rs | 62 ++++++++++ src/common/raydium/logs_parser.rs | 155 +++++++++++++++++++++++++ src/common/raydium/mod.rs | 9 ++ src/constants/mod.rs | 1 + src/constants/raydium/mod.rs | 85 ++++++++++++++ src/grpc/shred_stream.rs | 74 ++++++++++++ src/grpc/yellow_stone.rs | 109 +++++++++++++++++ src/main.rs | 68 ++++++++++- 13 files changed, 741 insertions(+), 137 deletions(-) delete mode 100755 src/common/pumpswap/logs_subscribe.rs create mode 100644 src/common/raydium/logs_data.rs create mode 100644 src/common/raydium/logs_events.rs create mode 100755 src/common/raydium/logs_filters.rs create mode 100755 src/common/raydium/logs_parser.rs create mode 100644 src/common/raydium/mod.rs create mode 100755 src/constants/raydium/mod.rs diff --git a/src/common/mod.rs b/src/common/mod.rs index fbbd475..72f857b 100755 --- a/src/common/mod.rs +++ b/src/common/mod.rs @@ -1,5 +1,6 @@ pub mod pumpfun; pub mod pumpswap; +pub mod raydium; pub mod address_lookup; pub mod nonce_cache; pub mod tip_cache; diff --git a/src/common/pumpswap/logs_subscribe.rs b/src/common/pumpswap/logs_subscribe.rs deleted file mode 100755 index f92d68a..0000000 --- a/src/common/pumpswap/logs_subscribe.rs +++ /dev/null @@ -1,134 +0,0 @@ -use solana_client::{ - nonblocking::pubsub_client::PubsubClient, - rpc_config::{RpcTransactionLogsConfig, RpcTransactionLogsFilter} -}; - -use solana_sdk::commitment_config::CommitmentConfig; -use std::sync::Arc; -use tokio::sync::mpsc; -use tokio::task::JoinHandle; -use futures::StreamExt; -use crate::common::pumpswap::{ - logs_events::PumpSwapEvent, - logs_filters::LogFilter -}; -use crate::constants::pumpswap::accounts; - -/// 订阅句柄，包含任务和取消订阅逻辑 -pub struct SubscriptionHandle { - pub task: JoinHandle<()>, - pub unsub_fn: Box, -} - -impl SubscriptionHandle { - pub async fn shutdown(self) { - (self.unsub_fn)(); - self.task.abort(); - } -} - -/// 创建PubSub客户端 -pub async fn create_pubsub_client(ws_url: &str) -> PubsubClient { - PubsubClient::new(ws_url).await.unwrap() -} - -/// 启动PumpSwap代币订阅 -pub async fn tokens_subscription( - ws_url: &str, - commitment: CommitmentConfig, - callback: F, -) -> Result> -where - F: Fn(PumpSwapEvent) + Send + Sync + 'static, -{ - // 使用constants中定义的AMM_PROGRAM - let program_address = accounts::AMM_PROGRAM.to_string(); - let logs_filter = RpcTransactionLogsFilter::Mentions(vec![program_address]); - - let logs_config = RpcTransactionLogsConfig { - commitment: Some(commitment), - }; - - // 创建PubsubClient - let sub_client = Arc::new(PubsubClient::new(ws_url).await.unwrap()); - - let sub_client_clone = Arc::clone(&sub_client); - - // 创建用于取消订阅的通道 - let (unsub_tx, _) = mpsc::channel(1); - - // 启动订阅任务 - let task = tokio::spawn(async move { - let (mut stream, _) = sub_client_clone.logs_subscribe(logs_filter, logs_config).await.unwrap(); - - loop { - let msg = stream.next().await; - match msg { - Some(msg) => { - if let Some(_err) = msg.value.err { - continue; - } - - let events = LogFilter::parse_pumpswap_logs(&msg.value.logs); - for mut event in events { - // 设置签名和slot - match &mut event { - PumpSwapEvent::Buy(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::Sell(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::CreatePool(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::Deposit(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::Withdraw(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::Disable(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::UpdateAdmin(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - PumpSwapEvent::UpdateFeeConfig(e) => { - e.signature = msg.value.signature.clone(); - e.slot = msg.context.slot; - }, - _ => {} - } - callback(event); - } - } - None => { - println!("PumpSwap subscription stream ended"); - } - } - } - }); - - // 返回订阅句柄和取消订阅逻辑 - Ok(SubscriptionHandle { - task, - unsub_fn: Box::new(move || { - let _ = unsub_tx.try_send(()); - }), - }) -} - - - -/// 停止订阅 -pub async fn stop_subscription(handle: SubscriptionHandle) { - handle.shutdown().await; -} diff --git a/src/common/pumpswap/mod.rs b/src/common/pumpswap/mod.rs index 8de9d17..856a6f0 100644 --- a/src/common/pumpswap/mod.rs +++ b/src/common/pumpswap/mod.rs @@ -1,11 +1,9 @@ pub mod logs_data; pub mod logs_parser; pub mod logs_filters; -pub mod logs_subscribe; pub mod logs_events; pub use logs_data::*; pub use logs_parser::*; pub use logs_filters::*; -pub use logs_subscribe::*; pub use logs_events::*; \ No newline at end of file diff --git a/src/common/raydium/logs_data.rs b/src/common/raydium/logs_data.rs new file mode 100644 index 0000000..53d1cb5 --- /dev/null +++ b/src/common/raydium/logs_data.rs @@ -0,0 +1,161 @@ +use borsh::{BorshDeserialize, BorshSerialize}; +use serde::{Deserialize, Serialize}; +use solana_sdk::pubkey::Pubkey; + +use crate::error::{ClientError, ClientResult}; + +/// Raydium指令类型 +#[derive(Debug)] +pub enum RaydiumInstruction { + V4Swap(V4SwapEvent), + SwapBaseInput(SwapBaseInputEvent), + SwapBaseOutput(SwapBaseOutputEvent), + Other, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct SwapBaseOutputEvent { + #[borsh(skip)] + pub timestamp: i64, + #[borsh(skip)] + pub slot: u64, + #[borsh(skip)] + pub signature: String, + pub max_amount_in: u64, + pub amount_out: u64, + + #[borsh(skip)] + pub payer: Pubkey, + #[borsh(skip)] + pub authority: Pubkey, + #[borsh(skip)] + pub amm_config: Pubkey, + #[borsh(skip)] + pub pool_state: Pubkey, + #[borsh(skip)] + pub input_token_account: Pubkey, + #[borsh(skip)] + pub output_token_account: Pubkey, + #[borsh(skip)] + pub input_vault: Pubkey, + #[borsh(skip)] + pub output_vault: Pubkey, + #[borsh(skip)] + pub input_token_program: Pubkey, + #[borsh(skip)] + pub output_token_program: Pubkey, + #[borsh(skip)] + pub input_token_mint: Pubkey, + #[borsh(skip)] + pub output_token_mint: Pubkey, + #[borsh(skip)] + pub observation_state: Pubkey, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct SwapBaseInputEvent { + #[borsh(skip)] + pub timestamp: i64, + #[borsh(skip)] + pub slot: u64, + #[borsh(skip)] + pub signature: String, + pub amount_in: u64, + pub minimum_amount_out: u64, + + #[borsh(skip)] + pub payer: Pubkey, + #[borsh(skip)] + pub authority: Pubkey, + #[borsh(skip)] + pub amm_config: Pubkey, + #[borsh(skip)] + pub pool_state: Pubkey, + #[borsh(skip)] + pub input_token_account: Pubkey, + #[borsh(skip)] + pub output_token_account: Pubkey, + #[borsh(skip)] + pub input_vault: Pubkey, + #[borsh(skip)] + pub output_vault: Pubkey, + #[borsh(skip)] + pub input_token_program: Pubkey, + #[borsh(skip)] + pub output_token_program: Pubkey, + #[borsh(skip)] + pub input_token_mint: Pubkey, + #[borsh(skip)] + pub output_token_mint: Pubkey, + #[borsh(skip)] + pub observation_state: Pubkey, +} + +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, BorshDeserialize)] +pub struct V4SwapEvent { + #[borsh(skip)] + pub timestamp: i64, + #[borsh(skip)] + pub slot: u64, + #[borsh(skip)] + pub signature: String, + pub amount_in: u64, + pub minimum_amount_out: u64, + + #[borsh(skip)] + pub amm: Pubkey, + #[borsh(skip)] + pub amm_authority: Pubkey, + #[borsh(skip)] + pub amm_open_orders: Pubkey, + #[borsh(skip)] + pub amm_target_orders: Pubkey, + #[borsh(skip)] + pub pool_coin_token_account: Pubkey, + #[borsh(skip)] + pub pool_pc_token_account: Pubkey, + #[borsh(skip)] + pub serum_program: Pubkey, + #[borsh(skip)] + pub serum_market: Pubkey, + #[borsh(skip)] + pub serum_bids: Pubkey, + #[borsh(skip)] + pub serum_asks: Pubkey, + #[borsh(skip)] + pub serum_event_queue: Pubkey, + #[borsh(skip)] + pub serum_coin_vault_account: Pubkey, + #[borsh(skip)] + pub serum_pc_vault_account: Pubkey, + #[borsh(skip)] + pub serum_vault_signer: Pubkey, + #[borsh(skip)] + pub user_source_token_account: Pubkey, + #[borsh(skip)] + pub user_destination_token_account: Pubkey, + #[borsh(skip)] + pub user_source_owner: Pubkey, +} + +/// 事件特性 +pub trait EventTrait: Sized + std::fmt::Debug { + fn from_bytes(bytes: &[u8]) -> ClientResult; + fn discriminator() -> &'static [u8]; +} + +/// 从字节中提取鉴别器 +pub fn extract_discriminator(length: usize, data: &[u8]) -> Option<(&[u8], &[u8])> { + if data.len() < length { + return None; + } + Some((&data[..length], &data[length..])) +} + +/// 事件鉴别器常量 +pub mod discriminators { + pub const V4_SWAP_IX: &u8 = &9; + + pub const SWAP_BASE_INPUT_IX: &[u8] = &[143, 190, 90, 218, 196, 30, 51, 222]; + pub const SWAP_BASE_OUTPUT_IX: &[u8] = &[55, 217, 98, 86, 163, 74, 180, 173]; +} diff --git a/src/common/raydium/logs_events.rs b/src/common/raydium/logs_events.rs new file mode 100644 index 0000000..7f32854 --- /dev/null +++ b/src/common/raydium/logs_events.rs @@ -0,0 +1,17 @@ +use crate::common::raydium::logs_data::{discriminators, SwapBaseInputEvent, V4SwapEvent}; +use crate::common::raydium::SwapBaseOutputEvent; +use base64::engine::general_purpose; +use base64::Engine; +use borsh::BorshDeserialize; + +pub const PROGRAM_DATA: &str = "Program data: "; +pub const PROGRAM_LOG_PREFIX: &str = "Program log: ray_log: "; + +/// Raydium事件枚举 +#[derive(Debug)] +pub enum RaydiumEvent { + V4Swap(V4SwapEvent), + SwapBaseInput(SwapBaseInputEvent), + SwapBaseOutput(SwapBaseOutputEvent), + Error(String), +} diff --git a/src/common/raydium/logs_filters.rs b/src/common/raydium/logs_filters.rs new file mode 100755 index 0000000..642ce4e --- /dev/null +++ b/src/common/raydium/logs_filters.rs @@ -0,0 +1,62 @@ +use crate::common::raydium::logs_data::RaydiumInstruction; +use crate::common::raydium::logs_parser::parse_raydium_instruction; +use crate::constants::raydium::accounts; +use crate::error::ClientResult; +use solana_sdk::transaction::VersionedTransaction; + +pub struct LogFilter; + +impl LogFilter { + /// 解析Raydium编译后的指令并返回指令类型和数据 + pub fn parse_raydium_compiled_instruction( + versioned_tx: VersionedTransaction, + ) -> ClientResult> { + let compiled_instructions = versioned_tx.message.instructions(); + let accounts = versioned_tx.message.static_account_keys(); + let ammv4_program_id = accounts::AMMV4_PROGRAM; + let cpmm_program_id = accounts::CPMM_PROGRAM; + let raydium_index = accounts.iter().position(|key| key == &ammv4_program_id); + let cpmm_index = accounts.iter().position(|key| key == &cpmm_program_id); + let mut instructions: Vec = Vec::new(); + if let Some(index) = raydium_index { + for instruction in compiled_instructions { + if instruction.program_id_index as usize == index { + let all_accounts_valid = instruction + .accounts + .iter() + .all(|&acc_idx| (acc_idx as usize) < accounts.len()); + if !all_accounts_valid { + continue; + } + + if let Some(parsed_instruction) = + parse_raydium_instruction(instruction, accounts, &ammv4_program_id) + { + instructions.push(parsed_instruction); + } + } + } + } + if let Some(index) = cpmm_index { + for instruction in compiled_instructions { + if instruction.program_id_index as usize == index { + let all_accounts_valid = instruction + .accounts + .iter() + .all(|&acc_idx| (acc_idx as usize) < accounts.len()); + if !all_accounts_valid { + continue; + } + + if let Some(parsed_instruction) = + parse_raydium_instruction(instruction, accounts, &cpmm_program_id) + { + instructions.push(parsed_instruction); + } + } + } + } + + Ok(instructions) + } +} diff --git a/src/common/raydium/logs_parser.rs b/src/common/raydium/logs_parser.rs new file mode 100755 index 0000000..f039273 --- /dev/null +++ b/src/common/raydium/logs_parser.rs @@ -0,0 +1,155 @@ +use crate::common::raydium::{ + logs_data::{discriminators, RaydiumInstruction, V4SwapEvent}, + logs_events::RaydiumEvent, +}; +use crate::error::ClientResult; + +use solana_sdk::instruction::CompiledInstruction; +use solana_sdk::pubkey::Pubkey; +use std::time::{SystemTime, UNIX_EPOCH}; + +/// 获取当前时间戳 +fn current_timestamp() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("Time went backwards") + .as_secs() as i64 +} + +/// 从指令中解析Raydium指令 +pub fn parse_raydium_instruction( + instruction: &CompiledInstruction, + _accounts: &[Pubkey], + program_id: &Pubkey, +) -> Option { + if instruction.data.len() < 8 { + return None; + } + + if program_id == &crate::constants::raydium::accounts::CPMM_PROGRAM { + let discriminator = &instruction.data[..8]; + let data = &instruction.data[8..]; + + let accounts: Vec = instruction + .accounts + .iter() + .map(|&idx| _accounts[idx as usize]) + .collect(); + + match discriminator { + d if d == discriminators::SWAP_BASE_INPUT_IX => { + if data.len() < 16 || accounts.len() < 12 { + return None; + } + let amount_in = u64::from_le_bytes(data[0..8].try_into().ok()?); + let minimum_amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); + + Some(RaydiumInstruction::SwapBaseInput( + crate::common::raydium::SwapBaseInputEvent { + timestamp: current_timestamp(), + amount_in: amount_in, + minimum_amount_out: minimum_amount_out, + + payer: accounts[0], + authority: accounts[1], + amm_config: accounts[2], + pool_state: accounts[3], + input_token_account: accounts[4], + output_token_account: accounts[5], + input_vault: accounts[6], + output_vault: accounts[7], + input_token_program: accounts[8], + output_token_program: accounts[9], + input_token_mint: accounts[10], + output_token_mint: accounts[11], + observation_state: accounts[12], + ..Default::default() + }, + )) + } + d if d == discriminators::SWAP_BASE_OUTPUT_IX => { + println!("data.len(): {:?}", data.len()); + println!("accounts.len(): {:?}", accounts.len()); + if data.len() < 16 || accounts.len() < 12 { + return None; + } + let max_amount_in = u64::from_le_bytes(data[0..8].try_into().ok()?); + let amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); + + Some(RaydiumInstruction::SwapBaseOutput( + crate::common::raydium::SwapBaseOutputEvent { + timestamp: current_timestamp(), + max_amount_in: max_amount_in, + amount_out: amount_out, + + payer: accounts[0], + authority: accounts[1], + amm_config: accounts[2], + pool_state: accounts[3], + input_token_account: accounts[4], + output_token_account: accounts[5], + input_vault: accounts[6], + output_vault: accounts[7], + input_token_program: accounts[8], + output_token_program: accounts[9], + input_token_mint: accounts[10], + output_token_mint: accounts[11], + observation_state: accounts[12], + ..Default::default() + }, + )) + } + _ => Some(RaydiumInstruction::Other), + } + } else if program_id == &crate::constants::raydium::accounts::AMMV4_PROGRAM { + let discriminator = &instruction.data[0]; + let data = &instruction.data[1..]; + + let mut accounts: Vec = instruction + .accounts + .iter() + .map(|&idx| _accounts[idx as usize]) + .collect(); + + match discriminator { + d if d == discriminators::V4_SWAP_IX => { + if data.len() < 16 || accounts.len() < 17 { + return None; + } + let amount_in = u64::from_le_bytes(data[0..8].try_into().ok()?); + let minimum_amount_out = u64::from_le_bytes(data[8..16].try_into().ok()?); + + if accounts.len() == 17 { + accounts.insert(4, Pubkey::default()); + } + + Some(RaydiumInstruction::V4Swap(V4SwapEvent { + timestamp: current_timestamp(), + amount_in: amount_in, + minimum_amount_out: minimum_amount_out, + amm: accounts[1], + amm_authority: accounts[2], + amm_open_orders: accounts[3], + amm_target_orders: accounts[4], + pool_coin_token_account: accounts[5], + pool_pc_token_account: accounts[6], + serum_program: accounts[7], + serum_market: accounts[8], + serum_bids: accounts[9], + serum_asks: accounts[10], + serum_event_queue: accounts[11], + serum_coin_vault_account: accounts[12], + serum_pc_vault_account: accounts[13], + serum_vault_signer: accounts[14], + user_source_token_account: accounts[15], + user_destination_token_account: accounts[16], + user_source_owner: accounts[17], + ..Default::default() + })) + } + _ => Some(RaydiumInstruction::Other), + } + } else { + None + } +} diff --git a/src/common/raydium/mod.rs b/src/common/raydium/mod.rs new file mode 100644 index 0000000..856a6f0 --- /dev/null +++ b/src/common/raydium/mod.rs @@ -0,0 +1,9 @@ +pub mod logs_data; +pub mod logs_parser; +pub mod logs_filters; +pub mod logs_events; + +pub use logs_data::*; +pub use logs_parser::*; +pub use logs_filters::*; +pub use logs_events::*; \ No newline at end of file diff --git a/src/constants/mod.rs b/src/constants/mod.rs index 945a789..561af57 100644 --- a/src/constants/mod.rs +++ b/src/constants/mod.rs @@ -1,5 +1,6 @@ pub mod pumpfun; pub mod pumpswap; +pub mod raydium; pub mod trade_type { pub const COPY_BUY: &'static str = "copy_buy"; diff --git a/src/constants/raydium/mod.rs b/src/constants/raydium/mod.rs new file mode 100755 index 0000000..376584a --- /dev/null +++ b/src/constants/raydium/mod.rs @@ -0,0 +1,85 @@ +//! Constants used by the crate. +//! +//! This module contains various constants used throughout the crate, including: +//! +//! - Seeds for deriving Program Derived Addresses (PDAs) +//! - Program account addresses and public keys +//! +//! The constants are organized into submodules for better organization: +//! +//! - `seeds`: Contains seed values used for PDA derivation +//! - `accounts`: Contains important program account addresses + +/// Constants used as seeds for deriving PDAs (Program Derived Addresses) +pub mod seeds { + /// Seed for the global state PDA + pub const GLOBAL_SEED: &[u8] = b"global"; +} + +/// Constants related to program accounts and authorities +pub mod accounts { + use solana_sdk::{pubkey, pubkey::Pubkey}; + + pub const JITO_TIP_ACCOUNTS: &[&str] = &[ + "96gYZGLnJYVFmbjzopPSU6QiEV5fGqZNyN9nmNhvrZU5", + "HFqU5x63VTqvQss8hp11i4wVV8bD44PvwucfZ2bU7gRe", + "Cw8CFyM9FkoMi7K7Crf6HNQqf4uEMzpKw6QNghXLvLkY", + "ADaUMid9yfUytqMBgopwjb2DTLSokTSzL1zt6iGPaS49", + "DfXygSm4jCyNCybVYYK6DwvWqjKee8pbDmJGcLWNDXjh", + "ADuUkR4vqLUMWXxW9gh6D6L8pMSawimctcNZ5pGwDcEt", + "DttWaMuVvTiduZRnguLF7jNxTgiMBZ1hyAumKUiL2KRL", + "3AVi9Tg9Uo68tJfuvoKvqKNWKkC5wPdSSdeBnizKZ6jT", + ]; + + /// Tip accounts + pub const NEXTBLOCK_TIP_ACCOUNTS: &[&str] = &[ + "NextbLoCkVtMGcV47JzewQdvBpLqT9TxQFozQkN98pE", + "NexTbLoCkWykbLuB1NkjXgFWkX9oAtcoagQegygXXA2", + "NeXTBLoCKs9F1y5PJS9CKrFNNLU1keHW71rfh7KgA1X", + "NexTBLockJYZ7QD7p2byrUa6df8ndV2WSd8GkbWqfbb", + "neXtBLock1LeC67jYd1QdAa32kbVeubsfPNTJC1V5At", + "nEXTBLockYgngeRmRrjDV31mGSekVPqZoMGhQEZtPVG", + "NEXTbLoCkB51HpLBLojQfpyVAMorm3zzKg7w9NFdqid", + "nextBLoCkPMgmG8ZgJtABeScP35qLa2AMCNKntAP7Xc", + ]; + + pub const ZEROSLOT_TIP_ACCOUNTS: &[&str] = &[ + "Eb2KpSC8uMt9GmzyAEm5Eb1AAAgTjRaXWFjKyFXHZxF3", + "FCjUJZ1qozm1e8romw216qyfQMaaWKxWsuySnumVCCNe", + "ENxTEjSQ1YabmUpXAdCgevnHQ9MHdLv8tzFiuiYJqa13", + "6rYLG55Q9RpsPGvqdPNJs4z5WTxJVatMB8zV3WJhs5EK", + "Cix2bHfqPcKcM233mzxbLk14kSggUUiz2A87fJtGivXr", + ]; + + pub const NOZOMI_TIP_ACCOUNTS: &[&str] = &[ + "TEMPaMeCRFAS9EKF53Jd6KpHxgL47uWLcpFArU1Fanq", + "noz3jAjPiHuBPqiSPkkugaJDkJscPuRhYnSpbi8UvC4", + "noz3str9KXfpKknefHji8L1mPgimezaiUyCHYMDv1GE", + "noz6uoYCDijhu1V7cutCpwxNiSovEwLdRHPwmgCGDNo", + "noz9EPNcT7WH6Sou3sr3GGjHQYVkN3DNirpbvDkv9YJ", + "nozc5yT15LazbLTFVZzoNZCwjh3yUtW86LoUyqsBu4L", + "nozFrhfnNGoyqwVuwPAW4aaGqempx4PU6g6D9CJMv7Z", + "nozievPk7HyK1Rqy1MPJwVQ7qQg2QoJGyP71oeDwbsu", + "noznbgwYnBLDHu8wcQVCEw6kDrXkPdKkydGJGNXGvL7", + "nozNVWs5N8mgzuD3qigrCG2UoKxZttxzZ85pvAQVrbP", + "nozpEGbwx4BcGp6pvEdAh1JoC2CQGZdU6HbNP1v2p6P", + "nozrhjhkCr3zXT3BiT4WCodYCUFeQvcdUkM7MqhKqge", + "nozrwQtWhEdrA6W8dkbt9gnUaMs52PdAv5byipnadq3", + "nozUacTVWub3cL4mJmGCYjKZTnE9RbdY5AP46iQgbPJ", + "nozWCyTPppJjRuw2fpzDhhWbW355fzosWSzrrMYB1Qk", + "nozWNju6dY353eMkMqURqwQEoM3SFgEKC6psLCSfUne", + "nozxNBgWohjR75vdspfxR5H9ceC7XXH99xpxhVGt3Bb", + ]; + + pub const AMMV4_PROGRAM: Pubkey = pubkey!("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"); + pub const CPMM_PROGRAM: Pubkey = pubkey!("CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C"); +} + +pub mod trade { + pub const TRADER_TIP_AMOUNT: u64 = 100000; // 0.0001 SOL in lamports + pub const DEFAULT_SLIPPAGE: u64 = 1000; // 10% + pub const DEFAULT_COMPUTE_UNIT_LIMIT: u32 = 78000; + pub const DEFAULT_COMPUTE_UNIT_PRICE: u64 = 500000; + pub const DEFAULT_BUY_TIP_FEE: u64 = 600000; // 0.0006 SOL in lamports + pub const DEFAULT_SELL_TIP_FEE: u64 = 100000; // 0.0001 SOL in lamports +} diff --git a/src/grpc/shred_stream.rs b/src/grpc/shred_stream.rs index a3c1646..c2d5faa 100755 --- a/src/grpc/shred_stream.rs +++ b/src/grpc/shred_stream.rs @@ -8,6 +8,7 @@ use log::error; use solana_sdk::transaction::VersionedTransaction; use crate::common::pumpswap::PumpSwapInstruction; +use crate::common::raydium::{RaydiumEvent, RaydiumInstruction}; use crate::common::AnyResult; use solana_sdk::pubkey::Pubkey; @@ -16,6 +17,7 @@ use crate::common::pumpfun::logs_events::PumpfunEvent; use crate::common::pumpswap::logs_events::PumpSwapEvent; use crate::common::pumpfun::logs_filters::LogFilter; use crate::common::pumpswap::logs_filters::LogFilter as PumpswapLogFilter; +use crate::common::raydium::logs_filters::LogFilter as RaydiumLogFilter; use crate::swqos::jito_grpc::shredstream::shredstream_proxy_client::ShredstreamProxyClient; use crate::swqos::jito_grpc::shredstream::SubscribeEntriesRequest; @@ -121,6 +123,47 @@ impl ShredStreamGrpc { Ok(()) } + pub async fn shredstream_subscribe_raydium(&self, callback: F) -> AnyResult<()> + where + F: Fn(RaydiumEvent) + Send + Sync + 'static, + { + let request = tonic::Request::new(SubscribeEntriesRequest {}); + let mut client = (*self.shredstream_client).clone(); + let mut stream = client.subscribe_entries(request).await?.into_inner(); + let (mut tx, mut rx) = mpsc::channel::(CHANNEL_SIZE); + let callback = Box::new(callback); + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Ok(entries) = bincode::deserialize::>(&msg.entries) { + for entry in entries { + for transaction in entry.transactions { + let _ = tx.try_send(TransactionWithSlot { + transaction: transaction.clone(), + slot: msg.slot, + }); + } + } + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + while let Some(transaction_with_slot) = rx.next().await { + if let Err(e) = Self::process_raydium_transaction(transaction_with_slot, &*callback).await { + error!("Error processing transaction: {:?}", e); + } + } + + Ok(()) + } + async fn process_pumpfun_transaction(transaction_with_slot: TransactionWithSlot, callback: &F, bot_wallet: Option) -> AnyResult<()> where F: Fn(PumpfunEvent) + Send + Sync, @@ -211,4 +254,35 @@ impl ShredStreamGrpc { } Ok(()) } + + async fn process_raydium_transaction(transaction_with_slot: TransactionWithSlot, callback: &F) -> AnyResult<()> + where + F: Fn(RaydiumEvent) + Send + Sync, + { + let slot = transaction_with_slot.slot; + let versioned_tx = transaction_with_slot.transaction; + let signature = versioned_tx.signatures[0].to_string(); + let instructions: Vec = RaydiumLogFilter::parse_raydium_compiled_instruction(versioned_tx).unwrap(); + for instruction in instructions { + match instruction { + RaydiumInstruction::V4Swap(mut v4_swap_event) => { + v4_swap_event.slot = slot; + v4_swap_event.signature = signature.clone(); + callback(RaydiumEvent::V4Swap(v4_swap_event)); + } + RaydiumInstruction::SwapBaseInput(mut swap_base_input_event) => { + swap_base_input_event.slot = slot; + swap_base_input_event.signature = signature.clone(); + callback(RaydiumEvent::SwapBaseInput(swap_base_input_event)); + } + RaydiumInstruction::SwapBaseOutput(mut swap_base_output_event) => { + swap_base_output_event.slot = slot; + swap_base_output_event.signature = signature.clone(); + callback(RaydiumEvent::SwapBaseOutput(swap_base_output_event)); + } + _ => {} + } + } + Ok(()) + } } diff --git a/src/grpc/yellow_stone.rs b/src/grpc/yellow_stone.rs index 40f809a..2c7a22f 100755 --- a/src/grpc/yellow_stone.rs +++ b/src/grpc/yellow_stone.rs @@ -535,4 +535,113 @@ impl YellowstoneGrpc { Ok(()) } + + // ------------------------------------------------------------ + // Raydium + // ------------------------------------------------------------ + + /// 订阅Raydium事件 + pub async fn subscribe_raydium(&self, callback: F) -> AnyResult<()> + where + F: Fn(crate::common::raydium::logs_events::RaydiumEvent) + Send + Sync + 'static, + { + // 使用constants中定义的AMM_PROGRAM + let raydium_v4_program_id = crate::constants::raydium::accounts::AMMV4_PROGRAM; + let raydium_cpmm_program_id = crate::constants::raydium::accounts::CPMM_PROGRAM; + let addrs = vec![raydium_v4_program_id.to_string(), raydium_cpmm_program_id.to_string()]; + + // 创建过滤器 + let transactions = self.get_subscribe_request_filter(addrs, vec![], vec![]); + + // 订阅事件 + let (mut subscribe_tx, mut stream) = self.subscribe_with_request(transactions).await?; + + // 创建通道 + let (mut tx, mut rx) = mpsc::channel::(1000); + + // 创建回调函数 + let callback = Box::new(callback); + + // 启动处理流的任务 + tokio::spawn(async move { + while let Some(message) = stream.next().await { + match message { + Ok(msg) => { + if let Err(e) = + Self::handle_stream_message(msg, &mut tx, &mut subscribe_tx).await + { + error!("Error handling message: {:?}", e); + break; + } + } + Err(error) => { + error!("Stream error: {error:?}"); + break; + } + } + } + }); + + // 处理交易 + while let Some(transaction_pretty) = rx.next().await { + if let Err(e) = Self::process_raydium_transaction(transaction_pretty, &*callback).await + { + error!("Error processing transaction: {:?}", e); + } + } + + Ok(()) + } + + /// 处理Raydium交易 + async fn process_raydium_transaction( + transaction_pretty: TransactionPretty, + callback: &F, + ) -> AnyResult<()> + where + F: Fn(crate::common::raydium::logs_events::RaydiumEvent) + Send + Sync, + { + let slot = transaction_pretty.slot; + let trade_raw: solana_transaction_status::EncodedTransactionWithStatusMeta = + transaction_pretty.tx; + + // 检查交易元数据 + let meta = trade_raw + .meta + .as_ref() + .ok_or_else(|| anyhow::anyhow!("Missing transaction metadata"))?; + + // 检查交易是否成功 + if meta.err.is_some() { + return Ok(()); + } + + if let Some(versioned_tx) = trade_raw.transaction.decode() { + let signature = versioned_tx.signatures[0].to_string(); + let instructions: Vec = + crate::common::raydium::logs_filters::LogFilter::parse_raydium_compiled_instruction(versioned_tx).unwrap(); + for instruction in instructions { + match instruction { + crate::common::raydium::logs_data::RaydiumInstruction::V4Swap(mut v4_swap_event) => { + v4_swap_event.slot = slot; + v4_swap_event.signature = signature.clone(); + callback(crate::common::raydium::logs_events::RaydiumEvent::V4Swap(v4_swap_event)); + } + crate::common::raydium::logs_data::RaydiumInstruction::SwapBaseInput(mut swap_base_input_event) => { + swap_base_input_event.slot = slot; + swap_base_input_event.signature = signature.clone(); + callback(crate::common::raydium::logs_events::RaydiumEvent::SwapBaseInput(swap_base_input_event)); + } + crate::common::raydium::logs_data::RaydiumInstruction::SwapBaseOutput(mut swap_base_output_event) => { + swap_base_output_event.slot = slot; + swap_base_output_event.signature = signature.clone(); + callback(crate::common::raydium::logs_events::RaydiumEvent::SwapBaseOutput(swap_base_output_event)); + } + _ => {} + } + } + } + + Ok(()) + } } diff --git a/src/main.rs b/src/main.rs index 2bbb703..6e0ce84 100755 --- a/src/main.rs +++ b/src/main.rs @@ -8,6 +8,7 @@ use pumpfun_sdk::{ logs_subscribe::{stop_subscription, tokens_subscription}, }, pumpswap::{self, PumpSwapEvent}, + raydium::{self, RaydiumEvent}, AnyResult, Cluster, PriorityFee, }, grpc::{ShredStreamGrpc, YellowstoneGrpc}, @@ -25,7 +26,9 @@ async fn main() -> Result<(), Box> { // test_pumpfun_with_grpc().await?; // test_pumpswap_with_shreds().await?; // test_pumpswap_with_grpc().await?; - test_sell().await?; + // test_raydium_with_shreds().await?; + test_raydium_with_grpc().await?; + // test_sell().await?; Ok(()) } @@ -190,6 +193,69 @@ async fn test_pumpswap_with_grpc() -> Result<(), Box> { Ok(()) } +async fn test_raydium_with_shreds() -> Result<(), Box> { + // 使用 ShredStream 客户端订阅 Raydium 事件 + println!("正在订阅 Raydium ShredStream 事件..."); + + let grpc_client = ShredStreamGrpc::new("http://140.82.2.197:10800".to_string()).await?; + + // 定义回调函数处理 Raydium 事件 + let callback = |event: RaydiumEvent| { + match event { + RaydiumEvent::V4Swap(v4_swap_event) => { + println!("v4_swap_event: {:?}", v4_swap_event); + } + RaydiumEvent::SwapBaseInput(swap_base_input_event) => { + println!("swap_base_input_event: {:?}", swap_base_input_event); + } + RaydiumEvent::SwapBaseOutput(swap_base_output_event) => { + println!("swap_base_output_event: {:?}", swap_base_output_event); + } + RaydiumEvent::Error(err) => { + println!("error: {}", err); + } + } + }; + // 订阅 Raydium 事件 + println!("开始监听 Raydium 事件，按 Ctrl+C 停止..."); + + grpc_client.shredstream_subscribe_raydium(callback).await?; + + Ok(()) +} + +async fn test_raydium_with_grpc() -> Result<(), Box> { + // 使用 GRPC 客户端订阅 Raydium 事件 + println!("正在订阅 Raydium GRPC 事件..."); + + let grpc = YellowstoneGrpc::new( + "https://solana-yellowstone-grpc.publicnode.com:443".to_string(), + None, + )?; + + // 定义回调函数处理 PumpSwap 事件 + let callback = |event: RaydiumEvent| match event { + RaydiumEvent::V4Swap(v4_swap_event) => { + println!("v4_swap_event: {:?}", v4_swap_event); + } + RaydiumEvent::SwapBaseInput(swap_base_input_event) => { + println!("swap_base_input_event: {:?}", swap_base_input_event); + } + RaydiumEvent::SwapBaseOutput(swap_base_output_event) => { + println!("swap_base_output_event: {:?}", swap_base_output_event); + } + RaydiumEvent::Error(err) => { + println!("error: {}", err); + } + }; + // 订阅 Raydium 事件 + println!("开始监听 Raydium 事件，按 Ctrl+C 停止..."); + + grpc.subscribe_raydium(callback).await?; + + Ok(()) +} + async fn test_sell() -> AnyResult<()> { let payer = Keypair::new(); // Define cluster configuration