gavindiaz/sol-trade-sdk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rebuild code

Browse Source
This commit is contained in:
wood
2025-07-06 22:06:44 +08:00
parent 9e202a1e5e
commit 3ac39508d9
34 changed files with 635 additions and 552 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/types.rs
+7 -4
View File
@@ -3,12 +3,15 @@ use std::sync::Arc;
use solana_client::rpc_client::RpcClient;
use solana_sdk::{commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Keypair};
use serde::Deserialize;
use crate::{constants::pumpfun::trade::{DEFAULT_BUY_TIP_FEE, DEFAULT_COMPUTE_UNIT_LIMIT, DEFAULT_COMPUTE_UNIT_PRICE, DEFAULT_RPC_UNIT_LIMIT, DEFAULT_RPC_UNIT_PRICE, DEFAULT_SELL_TIP_FEE}, swqos::FeeClient};
use crate::{constants::pumpfun::trade::{DEFAULT_BUY_TIP_FEE, DEFAULT_COMPUTE_UNIT_LIMIT, DEFAULT_COMPUTE_UNIT_PRICE, DEFAULT_RPC_UNIT_LIMIT, DEFAULT_RPC_UNIT_PRICE, DEFAULT_SELL_TIP_FEE}, swqos::SwqosClient};
#[derive(Debug, Clone, PartialEq)]
pub enum FeeType {
pub enum SwqosType {
Jito,
NextBlock,
ZeroSlot,
Nozomi,
Rpc,
}
#[derive(Debug, Clone)]
@@ -104,11 +107,11 @@ pub struct MethodArgs {
pub payer: Arc<Keypair>,
pub rpc: Arc<RpcClient>,
pub nonblocking_rpc: Arc<SolanaRpcClient>,
pub jito_client: Arc<FeeClient>,
pub jito_client: Arc<SwqosClient>,
}
impl MethodArgs {
pub fn new(payer: Arc<Keypair>, rpc: Arc<RpcClient>, nonblocking_rpc: Arc<SolanaRpcClient>, jito_client: Arc<FeeClient>) -> Self {
pub fn new(payer: Arc<Keypair>, rpc: Arc<RpcClient>, nonblocking_rpc: Arc<SolanaRpcClient>, jito_client: Arc<SwqosClient>) -> Self {
Self { payer, rpc, nonblocking_rpc, jito_client }
}
}
src/grpc/shred_stream.rs
+2 -2
View File
@@ -18,8 +18,8 @@ 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;
use crate::protos::shredstream::shredstream_proxy_client::ShredstreamProxyClient;
use crate::protos::shredstream::SubscribeEntriesRequest;
const CHANNEL_SIZE: usize = 1000;
src/lib.rs
+31 -25
View File
@@ -9,11 +9,12 @@ pub mod swqos;
pub mod pumpfun;
pub mod pumpswap;
pub mod trading;
pub mod protos;
use std::sync::Arc;
use std::sync::Mutex;
use swqos::{FeeClient, JitoClient, NextBlockClient, NozomiClient, SolRpcClient, ZeroSlotClient};
use swqos::SwqosClient;
use rustls::crypto::{ring::default_provider, CryptoProvider};
use solana_hash::Hash;
use solana_sdk::{
@@ -30,6 +31,11 @@ use constants::trade_type::{COPY_BUY, SNIPER_BUY};
use constants::trade_platform::{PUMPFUN, PUMPFUN_SWAP, RAYDIUM};
use accounts::BondingCurveAccount;
use crate::swqos::jito::JitoClient;
use crate::swqos::nextblock::NextBlockClient;
use crate::swqos::nozomi::NozomiClient;
use crate::swqos::solana_rpc::SolRpcClient;
use crate::swqos::zeroslot::ZeroSlotClient;
use crate::trading::core::params::PumpFunParams;
use crate::trading::core::params::PumpFunSellParams;
use crate::trading::core::params::PumpSwapParams;
@@ -40,7 +46,7 @@ use crate::trading::SellWithTipParams;
pub struct SolanaTrade {
pub payer: Arc<Keypair>,
pub rpc: Arc<SolanaRpcClient>,
pub fee_clients: Vec<Arc<FeeClient>>,
pub swqos_clients: Vec<Arc<SwqosClient>>,
pub priority_fee: PriorityFee,
pub cluster: Cluster,
}
@@ -52,7 +58,7 @@ impl Clone for SolanaTrade {
Self {
payer: self.payer.clone(),
rpc: self.rpc.clone(),
fee_clients: self.fee_clients.clone(),
swqos_clients: self.swqos_clients.clone(),
priority_fee: self.priority_fee.clone(),
cluster: self.cluster.clone(),
}
@@ -76,14 +82,14 @@ impl SolanaTrade {
cluster.clone().commitment
);
let rpc = Arc::new(rpc);
let mut fee_clients: Vec<Arc<FeeClient>> = vec![];
let mut swqos_clients: Vec<Arc<SwqosClient>> = vec![];
if cluster.clone().use_jito {
let jito_client = JitoClient::new(
cluster.clone().rpc_url,
cluster.clone().block_engine_url
).await.expect("Failed to create Jito client");
fee_clients.push(Arc::new(jito_client));
swqos_clients.push(Arc::new(jito_client));
}
if cluster.clone().use_zeroslot {
@@ -93,7 +99,7 @@ impl SolanaTrade {
cluster.clone().zeroslot_auth_token
);
fee_clients.push(Arc::new(zeroslot_client));
swqos_clients.push(Arc::new(zeroslot_client));
}
if cluster.clone().use_nozomi {
@@ -103,7 +109,7 @@ impl SolanaTrade {
cluster.clone().nozomi_auth_token
);
fee_clients.push(Arc::new(nozomi_client));
swqos_clients.push(Arc::new(nozomi_client));
}
if cluster.clone().use_nextblock {
@@ -113,18 +119,18 @@ impl SolanaTrade {
cluster.clone().nextblock_auth_token
);
fee_clients.push(Arc::new(nextblock_client));
swqos_clients.push(Arc::new(nextblock_client));
}
if cluster.clone().use_rpc {
let rpc_client = SolRpcClient::new(rpc.clone());
fee_clients.push(Arc::new(rpc_client));
swqos_clients.push(Arc::new(rpc_client));
}
let instance = Self {
payer,
rpc,
fee_clients,
swqos_clients,
priority_fee: cluster.clone().priority_fee,
cluster: cluster.clone(),
};
@@ -192,7 +198,7 @@ impl SolanaTrade {
) -> Result<(), anyhow::Error> {
pumpfun::create::create_and_buy_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
payer,
mint,
ipfs,
@@ -350,7 +356,7 @@ impl SolanaTrade {
priority_fee.buy_tip_fees = vec![custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap(),custom_buy_tip_fee.unwrap()];
}
pumpfun::buy::buy_with_tip(
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -384,7 +390,7 @@ impl SolanaTrade {
.as_any()
.downcast_ref::<PumpFunParams>() {
pumpfun::buy::buy_with_tip(
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -402,7 +408,7 @@ impl SolanaTrade {
.downcast_ref::<PumpSwapParams>() {
pumpswap::buy::buy_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -441,7 +447,7 @@ impl SolanaTrade {
}
if trade_platform == PUMPFUN {
pumpfun::buy::buy_with_tip(
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -456,7 +462,7 @@ impl SolanaTrade {
} else if trade_platform == PUMPFUN_SWAP {
pumpswap::buy::buy_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -595,7 +601,7 @@ impl SolanaTrade {
if trade_platform == PUMPFUN {
pumpfun::sell::sell_by_percent_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -608,7 +614,7 @@ impl SolanaTrade {
} else if trade_platform == PUMPFUN_SWAP {
pumpswap::sell::sell_by_percent_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -639,7 +645,7 @@ impl SolanaTrade {
if trade_platform == PUMPFUN {
pumpfun::sell::sell_by_amount_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -651,7 +657,7 @@ impl SolanaTrade {
} else if trade_platform == PUMPFUN_SWAP {
pumpswap::sell::sell_by_amount_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -681,7 +687,7 @@ impl SolanaTrade {
) -> Result<(), anyhow::Error> {
pumpfun::sell::sell_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -808,7 +814,7 @@ impl SolanaTrade {
.downcast_ref::<PumpFunSellParams>() {
pumpfun::sell::sell_by_percent_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -824,7 +830,7 @@ impl SolanaTrade {
.downcast_ref::<PumpSwapParams>() {
pumpswap::sell::sell_by_percent_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -858,7 +864,7 @@ impl SolanaTrade {
.downcast_ref::<PumpFunSellParams>() {
pumpfun::sell::sell_by_amount_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
@@ -873,7 +879,7 @@ impl SolanaTrade {
.downcast_ref::<PumpSwapParams>() {
pumpswap::sell::sell_by_amount_with_tip(
self.rpc.clone(),
self.fee_clients.clone(),
self.swqos_clients.clone(),
self.payer.clone(),
mint,
creator,
src/swqos/jito_grpc/auth.rs → src/protos/auth.rs
View File
src/swqos/jito_grpc/block.rs → src/protos/block.rs
View File
src/swqos/jito_grpc/block_engine.rs → src/protos/block_engine.rs
View File
src/swqos/jito_grpc/bundle.rs → src/protos/bundle.rs
View File
src/swqos/jito_grpc/convert.rs → src/protos/convert.rs
+1 -1
View File
@@ -11,7 +11,7 @@ use solana_sdk::{
transaction::VersionedTransaction,
};
use crate::swqos::jito_grpc::{
use crate::protos::{
packet::{
Meta as ProtoMeta, Packet as ProtoPacket, PacketBatch as ProtoPacketBatch,
PacketFlags as ProtoPacketFlags,
src/swqos/jito_grpc/mod.rs → src/protos/mod.rs
+3
View File
@@ -9,3 +9,6 @@ pub mod shared;
pub mod shredstream;
pub mod trace_shred;
pub mod convert;
pub mod nextblock_grpc;
pub mod searcher_client;
pub mod token_authenticator;
src/swqos/api.rs → src/protos/nextblock_grpc.rs
View File
src/swqos/jito_grpc/packet.rs → src/protos/packet.rs
View File
src/swqos/jito_grpc/relayer.rs → src/protos/relayer.rs
View File
src/swqos/jito_grpc/searcher.rs → src/protos/searcher.rs
View File
src/swqos/searcher_client.rs → src/protos/searcher_client.rs
+2 -3
View File
@@ -3,7 +3,7 @@ use std::{
time::{Duration, Instant},
};
use crate::swqos::jito_grpc::{
use crate::protos::{
bundle::{
Bundle, BundleResult,
},
@@ -25,8 +25,7 @@ use yellowstone_grpc_client::ClientTlsConfig;
use crate::swqos::common::poll_transaction_confirmation;
use crate::common::SolanaRpcClient;
use super::TradeType;
use crate::swqos::TradeType;
#[derive(Debug, Error)]
pub enum BlockEngineConnectionError {
src/swqos/jito_grpc/shared.rs → src/protos/shared.rs
View File
src/swqos/jito_grpc/shredstream.rs → src/protos/shredstream.rs
View File
src/swqos/token_authenticator.rs → src/protos/token_authenticator.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use std::{
time::{Duration, SystemTime},
};
use jito_protos::auth::{
use crate::protos::auth::{
auth_service_client::AuthServiceClient, GenerateAuthChallengeRequest,
GenerateAuthTokensRequest, RefreshAccessTokenRequest, Role, Token,
};
src/swqos/jito_grpc/trace_shred.rs → src/protos/trace_shred.rs
View File
src/pumpfun/buy.rs
+3 -3
View File
@@ -1,7 +1,7 @@
use crate::accounts::BondingCurveAccount;
use crate::{
common::{PriorityFee, SolanaRpcClient},
swqos::FeeClient,
swqos::SwqosClient,
trading::{core::params::PumpFunParams, factory::Protocol, BuyParams, TradeFactory},
};
use solana_hash::Hash;
@@ -49,7 +49,7 @@ pub async fn buy(
}
pub async fn buy_with_tip(
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -82,7 +82,7 @@ pub async fn buy_with_tip(
data_size_limit: MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT,
protocol_params,
};
let buy_with_tip_params = buy_params.with_tip(fee_clients);
let buy_with_tip_params = buy_params.with_tip(swqos_clients);
// 执行买入
executor.buy_with_tip(buy_with_tip_params).await?;
Ok(())
src/pumpfun/create.rs
+2 -2
View File
@@ -16,7 +16,7 @@ use spl_associated_token_account::instruction::create_associated_token_account;
use crate::{
common::{PriorityFee, SolanaRpcClient}, constants, instruction,
ipfs::TokenMetadataIPFS, swqos::{FeeClient, TradeType},
ipfs::TokenMetadataIPFS, swqos::{SwqosClient, TradeType},
};
use crate::pumpfun::common::{
@@ -86,7 +86,7 @@ pub async fn create_and_buy(
pub async fn create_and_buy_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
fee_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Keypair,
ipfs: TokenMetadataIPFS,
src/pumpfun/sell.rs
+4 -4
View File
@@ -3,7 +3,7 @@ use crate::trading::{
};
use crate::{
common::{PriorityFee, SolanaRpcClient},
swqos::FeeClient,
swqos::SwqosClient,
};
use anyhow::anyhow;
use solana_hash::Hash;
@@ -99,7 +99,7 @@ pub async fn sell_by_amount(
pub async fn sell_by_percent_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
fee_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -129,7 +129,7 @@ pub async fn sell_by_percent_with_tip(
pub async fn sell_by_amount_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
fee_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -158,7 +158,7 @@ pub async fn sell_by_amount_with_tip(
/// Sell tokens using Jito
pub async fn sell_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
fee_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
src/pumpswap/buy.rs
+3 -3
View File
@@ -2,7 +2,7 @@ use solana_hash::Hash;
use solana_sdk::{pubkey::Pubkey, signature::Keypair};
use std::sync::Arc;
use crate::swqos::FeeClient;
use crate::swqos::SwqosClient;
use crate::trading::{core::params::PumpSwapParams, factory::Protocol, BuyParams, TradeFactory};
use crate::{common::PriorityFee, SolanaRpcClient};
@@ -62,7 +62,7 @@ pub async fn buy(
// Buy tokens using a MEV service
pub async fn buy_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -104,7 +104,7 @@ pub async fn buy_with_tip(
data_size_limit: MAX_LOADED_ACCOUNTS_DATA_SIZE_LIMIT,
protocol_params,
};
let buy_with_tip_params = buy_params.with_tip(fee_clients);
let buy_with_tip_params = buy_params.with_tip(swqos_clients);
// 执行买入
executor.buy_with_tip(buy_with_tip_params).await?;
Ok(())
src/pumpswap/sell.rs
+7 -7
View File
@@ -5,7 +5,7 @@ use std::sync::Arc;
use crate::common::{PriorityFee, SolanaRpcClient};
use crate::pumpswap::common::get_token_balance;
use crate::swqos::FeeClient;
use crate::swqos::SwqosClient;
use crate::trading::{core::params::PumpSwapParams, factory::Protocol, SellParams, TradeFactory};
// Sell tokens to a Pumpswap pool
@@ -140,7 +140,7 @@ pub async fn sell_by_amount(
// Sell tokens using a MEV service
pub async fn sell_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -179,7 +179,7 @@ pub async fn sell_with_tip(
recent_blockhash,
protocol_params,
};
let sell_with_tip_params = sell_params.with_tip(fee_clients);
let sell_with_tip_params = sell_params.with_tip(swqos_clients);
// 执行卖出交易
executor.sell_with_tip(sell_with_tip_params).await?;
Ok(())
@@ -188,7 +188,7 @@ pub async fn sell_with_tip(
// Sell tokens by percentage using a MEV service
pub async fn sell_by_percent_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -212,7 +212,7 @@ pub async fn sell_by_percent_with_tip(
let amount = balance_u64 * percent / 100;
sell_with_tip(
rpc,
fee_clients,
swqos_clients,
payer,
mint,
creator,
@@ -233,7 +233,7 @@ pub async fn sell_by_percent_with_tip(
// Sell tokens by amount using a MEV service
pub async fn sell_by_amount_with_tip(
rpc: Arc<SolanaRpcClient>,
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
mint: Pubkey,
creator: Pubkey,
@@ -255,7 +255,7 @@ pub async fn sell_by_amount_with_tip(
sell_with_tip(
rpc,
fee_clients,
swqos_clients,
payer,
mint,
creator,
src/swqos/jito.rs
Executable
+66
View File
@@ -0,0 +1,66 @@
use tonic::transport::Channel;
use tokio::sync::Mutex;
use rand::{rng, seq::IteratorRandom};
use anyhow::{anyhow, Result};
use std::sync::Arc;
use solana_sdk::{transaction::VersionedTransaction, signature::Signature};
use crate::protos::searcher::searcher_service_client::SearcherServiceClient;
use crate::protos::searcher_client::{self, get_searcher_client_no_auth, send_bundle_with_confirmation};
use crate::swqos::{ClientType, TradeType};
use crate::swqos::SwqosClientTrait;
use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::JITO_TIP_ACCOUNTS};
pub struct JitoClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub searcher_client: Arc<Mutex<SearcherServiceClient<Channel>>>,
}
#[async_trait::async_trait]
impl SwqosClientTrait for JitoClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
self.send_bundle_with_confirmation(trade_type, &vec![transaction.clone()]).await?.first().cloned().ok_or(anyhow!("Failed to send transaction"))
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
self.send_bundle_with_confirmation(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
if let Some(acc) = JITO_TIP_ACCOUNTS.iter().choose(&mut rng()) {
Ok(acc.to_string())
} else {
Err(anyhow!("no valid tip accounts found"))
}
}
fn get_client_type(&self) -> ClientType {
ClientType::Jito
}
}
impl JitoClient {
pub async fn new(rpc_url: String, block_engine_url: String) -> Result<Self> {
let rpc_client = SolanaRpcClient::new(rpc_url);
let searcher_client = get_searcher_client_no_auth(block_engine_url.as_str()).await?;
Ok(Self { rpc_client: Arc::new(rpc_client), searcher_client: Arc::new(Mutex::new(searcher_client)) })
}
pub async fn send_bundle_with_confirmation(
&self,
trade_type: TradeType,
transactions: &Vec<VersionedTransaction>,
) -> Result<Vec<Signature>> {
send_bundle_with_confirmation(self.rpc_client.clone(), trade_type, &transactions, self.searcher_client.clone()).await
}
pub async fn send_bundle_no_wait(
&self,
transactions: &Vec<VersionedTransaction>,
) -> Result<Vec<Signature>> {
searcher_client::send_bundle_no_wait(&transactions, self.searcher_client.clone()).await
}
}
src/swqos/mod.rs
+14 -478
View File
@@ -1,35 +1,17 @@
use api::api_client::ApiClient;
use common::{poll_transaction_confirmation, serialize_smart_transaction_and_encode, serialize_transaction_and_encode};
use solana_client::rpc_config::RpcSendTransactionConfig;
use crate::swqos::jito_grpc::searcher::searcher_service_client::SearcherServiceClient;
use reqwest::Client;
use searcher_client::{get_searcher_client_no_auth, send_bundle_with_confirmation};
use serde_json::json;
use tonic::transport::Channel;
use yellowstone_grpc_client::Interceptor;
use std::{sync::Arc, time::Instant};
use tokio::sync::{Mutex, RwLock};
use solana_sdk::{commitment_config::CommitmentLevel, signature::Signature};
use std::str::FromStr;
use rustls::crypto::{ring::default_provider, CryptoProvider};
use tonic::{service::interceptor::InterceptedService, transport::Uri, Status};
use std::time::Duration;
use solana_transaction_status::UiTransactionEncoding;
use tonic::transport::ClientTlsConfig;
use anyhow::{anyhow, Result};
use rand::{rng, seq::{IndexedRandom, IteratorRandom}};
use solana_sdk::transaction::VersionedTransaction;
use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::{JITO_TIP_ACCOUNTS, NEXTBLOCK_TIP_ACCOUNTS, ZEROSLOT_TIP_ACCOUNTS, NOZOMI_TIP_ACCOUNTS}};
pub mod api;
pub mod common;
pub mod searcher_client;
pub mod jito_grpc;
pub mod solana_rpc;
pub mod jito;
pub mod nextblock;
pub mod zeroslot;
pub mod nozomi;
pub mod types;
use solana_sdk::signature::Signature;
use solana_sdk::transaction::VersionedTransaction;
use tokio::sync::RwLock;
use anyhow::Result;
lazy_static::lazy_static! {
static ref TIP_ACCOUNT_CACHE: RwLock<Vec<String>> = RwLock::new(Vec::new());
@@ -64,458 +46,12 @@ pub enum ClientType {
Rpc,
}
pub type FeeClient = dyn FeeClientTrait + Send + Sync + 'static;
pub type SwqosClient = dyn SwqosClientTrait + Send + Sync + 'static;
#[async_trait::async_trait]
pub trait FeeClientTrait {
pub trait SwqosClientTrait {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature>;
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>>;
fn get_tip_account(&self) -> Result<String>;
fn get_client_type(&self) -> ClientType;
}
#[derive(Clone)]
pub struct SolRpcClient {
pub rpc_client: Arc<SolanaRpcClient>,
}
#[async_trait::async_trait]
impl FeeClientTrait for SolRpcClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
let signature = self.rpc_client.send_transaction_with_config(transaction, RpcSendTransactionConfig{
skip_preflight: true,
preflight_commitment: Some(CommitmentLevel::Processed),
encoding: Some(UiTransactionEncoding::Base64),
max_retries: Some(3),
min_context_slot: Some(0),
}).await?;
let start_time = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" signature: {:?}", signature);
println!(" rpc{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
fn get_tip_account(&self) -> Result<String> {
Ok("".to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::Rpc
}
}
impl SolRpcClient {
pub fn new(rpc_client: Arc<SolanaRpcClient>) -> Self {
Self { rpc_client }
}
}
pub struct JitoClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub searcher_client: Arc<Mutex<SearcherServiceClient<Channel>>>,
}
#[async_trait::async_trait]
impl FeeClientTrait for JitoClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
self.send_bundle_with_confirmation(trade_type, &vec![transaction.clone()]).await?.first().cloned().ok_or(anyhow!("Failed to send transaction"))
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
self.send_bundle_with_confirmation(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String, anyhow::Error> {
if let Some(acc) = JITO_TIP_ACCOUNTS.iter().choose(&mut rng()) {
Ok(acc.to_string())
} else {
Err(anyhow!("no valid tip accounts found"))
}
}
fn get_client_type(&self) -> ClientType {
ClientType::Jito
}
}
impl JitoClient {
pub async fn new(rpc_url: String, block_engine_url: String) -> Result<Self> {
let rpc_client = SolanaRpcClient::new(rpc_url);
let searcher_client = get_searcher_client_no_auth(block_engine_url.as_str()).await?;
Ok(Self { rpc_client: Arc::new(rpc_client), searcher_client: Arc::new(Mutex::new(searcher_client)) })
}
pub async fn send_bundle_with_confirmation(
&self,
trade_type: TradeType,
transactions: &Vec<VersionedTransaction>,
) -> Result<Vec<Signature>, anyhow::Error> {
send_bundle_with_confirmation(self.rpc_client.clone(), trade_type, &transactions, self.searcher_client.clone()).await
}
pub async fn send_bundle_no_wait(
&self,
transactions: &Vec<VersionedTransaction>,
) -> Result<Vec<Signature>, anyhow::Error> {
searcher_client::send_bundle_no_wait(&transactions, self.searcher_client.clone()).await
}
}
#[derive(Clone)]
pub struct MyInterceptor {
auth_token: String,
}
impl MyInterceptor {
pub fn new(auth_token: String) -> Self {
Self { auth_token }
}
}
impl Interceptor for MyInterceptor {
fn call(&mut self, mut request: tonic::Request<()>) -> Result<tonic::Request<()>, Status> {
request.metadata_mut().insert(
"authorization",
tonic::metadata::MetadataValue::from_str(&self.auth_token)
.map_err(|_| Status::invalid_argument("Invalid auth token"))?
);
Ok(request)
}
}
#[derive(Clone)]
pub struct NextBlockClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub client: ApiClient<InterceptedService<Channel, MyInterceptor>>,
}
#[async_trait::async_trait]
impl FeeClientTrait for NextBlockClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *NEXTBLOCK_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NEXTBLOCK_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::NextBlock
}
}
impl NextBlockClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
if CryptoProvider::get_default().is_none() {
let _ = default_provider()
.install_default()
.map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e));
}
let endpoint = endpoint.parse::<Uri>().unwrap();
let tls = ClientTlsConfig::new().with_native_roots();
let channel = Channel::builder(endpoint)
.tls_config(tls).expect("Failed to create TLS config")
.tcp_keepalive(Some(Duration::from_secs(60)))
.http2_keep_alive_interval(Duration::from_secs(30))
.keep_alive_while_idle(true)
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.connect_lazy();
let client = ApiClient::with_interceptor(channel, MyInterceptor::new(auth_token));
let rpc_client = SolanaRpcClient::new(rpc_url);
Self { rpc_client: Arc::new(rpc_client), client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
let start_time = Instant::now();
let (content, signature) = serialize_smart_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
self.client.clone().post_submit_v2(api::PostSubmitRequest {
transaction: Some(api::TransactionMessage {
content,
is_cleanup: false,
}),
skip_pre_flight: true,
front_running_protection: Some(true),
experimental_front_running_protection: Some(true),
snipe_transaction: Some(true),
}).await?;
println!(" nextblock{}提交: {:?}", trade_type, start_time.elapsed());
let start_time: Instant = Instant::now();
let timeout: Duration = Duration::from_secs(10);
while Instant::now().duration_since(start_time) < timeout {
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => break,
Err(_) => continue,
}
}
println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
let mut entries = Vec::new();
let encoding = UiTransactionEncoding::Base64;
let mut signatures = Vec::new();
for transaction in transactions {
let (content, signature) = serialize_smart_transaction_and_encode(transaction, encoding).await?;
entries.push(api::PostSubmitRequestEntry {
transaction: Some(api::TransactionMessage {
content,
is_cleanup: false,
}),
skip_pre_flight: true,
});
signatures.push(signature);
}
self.client.clone().post_submit_batch_v2(api::PostSubmitBatchRequest {
entries,
submit_strategy: api::SubmitStrategy::PSubmitAll as i32,
use_bundle: Some(true),
front_running_protection: Some(true),
}).await?;
let start_time: Instant = Instant::now();
for signature in signatures.clone() {
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => continue,
Err(_) => continue,
}
}
println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signatures)
}
}
#[derive(Clone)]
pub struct ZeroSlotClient {
pub endpoint: String,
pub auth_token: String,
pub rpc_client: Arc<SolanaRpcClient>,
pub http_client: Client,
}
#[async_trait::async_trait]
impl FeeClientTrait for ZeroSlotClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *ZEROSLOT_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| ZEROSLOT_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::ZeroSlot
}
}
impl ZeroSlotClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
.pool_idle_timeout(Duration::from_secs(60))
.pool_max_idle_per_host(64)
.tcp_keepalive(Some(Duration::from_secs(1200)))
.http2_keep_alive_interval(Duration::from_secs(15))
.timeout(Duration::from_secs(10))
.connect_timeout(Duration::from_secs(5))
.build()
.unwrap();
Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
println!(" 交易编码base64: {:?}", start_time.elapsed());
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
content,
{ "encoding": "base64", "skipPreflight": true }
]
}))?;
let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20);
url.push_str(&self.endpoint);
url.push_str("/?api-key=");
url.push_str(&self.auth_token);
// 4. 直接使用 `text().await?`，避免 `json().await?` 的异步 JSON 解析
let response_text = self.http_client.post(&url)
.body(request_body) // 直接传字符串，避免 `json()` 开销
.header("Content-Type", "application/json") // 显式指定 JSON 头
.send()
.await?
.text()
.await?;
// 5. 用 `serde_json::from_str()` 解析 JSON，减少 `.json().await?` 额外等待
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" 0slot{}提交: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" 0slot{}提交失败: {:?}", trade_type, _error);
}
}
let start_time: Instant = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" 0slot{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
}
#[derive(Clone)]
pub struct NozomiClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub endpoint: String,
pub auth_token: String,
pub http_client: Client,
}
#[async_trait::async_trait]
impl FeeClientTrait for NozomiClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *NOZOMI_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NOZOMI_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::Nozomi
}
}
impl NozomiClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
.pool_idle_timeout(Duration::from_secs(60))
.pool_max_idle_per_host(64)
.tcp_keepalive(Some(Duration::from_secs(1200)))
.http2_keep_alive_interval(Duration::from_secs(15))
.timeout(Duration::from_secs(10))
.connect_timeout(Duration::from_secs(5))
.build()
.unwrap();
Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature, anyhow::Error> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
println!(" 交易编码base64: {:?}", start_time.elapsed());
// 按照 Nozomi 文档要求构建请求体
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
content,
{ "encoding": "base64" }
]
}))?;
let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20);
url.push_str(&self.endpoint);
url.push_str("/?c=");
url.push_str(&self.auth_token);
let response_text = self.http_client.post(&url)
.body(request_body)
.header("Content-Type", "application/json")
.send()
.await?
.text()
.await?;
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" nozomi{}提交: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
// eprintln!("nozomi交易提交失败: {:?}", _error);
}
}
let start_time: Instant = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" nozomi{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>, anyhow::Error> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
}
src/swqos/nextblock.rs
Executable
+166
View File
@@ -0,0 +1,166 @@
use crate::protos::nextblock_grpc;
use crate::protos::nextblock_grpc::api_client::ApiClient;
use crate::swqos::common::{poll_transaction_confirmation, serialize_smart_transaction_and_encode};
use rand::seq::IndexedRandom;
use rustls::crypto::ring::default_provider;
use rustls::crypto::CryptoProvider;
use tonic::transport::Channel;
use yellowstone_grpc_client::Interceptor;
use std::str::FromStr;
use std::{sync::Arc, time::Instant};
use solana_sdk::{signature::Signature};
use tonic::{service::interceptor::InterceptedService, transport::Uri, Status};
use std::time::Duration;
use solana_transaction_status::UiTransactionEncoding;
use tonic::transport::ClientTlsConfig;
use anyhow::Result;
use solana_sdk::transaction::VersionedTransaction;
use crate::swqos::{ClientType, TradeType};
use crate::swqos::SwqosClientTrait;
use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::NEXTBLOCK_TIP_ACCOUNTS};
#[derive(Clone)]
pub struct MyInterceptor {
auth_token: String,
}
impl MyInterceptor {
pub fn new(auth_token: String) -> Self {
Self { auth_token }
}
}
impl Interceptor for MyInterceptor {
fn call(&mut self, mut request: tonic::Request<()>) -> Result<tonic::Request<()>, Status> {
request.metadata_mut().insert(
"authorization",
tonic::metadata::MetadataValue::from_str(&self.auth_token)
.map_err(|_| Status::invalid_argument("Invalid auth token"))?
);
Ok(request)
}
}
#[derive(Clone)]
pub struct NextBlockClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub client: ApiClient<InterceptedService<Channel, MyInterceptor>>,
}
#[async_trait::async_trait]
impl SwqosClientTrait for NextBlockClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *NEXTBLOCK_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NEXTBLOCK_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::NextBlock
}
}
impl NextBlockClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
if CryptoProvider::get_default().is_none() {
let _ = default_provider()
.install_default()
.map_err(|e| anyhow::anyhow!("Failed to install crypto provider: {:?}", e));
}
let endpoint = endpoint.parse::<Uri>().unwrap();
let tls = ClientTlsConfig::new().with_native_roots();
let channel = Channel::builder(endpoint)
.tls_config(tls).expect("Failed to create TLS config")
.tcp_keepalive(Some(Duration::from_secs(60)))
.http2_keep_alive_interval(Duration::from_secs(30))
.keep_alive_while_idle(true)
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.connect_lazy();
let client = ApiClient::with_interceptor(channel, MyInterceptor::new(auth_token));
let rpc_client = SolanaRpcClient::new(rpc_url);
Self { rpc_client: Arc::new(rpc_client), client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
let start_time = Instant::now();
let (content, signature) = serialize_smart_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
self.client.clone().post_submit_v2(nextblock_grpc::PostSubmitRequest {
transaction: Some(nextblock_grpc::TransactionMessage {
content,
is_cleanup: false,
}),
skip_pre_flight: true,
front_running_protection: Some(true),
experimental_front_running_protection: Some(true),
snipe_transaction: Some(true),
}).await?;
println!(" nextblock{}提交: {:?}", trade_type, start_time.elapsed());
let start_time: Instant = Instant::now();
let timeout: Duration = Duration::from_secs(10);
while Instant::now().duration_since(start_time) < timeout {
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => break,
Err(_) => continue,
}
}
println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
let mut entries = Vec::new();
let encoding = UiTransactionEncoding::Base64;
let mut signatures = Vec::new();
for transaction in transactions {
let (content, signature) = serialize_smart_transaction_and_encode(transaction, encoding).await?;
entries.push(nextblock_grpc::PostSubmitRequestEntry {
transaction: Some(nextblock_grpc::TransactionMessage {
content,
is_cleanup: false,
}),
skip_pre_flight: true,
});
signatures.push(signature);
}
self.client.clone().post_submit_batch_v2(nextblock_grpc::PostSubmitBatchRequest {
entries,
submit_strategy: nextblock_grpc::SubmitStrategy::PSubmitAll as i32,
use_bundle: Some(true),
front_running_protection: Some(true),
}).await?;
let start_time: Instant = Instant::now();
for signature in signatures.clone() {
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => continue,
Err(_) => continue,
}
}
println!(" nextblock{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signatures)
}
}
src/swqos/nozomi.rs
Executable
+120
View File
@@ -0,0 +1,120 @@
use crate::swqos::common::{poll_transaction_confirmation, serialize_transaction_and_encode};
use rand::seq::IndexedRandom;
use reqwest::Client;
use serde_json::json;
use std::{sync::Arc, time::Instant};
use solana_sdk::{signature::Signature};
use std::time::Duration;
use solana_transaction_status::UiTransactionEncoding;
use anyhow::Result;
use solana_sdk::transaction::VersionedTransaction;
use crate::swqos::{ClientType, TradeType};
use crate::swqos::SwqosClientTrait;
use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::NOZOMI_TIP_ACCOUNTS};
#[derive(Clone)]
pub struct NozomiClient {
pub rpc_client: Arc<SolanaRpcClient>,
pub endpoint: String,
pub auth_token: String,
pub http_client: Client,
}
#[async_trait::async_trait]
impl SwqosClientTrait for NozomiClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *NOZOMI_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| NOZOMI_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::Nozomi
}
}
impl NozomiClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
.pool_idle_timeout(Duration::from_secs(60))
.pool_max_idle_per_host(64)
.tcp_keepalive(Some(Duration::from_secs(1200)))
.http2_keep_alive_interval(Duration::from_secs(15))
.timeout(Duration::from_secs(10))
.connect_timeout(Duration::from_secs(5))
.build()
.unwrap();
Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
println!(" 交易编码base64: {:?}", start_time.elapsed());
// 按照 Nozomi 文档要求构建请求体
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
content,
{ "encoding": "base64" }
]
}))?;
let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20);
url.push_str(&self.endpoint);
url.push_str("/?c=");
url.push_str(&self.auth_token);
let response_text = self.http_client.post(&url)
.body(request_body)
.header("Content-Type", "application/json")
.send()
.await?
.text()
.await?;
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" nozomi{}提交: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
// eprintln!("nozomi交易提交失败: {:?}", _error);
}
}
let start_time: Instant = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" nozomi{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
}
src/swqos/solana_rpc.rs
Executable
+64
View File
@@ -0,0 +1,64 @@
use std::{sync::Arc, time::Instant};
use solana_client::rpc_config::RpcSendTransactionConfig;
use solana_sdk::{
commitment_config::CommitmentLevel,
signature::Signature,
transaction::VersionedTransaction,
};
use solana_transaction_status::UiTransactionEncoding;
use crate::{common::SolanaRpcClient, swqos::{common::poll_transaction_confirmation, ClientType, TradeType}};
use crate::swqos::SwqosClientTrait;
use anyhow::Result;
#[derive(Clone)]
pub struct SolRpcClient {
pub rpc_client: Arc<SolanaRpcClient>,
}
#[async_trait::async_trait]
impl SwqosClientTrait for SolRpcClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
let signature = self.rpc_client.send_transaction_with_config(transaction, RpcSendTransactionConfig{
skip_preflight: true,
preflight_commitment: Some(CommitmentLevel::Processed),
encoding: Some(UiTransactionEncoding::Base64),
max_retries: Some(3),
min_context_slot: Some(0),
}).await?;
let start_time = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" signature: {:?}", signature);
println!(" rpc{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
fn get_tip_account(&self) -> Result<String> {
Ok("".to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::Rpc
}
}
impl SolRpcClient {
pub fn new(rpc_client: Arc<SolanaRpcClient>) -> Self {
Self { rpc_client }
}
}
src/swqos/types.rs
Executable
View File
src/swqos/zeroslot.rs
Executable
+119
View File
@@ -0,0 +1,119 @@
use crate::swqos::common::{poll_transaction_confirmation, serialize_transaction_and_encode};
use rand::seq::IndexedRandom;
use reqwest::Client;
use serde_json::json;
use std::{sync::Arc, time::Instant};
use std::time::Duration;
use solana_transaction_status::UiTransactionEncoding;
use anyhow::Result;
use solana_sdk::signature::Signature;
use solana_sdk::transaction::VersionedTransaction;
use crate::swqos::{ClientType, TradeType};
use crate::swqos::SwqosClientTrait;
use crate::{common::SolanaRpcClient, constants::pumpfun::accounts::ZEROSLOT_TIP_ACCOUNTS};
#[derive(Clone)]
pub struct ZeroSlotClient {
pub endpoint: String,
pub auth_token: String,
pub rpc_client: Arc<SolanaRpcClient>,
pub http_client: Client,
}
#[async_trait::async_trait]
impl SwqosClientTrait for ZeroSlotClient {
async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
self.send_transaction(trade_type, transaction).await
}
async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
self.send_transactions(trade_type, transactions).await
}
fn get_tip_account(&self) -> Result<String> {
let tip_account = *ZEROSLOT_TIP_ACCOUNTS.choose(&mut rand::rng()).or_else(|| ZEROSLOT_TIP_ACCOUNTS.first()).unwrap();
Ok(tip_account.to_string())
}
fn get_client_type(&self) -> ClientType {
ClientType::ZeroSlot
}
}
impl ZeroSlotClient {
pub fn new(rpc_url: String, endpoint: String, auth_token: String) -> Self {
let rpc_client = SolanaRpcClient::new(rpc_url);
let http_client = Client::builder()
.pool_idle_timeout(Duration::from_secs(60))
.pool_max_idle_per_host(64)
.tcp_keepalive(Some(Duration::from_secs(1200)))
.http2_keep_alive_interval(Duration::from_secs(15))
.timeout(Duration::from_secs(10))
.connect_timeout(Duration::from_secs(5))
.build()
.unwrap();
Self { rpc_client: Arc::new(rpc_client), endpoint, auth_token, http_client }
}
pub async fn send_transaction(&self, trade_type: TradeType, transaction: &VersionedTransaction) -> Result<Signature> {
let start_time = Instant::now();
let (content, signature) = serialize_transaction_and_encode(transaction, UiTransactionEncoding::Base64).await?;
println!(" 交易编码base64: {:?}", start_time.elapsed());
let request_body = serde_json::to_string(&json!({
"jsonrpc": "2.0",
"id": 1,
"method": "sendTransaction",
"params": [
content,
{ "encoding": "base64", "skipPreflight": true }
]
}))?;
let mut url = String::with_capacity(self.endpoint.len() + self.auth_token.len() + 20);
url.push_str(&self.endpoint);
url.push_str("/?api-key=");
url.push_str(&self.auth_token);
// 4. 直接使用 `text().await?`，避免 `json().await?` 的异步 JSON 解析
let response_text = self.http_client.post(&url)
.body(request_body) // 直接传字符串，避免 `json()` 开销
.header("Content-Type", "application/json") // 显式指定 JSON 头
.send()
.await?
.text()
.await?;
// 5. 用 `serde_json::from_str()` 解析 JSON，减少 `.json().await?` 额外等待
if let Ok(response_json) = serde_json::from_str::<serde_json::Value>(&response_text) {
if response_json.get("result").is_some() {
println!(" 0slot{}提交: {:?}", trade_type, start_time.elapsed());
} else if let Some(_error) = response_json.get("error") {
eprintln!(" 0slot{}提交失败: {:?}", trade_type, _error);
}
}
let start_time: Instant = Instant::now();
match poll_transaction_confirmation(&self.rpc_client, signature).await {
Ok(_) => (),
Err(_) => (),
}
println!(" 0slot{}确认: {:?}", trade_type, start_time.elapsed());
Ok(signature)
}
pub async fn send_transactions(&self, trade_type: TradeType, transactions: &Vec<VersionedTransaction>) -> Result<Vec<Signature>> {
let mut signatures = Vec::new();
for transaction in transactions {
let signature = self.send_transaction(trade_type, transaction).await?;
signatures.push(signature);
}
Ok(signatures)
}
}
src/trading/core/executor.rs
+2 -2
View File
@@ -107,7 +107,7 @@ impl TradeExecutor for GenericTradeExecutor {
// 并行执行交易
parallel_execute_with_tips(
params.fee_clients,
params.swqos_clients,
params.payer,
instructions,
params.priority_fee,
@@ -180,7 +180,7 @@ impl TradeExecutor for GenericTradeExecutor {
// 并行执行交易
parallel_execute_with_tips(
params.fee_clients,
params.swqos_clients,
params.payer,
instructions,
params.priority_fee,
src/trading/core/parallel.rs
+10 -10
View File
@@ -6,7 +6,7 @@ use tokio::task::JoinHandle;
use crate::{
common::PriorityFee,
swqos::{ClientType, FeeClient, TradeType},
swqos::{ClientType, SwqosClient, TradeType},
trading::common::{
build_rpc_transaction, build_sell_tip_transaction_with_priority_fee,
build_sell_transaction, build_tip_transaction_with_priority_fee,
@@ -15,7 +15,7 @@ use crate::{
/// 并行执行交易的通用函数
pub async fn parallel_execute_with_tips(
fee_clients: Vec<Arc<FeeClient>>,
swqos_clients: Vec<Arc<SwqosClient>>,
payer: Arc<Keypair>,
instructions: Vec<Instruction>,
priority_fee: PriorityFee,
@@ -27,8 +27,8 @@ pub async fn parallel_execute_with_tips(
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();
for i in 0..swqos_clients.len() {
let swqos_client = swqos_clients[i].clone();
let payer = payer.clone();
let instructions = instructions.clone();
let mut priority_fee = priority_fee.clone();
@@ -37,7 +37,7 @@ pub async fn parallel_execute_with_tips(
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
&& swqos_client.get_client_type() == ClientType::Rpc
{
build_sell_transaction(
payer,
@@ -48,9 +48,9 @@ pub async fn parallel_execute_with_tips(
)
.await?
} else if matches!(trade_type, TradeType::Sell)
&& fee_client.get_client_type() != ClientType::Rpc
&& swqos_client.get_client_type() != ClientType::Rpc
{
let tip_account = fee_client.get_tip_account()?;
let tip_account = swqos_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,
@@ -61,7 +61,7 @@ pub async fn parallel_execute_with_tips(
recent_blockhash,
)
.await?
} else if fee_client.get_client_type() == ClientType::Rpc {
} else if swqos_client.get_client_type() == ClientType::Rpc {
build_rpc_transaction(
payer,
&priority_fee,
@@ -72,7 +72,7 @@ pub async fn parallel_execute_with_tips(
)
.await?
} else {
let tip_account = fee_client.get_tip_account()?;
let tip_account = swqos_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];
@@ -88,7 +88,7 @@ pub async fn parallel_execute_with_tips(
.await?
};
fee_client
swqos_client
.send_transaction(trade_type, &transaction)
.await?;
Ok::<(), anyhow::Error>(())
src/trading/core/params.rs
+7 -7
View File
@@ -4,7 +4,7 @@ use std::sync::Arc;
use super::traits::ProtocolParams;
use crate::common::{PriorityFee, SolanaRpcClient};
use crate::swqos::FeeClient;
use crate::swqos::SwqosClient;
use crate::accounts::BondingCurveAccount;
/// 通用买入参数
@@ -27,7 +27,7 @@ pub struct BuyParams {
#[derive(Clone)]
pub struct BuyWithTipParams {
pub rpc: Option<Arc<SolanaRpcClient>>,
pub fee_clients: Vec<Arc<FeeClient>>,
pub swqos_clients: Vec<Arc<SwqosClient>>,
pub payer: Arc<Keypair>,
pub mint: Pubkey,
pub creator: Pubkey,
@@ -59,7 +59,7 @@ pub struct SellParams {
#[derive(Clone)]
pub struct SellWithTipParams {
pub rpc: Option<Arc<SolanaRpcClient>>,
pub fee_clients: Vec<Arc<FeeClient>>,
pub swqos_clients: Vec<Arc<SwqosClient>>,
pub payer: Arc<Keypair>,
pub mint: Pubkey,
pub creator: Pubkey,
@@ -124,10 +124,10 @@ impl ProtocolParams for PumpSwapParams {
impl BuyParams {
/// 转换为BuyWithTipParams
pub fn with_tip(self, fee_clients: Vec<Arc<FeeClient>>) -> BuyWithTipParams {
pub fn with_tip(self, swqos_clients: Vec<Arc<SwqosClient>>) -> BuyWithTipParams {
BuyWithTipParams {
rpc: self.rpc,
fee_clients,
swqos_clients,
payer: self.payer,
mint: self.mint,
creator: self.creator,
@@ -144,10 +144,10 @@ impl BuyParams {
impl SellParams {
/// 转换为SellWithTipParams
pub fn with_tip(self, fee_clients: Vec<Arc<FeeClient>>) -> SellWithTipParams {
pub fn with_tip(self, swqos_clients: Vec<Arc<SwqosClient>>) -> SellWithTipParams {
SellWithTipParams {
rpc: self.rpc,
fee_clients,
swqos_clients,
payer: self.payer,
mint: self.mint,
creator: self.creator,