mirror of
https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git
synced 2026-08-08 13:07:45 +00:00
feat: add logsSubscribe
This commit is contained in:
smypmsa
@@ -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
|
||||
@@ -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.
|
||||
@@ -0,0 +1,152 @@
|
||||
"""
|
||||
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)
|
||||
print(signature)
|
||||
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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user