mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-07-27 23:37:45 +00:00
feat(letsbonk): add idl manager
This commit is contained in:
smypmsa
@@ -2,7 +2,7 @@
|
||||
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.
|
||||
platform-specific implementations of the trading interfaces with IDL support.
|
||||
"""
|
||||
|
||||
from dataclasses import dataclass
|
||||
@@ -16,6 +16,10 @@ from interfaces.core import (
|
||||
InstructionBuilder,
|
||||
Platform,
|
||||
)
|
||||
from utils.idl_manager import get_idl_manager, has_idl_support
|
||||
from utils.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
@@ -64,7 +68,7 @@ class PlatformRegistry:
|
||||
client: SolanaClient,
|
||||
**kwargs: Any
|
||||
) -> PlatformImplementations:
|
||||
"""Create platform implementation instances.
|
||||
"""Create platform implementation instances with IDL support.
|
||||
|
||||
Args:
|
||||
platform: Platform to create implementations for
|
||||
@@ -89,11 +93,29 @@ class PlatformRegistry:
|
||||
|
||||
impl_classes = self._implementations[platform]
|
||||
|
||||
# Create instances - only curve_manager needs client
|
||||
# Check if platform has IDL support and prepare IDL parser
|
||||
idl_parser = None
|
||||
if has_idl_support(platform):
|
||||
try:
|
||||
idl_manager = get_idl_manager()
|
||||
idl_parser = idl_manager.get_parser(platform, verbose=kwargs.get('verbose_idl', False))
|
||||
logger.info(f"IDL parser loaded for {platform.value} platform implementations")
|
||||
except Exception as e:
|
||||
logger.warning(f"Failed to load IDL parser for {platform.value}: {e}")
|
||||
|
||||
# Create instances - pass IDL parser to classes that need it
|
||||
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']()
|
||||
|
||||
# For platforms with IDL support, pass the parser to relevant classes
|
||||
if idl_parser and platform in [Platform.LETS_BONK]: # Add other IDL-supported platforms here
|
||||
instruction_builder = impl_classes['instruction_builder'](idl_parser=idl_parser)
|
||||
curve_manager = impl_classes['curve_manager'](client, idl_parser=idl_parser)
|
||||
event_parser = impl_classes['event_parser'](idl_parser=idl_parser)
|
||||
else:
|
||||
# Fallback for platforms without IDL support (like pump.fun)
|
||||
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,
|
||||
@@ -138,10 +160,25 @@ class PlatformRegistry:
|
||||
True if platform is registered, False otherwise
|
||||
"""
|
||||
return platform in self._implementations
|
||||
|
||||
def clear_implementation_cache(self, platform: Platform | None = None) -> None:
|
||||
"""Clear cached platform implementations.
|
||||
|
||||
Args:
|
||||
platform: Specific platform to clear, or None to clear all
|
||||
"""
|
||||
if platform is None:
|
||||
logger.info("Clearing all cached platform implementations")
|
||||
self._instances.clear()
|
||||
else:
|
||||
keys_to_remove = [key for key in self._instances.keys() if key[0] == platform]
|
||||
for key in keys_to_remove:
|
||||
del self._instances[key]
|
||||
logger.info(f"Cleared cached implementations for {platform.value}")
|
||||
|
||||
|
||||
class PlatformFactory:
|
||||
"""Factory for creating platform-specific implementations."""
|
||||
"""Factory for creating platform-specific implementations with IDL support."""
|
||||
|
||||
def __init__(self):
|
||||
self.registry = PlatformRegistry()
|
||||
@@ -200,7 +237,7 @@ class PlatformFactory:
|
||||
Args:
|
||||
platform: Platform to create implementations for
|
||||
client: Solana RPC client
|
||||
**config: Platform-specific configuration
|
||||
**config: Platform-specific configuration (including verbose_idl)
|
||||
|
||||
Returns:
|
||||
PlatformImplementations containing all interface implementations
|
||||
@@ -266,6 +303,19 @@ class PlatformFactory:
|
||||
List of supported platforms
|
||||
"""
|
||||
return self.registry.get_supported_platforms()
|
||||
|
||||
def clear_caches(self, platform: Platform | None = None) -> None:
|
||||
"""Clear all caches for better memory management.
|
||||
|
||||
Args:
|
||||
platform: Specific platform to clear, or None to clear all
|
||||
"""
|
||||
# Clear implementation cache
|
||||
self.registry.clear_implementation_cache(platform)
|
||||
|
||||
# Clear IDL parser cache
|
||||
idl_manager = get_idl_manager()
|
||||
idl_manager.clear_cache(platform)
|
||||
|
||||
|
||||
# Global factory instance
|
||||
|
||||
@@ -5,7 +5,6 @@ This module handles LetsBonk (Raydium LaunchLab) specific pool operations
|
||||
by implementing the CurveManager interface using IDL-based decoding.
|
||||
"""
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from solders.pubkey import Pubkey
|
||||
@@ -23,30 +22,18 @@ logger = get_logger(__name__)
|
||||
class LetsBonkCurveManager(CurveManager):
|
||||
"""LetsBonk (Raydium LaunchLab) implementation of CurveManager interface."""
|
||||
|
||||
def __init__(self, client: SolanaClient):
|
||||
"""Initialize LetsBonk curve manager.
|
||||
def __init__(self, client: SolanaClient, idl_parser: IDLParser):
|
||||
"""Initialize LetsBonk curve manager with injected IDL parser.
|
||||
|
||||
Args:
|
||||
client: Solana RPC client
|
||||
idl_parser: Pre-loaded IDL parser for LetsBonk platform
|
||||
"""
|
||||
self.client = client
|
||||
self.address_provider = LetsBonkAddressProvider()
|
||||
self._idl_parser = self._load_idl_parser()
|
||||
self._idl_parser = idl_parser
|
||||
|
||||
logger.info("LetsBonk curve manager initialized with IDL-based account parsing")
|
||||
|
||||
def _load_idl_parser(self) -> IDLParser:
|
||||
"""Load the IDL parser for LetsBonk (Raydium LaunchLab)."""
|
||||
# Get the IDL file path relative to the project root
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.join(current_dir, "..", "..", "..")
|
||||
idl_path = os.path.join(project_root, "idl", "raydium_launchlab_idl.json")
|
||||
idl_path = os.path.normpath(idl_path)
|
||||
|
||||
if not os.path.exists(idl_path):
|
||||
raise FileNotFoundError(f"IDL file not found at {idl_path}")
|
||||
|
||||
return IDLParser(idl_path, verbose=False)
|
||||
logger.info("LetsBonk curve manager initialized with injected IDL parser")
|
||||
|
||||
@property
|
||||
def platform(self) -> Platform:
|
||||
@@ -67,7 +54,7 @@ class LetsBonkCurveManager(CurveManager):
|
||||
if not account.data:
|
||||
raise ValueError(f"No data in pool state account {pool_address}")
|
||||
|
||||
# Decode pool state using IDL parser
|
||||
# Decode pool state using injected IDL parser
|
||||
pool_state_data = self._decode_pool_state_with_idl(account.data)
|
||||
|
||||
return pool_state_data
|
||||
@@ -175,7 +162,7 @@ class LetsBonkCurveManager(CurveManager):
|
||||
return (pool_state["virtual_base"], pool_state["virtual_quote"])
|
||||
|
||||
def _decode_pool_state_with_idl(self, data: bytes) -> dict[str, Any]:
|
||||
"""Decode pool state data using IDL parser.
|
||||
"""Decode pool state data using injected IDL parser.
|
||||
|
||||
Args:
|
||||
data: Raw account data
|
||||
@@ -186,7 +173,7 @@ class LetsBonkCurveManager(CurveManager):
|
||||
Raises:
|
||||
ValueError: If IDL parsing fails
|
||||
"""
|
||||
# Use IDL parser to decode PoolState account data
|
||||
# Use injected IDL parser to decode PoolState account data
|
||||
decoded_pool_state = self._idl_parser.decode_account_data(
|
||||
data,
|
||||
"PoolState",
|
||||
|
||||
@@ -6,7 +6,6 @@ by implementing the EventParser interface with IDL-based parsing.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import os
|
||||
import struct
|
||||
from time import monotonic
|
||||
from typing import Any
|
||||
@@ -25,30 +24,21 @@ logger = get_logger(__name__)
|
||||
class LetsBonkEventParser(EventParser):
|
||||
"""LetsBonk implementation of EventParser interface with IDL-based parsing."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize LetsBonk event parser with IDL support."""
|
||||
self.address_provider = LetsBonkAddressProvider()
|
||||
self._idl_parser = self._load_idl_parser()
|
||||
def __init__(self, idl_parser: IDLParser):
|
||||
"""Initialize LetsBonk event parser with injected IDL parser.
|
||||
|
||||
# Get discriminators from IDL
|
||||
Args:
|
||||
idl_parser: Pre-loaded IDL parser for LetsBonk platform
|
||||
"""
|
||||
self.address_provider = LetsBonkAddressProvider()
|
||||
self._idl_parser = idl_parser
|
||||
|
||||
# Get discriminators from injected IDL parser
|
||||
discriminators = self._idl_parser.get_instruction_discriminators()
|
||||
self._initialize_discriminator_bytes = discriminators["initialize"]
|
||||
self._initialize_discriminator = struct.unpack("<Q", self._initialize_discriminator_bytes)[0]
|
||||
|
||||
logger.info("LetsBonk event parser initialized with IDL-based discriminators")
|
||||
|
||||
def _load_idl_parser(self) -> IDLParser:
|
||||
"""Load the IDL parser for LetsBonk (Raydium LaunchLab)."""
|
||||
# Get the IDL file path relative to the project root
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.join(current_dir, "..", "..", "..")
|
||||
idl_path = os.path.join(project_root, "idl", "raydium_launchlab_idl.json")
|
||||
idl_path = os.path.normpath(idl_path)
|
||||
|
||||
if not os.path.exists(idl_path):
|
||||
raise FileNotFoundError(f"IDL file not found at {idl_path}")
|
||||
|
||||
return IDLParser(idl_path, verbose=False)
|
||||
logger.info("LetsBonk event parser initialized with injected IDL parser")
|
||||
|
||||
@property
|
||||
def platform(self) -> Platform:
|
||||
@@ -79,7 +69,7 @@ class LetsBonkEventParser(EventParser):
|
||||
accounts: list[int],
|
||||
account_keys: list[bytes]
|
||||
) -> TokenInfo | None:
|
||||
"""Parse token creation from LetsBonk instruction data using IDL.
|
||||
"""Parse token creation from LetsBonk instruction data using injected IDL parser.
|
||||
|
||||
Args:
|
||||
instruction_data: Raw instruction data
|
||||
@@ -102,7 +92,7 @@ class LetsBonkEventParser(EventParser):
|
||||
return None
|
||||
return Pubkey.from_bytes(account_keys[account_index])
|
||||
|
||||
# Parse instruction data using IDL parser
|
||||
# Parse instruction data using injected IDL parser
|
||||
decoded = self._idl_parser.decode_instruction(instruction_data, account_keys, accounts)
|
||||
if not decoded or decoded['instruction_name'] != 'initialize':
|
||||
return None
|
||||
|
||||
@@ -6,7 +6,6 @@ by implementing the InstructionBuilder interface with IDL-based discriminators.
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import os
|
||||
import struct
|
||||
import time
|
||||
|
||||
@@ -26,29 +25,20 @@ logger = get_logger(__name__)
|
||||
class LetsBonkInstructionBuilder(InstructionBuilder):
|
||||
"""LetsBonk (Raydium LaunchLab) implementation of InstructionBuilder interface with IDL-based discriminators."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize LetsBonk instruction builder with IDL support."""
|
||||
self._idl_parser = self._load_idl_parser()
|
||||
def __init__(self, idl_parser: IDLParser):
|
||||
"""Initialize LetsBonk instruction builder with injected IDL parser.
|
||||
|
||||
# Get discriminators from IDL
|
||||
Args:
|
||||
idl_parser: Pre-loaded IDL parser for LetsBonk platform
|
||||
"""
|
||||
self._idl_parser = idl_parser
|
||||
|
||||
# Get discriminators from injected IDL parser
|
||||
discriminators = self._idl_parser.get_instruction_discriminators()
|
||||
self._buy_exact_in_discriminator = discriminators["buy_exact_in"]
|
||||
self._sell_exact_in_discriminator = discriminators["sell_exact_in"]
|
||||
|
||||
logger.info("LetsBonk instruction builder initialized with IDL-based discriminators")
|
||||
|
||||
def _load_idl_parser(self) -> IDLParser:
|
||||
"""Load the IDL parser for LetsBonk (Raydium LaunchLab)."""
|
||||
# Get the IDL file path relative to the project root
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.join(current_dir, "..", "..", "..")
|
||||
idl_path = os.path.join(project_root, "idl", "raydium_launchlab_idl.json")
|
||||
idl_path = os.path.normpath(idl_path)
|
||||
|
||||
if not os.path.exists(idl_path):
|
||||
raise FileNotFoundError(f"IDL file not found at {idl_path}")
|
||||
|
||||
return IDLParser(idl_path, verbose=False)
|
||||
logger.info("LetsBonk instruction builder initialized with injected IDL parser")
|
||||
|
||||
@property
|
||||
def platform(self) -> Platform:
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
"""
|
||||
Centralized IDL management for Solana platforms.
|
||||
|
||||
This module provides a single point of IDL loading and management to avoid
|
||||
duplicate loading across multiple platform implementation classes.
|
||||
"""
|
||||
|
||||
import os
|
||||
|
||||
from interfaces.core import Platform
|
||||
from utils.idl_parser import IDLParser
|
||||
from utils.logger import get_logger
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
|
||||
class IDLManager:
|
||||
"""Centralized manager for IDL parsers across all platforms."""
|
||||
|
||||
def __init__(self):
|
||||
"""Initialize the IDL manager."""
|
||||
self._parsers: dict[Platform, IDLParser] = {}
|
||||
self._idl_paths: dict[Platform, str] = {}
|
||||
self._setup_platform_idl_paths()
|
||||
|
||||
def _setup_platform_idl_paths(self) -> None:
|
||||
"""Setup IDL file paths for each platform."""
|
||||
# Get the project root directory (3 levels up from this file)
|
||||
current_dir = os.path.dirname(os.path.abspath(__file__))
|
||||
project_root = os.path.join(current_dir, "..", "..")
|
||||
project_root = os.path.normpath(project_root)
|
||||
|
||||
# Define IDL paths for each platform
|
||||
self._idl_paths = {
|
||||
Platform.LETS_BONK: os.path.join(project_root, "idl", "raydium_launchlab_idl.json"),
|
||||
# Platform.PUMP_FUN: os.path.join(project_root, "idl", "pump_fun_idl.json"), # Future
|
||||
}
|
||||
|
||||
def get_parser(self, platform: Platform, verbose: bool = False) -> IDLParser:
|
||||
"""Get or create an IDL parser for the specified platform.
|
||||
|
||||
Args:
|
||||
platform: Platform to get parser for
|
||||
verbose: Whether to enable verbose logging in the parser
|
||||
|
||||
Returns:
|
||||
IDLParser instance for the platform
|
||||
|
||||
Raises:
|
||||
ValueError: If platform is not supported or IDL file not found
|
||||
"""
|
||||
# Return cached parser if available
|
||||
if platform in self._parsers:
|
||||
return self._parsers[platform]
|
||||
|
||||
# Check if platform has IDL support
|
||||
if platform not in self._idl_paths:
|
||||
raise ValueError(f"Platform {platform.value} does not have IDL support configured")
|
||||
|
||||
idl_path = self._idl_paths[platform]
|
||||
|
||||
# Verify IDL file exists
|
||||
if not os.path.exists(idl_path):
|
||||
raise FileNotFoundError(f"IDL file not found for {platform.value} at {idl_path}")
|
||||
|
||||
# Load and cache the parser
|
||||
logger.info(f"Loading IDL parser for {platform.value} from {idl_path}")
|
||||
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")
|
||||
|
||||
return parser
|
||||
|
||||
def has_idl_support(self, platform: Platform) -> bool:
|
||||
"""Check if a platform has IDL support configured.
|
||||
|
||||
Args:
|
||||
platform: Platform to check
|
||||
|
||||
Returns:
|
||||
True if platform has IDL support
|
||||
"""
|
||||
return platform in self._idl_paths
|
||||
|
||||
def get_supported_platforms(self) -> list[Platform]:
|
||||
"""Get list of platforms with IDL support.
|
||||
|
||||
Returns:
|
||||
List of platforms that have IDL files configured
|
||||
"""
|
||||
return list(self._idl_paths.keys())
|
||||
|
||||
def clear_cache(self, platform: Platform | None = None) -> None:
|
||||
"""Clear cached parsers.
|
||||
|
||||
Args:
|
||||
platform: Specific platform to clear, or None to clear all
|
||||
"""
|
||||
if platform is None:
|
||||
logger.info("Clearing all cached IDL parsers")
|
||||
self._parsers.clear()
|
||||
elif platform in self._parsers:
|
||||
logger.info(f"Clearing cached IDL parser for {platform.value}")
|
||||
del self._parsers[platform]
|
||||
|
||||
def preload_parser(self, platform: Platform, verbose: bool = False) -> None:
|
||||
"""Preload IDL parser for a platform.
|
||||
|
||||
This can be useful for warming up the parser during initialization.
|
||||
|
||||
Args:
|
||||
platform: Platform to preload parser for
|
||||
verbose: Whether to enable verbose logging in the parser
|
||||
"""
|
||||
if platform not in self._parsers:
|
||||
logger.info(f"Preloading IDL parser for {platform.value}")
|
||||
self.get_parser(platform, verbose)
|
||||
else:
|
||||
logger.debug(f"IDL parser for {platform.value} already loaded")
|
||||
|
||||
def get_instruction_discriminators(self, platform: Platform) -> dict[str, bytes]:
|
||||
"""Get instruction discriminators for a platform.
|
||||
|
||||
Args:
|
||||
platform: Platform to get discriminators for
|
||||
|
||||
Returns:
|
||||
Dictionary mapping instruction names to discriminator bytes
|
||||
"""
|
||||
parser = self.get_parser(platform)
|
||||
return parser.get_instruction_discriminators()
|
||||
|
||||
def get_instruction_names(self, platform: Platform) -> list[str]:
|
||||
"""Get available instruction names for a platform.
|
||||
|
||||
Args:
|
||||
platform: Platform to get instruction names for
|
||||
|
||||
Returns:
|
||||
List of instruction names
|
||||
"""
|
||||
parser = self.get_parser(platform)
|
||||
return parser.get_instruction_names()
|
||||
|
||||
|
||||
# Global IDL manager instance
|
||||
_idl_manager: IDLManager | None = None
|
||||
|
||||
|
||||
def get_idl_manager() -> IDLManager:
|
||||
"""Get the global IDL manager instance.
|
||||
|
||||
Returns:
|
||||
Global IDLManager instance
|
||||
"""
|
||||
global _idl_manager
|
||||
if _idl_manager is None:
|
||||
_idl_manager = IDLManager()
|
||||
return _idl_manager
|
||||
|
||||
|
||||
def get_idl_parser(platform: Platform, verbose: bool = False) -> IDLParser:
|
||||
"""Convenience function to get an IDL parser for a platform.
|
||||
|
||||
Args:
|
||||
platform: Platform to get parser for
|
||||
verbose: Whether to enable verbose logging in the parser
|
||||
|
||||
Returns:
|
||||
IDLParser instance for the platform
|
||||
"""
|
||||
return get_idl_manager().get_parser(platform, verbose)
|
||||
|
||||
|
||||
def has_idl_support(platform: Platform) -> bool:
|
||||
"""Check if a platform has IDL support.
|
||||
|
||||
Args:
|
||||
platform: Platform to check
|
||||
|
||||
Returns:
|
||||
True if platform has IDL support
|
||||
"""
|
||||
return get_idl_manager().has_idl_support(platform)
|
||||
|
||||
|
||||
def preload_platform_idl(platform: Platform, verbose: bool = False) -> None:
|
||||
"""Preload IDL parser for a platform.
|
||||
|
||||
Args:
|
||||
platform: Platform to preload parser for
|
||||
verbose: Whether to enable verbose logging
|
||||
"""
|
||||
get_idl_manager().preload_parser(platform, verbose)
|
||||
Reference in New Issue
Block a user