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

feat: harden trading SDK examples

Browse Source
This commit is contained in:
0xfnzero
2026-07-12 14:35:58 +08:00
parent 47cef59d15
commit b09902c527
70 changed files with 1447 additions and 471 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/pumpswap_trading/src/main.rs
+234 -84
View File
@@ -1,5 +1,4 @@
use sol_trade_sdk::common::fast_fn::get_associated_token_address_with_program_id_fast_use_seed;
use sol_trade_sdk::common::TradeConfig;
use sol_trade_sdk::common::{clock::now_micros, SolanaRpcClient, TradeConfig};
use sol_trade_sdk::TradeTokenType;
use sol_trade_sdk::{
common::AnyResult,
@@ -8,11 +7,10 @@ use sol_trade_sdk::{
core::params::{DexParamEnum, PumpSwapParams},
factory::DexType,
},
SolanaTrade,
AccountPolicy, BuyAmount, SellAmount, SimpleBuyParams, SimpleSellParams, SolanaTrade,
};
use solana_commitment_config::CommitmentConfig;
use solana_sdk::signature::Keypair;
use solana_sdk::{pubkey::Pubkey, signer::Signer};
use solana_sdk::{hash::Hash, pubkey::Pubkey};
use solana_streamer_sdk::streaming::event_parser::{
common::filter::EventTypeFilter, protocols::pumpswap::PumpSwapBuyEvent,
};
@@ -25,24 +23,132 @@ use solana_streamer_sdk::streaming::YellowstoneGrpc;
use solana_streamer_sdk::{
match_event, streaming::event_parser::protocols::pumpswap::parser::PUMPSWAP_PROGRAM_ID,
};
use std::str::FromStr;
use std::sync::{
atomic::{AtomicBool, Ordering},
Arc,
};
use std::time::{Duration, Instant};
use tokio::sync::watch;
// Global static flag to ensure transaction is executed only once
static ALREADY_EXECUTED: AtomicBool = AtomicBool::new(false);
const BLOCKHASH_REFRESH_INTERVAL: Duration = Duration::from_millis(400);
const MAX_BLOCKHASH_AGE: Duration = Duration::from_secs(20);
#[derive(Clone, Copy)]
struct EventSelection {
target_mint: Option<Pubkey>,
target_pool: Option<Pubkey>,
max_event_age_ms: u64,
}
impl EventSelection {
fn from_env() -> AnyResult<Self> {
let target_mint = parse_optional_pubkey("TARGET_MINT")?;
let target_pool = parse_optional_pubkey("TARGET_POOL")?;
if target_mint.is_none() && target_pool.is_none() {
anyhow::bail!("set TARGET_MINT or TARGET_POOL before running this live example");
}
let max_event_age_ms = std::env::var("MAX_EVENT_AGE_MS")
.unwrap_or_else(|_| "1000".to_string())
.parse::<u64>()
.map_err(|_| anyhow::anyhow!("MAX_EVENT_AGE_MS must be a positive integer"))?;
if max_event_age_ms == 0 || max_event_age_ms > i64::MAX as u64 / 1_000 {
anyhow::bail!("MAX_EVENT_AGE_MS is outside the supported range");
}
Ok(Self { target_mint, target_pool, max_event_age_ms })
}
fn matches(self, pool: Pubkey, base_mint: Pubkey, quote_mint: Pubkey, recv_us: i64) -> bool {
if self.target_pool.is_some_and(|target| target != pool) {
return false;
}
if self.target_mint.is_some_and(|target| target != base_mint && target != quote_mint) {
return false;
}
is_event_fresh(recv_us, now_micros(), self.max_event_age_ms)
}
}
fn parse_optional_pubkey(key: &str) -> AnyResult<Option<Pubkey>> {
std::env::var(key)
.ok()
.map(|value| value.trim().to_string())
.filter(|value| !value.is_empty())
.map(|value| Pubkey::from_str(&value).map_err(anyhow::Error::from))
.transpose()
}
fn is_event_fresh(recv_us: i64, now_us: i64, max_age_ms: u64) -> bool {
recv_us > 0
&& now_us >= recv_us
&& now_us.saturating_sub(recv_us) <= (max_age_ms as i64).saturating_mul(1_000)
}
#[derive(Clone)]
struct CachedBlockhash {
hash: Hash,
fetched_at: Instant,
}
#[derive(Clone)]
struct BlockhashCache {
receiver: watch::Receiver<CachedBlockhash>,
}
impl BlockhashCache {
async fn start(rpc: Arc<SolanaRpcClient>) -> AnyResult<Self> {
let initial =
CachedBlockhash { hash: rpc.get_latest_blockhash().await?, fetched_at: Instant::now() };
let (sender, receiver) = watch::channel(initial);
tokio::spawn(async move {
let mut interval = tokio::time::interval(BLOCKHASH_REFRESH_INTERVAL);
interval.tick().await;
loop {
interval.tick().await;
match rpc.get_latest_blockhash().await {
Ok(hash) => {
if sender
.send(CachedBlockhash { hash, fetched_at: Instant::now() })
.is_err()
{
break;
}
}
Err(err) => eprintln!("warning: blockhash refresh failed: {err}"),
}
}
});
Ok(Self { receiver })
}
fn latest(&self) -> AnyResult<Hash> {
let cached = self.receiver.borrow().clone();
if cached.fetched_at.elapsed() > MAX_BLOCKHASH_AGE {
anyhow::bail!("cached blockhash is older than {} seconds", MAX_BLOCKHASH_AGE.as_secs());
}
Ok(cached.hash)
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
println!("Subscribing to GRPC events...");
let selection = EventSelection::from_env()?;
let trade_client = Arc::new(create_solana_trade_client().await?);
let blockhash_cache = BlockhashCache::start(trade_client.infrastructure.rpc.clone()).await?;
let grpc = YellowstoneGrpc::new(
"https://solana-yellowstone-grpc.publicnode.com:443".to_string(),
None,
std::env::var("GRPC_ENDPOINT")
.unwrap_or_else(|_| "https://solana-yellowstone-grpc.publicnode.com:443".to_string()),
std::env::var("GRPC_AUTH_TOKEN").ok(),
)?;
let callback = create_event_callback();
let callback = create_event_callback(trade_client, blockhash_cache, selection);
let protocols = vec![Protocol::PumpSwap];
// Filter accounts
let account_include = vec![
@@ -83,8 +189,12 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
}
/// Create an event callback function that handles different types of events
fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
|event: Box<dyn UnifiedEvent>| {
fn create_event_callback(
client: Arc<SolanaTrade>,
blockhash_cache: BlockhashCache,
selection: EventSelection,
) -> impl Fn(Box<dyn UnifiedEvent>) {
move |event: Box<dyn UnifiedEvent>| {
match_event!(event, {
PumpSwapBuyEvent => |e: PumpSwapBuyEvent| {
let is_wsol = e.base_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT || e.quote_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT;
@@ -92,13 +202,22 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
if !is_wsol && !is_usdc {
return;
}
if !selection.matches(e.pool, e.base_mint, e.quote_mint, e.metadata.recv_us) {
return;
}
// Test code, only test one transaction
if !ALREADY_EXECUTED.swap(true, Ordering::SeqCst) {
let event_clone = e.clone();
let client = client.clone();
let blockhash_cache = blockhash_cache.clone();
tokio::spawn(async move {
if let Err(err) = pumpswap_trade_with_grpc_buy_event(event_clone).await {
if let Err(err) = pumpswap_trade_with_grpc_buy_event(
client,
blockhash_cache,
event_clone,
).await {
eprintln!("Error in trade: {:?}", err);
std::process::exit(0);
std::process::exit(1);
}
});
}
@@ -109,13 +228,22 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
if !is_wsol && !is_usdc {
return;
}
if !selection.matches(e.pool, e.base_mint, e.quote_mint, e.metadata.recv_us) {
return;
}
// Test code, only test one transaction
if !ALREADY_EXECUTED.swap(true, Ordering::SeqCst) {
let event_clone = e.clone();
let client = client.clone();
let blockhash_cache = blockhash_cache.clone();
tokio::spawn(async move {
if let Err(err) = pumpswap_trade_with_grpc_sell_event(event_clone).await {
if let Err(err) = pumpswap_trade_with_grpc_sell_event(
client,
blockhash_cache,
event_clone,
).await {
eprintln!("Error in trade: {:?}", err);
std::process::exit(0);
std::process::exit(1);
}
});
}
@@ -127,9 +255,10 @@ fn create_event_callback() -> impl Fn(Box<dyn UnifiedEvent>) {
/// Create SolanaTrade client
/// Initializes a new SolanaTrade client with configuration
async fn create_solana_trade_client() -> AnyResult<SolanaTrade> {
println!("🚀 Initializing SolanaTrade client...");
let payer = Keypair::from_base58_string("your_payer_keypair_here");
let rpc_url = "https://api.mainnet-beta.solana.com".to_string();
println!("Initializing SolanaTrade client...");
let payer = sol_trade_sdk::common::keypair::load_keypair_from_env("PRIVATE_KEY")?;
let rpc_url = std::env::var("RPC_URL")
.unwrap_or_else(|_| "https://api.mainnet-beta.solana.com".to_string());
let commitment = CommitmentConfig::confirmed();
let swqos_configs: Vec<SwqosConfig> = vec![SwqosConfig::Default(rpc_url.clone())];
let trade_config = TradeConfig::builder(rpc_url, swqos_configs, commitment)
@@ -141,12 +270,15 @@ async fn create_solana_trade_client() -> AnyResult<SolanaTrade> {
// .mev_protection(false) // default: false
.build();
let solana_trade = SolanaTrade::new(Arc::new(payer), trade_config).await;
println!("✅ SolanaTrade client initialized successfully!");
println!("SolanaTrade client initialized successfully");
Ok(solana_trade)
}
async fn pumpswap_trade_with_grpc_buy_event(trade_info: PumpSwapBuyEvent) -> AnyResult<()> {
let client = create_solana_trade_client().await?;
async fn pumpswap_trade_with_grpc_buy_event(
client: Arc<SolanaTrade>,
blockhash_cache: BlockhashCache,
trade_info: PumpSwapBuyEvent,
) -> AnyResult<()> {
let params = PumpSwapParams::from_trade_with_fee_basis_points(
trade_info.pool,
trade_info.base_mint,
@@ -175,12 +307,16 @@ async fn pumpswap_trade_with_grpc_buy_event(trade_info: PumpSwapBuyEvent) -> Any
} else {
trade_info.base_mint
};
pumpswap_trade_with_grpc(&client, mint, params).await?;
pumpswap_trade_with_grpc(&client, &blockhash_cache, trade_info.metadata.recv_us, mint, params)
.await?;
Ok(())
}
async fn pumpswap_trade_with_grpc_sell_event(trade_info: PumpSwapSellEvent) -> AnyResult<()> {
let client = create_solana_trade_client().await?;
async fn pumpswap_trade_with_grpc_sell_event(
client: Arc<SolanaTrade>,
blockhash_cache: BlockhashCache,
trade_info: PumpSwapSellEvent,
) -> AnyResult<()> {
let params = PumpSwapParams::from_trade_with_fee_basis_points(
trade_info.pool,
trade_info.base_mint,
@@ -209,92 +345,106 @@ async fn pumpswap_trade_with_grpc_sell_event(trade_info: PumpSwapSellEvent) -> A
} else {
trade_info.base_mint
};
pumpswap_trade_with_grpc(&client, mint, params).await?;
pumpswap_trade_with_grpc(&client, &blockhash_cache, trade_info.metadata.recv_us, mint, params)
.await?;
Ok(())
}
async fn pumpswap_trade_with_grpc(
client: &SolanaTrade,
blockhash_cache: &BlockhashCache,
grpc_recv_us: i64,
mint_pubkey: Pubkey,
params: PumpSwapParams,
) -> AnyResult<()> {
println!("Testing PumpSwap trading...");
let slippage_basis_points = Some(500);
let recent_blockhash = client.infrastructure.rpc.get_latest_blockhash().await?;
let recent_blockhash = blockhash_cache.latest()?;
let gas_fee_strategy = sol_trade_sdk::common::GasFeeStrategy::new();
gas_fee_strategy.set_global_fee_strategy(150000, 150000, 500000, 500000, 0.001, 0.001);
let is_sol = params.base_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT
|| params.quote_mint == sol_trade_sdk::constants::WSOL_TOKEN_ACCOUNT;
let program_id = if params.base_mint == mint_pubkey {
params.base_token_program
} else if params.quote_mint == mint_pubkey {
params.quote_token_program
} else {
anyhow::bail!("target mint {} does not belong to pool {}", mint_pubkey, params.pool);
};
let balance_before =
client.get_payer_token_balance_with_program(&mint_pubkey, &program_id).await?;
// Buy tokens
println!("Buying tokens from PumpSwap...");
let buy_token_amount = 300_000;
let buy_params = sol_trade_sdk::TradeBuyParams {
dex_type: DexType::PumpSwap,
input_token_type: if is_sol { TradeTokenType::SOL } else { TradeTokenType::USDC },
mint: mint_pubkey,
input_token_amount: buy_token_amount,
slippage_basis_points: slippage_basis_points,
recent_blockhash: Some(recent_blockhash),
extension_params: DexParamEnum::PumpSwap(params.clone()),
address_lookup_table_accounts: Vec::new(),
wait_tx_confirmed: true,
wait_for_all_submits: false,
create_input_token_ata: is_sol,
close_input_token_ata: is_sol,
create_mint_ata: true,
durable_nonce: None,
fixed_output_token_amount: None,
gas_fee_strategy: gas_fee_strategy.clone(),
simulate: false,
use_exact_sol_amount: None,
grpc_recv_us: None,
};
client.buy(buy_params).await?;
let buy_params = SimpleBuyParams::new(
DexType::PumpSwap,
if is_sol { TradeTokenType::SOL } else { TradeTokenType::USDC },
mint_pubkey,
BuyAmount::WithMaxInput { quote_amount: buy_token_amount },
DexParamEnum::PumpSwap(params.clone()),
recent_blockhash,
gas_fee_strategy.clone(),
)
.slippage_basis_points(slippage_basis_points.unwrap_or(500))
.account_policy(AccountPolicy::Auto)
.wait_tx_confirmed(true)
.grpc_recv_us(grpc_recv_us);
let (ok, sigs, err, _) = client.buy_simple(buy_params).await?;
if !ok {
anyhow::bail!("buy failed: {:?}; signatures: {:?}", err, sigs);
}
// Sell tokens
println!("Selling tokens from PumpSwap...");
let rpc = client.infrastructure.rpc.clone();
let payer = client.payer.pubkey();
let program_id = if params.base_mint == mint_pubkey {
params.base_token_program
} else {
params.quote_token_program
};
let account = get_associated_token_address_with_program_id_fast_use_seed(
&payer,
&mint_pubkey,
&program_id,
client.use_seed_optimize,
);
let balance = rpc.get_token_account_balance(&account).await?;
let amount_token = balance.amount.parse::<u64>().unwrap();
let sell_params = sol_trade_sdk::TradeSellParams {
dex_type: DexType::PumpSwap,
output_token_type: if is_sol { TradeTokenType::SOL } else { TradeTokenType::USDC },
mint: mint_pubkey,
input_token_amount: amount_token,
slippage_basis_points: slippage_basis_points,
recent_blockhash: Some(recent_blockhash),
with_tip: false,
extension_params: DexParamEnum::PumpSwap(params.clone()),
address_lookup_table_accounts: Vec::new(),
wait_tx_confirmed: true,
wait_for_all_submits: false,
create_output_token_ata: is_sol,
close_output_token_ata: is_sol,
close_mint_token_ata: false,
durable_nonce: None,
fixed_output_token_amount: None,
gas_fee_strategy: gas_fee_strategy,
simulate: false,
grpc_recv_us: None,
};
client.sell(sell_params).await?;
let balance_after =
client.get_payer_token_balance_with_program(&mint_pubkey, &program_id).await?;
let position_amount = balance_after.checked_sub(balance_before).ok_or_else(|| {
anyhow::anyhow!(
"token balance decreased from {} to {}; refusing to sell existing holdings",
balance_before,
balance_after
)
})?;
if position_amount == 0 {
anyhow::bail!("confirmed buy did not increase token balance; refusing to sell");
}
let sell_params_from_rpc =
PumpSwapParams::from_pool_address_by_rpc(&client.infrastructure.rpc, &params.pool).await?;
let sell_params = SimpleSellParams::new(
DexType::PumpSwap,
if is_sol { TradeTokenType::SOL } else { TradeTokenType::USDC },
mint_pubkey,
SellAmount::ExactInput(position_amount),
DexParamEnum::PumpSwap(sell_params_from_rpc),
blockhash_cache.latest()?,
gas_fee_strategy,
)
.slippage_basis_points(slippage_basis_points.unwrap_or(500))
.account_policy(AccountPolicy::Auto)
.wait_tx_confirmed(true)
.with_tip(false);
let (ok, sigs, err, _) = client.sell_simple(sell_params).await?;
if !ok {
anyhow::bail!("sell failed: {:?}; signatures: {:?}", err, sigs);
}
// Exit program
std::process::exit(0);
}
#[cfg(test)]
mod tests {
use super::is_event_fresh;
#[test]
fn event_freshness_has_a_strict_boundary() {
assert!(!is_event_fresh(0, 1_000_000, 100));
assert!(!is_event_fresh(1_000_001, 1_000_000, 100));
assert!(!is_event_fresh(899_999, 1_000_000, 100));
assert!(is_event_fresh(900_000, 1_000_000, 100));
}
}