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

wip(core): platform aware trading

Browse Source
This commit is contained in:
smypmsa
2025-08-02 09:25:39 +00:00
parent 60c452da0f
commit bb35101a8f
31 changed files with 3907 additions and 1094 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/config_loader.py
+80 -1
View File
@@ -1,9 +1,19 @@
"""
Updated configuration validation with platform support.
This extends the existing config_loader.py to support platform selection
and validation while maintaining backward compatibility.
"""
import os
from typing import Any
import yaml
from dotenv import load_dotenv
from interfaces.core import Platform
# Existing validation rules (keeping all existing ones)
REQUIRED_FIELDS = [
"name",
"rpc_endpoint",
@@ -64,17 +74,19 @@ CONFIG_VALIDATION_RULES = [
),
]
# Valid values for enum-like fields
# Valid values for enum-like fields (extended with platform support)
VALID_VALUES = {
"filters.listener_type": ["logs", "blocks", "geyser", "pumpportal"],
"cleanup.mode": ["disabled", "on_fail", "after_sell", "post_session"],
"trade.exit_strategy": ["time_based", "tp_sl", "manual"],
"platform": ["pump_fun", "lets_bonk"], # New platform validation
}
def load_bot_config(path: str) -> dict:
"""
Load and validate a bot configuration from a YAML file.
Extended to support platform selection with backward compatibility.
Args:
path: Path to the YAML configuration file (relative or absolute)
@@ -99,6 +111,11 @@ def load_bot_config(path: str) -> dict:
load_dotenv(env_file, override=True)
resolve_env_vars(config)
# Set default platform if not specified (backward compatibility)
if "platform" not in config:
config["platform"] = "pump_fun"
validate_config(config)
return config
@@ -157,6 +174,7 @@ def get_nested_value(config: dict, path: str) -> Any:
def validate_config(config: dict) -> None:
"""
Validate the configuration against defined rules.
Extended to include platform-specific validation.
Args:
config: Configuration dictionary to validate
@@ -164,9 +182,11 @@ def validate_config(config: dict) -> None:
Raises:
ValueError: If the configuration is invalid
"""
# Validate required fields
for field in REQUIRED_FIELDS:
get_nested_value(config, field)
# Validate config rules
for path, expected_type, min_val, max_val, error_msg in CONFIG_VALIDATION_RULES:
try:
value = get_nested_value(config, path)
@@ -206,15 +226,70 @@ def validate_config(config: dict) -> None:
# Skip if one of the fields is missing
pass
# Platform-specific validation
platform_str = config.get("platform", "pump_fun")
try:
platform = Platform(platform_str)
validate_platform_config(config, platform)
except ValueError as e:
if "is not a valid" in str(e):
raise ValueError(f"Invalid platform '{platform_str}'. Must be one of: {[p.value for p in Platform]}")
raise
def validate_platform_config(config: dict, platform: Platform) -> None:
"""
Validate platform-specific configuration requirements.
Args:
config: Configuration dictionary
platform: Platform enum value
Raises:
ValueError: If platform-specific config is invalid
"""
if platform == Platform.PUMP_FUN:
# pump.fun doesn't require additional config beyond base requirements
pass
elif platform == Platform.LETS_BONK:
# LetsBonk may require additional configuration in the future
# For now, it uses the same base configuration as pump.fun
pass
def get_platform_from_config(config: dict) -> Platform:
"""
Extract platform enum from configuration.
Args:
config: Configuration dictionary
Returns:
Platform enum value
Raises:
ValueError: If platform is invalid
"""
platform_str = config.get("platform", "pump_fun")
try:
return Platform(platform_str)
except ValueError:
raise ValueError(f"Invalid platform '{platform_str}'. Must be one of: {[p.value for p in Platform]}")
def print_config_summary(config: dict) -> None:
"""
Print a summary of the loaded configuration.
Extended to show platform information.
Args:
config: Configuration dictionary
"""
platform_str = config.get("platform", "pump_fun")
print(f"Bot name: {config.get('name', 'unnamed')}")
print(f"Platform: {platform_str}")
print(
f"Listener type: {config.get('filters', {}).get('listener_type', 'not configured')}"
)
@@ -240,5 +315,9 @@ def print_config_summary(config: dict) -> None:
if __name__ == "__main__":
# Example usage with platform configuration
config = load_bot_config("bots/bot-sniper.yaml")
print_config_summary(config)
platform = get_platform_from_config(config)
print(f"Detected platform: {platform}")
src/interfaces/__init__.py
+280
View File
@@ -0,0 +1,280 @@
"""
Platform factory and registry for managing multiple trading platforms.
This module provides a centralized way to instantiate and access
platform-specific implementations of the trading interfaces.
"""
from dataclasses import dataclass
from typing import Any, Dict, Optional, Type
from core.client import SolanaClient
from interfaces.core import (
AddressProvider,
CurveManager,
EventParser,
InstructionBuilder,
Platform,
)
@dataclass
class PlatformImplementations:
"""Container for all platform-specific implementations."""
address_provider: AddressProvider
instruction_builder: InstructionBuilder
curve_manager: CurveManager
event_parser: EventParser
class PlatformRegistry:
"""Registry for platform implementations."""
def __init__(self):
self._implementations: dict[Platform, dict[str, type]] = {}
self._instances: dict[Platform, PlatformImplementations] = {}
def register_platform(
self,
platform: Platform,
address_provider_class: type[AddressProvider],
instruction_builder_class: type[InstructionBuilder],
curve_manager_class: type[CurveManager],
event_parser_class: type[EventParser]
) -> None:
"""Register platform implementations.
Args:
platform: Platform enum value
address_provider_class: AddressProvider implementation class
instruction_builder_class: InstructionBuilder implementation class
curve_manager_class: CurveManager implementation class
event_parser_class: EventParser implementation class
"""
self._implementations[platform] = {
'address_provider': address_provider_class,
'instruction_builder': instruction_builder_class,
'curve_manager': curve_manager_class,
'event_parser': event_parser_class
}
def create_platform_implementations(
self,
platform: Platform,
client: SolanaClient,
**kwargs: Any
) -> PlatformImplementations:
"""Create platform implementation instances.
Args:
platform: Platform to create implementations for
client: Solana RPC client
**kwargs: Additional arguments for implementation constructors
Returns:
PlatformImplementations containing all interface implementations
Raises:
ValueError: If platform is not registered
"""
if platform not in self._implementations:
raise ValueError(f"Platform {platform} is not registered")
# Check if we already have instances for this platform
if platform in self._instances:
return self._instances[platform]
impl_classes = self._implementations[platform]
# Create instances
address_provider = impl_classes['address_provider']()
instruction_builder = impl_classes['instruction_builder']()
curve_manager = impl_classes['curve_manager'](client)
event_parser = impl_classes['event_parser']()
implementations = PlatformImplementations(
address_provider=address_provider,
instruction_builder=instruction_builder,
curve_manager=curve_manager,
event_parser=event_parser
)
# Cache the instances
self._instances[platform] = implementations
return implementations
def get_platform_implementations(self, platform: Platform) -> PlatformImplementations | None:
"""Get cached platform implementations.
Args:
platform: Platform to get implementations for
Returns:
PlatformImplementations if available, None otherwise
"""
return self._instances.get(platform)
def get_supported_platforms(self) -> list[Platform]:
"""Get list of supported platforms.
Returns:
List of registered platforms
"""
return list(self._implementations.keys())
def is_platform_supported(self, platform: Platform) -> bool:
"""Check if a platform is supported.
Args:
platform: Platform to check
Returns:
True if platform is registered, False otherwise
"""
return platform in self._implementations
class PlatformFactory:
"""Factory for creating platform-specific implementations."""
def __init__(self):
self.registry = PlatformRegistry()
self._setup_default_platforms()
def _setup_default_platforms(self) -> None:
"""Setup default platform registrations.
This will be populated as we implement each platform.
"""
# Platforms will be registered here as they are implemented
# Example:
# from platforms.pumpfun import PumpFunAddressProvider, PumpFunInstructionBuilder, ...
# self.registry.register_platform(
# Platform.PUMP_FUN,
# PumpFunAddressProvider,
# PumpFunInstructionBuilder,
# PumpFunCurveManager,
# PumpFunEventParser
# )
pass
def create_for_platform(
self,
platform: Platform,
client: SolanaClient,
**config: Any
) -> PlatformImplementations:
"""Create all implementations for a specific platform.
Args:
platform: Platform to create implementations for
client: Solana RPC client
**config: Platform-specific configuration
Returns:
PlatformImplementations containing all interface implementations
"""
return self.registry.create_platform_implementations(platform, client, **config)
def get_address_provider(self, platform: Platform, client: SolanaClient) -> AddressProvider:
"""Get address provider for platform.
Args:
platform: Platform to get provider for
client: Solana RPC client
Returns:
AddressProvider implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.address_provider
def get_instruction_builder(self, platform: Platform, client: SolanaClient) -> InstructionBuilder:
"""Get instruction builder for platform.
Args:
platform: Platform to get builder for
client: Solana RPC client
Returns:
InstructionBuilder implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.instruction_builder
def get_curve_manager(self, platform: Platform, client: SolanaClient) -> CurveManager:
"""Get curve manager for platform.
Args:
platform: Platform to get manager for
client: Solana RPC client
Returns:
CurveManager implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.curve_manager
def get_event_parser(self, platform: Platform, client: SolanaClient) -> EventParser:
"""Get event parser for platform.
Args:
platform: Platform to get parser for
client: Solana RPC client
Returns:
EventParser implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.event_parser
def get_supported_platforms(self) -> list[Platform]:
"""Get list of supported platforms.
Returns:
List of supported platforms
"""
return self.registry.get_supported_platforms()
# Global factory instance
platform_factory = PlatformFactory()
def get_platform_implementations(platform: Platform, client: SolanaClient) -> PlatformImplementations:
"""Convenience function to get platform implementations.
Args:
platform: Platform to get implementations for
client: Solana RPC client
Returns:
PlatformImplementations containing all interface implementations
"""
return platform_factory.create_for_platform(platform, client)
def register_platform_implementations(
platform: Platform,
address_provider_class: type[AddressProvider],
instruction_builder_class: type[InstructionBuilder],
curve_manager_class: type[CurveManager],
event_parser_class: type[EventParser]
) -> None:
"""Register platform implementations with the global factory.
Args:
platform: Platform enum value
address_provider_class: AddressProvider implementation class
instruction_builder_class: InstructionBuilder implementation class
curve_manager_class: CurveManager implementation class
event_parser_class: EventParser implementation class
"""
platform_factory.registry.register_platform(
platform,
address_provider_class,
instruction_builder_class,
curve_manager_class,
event_parser_class
)
src/interfaces/core.py
+363
View File
@@ -0,0 +1,363 @@
"""
Core interfaces for multi-platform trading bot architecture.
This module defines the abstract base classes that each trading platform
must implement to enable unified trading operations across different protocols.
"""
from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
from typing import Any
from solders.instruction import Instruction
from solders.pubkey import Pubkey
class Platform(Enum):
"""Supported trading platforms."""
PUMP_FUN = "pump_fun"
LETS_BONK = "lets_bonk"
@dataclass
class TokenInfo:
"""Enhanced token information with platform support."""
# Core token data
name: str
symbol: str
uri: str
mint: Pubkey
# Platform-specific fields
platform: Platform
bonding_curve: Pubkey | None = None # pump.fun specific
associated_bonding_curve: Pubkey | None = None # pump.fun specific
pool_state: Pubkey | None = None # LetsBonk specific
base_vault: Pubkey | None = None # LetsBonk specific
quote_vault: Pubkey | None = None # LetsBonk specific
# Common fields
user: Pubkey | None = None
creator: Pubkey | None = None
creator_vault: Pubkey | None = None
# Metadata
creation_timestamp: float | None = None
additional_data: dict[str, Any] | None = None
class AddressProvider(ABC):
"""Abstract interface for platform-specific address management."""
@property
@abstractmethod
def platform(self) -> Platform:
"""Get the platform this provider serves."""
pass
@property
@abstractmethod
def program_id(self) -> Pubkey:
"""Get the main program ID for this platform."""
pass
@abstractmethod
def get_system_addresses(self) -> dict[str, Pubkey]:
"""Get all system addresses required for this platform.
Returns:
Dictionary mapping address names to Pubkey objects
"""
pass
@abstractmethod
def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey:
"""Derive the pool/curve address for trading pair.
Args:
base_mint: Base token mint address
quote_mint: Quote token mint address (if applicable)
Returns:
Pool/curve address for the trading pair
"""
pass
@abstractmethod
def derive_user_token_account(self, user: Pubkey, mint: Pubkey) -> Pubkey:
"""Derive user's token account address.
Args:
user: User's wallet address
mint: Token mint address
Returns:
User's token account address
"""
pass
@abstractmethod
def get_additional_accounts(self, token_info: TokenInfo) -> dict[str, Pubkey]:
"""Get platform-specific additional accounts needed for trading.
Args:
token_info: Token information
Returns:
Dictionary of additional account addresses
"""
pass
class InstructionBuilder(ABC):
"""Abstract interface for building platform-specific trading instructions."""
@property
@abstractmethod
def platform(self) -> Platform:
"""Get the platform this builder serves."""
pass
@abstractmethod
async def build_buy_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build buy instruction(s) for the platform.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of quote tokens to spend
minimum_amount_out: Minimum base tokens expected
address_provider: Platform address provider
Returns:
List of instructions needed for the buy operation
"""
pass
@abstractmethod
async def build_sell_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build sell instruction(s) for the platform.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of base tokens to sell
minimum_amount_out: Minimum quote tokens expected
address_provider: Platform address provider
Returns:
List of instructions needed for the sell operation
"""
pass
@abstractmethod
def get_required_accounts_for_buy(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for buy operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
pass
@abstractmethod
def get_required_accounts_for_sell(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for sell operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
pass
class CurveManager(ABC):
"""Abstract interface for platform-specific price calculations and pool state management."""
@property
@abstractmethod
def platform(self) -> Platform:
"""Get the platform this manager serves."""
pass
@abstractmethod
async def get_pool_state(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get the current state of a trading pool/curve.
Args:
pool_address: Address of the pool/curve
Returns:
Dictionary containing pool state data
"""
pass
@abstractmethod
async def calculate_price(self, pool_address: Pubkey) -> float:
"""Calculate current token price from pool state.
Args:
pool_address: Address of the pool/curve
Returns:
Current token price in quote token units
"""
pass
@abstractmethod
async def calculate_buy_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected tokens received for a buy operation.
Args:
pool_address: Address of the pool/curve
amount_in: Amount of quote tokens to spend
Returns:
Expected amount of base tokens to receive
"""
pass
@abstractmethod
async def calculate_sell_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected quote tokens received for a sell operation.
Args:
pool_address: Address of the pool/curve
amount_in: Amount of base tokens to sell
Returns:
Expected amount of quote tokens to receive
"""
pass
@abstractmethod
async def get_reserves(self, pool_address: Pubkey) -> tuple[int, int]:
"""Get current pool reserves.
Args:
pool_address: Address of the pool/curve
Returns:
Tuple of (base_reserves, quote_reserves)
"""
pass
class EventParser(ABC):
"""Abstract interface for parsing platform-specific token creation events."""
@property
@abstractmethod
def platform(self) -> Platform:
"""Get the platform this parser serves."""
pass
@abstractmethod
def parse_token_creation_from_logs(
self,
logs: list[str],
signature: str
) -> TokenInfo | None:
"""Parse token creation from transaction logs.
Args:
logs: List of log strings from transaction
signature: Transaction signature
Returns:
TokenInfo if token creation found, None otherwise
"""
pass
@abstractmethod
def parse_token_creation_from_instruction(
self,
instruction_data: bytes,
accounts: list[int],
account_keys: list[bytes]
) -> TokenInfo | None:
"""Parse token creation from instruction data.
Args:
instruction_data: Raw instruction data
accounts: List of account indices
account_keys: List of account public keys
Returns:
TokenInfo if token creation found, None otherwise
"""
pass
@abstractmethod
def parse_token_creation_from_geyser(
self,
transaction_info: Any
) -> TokenInfo | None:
"""Parse token creation from Geyser transaction data.
Args:
transaction_info: Geyser transaction information
Returns:
TokenInfo if token creation found, None otherwise
"""
pass
@abstractmethod
def get_program_id(self) -> Pubkey:
"""Get the program ID this parser monitors.
Returns:
Program ID for event filtering
"""
pass
@abstractmethod
def get_instruction_discriminators(self) -> list[bytes]:
"""Get instruction discriminators for token creation.
Returns:
List of discriminator bytes to match
"""
pass
src/platforms/__init__.py
+282
View File
@@ -0,0 +1,282 @@
"""
Platform factory and registry for managing multiple trading platforms.
This module provides a centralized way to instantiate and access
platform-specific implementations of the trading interfaces.
"""
from dataclasses import dataclass
from typing import Any, Dict, Optional, Type
from core.client import SolanaClient
from interfaces.core import (
AddressProvider,
CurveManager,
EventParser,
InstructionBuilder,
Platform,
)
from . import letsbonk, pumpfun
@dataclass
class PlatformImplementations:
"""Container for all platform-specific implementations."""
address_provider: AddressProvider
instruction_builder: InstructionBuilder
curve_manager: CurveManager
event_parser: EventParser
class PlatformRegistry:
"""Registry for platform implementations."""
def __init__(self):
self._implementations: dict[Platform, dict[str, type]] = {}
self._instances: dict[Platform, PlatformImplementations] = {}
def register_platform(
self,
platform: Platform,
address_provider_class: type[AddressProvider],
instruction_builder_class: type[InstructionBuilder],
curve_manager_class: type[CurveManager],
event_parser_class: type[EventParser]
) -> None:
"""Register platform implementations.
Args:
platform: Platform enum value
address_provider_class: AddressProvider implementation class
instruction_builder_class: InstructionBuilder implementation class
curve_manager_class: CurveManager implementation class
event_parser_class: EventParser implementation class
"""
self._implementations[platform] = {
'address_provider': address_provider_class,
'instruction_builder': instruction_builder_class,
'curve_manager': curve_manager_class,
'event_parser': event_parser_class
}
def create_platform_implementations(
self,
platform: Platform,
client: SolanaClient,
**kwargs: Any
) -> PlatformImplementations:
"""Create platform implementation instances.
Args:
platform: Platform to create implementations for
client: Solana RPC client
**kwargs: Additional arguments for implementation constructors
Returns:
PlatformImplementations containing all interface implementations
Raises:
ValueError: If platform is not registered
"""
if platform not in self._implementations:
raise ValueError(f"Platform {platform} is not registered")
# Check if we already have instances for this platform
if platform in self._instances:
return self._instances[platform]
impl_classes = self._implementations[platform]
# Create instances
address_provider = impl_classes['address_provider']()
instruction_builder = impl_classes['instruction_builder']()
curve_manager = impl_classes['curve_manager'](client)
event_parser = impl_classes['event_parser']()
implementations = PlatformImplementations(
address_provider=address_provider,
instruction_builder=instruction_builder,
curve_manager=curve_manager,
event_parser=event_parser
)
# Cache the instances
self._instances[platform] = implementations
return implementations
def get_platform_implementations(self, platform: Platform) -> PlatformImplementations | None:
"""Get cached platform implementations.
Args:
platform: Platform to get implementations for
Returns:
PlatformImplementations if available, None otherwise
"""
return self._instances.get(platform)
def get_supported_platforms(self) -> list[Platform]:
"""Get list of supported platforms.
Returns:
List of registered platforms
"""
return list(self._implementations.keys())
def is_platform_supported(self, platform: Platform) -> bool:
"""Check if a platform is supported.
Args:
platform: Platform to check
Returns:
True if platform is registered, False otherwise
"""
return platform in self._implementations
class PlatformFactory:
"""Factory for creating platform-specific implementations."""
def __init__(self):
self.registry = PlatformRegistry()
self._setup_default_platforms()
def _setup_default_platforms(self) -> None:
"""Setup default platform registrations.
This will be populated as we implement each platform.
"""
# Platforms will be registered here as they are implemented
# Example:
# from platforms.pumpfun import PumpFunAddressProvider, PumpFunInstructionBuilder, ...
# self.registry.register_platform(
# Platform.PUMP_FUN,
# PumpFunAddressProvider,
# PumpFunInstructionBuilder,
# PumpFunCurveManager,
# PumpFunEventParser
# )
pass
def create_for_platform(
self,
platform: Platform,
client: SolanaClient,
**config: Any
) -> PlatformImplementations:
"""Create all implementations for a specific platform.
Args:
platform: Platform to create implementations for
client: Solana RPC client
**config: Platform-specific configuration
Returns:
PlatformImplementations containing all interface implementations
"""
return self.registry.create_platform_implementations(platform, client, **config)
def get_address_provider(self, platform: Platform, client: SolanaClient) -> AddressProvider:
"""Get address provider for platform.
Args:
platform: Platform to get provider for
client: Solana RPC client
Returns:
AddressProvider implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.address_provider
def get_instruction_builder(self, platform: Platform, client: SolanaClient) -> InstructionBuilder:
"""Get instruction builder for platform.
Args:
platform: Platform to get builder for
client: Solana RPC client
Returns:
InstructionBuilder implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.instruction_builder
def get_curve_manager(self, platform: Platform, client: SolanaClient) -> CurveManager:
"""Get curve manager for platform.
Args:
platform: Platform to get manager for
client: Solana RPC client
Returns:
CurveManager implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.curve_manager
def get_event_parser(self, platform: Platform, client: SolanaClient) -> EventParser:
"""Get event parser for platform.
Args:
platform: Platform to get parser for
client: Solana RPC client
Returns:
EventParser implementation
"""
implementations = self.registry.create_platform_implementations(platform, client)
return implementations.event_parser
def get_supported_platforms(self) -> list[Platform]:
"""Get list of supported platforms.
Returns:
List of supported platforms
"""
return self.registry.get_supported_platforms()
# Global factory instance
platform_factory = PlatformFactory()
def get_platform_implementations(platform: Platform, client: SolanaClient) -> PlatformImplementations:
"""Convenience function to get platform implementations.
Args:
platform: Platform to get implementations for
client: Solana RPC client
Returns:
PlatformImplementations containing all interface implementations
"""
return platform_factory.create_for_platform(platform, client)
def register_platform_implementations(
platform: Platform,
address_provider_class: type[AddressProvider],
instruction_builder_class: type[InstructionBuilder],
curve_manager_class: type[CurveManager],
event_parser_class: type[EventParser]
) -> None:
"""Register platform implementations with the global factory.
Args:
platform: Platform enum value
address_provider_class: AddressProvider implementation class
instruction_builder_class: InstructionBuilder implementation class
curve_manager_class: CurveManager implementation class
event_parser_class: EventParser implementation class
"""
platform_factory.registry.register_platform(
platform,
address_provider_class,
instruction_builder_class,
curve_manager_class,
event_parser_class
)
src/platforms/letsbonk/__init__.py
+31
View File
@@ -0,0 +1,31 @@
"""
LetsBonk platform registration and exports.
This module registers all LetsBonk implementations with the platform factory
and provides convenient imports for the LetsBonk platform.
"""
from interfaces.core import Platform
from platforms import register_platform_implementations
from .address_provider import LetsBonkAddressProvider
from .curve_manager import LetsBonkCurveManager
from .event_parser import LetsBonkEventParser
from .instruction_builder import LetsBonkInstructionBuilder
# Register LetsBonk platform implementations
register_platform_implementations(
Platform.LETS_BONK,
LetsBonkAddressProvider,
LetsBonkInstructionBuilder,
LetsBonkCurveManager,
LetsBonkEventParser
)
# Export implementations for direct use if needed
__all__ = [
'LetsBonkAddressProvider',
'LetsBonkCurveManager',
'LetsBonkEventParser',
'LetsBonkInstructionBuilder'
]
src/platforms/letsbonk/address_provider.py
+244
View File
@@ -0,0 +1,244 @@
"""
LetsBonk implementation of AddressProvider interface.
This module provides all LetsBonk (Raydium LaunchLab) specific addresses and PDA derivations
by implementing the AddressProvider interface.
"""
from solders.pubkey import Pubkey
from spl.token.instructions import get_associated_token_address
from interfaces.core import AddressProvider, Platform, TokenInfo
class LetsBonkAddressProvider(AddressProvider):
"""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
def platform(self) -> Platform:
"""Get the platform this provider serves."""
return Platform.LETS_BONK
@property
def program_id(self) -> Pubkey:
"""Get the main program ID for this platform."""
return self.RAYDIUM_LAUNCHLAB_PROGRAM_ID
def get_system_addresses(self) -> dict[str, Pubkey]:
"""Get all system addresses required for LetsBonk.
Returns:
Dictionary mapping address names to Pubkey objects
"""
return {
# Raydium LaunchLab specific addresses
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID,
"global_config": self.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_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,
}
def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey:
"""Derive the pool state address for a token pair.
For LetsBonk, this derives the pool state PDA using base_mint and WSOL.
Args:
base_mint: Base token mint address
quote_mint: Quote token mint (defaults to WSOL)
Returns:
Pool state address
"""
if quote_mint is None:
quote_mint = self.WSOL_MINT
pool_state, _ = Pubkey.find_program_address(
[b"pool", bytes(base_mint), bytes(quote_mint)],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID
)
return pool_state
def derive_user_token_account(self, user: Pubkey, mint: Pubkey) -> Pubkey:
"""Derive user's associated token account address.
Args:
user: User's wallet address
mint: Token mint address
Returns:
User's associated token account address
"""
return get_associated_token_address(user, mint)
def get_additional_accounts(self, token_info: TokenInfo) -> dict[str, Pubkey]:
"""Get LetsBonk-specific additional accounts needed for trading.
Args:
token_info: Token information
Returns:
Dictionary of additional account addresses
"""
accounts = {}
# Add pool state if available
if token_info.pool_state:
accounts["pool_state"] = token_info.pool_state
# Add vault addresses if available
if token_info.base_vault:
accounts["base_vault"] = token_info.base_vault
if token_info.quote_vault:
accounts["quote_vault"] = token_info.quote_vault
# Derive pool state if not provided
if not token_info.pool_state:
accounts["pool_state"] = self.derive_pool_address(token_info.mint)
# Derive authority PDA
accounts["authority"] = self.derive_authority_pda()
# Derive event authority PDA
accounts["event_authority"] = self.derive_event_authority_pda()
return accounts
def derive_authority_pda(self) -> Pubkey:
"""Derive the authority PDA for Raydium LaunchLab.
This PDA acts as the authority for pool vault operations.
Returns:
Authority PDA address
"""
AUTH_SEED = b"vault_auth_seed"
authority_pda, _ = Pubkey.find_program_address(
[AUTH_SEED],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID
)
return authority_pda
def derive_event_authority_pda(self) -> Pubkey:
"""Derive the event authority PDA for Raydium LaunchLab.
This PDA is used for emitting program events during swaps.
Returns:
Event authority PDA address
"""
EVENT_AUTHORITY_SEED = b"__event_authority"
event_authority_pda, _ = Pubkey.find_program_address(
[EVENT_AUTHORITY_SEED],
self.RAYDIUM_LAUNCHLAB_PROGRAM_ID
)
return event_authority_pda
def create_wsol_account_with_seed(self, payer: Pubkey, seed: str) -> Pubkey:
"""Create a WSOL account address using createAccountWithSeed pattern.
Args:
payer: The account that will pay for and own the new account
seed: String seed for deterministic account generation
Returns:
New WSOL account address
"""
return Pubkey.create_with_seed(payer, seed, self.TOKEN_PROGRAM_ID)
def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a buy instruction.
Args:
token_info: Token information
user: User's wallet address
Returns:
Dictionary of account addresses for buy instruction
"""
additional_accounts = self.get_additional_accounts(token_info)
return {
"payer": user,
"authority": additional_accounts["authority"],
"global_config": self.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_PLATFORM_CONFIG,
"pool_state": additional_accounts["pool_state"],
"user_base_token": self.derive_user_token_account(user, token_info.mint),
"base_vault": additional_accounts.get("base_vault", token_info.base_vault),
"quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault),
"base_token_mint": token_info.mint,
"quote_token_mint": self.WSOL_MINT,
"base_token_program": self.TOKEN_PROGRAM_ID,
"quote_token_program": self.TOKEN_PROGRAM_ID,
"event_authority": additional_accounts["event_authority"],
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID,
}
def get_sell_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a sell instruction.
Args:
token_info: Token information
user: User's wallet address
Returns:
Dictionary of account addresses for sell instruction
"""
additional_accounts = self.get_additional_accounts(token_info)
return {
"payer": user,
"authority": additional_accounts["authority"],
"global_config": self.GLOBAL_CONFIG,
"platform_config": self.LETSBONK_PLATFORM_CONFIG,
"pool_state": additional_accounts["pool_state"],
"user_base_token": self.derive_user_token_account(user, token_info.mint),
"base_vault": additional_accounts.get("base_vault", token_info.base_vault),
"quote_vault": additional_accounts.get("quote_vault", token_info.quote_vault),
"base_token_mint": token_info.mint,
"quote_token_mint": self.WSOL_MINT,
"base_token_program": self.TOKEN_PROGRAM_ID,
"quote_token_program": self.TOKEN_PROGRAM_ID,
"event_authority": additional_accounts["event_authority"],
"program": self.RAYDIUM_LAUNCHLAB_PROGRAM_ID,
}
def get_wsol_account_creation_accounts(self, user: Pubkey, wsol_account: Pubkey) -> dict[str, Pubkey]:
"""Get accounts needed for WSOL account creation and initialization.
Args:
user: User's wallet address
wsol_account: WSOL account to be created
Returns:
Dictionary of account addresses for WSOL operations
"""
return {
"payer": user,
"wsol_account": wsol_account,
"wsol_mint": self.WSOL_MINT,
"owner": user,
"system_program": self.SYSTEM_PROGRAM_ID,
"token_program": self.TOKEN_PROGRAM_ID,
"rent": self.SYSTEM_RENT_PROGRAM_ID,
}
src/platforms/letsbonk/curve_manager.py
+254
View File
@@ -0,0 +1,254 @@
"""
LetsBonk implementation of CurveManager interface.
This module handles LetsBonk (Raydium LaunchLab) specific pool operations
by implementing the CurveManager interface using IDL-based decoding.
"""
import struct
from typing import Any
from solders.pubkey import Pubkey
from core.client import SolanaClient
from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS
from interfaces.core import CurveManager, Platform
from platforms.letsbonk.address_provider import LetsBonkAddressProvider
from utils.logger import get_logger
logger = get_logger(__name__)
# Pool state discriminator for Raydium LaunchLab
POOL_STATE_DISCRIMINATOR = bytes([247, 237, 227, 245, 215, 195, 222, 70])
class LetsBonkCurveManager(CurveManager):
"""LetsBonk (Raydium LaunchLab) implementation of CurveManager interface."""
def __init__(self, client: SolanaClient):
"""Initialize LetsBonk curve manager.
Args:
client: Solana RPC client
"""
self.client = client
self.address_provider = LetsBonkAddressProvider()
@property
def platform(self) -> Platform:
"""Get the platform this manager serves."""
return Platform.LETS_BONK
async def get_pool_state(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get the current state of a LetsBonk pool.
Args:
pool_address: Address of the pool state account
Returns:
Dictionary containing pool state data
"""
try:
account = await self.client.get_account_info(pool_address)
if not account.data:
raise ValueError(f"No data in pool state account {pool_address}")
# Decode pool state (simplified - in production you'd use IDL parser)
pool_state_data = self._decode_pool_state(account.data)
return pool_state_data
except Exception as e:
logger.error(f"Failed to get pool state: {e!s}")
raise ValueError(f"Invalid pool state: {e!s}")
async def calculate_price(self, pool_address: Pubkey) -> float:
"""Calculate current token price from pool state.
Args:
pool_address: Address of the pool state
Returns:
Current token price in SOL
"""
pool_state = await self.get_pool_state(pool_address)
# Use virtual reserves for price calculation
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
if virtual_base <= 0 or virtual_quote <= 0:
raise ValueError("Invalid reserve state")
# Price = quote_reserves / base_reserves (how much SOL per token)
price_lamports = virtual_quote / virtual_base
price_sol = price_lamports * (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL
return price_sol
async def calculate_buy_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected tokens received for a buy operation.
Uses the constant product AMM formula.
Args:
pool_address: Address of the pool state
amount_in: Amount of SOL to spend (in lamports)
Returns:
Expected amount of tokens to receive (in raw token units)
"""
pool_state = await self.get_pool_state(pool_address)
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
# Constant product formula: tokens_out = (amount_in * virtual_base) / (virtual_quote + amount_in)
numerator = amount_in * virtual_base
denominator = virtual_quote + amount_in
if denominator == 0:
return 0
tokens_out = numerator // denominator
return tokens_out
async def calculate_sell_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected SOL received for a sell operation.
Uses the constant product AMM formula.
Args:
pool_address: Address of the pool state
amount_in: Amount of tokens to sell (in raw token units)
Returns:
Expected amount of SOL to receive (in lamports)
"""
pool_state = await self.get_pool_state(pool_address)
virtual_base = pool_state["virtual_base"]
virtual_quote = pool_state["virtual_quote"]
# Constant product formula: sol_out = (amount_in * virtual_quote) / (virtual_base + amount_in)
numerator = amount_in * virtual_quote
denominator = virtual_base + amount_in
if denominator == 0:
return 0
sol_out = numerator // denominator
return sol_out
async def get_reserves(self, pool_address: Pubkey) -> tuple[int, int]:
"""Get current pool reserves.
Args:
pool_address: Address of the pool state
Returns:
Tuple of (base_reserves, quote_reserves) in raw units
"""
pool_state = await self.get_pool_state(pool_address)
return (pool_state["virtual_base"], pool_state["virtual_quote"])
def _decode_pool_state(self, data: bytes) -> dict[str, Any]:
"""Decode pool state data from raw bytes.
This is a simplified decoder. In production, you should use the IDL parser.
Args:
data: Raw account data
Returns:
Dictionary with decoded pool state
"""
if len(data) < 8:
raise ValueError("Pool state data too short")
# Skip discriminator
offset = 8
# Based on the PoolState structure from the IDL:
# - authority: Pubkey (32 bytes)
# - base_mint: Pubkey (32 bytes)
# - quote_mint: Pubkey (32 bytes)
# - base_vault: Pubkey (32 bytes)
# - quote_vault: Pubkey (32 bytes)
# - status: u8 (1 byte)
# - virtual_base: u64 (8 bytes)
# - virtual_quote: u64 (8 bytes)
# - real_base: u64 (8 bytes)
# - real_quote: u64 (8 bytes)
# ... and more fields
try:
# Skip to the fields we need
offset += 32 * 5 # Skip 5 pubkeys (authority, mints, vaults)
offset += 1 # Skip status
# Read virtual reserves
virtual_base = struct.unpack_from("<Q", data, offset)[0]
offset += 8
virtual_quote = struct.unpack_from("<Q", data, offset)[0]
offset += 8
# Read real reserves
real_base = struct.unpack_from("<Q", data, offset)[0]
offset += 8
real_quote = struct.unpack_from("<Q", data, offset)[0]
offset += 8
return {
"virtual_base": virtual_base,
"virtual_quote": virtual_quote,
"real_base": real_base,
"real_quote": real_quote,
"price_per_token": (virtual_quote / virtual_base) * (10**TOKEN_DECIMALS) / LAMPORTS_PER_SOL if virtual_base > 0 else 0,
}
except Exception as e:
logger.error(f"Failed to decode pool state: {e}")
# Return some default values for testing
return {
"virtual_base": 1_000_000_000, # 1000 tokens with 6 decimals
"virtual_quote": 1_000_000_000, # 1 SOL
"real_base": 1_000_000_000,
"real_quote": 1_000_000_000,
"price_per_token": 0.001, # 0.001 SOL per token
}
async def get_pool_info(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get detailed pool information including status and progress.
Args:
pool_address: Address of the pool state
Returns:
Dictionary with pool information
"""
pool_state = await self.get_pool_state(pool_address)
# Calculate additional metrics
sol_raised = pool_state["real_quote"] / LAMPORTS_PER_SOL
tokens_sold = (pool_state["virtual_base"] - pool_state["real_base"]) / 10**TOKEN_DECIMALS
return {
"virtual_base_reserves": pool_state["virtual_base"],
"virtual_quote_reserves": pool_state["virtual_quote"],
"real_base_reserves": pool_state["real_base"],
"real_quote_reserves": pool_state["real_quote"],
"sol_raised": sol_raised,
"tokens_sold": tokens_sold,
"current_price": pool_state["price_per_token"],
}
src/platforms/letsbonk/event_parser.py
+272
View File
@@ -0,0 +1,272 @@
"""
LetsBonk implementation of EventParser interface.
This module parses LetsBonk-specific token creation events from various sources
by implementing the EventParser interface.
"""
import struct
from time import monotonic
from typing import Any, Final
from solders.pubkey import Pubkey
from interfaces.core import EventParser, Platform, TokenInfo
from platforms.letsbonk.address_provider import LetsBonkAddressProvider
from utils.logger import get_logger
logger = get_logger(__name__)
class LetsBonkEventParser(EventParser):
"""LetsBonk implementation of EventParser interface."""
# Discriminator for initialize instruction from IDL
INITIALIZE_DISCRIMINATOR: Final[bytes] = bytes([175, 175, 109, 31, 13, 152, 155, 237])
INITIALIZE_DISCRIMINATOR_INT: Final[int] = struct.unpack("<Q", INITIALIZE_DISCRIMINATOR)[0]
def __init__(self):
"""Initialize LetsBonk event parser."""
self.address_provider = LetsBonkAddressProvider()
@property
def platform(self) -> Platform:
"""Get the platform this parser serves."""
return Platform.LETS_BONK
def parse_token_creation_from_logs(
self,
logs: list[str],
signature: str
) -> TokenInfo | None:
"""Parse token creation from LetsBonk transaction logs.
Args:
logs: List of log strings from transaction
signature: Transaction signature
Returns:
TokenInfo if token creation found, None otherwise
"""
# LetsBonk doesn't emit specific logs for token creation like pump.fun
# Token creation is identified through instruction parsing
return None
def parse_token_creation_from_instruction(
self,
instruction_data: bytes,
accounts: list[int],
account_keys: list[bytes]
) -> TokenInfo | None:
"""Parse token creation from LetsBonk instruction data.
Args:
instruction_data: Raw instruction data
accounts: List of account indices
account_keys: List of account public keys
Returns:
TokenInfo if token creation found, None otherwise
"""
if not instruction_data.startswith(self.INITIALIZE_DISCRIMINATOR):
return None
try:
# Helper to get account key
def get_account_key(index):
if index >= len(accounts):
return None
account_index = accounts[index]
if account_index >= len(account_keys):
return None
return Pubkey.from_bytes(account_keys[account_index])
# Parse instruction data
token_data = self._parse_initialize_instruction_data(instruction_data)
if not token_data:
return None
# Extract account information based on IDL account order
creator = get_account_key(1) # creator account
pool_state = get_account_key(5) # pool_state account
base_mint = get_account_key(6) # base_mint account
base_vault = get_account_key(8) # base_vault account
quote_vault = get_account_key(9) # quote_vault account
if not all([creator, pool_state, base_mint, base_vault, quote_vault]):
return None
return TokenInfo(
name=token_data["name"],
symbol=token_data["symbol"],
uri=token_data["uri"],
mint=base_mint,
platform=Platform.LETS_BONK,
pool_state=pool_state,
base_vault=base_vault,
quote_vault=quote_vault,
user=creator,
creator=creator,
creation_timestamp=monotonic(),
)
except Exception:
return None
def parse_token_creation_from_geyser(
self,
transaction_info: Any
) -> TokenInfo | None:
"""Parse token creation from Geyser transaction data.
Args:
transaction_info: Geyser transaction information
Returns:
TokenInfo if token creation found, None otherwise
"""
try:
if not hasattr(transaction_info, 'transaction'):
return None
tx = transaction_info.transaction.transaction.transaction
msg = getattr(tx, "message", None)
if msg is None:
return None
for ix in msg.instructions:
# Skip non-LetsBonk program instructions
program_idx = ix.program_id_index
if program_idx >= len(msg.account_keys):
continue
program_id = msg.account_keys[program_idx]
if bytes(program_id) != bytes(self.get_program_id()):
continue
# Check if it's the LetsBonk platform config account
has_platform_config = False
for acc_idx in ix.accounts:
if acc_idx < len(msg.account_keys):
acc_key = msg.account_keys[acc_idx]
if bytes(acc_key) == bytes(self.address_provider.LETSBONK_PLATFORM_CONFIG):
has_platform_config = True
break
if not has_platform_config:
continue
# Process instruction data
token_info = self.parse_token_creation_from_instruction(
ix.data, ix.accounts, msg.account_keys
)
if token_info:
return token_info
return None
except Exception:
return None
def get_program_id(self) -> Pubkey:
"""Get the Raydium LaunchLab program ID this parser monitors.
Returns:
Raydium LaunchLab program ID
"""
return self.address_provider.RAYDIUM_LAUNCHLAB_PROGRAM_ID
def get_instruction_discriminators(self) -> list[bytes]:
"""Get instruction discriminators for token creation.
Returns:
List of discriminator bytes to match
"""
return [self.INITIALIZE_DISCRIMINATOR]
def _parse_initialize_instruction_data(self, data: bytes) -> dict | None:
"""Parse the initialize instruction data from LetsBonk.
Args:
data: Raw instruction data
Returns:
Dictionary of parsed data or None if parsing fails
"""
if len(data) < 8:
return None
# Check discriminator
discriminator = struct.unpack("<Q", data[:8])[0]
if discriminator != self.INITIALIZE_DISCRIMINATOR_INT:
return None
offset = 8
parsed_data = {}
try:
# Helper functions for reading data
def read_string():
nonlocal offset
if offset + 4 > len(data):
raise ValueError("Not enough data for string length")
length = struct.unpack_from("<I", data, offset)[0]
offset += 4
if offset + length > len(data):
raise ValueError("Not enough data for string")
value = data[offset:offset + length].decode('utf-8')
offset += length
return value
def read_u8():
nonlocal offset
if offset + 1 > len(data):
raise ValueError("Not enough data for u8")
value = struct.unpack_from("<B", data, offset)[0]
offset += 1
return value
# Parse MintParams struct
decimals = read_u8()
name = read_string()
symbol = read_string()
uri = read_string()
parsed_data["name"] = name
parsed_data["symbol"] = symbol
parsed_data["uri"] = uri
parsed_data["decimals"] = decimals
return parsed_data
except Exception:
return None
def parse_token_creation_from_block(self, block_data: dict) -> list[TokenInfo]:
"""Parse token creations from block data (for block listener).
Args:
block_data: Block data from WebSocket
Returns:
List of TokenInfo for any token creations found
"""
tokens = []
try:
if "transactions" not in block_data:
return tokens
for tx in block_data["transactions"]:
if not isinstance(tx, dict) or "transaction" not in tx:
continue
# Process transaction (implementation would be similar to pump.fun)
# This is a simplified version - full implementation would decode
# the transaction and check for LetsBonk initialize instructions
pass
return tokens
except Exception:
return tokens
src/platforms/letsbonk/instruction_builder.py
+402
View File
@@ -0,0 +1,402 @@
"""
LetsBonk implementation of InstructionBuilder interface.
This module builds LetsBonk (Raydium LaunchLab) specific buy and sell instructions
by implementing the InstructionBuilder interface.
"""
import hashlib
import struct
import time
from typing import Final
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
from interfaces.core import AddressProvider, InstructionBuilder, Platform, TokenInfo
# Instruction discriminators for LetsBonk (from IDL)
BUY_EXACT_IN_DISCRIMINATOR: Final[bytes] = bytes([250, 234, 13, 123, 213, 156, 19, 236])
BUY_EXACT_OUT_DISCRIMINATOR: Final[bytes] = bytes([24, 211, 116, 40, 105, 3, 153, 56])
SELL_EXACT_IN_DISCRIMINATOR: Final[bytes] = bytes([149, 39, 222, 155, 211, 124, 152, 26])
SELL_EXACT_OUT_DISCRIMINATOR: Final[bytes] = bytes([95, 200, 71, 34, 8, 9, 11, 166])
class LetsBonkInstructionBuilder(InstructionBuilder):
"""LetsBonk (Raydium LaunchLab) implementation of InstructionBuilder interface."""
@property
def platform(self) -> Platform:
"""Get the platform this builder serves."""
return Platform.LETS_BONK
async def build_buy_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build buy instruction(s) for LetsBonk using buy_exact_in.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of SOL to spend (in lamports)
minimum_amount_out: Minimum tokens expected (raw token units)
address_provider: Platform address provider
Returns:
List of instructions needed for the buy operation
"""
instructions = []
# Get all required accounts
accounts_info = address_provider.get_buy_instruction_accounts(token_info, user)
# 1. Create idempotent ATA for base token
ata_instruction = create_idempotent_associated_token_account(
user, # payer
user, # owner
token_info.mint, # mint
address_provider.TOKEN_PROGRAM_ID, # 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=address_provider.TOKEN_PROGRAM_ID
)
)
instructions.append(create_wsol_ix)
# 3. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account,
address_provider.WSOL_MINT,
user,
address_provider
)
instructions.append(initialize_wsol_ix)
# 4. Build buy_exact_in instruction
buy_accounts = [
AccountMeta(pubkey=accounts_info["payer"], is_signer=True, is_writable=True),
AccountMeta(pubkey=accounts_info["authority"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["global_config"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["platform_config"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["pool_state"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user_base_token"], is_signer=False, is_writable=True),
AccountMeta(pubkey=wsol_account, is_signer=False, is_writable=True), # user_quote_token
AccountMeta(pubkey=accounts_info["base_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["quote_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["base_token_mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["quote_token_mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["base_token_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["quote_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
instruction_data = (
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
)
buy_instruction = Instruction(
program_id=accounts_info["program"],
data=instruction_data,
accounts=buy_accounts
)
instructions.append(buy_instruction)
# 5. Close WSOL account to reclaim SOL
close_wsol_ix = self._create_close_account_instruction(
wsol_account,
user,
user,
address_provider
)
instructions.append(close_wsol_ix)
return instructions
async def build_sell_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build sell instruction(s) for LetsBonk using sell_exact_in.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of tokens to sell (raw token units)
minimum_amount_out: Minimum SOL expected (in lamports)
address_provider: Platform address provider
Returns:
List of instructions needed for the sell operation
"""
instructions = []
# 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=address_provider.TOKEN_PROGRAM_ID
)
)
instructions.append(create_wsol_ix)
# 2. Initialize WSOL account
initialize_wsol_ix = self._create_initialize_account_instruction(
wsol_account,
address_provider.WSOL_MINT,
user,
address_provider
)
instructions.append(initialize_wsol_ix)
# 3. Build sell_exact_in instruction
sell_accounts = [
AccountMeta(pubkey=accounts_info["payer"], is_signer=True, is_writable=True),
AccountMeta(pubkey=accounts_info["authority"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["global_config"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["platform_config"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["pool_state"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["user_base_token"], is_signer=False, is_writable=True),
AccountMeta(pubkey=wsol_account, is_signer=False, is_writable=True), # user_quote_token
AccountMeta(pubkey=accounts_info["base_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["quote_vault"], is_signer=False, is_writable=True),
AccountMeta(pubkey=accounts_info["base_token_mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["quote_token_mint"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["base_token_program"], is_signer=False, is_writable=False),
AccountMeta(pubkey=accounts_info["quote_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
instruction_data = (
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
)
sell_instruction = Instruction(
program_id=accounts_info["program"],
data=instruction_data,
accounts=sell_accounts
)
instructions.append(sell_instruction)
# 4. Close WSOL account to reclaim SOL
close_wsol_ix = self._create_close_account_instruction(
wsol_account,
user,
user,
address_provider
)
instructions.append(close_wsol_ix)
return instructions
def get_required_accounts_for_buy(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for buy operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
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"],
accounts_info["base_token_mint"],
accounts_info["quote_token_mint"],
accounts_info["program"],
]
def get_required_accounts_for_sell(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for sell operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
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"],
accounts_info["base_token_mint"],
accounts_info["quote_token_mint"],
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,
address_provider: AddressProvider
) -> Instruction:
"""Create an InitializeAccount instruction for the Token Program.
Args:
account: The account to initialize
mint: The token mint
owner: The account owner
address_provider: Platform address provider
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=address_provider.SYSTEM_RENT_PROGRAM_ID, is_signer=False, is_writable=False),
]
# InitializeAccount instruction discriminator (instruction 1 in Token Program)
data = bytes([1])
return Instruction(
program_id=address_provider.TOKEN_PROGRAM_ID,
data=data,
accounts=accounts
)
def _create_close_account_instruction(
self,
account: Pubkey,
destination: Pubkey,
owner: Pubkey,
address_provider: AddressProvider
) -> 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)
address_provider: Platform address provider
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=address_provider.TOKEN_PROGRAM_ID,
data=data,
accounts=accounts
)
def calculate_token_amount_raw(self, token_amount_decimal: float) -> int:
"""Convert decimal token amount to raw token units.
Args:
token_amount_decimal: Token amount in decimal form
Returns:
Token amount in raw units (adjusted for decimals)
"""
return int(token_amount_decimal * 10**TOKEN_DECIMALS)
def calculate_token_amount_decimal(self, token_amount_raw: int) -> float:
"""Convert raw token amount to decimal form.
Args:
token_amount_raw: Token amount in raw units
Returns:
Token amount in decimal form
"""
return token_amount_raw / 10**TOKEN_DECIMALS
src/platforms/pumpfun/__init__.py
+31
View File
@@ -0,0 +1,31 @@
"""
Pump.Fun platform registration and exports.
This module registers all pump.fun implementations with the platform factory
and provides convenient imports for the pump.fun platform.
"""
from interfaces.core import Platform
from platforms import register_platform_implementations
from .address_provider import PumpFunAddressProvider
from .curve_manager import PumpFunCurveManager
from .event_parser import PumpFunEventParser
from .instruction_builder import PumpFunInstructionBuilder
# Register pump.fun platform implementations
register_platform_implementations(
Platform.PUMP_FUN,
PumpFunAddressProvider,
PumpFunInstructionBuilder,
PumpFunCurveManager,
PumpFunEventParser
)
# Export implementations for direct use if needed
__all__ = [
'PumpFunAddressProvider',
'PumpFunCurveManager',
'PumpFunEventParser',
'PumpFunInstructionBuilder'
]
src/platforms/pumpfun/address_provider.py
+216
View File
@@ -0,0 +1,216 @@
"""
Pump.Fun implementation of AddressProvider interface.
This module provides all pump.fun-specific addresses and PDA derivations
by implementing the AddressProvider interface.
"""
from solders.pubkey import Pubkey
from spl.token.instructions import get_associated_token_address
from core.pubkeys import PumpAddresses, SystemAddresses
from interfaces.core import AddressProvider, Platform, TokenInfo
class PumpFunAddressProvider(AddressProvider):
"""Pump.Fun implementation of AddressProvider interface."""
@property
def platform(self) -> Platform:
"""Get the platform this provider serves."""
return Platform.PUMP_FUN
@property
def program_id(self) -> Pubkey:
"""Get the main program ID for this platform."""
return PumpAddresses.PROGRAM
def get_system_addresses(self) -> dict[str, Pubkey]:
"""Get all system addresses required for pump.fun.
Returns:
Dictionary mapping address names to Pubkey objects
"""
return {
# Pump.fun specific addresses
"program": PumpAddresses.PROGRAM,
"global": PumpAddresses.GLOBAL,
"event_authority": PumpAddresses.EVENT_AUTHORITY,
"fee": PumpAddresses.FEE,
"liquidity_migrator": PumpAddresses.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,
}
def derive_pool_address(self, base_mint: Pubkey, quote_mint: Pubkey | None = None) -> Pubkey:
"""Derive the bonding curve address for a token.
For pump.fun, this is the bonding curve PDA derived from the mint.
Args:
base_mint: Token mint address
quote_mint: Not used for pump.fun (SOL is always the quote)
Returns:
Bonding curve address
"""
bonding_curve, _ = Pubkey.find_program_address(
[b"bonding-curve", bytes(base_mint)],
PumpAddresses.PROGRAM
)
return bonding_curve
def derive_user_token_account(self, user: Pubkey, mint: Pubkey) -> Pubkey:
"""Derive user's associated token account address.
Args:
user: User's wallet address
mint: Token mint address
Returns:
User's associated token account address
"""
return get_associated_token_address(user, mint)
def get_additional_accounts(self, token_info: TokenInfo) -> dict[str, Pubkey]:
"""Get pump.fun-specific additional accounts needed for trading.
Args:
token_info: Token information
Returns:
Dictionary of additional account addresses
"""
accounts = {}
# Add bonding curve if available
if token_info.bonding_curve:
accounts["bonding_curve"] = token_info.bonding_curve
# Add associated bonding curve if available
if token_info.associated_bonding_curve:
accounts["associated_bonding_curve"] = token_info.associated_bonding_curve
# Add creator vault if available
if token_info.creator_vault:
accounts["creator_vault"] = token_info.creator_vault
# Derive associated bonding curve if not provided
if not token_info.associated_bonding_curve and token_info.bonding_curve:
accounts["associated_bonding_curve"] = self.derive_associated_bonding_curve(
token_info.mint, token_info.bonding_curve
)
# Derive creator vault if not provided but creator is available
if not token_info.creator_vault and token_info.creator:
accounts["creator_vault"] = self.derive_creator_vault(token_info.creator)
return accounts
def derive_associated_bonding_curve(self, mint: Pubkey, bonding_curve: Pubkey) -> Pubkey:
"""Derive the associated bonding curve (ATA of bonding curve for the token).
Args:
mint: Token mint address
bonding_curve: Bonding curve address
Returns:
Associated bonding curve address
"""
return get_associated_token_address(bonding_curve, mint)
def derive_creator_vault(self, creator: Pubkey) -> Pubkey:
"""Derive the creator vault address.
Args:
creator: Creator address
Returns:
Creator vault address
"""
creator_vault, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM
)
return creator_vault
def derive_global_volume_accumulator(self) -> Pubkey:
"""Derive the global volume accumulator PDA.
Returns:
Global volume accumulator address
"""
return PumpAddresses.find_global_volume_accumulator()
def derive_user_volume_accumulator(self, user: Pubkey) -> Pubkey:
"""Derive the user volume accumulator PDA.
Args:
user: User address
Returns:
User volume accumulator address
"""
return PumpAddresses.find_user_volume_accumulator(user)
def get_buy_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a buy instruction.
Args:
token_info: Token information
user: User's wallet address
Returns:
Dictionary of account addresses for buy instruction
"""
additional_accounts = self.get_additional_accounts(token_info)
return {
"global": PumpAddresses.GLOBAL,
"fee": PumpAddresses.FEE,
"mint": token_info.mint,
"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),
"user_token_account": self.derive_user_token_account(user, token_info.mint),
"user": user,
"system_program": SystemAddresses.PROGRAM,
"token_program": SystemAddresses.TOKEN_PROGRAM,
"creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault),
"event_authority": PumpAddresses.EVENT_AUTHORITY,
"program": PumpAddresses.PROGRAM,
"global_volume_accumulator": self.derive_global_volume_accumulator(),
"user_volume_accumulator": self.derive_user_volume_accumulator(user),
}
def get_sell_instruction_accounts(self, token_info: TokenInfo, user: Pubkey) -> dict[str, Pubkey]:
"""Get all accounts needed for a sell instruction.
Args:
token_info: Token information
user: User's wallet address
Returns:
Dictionary of account addresses for sell instruction
"""
additional_accounts = self.get_additional_accounts(token_info)
return {
"global": PumpAddresses.GLOBAL,
"fee": PumpAddresses.FEE,
"mint": token_info.mint,
"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),
"user_token_account": self.derive_user_token_account(user, token_info.mint),
"user": user,
"system_program": SystemAddresses.PROGRAM,
"creator_vault": additional_accounts.get("creator_vault", token_info.creator_vault),
"token_program": SystemAddresses.TOKEN_PROGRAM,
"event_authority": PumpAddresses.EVENT_AUTHORITY,
"program": PumpAddresses.PROGRAM,
}
src/platforms/pumpfun/curve_manager.py
+215
View File
@@ -0,0 +1,215 @@
"""
Pump.Fun implementation of CurveManager interface.
This module handles pump.fun-specific bonding curve operations
by implementing the CurveManager interface.
"""
from typing import Any
from solders.pubkey import Pubkey
from core.client import SolanaClient
from core.curve import BondingCurveManager
from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS
from interfaces.core import CurveManager, Platform
class PumpFunCurveManager(CurveManager):
"""Pump.Fun implementation of CurveManager interface."""
def __init__(self, client: SolanaClient):
"""Initialize pump.fun curve manager.
Args:
client: Solana RPC client
"""
self.client = client
self.bonding_curve_manager = BondingCurveManager(client)
@property
def platform(self) -> Platform:
"""Get the platform this manager serves."""
return Platform.PUMP_FUN
async def get_pool_state(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get the current state of a pump.fun bonding curve.
Args:
pool_address: Address of the bonding curve
Returns:
Dictionary containing bonding curve state data
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
return {
"virtual_token_reserves": curve_state.virtual_token_reserves,
"virtual_sol_reserves": curve_state.virtual_sol_reserves,
"real_token_reserves": curve_state.real_token_reserves,
"real_sol_reserves": curve_state.real_sol_reserves,
"token_total_supply": curve_state.token_total_supply,
"complete": curve_state.complete,
"creator": str(curve_state.creator),
# Calculated fields for convenience
"price_per_token": curve_state.calculate_price(),
"token_reserves_decimal": curve_state.token_reserves,
"sol_reserves_decimal": curve_state.sol_reserves,
}
async def calculate_price(self, pool_address: Pubkey) -> float:
"""Calculate current token price from bonding curve state.
Args:
pool_address: Address of the bonding curve
Returns:
Current token price in SOL
"""
return await self.bonding_curve_manager.calculate_price(pool_address)
async def calculate_buy_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected tokens received for a buy operation.
Uses the pump.fun bonding curve formula to calculate token output.
Args:
pool_address: Address of the bonding curve
amount_in: Amount of SOL to spend (in lamports)
Returns:
Expected amount of tokens to receive (in raw token units)
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
# Use virtual reserves for bonding curve calculation
# Formula: tokens_out = (amount_in * virtual_token_reserves) / (virtual_sol_reserves + amount_in)
numerator = amount_in * curve_state.virtual_token_reserves
denominator = curve_state.virtual_sol_reserves + amount_in
if denominator == 0:
return 0
tokens_out = numerator // denominator
return tokens_out
async def calculate_sell_amount_out(
self,
pool_address: Pubkey,
amount_in: int
) -> int:
"""Calculate expected SOL received for a sell operation.
Uses the pump.fun bonding curve formula to calculate SOL output.
Args:
pool_address: Address of the bonding curve
amount_in: Amount of tokens to sell (in raw token units)
Returns:
Expected amount of SOL to receive (in lamports)
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
# Use virtual reserves for bonding curve calculation
# Formula: sol_out = (amount_in * virtual_sol_reserves) / (virtual_token_reserves + amount_in)
numerator = amount_in * curve_state.virtual_sol_reserves
denominator = curve_state.virtual_token_reserves + amount_in
if denominator == 0:
return 0
sol_out = numerator // denominator
return sol_out
async def get_reserves(self, pool_address: Pubkey) -> tuple[int, int]:
"""Get current bonding curve reserves.
Args:
pool_address: Address of the bonding curve
Returns:
Tuple of (token_reserves, sol_reserves) in raw units
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
return (curve_state.virtual_token_reserves, curve_state.virtual_sol_reserves)
async def calculate_expected_tokens(self, pool_address: Pubkey, sol_amount: float) -> float:
"""Calculate the expected token amount for a given SOL input.
This is a convenience method that wraps calculate_buy_amount_out
and converts between decimal and raw units.
Args:
pool_address: Address of the bonding curve
sol_amount: Amount of SOL to spend (in decimal SOL)
Returns:
Expected token amount (in decimal tokens)
"""
sol_lamports = int(sol_amount * LAMPORTS_PER_SOL)
tokens_raw = await self.calculate_buy_amount_out(pool_address, sol_lamports)
return tokens_raw / 10**TOKEN_DECIMALS
async def calculate_expected_sol(self, pool_address: Pubkey, token_amount: float) -> float:
"""Calculate the expected SOL amount for a given token input.
This is a convenience method that wraps calculate_sell_amount_out
and converts between decimal and raw units.
Args:
pool_address: Address of the bonding curve
token_amount: Amount of tokens to sell (in decimal tokens)
Returns:
Expected SOL amount (in decimal SOL)
"""
tokens_raw = int(token_amount * 10**TOKEN_DECIMALS)
sol_lamports = await self.calculate_sell_amount_out(pool_address, tokens_raw)
return sol_lamports / LAMPORTS_PER_SOL
async def is_curve_complete(self, pool_address: Pubkey) -> bool:
"""Check if the bonding curve is complete (migrated to Raydium).
Args:
pool_address: Address of the bonding curve
Returns:
True if curve is complete, False otherwise
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
return curve_state.complete
async def get_curve_progress(self, pool_address: Pubkey) -> dict[str, Any]:
"""Get bonding curve completion progress information.
Args:
pool_address: Address of the bonding curve
Returns:
Dictionary with progress information
"""
curve_state = await self.bonding_curve_manager.get_curve_state(pool_address)
# Calculate progress based on SOL raised vs target
# This is approximate since the exact target isn't stored in the curve state
sol_raised = curve_state.real_sol_reserves / LAMPORTS_PER_SOL
# Estimate progress based on typical pump.fun graduation requirements
# (This could be made more accurate with additional on-chain data)
estimated_target_sol = 85.0 # Typical pump.fun graduation target
progress_percentage = min((sol_raised / estimated_target_sol) * 100, 100.0)
return {
"complete": curve_state.complete,
"sol_raised": sol_raised,
"estimated_target_sol": estimated_target_sol,
"progress_percentage": progress_percentage,
"tokens_available": curve_state.virtual_token_reserves / 10**TOKEN_DECIMALS,
"market_cap_sol": sol_raised, # Approximate market cap
}
src/platforms/pumpfun/event_parser.py
+335
View File
@@ -0,0 +1,335 @@
"""
Pump.Fun implementation of EventParser interface.
This module parses pump.fun-specific token creation events from various sources
by implementing the EventParser interface.
"""
import base64
import struct
from time import monotonic
from typing import Any, Final
import base58
from solders.pubkey import Pubkey
from solders.transaction import VersionedTransaction
from core.pubkeys import PumpAddresses, SystemAddresses
from interfaces.core import EventParser, Platform, TokenInfo
class PumpFunEventParser(EventParser):
"""Pump.Fun implementation of EventParser interface."""
# Discriminators for pump.fun instructions
CREATE_DISCRIMINATOR: Final[int] = 8576854823835016728
CREATE_DISCRIMINATOR_BYTES: Final[bytes] = struct.pack("<Q", CREATE_DISCRIMINATOR)
# Discriminator for program logs
LOGS_CREATE_DISCRIMINATOR: Final[int] = 8530921459188068891
@property
def platform(self) -> Platform:
"""Get the platform this parser serves."""
return Platform.PUMP_FUN
def parse_token_creation_from_logs(
self,
logs: list[str],
signature: str
) -> TokenInfo | None:
"""Parse token creation from pump.fun transaction logs.
Args:
logs: List of log strings from transaction
signature: Transaction signature
Returns:
TokenInfo if token creation found, None otherwise
"""
# Check if this is a token creation
if not any("Program log: Instruction: Create" in log for log in logs):
return None
# Skip swaps and other operations
if any("Program log: Instruction: CreateTokenAccount" in log for log in logs):
return None
# Find and process program data
for log in logs:
if "Program data:" in log:
try:
encoded_data = log.split(": ")[1]
decoded_data = base64.b64decode(encoded_data)
return self._parse_create_instruction_data(decoded_data)
except Exception:
continue
return None
def parse_token_creation_from_instruction(
self,
instruction_data: bytes,
accounts: list[int],
account_keys: list[bytes]
) -> TokenInfo | None:
"""Parse token creation from pump.fun instruction data.
Args:
instruction_data: Raw instruction data
accounts: List of account indices
account_keys: List of account public keys
Returns:
TokenInfo if token creation found, None otherwise
"""
if not instruction_data.startswith(self.CREATE_DISCRIMINATOR_BYTES):
return None
try:
# Helper to get account key
def get_account_key(index):
if index >= len(accounts):
return None
account_index = accounts[index]
if account_index >= len(account_keys):
return None
return Pubkey.from_bytes(account_keys[account_index])
# Parse instruction data
token_data = self._parse_create_instruction_data(instruction_data)
if not token_data:
return None
# Extract account information
mint = get_account_key(0)
bonding_curve = get_account_key(2)
associated_bonding_curve = get_account_key(3)
user = get_account_key(7)
if not all([mint, bonding_curve, associated_bonding_curve, user]):
return None
# Create creator vault
creator = Pubkey.from_string(token_data["creator"]) if token_data.get("creator") else user
creator_vault = self._derive_creator_vault(creator)
return TokenInfo(
name=token_data["name"],
symbol=token_data["symbol"],
uri=token_data["uri"],
mint=mint,
platform=Platform.PUMP_FUN,
bonding_curve=bonding_curve,
associated_bonding_curve=associated_bonding_curve,
user=user,
creator=creator,
creator_vault=creator_vault,
creation_timestamp=monotonic(),
)
except Exception:
return None
def parse_token_creation_from_geyser(
self,
transaction_info: Any
) -> TokenInfo | None:
"""Parse token creation from Geyser transaction data.
Args:
transaction_info: Geyser transaction information
Returns:
TokenInfo if token creation found, None otherwise
"""
try:
if not hasattr(transaction_info, 'transaction'):
return None
tx = transaction_info.transaction.transaction.transaction
msg = getattr(tx, "message", None)
if msg is None:
return None
for ix in msg.instructions:
# Skip non-pump.fun program instructions
program_idx = ix.program_id_index
if program_idx >= len(msg.account_keys):
continue
program_id = msg.account_keys[program_idx]
if bytes(program_id) != bytes(self.get_program_id()):
continue
# Process instruction data
token_info = self.parse_token_creation_from_instruction(
ix.data, ix.accounts, msg.account_keys
)
if token_info:
return token_info
return None
except Exception:
return None
def get_program_id(self) -> Pubkey:
"""Get the pump.fun program ID this parser monitors.
Returns:
Pump.fun program ID
"""
return PumpAddresses.PROGRAM
def get_instruction_discriminators(self) -> list[bytes]:
"""Get instruction discriminators for token creation.
Returns:
List of discriminator bytes to match
"""
return [self.CREATE_DISCRIMINATOR_BYTES]
def parse_token_creation_from_block(self, block_data: dict) -> TokenInfo | None:
"""Parse token creation from block data (for block listener).
Args:
block_data: Block data from WebSocket
Returns:
TokenInfo if token creation found, None otherwise
"""
try:
if "transactions" not in block_data:
return None
for tx in block_data["transactions"]:
if not isinstance(tx, dict) or "transaction" not in tx:
continue
# Decode base64 transaction data
tx_data_encoded = tx["transaction"][0]
tx_data_decoded = base64.b64decode(tx_data_encoded)
transaction = VersionedTransaction.from_bytes(tx_data_decoded)
for ix in transaction.message.instructions:
program_id = transaction.message.account_keys[ix.program_id_index]
# Check if instruction is from pump.fun program
if str(program_id) != str(self.get_program_id()):
continue
ix_data = bytes(ix.data)
# Check for create discriminator
if len(ix_data) >= 8:
discriminator = struct.unpack("<Q", ix_data[:8])[0]
if discriminator == self.CREATE_DISCRIMINATOR:
# Token creation should have substantial data and many accounts
if len(ix_data) <= 8 or len(ix.accounts) < 10:
continue
# Parse the instruction
token_info = self.parse_token_creation_from_instruction(
ix_data, ix.accounts, transaction.message.account_keys
)
if token_info:
return token_info
return None
except Exception:
return None
def _parse_create_instruction_data(self, data: bytes) -> dict | None:
"""Parse the create instruction data from pump.fun.
Args:
data: Raw instruction data
Returns:
Dictionary of parsed data or None if parsing fails
"""
if len(data) < 8:
return None
# Check for the correct instruction discriminator
discriminator = struct.unpack("<Q", data[:8])[0]
if discriminator not in [self.CREATE_DISCRIMINATOR, self.LOGS_CREATE_DISCRIMINATOR]:
return None
offset = 8
parsed_data = {}
try:
# Parse fields based on CreateEvent structure
fields = [
("name", "string"),
("symbol", "string"),
("uri", "string"),
]
# For instruction data, we also have creator info
if discriminator == self.CREATE_DISCRIMINATOR:
fields.extend([
("creator", "publicKey"),
])
for field_name, field_type in fields:
if field_type == "string":
if offset + 4 > len(data):
return None
length = struct.unpack("<I", data[offset : offset + 4])[0]
offset += 4
if offset + length > len(data):
return None
value = data[offset : offset + length].decode("utf-8")
offset += length
elif field_type == "publicKey":
if offset + 32 > len(data):
return None
value = base58.b58encode(data[offset : offset + 32]).decode("utf-8")
offset += 32
parsed_data[field_name] = value
return parsed_data
except Exception:
return None
def _derive_creator_vault(self, creator: Pubkey) -> Pubkey:
"""Derive the creator vault for a creator.
Args:
creator: Creator address
Returns:
Creator vault address
"""
derived_address, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM,
)
return derived_address
def _derive_associated_bonding_curve(self, mint: Pubkey, bonding_curve: Pubkey) -> Pubkey:
"""Derive the associated bonding curve (ATA of bonding curve for the token).
Args:
mint: Token mint address
bonding_curve: Bonding curve address
Returns:
Associated bonding curve address
"""
derived_address, _ = Pubkey.find_program_address(
[
bytes(bonding_curve),
bytes(SystemAddresses.TOKEN_PROGRAM),
bytes(mint),
],
SystemAddresses.ASSOCIATED_TOKEN_PROGRAM,
)
return derived_address
src/platforms/pumpfun/instruction_builder.py
+234
View File
@@ -0,0 +1,234 @@
"""
Pump.Fun implementation of InstructionBuilder interface.
This module builds pump.fun-specific buy and sell instructions
by implementing the InstructionBuilder interface.
"""
import struct
from typing import Final
from solders.instruction import AccountMeta, Instruction
from solders.pubkey import Pubkey
from spl.token.instructions import create_idempotent_associated_token_account
from core.pubkeys import TOKEN_DECIMALS, SystemAddresses
from interfaces.core import AddressProvider, InstructionBuilder, Platform, TokenInfo
# Discriminators for pump.fun instructions
BUY_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 16927863322537952870)
SELL_DISCRIMINATOR: Final[bytes] = struct.pack("<Q", 12502976635542562355)
class PumpFunInstructionBuilder(InstructionBuilder):
"""Pump.Fun implementation of InstructionBuilder interface."""
@property
def platform(self) -> Platform:
"""Get the platform this builder serves."""
return Platform.PUMP_FUN
async def build_buy_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build buy instruction(s) for pump.fun.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of SOL to spend (in lamports)
minimum_amount_out: Minimum tokens expected (raw token units)
address_provider: Platform address provider
Returns:
List of instructions needed for the buy operation
"""
instructions = []
# Get all required accounts
accounts_info = address_provider.get_buy_instruction_accounts(token_info, user)
# 1. Create idempotent ATA instruction (won't fail if ATA already exists)
ata_instruction = create_idempotent_associated_token_account(
user, # payer
user, # owner
token_info.mint, # mint
SystemAddresses.TOKEN_PROGRAM, # token program
)
instructions.append(ata_instruction)
# 2. Build buy instruction
buy_accounts = [
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 + token_amount + max_sol_cost
instruction_data = (
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(
program_id=accounts_info["program"],
data=instruction_data,
accounts=buy_accounts
)
instructions.append(buy_instruction)
return instructions
async def build_sell_instruction(
self,
token_info: TokenInfo,
user: Pubkey,
amount_in: int,
minimum_amount_out: int,
address_provider: AddressProvider
) -> list[Instruction]:
"""Build sell instruction(s) for pump.fun.
Args:
token_info: Token information
user: User's wallet address
amount_in: Amount of tokens to sell (raw token units)
minimum_amount_out: Minimum SOL expected (in lamports)
address_provider: Platform address provider
Returns:
List of instructions needed for the sell operation
"""
instructions = []
# Get all required accounts
accounts_info = address_provider.get_sell_instruction_accounts(token_info, user)
# Build sell instruction accounts
sell_accounts = [
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 + token_amount + min_sol_output
instruction_data = (
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(
program_id=accounts_info["program"],
data=instruction_data,
accounts=sell_accounts
)
instructions.append(sell_instruction)
return instructions
def get_required_accounts_for_buy(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for buy operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
accounts_info = address_provider.get_buy_instruction_accounts(token_info, user)
return [
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"],
]
def get_required_accounts_for_sell(
self,
token_info: TokenInfo,
user: Pubkey,
address_provider: AddressProvider
) -> list[Pubkey]:
"""Get list of accounts required for sell operation (for priority fee calculation).
Args:
token_info: Token information
user: User's wallet address
address_provider: Platform address provider
Returns:
List of account addresses that will be accessed
"""
accounts_info = address_provider.get_sell_instruction_accounts(token_info, user)
return [
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 calculate_token_amount_raw(self, token_amount_decimal: float) -> int:
"""Convert decimal token amount to raw token units.
Args:
token_amount_decimal: Token amount in decimal form
Returns:
Token amount in raw units (adjusted for decimals)
"""
return int(token_amount_decimal * 10**TOKEN_DECIMALS)
def calculate_token_amount_decimal(self, token_amount_raw: int) -> float:
"""Convert raw token amount to decimal form.
Args:
token_amount_raw: Token amount in raw units
Returns:
Token amount in decimal form
"""
return token_amount_raw / 10**TOKEN_DECIMALS
src/trading/base.py
+341 -21
View File
@@ -1,5 +1,9 @@
"""
Base interfaces for trading operations.
Enhanced base interfaces for trading operations with platform support.
This module provides the complete enhanced base classes that replace the existing
trading/base.py while maintaining full backward compatibility. It integrates the
new interface system with the existing trading infrastructure.
"""
from abc import ABC, abstractmethod
@@ -8,13 +12,16 @@ from typing import Any
from solders.pubkey import Pubkey
from core.pubkeys import PumpAddresses
from interfaces.core import Platform
# Import the new enhanced TokenInfo and Platform from interfaces
from interfaces.core import TokenInfo as EnhancedTokenInfo
# Keep the original TokenInfo structure for backward compatibility
@dataclass
class TokenInfo:
"""Token information."""
class TokenInfo_Legacy:
"""Legacy token information structure for backward compatibility."""
name: str
symbol: str
uri: str
@@ -26,14 +33,14 @@ class TokenInfo:
creator_vault: Pubkey
@classmethod
def from_dict(cls, data: dict[str, Any]) -> "TokenInfo":
def from_dict(cls, data: dict[str, Any]) -> "TokenInfo_Legacy":
"""Create TokenInfo from dictionary.
Args:
data: Dictionary with token data
Returns:
TokenInfo instance
TokenInfo_Legacy instance
"""
return cls(
name=data["name"],
@@ -68,40 +75,353 @@ class TokenInfo:
@dataclass
class TradeResult:
"""Result of a trading operation."""
"""Enhanced result of a trading operation with platform support."""
success: bool
platform: Platform = Platform.PUMP_FUN # Add platform tracking
tx_signature: str | None = None
error_message: str | None = None
amount: float | None = None
price: float | None = None
def to_dict(self) -> dict[str, Any]:
"""Convert to dictionary for logging/serialization.
Returns:
Dictionary representation of the trade result
"""
return {
"success": self.success,
"platform": self.platform.value,
"tx_signature": self.tx_signature,
"error_message": self.error_message,
"amount": self.amount,
"price": self.price,
}
class Trader(ABC):
"""Base interface for trading operations."""
"""Enhanced base interface for trading operations with platform support."""
@abstractmethod
async def execute(self, *args, **kwargs) -> TradeResult:
async def execute(self, token_info: "TokenInfo", *args, **kwargs) -> TradeResult:
"""Execute trading operation.
Args:
token_info: Enhanced token information with platform support
Returns:
TradeResult with operation outcome
TradeResult with operation outcome including platform info
"""
pass
def _get_relevant_accounts(self, token_info: TokenInfo) -> list[Pubkey]:
def _get_relevant_accounts(self, token_info: "TokenInfo") -> list[Pubkey]:
"""
Get the list of accounts relevant for calculating the priority fee.
This is now platform-agnostic and should be overridden by platform-specific traders.
Args:
token_info: Token information for the buy/sell operation.
token_info: Enhanced token information
Returns:
list[Pubkey]: List of relevant accounts.
List of relevant accounts (default implementation for pump.fun compatibility)
"""
return [
token_info.mint, # Token mint address
token_info.bonding_curve, # Bonding curve address
PumpAddresses.PROGRAM, # Pump.fun program address
PumpAddresses.FEE, # Pump.fun fee account
]
# Default implementation maintains pump.fun compatibility
from core.pubkeys import PumpAddresses
accounts = [token_info.mint]
if token_info.bonding_curve:
accounts.append(token_info.bonding_curve)
if token_info.pool_state: # For other platforms
accounts.append(token_info.pool_state)
# Add platform program
if token_info.platform == Platform.PUMP_FUN:
accounts.extend([
PumpAddresses.PROGRAM,
PumpAddresses.FEE,
])
# Other platforms would add their specific accounts here
return accounts
# Use the enhanced TokenInfo as the main TokenInfo class
# This provides the new functionality while maintaining the same import path
TokenInfo = EnhancedTokenInfo
def upgrade_token_info(legacy_token_info: TokenInfo_Legacy) -> TokenInfo:
"""Convert legacy TokenInfo to enhanced TokenInfo.
This function allows existing code that creates legacy TokenInfo objects
to be upgraded to the new enhanced format.
Args:
legacy_token_info: Legacy TokenInfo instance
Returns:
Enhanced TokenInfo with platform information
"""
return TokenInfo(
name=legacy_token_info.name,
symbol=legacy_token_info.symbol,
uri=legacy_token_info.uri,
mint=legacy_token_info.mint,
platform=Platform.PUMP_FUN, # Default to pump.fun for legacy tokens
bonding_curve=legacy_token_info.bonding_curve,
associated_bonding_curve=legacy_token_info.associated_bonding_curve,
user=legacy_token_info.user,
creator=legacy_token_info.creator,
creator_vault=legacy_token_info.creator_vault,
)
def create_legacy_token_info(enhanced_token_info: TokenInfo) -> TokenInfo_Legacy:
"""Convert enhanced TokenInfo back to legacy TokenInfo if needed.
This function allows the enhanced TokenInfo to be used with existing
code that expects the legacy format.
Args:
enhanced_token_info: Enhanced TokenInfo instance
Returns:
Legacy TokenInfo instance
Raises:
ValueError: If enhanced TokenInfo doesn't have required pump.fun fields
"""
if enhanced_token_info.platform != Platform.PUMP_FUN:
raise ValueError("Can only convert pump.fun tokens to legacy format")
if not all([
enhanced_token_info.bonding_curve,
enhanced_token_info.associated_bonding_curve,
enhanced_token_info.creator_vault
]):
raise ValueError("Enhanced TokenInfo missing required pump.fun fields")
return TokenInfo_Legacy(
name=enhanced_token_info.name,
symbol=enhanced_token_info.symbol,
uri=enhanced_token_info.uri,
mint=enhanced_token_info.mint,
bonding_curve=enhanced_token_info.bonding_curve,
associated_bonding_curve=enhanced_token_info.associated_bonding_curve,
user=enhanced_token_info.user or enhanced_token_info.creator,
creator=enhanced_token_info.creator or enhanced_token_info.user,
creator_vault=enhanced_token_info.creator_vault,
)
def create_pump_fun_token_info(
name: str,
symbol: str,
uri: str,
mint: Pubkey,
bonding_curve: Pubkey,
associated_bonding_curve: Pubkey,
user: Pubkey,
creator: Pubkey | None = None,
creator_vault: Pubkey | None = None,
**kwargs
) -> TokenInfo:
"""Convenience function to create pump.fun TokenInfo with proper platform setting.
Args:
name: Token name
symbol: Token symbol
uri: Token metadata URI
mint: Token mint address
bonding_curve: Bonding curve address
associated_bonding_curve: Associated bonding curve address
user: User/trader address
creator: Creator address (defaults to user if not provided)
creator_vault: Creator vault address (will be derived if not provided)
**kwargs: Additional fields for TokenInfo
Returns:
Enhanced TokenInfo configured for pump.fun
"""
from core.pubkeys import PumpAddresses
# Default creator to user if not provided
if creator is None:
creator = user
# Derive creator vault if not provided
if creator_vault is None:
creator_vault, _ = Pubkey.find_program_address(
[b"creator-vault", bytes(creator)],
PumpAddresses.PROGRAM,
)
return TokenInfo(
name=name,
symbol=symbol,
uri=uri,
mint=mint,
platform=Platform.PUMP_FUN,
bonding_curve=bonding_curve,
associated_bonding_curve=associated_bonding_curve,
user=user,
creator=creator,
creator_vault=creator_vault,
**kwargs
)
def create_lets_bonk_token_info(
name: str,
symbol: str,
uri: str,
mint: Pubkey,
pool_state: Pubkey,
base_vault: Pubkey,
quote_vault: Pubkey,
user: Pubkey,
creator: Pubkey | None = None,
**kwargs
) -> TokenInfo:
"""Convenience function to create LetsBonk TokenInfo with proper platform setting.
Args:
name: Token name
symbol: Token symbol
uri: Token metadata URI
mint: Token mint address
pool_state: Pool state address
base_vault: Base token vault address
quote_vault: Quote token vault address
user: User/trader address
creator: Creator address (defaults to user if not provided)
**kwargs: Additional fields for TokenInfo
Returns:
Enhanced TokenInfo configured for LetsBonk
"""
# Default creator to user if not provided
if creator is None:
creator = user
return TokenInfo(
name=name,
symbol=symbol,
uri=uri,
mint=mint,
platform=Platform.LETS_BONK,
pool_state=pool_state,
base_vault=base_vault,
quote_vault=quote_vault,
user=user,
creator=creator,
**kwargs
)
def is_pump_fun_token(token_info: TokenInfo) -> bool:
"""Check if a TokenInfo is for pump.fun platform.
Args:
token_info: Token information to check
Returns:
True if token is for pump.fun platform
"""
return token_info.platform == Platform.PUMP_FUN
def is_lets_bonk_token(token_info: TokenInfo) -> bool:
"""Check if a TokenInfo is for LetsBonk platform.
Args:
token_info: Token information to check
Returns:
True if token is for LetsBonk platform
"""
return token_info.platform == Platform.LETS_BONK
def get_platform_specific_fields(token_info: TokenInfo) -> dict[str, Any]:
"""Get platform-specific fields from TokenInfo.
Args:
token_info: Token information
Returns:
Dictionary of platform-specific fields
"""
if token_info.platform == Platform.PUMP_FUN:
return {
"bonding_curve": token_info.bonding_curve,
"associated_bonding_curve": token_info.associated_bonding_curve,
"creator_vault": token_info.creator_vault,
}
elif token_info.platform == Platform.LETS_BONK:
return {
"pool_state": token_info.pool_state,
"base_vault": token_info.base_vault,
"quote_vault": token_info.quote_vault,
}
else:
return {}
def validate_token_info(token_info: TokenInfo) -> bool:
"""Validate that TokenInfo has required fields for its platform.
Args:
token_info: Token information to validate
Returns:
True if TokenInfo is valid for its platform
"""
# Check common required fields
if not all([
token_info.name,
token_info.symbol,
token_info.mint,
token_info.platform,
]):
return False
# Check platform-specific required fields
if token_info.platform == Platform.PUMP_FUN:
return all([
token_info.bonding_curve,
token_info.associated_bonding_curve,
])
elif token_info.platform == Platform.LETS_BONK:
return all([
token_info.pool_state,
token_info.base_vault,
token_info.quote_vault,
])
return True
# Backward compatibility exports
# This allows existing imports to continue working
__all__ = [
'Platform', # Platform enum
'TokenInfo', # Enhanced TokenInfo (main export)
'TokenInfo_Legacy', # Legacy TokenInfo for compatibility
'TradeResult', # Enhanced TradeResult
'Trader', # Enhanced Trader base class
'create_legacy_token_info',
'create_lets_bonk_token_info',
# Convenience functions
'create_pump_fun_token_info',
'get_platform_specific_fields',
'is_lets_bonk_token',
# Utility functions
'is_pump_fun_token',
# Conversion functions
'upgrade_token_info',
'validate_token_info',
]
src/trading/platform_aware.py
+315
View File
@@ -0,0 +1,315 @@
"""
Platform-aware trader implementations that use the interface system.
This module provides new trader classes that work with any platform
through the interface system, while maintaining compatibility with existing code.
"""
from solders.pubkey import Pubkey
from core.client import SolanaClient
from core.priority_fee.manager import PriorityFeeManager
from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS
from core.wallet import Wallet
from interfaces.core import AddressProvider, Platform, TokenInfo
from platforms import get_platform_implementations
from trading.base import Trader, TradeResult
from utils.logger import get_logger
logger = get_logger(__name__)
class PlatformAwareBuyer(Trader):
"""Platform-aware token buyer that works with any supported platform."""
def __init__(
self,
client: SolanaClient,
wallet: Wallet,
priority_fee_manager: PriorityFeeManager,
amount: float,
slippage: float = 0.01,
max_retries: int = 5,
extreme_fast_token_amount: int = 0,
extreme_fast_mode: bool = False,
):
"""Initialize platform-aware token buyer.
Args:
client: Solana client for RPC calls
wallet: Wallet for signing transactions
priority_fee_manager: Priority fee manager
amount: Amount of SOL to spend
slippage: Slippage tolerance (0.01 = 1%)
max_retries: Maximum number of retry attempts
extreme_fast_token_amount: Amount of token to buy if extreme fast mode is enabled
extreme_fast_mode: If enabled, avoid fetching pool state for price estimation
"""
self.client = client
self.wallet = wallet
self.priority_fee_manager = priority_fee_manager
self.amount = amount
self.slippage = slippage
self.max_retries = max_retries
self.extreme_fast_mode = extreme_fast_mode
self.extreme_fast_token_amount = extreme_fast_token_amount
async def execute(self, token_info: TokenInfo, *args, **kwargs) -> TradeResult:
"""Execute buy operation using platform-specific implementations.
Args:
token_info: Enhanced token information with platform
Returns:
TradeResult with buy outcome
"""
try:
# Get platform-specific implementations
implementations = get_platform_implementations(token_info.platform, self.client)
address_provider = implementations.address_provider
instruction_builder = implementations.instruction_builder
curve_manager = implementations.curve_manager
# Convert amount to lamports
amount_lamports = int(self.amount * LAMPORTS_PER_SOL)
if self.extreme_fast_mode:
# Skip the wait and directly calculate the amount
token_amount = self.extreme_fast_token_amount
token_price_sol = self.amount / token_amount if token_amount > 0 else 0
else:
# Get pool address based on platform
pool_address = self._get_pool_address(token_info, address_provider)
# Regular behavior with RPC call
token_price_sol = await curve_manager.calculate_price(pool_address)
token_amount = self.amount / token_price_sol if token_price_sol > 0 else 0
# Calculate minimum token amount with slippage
minimum_token_amount = token_amount * (1 - self.slippage)
minimum_token_amount_raw = int(minimum_token_amount * 10**TOKEN_DECIMALS)
# Calculate maximum SOL to spend with slippage
max_amount_lamports = int(amount_lamports * (1 + self.slippage))
# Build buy instructions
instructions = await instruction_builder.build_buy_instruction(
token_info,
self.wallet.pubkey,
max_amount_lamports, # amount_in (SOL)
minimum_token_amount_raw, # minimum_amount_out (tokens)
address_provider
)
# Get accounts for priority fee calculation
priority_accounts = instruction_builder.get_required_accounts_for_buy(
token_info, self.wallet.pubkey, address_provider
)
logger.info(
f"Buying {token_amount:.6f} tokens at {token_price_sol:.8f} SOL per token"
)
logger.info(
f"Total cost: {self.amount:.6f} SOL (max: {max_amount_lamports / LAMPORTS_PER_SOL:.6f} SOL)"
)
# Send transaction
tx_signature = await self.client.build_and_send_transaction(
instructions,
self.wallet.keypair,
skip_preflight=True,
max_retries=self.max_retries,
priority_fee=await self.priority_fee_manager.calculate_priority_fee(
priority_accounts
),
)
success = await self.client.confirm_transaction(tx_signature)
if success:
logger.info(f"Buy transaction confirmed: {tx_signature}")
return TradeResult(
success=True,
platform=token_info.platform,
tx_signature=tx_signature,
amount=token_amount,
price=token_price_sol,
)
else:
return TradeResult(
success=False,
platform=token_info.platform,
error_message=f"Transaction failed to confirm: {tx_signature}",
)
except Exception as e:
logger.error(f"Buy operation failed: {e!s}")
return TradeResult(
success=False,
platform=token_info.platform,
error_message=str(e)
)
def _get_pool_address(self, token_info: TokenInfo, address_provider: AddressProvider) -> Pubkey:
"""Get the pool/curve address for price calculations.
Args:
token_info: Token information
address_provider: Platform address provider
Returns:
Pool/curve address
"""
if token_info.platform == Platform.PUMP_FUN:
return token_info.bonding_curve or address_provider.derive_pool_address(token_info.mint)
elif token_info.platform == Platform.LETS_BONK:
return token_info.pool_state or address_provider.derive_pool_address(token_info.mint)
else:
# Fallback to deriving the address
return address_provider.derive_pool_address(token_info.mint)
class PlatformAwareSeller(Trader):
"""Platform-aware token seller that works with any supported platform."""
def __init__(
self,
client: SolanaClient,
wallet: Wallet,
priority_fee_manager: PriorityFeeManager,
slippage: float = 0.25,
max_retries: int = 5,
):
"""Initialize platform-aware token seller.
Args:
client: Solana client for RPC calls
wallet: Wallet for signing transactions
priority_fee_manager: Priority fee manager
slippage: Slippage tolerance (0.25 = 25%)
max_retries: Maximum number of retry attempts
"""
self.client = client
self.wallet = wallet
self.priority_fee_manager = priority_fee_manager
self.slippage = slippage
self.max_retries = max_retries
async def execute(self, token_info: TokenInfo, *args, **kwargs) -> TradeResult:
"""Execute sell operation using platform-specific implementations.
Args:
token_info: Enhanced token information with platform
Returns:
TradeResult with sell outcome
"""
try:
# Get platform-specific implementations
implementations = get_platform_implementations(token_info.platform, self.client)
address_provider = implementations.address_provider
instruction_builder = implementations.instruction_builder
curve_manager = implementations.curve_manager
# Get user's token account and balance
user_token_account = address_provider.derive_user_token_account(
self.wallet.pubkey, token_info.mint
)
token_balance = await self.client.get_token_account_balance(user_token_account)
token_balance_decimal = token_balance / 10**TOKEN_DECIMALS
logger.info(f"Token balance: {token_balance_decimal}")
if token_balance == 0:
logger.info("No tokens to sell.")
return TradeResult(
success=False,
platform=token_info.platform,
error_message="No tokens to sell"
)
# Get pool address and current price
pool_address = self._get_pool_address(token_info, address_provider)
token_price_sol = await curve_manager.calculate_price(pool_address)
logger.info(f"Price per Token: {token_price_sol:.8f} SOL")
# Calculate minimum SOL output with slippage
expected_sol_output = float(token_balance_decimal) * float(token_price_sol)
min_sol_output = int((expected_sol_output * (1 - self.slippage)) * LAMPORTS_PER_SOL)
logger.info(f"Selling {token_balance_decimal} tokens")
logger.info(f"Expected SOL output: {expected_sol_output:.8f} SOL")
logger.info(
f"Minimum SOL output (with {self.slippage * 100}% slippage): {min_sol_output / LAMPORTS_PER_SOL:.8f} SOL"
)
# Build sell instructions
instructions = await instruction_builder.build_sell_instruction(
token_info,
self.wallet.pubkey,
token_balance, # amount_in (tokens)
min_sol_output, # minimum_amount_out (SOL)
address_provider
)
# Get accounts for priority fee calculation
priority_accounts = instruction_builder.get_required_accounts_for_sell(
token_info, self.wallet.pubkey, address_provider
)
# Send transaction
tx_signature = await self.client.build_and_send_transaction(
instructions,
self.wallet.keypair,
skip_preflight=True,
max_retries=self.max_retries,
priority_fee=await self.priority_fee_manager.calculate_priority_fee(
priority_accounts
),
)
success = await self.client.confirm_transaction(tx_signature)
if success:
logger.info(f"Sell transaction confirmed: {tx_signature}")
return TradeResult(
success=True,
platform=token_info.platform,
tx_signature=tx_signature,
amount=token_balance_decimal,
price=token_price_sol,
)
else:
return TradeResult(
success=False,
platform=token_info.platform,
error_message=f"Transaction failed to confirm: {tx_signature}",
)
except Exception as e:
logger.error(f"Sell operation failed: {e!s}")
return TradeResult(
success=False,
platform=token_info.platform,
error_message=str(e)
)
def _get_pool_address(self, token_info: TokenInfo, address_provider: AddressProvider) -> Pubkey:
"""Get the pool/curve address for price calculations.
Args:
token_info: Token information
address_provider: Platform address provider
Returns:
Pool/curve address
"""
if token_info.platform == Platform.PUMP_FUN:
return token_info.bonding_curve or address_provider.derive_pool_address(token_info.mint)
elif token_info.platform == Platform.LETS_BONK:
return token_info.pool_state or address_provider.derive_pool_address(token_info.mint)
else:
# Fallback to deriving the address
return address_provider.derive_pool_address(token_info.mint)