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

feat(core): add event idl parser

Browse Source
This commit is contained in:
smypmsa
2025-08-02 19:19:20 +00:00
parent bae03c3fab
commit 9b6563cecc
5 changed files with 385 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/platforms/pumpfun/event_parser.py
+111 -67
View File
@@ -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("<Q", self._create_discriminator_bytes)[0]
event_discriminators = self._idl_parser.get_event_discriminators()
self._create_event_discriminator_bytes = event_discriminators["CreateEvent"]
self._create_event_discriminator = struct.unpack("<Q", self._create_event_discriminator_bytes)[0]
logger.info("Pump.Fun event parser initialized with injected IDL parser")
instruction_discriminators = self._idl_parser.get_instruction_discriminators()
self._create_instruction_discriminator_bytes = instruction_discriminators["create"]
self._create_instruction_discriminator = struct.unpack("<Q", self._create_instruction_discriminator_bytes)[0]
logger.info("Pump.Fun event parser initialized with IDL-based event and instruction parsing")
logger.info(f"CreateEvent discriminator: {self._create_event_discriminator_bytes.hex()}")
logger.info(f"create instruction discriminator: {self._create_instruction_discriminator_bytes.hex()}")
@property
def platform(self) -> 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("<Q", discriminator)[0]
# Debug: Print all discriminators
print(f"🔍 Program data discriminator found: {discriminator.hex()} (int: {discriminator_int})")
print(f"🎯 Expected CreateEvent discriminator: {self._create_event_discriminator_bytes.hex()} (int: {self._create_event_discriminator})")
print(f"🛠️ Expected create instruction discriminator: {self._create_instruction_discriminator_bytes.hex()} (int: {self._create_instruction_discriminator})")
# Check if it matches CreateEvent discriminator
decoded_event = None
if discriminator_int == self._create_event_discriminator:
print("✅ Matches CreateEvent discriminator - proceeding with event parsing")
# Use IDL parser to decode the event data
decoded_event = self._idl_parser.decode_event_data(decoded_data, "CreateEvent")
if not decoded_event or decoded_event['event_name'] != 'CreateEvent':
print("❌ IDL parser failed to decode as CreateEvent")
continue
else:
print(f"⚠️ Discriminator mismatch - found {discriminator_int}, expected {self._create_event_discriminator}")
# Try to decode anyway (maybe the discriminator calculation is wrong)
print("🔄 Trying to decode anyway...")
decoded_event = self._idl_parser.decode_event_data(decoded_data, "CreateEvent")
if not decoded_event or decoded_event['event_name'] != 'CreateEvent':
print("❌ IDL parser also failed with mismatched discriminator")
continue
print("✅ IDL parser succeeded despite discriminator mismatch!")
print(f"✅ Successfully decoded event: {decoded_event.get('event_name', 'Unknown')}")
print(f"🔍 Event fields: {list(decoded_event.get('fields', {}).keys())}")
fields = decoded_event.get('fields', {})
if not fields:
print("❌ No fields found in decoded event")
continue
# Convert to TokenInfo
mint = Pubkey.from_string(fields["mint"])
bonding_curve = Pubkey.from_string(fields["bonding_curve"])
user = Pubkey.from_string(fields["user"])
creator = Pubkey.from_string(fields["creator"])
# Derive additional addresses
associated_bonding_curve = self._derive_associated_bonding_curve(mint, bonding_curve)
creator_vault = self._derive_creator_vault(creator)
return TokenInfo(
name=fields["name"],
symbol=fields["symbol"],
uri=fields["uri"],
mint=mint,
platform=Platform.PUMP_FUN,
bonding_curve=bonding_curve,
associated_bonding_curve=associated_bonding_curve,
user=user,
creator=creator,
creator_vault=creator_vault,
creation_timestamp=monotonic(),
)
except Exception as e:
logger.debug(f"Failed to decode log data: {e}")
continue
return None
except Exception as e:
logger.debug(f"Failed to parse token creation from logs: {e}")
return None
def parse_token_creation_from_instruction(
self,
@@ -95,7 +172,7 @@ class PumpFunEventParser(EventParser):
Returns:
TokenInfo if token creation found, None otherwise
"""
if not instruction_data.startswith(self._create_discriminator_bytes):
if not instruction_data.startswith(self._create_instruction_discriminator_bytes):
return None
try:
@@ -177,7 +254,6 @@ class PumpFunEventParser(EventParser):
if bytes(program_id) != bytes(self.get_program_id()):
continue
# Process instruction data
token_info = self.parse_token_creation_from_instruction(
ix.data, ix.accounts, msg.account_keys
)
@@ -204,7 +280,15 @@ class PumpFunEventParser(EventParser):
Returns:
List of discriminator bytes to match
"""
return [self._create_discriminator_bytes]
return [self._create_instruction_discriminator_bytes]
def get_event_discriminators(self) -> 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("<Q", ix_data[:8])[0]
if discriminator == self._create_discriminator:
if discriminator == self._create_instruction_discriminator:
# Token creation should have substantial data and many accounts
if len(ix_data) <= 8 or len(ix.accounts) < 10:
continue
@@ -290,7 +374,7 @@ class PumpFunEventParser(EventParser):
if len(ix_data) >= 8:
discriminator = struct.unpack("<Q", ix_data[:8])[0]
if discriminator == self._create_discriminator:
if discriminator == self._create_instruction_discriminator:
if len(ix_data) <= 8 or len(ix["accounts"]) < 10:
continue
@@ -315,47 +399,7 @@ class PumpFunEventParser(EventParser):
except Exception as e:
logger.debug(f"Failed to parse block data: {e}")
return None
def _parse_create_instruction_data(self, data: bytes) -> 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("<Q", data[:8])[0]
if discriminator != self._create_discriminator:
return None
try:
# Use IDL parser to decode the instruction data
# For log data parsing, we need to create dummy account info
dummy_accounts = list(range(20)) # Assume enough accounts
dummy_account_keys = [b'\x00' * 32] * 20 # Dummy keys
decoded = self._idl_parser.decode_instruction(data, dummy_account_keys, dummy_accounts)
if not decoded or decoded['instruction_name'] != 'create':
return None
# Extract the arguments
args = decoded.get('args', {})
return {
"name": args.get("name", ""),
"symbol": args.get("symbol", ""),
"uri": args.get("uri", ""),
"creator": args.get("creator", ""),
}
except Exception as e:
logger.debug(f"Failed to parse instruction data with IDL: {e}")
return None
def _derive_creator_vault(self, creator: Pubkey) -> Pubkey:
"""Derive the creator vault for a creator.
src/utils/idl_manager.py
+155 -2
View File
@@ -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)
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)
src/utils/idl_parser.py
+117 -3
View File
@@ -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', []):