diff --git a/bots/bot-sniper-2-logs.yaml b/bots/bot-sniper-2-logs.yaml index 47780b8..46562e5 100644 --- a/bots/bot-sniper-2-logs.yaml +++ b/bots/bot-sniper-2-logs.yaml @@ -8,7 +8,7 @@ rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}" wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}" private_key: "${SOLANA_PRIVATE_KEY}" -enabled: false # You can turn off the bot w/o removing its config +enabled: true # You can turn off the bot w/o removing its config separate_process: true # Options: "pump_fun" (default), "lets_bonk" diff --git a/bots/bot-sniper-3-blocks.yaml b/bots/bot-sniper-3-blocks.yaml index f3e5a4e..f2acd0f 100644 --- a/bots/bot-sniper-3-blocks.yaml +++ b/bots/bot-sniper-3-blocks.yaml @@ -8,7 +8,7 @@ rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}" wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}" private_key: "${SOLANA_PRIVATE_KEY}" -enabled: true # You can turn off the bot w/o removing its config +enabled: false # You can turn off the bot w/o removing its config separate_process: true # Options: "pump_fun" (default), "lets_bonk" diff --git a/src/platforms/pumpfun/event_parser.py b/src/platforms/pumpfun/event_parser.py index b5a7c29..af5417f 100644 --- a/src/platforms/pumpfun/event_parser.py +++ b/src/platforms/pumpfun/event_parser.py @@ -2,7 +2,7 @@ Pump.Fun implementation of EventParser interface. This module parses pump.fun-specific token creation events from various sources -by implementing the EventParser interface with IDL-based parsing. +by implementing the EventParser interface with IDL-based event parsing. """ import base64 @@ -23,8 +23,8 @@ logger = get_logger(__name__) class PumpFunEventParser(EventParser): - """Pump.Fun implementation of EventParser interface with IDL-based parsing.""" - + """Pump.Fun implementation of EventParser interface with IDL-based event parsing.""" + def __init__(self, idl_parser: IDLParser): """Initialize pump.fun event parser with injected IDL parser. @@ -33,12 +33,17 @@ class PumpFunEventParser(EventParser): """ self._idl_parser = idl_parser - # Get discriminators from injected IDL parser - discriminators = self._idl_parser.get_instruction_discriminators() - self._create_discriminator_bytes = discriminators["create"] - self._create_discriminator = struct.unpack(" Platform: @@ -50,7 +55,7 @@ class PumpFunEventParser(EventParser): logs: list[str], signature: str ) -> TokenInfo | None: - """Parse token creation from pump.fun transaction logs. + """Parse token creation from pump.fun transaction logs using IDL instruction parsing. Args: logs: List of log strings from transaction @@ -59,25 +64,97 @@ class PumpFunEventParser(EventParser): Returns: TokenInfo if token creation found, None otherwise """ - # Check if this is a token creation + # Check if this is a token creation transaction if not any("Program log: Instruction: Create" in log for log in logs): return None - # Skip swaps and other operations + # 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) - return self._parse_create_instruction_data(decoded_data) - except Exception: - continue - - return None + # Look for event data in the logs (CreateEvent data!) + try: + for log in logs: + if "Program data:" in log: + try: + # Extract base64 encoded event data + encoded_data = log.split("Program data: ")[1].strip() + decoded_data = base64.b64decode(encoded_data) + + # Parse as event data (CreateEvent) + # The "Program data:" in logs contains event data, not instruction data + if len(decoded_data) < 8: + continue + + # Check discriminator from program data + discriminator = decoded_data[:8] + discriminator_int = struct.unpack(" list[bytes]: + """Get event discriminators for token creation. + + Returns: + List of event discriminator bytes to match + """ + return [self._create_event_discriminator_bytes] def parse_token_creation_from_block(self, block_data: dict) -> TokenInfo | None: """Parse token creation from block data (for block listener). @@ -244,7 +328,7 @@ class PumpFunEventParser(EventParser): if len(ix_data) >= 8: discriminator = struct.unpack("= 8: discriminator = struct.unpack(" dict | None: - """Parse the create instruction data from pump.fun using injected IDL parser. - 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: """Derive the creator vault for a creator. diff --git a/src/utils/idl_manager.py b/src/utils/idl_manager.py index c5ed998..ec82af3 100644 --- a/src/utils/idl_manager.py +++ b/src/utils/idl_manager.py @@ -6,6 +6,7 @@ duplicate loading across multiple platform implementation classes. """ import os +from typing import Any from interfaces.core import Platform from utils.idl_parser import IDLParser @@ -68,7 +69,9 @@ class IDLManager: parser = IDLParser(idl_path, verbose=verbose) self._parsers[platform] = parser - logger.info(f"IDL parser loaded for {platform.value} with {len(parser.get_instruction_names())} instructions") + instruction_count = len(parser.get_instruction_names()) + event_count = len(parser.get_event_names()) + logger.info(f"IDL parser loaded for {platform.value} with {instruction_count} instructions and {event_count} events") return parser @@ -119,6 +122,10 @@ class IDLManager: else: logger.debug(f"IDL parser for {platform.value} already loaded") + # -------------------------------------------------------------------------- + # Instruction-related convenience methods + # -------------------------------------------------------------------------- + def get_instruction_discriminators(self, platform: Platform) -> dict[str, bytes]: """Get instruction discriminators for a platform. @@ -142,6 +149,110 @@ class IDLManager: """ parser = self.get_parser(platform) return parser.get_instruction_names() + + # -------------------------------------------------------------------------- + # Event-related convenience methods + # -------------------------------------------------------------------------- + + def get_event_discriminators(self, platform: Platform) -> dict[str, bytes]: + """Get event discriminators for a platform. + + Args: + platform: Platform to get event discriminators for + + Returns: + Dictionary mapping event names to discriminator bytes + """ + parser = self.get_parser(platform) + return parser.get_event_discriminators() + + def get_event_names(self, platform: Platform) -> list[str]: + """Get available event names for a platform. + + Args: + platform: Platform to get event names for + + Returns: + List of event names + """ + parser = self.get_parser(platform) + return parser.get_event_names() + + def decode_event_from_logs(self, platform: Platform, logs: list[str], event_name: str | None = None) -> dict | None: + """Decode event data from transaction logs for a platform. + + Args: + platform: Platform to use for decoding + logs: List of log strings from transaction + event_name: Optional specific event name to look for + + Returns: + Decoded event data if found, None otherwise + """ + parser = self.get_parser(platform) + return parser.find_event_in_logs(logs, event_name) + + def decode_event_data(self, platform: Platform, event_data: bytes, event_name: str | None = None) -> dict | None: + """Decode raw event data for a platform. + + Args: + platform: Platform to use for decoding + event_data: Raw event data bytes + event_name: Optional event name to decode as + + Returns: + Decoded event data if successful, None otherwise + """ + parser = self.get_parser(platform) + return parser.decode_event_data(event_data, event_name) + + # -------------------------------------------------------------------------- + # Platform information methods + # -------------------------------------------------------------------------- + + def get_platform_capabilities(self, platform: Platform) -> dict[str, Any]: + """Get comprehensive capability information for a platform. + + Args: + platform: Platform to get capabilities for + + Returns: + Dictionary with platform capabilities + """ + if not self.has_idl_support(platform): + return { + "platform": platform.value, + "has_idl_support": False, + "instructions": [], + "events": [], + "instruction_count": 0, + "event_count": 0, + } + + try: + parser = self.get_parser(platform) + instruction_names = parser.get_instruction_names() + event_names = parser.get_event_names() + + return { + "platform": platform.value, + "has_idl_support": True, + "instructions": instruction_names, + "events": event_names, + "instruction_count": len(instruction_names), + "event_count": len(event_names), + } + except Exception as e: + logger.error(f"Failed to get capabilities for {platform.value}: {e}") + return { + "platform": platform.value, + "has_idl_support": False, + "error": str(e), + "instructions": [], + "events": [], + "instruction_count": 0, + "event_count": 0, + } # Global IDL manager instance @@ -192,4 +303,46 @@ def preload_platform_idl(platform: Platform, verbose: bool = False) -> None: platform: Platform to preload parser for verbose: Whether to enable verbose logging """ - get_idl_manager().preload_parser(platform, verbose) \ No newline at end of file + get_idl_manager().preload_parser(platform, verbose) + + +# -------------------------------------------------------------------------- +# Convenience functions for event handling +# -------------------------------------------------------------------------- + +def get_event_discriminators(platform: Platform) -> dict[str, bytes]: + """Convenience function to get event discriminators for a platform. + + Args: + platform: Platform to get event discriminators for + + Returns: + Dictionary mapping event names to discriminator bytes + """ + return get_idl_manager().get_event_discriminators(platform) + + +def get_event_names(platform: Platform) -> list[str]: + """Convenience function to get event names for a platform. + + Args: + platform: Platform to get event names for + + Returns: + List of event names + """ + return get_idl_manager().get_event_names(platform) + + +def decode_event_from_logs(platform: Platform, logs: list[str], event_name: str | None = None) -> dict | None: + """Convenience function to decode events from logs. + + Args: + platform: Platform to use for decoding + logs: List of log strings from transaction + event_name: Optional specific event name to look for + + Returns: + Decoded event data if found, None otherwise + """ + return get_idl_manager().decode_event_from_logs(platform, logs, event_name) \ No newline at end of file diff --git a/src/utils/idl_parser.py b/src/utils/idl_parser.py index 272f580..ffcc272 100644 --- a/src/utils/idl_parser.py +++ b/src/utils/idl_parser.py @@ -1,8 +1,9 @@ """ IDL Parser module for Solana programs. -Provides functionality to load and parse Anchor IDL files and decode instruction data. +Provides functionality to load and parse Anchor IDL files and decode instruction data and events. """ +import base64 import json import struct from typing import Any @@ -17,7 +18,7 @@ ENUM_DISCRIMINATOR_SIZE = 1 class IDLParser: - """Parser for automatically decoding instructions using IDL definitions.""" + """Parser for automatically decoding instructions and events using IDL definitions.""" # A single source of truth for primitive type information, mapping the type name # to its struct format character and size in bytes. @@ -48,14 +49,16 @@ class IDLParser: with open(idl_path) as f: self.idl = json.load(f) self.instructions: dict[bytes, dict[str, Any]] = {} + self.events: dict[bytes, dict[str, Any]] = {} self.types: dict[str, dict[str, Any]] = {} self.instruction_min_sizes: dict[bytes, int] = {} self._build_instruction_map() + self._build_event_map() self._build_type_map() self._calculate_instruction_sizes() # -------------------------------------------------------------------------- - # Public Methods (External API) + # Public Methods (External API) - Instructions # -------------------------------------------------------------------------- def get_instruction_discriminators(self) -> dict[str, bytes]: @@ -132,6 +135,108 @@ class IDLParser: 'accounts': account_info } + # -------------------------------------------------------------------------- + # Public Methods (External API) - Events + # -------------------------------------------------------------------------- + + def get_event_discriminators(self) -> dict[str, bytes]: + """Get a mapping of event names to their discriminators.""" + return {event['name']: disc for disc, event in self.events.items()} + + def get_event_names(self) -> list[str]: + """Get a list of all available event names.""" + return [event['name'] for event in self.events.values()] + + def decode_event_data(self, event_data: bytes, event_name: str | None = None) -> dict[str, Any] | None: + """ + Decode event data using IDL event definitions. + + Args: + event_data: Raw event data bytes (typically from base64 decoded log data) + event_name: Optional event name to decode as. If None, will try to match discriminator. + + Returns: + Decoded event data as a dictionary, or None if decoding fails. + """ + if len(event_data) < DISCRIMINATOR_SIZE: + return None + + discriminator = event_data[:DISCRIMINATOR_SIZE] + + # If event_name provided, validate it matches the discriminator + if event_name: + event_discriminators = self.get_event_discriminators() + if event_name not in event_discriminators: + if self.verbose: + print(f"Unknown event name: {event_name}") + return None + if event_discriminators[event_name] != discriminator: + if self.verbose: + print(f"Event discriminator mismatch for {event_name}") + return None + event_def = self.events[discriminator] + else: + # Try to find event by discriminator + if discriminator not in self.events: + if self.verbose: + print(f"Unknown event discriminator: {discriminator.hex()}") + return None + event_def = self.events[discriminator] + + # Decode event fields + try: + event_fields = {} + data_part = event_data[DISCRIMINATOR_SIZE:] + decode_offset = 0 + + for field in event_def.get('fields', []): + value, decode_offset = self._decode_type(data_part, decode_offset, field['type']) + event_fields[field['name']] = value + + return { + 'event_name': event_def['name'], + 'fields': event_fields + } + + except Exception as e: + if self.verbose: + print(f"❌ Error decoding event {event_def['name']}: {e}") + return None + + def find_event_in_logs(self, logs: list[str], target_event_name: str | None = None) -> dict[str, Any] | None: + """ + Find and decode event data from transaction logs. + + Args: + logs: List of log strings from a transaction + target_event_name: Optional specific event name to look for + + Returns: + Decoded event data if found, None otherwise + """ + for log in logs: + if "Program data:" in log: + try: + # Extract base64 encoded data + encoded_data = log.split("Program data: ")[1].strip() + decoded_data = base64.b64decode(encoded_data) + + # Try to decode as event + event_data = self.decode_event_data(decoded_data, target_event_name) + if event_data: + return event_data + + except Exception as e: + if self.verbose: + print(f"Failed to decode log data: {e}") + continue + + return None + + # -------------------------------------------------------------------------- + # Public Methods (External API) - Account Data + # -------------------------------------------------------------------------- + def decode_account_data(self, account_data: bytes, account_type_name: str, skip_discriminator: bool = True) -> dict[str, Any] | None: """ Decode account data using a specific account type from the IDL. @@ -179,6 +284,15 @@ class IDLParser: discriminator = bytes(instruction['discriminator']) self.instructions[discriminator] = instruction + def _build_event_map(self): + """Build a map of discriminators to event definitions.""" + for event in self.idl.get('events', []): + # The discriminator from the JSON IDL is a list of u8 integers. + discriminator = bytes(event['discriminator']) + self.events[discriminator] = event + if self.verbose: + print(f"📅 Loaded event: {event['name']} with discriminator {discriminator.hex()}") + def _build_type_map(self): """Build a map of type names to their definitions.""" for type_def in self.idl.get('types', []):