gavindiaz/pumpfun-bonkfun-bot_github
1
0
Fork 0
mirror of https://github.com/chainstacklabs/pumpfun-bonkfun-bot.git synced 2026-07-27 23:37:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove redundant tests

Browse Source
This commit is contained in:
smypmsa
2025-08-02 15:12:49 +00:00
parent 93332794a3
commit 72a3d23e6b
4 changed files with 0 additions and 522 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/compare_listeners.py
-209
View File
@@ -1,209 +0,0 @@
"""
Test script to compare BlockListener, LogsListener, and GeyserListener
Runs all listeners simultaneously to compare their performance
"""
import asyncio
import logging
import os
import sys
import time
from pathlib import Path
from dotenv import load_dotenv
sys.path.append(str(Path(__file__).parent.parent / "src"))
from core.pubkeys import PumpAddresses
from monitoring.block_listener import BlockListener
from monitoring.geyser_listener import GeyserListener
from monitoring.logs_listener import LogsListener
from trading.base import TokenInfo
load_dotenv()
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 listen_with_timeout(listener, callback, timeout):
"""Run a listener for a specified duration"""
try:
listen_task = asyncio.create_task(
listener.listen_for_tokens(callback.on_token_created)
)
await asyncio.sleep(timeout)
listen_task.cancel()
try:
await listen_task
except asyncio.CancelledError:
pass
except Exception as e:
logger.error(f"Error in listener {callback.name}: {e}")
async def run_comparison(test_duration: int = 300):
"""Run all listeners and compare their performance"""
wss_endpoint = os.environ.get("SOLANA_NODE_WSS_ENDPOINT")
geyser_endpoint = os.environ.get("GEYSER_ENDPOINT")
geyser_api_token = os.environ.get("GEYSER_API_TOKEN")
geyser_auth_type = os.environ.get("GEYSER_AUTH_TYPE", "x-token")
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")
listener_tasks = [
listen_with_timeout(block_listener, block_callback, test_duration),
listen_with_timeout(logs_listener, logs_callback, test_duration)
]
callbacks = [block_callback, logs_callback]
listener_names = ["BlockListener", "LogsListener"]
# Initialize Geyser listener if credentials are available
if geyser_endpoint and geyser_api_token:
logger.info(f"Connecting to Geyser API: {geyser_endpoint}")
geyser_listener = GeyserListener(geyser_endpoint, geyser_api_token, geyser_auth_type, PumpAddresses.PROGRAM)
geyser_callback = TimingTokenCallback("GeyserListener")
listener_tasks.append(
listen_with_timeout(geyser_listener, geyser_callback, test_duration)
)
callbacks.append(geyser_callback)
listener_names.append("GeyserListener")
else:
logger.warning("Geyser API credentials not found. Running without Geyser listener.")
logger.info("Starting all listeners simultaneously...")
logger.info(f"Comparison running for {test_duration} seconds...")
try:
# Start all listeners at the same time
start_time = time.time()
await asyncio.gather(*listener_tasks)
end_time = time.time()
logger.info(f"Test completed in {end_time - start_time:.2f} seconds")
except KeyboardInterrupt:
logger.info("Test interrupted by user")
# No need for explicit cancellation as gather() will be interrupted
for i, callback in enumerate(callbacks):
logger.info(f"{listener_names[i]} detected {len(callback.detected_tokens)} tokens")
# Find tokens detected by multiple listeners
all_mints = {}
for i, callback in enumerate(callbacks):
mints = {str(token.mint) for token in callback.detected_tokens}
all_mints[listener_names[i]] = mints
# Analyze common detections between all listeners
if len(callbacks) > 1:
logger.info("\nAnalyzing token detection across listeners:")
# Find tokens detected by all listeners
if len(callbacks) > 2: # If we have all 3 listeners
common_to_all = set.intersection(*all_mints.values())
logger.info(f"Tokens detected by all listeners: {len(common_to_all)}")
# Compare pairs of listeners
listeners = list(all_mints.keys())
for i in range(len(listeners)):
for j in range(i+1, len(listeners)):
listener1 = listeners[i]
listener2 = listeners[j]
common = all_mints[listener1].intersection(all_mints[listener2])
logger.info(f"Tokens detected by both {listener1} and {listener2}: {len(common)}")
unique1 = all_mints[listener1] - all_mints[listener2]
unique2 = all_mints[listener2] - all_mints[listener1]
logger.info(f"Tokens unique to {listener1}: {len(unique1)}")
logger.info(f"Tokens unique to {listener2}: {len(unique2)}")
# Find tokens detected by at least one listener
all_detected = set.union(*all_mints.values())
logger.info(f"Total unique tokens detected by any listener: {len(all_detected)}")
logger.info("\nDetection speed comparison:")
# Collect all tokens detected by at least two listeners
detection_comparisons = []
for mint in set.union(*all_mints.values()):
detections = {}
for i, callback in enumerate(callbacks):
if mint in callback.detection_times:
detections[listener_names[i]] = callback.detection_times[mint]
if len(detections) > 1: # Only consider tokens detected by multiple listeners
detection_comparisons.append((mint, detections))
if detection_comparisons:
logger.info("Token | " + " | ".join(listener_names) + " | Fastest")
logger.info("-" * 80)
for mint, detections in detection_comparisons:
# Create row with detection times or "N/A" if not detected
times = []
for name in listener_names:
time_str = f"{detections.get(name, 0):.6f}" if name in detections else "N/A"
times.append(time_str)
# Determine fastest listener
valid_times = {name: time for name, time in detections.items() if time > 0}
fastest = min(valid_times.items(), key=lambda x: x[1])[0] if valid_times else "N/A"
logger.info(f"{mint[:10]}... | " + " | ".join(times) + f" | {fastest}")
else:
logger.info("No tokens were detected by multiple listeners for timing comparison")
if __name__ == "__main__":
test_duration = 60 # 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
-102
View File
@@ -1,102 +0,0 @@
"""
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
from dotenv import load_dotenv
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
load_dotenv()
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"User: {token_info.user}")
print(f"Creator: {token_info.creator}")
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 = 30
logger.info("Starting block listener test (using blockSubscribe)")
asyncio.run(test_block_listener(match_string, creator_address, test_duration))
tests/test_geyser_listener.py
-109
View File
@@ -1,109 +0,0 @@
"""
Test script for GeyserListener
Tests gRPC monitoring for new pump.fun tokens using Geyser
"""
import asyncio
import logging
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
sys.path.append(str(Path(__file__).parent.parent / "src"))
from core.pubkeys import PumpAddresses
from monitoring.geyser_listener import GeyserListener
from trading.base import TokenInfo
load_dotenv()
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("geyser-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"User: {token_info.user}")
print(f"Creator: {token_info.creator}")
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_geyser_listener(
match_string: str | None = None,
creator_address: str | None = None,
test_duration: int = 60,
):
"""Test the Geyser listener functionality"""
geyser_endpoint = os.environ.get("GEYSER_ENDPOINT")
geyser_api_token = os.environ.get("GEYSER_API_TOKEN")
geyser_auth_type = os.environ.get("GEYSER_AUTH_TYPE", "x-token")
if not geyser_endpoint:
logger.error("GEYSER_ENDPOINT environment variable is not set")
return []
if not geyser_api_token:
logger.error("GEYSER_API_TOKEN environment variable is not set")
return []
logger.info(f"Connecting to Geyser API: {geyser_endpoint}")
listener = GeyserListener(geyser_endpoint, geyser_api_token, geyser_auth_type, 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 = 30
logger.info("Starting Geyser listener test (using Geyser API)")
asyncio.run(test_geyser_listener(match_string, creator_address, test_duration))
tests/test_logs_listener.py
-102
View File
@@ -1,102 +0,0 @@
"""
Test script for LogsListener
Tests websocket monitoring for new pump.fun tokens using logsSubscribe
"""
import asyncio
import logging
import os
import sys
from pathlib import Path
from dotenv import load_dotenv
sys.path.append(str(Path(__file__).parent.parent / "src"))
from core.pubkeys import PumpAddresses
from monitoring.logs_listener import LogsListener
from trading.base import TokenInfo
load_dotenv()
logging.basicConfig(
level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger("logs-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"User: {token_info.user}")
print(f"Creator: {token_info.creator}")
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_logs_listener(
match_string: str | None = None,
creator_address: str | None = None,
test_duration: int = 60,
):
"""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 = LogsListener(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 = 30
logger.info("Starting logs listener test (using logsSubscribe)")
asyncio.run(test_logs_listener(match_string, creator_address, test_duration))