diff --git a/src/cli.py b/src/cli.py index fe35bdd..983f880 100644 --- a/src/cli.py +++ b/src/cli.py @@ -88,12 +88,6 @@ async def main() -> None: args.sell_slippage if args.sell_slippage is not None else config.SELL_SLIPPAGE ) - # Not implemented parameters - enable_dynamic_prior__fee = ( - config.ENABLE_DYNAMIC_PRIORITY_FEE - ) # TODO: to be implemented - prior_fee_multiplier = config.EXTRA_PRIORITY_FEE # TODO: to be implemented - trader: PumpTrader = PumpTrader( rpc_endpoint=rpc_endpoint, # type: ignore wss_endpoint=wss_endpoint, # type: ignore @@ -102,6 +96,7 @@ async def main() -> None: buy_slippage=buy_slippage, sell_slippage=sell_slippage, max_retries=config.MAX_RETRIES, + listener_type=config.LISTENER_TYPE, ) try: diff --git a/src/config.py b/src/config.py index 4e499be..a2b8d15 100644 --- a/src/config.py +++ b/src/config.py @@ -20,6 +20,10 @@ HARD_CAP_PRIOR_FEE: int = ( ) +# Listener configuration +LISTENER_TYPE = "block" # Options: "block" or "logs" + + # Retries and timeouts MAX_RETRIES: int = 10 # Number of retries for transaction sending # TODO: waiting times will be replaced with retries to shorten delays diff --git a/src/monitoring/base_listener.py b/src/monitoring/base_listener.py new file mode 100644 index 0000000..0a5e191 --- /dev/null +++ b/src/monitoring/base_listener.py @@ -0,0 +1,29 @@ +""" +Base class for WebSocket token listeners. +""" + +from abc import ABC, abstractmethod +from collections.abc import Awaitable, Callable + +from trading.base import TokenInfo + + +class BaseTokenListener(ABC): + """Base abstract class for token listeners.""" + + @abstractmethod + async def listen_for_tokens( + self, + token_callback: Callable[[TokenInfo], Awaitable[None]], + match_string: str | None = None, + creator_address: str | None = None, + ) -> None: + """ + Listen for new token creations. + + 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 + """ + pass diff --git a/src/monitoring/events.py b/src/monitoring/block_event_processor.py similarity index 100% rename from src/monitoring/events.py rename to src/monitoring/block_event_processor.py diff --git a/src/monitoring/listener.py b/src/monitoring/block_listener.py similarity index 96% rename from src/monitoring/listener.py rename to src/monitoring/block_listener.py index 7af8647..846b761 100644 --- a/src/monitoring/listener.py +++ b/src/monitoring/block_listener.py @@ -9,15 +9,16 @@ from collections.abc import Awaitable, Callable import websockets from solders.pubkey import Pubkey -from monitoring.events import PumpEventProcessor +from monitoring.base_listener import BaseTokenListener +from monitoring.block_event_processor import PumpEventProcessor from trading.base import TokenInfo from utils.logger import get_logger logger = get_logger(__name__) -class PumpTokenListener: - """WebSocket listener for pump.fun token creation events.""" +class BlockListener(BaseTokenListener): + """WebSocket listener for pump.fun token creation events using blockSubscribe.""" def __init__(self, wss_endpoint: str, pump_program: Pubkey): """Initialize token listener. diff --git a/src/monitoring/logs_event_processor.py b/src/monitoring/logs_event_processor.py new file mode 100644 index 0000000..9aff061 --- /dev/null +++ b/src/monitoring/logs_event_processor.py @@ -0,0 +1,151 @@ +""" +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 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 + ) + + 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"]), + ) + 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 diff --git a/src/monitoring/logs_listener.py b/src/monitoring/logs_listener.py new file mode 100644 index 0000000..4af072e --- /dev/null +++ b/src/monitoring/logs_listener.py @@ -0,0 +1,167 @@ +""" +WebSocket monitoring for pump.fun tokens using logsSubscribe. +""" + +import asyncio +import json +from collections.abc import Awaitable, Callable + +import websockets +from solders.pubkey import Pubkey + +from monitoring.base_listener import BaseTokenListener +from monitoring.logs_event_processor import LogsEventProcessor +from trading.base import TokenInfo +from utils.logger import get_logger + +logger = get_logger(__name__) + + +class LogsListener(BaseTokenListener): + """WebSocket listener for pump.fun token creation events using logsSubscribe.""" + + def __init__(self, wss_endpoint: str, pump_program: Pubkey): + """Initialize token listener. + + Args: + wss_endpoint: WebSocket endpoint URL + pump_program: Pump.fun program address + """ + self.wss_endpoint = wss_endpoint + self.pump_program = pump_program + self.event_processor = LogsEventProcessor(pump_program) + self.ping_interval = 20 # seconds + + async def listen_for_tokens( + self, + token_callback: Callable[[TokenInfo], Awaitable[None]], + match_string: str | None = None, + creator_address: str | None = None, + ) -> None: + """Listen for new token creations using logsSubscribe. + + 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 + """ + while True: + try: + async with websockets.connect(self.wss_endpoint) as websocket: + await self._subscribe_to_logs(websocket) + ping_task = asyncio.create_task(self._ping_loop(websocket)) + + try: + while True: + token_info = await self._wait_for_token_creation(websocket) + if not token_info: + continue + + logger.info( + f"New token detected: {token_info.name} ({token_info.symbol})" + ) + + if match_string and not ( + match_string.lower() in token_info.name.lower() + or match_string.lower() in token_info.symbol.lower() + ): + logger.info( + f"Token does not match filter '{match_string}'. Skipping..." + ) + continue + + if ( + creator_address + and str(token_info.user) != creator_address + ): + logger.info( + f"Token not created by {creator_address}. Skipping..." + ) + continue + + await token_callback(token_info) + + except websockets.exceptions.ConnectionClosed: + logger.warning("WebSocket connection closed. Reconnecting...") + ping_task.cancel() + + except Exception as e: + logger.error(f"WebSocket connection error: {str(e)}") + 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. + + Args: + websocket: Active WebSocket connection + """ + subscription_message = json.dumps( + { + "jsonrpc": "2.0", + "id": 1, + "method": "logsSubscribe", + "params": [ + {"mentions": [str(self.pump_program)]}, + {"commitment": "processed"}, + ], + } + ) + + await websocket.send(subscription_message) + logger.info(f"Subscribed to logs mentioning program: {self.pump_program}") + + # 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. + + Args: + websocket: Active WebSocket connection + """ + try: + while True: + await asyncio.sleep(self.ping_interval) + try: + pong_waiter = await websocket.ping() + await asyncio.wait_for(pong_waiter, timeout=10) + except asyncio.TimeoutError: + logger.warning("Ping timeout - server not responding") + # Force reconnection + await websocket.close() + return + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Ping error: {str(e)}") + + async def _wait_for_token_creation(self, websocket) -> TokenInfo | None: + try: + response = await asyncio.wait_for(websocket.recv(), timeout=30) + data = json.loads(response) + + if "method" not in data or data["method"] != "logsNotification": + return None + + log_data = data["params"]["result"]["value"] + 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) + + except asyncio.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)}") + + return None diff --git a/src/trading/trader.py b/src/trading/trader.py index e9def64..6204217 100644 --- a/src/trading/trader.py +++ b/src/trading/trader.py @@ -14,7 +14,8 @@ from core.curve import BondingCurveManager from core.priority_fee.manager import PriorityFeeManager from core.pubkeys import PumpAddresses from core.wallet import Wallet -from monitoring.listener import PumpTokenListener +from monitoring.block_listener import BlockListener +from monitoring.logs_listener import LogsListener from trading.base import TokenInfo, TradeResult from trading.buyer import TokenBuyer from trading.seller import TokenSeller @@ -35,6 +36,7 @@ class PumpTrader: buy_slippage: float, sell_slippage: float, max_retries: int = 5, + listener_type: str = "block", # Add this parameter ): """Initialize the pump trader. @@ -46,6 +48,7 @@ class PumpTrader: buy_slippage: Slippage tolerance for buys sell_slippage: Slippage tolerance for sells max_retries: Maximum number of retry attempts + listener_type: Type of listener to use ('block' or 'logs') """ self.solana_client = SolanaClient(rpc_endpoint) self.wallet = Wallet(private_key) @@ -79,7 +82,13 @@ class PumpTrader: max_retries, ) - self.token_listener = PumpTokenListener(wss_endpoint, PumpAddresses.PROGRAM) + # Initialize the appropriate listener type + if listener_type.lower() == "logs": + self.token_listener = LogsListener(wss_endpoint, PumpAddresses.PROGRAM) + logger.info("Using logsSubscribe listener for token monitoring") + else: + self.token_listener = BlockListener(wss_endpoint, PumpAddresses.PROGRAM) + logger.info("Using blockSubscribe listener for token monitoring") self.buy_amount = buy_amount self.buy_slippage = buy_slippage @@ -92,7 +101,7 @@ class PumpTrader: self.processing = False self.processed_tokens: set[str] = set() self.token_timestamps: dict[str, float] = {} - + async def start( self, match_string: str | None = None, diff --git a/tests/compare_listeners.py b/tests/compare_listeners.py new file mode 100644 index 0000000..809b7c7 --- /dev/null +++ b/tests/compare_listeners.py @@ -0,0 +1,137 @@ +""" +Test script to compare BlockListener and LogsListener +Runs both listeners simultaneously to compare their performance +""" + +import asyncio +import logging +import os +import sys +import time +from pathlib import Path + +sys.path.append(str(Path(__file__).parent.parent / "src")) + +from core.pubkeys import PumpAddresses +from monitoring.block_listener import BlockListener +from monitoring.logs_listener import LogsListener +from trading.base import TokenInfo + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger("listener-comparison") + + +class TimingTokenCallback: + def __init__(self, name: str): + self.name = name + self.detected_tokens = [] + self.detection_times = {} + + async def on_token_created(self, token_info: TokenInfo) -> None: + """Process detected token with timing information""" + token_key = str(token_info.mint) + detection_time = time.time() + + self.detected_tokens.append(token_info) + self.detection_times[token_key] = detection_time + + logger.info(f"[{self.name}] Detected: {token_info.name} ({token_info.symbol})") + print(f"\n{'=' * 50}") + print(f"[{self.name}] NEW TOKEN: {token_info.name}") + print(f"Symbol: {token_info.symbol}") + print(f"Mint: {token_info.mint}") + print(f"Detection time: {detection_time}") + print(f"{'=' * 50}\n") + + +async def run_comparison(test_duration: int = 300): + """Run both listeners and compare their performance""" + wss_endpoint = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") + if not wss_endpoint: + logger.error("SOLANA_NODE_WSS_ENDPOINT environment variable is not set") + return + + logger.info(f"Connecting to WebSocket: {wss_endpoint}") + + block_listener = BlockListener(wss_endpoint, PumpAddresses.PROGRAM) + logs_listener = LogsListener(wss_endpoint, PumpAddresses.PROGRAM) + + block_callback = TimingTokenCallback("BlockListener") + logs_callback = TimingTokenCallback("LogsListener") + + logger.info("Starting both listeners...") + block_task = asyncio.create_task( + block_listener.listen_for_tokens(block_callback.on_token_created) + ) + logs_task = asyncio.create_task( + logs_listener.listen_for_tokens(logs_callback.on_token_created) + ) + + logger.info(f"Comparison running for {test_duration} seconds...") + try: + await asyncio.sleep(test_duration) + except KeyboardInterrupt: + logger.info("Test interrupted by user") + finally: + block_task.cancel() + logs_task.cancel() + try: + await asyncio.gather(block_task, logs_task, return_exceptions=True) + except asyncio.CancelledError: + pass + + logger.info(f"BlockListener detected {len(block_callback.detected_tokens)} tokens") + logger.info(f"LogsListener detected {len(logs_callback.detected_tokens)} tokens") + + # Find tokens detected by both listeners + block_mints = {str(token.mint) for token in block_callback.detected_tokens} + logs_mints = {str(token.mint) for token in logs_callback.detected_tokens} + common_mints = block_mints.intersection(logs_mints) + + logger.info(f"Tokens detected by both listeners: {len(common_mints)}") + + # Compare detection times for common tokens + if common_mints: + logger.info("\nPerformance comparison for tokens detected by both listeners:") + logger.info("Token Mint | BlockListener Time | LogsListener Time | Difference (ms)") + logger.info("-" * 80) + + for mint in common_mints: + block_time = block_callback.detection_times.get(mint) + logs_time = logs_callback.detection_times.get(mint) + + if block_time and logs_time: + diff_ms = abs(block_time - logs_time) * 1000 # Convert to milliseconds + faster = "BlockListener" if block_time < logs_time else "LogsListener" + + logger.info(f"{mint[:10]}... | {block_time:.6f} | {logs_time:.6f} | {diff_ms:.2f}ms ({faster} faster)") + + # Report tokens only detected by one listener + block_only = block_mints - logs_mints + logs_only = logs_mints - block_mints + + if block_only: + logger.info(f"\nTokens only detected by BlockListener: {len(block_only)}") + for mint in block_only: + logger.info(f" - {mint}") + + if logs_only: + logger.info(f"\nTokens only detected by LogsListener: {len(logs_only)}") + for mint in logs_only: + logger.info(f" - {mint}") + + +if __name__ == "__main__": + test_duration = 30 # seconds + + if len(sys.argv) > 1: + try: + test_duration = int(sys.argv[1]) + except ValueError: + logger.error(f"Invalid test duration: {sys.argv[1]}. Using default of {test_duration} seconds.") + + logger.info("Starting listener comparison test") + logger.info(f"Will run for {test_duration} seconds") + asyncio.run(run_comparison(test_duration)) diff --git a/tests/test_block_listener.py b/tests/test_block_listener.py new file mode 100644 index 0000000..9648f21 --- /dev/null +++ b/tests/test_block_listener.py @@ -0,0 +1,97 @@ +""" +Test script for BlockListener +Tests websocket monitoring for new pump.fun tokens using blockSubscribe +""" + +import asyncio +import logging +import os +import sys +from pathlib import Path + +sys.path.append(str(Path(__file__).parent.parent / "src")) + +from core.pubkeys import PumpAddresses +from monitoring.block_listener import BlockListener +from trading.base import TokenInfo + +logging.basicConfig( + level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" +) +logger = logging.getLogger("block-listener-test") + + +class TestTokenCallback: + def __init__(self): + self.detected_tokens = [] + + async def on_token_created(self, token_info: TokenInfo) -> None: + """Process detected token""" + logger.info(f"New token detected: {token_info.name} ({token_info.symbol})") + logger.info(f"Mint: {token_info.mint}") + self.detected_tokens.append(token_info) + print(f"\n{'=' * 50}") + print(f"NEW TOKEN: {token_info.name}") + print(f"Symbol: {token_info.symbol}") + print(f"Mint: {token_info.mint}") + print(f"URI: {token_info.uri}") + print(f"Creator: {token_info.user}") + print(f"Bonding Curve: {token_info.bonding_curve}") + print(f"Associated Bonding Curve: {token_info.associated_bonding_curve}") + print(f"{'=' * 50}\n") + + +async def test_block_listener( + match_string: str | None = None, + creator_address: str | None = None, + test_duration: int = 60, +): + """Test the block listener functionality""" + wss_endpoint = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") + if not wss_endpoint: + logger.error("SOLANA_NODE_WSS_ENDPOINT environment variable is not set") + return [] + + logger.info(f"Connecting to WebSocket: {wss_endpoint}") + listener = BlockListener(wss_endpoint, PumpAddresses.PROGRAM) + callback = TestTokenCallback() + + if match_string: + logger.info(f"Filtering tokens matching: {match_string}") + if creator_address: + logger.info(f"Filtering tokens by creator: {creator_address}") + + listen_task = asyncio.create_task( + listener.listen_for_tokens( + callback.on_token_created, + match_string=match_string, + creator_address=creator_address, + ) + ) + + logger.info(f"Listening for {test_duration} seconds...") + try: + await asyncio.sleep(test_duration) + except KeyboardInterrupt: + logger.info("Test interrupted by user") + finally: + listen_task.cancel() + try: + await listen_task + except asyncio.CancelledError: + pass + + logger.info(f"Detected {len(callback.detected_tokens)} tokens") + for token in callback.detected_tokens: + logger.info(f" - {token.name} ({token.symbol}): {token.mint}") + + return callback.detected_tokens + + +if __name__ == "__main__": + match_string = None # Update if you want to filter tokens by name/symbol + creator_address = None # Update if you want to filter tokens by creator address + test_duration = 15 + + logger.info("Starting block listener test (using blockSubscribe)") + asyncio.run(test_block_listener(match_string, creator_address, test_duration)) diff --git a/tests/test_listener.py b/tests/test_logs_listener.py similarity index 78% rename from tests/test_listener.py rename to tests/test_logs_listener.py index 09a5063..8dcc05f 100644 --- a/tests/test_listener.py +++ b/tests/test_logs_listener.py @@ -1,6 +1,6 @@ """ -Test script for PumpTokenListener -Tests websocket monitoring for new pump.fun tokens +Test script for LogsListener +Tests websocket monitoring for new pump.fun tokens using logsSubscribe """ import asyncio @@ -12,13 +12,13 @@ from pathlib import Path sys.path.append(str(Path(__file__).parent.parent / "src")) from core.pubkeys import PumpAddresses -from monitoring.listener import PumpTokenListener +from monitoring.logs_listener import LogsListener from trading.base import TokenInfo logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s" ) -logger = logging.getLogger("token-listener-test") +logger = logging.getLogger("logs-listener-test") class TestTokenCallback: @@ -41,16 +41,19 @@ class TestTokenCallback: print(f"{'=' * 50}\n") -async def test_pump_token_listener( +async def test_logs_listener( match_string: str | None = None, creator_address: str | None = None, test_duration: int = 60, ): - """Test the token listener functionality""" - wss_endpoint = os.environ.get( - "SOLANA_NODE_WSS_ENDPOINT") + """Test the logs listener functionality""" + wss_endpoint = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") + if not wss_endpoint: + logger.error("SOLANA_NODE_WSS_ENDPOINT environment variable is not set") + return [] + logger.info(f"Connecting to WebSocket: {wss_endpoint}") - listener = PumpTokenListener(wss_endpoint, PumpAddresses.PROGRAM) + listener = LogsListener(wss_endpoint, PumpAddresses.PROGRAM) callback = TestTokenCallback() if match_string: @@ -90,5 +93,5 @@ if __name__ == "__main__": creator_address = None # Update if you want to filter tokens by creator address test_duration = 15 - logger.info("Starting token listener test") - asyncio.run(test_pump_token_listener(match_string, creator_address, test_duration)) + logger.info("Starting logs listener test (using logsSubscribe)") + asyncio.run(test_logs_listener(match_string, creator_address, test_duration))