gavindiaz/pumpfun-bonkfun-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(examples): clarify cu optimized buy script

Browse Source
This commit is contained in:
smypmsa
2025-10-27 22:01:13 +00:00
parent 8367611ed7
commit 3f20649d0e
1 changed files with 160 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files
learning-examples/manual_buy_cu_optimized.py
+160 -207
View File
@@ -1,3 +1,15 @@
"""
Pump.fun Token Buy Script with Compute Unit Optimization
This script is identical to manual_buy.py but adds SetLoadedAccountsDataSizeLimit
instruction. By default, Solana transactions can load up to 64MB of account data
(costing 16k CU). By setting a lower limit (512KB), we reduce CU consumption and
improve transaction priority.
Key difference from manual_buy.py:
- Adds set_loaded_accounts_data_size_limit(512_000) before other instructions
"""
import asyncio
import base64
import hashlib
@@ -16,7 +28,7 @@ from solders.instruction import AccountMeta, Instruction
from solders.keypair import Keypair
from solders.message import Message
from solders.pubkey import Pubkey
from solders.transaction import Transaction
from solders.transaction import Transaction, VersionedTransaction
from spl.token.instructions import (
create_idempotent_associated_token_account,
get_associated_token_address,
@@ -61,10 +73,14 @@ class BondingCurveState:
)
def __init__(self, data: bytes) -> None:
"""Parse bonding curve data."""
if data[:8] != EXPECTED_DISCRIMINATOR:
raise ValueError("Invalid curve state discriminator")
parsed = self._STRUCT.parse(data[8:])
self.__dict__.update(parsed)
# Convert raw bytes to Pubkey for creator field
if hasattr(self, "creator") and isinstance(self.creator, bytes):
self.creator = Pubkey.from_bytes(self.creator)
@@ -75,15 +91,18 @@ async def get_pump_curve_state(
response = await conn.get_account_info(curve_address, encoding="base64")
if not response.value or not response.value.data:
raise ValueError("Invalid curve state: No data")
data = response.value.data
if data[:8] != EXPECTED_DISCRIMINATOR:
raise ValueError("Invalid curve state discriminator")
return BondingCurveState(data)
def calculate_pump_curve_price(curve_state: BondingCurveState) -> float:
if curve_state.virtual_token_reserves <= 0 or curve_state.virtual_sol_reserves <= 0:
raise ValueError("Invalid reserve state")
return (curve_state.virtual_sol_reserves / LAMPORTS_PER_SOL) / (
curve_state.virtual_token_reserves / 10**TOKEN_DECIMALS
)
@@ -91,130 +110,53 @@ def calculate_pump_curve_price(curve_state: BondingCurveState) -> float:
def _find_creator_vault(creator: Pubkey) -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)], PUMP_PROGRAM
[b"creator-vault", bytes(creator)],
PUMP_PROGRAM,
)
return derived_address
def _find_global_volume_accumulator() -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[b"global_volume_accumulator"], PUMP_PROGRAM
[b"global_volume_accumulator"],
PUMP_PROGRAM,
)
return derived_address
def _find_user_volume_accumulator(user: Pubkey) -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[b"user_volume_accumulator", bytes(user)], PUMP_PROGRAM
[b"user_volume_accumulator", bytes(user)],
PUMP_PROGRAM,
)
return derived_address
def _find_fee_config() -> Pubkey:
derived_address, _ = Pubkey.find_program_address(
[b"fee_config", bytes(PUMP_PROGRAM)], PUMP_FEE_PROGRAM
[b"fee_config", bytes(PUMP_PROGRAM)],
PUMP_FEE_PROGRAM,
)
return derived_address
def set_loaded_accounts_data_size_limit(bytes_limit: int) -> Instruction:
"""Create SetLoadedAccountsDataSizeLimit compute budget instruction."""
"""
Create SetLoadedAccountsDataSizeLimit instruction to reduce CU consumption.
Solana defaults to 64MB loaded data limit (16k CU cost: 8 CU per 32KB).
By setting a lower limit, you reduce CU consumption and improve tx priority.
Args:
bytes_limit: Max account data size in bytes (e.g., 512_000 = 512KB)
Returns:
Compute Budget instruction (discriminator 4)
"""
data = struct.pack("<BI", 4, bytes_limit)
return Instruction(COMPUTE_BUDGET_PROGRAM, data, [])
def build_buy_instruction(
payer: Keypair,
mint: Pubkey,
bonding_curve: Pubkey,
associated_bonding_curve: Pubkey,
creator_vault: Pubkey,
token_amount: int,
max_amount_lamports: int,
):
"""Build the buy instruction with all accounts."""
associated_token_account = get_associated_token_address(payer.pubkey(), mint)
accounts = [
AccountMeta(pubkey=PUMP_GLOBAL, is_signer=False, is_writable=False),
AccountMeta(pubkey=PUMP_FEE, is_signer=False, is_writable=True),
AccountMeta(pubkey=mint, is_signer=False, is_writable=False),
AccountMeta(pubkey=bonding_curve, is_signer=False, is_writable=True),
AccountMeta(pubkey=associated_bonding_curve, is_signer=False, is_writable=True),
AccountMeta(pubkey=associated_token_account, is_signer=False, is_writable=True),
AccountMeta(pubkey=payer.pubkey(), is_signer=True, is_writable=True),
AccountMeta(pubkey=SYSTEM_PROGRAM, is_signer=False, is_writable=False),
AccountMeta(pubkey=SYSTEM_TOKEN_PROGRAM, is_signer=False, is_writable=False),
AccountMeta(pubkey=creator_vault, is_signer=False, is_writable=True),
AccountMeta(pubkey=PUMP_EVENT_AUTHORITY, is_signer=False, is_writable=False),
AccountMeta(pubkey=PUMP_PROGRAM, is_signer=False, is_writable=False),
AccountMeta(
pubkey=_find_global_volume_accumulator(), is_signer=False, is_writable=True
),
AccountMeta(
pubkey=_find_user_volume_accumulator(payer.pubkey()),
is_signer=False,
is_writable=True,
),
AccountMeta(pubkey=_find_fee_config(), is_signer=False, is_writable=False),
AccountMeta(pubkey=PUMP_FEE_PROGRAM, is_signer=False, is_writable=False),
]
discriminator = struct.pack("<Q", 16927863322537952870)
data = (
discriminator
+ struct.pack("<Q", token_amount)
+ struct.pack("<Q", max_amount_lamports)
)
return Instruction(PUMP_PROGRAM, data, accounts)
async def simulate_buy(
mint: Pubkey,
bonding_curve: Pubkey,
associated_bonding_curve: Pubkey,
creator_vault: Pubkey,
amount: float,
slippage: float = 0.25,
):
"""Simulate buy transaction and return CU consumption."""
private_key = base58.b58decode(os.environ.get("SOLANA_PRIVATE_KEY"))
payer = Keypair.from_bytes(private_key)
async with AsyncClient(RPC_ENDPOINT) as client:
amount_lamports = int(amount * LAMPORTS_PER_SOL)
curve_state = await get_pump_curve_state(client, bonding_curve)
token_price_sol = calculate_pump_curve_price(curve_state)
token_amount = int((amount / token_price_sol) * 10**6)
max_amount_lamports = int(amount_lamports * (1 + slippage))
buy_ix = build_buy_instruction(
payer,
mint,
bonding_curve,
associated_bonding_curve,
creator_vault,
token_amount,
max_amount_lamports,
)
idempotent_ata_ix = create_idempotent_associated_token_account(
payer.pubkey(), payer.pubkey(), mint
)
msg = Message(
[set_compute_unit_price(1_000), idempotent_ata_ix, buy_ix], payer.pubkey()
)
recent_blockhash = await client.get_latest_blockhash()
tx = Transaction([payer], msg, recent_blockhash.value.blockhash)
sim_result = await client.simulate_transaction(tx)
if sim_result.value.err:
print(f"Simulation error: {sim_result.value.err}")
return None
return sim_result.value.units_consumed
async def buy_token(
mint: Pubkey,
bonding_curve: Pubkey,
@@ -222,62 +164,131 @@ async def buy_token(
creator_vault: Pubkey,
amount: float,
slippage: float = 0.25,
use_cu_optimization: bool = False,
max_retries=5,
):
"""Buy token with or without CU optimization."""
private_key = base58.b58decode(os.environ.get("SOLANA_PRIVATE_KEY"))
payer = Keypair.from_bytes(private_key)
async with AsyncClient(RPC_ENDPOINT) as client:
associated_token_account = get_associated_token_address(payer.pubkey(), mint)
amount_lamports = int(amount * LAMPORTS_PER_SOL)
# Fetch the token price
curve_state = await get_pump_curve_state(client, bonding_curve)
token_price_sol = calculate_pump_curve_price(curve_state)
token_amount = int((amount / token_price_sol) * 10**6)
token_amount = amount / token_price_sol
# Calculate maximum SOL to spend with slippage
max_amount_lamports = int(amount_lamports * (1 + slippage))
buy_ix = build_buy_instruction(
payer,
mint,
bonding_curve,
associated_bonding_curve,
creator_vault,
token_amount,
max_amount_lamports,
accounts = [
AccountMeta(pubkey=PUMP_GLOBAL, is_signer=False, is_writable=False),
AccountMeta(pubkey=PUMP_FEE, is_signer=False, is_writable=True),
AccountMeta(pubkey=mint, is_signer=False, is_writable=False),
AccountMeta(pubkey=bonding_curve, is_signer=False, is_writable=True),
AccountMeta(
pubkey=associated_bonding_curve,
is_signer=False,
is_writable=True,
),
AccountMeta(
pubkey=associated_token_account,
is_signer=False,
is_writable=True,
),
AccountMeta(pubkey=payer.pubkey(), is_signer=True, is_writable=True),
AccountMeta(pubkey=SYSTEM_PROGRAM, is_signer=False, is_writable=False),
AccountMeta(
pubkey=SYSTEM_TOKEN_PROGRAM, is_signer=False, is_writable=False
),
AccountMeta(pubkey=creator_vault, is_signer=False, is_writable=True),
AccountMeta(
pubkey=PUMP_EVENT_AUTHORITY, is_signer=False, is_writable=False
),
AccountMeta(pubkey=PUMP_PROGRAM, is_signer=False, is_writable=False),
AccountMeta(
pubkey=_find_global_volume_accumulator(),
is_signer=False,
is_writable=True,
),
AccountMeta(
pubkey=_find_user_volume_accumulator(payer.pubkey()),
is_signer=False,
is_writable=True,
),
# Index 14: fee_config (readonly)
AccountMeta(
pubkey=_find_fee_config(),
is_signer=False,
is_writable=False,
),
# Index 15: fee_program (readonly)
AccountMeta(
pubkey=PUMP_FEE_PROGRAM,
is_signer=False,
is_writable=False,
),
]
discriminator = struct.pack("<Q", 16927863322537952870)
data = (
discriminator
+ struct.pack("<Q", int(token_amount * 10**6))
+ struct.pack("<Q", max_amount_lamports)
)
buy_ix = Instruction(PUMP_PROGRAM, data, accounts)
idempotent_ata_ix = create_idempotent_associated_token_account(
payer.pubkey(), payer.pubkey(), mint
)
instructions = [set_compute_unit_price(1_000)]
if use_cu_optimization:
instructions.insert(0, set_loaded_accounts_data_size_limit(512_000))
instructions.extend([idempotent_ata_ix, buy_ix])
# CU OPTIMIZATION: Limit account data to 512KB (down from 64MB default)
# This reduces CU cost from 16k to ~128 CU and improves tx priority.
# Must be placed FIRST in the instruction list.
cu_limit_ix = set_loaded_accounts_data_size_limit(512_000)
msg = Message(instructions, payer.pubkey())
msg = Message(
[cu_limit_ix, set_compute_unit_price(1_000), idempotent_ata_ix, buy_ix],
payer.pubkey(),
)
recent_blockhash = await client.get_latest_blockhash()
tx = Transaction([payer], msg, recent_blockhash.value.blockhash)
opts = TxOpts(skip_preflight=True, preflight_commitment=Confirmed)
tx_result = await client.send_transaction(tx, opts=opts)
tx_hash = tx_result.value
print(f" TX: https://explorer.solana.com/tx/{tx_hash}")
await client.confirm_transaction(
tx_hash, commitment="confirmed", sleep_seconds=1
)
for attempt in range(max_retries):
try:
tx_buy = await client.send_transaction(
Transaction(
[payer],
msg,
recent_blockhash.value.blockhash,
),
opts=opts,
)
tx_hash = tx_buy.value
print(f"Transaction sent: https://explorer.solana.com/tx/{tx_hash}")
await client.confirm_transaction(
tx_hash, commitment="confirmed", sleep_seconds=1
)
print("Transaction confirmed")
return # Success, exit the function
except Exception as e:
print(f"Attempt {attempt + 1} failed: {str(e)[:50]}")
if attempt < max_retries - 1:
wait_time = 2**attempt
print(f"Retrying in {wait_time} seconds...")
await asyncio.sleep(wait_time)
else:
print("Max retries reached. Unable to complete the transaction.")
# Get CU consumption
tx_details = await client.get_transaction(
tx_hash, encoding="json", max_supported_transaction_version=0
)
if (
tx_details.value
and tx_details.value.transaction
and tx_details.value.transaction.meta
):
cu_consumed = tx_details.value.transaction.meta.compute_units_consumed
return cu_consumed
return None
def load_idl(file_path):
with open(file_path) as f:
return json.load(f)
def calculate_discriminator(instruction_name):
sha = hashlib.sha256()
sha.update(instruction_name.encode("utf-8"))
return struct.unpack("<Q", sha.digest()[:8])[0]
def decode_create_instruction(ix_data, ix_def, accounts):
@@ -301,18 +312,15 @@ def decode_create_instruction(ix_data, ix_def, accounts):
args["bondingCurve"] = str(accounts[2])
args["associatedBondingCurve"] = str(accounts[3])
args["user"] = str(accounts[7])
return args
async def listen_for_create_transaction():
"""Listen for new token creation on pump.fun."""
idl_path = os.path.join(os.path.dirname(__file__), "..", "idl", "pump_fun_idl.json")
with open(idl_path) as f:
idl = json.load(f)
create_discriminator = struct.unpack(
"<Q", hashlib.sha256(b"global:create").digest()[:8]
)[0]
idl = load_idl(idl_path)
create_discriminator = calculate_discriminator("global:create")
async with websockets.connect(RPC_WEBSOCKET) as websocket:
subscription_message = json.dumps(
@@ -347,8 +355,6 @@ async def listen_for_create_transaction():
if "transactions" in block:
for tx in block["transactions"]:
if isinstance(tx, dict) and "transaction" in tx:
from solders.transaction import VersionedTransaction
tx_data_decoded = base64.b64decode(
tx["transaction"][0]
)
@@ -401,79 +407,26 @@ async def main():
mint = Pubkey.from_string(token_data["mint"])
bonding_curve = Pubkey.from_string(token_data["bondingCurve"])
associated_bonding_curve = Pubkey.from_string(token_data["associatedBondingCurve"])
creator_vault = _find_creator_vault(Pubkey.from_string(token_data["creator"]))
# Get creator from bonding curve state
# Fetch the token price
async with AsyncClient(RPC_ENDPOINT) as client:
curve_state = await get_pump_curve_state(client, bonding_curve)
creator_vault = _find_creator_vault(curve_state.creator)
token_price_sol = calculate_pump_curve_price(curve_state)
amount = 0.001 # 0.001 SOL
slippage = 0.3
# Amount of SOL to spend (adjust as needed)
amount = 0.000_001 # 0.00001 SOL
slippage = 0.3 # 30% slippage tolerance
print(f"\nToken price: {token_price_sol:.10f} SOL")
print(f"Buying {amount:.6f} SOL worth with {slippage * 100:.1f}% slippage\n")
# 1. Simulate
print("=" * 60)
print("1. SIMULATION")
print("=" * 60)
sim_cu = await simulate_buy(
print(f"Bonding curve address: {bonding_curve}")
print(f"Token price: {token_price_sol:.10f} SOL")
print(
f"Buying {amount:.6f} SOL worth of the new token with {slippage * 100:.1f}% slippage tolerance..."
)
print("CU Optimization: Enabled (512KB account data limit)")
await buy_token(
mint, bonding_curve, associated_bonding_curve, creator_vault, amount, slippage
)
if sim_cu:
print(f" Simulated CU consumption: {sim_cu:,}")
# 2. Buy without optimization
print("\n" + "=" * 60)
print("2. BUY WITHOUT CU OPTIMIZATION")
print("=" * 60)
cu_no_opt = await buy_token(
mint,
bonding_curve,
associated_bonding_curve,
creator_vault,
amount,
slippage,
use_cu_optimization=False,
)
if cu_no_opt:
print(f" Actual CU consumed: {cu_no_opt:,}")
# 3. Buy with optimization
print("\n" + "=" * 60)
print("3. BUY WITH CU OPTIMIZATION (setLoadedAccountsDataSizeLimit)")
print(" Setting limit to 500 KB (512,000 bytes)")
print("=" * 60)
cu_with_opt = await buy_token(
mint,
bonding_curve,
associated_bonding_curve,
creator_vault,
amount,
slippage,
use_cu_optimization=True,
)
if cu_with_opt:
print(f" ✓ Actual CU consumed (optimized): {cu_with_opt:,}")
if cu_no_opt:
immediate_savings = cu_no_opt - cu_with_opt
print(f" ✓ Immediate savings: {immediate_savings:,} CU")
# Summary
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
if sim_cu:
print(f"Simulated: {sim_cu:,} CU")
if cu_no_opt:
print(f"Without optimize: {cu_no_opt:,} CU")
if cu_with_opt:
print(f"With optimize: {cu_with_opt:,} CU")
if cu_no_opt:
savings = cu_no_opt - cu_with_opt
pct = (savings / cu_no_opt) * 100
print(f"Savings: {savings:,} CU ({pct:.1f}%)")
if __name__ == "__main__":