Merge pull request #65 from chainstacklabs/feature/logs-subscribe

Add logsSubscribe listener

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:

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

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

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:
+class BlockListener(BaseTokenListener):
-    """WebSocket listener for pump.fun token creation events."""
+    """WebSocket listener for pump.fun token creation events using blockSubscribe."""
 
     def __init__(self, wss_endpoint: str, pump_program: Pubkey):
         """Initialize token listener.

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("<Q", data[:8])[0]
+        if discriminator != self.CREATE_DISCRIMINATOR:
+            logger.info(f"Skipping non-Create instruction with discriminator: {discriminator}")
+            return None
+
+        offset = 8
+        parsed_data = {}
+
+        # Parse fields based on CreateEvent structure
+        fields = [
+            ("name", "string"),
+            ("symbol", "string"),
+            ("uri", "string"),
+            ("mint", "publicKey"),
+            ("bondingCurve", "publicKey"),
+            ("user", "publicKey"),
+        ]
+
+        try:
+            for field_name, field_type in fields:
+                if field_type == "string":
+                    length = struct.unpack("<I", data[offset : offset + 4])[0]
+                    offset += 4
+                    value = data[offset : offset + length].decode("utf-8")
+                    offset += length
+                elif field_type == "publicKey":
+                    value = base58.b58encode(data[offset : offset + 32]).decode("utf-8")
+                    offset += 32
+
+                parsed_data[field_name] = value
+
+            return parsed_data
+        except Exception as e:
+            logger.error(f"Failed to parse create instruction: {e}")
+            return None
+
+    def _find_associated_bonding_curve(
+        self, mint: Pubkey, bonding_curve: Pubkey
+    ) -> 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

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

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

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))

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))

tests/test_logs_listener.py

@@ -1,6 +1,6 @@
 """
-Test script for PumpTokenListener
+Test script for LogsListener
-Tests websocket monitoring for new pump.fun tokens
+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"""
+    """Test the logs listener functionality"""
-    wss_endpoint = os.environ.get(
+    wss_endpoint = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
-        "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")
+    logger.info("Starting logs listener test (using logsSubscribe)")
-    asyncio.run(test_pump_token_listener(match_string, creator_address, test_duration))
+    asyncio.run(test_logs_listener(match_string, creator_address, test_duration))