gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-08-16 08:48:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(core): cleanup address management system

Browse Source
This commit is contained in:
smypmsa
2025-08-02 14:03:41 +00:00
parent 227ef0b3bd
commit 8ee6b95d2d
7 changed files with 183 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/core/pubkeys.py
+24 -53
View File
@@ -1,5 +1,7 @@
""" """
System and program addresses for Solana and pump.fun interactions. System addresses and constants for Solana blockchain operations.
This module contains only system-level addresses that are shared across all platforms.
Platform-specific addresses are handled by their respective AddressProvider implementations.
""" """
from dataclasses import dataclass from dataclasses import dataclass
@@ -7,76 +9,45 @@ from typing import Final
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
# Constants
LAMPORTS_PER_SOL: Final[int] = 1_000_000_000 LAMPORTS_PER_SOL: Final[int] = 1_000_000_000
TOKEN_DECIMALS: Final[int] = 6 TOKEN_DECIMALS: Final[int] = 6
@dataclass @dataclass
class SystemAddresses: class SystemAddresses:
"""System-level Solana addresses.""" """System-level Solana addresses shared across all platforms."""
PROGRAM: Final[Pubkey] = Pubkey.from_string("11111111111111111111111111111111") # Core system programs
SYSTEM_PROGRAM: Final[Pubkey] = Pubkey.from_string("11111111111111111111111111111111")
TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string( TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string(
"TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
) )
ASSOCIATED_TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string( ASSOCIATED_TOKEN_PROGRAM: Final[Pubkey] = Pubkey.from_string(
"ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL" "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL"
) )
# System accounts
RENT: Final[Pubkey] = Pubkey.from_string( RENT: Final[Pubkey] = Pubkey.from_string(
"SysvarRent111111111111111111111111111111111" "SysvarRent111111111111111111111111111111111"
) )
SOL: Final[Pubkey] = Pubkey.from_string(
# Native SOL token
SOL_MINT: Final[Pubkey] = Pubkey.from_string(
"So11111111111111111111111111111111111111112" "So11111111111111111111111111111111111111112"
) )
@classmethod
@dataclass def get_all_system_addresses(cls) -> dict[str, Pubkey]:
class PumpAddresses: """Get all system addresses as a dictionary.
"""Pump.fun program addresses."""
PROGRAM: Final[Pubkey] = Pubkey.from_string(
"6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
)
GLOBAL: Final[Pubkey] = Pubkey.from_string(
"4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf"
)
EVENT_AUTHORITY: Final[Pubkey] = Pubkey.from_string(
"Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1"
)
FEE: Final[Pubkey] = Pubkey.from_string(
"CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM"
)
LIQUIDITY_MIGRATOR: Final[Pubkey] = Pubkey.from_string(
"39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg"
)
@staticmethod
def find_global_volume_accumulator() -> Pubkey:
"""
Derive the Program Derived Address (PDA) for the global volume accumulator.
Returns: Returns:
Pubkey of the derived global volume accumulator account Dictionary mapping address names to Pubkey objects
""" """
derived_address, _ = Pubkey.find_program_address( return {
[b"global_volume_accumulator"], "system_program": cls.SYSTEM_PROGRAM,
PumpAddresses.PROGRAM, "token_program": cls.TOKEN_PROGRAM,
) "associated_token_program": cls.ASSOCIATED_TOKEN_PROGRAM,
return derived_address "rent": cls.RENT,
"sol_mint": cls.SOL_MINT,
@staticmethod }
def find_user_volume_accumulator(user: Pubkey) -> Pubkey:
"""
Derive the Program Derived Address (PDA) for a user's volume accumulator.
Args:
user: Pubkey of the user account
Returns:
Pubkey of the derived user volume accumulator account
"""
derived_address, _ = Pubkey.find_program_address(
[b"user_volume_accumulator", bytes(user)],
PumpAddresses.PROGRAM,
)
return derived_address
src/monitoring/pumpportal_event_processor.py
+9 -7
View File
@@ -4,8 +4,9 @@ Event processing for pump.fun tokens using PumpPortal data.
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
from core.pubkeys import PumpAddresses, SystemAddresses from core.pubkeys import SystemAddresses
from trading.base import TokenInfo from interfaces.core import Platform, TokenInfo
from platforms.pumpfun.address_provider import PumpFunAddresses
from utils.logger import get_logger from utils.logger import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -14,13 +15,13 @@ logger = get_logger(__name__)
class PumpPortalEventProcessor: class PumpPortalEventProcessor:
"""Processes token creation events from PumpPortal WebSocket.""" """Processes token creation events from PumpPortal WebSocket."""
def __init__(self, pump_program: Pubkey): def __init__(self, pump_program: Pubkey | None = None):
"""Initialize event processor. """Initialize event processor.
Args: Args:
pump_program: Pump.fun program address pump_program: Pump.fun program address (optional, will use default if None)
""" """
self.pump_program = pump_program self.pump_program = pump_program or PumpFunAddresses.PROGRAM
def process_token_data(self, token_data: dict) -> TokenInfo | None: def process_token_data(self, token_data: dict) -> TokenInfo | None:
"""Process token data from PumpPortal and extract token creation info. """Process token data from PumpPortal and extract token creation info.
@@ -71,6 +72,7 @@ class PumpPortalEventProcessor:
symbol=symbol, symbol=symbol,
uri=uri, uri=uri,
mint=mint, mint=mint,
platform=Platform.PUMP_FUN,
bonding_curve=bonding_curve, bonding_curve=bonding_curve,
associated_bonding_curve=associated_bonding_curve, associated_bonding_curve=associated_bonding_curve,
user=user, user=user,
@@ -118,6 +120,6 @@ class PumpPortalEventProcessor:
""" """
derived_address, _ = Pubkey.find_program_address( derived_address, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)], [b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM, PumpFunAddresses.PROGRAM,
) )
return derived_address return derived_address
src/monitoring/pumpportal_listener.py
+8 -6
View File
@@ -9,9 +9,10 @@ from collections.abc import Awaitable, Callable
import websockets import websockets
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
from interfaces.core import TokenInfo
from monitoring.base_listener import BaseTokenListener from monitoring.base_listener import BaseTokenListener
from monitoring.pumpportal_event_processor import PumpPortalEventProcessor from monitoring.pumpportal_event_processor import PumpPortalEventProcessor
from trading.base import TokenInfo from platforms.pumpfun.address_provider import PumpFunAddresses
from utils.logger import get_logger from utils.logger import get_logger
logger = get_logger(__name__) logger = get_logger(__name__)
@@ -22,18 +23,19 @@ class PumpPortalListener(BaseTokenListener):
def __init__( def __init__(
self, self,
pump_program: Pubkey, pump_program: Pubkey | None = None,
pumpportal_url: str = "wss://pumpportal.fun/api/data", pumpportal_url: str = "wss://pumpportal.fun/api/data",
): ):
"""Initialize token listener. """Initialize token listener.
Args: Args:
pump_program: Pump.fun program address pump_program: Pump.fun program address (optional, will use default if None)
pumpportal_url: PumpPortal WebSocket URL pumpportal_url: PumpPortal WebSocket URL
""" """
self.pump_program = pump_program super().__init__()
self.pump_program = pump_program or PumpFunAddresses.PROGRAM
self.pumpportal_url = pumpportal_url self.pumpportal_url = pumpportal_url
self.event_processor = PumpPortalEventProcessor(pump_program) self.event_processor = PumpPortalEventProcessor(self.pump_program)
self.ping_interval = 20 # seconds self.ping_interval = 20 # seconds
async def listen_for_tokens( async def listen_for_tokens(
@@ -171,4 +173,4 @@ class PumpPortalListener(BaseTokenListener):
except Exception as e: except Exception as e:
logger.error(f"Error processing PumpPortal WebSocket message: {e}") logger.error(f"Error processing PumpPortal WebSocket message: {e}")
return None return None
src/platforms/letsbonk/address_provider.py
+46 -45
View File
@@ -5,28 +5,29 @@ This module provides all LetsBonk (Raydium LaunchLab) specific addresses and PDA
by implementing the AddressProvider interface. by implementing the AddressProvider interface.
""" """
from dataclasses import dataclass
from typing import Final
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
from spl.token.instructions import get_associated_token_address from spl.token.instructions import get_associated_token_address
from core.pubkeys import SystemAddresses
from interfaces.core import AddressProvider, Platform, TokenInfo from interfaces.core import AddressProvider, Platform, TokenInfo
@dataclass
class LetsBonkAddresses:
"""LetsBonk (Raydium LaunchLab) program addresses."""
# Raydium LaunchLab program addresses
PROGRAM: Final[Pubkey] = Pubkey.from_string("LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj")
GLOBAL_CONFIG: Final[Pubkey] = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
PLATFORM_CONFIG: Final[Pubkey] = Pubkey.from_string("FfYek5vEz23cMkWsdJwG2oa6EphsvXSHrGpdALN4g6W1")
class LetsBonkAddressProvider(AddressProvider): class LetsBonkAddressProvider(AddressProvider):
"""LetsBonk (Raydium LaunchLab) implementation of AddressProvider interface.""" """LetsBonk (Raydium LaunchLab) implementation of AddressProvider interface."""
# Raydium LaunchLab program addresses
RAYDIUM_LAUNCHLAB_PROGRAM_ID = Pubkey.from_string("LanMV9sAd7wArD4vJFi2qDdfnVhFxYSUg6eADduJ3uj")
GLOBAL_CONFIG = Pubkey.from_string("6s1xP3hpbAfFoNtUNF8mfHsjr2Bd97JxFJRWLbL6aHuX")
LETSBONK_PLATFORM_CONFIG = Pubkey.from_string("FfYek5vEz23cMkWsdJwG2oa6EphsvXSHrGpdALN4g6W1")
# System program addresses
TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA")
SYSTEM_PROGRAM_ID = Pubkey.from_string("11111111111111111111111111111111")
WSOL_MINT = Pubkey.from_string("So11111111111111111111111111111111111111112")
ASSOCIATED_TOKEN_PROGRAM_ID = Pubkey.from_string("ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJA8knL")
SYSTEM_RENT_PROGRAM_ID = Pubkey.from_string("SysvarRent111111111111111111111111111111111")
@property @property
def platform(self) -> Platform: def platform(self) -> Platform:
"""Get the platform this provider serves.""" """Get the platform this provider serves."""
@@ -35,7 +36,7 @@ class LetsBonkAddressProvider(AddressProvider):
@property @property
def program_id(self) -> Pubkey: def program_id(self) -> Pubkey:
"""Get the main program ID for this platform.""" """Get the main program ID for this platform."""
return self.RAYDIUM_LAUNCHLAB_PROGRAM_ID return LetsBonkAddresses.PROGRAM
def get_system_addresses(self) -> dict[str, Pubkey]: def get_system_addresses(self) -> dict[str, Pubkey]:
"""Get all system addresses required for LetsBonk. """Get all system addresses required for LetsBonk.
@@ -43,19 +44,19 @@ class LetsBonkAddressProvider(AddressProvider):
Returns: Returns:
Dictionary mapping address names to Pubkey objects Dictionary mapping address names to Pubkey objects
""" """
return { # Get system addresses from the single source of truth
system_addresses = SystemAddresses.get_all_system_addresses()
# Add LetsBonk specific addresses
letsbonk_addresses = {
# Raydium LaunchLab specific addresses # Raydium LaunchLab specific addresses
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID, "program": LetsBonkAddresses.PROGRAM,
"global_config": self.GLOBAL_CONFIG, "global_config": LetsBonkAddresses.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_PLATFORM_CONFIG, "platform_config": LetsBonkAddresses.PLATFORM_CONFIG,
# System addresses
"system_program": self.SYSTEM_PROGRAM_ID,
"token_program": self.TOKEN_PROGRAM_ID,
"associated_token_program": self.ASSOCIATED_TOKEN_PROGRAM_ID,
"rent": self.SYSTEM_RENT_PROGRAM_ID,
"wsol_mint": self.WSOL_MINT,
} }
# Combine system and platform-specific addresses
return {**system_addresses, **letsbonk_addresses}
def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey: def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey:
"""Derive the pool state address for a token pair. """Derive the pool state address for a token pair.
@@ -70,11 +71,11 @@ class LetsBonkAddressProvider(AddressProvider):
Pool state address Pool state address
""" """
if quote_mint is None: if quote_mint is None:
quote_mint = self.WSOL_MINT quote_mint = SystemAddresses.SOL_MINT
pool_state, _ = Pubkey.find_program_address( pool_state, _ = Pubkey.find_program_address(
[b"pool", bytes(base_mint), bytes(quote_mint)], [b"pool", bytes(base_mint), bytes(quote_mint)],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID LetsBonkAddresses.PROGRAM
) )
return pool_state return pool_state
@@ -134,7 +135,7 @@ class LetsBonkAddressProvider(AddressProvider):
AUTH_SEED = b"vault_auth_seed" AUTH_SEED = b"vault_auth_seed"
authority_pda, _ = Pubkey.find_program_address( authority_pda, _ = Pubkey.find_program_address(
[AUTH_SEED], [AUTH_SEED],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID LetsBonkAddresses.PROGRAM
) )
return authority_pda return authority_pda
@@ -149,7 +150,7 @@ class LetsBonkAddressProvider(AddressProvider):
EVENT_AUTHORITY_SEED = b"__event_authority" EVENT_AUTHORITY_SEED = b"__event_authority"
event_authority_pda, _ = Pubkey.find_program_address( event_authority_pda, _ = Pubkey.find_program_address(
[EVENT_AUTHORITY_SEED], [EVENT_AUTHORITY_SEED],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID LetsBonkAddresses.PROGRAM
) )
return event_authority_pda return event_authority_pda
@@ -163,7 +164,7 @@ class LetsBonkAddressProvider(AddressProvider):
Returns: Returns:
New WSOL account address New WSOL account address
""" """
return Pubkey.create_with_seed(payer, seed, self.TOKEN_PROGRAM_ID) return Pubkey.create_with_seed(payer, seed, SystemAddresses.TOKEN_PROGRAM)
def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]: def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a buy instruction. """Get all accounts needed for a buy instruction.
@@ -180,18 +181,18 @@ class LetsBonkAddressProvider(AddressProvider):
return { return {
"payer": user, "payer": user,
"authority": additional_accounts["authority"], "authority": additional_accounts["authority"],
"global_config": self.GLOBAL_CONFIG, "global_config": LetsBonkAddresses.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_PLATFORM_CONFIG, "platform_config": LetsBonkAddresses.PLATFORM_CONFIG,
"pool_state": additional_accounts["pool_state"], "pool_state": additional_accounts["pool_state"],
"user_base_token": self.derive_user_token_account(user, token_info.mint), "user_base_token": self.derive_user_token_account(user, token_info.mint),
"base_vault": additional_accounts.get("base_vault", token_info.base_vault), "base_vault": additional_accounts.get("base_vault", token_info.base_vault),
"quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault), "quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault),
"base_token_mint": token_info.mint, "base_token_mint": token_info.mint,
"quote_token_mint": self.WSOL_MINT, "quote_token_mint": SystemAddresses.SOL_MINT,
"base_token_program": self.TOKEN_PROGRAM_ID, "base_token_program": SystemAddresses.TOKEN_PROGRAM,
"quote_token_program": self.TOKEN_PROGRAM_ID, "quote_token_program": SystemAddresses.TOKEN_PROGRAM,
"event_authority": additional_accounts["event_authority"], "event_authority": additional_accounts["event_authority"],
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID, "program": LetsBonkAddresses.PROGRAM,
} }
def get_sell_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]: def get_sell_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
@@ -209,18 +210,18 @@ class LetsBonkAddressProvider(AddressProvider):
return { return {
"payer": user, "payer": user,
"authority": additional_accounts["authority"], "authority": additional_accounts["authority"],
"global_config": self.GLOBAL_CONFIG, "global_config": LetsBonkAddresses.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_PLATFORM_CONFIG, "platform_config": LetsBonkAddresses.PLATFORM_CONFIG,
"pool_state": additional_accounts["pool_state"], "pool_state": additional_accounts["pool_state"],
"user_base_token": self.derive_user_token_account(user, token_info.mint), "user_base_token": self.derive_user_token_account(user, token_info.mint),
"base_vault": additional_accounts.get("base_vault", token_info.base_vault), "base_vault": additional_accounts.get("base_vault", token_info.base_vault),
"quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault), "quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault),
"base_token_mint": token_info.mint, "base_token_mint": token_info.mint,
"quote_token_mint": self.WSOL_MINT, "quote_token_mint": SystemAddresses.SOL_MINT,
"base_token_program": self.TOKEN_PROGRAM_ID, "base_token_program": SystemAddresses.TOKEN_PROGRAM,
"quote_token_program": self.TOKEN_PROGRAM_ID, "quote_token_program": SystemAddresses.TOKEN_PROGRAM,
"event_authority": additional_accounts["event_authority"], "event_authority": additional_accounts["event_authority"],
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID, "program": LetsBonkAddresses.PROGRAM,
} }
def get_wsol_account_creation_accounts(self, user: Pubkey, wsol_account: Pubkey) -> dict[str, Pubkey]: def get_wsol_account_creation_accounts(self, user: Pubkey, wsol_account: Pubkey) -> dict[str, Pubkey]:
@@ -236,9 +237,9 @@ class LetsBonkAddressProvider(AddressProvider):
return { return {
"payer": user, "payer": user,
"wsol_account": wsol_account, "wsol_account": wsol_account,
"wsol_mint": self.WSOL_MINT, "wsol_mint": SystemAddresses.SOL_MINT,
"owner": user, "owner": user,
"system_program": self.SYSTEM_PROGRAM_ID, "system_program": SystemAddresses.SYSTEM_PROGRAM,
"token_program": self.TOKEN_PROGRAM_ID, "token_program": SystemAddresses.TOKEN_PROGRAM,
"rent": self.SYSTEM_RENT_PROGRAM_ID, "rent": SystemAddresses.RENT,
} }
src/platforms/letsbonk/instruction_builder.py
+9 -9
View File
@@ -15,7 +15,7 @@ from solders.pubkey import Pubkey
from solders.system_program import CreateAccountWithSeedParams, create_account_with_seed from solders.system_program import CreateAccountWithSeedParams, create_account_with_seed
from spl.token.instructions import create_idempotent_associated_token_account from spl.token.instructions import create_idempotent_associated_token_account
from core.pubkeys import TOKEN_DECIMALS from core.pubkeys import TOKEN_DECIMALS, SystemAddresses
from interfaces.core import AddressProvider, InstructionBuilder, Platform, TokenInfo from interfaces.core import AddressProvider, InstructionBuilder, Platform, TokenInfo
# Instruction discriminators for LetsBonk (from IDL) # Instruction discriminators for LetsBonk (from IDL)
@@ -63,7 +63,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
user, # payer user, # payer
user, # owner user, # owner
token_info.mint, # mint token_info.mint, # mint
address_provider.TOKEN_PROGRAM_ID, # token program SystemAddresses.TOKEN_PROGRAM, # token program
) )
instructions.append(ata_instruction) instructions.append(ata_instruction)
@@ -83,7 +83,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
seed=wsol_seed, seed=wsol_seed,
lamports=total_lamports, lamports=total_lamports,
space=165, # Size of a token account space=165, # Size of a token account
owner=address_provider.TOKEN_PROGRAM_ID owner=SystemAddresses.TOKEN_PROGRAM
) )
) )
instructions.append(create_wsol_ix) instructions.append(create_wsol_ix)
@@ -91,7 +91,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
# 3. Initialize WSOL account # 3. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction( initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account, wsol_account,
address_provider.WSOL_MINT, SystemAddresses.SOL_MINT,
user, user,
address_provider address_provider
) )
@@ -183,7 +183,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
seed=wsol_seed, seed=wsol_seed,
lamports=account_creation_lamports, lamports=account_creation_lamports,
space=165, # Size of a token account space=165, # Size of a token account
owner=address_provider.TOKEN_PROGRAM_ID owner=SystemAddresses.TOKEN_PROGRAM
) )
) )
instructions.append(create_wsol_ix) instructions.append(create_wsol_ix)
@@ -191,7 +191,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
# 2. Initialize WSOL account # 2. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction( initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account, wsol_account,
address_provider.WSOL_MINT, SystemAddresses.SOL_MINT,
user, user,
address_provider address_provider
) )
@@ -334,14 +334,14 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
AccountMeta(pubkey=account, is_signer=False, is_writable=True), AccountMeta(pubkey=account, is_signer=False, is_writable=True),
AccountMeta(pubkey=mint, is_signer=False, is_writable=False), AccountMeta(pubkey=mint, is_signer=False, is_writable=False),
AccountMeta(pubkey=owner, is_signer=False, is_writable=False), AccountMeta(pubkey=owner, is_signer=False, is_writable=False),
AccountMeta(pubkey=address_provider.SYSTEM_RENT_PROGRAM_ID, is_signer=False, is_writable=False), AccountMeta(pubkey=SystemAddresses.RENT, is_signer=False, is_writable=False),
] ]
# InitializeAccount instruction discriminator (instruction 1 in Token Program) # InitializeAccount instruction discriminator (instruction 1 in Token Program)
data = bytes([1]) data = bytes([1])
return Instruction( return Instruction(
program_id=address_provider.TOKEN_PROGRAM_ID, program_id=SystemAddresses.TOKEN_PROGRAM,
data=data, data=data,
accounts=accounts accounts=accounts
) )
@@ -374,7 +374,7 @@ class LetsBonkInstructionBuilder(InstructionBuilder):
data = bytes([9]) data = bytes([9])
return Instruction( return Instruction(
program_id=address_provider.TOKEN_PROGRAM_ID, program_id=SystemAddresses.TOKEN_PROGRAM,
data=data, data=data,
accounts=accounts accounts=accounts
) )
src/platforms/pumpfun/address_provider.py
+83 -29
View File
@@ -5,14 +5,68 @@ This module provides all pump.fun-specific addresses and PDA derivations
by implementing the AddressProvider interface. by implementing the AddressProvider interface.
""" """
from dataclasses import dataclass
from typing import Final
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
from spl.token.instructions import get_associated_token_address from spl.token.instructions import get_associated_token_address
from core.pubkeys import PumpAddresses, SystemAddresses from core.pubkeys import SystemAddresses
from interfaces.core import AddressProvider, Platform, TokenInfo from interfaces.core import AddressProvider, Platform, TokenInfo
@dataclass
class PumpFunAddresses:
"""Pump.fun program addresses."""
PROGRAM: Final[Pubkey] = Pubkey.from_string(
"6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P"
)
GLOBAL: Final[Pubkey] = Pubkey.from_string(
"4wTV1YmiEkRvAtNtsSGPtUrqRYQMe5SKy2uB4Jjaxnjf"
)
EVENT_AUTHORITY: Final[Pubkey] = Pubkey.from_string(
"Ce6TQqeHC9p8KetsN6JsjHK7UTZk7nasjjnr7XxXp9F1"
)
FEE: Final[Pubkey] = Pubkey.from_string(
"CebN5WGQ4jvEPvsVU4EoHEpgzq1VV7AbicfhtW4xC9iM"
)
LIQUIDITY_MIGRATOR: Final[Pubkey] = Pubkey.from_string(
"39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg"
)
@staticmethod
def find_global_volume_accumulator() -> Pubkey:
"""
Derive the Program Derived Address (PDA) for the global volume accumulator.
Returns:
Pubkey of the derived global volume accumulator account
"""
derived_address, _ = Pubkey.find_program_address(
[b"global_volume_accumulator"],
PumpFunAddresses.PROGRAM,
)
return derived_address
@staticmethod
def find_user_volume_accumulator(user: Pubkey) -> Pubkey:
"""
Derive the Program Derived Address (PDA) for a user's volume accumulator.
Args:
user: Pubkey of the user account
Returns:
Pubkey of the derived user volume accumulator account
"""
derived_address, _ = Pubkey.find_program_address(
[b"user_volume_accumulator", bytes(user)],
PumpFunAddresses.PROGRAM,
)
return derived_address
class PumpFunAddressProvider(AddressProvider): class PumpFunAddressProvider(AddressProvider):
"""Pump.Fun implementation of AddressProvider interface.""" """Pump.Fun implementation of AddressProvider interface."""
@@ -24,7 +78,7 @@ class PumpFunAddressProvider(AddressProvider):
@property @property
def program_id(self) -> Pubkey: def program_id(self) -> Pubkey:
"""Get the main program ID for this platform.""" """Get the main program ID for this platform."""
return PumpAddresses.PROGRAM return PumpFunAddresses.PROGRAM
def get_system_addresses(self) -> dict[str, Pubkey]: def get_system_addresses(self) -> dict[str, Pubkey]:
"""Get all system addresses required for pump.fun. """Get all system addresses required for pump.fun.
@@ -32,21 +86,21 @@ class PumpFunAddressProvider(AddressProvider):
Returns: Returns:
Dictionary mapping address names to Pubkey objects Dictionary mapping address names to Pubkey objects
""" """
return { # Get system addresses from the single source of truth
system_addresses = SystemAddresses.get_all_system_addresses()
# Add pump.fun specific addresses
pumpfun_addresses = {
# Pump.fun specific addresses # Pump.fun specific addresses
"program": PumpAddresses.PROGRAM, "program": PumpFunAddresses.PROGRAM,
"global": PumpAddresses.GLOBAL, "global": PumpFunAddresses.GLOBAL,
"event_authority": PumpAddresses.EVENT_AUTHORITY, "event_authority": PumpFunAddresses.EVENT_AUTHORITY,
"fee": PumpAddresses.FEE, "fee": PumpFunAddresses.FEE,
"liquidity_migrator": PumpAddresses.LIQUIDITY_MIGRATOR, "liquidity_migrator": PumpFunAddresses.LIQUIDITY_MIGRATOR,
# System addresses
"system_program": SystemAddresses.PROGRAM,
"token_program": SystemAddresses.TOKEN_PROGRAM,
"associated_token_program": SystemAddresses.ASSOCIATED_TOKEN_PROGRAM,
"rent": SystemAddresses.RENT,
"sol_mint": SystemAddresses.SOL,
} }
# Combine system and platform-specific addresses
return {**system_addresses, **pumpfun_addresses}
def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey: def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey:
"""Derive the bonding curve address for a token. """Derive the bonding curve address for a token.
@@ -62,7 +116,7 @@ class PumpFunAddressProvider(AddressProvider):
""" """
bonding_curve, _ = Pubkey.find_program_address( bonding_curve, _ = Pubkey.find_program_address(
[b"bonding-curve", bytes(base_mint)], [b"bonding-curve", bytes(base_mint)],
PumpAddresses.PROGRAM PumpFunAddresses.PROGRAM
) )
return bonding_curve return bonding_curve
@@ -136,7 +190,7 @@ class PumpFunAddressProvider(AddressProvider):
""" """
creator_vault, _ = Pubkey.find_program_address( creator_vault, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)], [b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM PumpFunAddresses.PROGRAM
) )
return creator_vault return creator_vault
@@ -146,7 +200,7 @@ class PumpFunAddressProvider(AddressProvider):
Returns: Returns:
Global volume accumulator address Global volume accumulator address
""" """
return PumpAddresses.find_global_volume_accumulator() return PumpFunAddresses.find_global_volume_accumulator()
def derive_user_volume_accumulator(self, user: Pubkey) -> Pubkey: def derive_user_volume_accumulator(self, user: Pubkey) -> Pubkey:
"""Derive the user volume accumulator PDA. """Derive the user volume accumulator PDA.
@@ -157,7 +211,7 @@ class PumpFunAddressProvider(AddressProvider):
Returns: Returns:
User volume accumulator address User volume accumulator address
""" """
return PumpAddresses.find_user_volume_accumulator(user) return PumpFunAddresses.find_user_volume_accumulator(user)
def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]: def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a buy instruction. """Get all accounts needed for a buy instruction.
@@ -172,18 +226,18 @@ class PumpFunAddressProvider(AddressProvider):
additional_accounts = self.get_additional_accounts(token_info) additional_accounts = self.get_additional_accounts(token_info)
return { return {
"global": PumpAddresses.GLOBAL, "global": PumpFunAddresses.GLOBAL,
"fee": PumpAddresses.FEE, "fee": PumpFunAddresses.FEE,
"mint": token_info.mint, "mint": token_info.mint,
"bonding_curve": additional_accounts.get("bonding_curve", token_info.bonding_curve), "bonding_curve": additional_accounts.get("bonding_curve", token_info.bonding_curve),
"associated_bonding_curve": additional_accounts.get("associated_bonding_curve", token_info.associated_bonding_curve), "associated_bonding_curve": additional_accounts.get("associated_bonding_curve", token_info.associated_bonding_curve),
"user_token_account": self.derive_user_token_account(user, token_info.mint), "user_token_account": self.derive_user_token_account(user, token_info.mint),
"user": user, "user": user,
"system_program": SystemAddresses.PROGRAM, "system_program": SystemAddresses.SYSTEM_PROGRAM,
"token_program": SystemAddresses.TOKEN_PROGRAM, "token_program": SystemAddresses.TOKEN_PROGRAM,
"creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault), "creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault),
"event_authority": PumpAddresses.EVENT_AUTHORITY, "event_authority": PumpFunAddresses.EVENT_AUTHORITY,
"program": PumpAddresses.PROGRAM, "program": PumpFunAddresses.PROGRAM,
"global_volume_accumulator": self.derive_global_volume_accumulator(), "global_volume_accumulator": self.derive_global_volume_accumulator(),
"user_volume_accumulator": self.derive_user_volume_accumulator(user), "user_volume_accumulator": self.derive_user_volume_accumulator(user),
} }
@@ -201,16 +255,16 @@ class PumpFunAddressProvider(AddressProvider):
additional_accounts = self.get_additional_accounts(token_info) additional_accounts = self.get_additional_accounts(token_info)
return { return {
"global": PumpAddresses.GLOBAL, "global": PumpFunAddresses.GLOBAL,
"fee": PumpAddresses.FEE, "fee": PumpFunAddresses.FEE,
"mint": token_info.mint, "mint": token_info.mint,
"bonding_curve": additional_accounts.get("bonding_curve", token_info.bonding_curve), "bonding_curve": additional_accounts.get("bonding_curve", token_info.bonding_curve),
"associated_bonding_curve": additional_accounts.get("associated_bonding_curve", token_info.associated_bonding_curve), "associated_bonding_curve": additional_accounts.get("associated_bonding_curve", token_info.associated_bonding_curve),
"user_token_account": self.derive_user_token_account(user, token_info.mint), "user_token_account": self.derive_user_token_account(user, token_info.mint),
"user": user, "user": user,
"system_program": SystemAddresses.PROGRAM, "system_program": SystemAddresses.SYSTEM_PROGRAM,
"creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault), "creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault),
"token_program": SystemAddresses.TOKEN_PROGRAM, "token_program": SystemAddresses.TOKEN_PROGRAM,
"event_authority": PumpAddresses.EVENT_AUTHORITY, "event_authority": PumpFunAddresses.EVENT_AUTHORITY,
"program": PumpAddresses.PROGRAM, "program": PumpFunAddresses.PROGRAM,
} }
src/platforms/pumpfun/event_parser.py
+4 -3
View File
@@ -14,8 +14,9 @@ import base58
from solders.pubkey import Pubkey from solders.pubkey import Pubkey
from solders.transaction import VersionedTransaction from solders.transaction import VersionedTransaction
from core.pubkeys import PumpAddresses, SystemAddresses from core.pubkeys import SystemAddresses
from interfaces.core import EventParser, Platform, TokenInfo from interfaces.core import EventParser, Platform, TokenInfo
from platforms.pumpfun.address_provider import PumpFunAddresses
class PumpFunEventParser(EventParser): class PumpFunEventParser(EventParser):
@@ -180,7 +181,7 @@ class PumpFunEventParser(EventParser):
Returns: Returns:
Pump.fun program ID Pump.fun program ID
""" """
return PumpAddresses.PROGRAM return PumpFunAddresses.PROGRAM
def get_instruction_discriminators(self) -> list[bytes]: def get_instruction_discriminators(self) -> list[bytes]:
"""Get instruction discriminators for token creation. """Get instruction discriminators for token creation.
@@ -310,7 +311,7 @@ class PumpFunEventParser(EventParser):
""" """
derived_address, _ = Pubkey.find_program_address( derived_address, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)], [b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM, PumpFunAddresses.PROGRAM,
) )
return derived_address return derived_address