diff --git a/src/platforms/__init__.py b/src/platforms/__init__.py index 7c0db22..feb53f0 100644 --- a/src/platforms/__init__.py +++ b/src/platforms/__init__.py @@ -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 diff --git a/src/platforms/letsbonk/curve_manager.py b/src/platforms/letsbonk/curve_manager.py index 45540fc..395fe6e 100644 --- a/src/platforms/letsbonk/curve_manager.py +++ b/src/platforms/letsbonk/curve_manager.py @@ -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", diff --git a/src/platforms/letsbonk/event_parser.py b/src/platforms/letsbonk/event_parser.py index 2466503..c5f6f90 100644 --- a/src/platforms/letsbonk/event_parser.py +++ b/src/platforms/letsbonk/event_parser.py @@ -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(" 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 diff --git a/src/platforms/letsbonk/instruction_builder.py b/src/platforms/letsbonk/instruction_builder.py index 2e3c44b..b16a6bd 100644 --- a/src/platforms/letsbonk/instruction_builder.py +++ b/src/platforms/letsbonk/instruction_builder.py @@ -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: diff --git a/src/utils/idl_manager.py b/src/utils/idl_manager.py new file mode 100644 index 0000000..bfb208c --- /dev/null +++ b/src/utils/idl_manager.py @@ -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) \ No newline at end of file