diff --git a/src/bot_runner.py b/src/bot_runner.py index 9a73351..0dc79da 100644 --- a/src/bot_runner.py +++ b/src/bot_runner.py @@ -8,8 +8,12 @@ import uvloop asyncio.set_event_loop_policy(uvloop.EventLoopPolicy()) -from config_loader import load_bot_config, print_config_summary -from trading.trader import PumpTrader +from config_loader import ( + get_platform_from_config, + load_bot_config, + print_config_summary, +) +from trading.universal_trader import UniversalTrader from utils.logger import setup_file_logging @@ -40,34 +44,48 @@ async def start_bot(config_path: str): setup_logging(cfg["name"]) print_config_summary(cfg) - trader = PumpTrader( + # Get platform from configuration + platform = get_platform_from_config(cfg) + + # Initialize universal trader with platform-specific configuration + trader = UniversalTrader( # Connection settings rpc_endpoint=cfg["rpc_endpoint"], wss_endpoint=cfg["wss_endpoint"], private_key=cfg["private_key"], + + # Platform configuration + platform=platform, + # Trade parameters buy_amount=cfg["trade"]["buy_amount"], buy_slippage=cfg["trade"]["buy_slippage"], sell_slippage=cfg["trade"]["sell_slippage"], + # Extreme fast mode settings extreme_fast_mode=cfg["trade"].get("extreme_fast_mode", False), extreme_fast_token_amount=cfg["trade"].get("extreme_fast_token_amount", 30), + # Exit strategy configuration exit_strategy=cfg["trade"].get("exit_strategy", "time_based"), take_profit_percentage=cfg["trade"].get("take_profit_percentage"), stop_loss_percentage=cfg["trade"].get("stop_loss_percentage"), max_hold_time=cfg["trade"].get("max_hold_time"), price_check_interval=cfg["trade"].get("price_check_interval", 10), + # Listener configuration listener_type=cfg["filters"]["listener_type"], + # Geyser configuration (if applicable) geyser_endpoint=cfg.get("geyser", {}).get("endpoint"), geyser_api_token=cfg.get("geyser", {}).get("api_token"), - geyser_auth_type=cfg.get("geyser", {}).get("auth_type"), + geyser_auth_type=cfg.get("geyser", {}).get("auth_type", "x-token"), + # PumpPortal configuration (if applicable) pumpportal_url=cfg.get("pumpportal", {}).get( "url", "wss://pumpportal.fun/api/data" ), + # Priority fee configuration enable_dynamic_priority_fee=cfg.get("priority_fees", {}).get( "enable_dynamic", False @@ -78,6 +96,7 @@ async def start_bot(config_path: str): fixed_priority_fee=cfg.get("priority_fees", {}).get("fixed_amount", 500000), extra_priority_fee=cfg.get("priority_fees", {}).get("extra_percentage", 0.0), hard_cap_prior_fee=cfg.get("priority_fees", {}).get("hard_cap", 500000), + # Retry and timeout settings max_retries=cfg.get("retries", {}).get("max_attempts", 10), wait_time_after_creation=cfg.get("retries", {}).get("wait_after_creation", 15), @@ -85,8 +104,9 @@ async def start_bot(config_path: str): wait_time_before_new_token=cfg.get("retries", {}).get( "wait_before_new_token", 15 ), - max_token_age=cfg.get("timing", {}).get("max_token_age", 0.001), + max_token_age=cfg.get("filters", {}).get("max_token_age", 0.001), token_wait_timeout=cfg.get("timing", {}).get("token_wait_timeout", 120), + # Cleanup settings cleanup_mode=cfg.get("cleanup", {}).get("mode", "disabled"), cleanup_force_close_with_burn=cfg.get("cleanup", {}).get( @@ -95,6 +115,7 @@ async def start_bot(config_path: str): cleanup_with_priority_fee=cfg.get("cleanup", {}).get( "with_priority_fee", False ), + # Trading filters match_string=cfg["filters"].get("match_string"), bro_address=cfg["filters"].get("bro_address"), @@ -141,21 +162,34 @@ def run_all_bots(): skipped_bots += 1 continue + # Validate platform support + try: + platform = get_platform_from_config(cfg) + from platforms import platform_factory + if not platform_factory.registry.is_platform_supported(platform): + logging.error(f"Platform {platform.value} is not supported for bot '{bot_name}'. Skipping...") + skipped_bots += 1 + continue + except Exception as e: + logging.exception(f"Invalid platform configuration for bot '{bot_name}': {e}. Skipping...") + skipped_bots += 1 + continue + if cfg.get("separate_process", False): - logging.info(f"Starting bot '{bot_name}' in separate process") + logging.info(f"Starting bot '{bot_name}' ({platform.value}) in separate process") p = multiprocessing.Process( target=run_bot_process, args=(str(file),), name=f"bot-{bot_name}" ) p.start() processes.append(p) else: - logging.info(f"Starting bot '{bot_name}' in main process") + logging.info(f"Starting bot '{bot_name}' ({platform.value}) in main process") asyncio.run(start_bot(str(file))) except Exception as e: logging.exception(f"Failed to start bot from {file}: {e}") logging.info( - f"Started {len(bot_files) - skipped_bots} bots, skipped {skipped_bots} disabled bots" + f"Started {len(bot_files) - skipped_bots} bots, skipped {skipped_bots} disabled/invalid bots" ) for p in processes: @@ -169,8 +203,16 @@ def main() -> None: format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", ) + # Log supported platforms + try: + from platforms import platform_factory + supported_platforms = platform_factory.get_supported_platforms() + logging.info(f"Supported platforms: {[p.value for p in supported_platforms]}") + except Exception as e: + logging.warning(f"Could not load platform information: {e}") + run_all_bots() if __name__ == "__main__": - main() + main() \ No newline at end of file diff --git a/src/config_loader.py b/src/config_loader.py index 79e3c47..8d0313d 100644 --- a/src/config_loader.py +++ b/src/config_loader.py @@ -79,7 +79,7 @@ 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 + "platform": ["pump_fun", "lets_bonk"], # Platform validation } diff --git a/src/core/curve.py b/src/core/curve.py deleted file mode 100644 index e56651d..0000000 --- a/src/core/curve.py +++ /dev/null @@ -1,140 +0,0 @@ -""" -Bonding curve operations for pump.fun tokens. -""" - -import struct -from typing import Final - -from construct import Bytes, Flag, Int64ul, Struct -from solders.pubkey import Pubkey - -from core.client import SolanaClient -from core.pubkeys import LAMPORTS_PER_SOL, TOKEN_DECIMALS -from utils.logger import get_logger - -logger = get_logger(__name__) - -# Discriminator for the bonding curve account -EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" None: - """Parse bonding curve data. - - Args: - data: Raw account data - - Raises: - ValueError: If data cannot be parsed - """ - if data[:8] != EXPECTED_DISCRIMINATOR: - raise ValueError("Invalid curve state discriminator") - - parsed = self._STRUCT.parse(data[8:]) - self.__dict__.update(parsed) - - # Convert raw bytes to Pubkey for creator field - if hasattr(self, "creator") and isinstance(self.creator, bytes): - self.creator = Pubkey.from_bytes(self.creator) - - def calculate_price(self) -> float: - """Calculate token price in SOL. - - Returns: - Token price in SOL - - Raises: - ValueError: If reserve state is invalid - """ - if self.virtual_token_reserves <= 0 or self.virtual_sol_reserves <= 0: - raise ValueError("Invalid reserve state") - - return (self.virtual_sol_reserves / LAMPORTS_PER_SOL) / ( - self.virtual_token_reserves / 10**TOKEN_DECIMALS - ) - - @property - def token_reserves(self) -> float: - """Get token reserves in decimal form.""" - return self.virtual_token_reserves / 10**TOKEN_DECIMALS - - @property - def sol_reserves(self) -> float: - """Get SOL reserves in decimal form.""" - return self.virtual_sol_reserves / LAMPORTS_PER_SOL - - -class BondingCurveManager: - """Manager for bonding curve operations.""" - - def __init__(self, client: SolanaClient): - """Initialize with Solana client. - - Args: - client: Solana client for RPC calls - """ - self.client = client - - async def get_curve_state(self, curve_address: Pubkey) -> BondingCurveState: - """Get the state of a bonding curve. - - Args: - curve_address: Address of the bonding curve account - - Returns: - Bonding curve state - - Raises: - ValueError: If curve data is invalid - """ - try: - account = await self.client.get_account_info(curve_address) - if not account.data: - raise ValueError(f"No data in bonding curve account {curve_address}") - - return BondingCurveState(account.data) - - except Exception as e: - logger.error(f"Failed to get curve state: {e!s}") - raise ValueError(f"Invalid curve state: {e!s}") - - async def calculate_price(self, curve_address: Pubkey) -> float: - """Calculate the current price of a token. - - Args: - curve_address: Address of the bonding curve account - - Returns: - Token price in SOL - """ - curve_state = await self.get_curve_state(curve_address) - return curve_state.calculate_price() - - async def calculate_expected_tokens( - self, curve_address: Pubkey, sol_amount: float - ) -> float: - """Calculate the expected token amount for a given SOL input. - - Args: - curve_address: Address of the bonding curve account - sol_amount: Amount of SOL to spend - - Returns: - Expected token amount - """ - curve_state = await self.get_curve_state(curve_address) - price = curve_state.calculate_price() - return sol_amount / price diff --git a/src/interfaces/__init__.py b/src/interfaces/__init__.py deleted file mode 100644 index c2b3dbf..0000000 --- a/src/interfaces/__init__.py +++ /dev/null @@ -1,280 +0,0 @@ -""" -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 - ) \ No newline at end of file diff --git a/src/monitoring/base_listener.py b/src/monitoring/base_listener.py index 0a5e191..10184e2 100644 --- a/src/monitoring/base_listener.py +++ b/src/monitoring/base_listener.py @@ -1,15 +1,23 @@ """ -Base class for WebSocket token listeners. +Base class for WebSocket token listeners - now platform-agnostic. """ from abc import ABC, abstractmethod from collections.abc import Awaitable, Callable -from trading.base import TokenInfo +from interfaces.core import Platform, TokenInfo class BaseTokenListener(ABC): - """Base abstract class for token listeners.""" + """Base abstract class for token listeners - now platform-agnostic.""" + + def __init__(self, platform: Platform | None = None): + """Initialize the listener with optional platform specification. + + Args: + platform: Platform to monitor (if None, monitor all platforms) + """ + self.platform = platform @abstractmethod async def listen_for_tokens( @@ -27,3 +35,16 @@ class BaseTokenListener(ABC): creator_address: Optional creator address to filter by """ pass + + def should_process_token(self, token_info: TokenInfo) -> bool: + """Check if a token should be processed based on platform filter. + + Args: + token_info: Token information + + Returns: + True if token should be processed + """ + if self.platform is None: + return True # Process all platforms + return token_info.platform == self.platform \ No newline at end of file diff --git a/src/monitoring/block_event_processor.py b/src/monitoring/block_event_processor.py deleted file mode 100644 index a3defbe..0000000 --- a/src/monitoring/block_event_processor.py +++ /dev/null @@ -1,190 +0,0 @@ -""" -Event processing for pump.fun tokens. -""" - -import base64 -import json -import struct -from typing import Any - -import base58 -from solders.pubkey import Pubkey -from solders.transaction import VersionedTransaction - -from core.pubkeys import PumpAddresses -from trading.base import TokenInfo -from utils.logger import get_logger - -logger = get_logger(__name__) - - -class PumpEventProcessor: - """Processes events from pump.fun program.""" - - # Discriminator for create instruction - CREATE_DISCRIMINATOR = 8576854823835016728 - - def __init__(self, pump_program: Pubkey): - """Initialize event processor. - - Args: - pump_program: Pump.fun program address - """ - self.pump_program = pump_program - self._idl = self._load_idl() - - def _load_idl(self) -> dict[str, Any]: - """Load IDL from file. - - Returns: - IDL as dictionary - """ - try: - with open("idl/pump_fun_idl.json") as f: - return json.load(f) - except Exception as e: - logger.error(f"Failed to load IDL: {e!s}") - # Create a minimal IDL with just what we need - return { - "instructions": [ - { - "name": "create", - "args": [ - {"name": "name", "type": "string"}, - {"name": "symbol", "type": "string"}, - {"name": "uri", "type": "string"}, - ], - } - ] - } - - def process_transaction(self, tx_data: str) -> TokenInfo | None: - """Process a transaction and extract token info. - - Args: - tx_data: Base64 encoded transaction data - - Returns: - TokenInfo if a token creation is found, None otherwise - """ - try: - tx_data_decoded = base64.b64decode(tx_data) - transaction = VersionedTransaction.from_bytes(tx_data_decoded) - - for ix in transaction.message.instructions: - # Check if instruction is from pump.fun program - program_id_index = ix.program_id_index - if program_id_index >= len(transaction.message.account_keys): - continue - - program_id = transaction.message.account_keys[program_id_index] - - if str(program_id) != str(self.pump_program): - continue - - ix_data = bytes(ix.data) - - # Check if it's a create instruction - if len(ix_data) < 8: - continue - - discriminator = struct.unpack(" dict[str, Any]: - """Decode create instruction data. - - Args: - ix_data: Instruction data bytes - ix_def: Instruction definition from IDL - accounts: List of account pubkeys - - Returns: - Decoded instruction arguments - """ - args = {} - offset = 8 # Skip 8-byte discriminator - - for arg in ix_def["args"]: - if arg["type"] == "string": - length = struct.unpack_from(" Pubkey: - """ - Find 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 diff --git a/src/monitoring/geyser_event_processor.py b/src/monitoring/geyser_event_processor.py deleted file mode 100644 index 3331c1b..0000000 --- a/src/monitoring/geyser_event_processor.py +++ /dev/null @@ -1,125 +0,0 @@ -""" -Event processing for pump.fun tokens using Geyser data. -""" - -import struct -from typing import Final - -import base58 -from solders.pubkey import Pubkey - -from core.pubkeys import PumpAddresses -from trading.base import TokenInfo -from utils.logger import get_logger - -logger = get_logger(__name__) - - -class GeyserEventProcessor: - """Processes token creation events from Geyser stream.""" - - CREATE_DISCRIMINATOR: Final[bytes] = struct.pack(" TokenInfo | None: - """Process transaction data and extract token creation info. - - Args: - instruction_data: Raw instruction data - accounts: List of account indices - keys: List of account public keys - - Returns: - TokenInfo if token creation found, None otherwise - """ - if not instruction_data.startswith(self.CREATE_DISCRIMINATOR): - return None - - try: - # Skip past the 8-byte discriminator - offset = 8 - - # Helper to read strings (prefixed with length) - def read_string(): - nonlocal offset - # Get string length (4-byte uint) - length = struct.unpack_from("= len(accounts): - return None - account_index = accounts[index] - if account_index >= len(keys): - return None - return Pubkey.from_bytes(keys[account_index]) - - name = read_string() - symbol = read_string() - uri = read_string() - creator = read_pubkey() - - mint = get_account_key(0) - bonding_curve = get_account_key(2) - associated_bonding_curve = get_account_key(3) - user = get_account_key(7) - - creator_vault = self._find_creator_vault(creator) - - if not all([mint, bonding_curve, associated_bonding_curve, user]): - logger.warning("Missing required account keys in token creation") - return None - - return TokenInfo( - name=name, - symbol=symbol, - uri=uri, - mint=mint, - bonding_curve=bonding_curve, - associated_bonding_curve=associated_bonding_curve, - user=user, - creator=creator, - creator_vault=creator_vault, - ) - - except Exception as e: - logger.error(f"Failed to process transaction data: {e}") - return None - - def _find_creator_vault(self, creator: Pubkey) -> Pubkey: - """ - Find 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 diff --git a/src/monitoring/listener_factory.py b/src/monitoring/listener_factory.py new file mode 100644 index 0000000..93dca13 --- /dev/null +++ b/src/monitoring/listener_factory.py @@ -0,0 +1,137 @@ +""" +Factory for creating platform-aware token listeners. +""" + +from interfaces.core import Platform +from monitoring.base_listener import BaseTokenListener +from utils.logger import get_logger + +logger = get_logger(__name__) + + +class ListenerFactory: + """Factory for creating appropriate token listeners based on configuration.""" + + @staticmethod + def create_listener( + listener_type: str, + wss_endpoint: str | None = None, + geyser_endpoint: str | None = None, + geyser_api_token: str | None = None, + geyser_auth_type: str = "x-token", + pumpportal_url: str = "wss://pumpportal.fun/api/data", + platforms: list[Platform] | None = None, + **kwargs + ) -> BaseTokenListener: + """Create a token listener based on the specified type. + + Args: + listener_type: Type of listener ('logs', 'blocks', 'geyser', or 'pumpportal') + wss_endpoint: WebSocket endpoint URL (for logs/blocks listeners) + geyser_endpoint: Geyser gRPC endpoint URL (for geyser listener) + geyser_api_token: Geyser API token (for geyser listener) + geyser_auth_type: Geyser authentication type + pumpportal_url: PumpPortal WebSocket URL (for pumpportal listener) + platforms: List of platforms to monitor (if None, monitor all) + **kwargs: Additional arguments + + Returns: + Configured token listener + + Raises: + ValueError: If listener type is invalid or required parameters are missing + """ + listener_type = listener_type.lower() + + if listener_type == "geyser": + if not geyser_endpoint or not geyser_api_token: + raise ValueError( + "Geyser endpoint and API token are required for geyser listener" + ) + + from monitoring.universal_geyser_listener import UniversalGeyserListener + + listener = UniversalGeyserListener( + geyser_endpoint=geyser_endpoint, + geyser_api_token=geyser_api_token, + geyser_auth_type=geyser_auth_type, + platforms=platforms, + ) + logger.info("Created Universal Geyser listener for token monitoring") + return listener + + elif listener_type == "logs": + if not wss_endpoint: + raise ValueError("WebSocket endpoint is required for logs listener") + + from monitoring.universal_logs_listener import UniversalLogsListener + + listener = UniversalLogsListener( + wss_endpoint=wss_endpoint, + platforms=platforms, + ) + logger.info("Created Universal Logs listener for token monitoring") + return listener + + elif listener_type == "blocks": + if not wss_endpoint: + raise ValueError("WebSocket endpoint is required for blocks listener") + + from monitoring.universal_block_listener import UniversalBlockListener + + listener = UniversalBlockListener( + wss_endpoint=wss_endpoint, + platforms=platforms, + ) + logger.info("Created Universal Block listener for token monitoring") + return listener + + elif listener_type == "pumpportal": + # PumpPortal is pump.fun specific, so filter platforms + pumpfun_platforms = [Platform.PUMP_FUN] + if platforms: + pumpfun_platforms = [p for p in platforms if p == Platform.PUMP_FUN] + + if not pumpfun_platforms: + raise ValueError("PumpPortal listener only supports pump.fun platform") + + from monitoring.pumpportal_listener import PumpPortalListener + + listener = PumpPortalListener( + pump_program=None, # Will be determined from platform + pumpportal_url=pumpportal_url, + ) + logger.info("Created PumpPortal listener for token monitoring") + return listener + + else: + raise ValueError( + f"Invalid listener type '{listener_type}'. " + f"Must be one of: 'logs', 'blocks', 'geyser', 'pumpportal'" + ) + + @staticmethod + def get_supported_listener_types() -> list[str]: + """Get list of supported listener types. + + Returns: + List of supported listener type strings + """ + return ["logs", "blocks", "geyser", "pumpportal"] + + @staticmethod + def get_platform_compatible_listeners(platform: Platform) -> list[str]: + """Get list of listener types compatible with a specific platform. + + Args: + platform: Platform to check compatibility for + + Returns: + List of compatible listener types + """ + if platform == Platform.PUMP_FUN: + return ["logs", "blocks", "geyser", "pumpportal"] + elif platform == Platform.LETS_BONK: + return ["logs", "blocks", "geyser"] # PumpPortal is pump.fun only + else: + return ["logs", "blocks", "geyser"] # Default universal listeners \ No newline at end of file diff --git a/src/monitoring/logs_event_processor.py b/src/monitoring/logs_event_processor.py deleted file mode 100644 index b1bca10..0000000 --- a/src/monitoring/logs_event_processor.py +++ /dev/null @@ -1,174 +0,0 @@ -""" -Event processing for pump.fun tokens using logsSubscribe data. -""" - -import base64 -import struct -from typing import Final - -import base58 -from solders.pubkey import Pubkey - -from core.pubkeys import PumpAddresses, SystemAddresses -from trading.base import TokenInfo -from utils.logger import get_logger - -logger = get_logger(__name__) - - -class LogsEventProcessor: - """Processes events from pump.fun program logs.""" - - # Discriminator for create instruction to avoid non-create transactions - CREATE_DISCRIMINATOR: Final[int] = 8530921459188068891 - - def __init__(self, pump_program: Pubkey): - """Initialize event processor. - - Args: - pump_program: Pump.fun program address - """ - self.pump_program = pump_program - - def process_program_logs(self, logs: list[str], signature: str) -> TokenInfo | None: - """Process program logs and extract token info. - - Args: - logs: List of log strings from the notification - signature: Transaction signature - - Returns: - TokenInfo if a token creation is 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 as the first condition may pass them - 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) - parsed_data = self._parse_create_instruction(decoded_data) - - if parsed_data and "name" in parsed_data: - mint = Pubkey.from_string(parsed_data["mint"]) - bonding_curve = Pubkey.from_string(parsed_data["bondingCurve"]) - associated_curve = self._find_associated_bonding_curve( - mint, bonding_curve - ) - creator = Pubkey.from_string(parsed_data["creator"]) - creator_vault = self._find_creator_vault(creator) - - return TokenInfo( - name=parsed_data["name"], - symbol=parsed_data["symbol"], - uri=parsed_data["uri"], - mint=mint, - bonding_curve=bonding_curve, - associated_bonding_curve=associated_curve, - user=Pubkey.from_string(parsed_data["user"]), - creator=creator, - creator_vault=creator_vault, - ) - except Exception as e: - logger.error(f"Failed to process log data: {e}") - - return None - - def _parse_create_instruction(self, data: bytes) -> dict | None: - """Parse the create instruction data. - - 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(" Pubkey: - """ - Find the associated bonding curve for a given mint and bonding curve. - This uses the standard ATA derivation. - - 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 - - def _find_creator_vault(self, creator: Pubkey) -> Pubkey: - """ - Find 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 diff --git a/src/monitoring/block_listener.py b/src/monitoring/universal_block_listener.py similarity index 53% rename from src/monitoring/block_listener.py rename to src/monitoring/universal_block_listener.py index 3775e79..fc47448 100644 --- a/src/monitoring/block_listener.py +++ b/src/monitoring/universal_block_listener.py @@ -1,5 +1,5 @@ """ -WebSocket monitoring for pump.fun tokens. +Universal block listener that works with any platform through the interface system. """ import asyncio @@ -7,30 +7,61 @@ import json from collections.abc import Awaitable, Callable import websockets -from solders.pubkey import Pubkey +from interfaces.core import Platform, TokenInfo from monitoring.base_listener import BaseTokenListener -from monitoring.block_event_processor import PumpEventProcessor -from trading.base import TokenInfo +from platforms import get_platform_implementations from utils.logger import get_logger logger = get_logger(__name__) -class BlockListener(BaseTokenListener): - """WebSocket listener for pump.fun token creation events using blockSubscribe.""" +class UniversalBlockListener(BaseTokenListener): + """Universal block listener that works with any platform.""" - def __init__(self, wss_endpoint: str, pump_program: Pubkey): - """Initialize token listener. + def __init__( + self, + wss_endpoint: str, + platforms: list[Platform] | None = None, + ): + """Initialize universal block listener. Args: wss_endpoint: WebSocket endpoint URL - pump_program: Pump.fun program address + platforms: List of platforms to monitor (if None, monitor all supported platforms) """ + super().__init__() self.wss_endpoint = wss_endpoint - self.pump_program = pump_program - self.event_processor = PumpEventProcessor(pump_program) self.ping_interval = 20 # seconds + + # Import platform factory and get supported platforms + from platforms import platform_factory + + if platforms is None: + # Monitor all supported platforms + self.platforms = platform_factory.get_supported_platforms() + else: + self.platforms = platforms + + # Get event parsers for all platforms + self.platform_parsers = {} + self.platform_program_ids = [] + + for platform in self.platforms: + try: + # We'll need a dummy client for getting the parser + from core.client import SolanaClient + dummy_client = SolanaClient("http://localhost") # Won't be used for parsing + + implementations = get_platform_implementations(platform, dummy_client) + parser = implementations.event_parser + self.platform_parsers[platform] = parser + self.platform_program_ids.append(str(parser.get_program_id())) + + logger.info(f"Registered platform {platform.value} with program ID {parser.get_program_id()}") + + except Exception as e: + logger.warning(f"Could not register platform {platform.value}: {e}") async def listen_for_tokens( self, @@ -38,17 +69,21 @@ class BlockListener(BaseTokenListener): match_string: str | None = None, creator_address: str | None = None, ) -> None: - """Listen for new token creations. + """Listen for new token creations using blockSubscribe. Args: token_callback: Callback function for new tokens match_string: Optional string to match in token name/symbol creator_address: Optional creator address to filter by """ + if not self.platform_parsers: + logger.error("No platform parsers available. Cannot listen for tokens.") + return + while True: try: async with websockets.connect(self.wss_endpoint) as websocket: - await self._subscribe_to_program(websocket) + await self._subscribe_to_programs(websocket) ping_task = asyncio.create_task(self._ping_loop(websocket)) try: @@ -58,9 +93,10 @@ class BlockListener(BaseTokenListener): continue logger.info( - f"New token detected: {token_info.name} ({token_info.symbol})" + f"New token detected: {token_info.name} ({token_info.symbol}) on {token_info.platform.value}" ) + # Apply filters if match_string and not ( match_string.lower() in token_info.name.lower() or match_string.lower() in token_info.symbol.lower() @@ -70,10 +106,7 @@ class BlockListener(BaseTokenListener): ) continue - if ( - creator_address - and str(token_info.user) != creator_address - ): + if creator_address and str(token_info.user) != creator_address: logger.info( f"Token not created by {creator_address}. Skipping..." ) @@ -90,32 +123,35 @@ class BlockListener(BaseTokenListener): logger.info("Reconnecting in 5 seconds...") await asyncio.sleep(5) - async def _subscribe_to_program(self, websocket) -> None: - """Subscribe to blocks mentioning the pump.fun program. + async def _subscribe_to_programs(self, websocket) -> None: + """Subscribe to blocks mentioning any of the monitored program IDs. Args: websocket: Active WebSocket connection """ - subscription_message = json.dumps( - { - "jsonrpc": "2.0", - "id": 1, - "method": "blockSubscribe", - "params": [ - {"mentionsAccountOrProgram": str(self.pump_program)}, - { - "commitment": "confirmed", - "encoding": "base64", # base64 is faster than other encoding options - "showRewards": False, - "transactionDetails": "full", - "maxSupportedTransactionVersion": 0, - }, - ], - } - ) + # For block subscriptions, we can use mentionsAccountOrProgram to monitor multiple programs + # We'll create separate subscriptions for each program to be more specific + for i, program_id in enumerate(self.platform_program_ids): + subscription_message = json.dumps( + { + "jsonrpc": "2.0", + "id": i + 1, + "method": "blockSubscribe", + "params": [ + {"mentionsAccountOrProgram": program_id}, + { + "commitment": "confirmed", + "encoding": "base64", + "showRewards": False, + "transactionDetails": "full", + "maxSupportedTransactionVersion": 0, + }, + ], + } + ) - await websocket.send(subscription_message) - logger.info(f"Subscribed to blocks mentioning program: {self.pump_program}") + await websocket.send(subscription_message) + logger.info(f"Subscribed to blocks mentioning program: {program_id}") async def _ping_loop(self, websocket) -> None: """Keep connection alive with pings. @@ -140,7 +176,7 @@ class BlockListener(BaseTokenListener): logger.error(f"Ping error: {e!s}") async def _wait_for_token_creation(self, websocket) -> TokenInfo | None: - """Wait for token creation event. + """Wait for token creation event from any platform. Args: websocket: Active WebSocket connection @@ -166,15 +202,21 @@ class BlockListener(BaseTokenListener): if "transactions" not in block: return None + # Try each platform's event parser on each transaction for tx in block["transactions"]: if not isinstance(tx, dict) or "transaction" not in tx: continue - token_info = self.event_processor.process_transaction( - tx["transaction"][0] - ) - if token_info: - return token_info + for platform, parser in self.platform_parsers.items(): + # Check if the parser has a block parsing method + if hasattr(parser, 'parse_token_creation_from_block'): + token_info = parser.parse_token_creation_from_block({ + "transactions": [tx] + }) + if token_info: + return token_info + + return None except TimeoutError: logger.debug("No data received for 30 seconds") @@ -184,4 +226,4 @@ class BlockListener(BaseTokenListener): except Exception as e: logger.error(f"Error processing WebSocket message: {e!s}") - return None + return None \ No newline at end of file diff --git a/src/monitoring/geyser_listener.py b/src/monitoring/universal_geyser_listener.py similarity index 60% rename from src/monitoring/geyser_listener.py rename to src/monitoring/universal_geyser_listener.py index b1ba4ff..0bac6f9 100644 --- a/src/monitoring/geyser_listener.py +++ b/src/monitoring/universal_geyser_listener.py @@ -1,5 +1,5 @@ """ -Geyser monitoring for pump.fun tokens. +Universal Geyser listener that works with any platform through the interface system. """ import asyncio @@ -9,34 +9,36 @@ import grpc from solders.pubkey import Pubkey from geyser.generated import geyser_pb2, geyser_pb2_grpc +from interfaces.core import Platform, TokenInfo from monitoring.base_listener import BaseTokenListener -from monitoring.geyser_event_processor import GeyserEventProcessor -from trading.base import TokenInfo +from platforms import get_platform_implementations from utils.logger import get_logger logger = get_logger(__name__) -class GeyserListener(BaseTokenListener): - """Geyser listener for pump.fun token creation events.""" +class UniversalGeyserListener(BaseTokenListener): + """Universal Geyser listener that works with any platform.""" def __init__( self, geyser_endpoint: str, geyser_api_token: str, geyser_auth_type: str, - pump_program: Pubkey, + platforms: list[Platform] | None = None, ): - """Initialize token listener. + """Initialize universal Geyser listener. Args: geyser_endpoint: Geyser gRPC endpoint URL geyser_api_token: API token for authentication geyser_auth_type: authentication type ('x-token' or 'basic') - pump_program: Pump.fun program address + platforms: List of platforms to monitor (if None, monitor all supported platforms) """ + super().__init__() self.geyser_endpoint = geyser_endpoint self.geyser_api_token = geyser_api_token + valid_auth_types = {"x-token", "basic"} self.auth_type: str = (geyser_auth_type or "x-token").lower() if self.auth_type not in valid_auth_types: @@ -44,8 +46,35 @@ class GeyserListener(BaseTokenListener): f"Unsupported auth_type={self.auth_type!r}. " f"Expected one of {valid_auth_types}" ) - self.pump_program = pump_program - self.event_processor = GeyserEventProcessor(pump_program) + + # Import platform factory and get supported platforms + from platforms import platform_factory + + if platforms is None: + # Monitor all supported platforms + self.platforms = platform_factory.get_supported_platforms() + else: + self.platforms = platforms + + # Get event parsers for all platforms + self.platform_parsers = {} + self.platform_program_ids = set() + + for platform in self.platforms: + try: + # We'll need a dummy client for getting the parser - this is a design issue we should fix + from core.client import SolanaClient + dummy_client = SolanaClient("http://localhost") # Won't be used for parsing + + implementations = get_platform_implementations(platform, dummy_client) + parser = implementations.event_parser + self.platform_parsers[platform] = parser + self.platform_program_ids.add(parser.get_program_id()) + + logger.info(f"Registered platform {platform.value} with program ID {parser.get_program_id()}") + + except Exception as e: + logger.warning(f"Could not register platform {platform.value}: {e}") async def _create_geyser_connection(self): """Establish a secure connection to the Geyser endpoint.""" @@ -66,12 +95,15 @@ class GeyserListener(BaseTokenListener): return geyser_pb2_grpc.GeyserStub(channel), channel def _create_subscription_request(self): - """Create a subscription request for Pump.fun transactions.""" + """Create a subscription request for all monitored platforms.""" request = geyser_pb2.SubscribeRequest() - request.transactions["pump_filter"].account_include.append( - str(self.pump_program) - ) - request.transactions["pump_filter"].failed = False + + # Add all platform program IDs to the filter + for program_id in self.platform_program_ids: + filter_name = f"platform_filter_{program_id}" + request.transactions[filter_name].account_include.append(str(program_id)) + request.transactions[filter_name].failed = False + request.commitment = geyser_pb2.CommitmentLevel.PROCESSED return request @@ -88,15 +120,18 @@ class GeyserListener(BaseTokenListener): match_string: Optional string to match in token name/symbol creator_address: Optional creator address to filter by """ + if not self.platform_parsers: + logger.error("No platform parsers available. Cannot listen for tokens.") + return + while True: try: stub, channel = await self._create_geyser_connection() request = self._create_subscription_request() logger.info(f"Connected to Geyser endpoint: {self.geyser_endpoint}") - logger.info( - f"Monitoring for transactions involving program: {self.pump_program}" - ) + logger.info(f"Monitoring platforms: {[p.value for p in self.platforms]}") + logger.info(f"Monitoring program IDs: {[str(pid) for pid in self.platform_program_ids]}") try: async for update in stub.Subscribe(iter([request])): @@ -105,9 +140,10 @@ class GeyserListener(BaseTokenListener): continue logger.info( - f"New token detected: {token_info.name} ({token_info.symbol})" + f"New token detected: {token_info.name} ({token_info.symbol}) on {token_info.platform.value}" ) + # Apply filters if match_string and not ( match_string.lower() in token_info.name.lower() or match_string.lower() in token_info.symbol.lower() @@ -156,24 +192,25 @@ class GeyserListener(BaseTokenListener): return None for ix in msg.instructions: - # Skip non-Pump.fun program instructions + # Check which platform this instruction belongs to 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.pump_program): - continue - - # Process instruction data - token_info = self.event_processor.process_transaction_data( - ix.data, ix.accounts, msg.account_keys - ) - if token_info: - return token_info + program_id = Pubkey.from_bytes(msg.account_keys[program_idx]) + + # Find the matching platform parser + for platform, parser in self.platform_parsers.items(): + if program_id == parser.get_program_id(): + # Use the platform's event parser + token_info = parser.parse_token_creation_from_instruction( + ix.data, ix.accounts, msg.account_keys + ) + if token_info: + return token_info return None except Exception as e: logger.error(f"Error processing Geyser update: {e}") - return None + return None \ No newline at end of file diff --git a/src/monitoring/logs_listener.py b/src/monitoring/universal_logs_listener.py similarity index 51% rename from src/monitoring/logs_listener.py rename to src/monitoring/universal_logs_listener.py index 4af072e..56c5f03 100644 --- a/src/monitoring/logs_listener.py +++ b/src/monitoring/universal_logs_listener.py @@ -1,5 +1,5 @@ """ -WebSocket monitoring for pump.fun tokens using logsSubscribe. +Universal logs listener that works with any platform through the interface system. """ import asyncio @@ -7,30 +7,61 @@ import json from collections.abc import Awaitable, Callable import websockets -from solders.pubkey import Pubkey +from interfaces.core import Platform, TokenInfo from monitoring.base_listener import BaseTokenListener -from monitoring.logs_event_processor import LogsEventProcessor -from trading.base import TokenInfo +from platforms import get_platform_implementations from utils.logger import get_logger logger = get_logger(__name__) -class LogsListener(BaseTokenListener): - """WebSocket listener for pump.fun token creation events using logsSubscribe.""" +class UniversalLogsListener(BaseTokenListener): + """Universal logs listener that works with any platform.""" - def __init__(self, wss_endpoint: str, pump_program: Pubkey): - """Initialize token listener. + def __init__( + self, + wss_endpoint: str, + platforms: list[Platform] | None = None, + ): + """Initialize universal logs listener. Args: wss_endpoint: WebSocket endpoint URL - pump_program: Pump.fun program address + platforms: List of platforms to monitor (if None, monitor all supported platforms) """ + super().__init__() self.wss_endpoint = wss_endpoint - self.pump_program = pump_program - self.event_processor = LogsEventProcessor(pump_program) self.ping_interval = 20 # seconds + + # Import platform factory and get supported platforms + from platforms import platform_factory + + if platforms is None: + # Monitor all supported platforms + self.platforms = platform_factory.get_supported_platforms() + else: + self.platforms = platforms + + # Get event parsers for all platforms + self.platform_parsers = {} + self.platform_program_ids = [] + + for platform in self.platforms: + try: + # We'll need a dummy client for getting the parser + from core.client import SolanaClient + dummy_client = SolanaClient("http://localhost") # Won't be used for parsing + + implementations = get_platform_implementations(platform, dummy_client) + parser = implementations.event_parser + self.platform_parsers[platform] = parser + self.platform_program_ids.append(str(parser.get_program_id())) + + logger.info(f"Registered platform {platform.value} with program ID {parser.get_program_id()}") + + except Exception as e: + logger.warning(f"Could not register platform {platform.value}: {e}") async def listen_for_tokens( self, @@ -45,6 +76,10 @@ class LogsListener(BaseTokenListener): match_string: Optional string to match in token name/symbol creator_address: Optional creator address to filter by """ + if not self.platform_parsers: + logger.error("No platform parsers available. Cannot listen for tokens.") + return + while True: try: async with websockets.connect(self.wss_endpoint) as websocket: @@ -58,9 +93,10 @@ class LogsListener(BaseTokenListener): continue logger.info( - f"New token detected: {token_info.name} ({token_info.symbol})" + f"New token detected: {token_info.name} ({token_info.symbol}) on {token_info.platform.value}" ) + # Apply filters if match_string and not ( match_string.lower() in token_info.name.lower() or match_string.lower() in token_info.symbol.lower() @@ -70,10 +106,7 @@ class LogsListener(BaseTokenListener): ) continue - if ( - creator_address - and str(token_info.user) != creator_address - ): + if creator_address and str(token_info.user) != creator_address: logger.info( f"Token not created by {creator_address}. Skipping..." ) @@ -86,38 +119,40 @@ class LogsListener(BaseTokenListener): ping_task.cancel() except Exception as e: - logger.error(f"WebSocket connection error: {str(e)}") + logger.error(f"WebSocket connection error: {e!s}") logger.info("Reconnecting in 5 seconds...") await asyncio.sleep(5) async def _subscribe_to_logs(self, websocket) -> None: - """Subscribe to logs mentioning the pump.fun program. + """Subscribe to logs mentioning any of the monitored program IDs. Args: websocket: Active WebSocket connection """ - subscription_message = json.dumps( - { - "jsonrpc": "2.0", - "id": 1, - "method": "logsSubscribe", - "params": [ - {"mentions": [str(self.pump_program)]}, - {"commitment": "processed"}, - ], - } - ) + # Subscribe to logs for all monitored platforms + for program_id in self.platform_program_ids: + subscription_message = json.dumps( + { + "jsonrpc": "2.0", + "id": len(self.platform_program_ids), # Use different IDs + "method": "logsSubscribe", + "params": [ + {"mentions": [program_id]}, + {"commitment": "processed"}, + ], + } + ) - await websocket.send(subscription_message) - logger.info(f"Subscribed to logs mentioning program: {self.pump_program}") + await websocket.send(subscription_message) + logger.info(f"Subscribed to logs mentioning program: {program_id}") - # Wait for subscription confirmation - response = await websocket.recv() - response_data = json.loads(response) - if "result" in response_data: - logger.info(f"Subscription confirmed with ID: {response_data['result']}") - else: - logger.warning(f"Unexpected subscription response: {response}") + # Wait for subscription confirmation + response = await websocket.recv() + response_data = json.loads(response) + if "result" in response_data: + logger.info(f"Subscription confirmed with ID: {response_data['result']}") + else: + logger.warning(f"Unexpected subscription response: {response}") async def _ping_loop(self, websocket) -> None: """Keep connection alive with pings. @@ -131,7 +166,7 @@ class LogsListener(BaseTokenListener): try: pong_waiter = await websocket.ping() await asyncio.wait_for(pong_waiter, timeout=10) - except asyncio.TimeoutError: + except TimeoutError: logger.warning("Ping timeout - server not responding") # Force reconnection await websocket.close() @@ -139,9 +174,17 @@ class LogsListener(BaseTokenListener): except asyncio.CancelledError: pass except Exception as e: - logger.error(f"Ping error: {str(e)}") + logger.error(f"Ping error: {e!s}") async def _wait_for_token_creation(self, websocket) -> TokenInfo | None: + """Wait for token creation events from any platform. + + Args: + websocket: Active WebSocket connection + + Returns: + TokenInfo if a token creation is found, None otherwise + """ try: response = await asyncio.wait_for(websocket.recv(), timeout=30) data = json.loads(response) @@ -153,15 +196,20 @@ class LogsListener(BaseTokenListener): logs = log_data.get("logs", []) signature = log_data.get("signature", "unknown") - # Use the processor to extract token info - return self.event_processor.process_program_logs(logs, signature) + # Try each platform's event parser + for platform, parser in self.platform_parsers.items(): + token_info = parser.parse_token_creation_from_logs(logs, signature) + if token_info: + return token_info - except asyncio.TimeoutError: + return None + + except TimeoutError: logger.debug("No data received for 30 seconds") except websockets.exceptions.ConnectionClosed: logger.warning("WebSocket connection closed") raise except Exception as e: - logger.error(f"Error processing WebSocket message: {str(e)}") + logger.error(f"Error processing WebSocket message: {e!s}") - return None + return None \ No newline at end of file diff --git a/src/platforms/__init__.py b/src/platforms/__init__.py index ea584fb..9348112 100644 --- a/src/platforms/__init__.py +++ b/src/platforms/__init__.py @@ -6,7 +6,7 @@ platform-specific implementations of the trading interfaces. """ from dataclasses import dataclass -from typing import Any, Dict, Optional, Type +from typing import Any from core.client import SolanaClient from interfaces.core import ( @@ -17,8 +17,6 @@ from interfaces.core import ( Platform, ) -from . import letsbonk, pumpfun - @dataclass class PlatformImplementations: @@ -145,21 +143,45 @@ class PlatformFactory: self._setup_default_platforms() def _setup_default_platforms(self) -> None: - """Setup default platform registrations. + """Setup default platform registrations.""" + # Import platform implementations dynamically to avoid circular imports + try: + from platforms.pumpfun import ( + PumpFunAddressProvider, + PumpFunCurveManager, + PumpFunEventParser, + PumpFunInstructionBuilder, + ) + + self.registry.register_platform( + Platform.PUMP_FUN, + PumpFunAddressProvider, + PumpFunInstructionBuilder, + PumpFunCurveManager, + PumpFunEventParser + ) + + except ImportError as e: + print(f"Warning: Could not register pump.fun platform: {e}") - 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 + try: + from platforms.letsbonk import ( + LetsBonkAddressProvider, + LetsBonkCurveManager, + LetsBonkEventParser, + LetsBonkInstructionBuilder, + ) + + self.registry.register_platform( + Platform.LETS_BONK, + LetsBonkAddressProvider, + LetsBonkInstructionBuilder, + LetsBonkCurveManager, + LetsBonkEventParser + ) + + except ImportError as e: + print(f"Warning: Could not register LetsBonk platform: {e}") def create_for_platform( self, diff --git a/src/platforms/letsbonk/__init__.py b/src/platforms/letsbonk/__init__.py index 8e494e8..60c1cfc 100644 --- a/src/platforms/letsbonk/__init__.py +++ b/src/platforms/letsbonk/__init__.py @@ -1,27 +1,15 @@ """ -LetsBonk platform registration and exports. +LetsBonk platform exports. -This module registers all LetsBonk implementations with the platform factory -and provides convenient imports for the LetsBonk platform. +This module provides convenient imports for the LetsBonk platform implementations. +Platform registration is now handled by the main platform factory. """ -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', diff --git a/src/platforms/pumpfun/__init__.py b/src/platforms/pumpfun/__init__.py index 38c8f53..bc31eb3 100644 --- a/src/platforms/pumpfun/__init__.py +++ b/src/platforms/pumpfun/__init__.py @@ -1,27 +1,15 @@ """ -Pump.Fun platform registration and exports. +Pump.Fun platform exports. -This module registers all pump.fun implementations with the platform factory -and provides convenient imports for the pump.fun platform. +This module provides convenient imports for the pump.fun platform implementations. +Platform registration is now handled by the main platform factory. """ -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', diff --git a/src/trading/base.py b/src/trading/base.py index ee7eb63..e2cb83d 100644 --- a/src/trading/base.py +++ b/src/trading/base.py @@ -12,13 +12,76 @@ from typing import Any from solders.pubkey import Pubkey -from interfaces.core import Platform - -# Import the new enhanced TokenInfo and Platform from interfaces -from interfaces.core import TokenInfo as EnhancedTokenInfo +# Import from interfaces to avoid duplication +from interfaces.core import Platform, TokenInfo -# Keep the original TokenInfo structure for backward compatibility +@dataclass +class TradeResult: + """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): + """Enhanced base interface for trading operations with platform support.""" + + @abstractmethod + 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 including platform info + """ + pass + + 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: Enhanced token information + + Returns: + List of relevant accounts (basic implementation) + """ + # Basic implementation - platform-specific traders should override this + 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) + + return accounts + + +# Legacy TokenInfo for backward compatibility (keep pump.fun specific) @dataclass class TokenInfo_Legacy: """Legacy token information structure for backward compatibility.""" @@ -73,86 +136,6 @@ class TokenInfo_Legacy: } -@dataclass -class TradeResult: - """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): - """Enhanced base interface for trading operations with platform support.""" - - @abstractmethod - 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 including platform info - """ - pass - - 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: Enhanced token information - - Returns: - List of relevant accounts (default implementation for pump.fun compatibility) - """ - # 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. @@ -246,18 +229,11 @@ def create_pump_fun_token_info( 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, - ) + # Derive creator vault if not provided (import here to avoid circular imports) + if creator_vault is None and creator: + # We can't import PumpAddresses here, so this will need to be handled elsewhere + # For now, leave it as None and let the platform implementation handle it + pass return TokenInfo( name=name, @@ -268,7 +244,7 @@ def create_pump_fun_token_info( bonding_curve=bonding_curve, associated_bonding_curve=associated_bonding_curve, user=user, - creator=creator, + creator=creator or user, creator_vault=creator_vault, **kwargs ) @@ -303,10 +279,6 @@ def create_lets_bonk_token_info( 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, @@ -317,7 +289,7 @@ def create_lets_bonk_token_info( base_vault=base_vault, quote_vault=quote_vault, user=user, - creator=creator, + creator=creator or user, **kwargs ) @@ -406,7 +378,6 @@ def validate_token_info(token_info: TokenInfo) -> bool: # Backward compatibility exports -# This allows existing imports to continue working __all__ = [ 'Platform', # Platform enum 'TokenInfo', # Enhanced TokenInfo (main export) @@ -415,13 +386,10 @@ __all__ = [ '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', ] \ No newline at end of file diff --git a/src/trading/buyer.py b/src/trading/buyer.py deleted file mode 100644 index 12dc3f4..0000000 --- a/src/trading/buyer.py +++ /dev/null @@ -1,416 +0,0 @@ -""" -Buy operations for pump.fun tokens. -""" - -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.client import SolanaClient -from core.curve import BondingCurveManager -from core.priority_fee.manager import PriorityFeeManager -from core.pubkeys import ( - LAMPORTS_PER_SOL, - TOKEN_DECIMALS, - PumpAddresses, - SystemAddresses, -) -from core.wallet import Wallet -from trading.base import TokenInfo, Trader, TradeResult -from utils.logger import get_logger - -logger = get_logger(__name__) - -# Discriminator for the buy instruction -EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" TradeResult: - """Execute buy operation. - - Args: - token_info: Token information - - Returns: - TradeResult with buy outcome - """ - try: - # 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 - # logger.info(f"EXTREME FAST Mode: Buying {token_amount} tokens.") - else: - # Regular behavior with RPC call - curve_state = await self.curve_manager.get_curve_state( - token_info.bonding_curve - ) - token_price_sol = curve_state.calculate_price() - token_amount = self.amount / token_price_sol - - # Calculate maximum SOL to spend with slippage - max_amount_lamports = int(amount_lamports * (1 + self.slippage)) - - associated_token_account = self.wallet.get_associated_token_address( - token_info.mint - ) - - tx_signature = await self._send_buy_transaction( - token_info, - associated_token_account, - token_amount, - max_amount_lamports, - ) - - 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)" - ) - - success = await self.client.confirm_transaction(tx_signature) - - if success: - # Get actual execution data from bonding curve balance changes - actual_price, actual_tokens = await self._get_actual_execution_price( - tx_signature, token_info - ) - - logger.info(f"Buy transaction confirmed: {tx_signature}") - logger.info( - f"Actual price paid to bonding curve: {actual_price:.8f} SOL per token" - ) - - return TradeResult( - success=True, - tx_signature=tx_signature, - amount=actual_tokens, # Actual tokens received - price=actual_price, # Actual price based on bonding curve SOL flow - ) - else: - return TradeResult( - success=False, - 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, error_message=str(e)) - - async def _send_buy_transaction( - self, - token_info: TokenInfo, - associated_token_account: Pubkey, - token_amount: float, - max_amount_lamports: int, - ) -> str: - """Send buy transaction. - - Args: - token_info: Token information - associated_token_account: User's token account - token_amount: Amount of tokens to buy - max_amount_lamports: Maximum SOL to spend in lamports - - Returns: - Transaction signature - - Raises: - Exception: If transaction fails after all retries - """ - accounts = [ - AccountMeta( - pubkey=PumpAddresses.GLOBAL, is_signer=False, is_writable=False - ), - AccountMeta(pubkey=PumpAddresses.FEE, is_signer=False, is_writable=True), - AccountMeta(pubkey=token_info.mint, is_signer=False, is_writable=False), - AccountMeta( - pubkey=token_info.bonding_curve, is_signer=False, is_writable=True - ), - AccountMeta( - pubkey=token_info.associated_bonding_curve, - is_signer=False, - is_writable=True, - ), - AccountMeta( - pubkey=associated_token_account, is_signer=False, is_writable=True - ), - AccountMeta(pubkey=self.wallet.pubkey, is_signer=True, is_writable=True), - AccountMeta( - pubkey=SystemAddresses.PROGRAM, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=token_info.creator_vault, is_signer=False, is_writable=True - ), - AccountMeta( - pubkey=PumpAddresses.EVENT_AUTHORITY, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=PumpAddresses.PROGRAM, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=PumpAddresses.find_global_volume_accumulator(), is_signer=False, is_writable=True - ), - AccountMeta( - pubkey=PumpAddresses.find_user_volume_accumulator(self.wallet.pubkey), is_signer=False, is_writable=True - ), - ] - - # Prepare idempotent create ATA instruction: it will not fail if ATA already exists - idempotent_ata_ix = create_idempotent_associated_token_account( - self.wallet.pubkey, - self.wallet.pubkey, - token_info.mint, - SystemAddresses.TOKEN_PROGRAM, - ) - - # Prepare buy instruction data - token_amount_raw = int(token_amount * 10**TOKEN_DECIMALS) - data = ( - EXPECTED_DISCRIMINATOR - + struct.pack(" tuple[float, float]: - """Get actual execution price from bonding curve SOL balance changes.""" - try: - client = await self.client.get_client() - - tx_response = await client.get_transaction( - tx_signature, - encoding="jsonParsed", - commitment="confirmed", - max_supported_transaction_version=0, - ) - - if not tx_response.value or not tx_response.value.transaction: - raise ValueError("Transaction not found") - - meta = tx_response.value.transaction.meta - if not meta or not meta.pre_balances or not meta.post_balances: - raise ValueError("Transaction balance data not found") - - # Get accounts - they're ParsedAccountTxStatus objects, need to extract pubkey - accounts = tx_response.value.transaction.transaction.message.account_keys - - # Find bonding curve account index in the transaction - bonding_curve_index = None - for i, account in enumerate(accounts): - # Extract pubkey from ParsedAccountTxStatus object - account_pubkey = ( - str(account.pubkey) if hasattr(account, "pubkey") else str(account) - ) - - if account_pubkey == str(token_info.bonding_curve): - bonding_curve_index = i - break - - if bonding_curve_index is None: - raise ValueError("Bonding curve not found in transaction accounts") - - pre_balance_lamports = meta.pre_balances[bonding_curve_index] - post_balance_lamports = meta.post_balances[bonding_curve_index] - - sol_sent_to_curve = ( - post_balance_lamports - pre_balance_lamports - ) / LAMPORTS_PER_SOL - - if sol_sent_to_curve <= 0: - raise ValueError(f"No SOL sent to bonding curve: {sol_sent_to_curve}") - - tokens_received = await self._get_tokens_received_from_tx( - tx_response, token_info - ) - - if tokens_received == 0: - raise ValueError("Cannot compute execution price: zero tokens received") - actual_price = sol_sent_to_curve / tokens_received - - logger.info(f"Bonding curve received: {sol_sent_to_curve:.6f} SOL") - logger.info(f"We received: {tokens_received:.6f} tokens") - logger.info(f"Actual execution price: {actual_price:.8f} SOL per token") - - return actual_price, tokens_received - - except Exception as e: - logger.warning( - f"Failed to get actual execution price from bonding curve: {e}" - ) - # Fallback to EXTREME_FAST estimate - tokens_received = ( - self.extreme_fast_token_amount - if self.extreme_fast_mode - else self.amount - / await self.curve_manager.calculate_price(token_info.bonding_curve) - ) - if tokens_received == 0: - logger.error("Fallback failed – unable to determine tokens received") - return 0.0, 0.0 - return self.amount / tokens_received, tokens_received - - async def _get_tokens_received_from_tx( - self, tx_response, token_info: TokenInfo - ) -> float: - """Extract tokens received from transaction token balance changes.""" - meta = tx_response.value.transaction.meta - - pre_token_balance = 0 - post_token_balance = 0 - - wallet_str = str(self.wallet.pubkey) - mint_str = str(token_info.mint) - - if meta.pre_token_balances: - for balance in meta.pre_token_balances: - # Convert to string for comparison - balance_owner = ( - str(balance.owner) - if hasattr(balance, "owner") - else str(getattr(balance, "owner", "")) - ) - balance_mint = ( - str(balance.mint) - if hasattr(balance, "mint") - else str(getattr(balance, "mint", "")) - ) - - if balance_owner == wallet_str and balance_mint == mint_str: - try: - # Try multiple ways to get the amount - if hasattr(balance, "ui_token_amount"): - amount_obj = balance.ui_token_amount - if ( - hasattr(amount_obj, "amount") - and amount_obj.amount is not None - ): - pre_token_balance = int(amount_obj.amount) - elif ( - hasattr(amount_obj, "ui_amount") - and amount_obj.ui_amount is not None - ): - pre_token_balance = int( - float(amount_obj.ui_amount) * (10**TOKEN_DECIMALS) - ) - except (ValueError, TypeError) as e: - logger.warning(f"Error parsing pre-token balance: {e}") - break - - # Check post-token balances - if meta.post_token_balances: - for balance in meta.post_token_balances: - # Convert to string for comparison - balance_owner = ( - str(balance.owner) - if hasattr(balance, "owner") - else str(getattr(balance, "owner", "")) - ) - balance_mint = ( - str(balance.mint) - if hasattr(balance, "mint") - else str(getattr(balance, "mint", "")) - ) - - if balance_owner == wallet_str and balance_mint == mint_str: - try: - # Try multiple ways to get the amount - if hasattr(balance, "ui_token_amount"): - amount_obj = balance.ui_token_amount - if ( - hasattr(amount_obj, "amount") - and amount_obj.amount is not None - ): - post_token_balance = int(amount_obj.amount) - elif ( - hasattr(amount_obj, "ui_amount") - and amount_obj.ui_amount is not None - ): - post_token_balance = int( - float(amount_obj.ui_amount) * (10**TOKEN_DECIMALS) - ) - except (ValueError, TypeError) as e: - logger.warning(f"Error parsing post-token balance: {e}") - break - - # Calculate tokens received - if pre_token_balance == 0 and post_token_balance > 0: - tokens_received_raw = post_token_balance - else: - tokens_received_raw = post_token_balance - pre_token_balance - - if tokens_received_raw <= 0: - logger.warning( - "Token balance search failed. Using fallback from EXTREME_FAST estimate." - ) - # Fallback: use the amount we know we bought - if self.extreme_fast_mode and self.extreme_fast_token_amount > 0: - return self.extreme_fast_token_amount - else: - logger.error("Cannot determine tokens received from transaction") - return 0.0 - - return tokens_received_raw / 10**TOKEN_DECIMALS diff --git a/src/trading/seller.py b/src/trading/seller.py deleted file mode 100644 index ac8b3e5..0000000 --- a/src/trading/seller.py +++ /dev/null @@ -1,213 +0,0 @@ -""" -Sell operations for pump.fun tokens. -""" - -import struct -from typing import Final - -from solders.instruction import AccountMeta, Instruction -from solders.pubkey import Pubkey - -from core.client import SolanaClient -from core.curve import BondingCurveManager -from core.priority_fee.manager import PriorityFeeManager -from core.pubkeys import ( - LAMPORTS_PER_SOL, - TOKEN_DECIMALS, - PumpAddresses, - SystemAddresses, -) -from core.wallet import Wallet -from trading.base import TokenInfo, Trader, TradeResult -from utils.logger import get_logger - -logger = get_logger(__name__) - -# Discriminator for the sell instruction -EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" TradeResult: - """Execute sell operation. - - Args: - token_info: Token information - - Returns: - TradeResult with sell outcome - """ - try: - # Get associated token account - associated_token_account = self.wallet.get_associated_token_address( - token_info.mint - ) - - # Get token balance - token_balance = await self.client.get_token_account_balance( - associated_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, error_message="No tokens to sell") - - # Fetch token price - curve_state = await self.curve_manager.get_curve_state( - token_info.bonding_curve - ) - token_price_sol = curve_state.calculate_price() - - logger.info(f"Price per Token: {token_price_sol:.8f} SOL") - - # Calculate minimum SOL output with slippage - amount = token_balance - expected_sol_output = float(token_balance_decimal) * float(token_price_sol) - slippage_factor = 1 - self.slippage - min_sol_output = int( - (expected_sol_output * slippage_factor) * 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" - ) - - tx_signature = await self._send_sell_transaction( - token_info, - associated_token_account, - amount, - min_sol_output, - ) - - success = await self.client.confirm_transaction(tx_signature) - - if success: - logger.info(f"Sell transaction confirmed: {tx_signature}") - return TradeResult( - success=True, - tx_signature=tx_signature, - amount=token_balance_decimal, - price=token_price_sol, - ) - else: - return TradeResult( - success=False, - 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, error_message=str(e)) - - async def _send_sell_transaction( - self, - token_info: TokenInfo, - associated_token_account: Pubkey, - token_amount: int, - min_sol_output: int, - ) -> str: - """Send sell transaction. - - Args: - mint: Token information - associated_token_account: User's token account - token_amount: Amount of tokens to sell in raw units - min_sol_output: Minimum SOL to receive in lamports - - Returns: - Transaction signature - - Raises: - Exception: If transaction fails after all retries - """ - # Prepare sell instruction accounts - accounts = [ - AccountMeta( - pubkey=PumpAddresses.GLOBAL, is_signer=False, is_writable=False - ), - AccountMeta(pubkey=PumpAddresses.FEE, is_signer=False, is_writable=True), - AccountMeta(pubkey=token_info.mint, is_signer=False, is_writable=False), - AccountMeta( - pubkey=token_info.bonding_curve, is_signer=False, is_writable=True - ), - AccountMeta( - pubkey=token_info.associated_bonding_curve, - is_signer=False, - is_writable=True, - ), - AccountMeta( - pubkey=associated_token_account, is_signer=False, is_writable=True - ), - AccountMeta(pubkey=self.wallet.pubkey, is_signer=True, is_writable=True), - AccountMeta( - pubkey=SystemAddresses.PROGRAM, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=token_info.creator_vault, - is_signer=False, - is_writable=True, - ), - AccountMeta( - pubkey=SystemAddresses.TOKEN_PROGRAM, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=PumpAddresses.EVENT_AUTHORITY, is_signer=False, is_writable=False - ), - AccountMeta( - pubkey=PumpAddresses.PROGRAM, is_signer=False, is_writable=False - ), - ] - - # Prepare sell instruction data - data = ( - EXPECTED_DISCRIMINATOR - + struct.pack(" None: """Start the trading bot and listen for new tokens.""" - logger.info("Starting pump.fun trader") - logger.info( - f"Match filter: {self.match_string if self.match_string else 'None'}" - ) - logger.info( - f"Creator filter: {self.bro_address if self.bro_address else 'None'}" - ) + logger.info(f"Starting Universal Trader for {self.platform.value}") + logger.info(f"Match filter: {self.match_string if self.match_string else 'None'}") + logger.info(f"Creator filter: {self.bro_address if self.bro_address else 'None'}") logger.info(f"Marry mode: {self.marry_mode}") logger.info(f"YOLO mode: {self.yolo_mode}") logger.info(f"Exit strategy: {self.exit_strategy}") + if self.exit_strategy == "tp_sl": - logger.info( - f"Take profit: {self.take_profit_percentage * 100 if self.take_profit_percentage else 'None'}%" - ) - logger.info( - f"Stop loss: {self.stop_loss_percentage * 100 if self.stop_loss_percentage else 'None'}%" - ) - logger.info( - f"Max hold time: {self.max_hold_time if self.max_hold_time else 'None'} seconds" - ) + logger.info(f"Take profit: {self.take_profit_percentage * 100 if self.take_profit_percentage else 'None'}%") + logger.info(f"Stop loss: {self.stop_loss_percentage * 100 if self.stop_loss_percentage else 'None'}%") + logger.info(f"Max hold time: {self.max_hold_time if self.max_hold_time else 'None'} seconds") + logger.info(f"Max token age: {self.max_token_age} seconds") try: @@ -262,22 +218,16 @@ class PumpTrader: # Choose operating mode based on yolo_mode if not self.yolo_mode: # Single token mode: process one token and exit - logger.info( - "Running in single token mode - will process one token and exit" - ) + logger.info("Running in single token mode - will process one token and exit") token_info = await self._wait_for_token() if token_info: await self._handle_token(token_info) logger.info("Finished processing single token. Exiting...") else: - logger.info( - f"No suitable token found within timeout period ({self.token_wait_timeout}s). Exiting..." - ) + logger.info(f"No suitable token found within timeout period ({self.token_wait_timeout}s). Exiting...") else: # Continuous mode: process tokens until interrupted - logger.info( - "Running in continuous mode - will process tokens until interrupted" - ) + logger.info("Running in continuous mode - will process tokens until interrupted") processor_task = asyncio.create_task(self._process_token_queue()) try: @@ -300,7 +250,7 @@ class PumpTrader: finally: await self._cleanup_resources() - logger.info("Pump trader has shut down") + logger.info("Universal Trader has shut down") async def _wait_for_token(self) -> TokenInfo | None: """Wait for a single token to be detected. @@ -334,16 +284,12 @@ class PumpTrader: # Wait for a token with a timeout try: - logger.info( - f"Waiting for a suitable token (timeout: {self.token_wait_timeout}s)..." - ) + logger.info(f"Waiting for a suitable token (timeout: {self.token_wait_timeout}s)...") await asyncio.wait_for(token_found.wait(), timeout=self.token_wait_timeout) logger.info(f"Found token: {found_token.symbol} ({found_token.mint})") return found_token except TimeoutError: - logger.info( - f"Timed out after waiting {self.token_wait_timeout}s for a token" - ) + logger.info(f"Timed out after waiting {self.token_wait_timeout}s for a token") return None finally: listener_task.cancel() @@ -391,7 +337,7 @@ class PumpTrader: self.token_timestamps[token_key] = monotonic() await self.token_queue.put(token_info) - logger.info(f"Queued new token: {token_info.symbol} ({token_info.mint})") + logger.info(f"Queued new token: {token_info.symbol} ({token_info.mint}) on {token_info.platform.value}") async def _process_token_queue(self) -> None: """Continuously process tokens from the queue, only if they're fresh.""" @@ -402,21 +348,15 @@ class PumpTrader: # Check if token is still "fresh" current_time = monotonic() - token_age = current_time - self.token_timestamps.get( - token_key, current_time - ) + token_age = current_time - self.token_timestamps.get(token_key, current_time) if token_age > self.max_token_age: - logger.info( - f"Skipping token {token_info.symbol} - too old ({token_age:.1f}s > {self.max_token_age}s)" - ) + logger.info(f"Skipping token {token_info.symbol} - too old ({token_age:.1f}s > {self.max_token_age}s)") continue self.processed_tokens.add(token_key) - logger.info( - f"Processing fresh token: {token_info.symbol} (age: {token_age:.1f}s)" - ) + logger.info(f"Processing fresh token: {token_info.symbol} (age: {token_age:.1f}s)") await self._handle_token(token_info) except asyncio.CancelledError: @@ -435,19 +375,19 @@ class PumpTrader: token_info: Token information """ try: - # Wait for bonding curve to stabilize (unless in extreme fast mode) + # Validate that token is for our platform + if token_info.platform != self.platform: + logger.warning(f"Token platform mismatch: expected {self.platform.value}, got {token_info.platform.value}") + return + + # Wait for pool/curve to stabilize (unless in extreme fast mode) if not self.extreme_fast_mode: - # Save token info to file - # await self._save_token_info(token_info) - logger.info( - f"Waiting for {self.wait_time_after_creation} seconds for the bonding curve to stabilize..." - ) + await self._save_token_info(token_info) + logger.info(f"Waiting for {self.wait_time_after_creation} seconds for the pool/curve to stabilize...") await asyncio.sleep(self.wait_time_after_creation) # Buy token - logger.info( - f"Buying {self.buy_amount:.6f} SOL worth of {token_info.symbol}..." - ) + logger.info(f"Buying {self.buy_amount:.6f} SOL worth of {token_info.symbol} on {token_info.platform.value}...") buy_result: TradeResult = await self.buyer.execute(token_info) if buy_result.success: @@ -457,31 +397,21 @@ class PumpTrader: # Only wait for next token in yolo mode if self.yolo_mode: - logger.info( - f"YOLO mode enabled. Waiting {self.wait_time_before_new_token} seconds before looking for next token..." - ) + logger.info(f"YOLO mode enabled. Waiting {self.wait_time_before_new_token} seconds before looking for next token...") await asyncio.sleep(self.wait_time_before_new_token) except Exception as e: logger.error(f"Error handling token {token_info.symbol}: {e!s}") - async def _handle_successful_buy( - self, token_info: TokenInfo, buy_result: TradeResult - ) -> None: + async def _handle_successful_buy(self, token_info: TokenInfo, buy_result: TradeResult) -> None: """Handle successful token purchase. Args: token_info: Token information buy_result: The result of the buy operation """ - logger.info(f"Successfully bought {token_info.symbol}") - self._log_trade( - "buy", - token_info, - buy_result.price, # type: ignore - buy_result.amount, # type: ignore - buy_result.tx_signature, - ) + logger.info(f"Successfully bought {token_info.symbol} on {token_info.platform.value}") + self._log_trade("buy", token_info, buy_result.price, buy_result.amount, buy_result.tx_signature) self.traded_mints.add(token_info.mint) # Choose exit strategy @@ -495,9 +425,7 @@ class PumpTrader: else: logger.info("Marry mode enabled. Skipping sell operation.") - async def _handle_failed_buy( - self, token_info: TokenInfo, buy_result: TradeResult - ) -> None: + async def _handle_failed_buy(self, token_info: TokenInfo, buy_result: TradeResult) -> None: """Handle failed token purchase. Args: @@ -516,9 +444,7 @@ class PumpTrader: self.cleanup_force_close_with_burn, ) - async def _handle_tp_sl_exit( - self, token_info: TokenInfo, buy_result: TradeResult - ) -> None: + async def _handle_tp_sl_exit(self, token_info: TokenInfo, buy_result: TradeResult) -> None: """Handle take profit/stop loss exit strategy. Args: @@ -529,8 +455,8 @@ class PumpTrader: position = Position.create_from_buy_result( mint=token_info.mint, symbol=token_info.symbol, - entry_price=buy_result.price, # type: ignore - quantity=buy_result.amount, # type: ignore + entry_price=buy_result.price, + quantity=buy_result.amount, take_profit_percentage=self.take_profit_percentage, stop_loss_percentage=self.stop_loss_percentage, max_hold_time=self.max_hold_time, @@ -559,13 +485,7 @@ class PumpTrader: if sell_result.success: logger.info(f"Successfully sold {token_info.symbol}") - self._log_trade( - "sell", - token_info, - sell_result.price, # type: ignore - sell_result.amount, # type: ignore - sell_result.tx_signature, - ) + self._log_trade("sell", token_info, sell_result.price, sell_result.amount, sell_result.tx_signature) # Close ATA if enabled await handle_cleanup_after_sell( self.solana_client, @@ -577,29 +497,25 @@ class PumpTrader: self.cleanup_force_close_with_burn, ) else: - logger.error( - f"Failed to sell {token_info.symbol}: {sell_result.error_message}" - ) + logger.error(f"Failed to sell {token_info.symbol}: {sell_result.error_message}") - async def _monitor_position_until_exit( - self, token_info: TokenInfo, position: Position - ) -> None: + async def _monitor_position_until_exit(self, token_info: TokenInfo, position: Position) -> None: """Monitor a position until exit conditions are met. Args: token_info: Token information position: Position to monitor """ - logger.info( - f"Starting position monitoring (check interval: {self.price_check_interval}s)" - ) + logger.info(f"Starting position monitoring (check interval: {self.price_check_interval}s)") + + # Get pool address for price monitoring + pool_address = self._get_pool_address(token_info) + curve_manager = self.platform_implementations.curve_manager while position.is_active: try: - # Get current price from bonding curve - current_price = await self.curve_manager.calculate_price( - token_info.bonding_curve - ) + # Get current price from pool/curve + current_price = await curve_manager.calculate_price(pool_address) # Check if position should be exited should_exit, exit_reason = position.should_exit(current_price) @@ -610,33 +526,21 @@ class PumpTrader: # Log PnL before exit pnl = position.get_pnl(current_price) - logger.info( - f"Position PnL: {pnl['price_change_pct']:.2f}% ({pnl['unrealized_pnl_sol']:.6f} SOL)" - ) + logger.info(f"Position PnL: {pnl['price_change_pct']:.2f}% ({pnl['unrealized_pnl_sol']:.6f} SOL)") # Execute sell sell_result = await self.seller.execute(token_info) if sell_result.success: # Close position with actual exit price - position.close_position(sell_result.price, exit_reason) # type: ignore + position.close_position(sell_result.price, exit_reason) - logger.info( - f"Successfully exited position: {exit_reason.value}" - ) - self._log_trade( - "sell", - token_info, - sell_result.price, # type: ignore - sell_result.amount, # type: ignore - sell_result.tx_signature, - ) + logger.info(f"Successfully exited position: {exit_reason.value}") + self._log_trade("sell", token_info, sell_result.price, sell_result.amount, sell_result.tx_signature) # Log final PnL final_pnl = position.get_pnl() - logger.info( - f"Final PnL: {final_pnl['price_change_pct']:.2f}% ({final_pnl['unrealized_pnl_sol']:.6f} SOL)" - ) + logger.info(f"Final PnL: {final_pnl['price_change_pct']:.2f}% ({final_pnl['unrealized_pnl_sol']:.6f} SOL)") # Close ATA if enabled await handle_cleanup_after_sell( @@ -649,27 +553,40 @@ class PumpTrader: self.cleanup_force_close_with_burn, ) else: - logger.error( - f"Failed to exit position: {sell_result.error_message}" - ) + logger.error(f"Failed to exit position: {sell_result.error_message}") # Keep monitoring in case sell can be retried break else: # Log current status pnl = position.get_pnl(current_price) - logger.debug( - f"Position status: {current_price:.8f} SOL ({pnl['price_change_pct']:+.2f}%)" - ) + logger.debug(f"Position status: {current_price:.8f} SOL ({pnl['price_change_pct']:+.2f}%)") # Wait before next price check await asyncio.sleep(self.price_check_interval) except Exception as e: logger.error(f"Error monitoring position: {e}") - await asyncio.sleep( - self.price_check_interval - ) # Continue monitoring despite errors + await asyncio.sleep(self.price_check_interval) # Continue monitoring despite errors + + def _get_pool_address(self, token_info: TokenInfo) -> Pubkey: + """Get the pool/curve address for price monitoring. + + Args: + token_info: Token information + + Returns: + Pool/curve address + """ + address_provider = self.platform_implementations.address_provider + + 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) async def _save_token_info(self, token_info: TokenInfo) -> None: """Save token information to a file. @@ -681,21 +598,32 @@ class PumpTrader: os.makedirs("trades", exist_ok=True) file_name = os.path.join("trades", f"{token_info.mint}.txt") + # Convert to dictionary for saving + token_dict = { + "name": token_info.name, + "symbol": token_info.symbol, + "uri": token_info.uri, + "mint": str(token_info.mint), + "platform": token_info.platform.value, + "user": str(token_info.user) if token_info.user else None, + "creator": str(token_info.creator) if token_info.creator else None, + # Platform-specific fields + "bonding_curve": str(token_info.bonding_curve) if token_info.bonding_curve else None, + "associated_bonding_curve": str(token_info.associated_bonding_curve) if token_info.associated_bonding_curve else None, + "creator_vault": str(token_info.creator_vault) if token_info.creator_vault else None, + "pool_state": str(token_info.pool_state) if token_info.pool_state else None, + "base_vault": str(token_info.base_vault) if token_info.base_vault else None, + "quote_vault": str(token_info.quote_vault) if token_info.quote_vault else None, + } + with open(file_name, "w") as file: - file.write(json.dumps(token_info.to_dict(), indent=2)) + file.write(json.dumps(token_dict, indent=2)) logger.info(f"Token information saved to {file_name}") except Exception as e: logger.error(f"Failed to save token information: {e!s}") - def _log_trade( - self, - action: str, - token_info: TokenInfo, - price: float, - amount: float, - tx_hash: str | None, - ) -> None: + def _log_trade(self, action: str, token_info: TokenInfo, price: float, amount: float, tx_hash: str | None) -> None: """Log trade information. Args: @@ -711,6 +639,7 @@ class PumpTrader: log_entry = { "timestamp": datetime.utcnow().isoformat(), "action": action, + "platform": token_info.platform.value, "token_address": str(token_info.mint), "symbol": token_info.symbol, "price": price, @@ -722,3 +651,7 @@ class PumpTrader: log_file.write(json.dumps(log_entry) + "\n") except Exception as e: logger.error(f"Failed to log trade information: {e!s}") + + +# Backward compatibility alias +PumpTrader = UniversalTrader # Legacy name for backward compatibility \ No newline at end of file