gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-07-27 23:37:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): platform listener check

Browse Source
This commit is contained in:
smypmsa
2025-08-02 16:12:45 +00:00
parent e7107751f5
commit 26c048c86a
8 changed files with 115 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bots/bot-sniper-1-geyser.yaml
+2 -2
View File
@@ -8,11 +8,11 @@ rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}"
wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}"
private_key: "${SOLANA_PRIVATE_KEY}"
enabled: true # You can turn off the bot w/o removing its config
enabled: false # You can turn off the bot w/o removing its config
separate_process: true
# Options: "pump_fun" (default), "lets_bonk"
platform: "lets_bonk"
platform: "pump_fun"
# Geyser configuration (fastest method for getting updates)
geyser:
bots/bot-sniper-3-blocks.yaml
+1 -1
View File
@@ -8,7 +8,7 @@ rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}"
wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}"
private_key: "${SOLANA_PRIVATE_KEY}"
enabled: false # You can turn off the bot w/o removing its config
enabled: true # You can turn off the bot w/o removing its config
separate_process: true
# Options: "pump_fun" (default), "lets_bonk"
src/config_loader.py
+1 -1
View File
@@ -45,7 +45,7 @@ VALID_VALUES = {
# Platform-specific listener compatibility
PLATFORM_LISTENER_COMPATIBILITY = {
Platform.PUMP_FUN: ["logs", "blocks", "geyser", "pumpportal"],
Platform.LETS_BONK: ["logs", "blocks", "geyser"], # PumpPortal is pump.fun only
Platform.LETS_BONK: ["blocks", "geyser"], # LetsBonk only supports geyser and blocks
}
src/monitoring/listener_factory.py
+2 -2
View File
@@ -132,6 +132,6 @@ class ListenerFactory:
if platform == Platform.PUMP_FUN:
return ["logs", "blocks", "geyser", "pumpportal"]
elif platform == Platform.LETS_BONK:
return ["logs", "blocks", "geyser"] # PumpPortal is pump.fun only
return ["blocks", "geyser"] # LetsBonk only supports geyser and blocks
else:
return ["logs", "blocks", "geyser"] # Default universal listeners
return ["blocks", "geyser"] # Default universal listeners
src/monitoring/universal_block_listener.py
+13 -2
View File
@@ -49,9 +49,20 @@ class UniversalBlockListener(BaseTokenListener):
for platform in self.platforms:
try:
# We'll need a dummy client for getting the parser
# Create a simple dummy client that doesn't start blockhash updater
from core.client import SolanaClient
dummy_client = SolanaClient("http://localhost") # Won't be used for parsing
# Create a mock client class to avoid network operations
class DummyClient(SolanaClient):
def __init__(self):
# Skip SolanaClient.__init__ to avoid starting blockhash updater
self.rpc_endpoint = "http://dummy"
self._client = None
self._cached_blockhash = None
self._blockhash_lock = None
self._blockhash_updater_task = None
dummy_client = DummyClient()
implementations = get_platform_implementations(platform, dummy_client)
parser = implementations.event_parser
src/monitoring/universal_geyser_listener.py
+16 -5
View File
@@ -50,13 +50,24 @@ class UniversalGeyserListener(BaseTokenListener):
self.platform_parsers = {}
self.platform_program_ids = set()
# Create a temporary client for getting parsers
from core.client import SolanaClient
temp_client = SolanaClient("http://temp")
for platform in self.platforms:
try:
implementations = platform_factory.create_for_platform(platform, temp_client)
# Create a simple dummy client that doesn't start blockhash updater
from core.client import SolanaClient
# Create a mock client class to avoid network operations
class DummyClient(SolanaClient):
def __init__(self):
# Skip SolanaClient.__init__ to avoid starting blockhash updater
self.rpc_endpoint = "http://dummy"
self._client = None
self._cached_blockhash = None
self._blockhash_lock = None
self._blockhash_updater_task = None
dummy_client = DummyClient()
implementations = platform_factory.create_for_platform(platform, dummy_client)
parser = implementations.event_parser
self.platform_parsers[platform] = parser
self.platform_program_ids.add(parser.get_program_id())
src/monitoring/universal_logs_listener.py
+16 -5
View File
@@ -45,13 +45,24 @@ class UniversalLogsListener(BaseTokenListener):
self.platform_parsers = {}
self.platform_program_ids = []
# Create a temporary client for getting parsers (stateless parsers don't use it)
from core.client import SolanaClient
temp_client = SolanaClient("http://temp")
for platform in self.platforms:
try:
implementations = platform_factory.create_for_platform(platform, temp_client)
# Create a simple dummy client that doesn't start blockhash updater
from core.client import SolanaClient
# Create a mock client class to avoid network operations
class DummyClient(SolanaClient):
def __init__(self):
# Skip SolanaClient.__init__ to avoid starting blockhash updater
self.rpc_endpoint = "http://dummy"
self._client = None
self._cached_blockhash = None
self._blockhash_lock = None
self._blockhash_updater_task = None
dummy_client = DummyClient()
implementations = platform_factory.create_for_platform(platform, dummy_client)
parser = implementations.event_parser
self.platform_parsers[platform] = parser
self.platform_program_ids.append(str(parser.get_program_id()))
src/platforms/pumpfun/instruction_builder.py
+64 -223
View File
@@ -1,17 +1,14 @@
"""
LetsBonk implementation of InstructionBuilder interface.
Pump.Fun implementation of InstructionBuilder interface.
This module builds LetsBonk (Raydium LaunchLab) specific buy and sell instructions
This module builds pump.fun-specific buy and sell instructions
by implementing the InstructionBuilder interface with IDL-based discriminators.
"""
import hashlib
import struct
import time
from solders.instruction import AccountMeta, Instruction
from solders.pubkey import Pubkey
from solders.system_program import CreateAccountWithSeedParams, create_account_with_seed
from spl.token.instructions import create_idempotent_associated_token_account
from core.pubkeys import TOKEN_DECIMALS, SystemAddresses
@@ -22,28 +19,28 @@ from utils.logger import get_logger
logger = get_logger(__name__)
class LetsBonkInstructionBuilder(InstructionBuilder):
"""LetsBonk (Raydium LaunchLab) implementation of InstructionBuilder interface with IDL-based discriminators."""
class PumpFunInstructionBuilder(InstructionBuilder):
"""Pump.Fun implementation of InstructionBuilder interface with IDL-based discriminators."""
def __init__(self, idl_parser: IDLParser):
"""Initialize LetsBonk instruction builder with injected IDL parser.
"""Initialize pump.fun instruction builder with injected IDL parser.
Args:
idl_parser: Pre-loaded IDL parser for LetsBonk platform
idl_parser: Pre-loaded IDL parser for pump.fun platform
"""
self._idl_parser = idl_parser
# Get discriminators from injected IDL parser
discriminators = self._idl_parser.get_instruction_discriminators()
self._buy_exact_in_discriminator = discriminators["buy_exact_in"]
self._sell_exact_in_discriminator = discriminators["sell_exact_in"]
self._buy_discriminator = discriminators["buy"]
self._sell_discriminator = discriminators["sell"]
logger.info("LetsBonk instruction builder initialized with injected IDL parser")
logger.info("Pump.Fun instruction builder initialized with injected IDL parser")
@property
def platform(self) -> Platform:
"""Get the platform this builder serves."""
return Platform.LETS_BONK
return Platform.PUMP_FUN
async def build_buy_instruction(
self,
@@ -53,7 +50,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build buy instruction(s) for LetsBonk using buy_exact_in.
"""Build buy instruction(s) for pump.fun.
Args:
token_info: Token information
@@ -70,71 +67,38 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
# Get all required accounts
accounts_info = address_provider.get_buy_instruction_accounts(token_info, user)
# 1. Create idempotent ATA for base token
# 1. Create idempotent ATA instruction (won't fail if ATA already exists)
ata_instruction = create_idempotent_associated_token_account(
user, # payer
user, # owner
user, # owner
token_info.mint, # mint
SystemAddresses.TOKEN_PROGRAM, # token program
)
instructions.append(ata_instruction)
# 2. Create WSOL account with seed (temporary account for the transaction)
wsol_seed = self._generate_wsol_seed(user)
wsol_account = address_provider.create_wsol_account_with_seed(user, wsol_seed)
# Account creation cost + amount to spend
account_creation_lamports = 2_039_280 # Standard account creation cost
total_lamports = amount_in + account_creation_lamports
create_wsol_ix = create_account_with_seed(
CreateAccountWithSeedParams(
from_pubkey=user,
to_pubkey=wsol_account,
base=user,
seed=wsol_seed,
lamports=total_lamports,
space=165, # Size of a token account
owner=SystemAddresses.TOKEN_PROGRAM
)
)
instructions.append(create_wsol_ix)
# 3. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account,
SystemAddresses.SOL_MINT,
user
)
instructions.append(initialize_wsol_ix)
# 4. Build buy_exact_in instruction with correct account ordering
# Based on the IDL and manual examples, the account order should be:
# 2. Build buy instruction
buy_accounts = [
AccountMeta(pubkey=user, is_signer=True, is_writable=True), # payer
AccountMeta(pubkey=accounts_info["authority"], is_signer=False, is_writable=False), # authority
AccountMeta(pubkey=accounts_info["global_config"], is_signer=False, is_writable=False), # global_config
AccountMeta(pubkey=accounts_info["platform_config"], is_signer=False, is_writable=False), # platform_config
AccountMeta(pubkey=accounts_info["pool_state"], is_signer=False, is_writable=True), # pool_state
AccountMeta(pubkey=accounts_info["user_base_token"], is_signer=False, is_writable=True), # user_base_token
AccountMeta(pubkey=wsol_account, is_signer=False, is_writable=True), # user_quote_token (WSOL account)
AccountMeta(pubkey=accounts_info["base_vault"], is_signer=False, is_writable=True), # base_vault
AccountMeta(pubkey=accounts_info["quote_vault"], is_signer=False, is_writable=True), # quote_vault
AccountMeta(pubkey=token_info.mint, is_signer=False, is_writable=False), # base_token_mint
AccountMeta(pubkey=SystemAddresses.SOL_MINT, is_signer=False, is_writable=False), # quote_token_mint
AccountMeta(pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False), # base_token_program
AccountMeta(pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False), # quote_token_program
AccountMeta(pubkey=accounts_info["event_authority"], is_signer=False, is_writable=False), # event_authority
AccountMeta(pubkey=accounts_info["program"], is_signer=False, is_writable=False), # program
AccountMeta(pubkey=accounts_info["global"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["fee"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["bonding_curve"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["associated_bonding_curve"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user_token_account"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user"], is_signer=True, is_writable=True),
AccountMeta(pubkey=accounts_info["system_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["token_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["creator_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["event_authority"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["global_volume_accumulator"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user_volume_accumulator"], is_signer=False, is_writable=True),
]
# Build instruction data: discriminator + amount_in + minimum_amount_out + share_fee_rate
SHARE_FEE_RATE = 0 # No sharing fee
# Build instruction data: discriminator + token_amount + max_sol_cost
instruction_data = (
self._buy_exact_in_discriminator +
struct.pack("<Q", amount_in) + # amount_in (u64) - SOL to spend
struct.pack("<Q", minimum_amount_out) + # minimum_amount_out (u64) - min tokens
struct.pack("<Q", SHARE_FEE_RATE) # share_fee_rate (u64): 0
self._buy_discriminator +
struct.pack("<Q", minimum_amount_out) + # token amount in raw units
struct.pack("<Q", amount_in) # max SOL cost in lamports
)
buy_instruction = Instruction(
@@ -144,14 +108,6 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
)
instructions.append(buy_instruction)
# 5. Close WSOL account to reclaim SOL
close_wsol_ix = self._create_close_account_instruction(
wsol_account,
user,
user
)
instructions.append(close_wsol_ix)
return instructions
async def build_sell_instruction(
@@ -162,7 +118,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build sell instruction(s) for LetsBonk using sell_exact_in.
"""Build sell instruction(s) for pump.fun.
Args:
token_info: Token information
@@ -179,60 +135,27 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
# Get all required accounts
accounts_info = address_provider.get_sell_instruction_accounts(token_info, user)
# 1. Create WSOL account with seed (to receive SOL)
wsol_seed = self._generate_wsol_seed(user)
wsol_account = address_provider.create_wsol_account_with_seed(user, wsol_seed)
# Minimal account creation cost
account_creation_lamports = 2_039_280
create_wsol_ix = create_account_with_seed(
CreateAccountWithSeedParams(
from_pubkey=user,
to_pubkey=wsol_account,
base=user,
seed=wsol_seed,
lamports=account_creation_lamports,
space=165, # Size of a token account
owner=SystemAddresses.TOKEN_PROGRAM
)
)
instructions.append(create_wsol_ix)
# 2. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account,
SystemAddresses.SOL_MINT,
user
)
instructions.append(initialize_wsol_ix)
# 3. Build sell_exact_in instruction with correct account ordering
# Build sell instruction accounts
sell_accounts = [
AccountMeta(pubkey=user, is_signer=True, is_writable=True), # payer
AccountMeta(pubkey=accounts_info["authority"], is_signer=False, is_writable=False), # authority
AccountMeta(pubkey=accounts_info["global_config"], is_signer=False, is_writable=False), # global_config
AccountMeta(pubkey=accounts_info["platform_config"], is_signer=False, is_writable=False), # platform_config
AccountMeta(pubkey=accounts_info["pool_state"], is_signer=False, is_writable=True), # pool_state
AccountMeta(pubkey=accounts_info["user_base_token"], is_signer=False, is_writable=True), # user_base_token (tokens being sold)
AccountMeta(pubkey=wsol_account, is_signer=False, is_writable=True), # user_quote_token (WSOL received)
AccountMeta(pubkey=accounts_info["base_vault"], is_signer=False, is_writable=True), # base_vault (receives tokens)
AccountMeta(pubkey=accounts_info["quote_vault"], is_signer=False, is_writable=True), # quote_vault (sends WSOL)
AccountMeta(pubkey=token_info.mint, is_signer=False, is_writable=False), # base_token_mint
AccountMeta(pubkey=SystemAddresses.SOL_MINT, is_signer=False, is_writable=False), # quote_token_mint
AccountMeta(pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False), # base_token_program
AccountMeta(pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False), # quote_token_program
AccountMeta(pubkey=accounts_info["event_authority"], is_signer=False, is_writable=False), # event_authority
AccountMeta(pubkey=accounts_info["program"], is_signer=False, is_writable=False), # program
AccountMeta(pubkey=accounts_info["global"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["fee"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["bonding_curve"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["associated_bonding_curve"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user_token_account"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user"], is_signer=True, is_writable=True),
AccountMeta(pubkey=accounts_info["system_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["creator_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["token_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["event_authority"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["program"], is_signer=False, is_writable=False),
]
# Build instruction data: discriminator + amount_in + minimum_amount_out + share_fee_rate
SHARE_FEE_RATE = 0 # No sharing fee
# Build instruction data: discriminator + token_amount + min_sol_output
instruction_data = (
self._sell_exact_in_discriminator +
struct.pack("<Q", amount_in) + # amount_in (u64) - tokens to sell
struct.pack("<Q", minimum_amount_out) + # minimum_amount_out (u64) - min SOL
struct.pack("<Q", SHARE_FEE_RATE) # share_fee_rate (u64): 0
self._sell_discriminator +
struct.pack("<Q", amount_in) + # token amount in raw units
struct.pack("<Q", minimum_amount_out) # min SOL output in lamports
)
sell_instruction = Instruction(
@@ -242,14 +165,6 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
)
instructions.append(sell_instruction)
# 4. Close WSOL account to reclaim SOL
close_wsol_ix = self._create_close_account_instruction(
wsol_account,
user,
user
)
instructions.append(close_wsol_ix)
return instructions
def get_required_accounts_for_buy(
@@ -271,12 +186,14 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
accounts_info = address_provider.get_buy_instruction_accounts(token_info, user)
return [
accounts_info["pool_state"],
accounts_info["user_base_token"],
accounts_info["base_vault"],
accounts_info["quote_vault"],
token_info.mint,
SystemAddresses.SOL_MINT,
accounts_info["mint"],
accounts_info["bonding_curve"],
accounts_info["associated_bonding_curve"],
accounts_info["user_token_account"],
accounts_info["fee"],
accounts_info["creator_vault"],
accounts_info["global_volume_accumulator"],
accounts_info["user_volume_accumulator"],
accounts_info["program"],
]
@@ -299,91 +216,15 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
accounts_info = address_provider.get_sell_instruction_accounts(token_info, user)
return [
accounts_info["pool_state"],
accounts_info["user_base_token"],
accounts_info["base_vault"],
accounts_info["quote_vault"],
token_info.mint,
SystemAddresses.SOL_MINT,
accounts_info["mint"],
accounts_info["bonding_curve"],
accounts_info["associated_bonding_curve"],
accounts_info["user_token_account"],
accounts_info["fee"],
accounts_info["creator_vault"],
accounts_info["program"],
]
def _generate_wsol_seed(self, user: Pubkey) -> str:
"""Generate a unique seed for WSOL account creation.
Args:
user: User's wallet address
Returns:
Unique seed string for WSOL account
"""
# Generate a unique seed based on timestamp and user pubkey
seed_data = f"{int(time.time())}{user!s}"
return hashlib.sha256(seed_data.encode()).hexdigest()[:32]
def _create_initialize_account_instruction(
self,
account: Pubkey,
mint: Pubkey,
owner: Pubkey
) -> Instruction:
"""Create an InitializeAccount instruction for the Token Program.
Args:
account: The account to initialize
mint: The token mint
owner: The account owner
Returns:
Instruction for initializing the account
"""
accounts = [
AccountMeta(pubkey=account, is_signer=False, is_writable=True),
AccountMeta(pubkey=mint, is_signer=False, is_writable=False),
AccountMeta(pubkey=owner, is_signer=False, is_writable=False),
AccountMeta(pubkey=SystemAddresses.RENT, is_signer=False, is_writable=False),
]
# InitializeAccount instruction discriminator (instruction 1 in Token Program)
data = bytes([1])
return Instruction(
program_id=SystemAddresses.TOKEN_PROGRAM,
data=data,
accounts=accounts
)
def _create_close_account_instruction(
self,
account: Pubkey,
destination: Pubkey,
owner: Pubkey
) -> Instruction:
"""Create a CloseAccount instruction for the Token Program.
Args:
account: The account to close
destination: Where to send the remaining lamports
owner: The account owner (must sign)
Returns:
Instruction for closing the account
"""
accounts = [
AccountMeta(pubkey=account, is_signer=False, is_writable=True),
AccountMeta(pubkey=destination, is_signer=False, is_writable=True),
AccountMeta(pubkey=owner, is_signer=True, is_writable=False),
]
# CloseAccount instruction discriminator (instruction 9 in Token Program)
data = bytes([9])
return Instruction(
program_id=SystemAddresses.TOKEN_PROGRAM,
data=data,
accounts=accounts
)
def calculate_token_amount_raw(self, token_amount_decimal: float) -> int:
"""Convert decimal token amount to raw token units.