feat: add pump portal listener

bots/bot-sniper-3-blocks.yaml

@@ -8,7 +8,7 @@ rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}"
 wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}"
 private_key: "${SOLANA_PRIVATE_KEY}"
 
-enabled: true # You can turn off the bot w/o removing its config
+enabled: false # You can turn off the bot w/o removing its config
 separate_process: true
 
 # Geyser configuration (fastest method for getting updates)

bots/bot-sniper-4-pp.yaml

@@ -0,0 +1,70 @@
+# This file defines comprehensive parameters and settings for the trading bot.
+# Carefully review and adjust values to match your trading strategy and risk tolerance.
+
+# Bot identification and connection settings
+name: "bot-sniper-pumpportal"
+env_file: ".env"
+rpc_endpoint: "${SOLANA_NODE_RPC_ENDPOINT}"
+wss_endpoint: "${SOLANA_NODE_WSS_ENDPOINT}"
+private_key: "${SOLANA_PRIVATE_KEY}"
+
+enabled: true # You can turn off the bot w/o removing its config
+separate_process: true
+
+# PumpPortal configuration (optional - uses default URL if not specified)
+pumpportal:
+  url: "wss://pumpportal.fun/api/data" # Default PumpPortal WebSocket URL
+
+# Trading parameters
+# Control trade execution: amount of SOL per trade and acceptable price deviation
+trade:
+  buy_amount: 0.0001 # Amount of SOL to spend when buying (in SOL)
+  buy_slippage: 0.3 # Maximum acceptable price deviation (0.3 = 30%)
+  sell_slippage: 0.3
+
+  # EXTREME FAST mode configuration
+  # When enabled, skips waiting for the bonding curve to stabilize and RPC price check.
+  # The bot buys the specified number of tokens directly, making the process faster but less precise.
+  extreme_fast_mode: true
+  extreme_fast_token_amount: 20 # Amount of tokens to buy
+
+# Priority fee configuration
+# Manage transaction speed and cost on the Solana network.
+# Note: dynamic mode requires an additional RPC call, which slows down the buying process.
+priority_fees:
+  enable_dynamic: false # Use latest transactions to estimate required fee (getRecentPrioritizationFees)
+  enable_fixed: true # Use fixed amount below
+  fixed_amount: 200_000 # Base fee in microlamports
+  extra_percentage: 0.0 # Percentage increase on riority fee regardless of the calculation method (0.1 = 10%)
+  hard_cap: 200_000 # Maximum allowable fee in microlamports to prevent excessive spending
+
+# Filters for token selection
+filters:
+  match_string: null # Only process tokens with this string in name/symbol
+  bro_address: null # Only trade tokens created by this user address
+  listener_type: "pumpportal" # Method for detecting new tokens: "logs", "blocks", "geyser", or "pumpportal"
+  max_token_age: 0.001 # Maximum token age in seconds for processing
+  marry_mode: false # Only buy tokens, skip selling
+  yolo_mode: false # Continuously trade tokens
+
+# Retry and timeout settings
+retries:
+  max_attempts: 1 # Number of attempts for transaction submission
+  wait_after_creation: 15 # Seconds to wait after token creation (only if EXTREME FAST is disabled)
+  wait_after_buy: 15 # Holding period after buy transaction
+  wait_before_new_token: 15 # Pause between token trades
+
+# Token and account management
+cleanup:
+  # Cleanup mode determines when to manage token accounts. Options:
+  # "disabled": no cleanup will occur.
+  # "on_fail": only clean up if a buy transaction fails.
+  # "after_sell": clean up after selling.
+  # "post_session": clean up all empty accounts after a trading session ends.
+  mode: "post_session"
+  force_close_with_burn: false # Force burning remaining tokens before closing account
+  with_priority_fee: false # Use priority fees for cleanup transactions
+
+# Node provider configuration (not implemented)
+node:
+  max_rps: 25 # Maximum requests per second

src/bot_runner.py

@@ -62,6 +62,9 @@ async def start_bot(config_path: str):
         geyser_api_token=cfg.get("geyser", {}).get("api_token"),
         geyser_auth_type=cfg.get("geyser", {}).get("auth_type"),
         
+        # PumpPortal configuration (if applicable)
+        pumpportal_url=cfg.get("pumpportal", {}).get("url", "wss://pumpportal.fun/api/data"),
+        
         # Priority fee configuration
         enable_dynamic_priority_fee=cfg.get("priority_fees", {}).get("enable_dynamic", False),
         enable_fixed_priority_fee=cfg.get("priority_fees", {}).get("enable_fixed", True),

src/monitoring/pumpportal_event_processor.py

@@ -0,0 +1,126 @@
+"""
+Event processing for pump.fun tokens using PumpPortal data.
+"""
+
+from solders.pubkey import Pubkey
+
+from core.pubkeys import PumpAddresses, SystemAddresses
+from trading.base import TokenInfo
+from utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class PumpPortalEventProcessor:
+    """Processes token creation events from PumpPortal WebSocket."""
+
+    def __init__(self, pump_program: Pubkey):
+        """Initialize event processor.
+
+        Args:
+            pump_program: Pump.fun program address
+        """
+        self.pump_program = pump_program
+
+    def process_token_data(self, token_data: dict) -> TokenInfo | None:
+        """Process token data from PumpPortal and extract token creation info.
+
+        Args:
+            token_data: Token data from PumpPortal WebSocket
+
+        Returns:
+            TokenInfo if token creation found, None otherwise
+        """
+        try:
+            # Extract required fields
+            name = token_data.get("name", "")
+            symbol = token_data.get("symbol", "")
+            mint_str = token_data.get("mint")
+            bonding_curve_str = token_data.get("bondingCurveKey")
+            creator_str = token_data.get("traderPublicKey")  # Maps to user field
+            uri = token_data.get("uri", "")
+
+            # Additional fields available from PumpPortal but not used:
+            # - initialBuy: Initial buy amount in SOL
+            # - marketCapSol: Market cap in SOL
+            # - vSolInBondingCurve: Virtual SOL in bonding curve
+            # - vTokensInBondingCurve: Virtual tokens in bonding curve
+            # - signature: Transaction signature
+
+            if not all([name, symbol, mint_str, bonding_curve_str, creator_str]):
+                logger.warning("Missing required fields in PumpPortal token data")
+                return None
+
+            # Convert string addresses to Pubkey objects
+            mint = Pubkey.from_string(mint_str)
+            bonding_curve = Pubkey.from_string(bonding_curve_str)
+            user = Pubkey.from_string(creator_str)
+            
+            # For PumpPortal, we assume the creator is the same as the user
+            # since PumpPortal doesn't distinguish between them
+            creator = user
+
+            # Calculate derived addresses
+            associated_bonding_curve = self._find_associated_bonding_curve(
+                mint, bonding_curve
+            )
+            creator_vault = self._find_creator_vault(creator)
+
+            return TokenInfo(
+                name=name,
+                symbol=symbol,
+                uri=uri,
+                mint=mint,
+                bonding_curve=bonding_curve,
+                associated_bonding_curve=associated_bonding_curve,
+                user=user,
+                creator=creator,
+                creator_vault=creator_vault,
+            )
+
+        except Exception as e:
+            logger.error(f"Failed to process PumpPortal token data: {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
+
+    def _find_creator_vault(self, creator: Pubkey) -> Pubkey:
+        """
+        Find the creator vault for a creator.
+
+        Args:
+            creator: Creator address
+
+        Returns:
+            Creator vault address
+        """
+        derived_address, _ = Pubkey.find_program_address(
+            [
+                b"creator-vault",
+                bytes(creator)
+            ],
+            PumpAddresses.PROGRAM,
+        )
+        return derived_address

src/monitoring/pumpportal_listener.py

@@ -0,0 +1,166 @@
+"""
+PumpPortal monitoring for pump.fun tokens.
+"""
+
+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.pumpportal_event_processor import PumpPortalEventProcessor
+from trading.base import TokenInfo
+from utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class PumpPortalListener(BaseTokenListener):
+    """PumpPortal listener for pump.fun token creation events."""
+
+    def __init__(self, pump_program: Pubkey, pumpportal_url: str = "wss://pumpportal.fun/api/data"):
+        """Initialize token listener.
+
+        Args:
+            pump_program: Pump.fun program address
+            pumpportal_url: PumpPortal WebSocket URL
+        """
+        self.pump_program = pump_program
+        self.pumpportal_url = pumpportal_url
+        self.event_processor = PumpPortalEventProcessor(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 PumpPortal WebSocket.
+
+        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.pumpportal_url) as websocket:
+                    await self._subscribe_to_new_tokens(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("PumpPortal WebSocket connection closed. Reconnecting...")
+                        ping_task.cancel()
+
+            except Exception as e:
+                logger.error(f"PumpPortal WebSocket connection error: {e}")
+                logger.info("Reconnecting in 5 seconds...")
+                await asyncio.sleep(5)
+
+    async def _subscribe_to_new_tokens(self, websocket) -> None:
+        """Subscribe to new token events from PumpPortal.
+
+        Args:
+            websocket: Active WebSocket connection
+        """
+        subscription_message = json.dumps({
+            "method": "subscribeNewToken",
+            "params": []
+        })
+
+        await websocket.send(subscription_message)
+        logger.info("Subscribed to PumpPortal new token events")
+
+    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 TimeoutError:
+                    logger.warning("Ping timeout - PumpPortal server not responding")
+                    # Force reconnection
+                    await websocket.close()
+                    return
+        except asyncio.CancelledError:
+            pass
+        except Exception as e:
+            logger.error(f"Ping error: {e}")
+
+    async def _wait_for_token_creation(self, websocket) -> TokenInfo | None:
+        """Wait for token creation event from PumpPortal.
+
+        Args:
+            websocket: Active WebSocket connection
+
+        Returns:
+            TokenInfo if a token creation is found, None otherwise
+        """
+        try:
+            response = await asyncio.wait_for(websocket.recv(), timeout=30)
+            data = json.loads(response)
+
+            # Handle different message formats from PumpPortal
+            token_info = None
+            if "method" in data and data["method"] == "newToken":
+                # Standard newToken method format
+                params = data.get("params", [])
+                if params and len(params) > 0:
+                    token_data = params[0]
+                    token_info = self.event_processor.process_token_data(token_data)
+            elif "signature" in data and "mint" in data:
+                # Direct token data format
+                token_info = self.event_processor.process_token_data(data)
+
+            return token_info
+
+        except TimeoutError:
+            logger.debug("No data received from PumpPortal for 30 seconds")
+        except websockets.exceptions.ConnectionClosed:
+            logger.warning("PumpPortal WebSocket connection closed")
+            raise
+        except json.JSONDecodeError as e:
+            logger.error(f"Failed to decode PumpPortal message: {e}")
+        except Exception as e:
+            logger.error(f"Error processing PumpPortal WebSocket message: {e}")
+
+        return None

src/trading/trader.py

@@ -25,6 +25,7 @@ from core.wallet import Wallet
 from monitoring.block_listener import BlockListener
 from monitoring.geyser_listener import GeyserListener
 from monitoring.logs_listener import LogsListener
+from monitoring.pumpportal_listener import PumpPortalListener
 from trading.base import TokenInfo, TradeResult
 from trading.buyer import TokenBuyer
 from trading.seller import TokenSeller
@@ -49,6 +50,7 @@ class PumpTrader:
         geyser_endpoint: str | None = None,
         geyser_api_token: str | None = None,
         geyser_auth_type: str = "x-token",
+        pumpportal_url: str = "wss://pumpportal.fun/api/data",
 
         extreme_fast_mode: bool = False,
         extreme_fast_token_amount: int = 30,
@@ -88,10 +90,11 @@ class PumpTrader:
             buy_slippage: Slippage tolerance for buys
             sell_slippage: Slippage tolerance for sells
 
-            listener_type: Type of listener to use ('logs', 'blocks', or 'geyser')
+            listener_type: Type of listener to use ('logs', 'blocks', 'geyser', or 'pumpportal')
             geyser_endpoint: Geyser endpoint URL (required for geyser listener)
             geyser_api_token: Geyser API token (required for geyser listener)
             geyser_auth_type: Geyser authentication type ('x-token' or 'basic')
+            pumpportal_url: PumpPortal WebSocket URL (default: wss://pumpportal.fun/api/data)
 
             extreme_fast_mode: Whether to enable extreme fast mode
             extreme_fast_token_amount: Maximum token amount for extreme fast mode
@@ -165,6 +168,9 @@ class PumpTrader:
         elif listener_type == "logs":
             self.token_listener = LogsListener(wss_endpoint, PumpAddresses.PROGRAM)
             logger.info("Using logsSubscribe listener for token monitoring")
+        elif listener_type == "pumpportal":
+            self.token_listener = PumpPortalListener(PumpAddresses.PROGRAM, pumpportal_url)
+            logger.info("Using PumpPortal listener for token monitoring")
         else:
             self.token_listener = BlockListener(wss_endpoint, PumpAddresses.PROGRAM)
             logger.info("Using blockSubscribe listener for token monitoring")