diff --git a/learning-examples/check_boding_curve_status.py b/learning-examples/bonding-curve-progress/get_bonding_curve_status.py similarity index 52% rename from learning-examples/check_boding_curve_status.py rename to learning-examples/bonding-curve-progress/get_bonding_curve_status.py index d66fafa..29974f3 100644 --- a/learning-examples/check_boding_curve_status.py +++ b/learning-examples/bonding-curve-progress/get_bonding_curve_status.py @@ -1,25 +1,42 @@ -import argparse +""" +Module for checking the status of a token's bonding curve on the Solana network using +the Pump.fun program. It allows querying the bonding curve state and completion status. +""" + import asyncio import os import struct -import sys from typing import Final from construct import Flag, Int64ul, Struct +from dotenv import load_dotenv from solana.rpc.async_api import AsyncClient from solders.pubkey import Pubkey -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) - -from core.pubkeys import PumpAddresses - -# Constants -EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" tuple[Pubkey, int]: """ - Derives the associated bonding curve address for a given mint + Derives the associated bonding curve address for a given mint. + + Args: + mint: The token mint address + program_id: The program ID for the bonding curve + + Returns: + Tuple of (bonding curve address, bump seed) """ return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id) @@ -46,6 +70,19 @@ def get_associated_bonding_curve_address( async def get_bonding_curve_state( conn: AsyncClient, curve_address: Pubkey ) -> BondingCurveState: + """ + Fetches and validates the state of a bonding curve account. + + Args: + conn: AsyncClient connection to Solana RPC + curve_address: Address of the bonding curve account + + Returns: + BondingCurveState object containing parsed account data + + Raises: + ValueError: If account data is invalid or missing + """ response = await conn.get_account_info(curve_address, encoding="base64") if not response.value or not response.value.data: raise ValueError("Invalid curve state: No data") @@ -58,19 +95,25 @@ async def get_bonding_curve_state( async def check_token_status(mint_address: str) -> None: + """ + Checks and prints the status of a token and its bonding curve. + + Args: + mint_address: The token mint address as a string + """ try: mint = Pubkey.from_string(mint_address) # Get the associated bonding curve address bonding_curve_address, bump = get_associated_bonding_curve_address( - mint, PumpAddresses.PROGRAM + mint, PUMP_PROGRAM_ID ) - print("\nToken Status:") + print("\nToken status:") print("-" * 50) - print(f"Token Mint: {mint}") - print(f"Associated Bonding Curve: {bonding_curve_address}") - print(f"Bump Seed: {bump}") + print(f"Token mint: {mint}") + print(f"Associated bonding curve: {bonding_curve_address}") + print(f"Bump seed: {bump}") print("-" * 50) # Check completion status @@ -80,14 +123,14 @@ async def check_token_status(mint_address: str) -> None: client, bonding_curve_address ) - print("\nBonding Curve Status:") + print("\nBonding curve status:") print("-" * 50) print( - f"Completion Status: {'Completed' if curve_state.complete else 'Not Completed'}" + f"Completion status: {'Completed' if curve_state.complete else 'Not completed'}" ) if curve_state.complete: print( - "\nNote: This bonding curve has completed and liquidity has been migrated to Raydium." + "\nNote: This bonding curve has completed and liquidity has been migrated to PumpSwap." ) print("-" * 50) @@ -100,12 +143,13 @@ async def check_token_status(mint_address: str) -> None: print(f"\nUnexpected error: {e}") -def main(): - parser = argparse.ArgumentParser(description="Check token bonding curve status") - parser.add_argument("mint_address", help="The token mint address") - - args = parser.parse_args() - asyncio.run(check_token_status(args.mint_address)) +def main() -> None: + """Main entry point for the token status checker.""" + #parser = argparse.ArgumentParser(description="Check token bonding curve status") + #parser.add_argument("mint_address", help="The token mint address" + #args = parser.parse_args() + + asyncio.run(check_token_status(TOKEN_MINT)) if __name__ == "__main__": diff --git a/learning-examples/get_graduating_tokens.py b/learning-examples/bonding-curve-progress/get_graduating_tokens.py similarity index 59% rename from learning-examples/get_graduating_tokens.py rename to learning-examples/bonding-curve-progress/get_graduating_tokens.py index df72be3..e6d2540 100644 --- a/learning-examples/get_graduating_tokens.py +++ b/learning-examples/bonding-curve-progress/get_graduating_tokens.py @@ -1,6 +1,15 @@ +""" +Module for querying and analyzing soon-to-gradute tokens in the Pump.fun program. +It includes functionality to fetch bonding curves based on token reserves and +find associated SPL token accounts. + +Note: getProgramAccounts may be slow as it is a pretty heavy method for RPC. +""" + import asyncio import os import struct +from typing import Final from dotenv import load_dotenv from solana.rpc.async_api import AsyncClient @@ -9,19 +18,18 @@ from solders.pubkey import Pubkey load_dotenv() -RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT1") -PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") -TOKEN_PROGRAM_ID = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA") +# Constants +RPC_ENDPOINT: Final[str] = os.environ.get("SOLANA_NODE_RPC_ENDPOINT") +PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") +TOKEN_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA") # The 8-byte discriminator for bonding curve accounts in Pump.fun -# Used to identify account types in getProgramAccounts requests -BONDING_CURVE_DISCRIMINATOR_BYTES = bytes.fromhex("17b7f83760d8ac60") +BONDING_CURVE_DISCRIMINATOR_BYTES: Final[bytes] = bytes.fromhex("17b7f83760d8ac60") async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> list: """ - Fetch bonding curve accounts from the Pump.fun program that have real_token_reserves - below a defined threshold and are not marked as complete (i.e., not migrated). + Fetch bonding curve accounts with real token reserves below a threshold. Args: client: Optional AsyncClient instance. If None, a new one will be created. @@ -29,31 +37,22 @@ async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> l Returns: List of bonding curve accounts matching the criteria """ - # Define the reserve threshold we're interested in - # 100 trillion (in token base units) - threshold = 100_000_000_000_000 - - # Convert the threshold into 8-byte little-endian format (as stored on-chain) - threshold_bytes = threshold.to_bytes(8, 'little') - - # Extract the 2 most significant bytes to pre-filter values less than 2^48 (~281T) - # This optimization reduces the number of accounts returned by the RPC - msb_prefix = threshold_bytes[6:] + # Define the reserve threshold (100 trillion in token base units) + threshold: int = 100_000_000_000_000 + threshold_bytes: bytes = threshold.to_bytes(8, "little") + msb_prefix: bytes = threshold_bytes[6:] # Most significant bytes for pre-filtering - should_close_client = client is None + should_close_client: bool = client is None try: if should_close_client: - client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120) + client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=180) await client.is_connected() # Define on-chain filters for getProgramAccounts filters = [ - # Match only bonding curve accounts - MemcmpOpts(offset=0, bytes=BONDING_CURVE_DISCRIMINATOR_BYTES), - # Real token reserves MSB bytes (pre-filter) - MemcmpOpts(offset=30, bytes=msb_prefix), - # Complete flag is False (not migrated) - MemcmpOpts(offset=48, bytes=b'\x00'), + MemcmpOpts(offset=0, bytes=BONDING_CURVE_DISCRIMINATOR_BYTES), # Match bonding curve accounts + MemcmpOpts(offset=30, bytes=msb_prefix), # Pre-filter by real token reserves MSB + MemcmpOpts(offset=48, bytes=b"\x00"), # Ensure complete flag is False ] # Query accounts matching filters @@ -67,26 +66,15 @@ async def get_bonding_curves_by_reserves(client: AsyncClient | None = None) -> l for acc in response.value: raw = acc.account.data - # Parse account data according to the Pump.fun bonding curve layout: - # [8] discriminator - # [8] virtual_token_reserves (u64) - # [8] virtual_sol_reserves (u64) - # [8] real_token_reserves (u64) - # [8] real_sol_reserves (u64) - # [8] token_total_supply (u64) - # [1] complete (bool) - - # Skip to real_token_reserves field (8 + 8 + 8 = 24 bytes offset) - offset = 24 - # Extract real_token_reserves (u64 = 8 bytes, little-endian) - real_token_reserves = struct.unpack(" l await client.close() -async def find_associated_bonding_curve(bonding_curve_address: str, client: AsyncClient | None = None): +async def find_associated_bonding_curve( + bonding_curve_address: str, client: AsyncClient | None = None +) -> dict | None: """ Find the SPL token account owned by a bonding curve. - A bonding curve typically owns exactly one SPL token account that represents - the token being traded through the curve. - Args: bonding_curve_address: The bonding curve public key (as a string) client: Optional AsyncClient instance. If None, a new one will be created. @@ -109,7 +96,7 @@ async def find_associated_bonding_curve(bonding_curve_address: str, client: Asyn Returns: The associated SPL token account data or None if not found """ - should_close_client = client is None + should_close_client: bool = client is None try: if should_close_client: client = AsyncClient(RPC_ENDPOINT) @@ -137,26 +124,23 @@ def get_mint_address(data: bytes) -> str: """ Extract the mint address from SPL token account data. - In SPL token account data, the mint address is stored in the first 32 bytes. - Args: data: The token account data as bytes Returns: The mint address as a base58-encoded string """ - # The mint address is stored in the first 32 bytes - mint_bytes = data[:32] - return str(Pubkey(mint_bytes)) + return str(Pubkey(data[:32])) -async def main(): +async def main() -> None: + """Main entry point for querying and processing bonding curves.""" async with AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120) as client: await client.is_connected() bonding_curves = await get_bonding_curves_by_reserves(client) print(f"Total matches: {len(bonding_curves)}") - print("="*50) + print("=" * 50) for bonding_curve in bonding_curves: # Find the SPL token account owned by the bonding curve @@ -168,7 +152,7 @@ async def main(): mint_address = get_mint_address(associated_token_account.data) print(f"Bonding curve: {bonding_curve.pubkey}") print(f"Mint address: {mint_address}") - print("="*50) + print("=" * 50) # For demonstration, only process the first curve break diff --git a/learning-examples/bonding-curve-progress/poll_bonding_curve_progress.py b/learning-examples/bonding-curve-progress/poll_bonding_curve_progress.py new file mode 100644 index 0000000..1ab52c3 --- /dev/null +++ b/learning-examples/bonding-curve-progress/poll_bonding_curve_progress.py @@ -0,0 +1,135 @@ +""" +Module for tracking the progress of a bonding curve for a Pump.fun token. +It continuously polls the bonding curve state and prints updates at regular intervals. +""" + +import asyncio +import os +import struct +from typing import Final + +from dotenv import load_dotenv +from solana.rpc.async_api import AsyncClient +from solders.pubkey import Pubkey + +load_dotenv() + +# Constants +RPC_URL: Final[str] = os.getenv("SOLANA_NODE_RPC_ENDPOINT") +TOKEN_MINT: Final[str] = "xWrzYY4c1LnbSkLrd2LDUg9vw7YtVyJhGmw7MABpump" +PUMP_PROGRAM_ID: Final[Pubkey] = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") +LAMPORTS_PER_SOL: Final[int] = 1_000_000_000 +TOKEN_DECIMALS: Final[int] = 6 +EXPECTED_DISCRIMINATOR: Final[bytes] = struct.pack(" Pubkey: + """ + Derive the bonding curve PDA address from a mint address. + + Args: + mint: The token mint address + program_id: The program ID for the bonding curve + + Returns: + The bonding curve address + """ + return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)[0] + + +async def get_account_data(client: AsyncClient, pubkey: Pubkey) -> bytes: + """ + Fetch raw account data for a given public key. + + Args: + client: AsyncClient connection to Solana RPC + pubkey: The public key of the account to fetch + + Returns: + The raw account data as bytes + + Raises: + ValueError: If the account is not found or has no data + """ + resp = await client.get_account_info(pubkey, encoding="base64") + if not resp.value or not resp.value.data: + raise ValueError(f"Account {pubkey} not found or has no data") + + return resp.value.data + + +def parse_curve_state(data: bytes) -> dict: + """ + Decode bonding curve account data into a readable format. + + Args: + data: The raw bonding curve account data + + Returns: + A dictionary containing parsed bonding curve fields + + Raises: + ValueError: If the account discriminator is invalid + """ + if data[:8] != EXPECTED_DISCRIMINATOR: + raise ValueError("Invalid discriminator for bonding curve") + + fields = struct.unpack_from(" None: + """ + Print the current status of the bonding curve in a readable format. + + Args: + state: The parsed bonding curve state dictionary + """ + progress = 0 + if state["token_total_supply"]: + progress = 100 - (100 * state["real_token_reserves"] / state["token_total_supply"]) + + print("=" * 30) + print(f"Complete: {'✅' if state['complete'] else '❌'}") + print(f"Progress: {progress:.2f}%") + print(f"Token reserves: {state['real_token_reserves']:.4f}") + print(f"SOL reserves: {state['real_sol_reserves']:.4f}") + print("=" * 30, "\n") + + +async def track_curve() -> None: + """ + Continuously track and display the state of a bonding curve. + """ + if not RPC_URL or not TOKEN_MINT: + print("❌ Set SOLANA_NODE_RPC_ENDPOINT and TOKEN_MINT in .env") + return + + mint_pubkey: Pubkey = Pubkey.from_string(TOKEN_MINT) + curve_pubkey: Pubkey = get_associated_bonding_curve_address(mint_pubkey, PUMP_PROGRAM_ID) + + print("Tracking bonding curve for:", mint_pubkey) + print("Curve address:", curve_pubkey, "\n") + + async with AsyncClient(RPC_URL) as client: + while True: + try: + data = await get_account_data(client, curve_pubkey) + state = parse_curve_state(data) + print_curve_status(state) + except Exception as e: + print(f"⚠️ Error: {e}") + + await asyncio.sleep(POLL_INTERVAL) + + +if __name__ == "__main__": + asyncio.run(track_curve()) diff --git a/learning-examples/listen-migrations/compare_migration_listeners.py b/learning-examples/listen-migrations/compare_migration_listeners.py new file mode 100644 index 0000000..afe3a4a --- /dev/null +++ b/learning-examples/listen-migrations/compare_migration_listeners.py @@ -0,0 +1,637 @@ +""" +This script compares two methods of detecting migrations: +1. Migration program listener (listens Migration program) - detects markets via successful migration transactions +2. Direct market account listener (listens Pump Fun AMM program aka PumpSwap) - detects markets via program account subscription + +The script tracks which method detects new markets first and provides detailed performance statistics. + +Note: multiple endpoints available. Scroll down to change providers which you want to test. +""" + +import asyncio +import base64 +import json +import os +import struct +import time + +import aiohttp +import base58 +import websockets +from dotenv import load_dotenv +from solders.pubkey import Pubkey + +load_dotenv() + +RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT") +MIGRATION_PROGRAM_ID = Pubkey.from_string("39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg") +PUMP_AMM_PROGRAM_ID = Pubkey.from_string("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA") +QUOTE_MINT_SOL = base58.b58encode(bytes(Pubkey.from_string("So11111111111111111111111111111111111111112"))).decode() + +MARKET_DISCRIMINATOR = base58.b58encode(b'\xf1\x9am\x04\x11\xb1m\xbc').decode() +MARKET_ACCOUNT_LENGTH = 8 + 1 + 2 + 32 * 6 + 8 # total size of known market structure + + +class DetectionTracker: + """Tracks and analyzes detection times for both methods across providers""" + + def __init__(self): + self.migrations = {} # {base_mint: {provider: timestamp}} + self.markets = {} # {base_mint: {provider: timestamp}} + self.migration_messages = {} # {provider: count} + self.market_messages = {} # {provider: count} + self.start_time = time.time() + + def add_migration(self, base_mint, provider, timestamp): + """Record a migration detection event""" + if base_mint not in self.migrations: + self.migrations[base_mint] = {} + self.migrations[base_mint][provider] = timestamp + print(f"[MIGRATION] base_mint={base_mint} provider={provider} time={timestamp:.3f}") + + def add_market(self, base_mint, provider, timestamp): + """Record a market detection event""" + if base_mint not in self.markets: + self.markets[base_mint] = {} + self.markets[base_mint][provider] = timestamp + print(f"[MARKET] base_mint={base_mint} provider={provider} time={timestamp:.3f}") + + def increment_migration_messages(self, provider): + """Count WebSocket messages received by migration listener""" + if provider not in self.migration_messages: + self.migration_messages[provider] = 0 + self.migration_messages[provider] += 1 + + def increment_market_messages(self, provider): + """Count WebSocket messages received by market listener""" + if provider not in self.market_messages: + self.market_messages[provider] = 0 + self.market_messages[provider] += 1 + + def print_summary(self): + """Print detailed summary statistics of the comparison test""" + test_duration = time.time() - self.start_time + + # Count total messages + total_migration_messages = sum(self.migration_messages.values()) + total_market_messages = sum(self.market_messages.values()) + + print("\n=== Test Summary ===") + print(f"Test duration: {test_duration:.2f} seconds") + print(f"WebSocket messages received: {total_migration_messages + total_market_messages}") + print(f" - Migration events: {total_migration_messages}") + print(f" - Market events: {total_market_messages}") + + # Count unique tokens detected by each method + unique_migrations = set(self.migrations.keys()) + unique_markets = set(self.markets.keys()) + common_tokens = unique_migrations & unique_markets + + print(f"Tokens detected: {len(unique_migrations | unique_markets)}") + print(f" - Migration events: {len(unique_migrations)}") + print(f" - Market events: {len(unique_markets)}") + print(f" - Detected in both: {len(common_tokens)}\n") + + print("=== Provider Message Counts ===") + print("Provider | Migration Messages | Market Messages | Total Messages") + print("-" * 80) + all_providers = set(self.migration_messages.keys()) | set(self.market_messages.keys()) + for provider in sorted(all_providers): + migration_count = self.migration_messages.get(provider, 0) + market_count = self.market_messages.get(provider, 0) + total = migration_count + market_count + print(f"{provider:<22} | {migration_count:<18} | {market_count:<14} | {total}") + print() + + print("=== Migration Event Provider Performance ===") + self._print_provider_performance(self.migrations) + + print("\n=== Market Event Provider Performance ===") + self._print_provider_performance(self.markets) + + # Compare detection methods for tokens detected by both + if common_tokens: + print("\n=== Detection Timing Comparison: Migration vs Market ===") + print("Base Mint | First Detection Method | First Provider | Time Delta (ms)") + print("-" * 100) + + migration_first = 0 + market_first = 0 + total_delta_ms = 0 + + for base_mint in sorted(common_tokens): + # Find earliest time for each method + migration_time = min(self.migrations[base_mint].values()) if base_mint in self.migrations else float('inf') + market_time = min(self.markets[base_mint].values()) if base_mint in self.markets else float('inf') + + # Find provider with earliest time for each method + migration_provider = None + if base_mint in self.migrations: + migration_provider = min(self.migrations[base_mint].items(), key=lambda x: x[1])[0] + + market_provider = None + if base_mint in self.markets: + market_provider = min(self.markets[base_mint].items(), key=lambda x: x[1])[0] + + delta_ms = abs(migration_time - market_time) * 1000 + total_delta_ms += delta_ms + + if migration_time < market_time: + first_method = "Migration" + first_provider = migration_provider + migration_first += 1 + else: + first_method = "Market" + first_provider = market_provider + market_first += 1 + + print(f"{base_mint} | {first_method:<21} | {first_provider:<14} | {delta_ms:8.1f}") + + # Print statistics summary + if common_tokens: + avg_delta_ms = total_delta_ms / len(common_tokens) + print("\nSummary statistics:") + print(f" - Migration detected first: {migration_first}/{len(common_tokens)} ({migration_first/len(common_tokens)*100:.1f}%)") + print(f" - Market detected first: {market_first}/{len(common_tokens)} ({market_first/len(common_tokens)*100:.1f}%)") + print(f" - Average timing difference: {avg_delta_ms:.1f} ms") + + def _print_provider_performance(self, events_dict): + """Print performance metrics for providers using a specific detection method""" + # Count how many times each provider was first + first_count = {} + total_events = 0 + + for base_mint, providers in events_dict.items(): + total_events += 1 + if not providers: + continue + + # Find the fastest provider for this event + fastest_provider = min(providers.items(), key=lambda x: x[1])[0] + if fastest_provider not in first_count: + first_count[fastest_provider] = 0 + first_count[fastest_provider] += 1 + + if not first_count: + print("No events detected") + return + + # Print rankings + print("Provider | First Detections | Percentage") + print("-" * 60) + + for provider, count in sorted(first_count.items(), key=lambda x: x[1], reverse=True): + percentage = (count / total_events) * 100 if total_events > 0 else 0 + print(f"{provider:<22} | {count:<16} | {percentage:.1f}%") + + # Calculate average latency between providers + self._print_provider_latency_matrix(events_dict) + + def _print_provider_latency_matrix(self, events_dict): + """Print a matrix of average latency between providers""" + # Get unique providers + all_providers = set() + for providers_data in events_dict.values(): + all_providers.update(providers_data.keys()) + + if len(all_providers) <= 1: + return + + providers_list = sorted(all_providers) + + print("\nAverage Latency Matrix (ms):") + # Print header + header = " |" + for provider in providers_list: + header += f" {provider[:8]:>8} |" + print(header) + print("-" * len(header)) + + # Calculate and print latency matrix + for provider1 in providers_list: + row = f"{provider1[:8]:>8} |" + for provider2 in providers_list: + if provider1 == provider2: + row += " — |" + continue + + # Calculate average latency + latencies = [] + for base_mint, providers_data in events_dict.items(): + if provider1 in providers_data and provider2 in providers_data: + latency_ms = (providers_data[provider2] - providers_data[provider1]) * 1000 + latencies.append(latency_ms) + + if latencies: + avg_latency = sum(latencies) / len(latencies) + row += f" {avg_latency:>+7.1f} |" + else: + row += " ? |" + print(row) + + +# ============ MARKET DETECTION METHODS ============ + +async def fetch_existing_market_pubkeys(): + """ + Fetch existing AMM market accounts from the blockchain + + Used to filter out already existing markets when detecting new ones + """ + headers = {"Content-Type": "application/json"} + body = { + "jsonrpc": "2.0", + "id": 1, + "method": "getProgramAccounts", + "params": [ + str(PUMP_AMM_PROGRAM_ID), + { + "encoding": "base64", + "commitment": "processed", + "filters": [ + {"dataSize": MARKET_ACCOUNT_LENGTH}, + {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}}, + {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}} + ] + } + ] + } + + async with aiohttp.ClientSession() as session: + async with session.post(RPC_ENDPOINT, headers=headers, json=body) as resp: + res = await resp.json() + return {account["pubkey"] for account in res.get("result", [])} + + +def parse_market_account_data(data): + """ + Parse binary market account data into a structured format + + This function matches the parser from the market listener script + """ + parsed_data = {} + offset = 8 # Skip discriminator + + fields = [ + ("pool_bump", "u8"), + ("index", "u16"), + ("creator", "pubkey"), + ("base_mint", "pubkey"), + ("quote_mint", "pubkey"), + ("lp_mint", "pubkey"), + ("pool_base_token_account", "pubkey"), + ("pool_quote_token_account", "pubkey"), + ("lp_supply", "u64"), + ] + + try: + for field_name, field_type in fields: + if field_type == "pubkey": + value = data[offset:offset + 32] + parsed_data[field_name] = base58.b58encode(value).decode("utf-8") + offset += 32 + elif field_type in {"u64", "i64"}: + value = struct.unpack(" no parsed data). +To cover those cases, please use an additional RPC call (get transaction data) or additional listener not based on logs. +""" + import asyncio import base64 import json @@ -6,8 +14,11 @@ import struct import base58 import websockets +from dotenv import load_dotenv from solders.pubkey import Pubkey +load_dotenv() + WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") MIGRATION_PROGRAM_ID = Pubkey.from_string("39azUYFWPz3VHgKCf3VChUwbpURdCHRxjWVowf5jUJjg") @@ -77,9 +88,8 @@ def is_transaction_successful(logs): def print_transaction_details(log_data): - signature = log_data.get('signature', 'N/A') - print(f"\n[INFO] Transaction Signature: {signature}") logs = log_data.get("logs", []) + parsed_data = {} for log in logs: if log.startswith("Program data:"): @@ -87,16 +97,14 @@ def print_transaction_details(log_data): data = base64.b64decode(log.split(": ")[1]) parsed_data = parse_migrate_instruction(data) if parsed_data: - print("\n[INFO] Parsed Migration Instruction Data:") + print("[INFO] Parsed from Program data:") for key, value in parsed_data.items(): print(f" {key}: {value}") except Exception as e: - print(f"[ERROR] Base64 decode failed: {log}, Error: {e}") + print(f"[ERROR] Failed to decode Program data: {e}") -def print_log_details(logs): - print("\n[INFO] Processing logs:") - for log in logs: - print(f" Log: {log}") + if not parsed_data: + print("[ERROR] Failed to parse migration data: parsed data is empty") async def listen_for_migrations(): @@ -130,16 +138,19 @@ async def listen_for_migrations(): log_data = data["params"]["result"]["value"] logs = log_data.get("logs", []) + signature = log_data.get('signature', 'N/A') + print(f"\n[INFO] Transaction signature: {signature}") + if is_transaction_successful(logs): if not any("Program log: Instruction: Migrate" in log for log in logs): - print("[INFO] Skipping: No Migrate instruction") + print("[INFO] Skipping: no migrate instruction") continue if any("Program log: Bonding curve already migrated" in log for log in logs): - print("[INFO] Skipping: Bonding curve already migrated") + print("[INFO] Skipping: bonding curve already migrated") continue - print("[INFO] Processing migration instruction!") + print("[INFO] Processing migration instruction...") print_transaction_details(log_data) else: print("[INFO] Skipping failed transaction.") @@ -155,4 +166,4 @@ async def listen_for_migrations(): await asyncio.sleep(5) if __name__ == "__main__": - asyncio.run(listen_for_migrations()) + asyncio.run(listen_for_migrations()) \ No newline at end of file diff --git a/learning-examples/listen-migrations/listen_programsubscribe.py b/learning-examples/listen-migrations/listen_programsubscribe.py new file mode 100644 index 0000000..4045703 --- /dev/null +++ b/learning-examples/listen-migrations/listen_programsubscribe.py @@ -0,0 +1,169 @@ +""" +Monitors Solana for new Pump AMM markets via WebSocket. +Fetches existing markets to filter out already existing ones, parses market account data (e.g., mints, token accounts, creator), +and excludes user-created markets. May also detect non-migration-based markets (if they created by a program). + +Note: this method consumes HUGE AMOUNT OF MESSAGES from a WebSocket. +""" + +import asyncio +import base64 +import json +import os +import struct + +import aiohttp +import base58 +import websockets +from dotenv import load_dotenv +from solders.pubkey import Pubkey + +load_dotenv() + +WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") +RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT") +PUMP_AMM_PROGRAM_ID = Pubkey.from_string("pAMMBay6oceH9fJKBRHGP5D4bD4sWpmSwMn52FMfXEA") + +MARKET_ACCOUNT_LENGTH = 8 + 1 + 2 + 32 * 6 + 8 # total size of known market structure +MARKET_DISCRIMINATOR = base58.b58encode(b'\xf1\x9am\x04\x11\xb1m\xbc').decode() +QUOTE_MINT_SOL = base58.b58encode(bytes(Pubkey.from_string("So11111111111111111111111111111111111111112"))).decode() + + +async def fetch_existing_market_pubkeys(): + headers = {"Content-Type": "application/json"} + body = { + "jsonrpc": "2.0", + "id": 1, + "method": "getProgramAccounts", + "params": [ + str(PUMP_AMM_PROGRAM_ID), + { + "encoding": "base64", + "commitment": "processed", + "filters": [ + {"dataSize": MARKET_ACCOUNT_LENGTH}, + {"memcmp": {"offset": 0, "bytes": MARKET_DISCRIMINATOR}}, + {"memcmp": {"offset": 75, "bytes": QUOTE_MINT_SOL}} + ] + } + ] + } + + async with aiohttp.ClientSession() as session: + async with session.post(RPC_ENDPOINT, headers=headers, json=body) as resp: + res = await resp.json() + return {account["pubkey"] for account in res.get("result", [])} + + +def parse_market_account_data(data): + parsed_data = {} + offset = 8 # Discriminator + + fields = [ + ("pool_bump", "u8"), + ("index", "u16"), + ("creator", "pubkey"), + ("base_mint", "pubkey"), + ("quote_mint", "pubkey"), + ("lp_mint", "pubkey"), + ("pool_base_token_account", "pubkey"), + ("pool_quote_token_account", "pubkey"), + ("lp_supply", "u64"), + ] + + try: + for field_name, field_type in fields: + if field_type == "pubkey": + value = data[offset:offset + 32] + parsed_data[field_name] = base58.b58encode(value).decode("utf-8") + offset += 32 + elif field_type in {"u64", "i64"}: + value = struct.unpack(" Pubkey: """ @@ -23,24 +32,14 @@ def find_associated_bonding_curve(mint: Pubkey, bonding_curve: Pubkey) -> Pubkey derived_address, _ = Pubkey.find_program_address( [ bytes(bonding_curve), - bytes(SystemAddresses.TOKEN_PROGRAM), + bytes(TOKEN_PROGRAM_ID), bytes(mint), ], - SystemAddresses.ASSOCIATED_TOKEN_PROGRAM, + ASSOCIATED_TOKEN_PROGRAM_ID, ) return derived_address -# Load the IDL JSON file -with open("idl/pump_fun_idl.json") as f: - idl = json.load(f) - -# Extract the "create" instruction definition -create_instruction = next( - instr for instr in idl["instructions"] if instr["name"] == "create" -) - - def parse_create_instruction(data): if len(data) < 8: return None @@ -75,18 +74,6 @@ def parse_create_instruction(data): return None -def print_transaction_details(log_data): - print(f"Signature: {log_data.get('signature')}") - - for log in log_data.get("logs", []): - if log.startswith("Program data:"): - try: - data = base58.b58decode(log.split(": ")[1]).decode("utf-8") - print(f"Data: {data}") - except: - pass - - async def listen_for_new_tokens(): while True: try: @@ -97,14 +84,14 @@ async def listen_for_new_tokens(): "id": 1, "method": "logsSubscribe", "params": [ - {"mentions": [str(PumpAddresses.PROGRAM)]}, + {"mentions": [str(PUMP_PROGRAM_ID)]}, {"commitment": "processed"}, ], } ) await websocket.send(subscription_message) print( - f"Listening for new token creations from program: {PumpAddresses.PROGRAM}" + f"Listening for new token creations from program: {PUMP_PROGRAM_ID}" ) # Wait for subscription confirmation diff --git a/learning-examples/listen_new_direct.py b/learning-examples/listen-new-tokens/listen_logsubscribe.py similarity index 84% rename from learning-examples/listen_new_direct.py rename to learning-examples/listen-new-tokens/listen_logsubscribe.py index 00962de..6d9506a 100644 --- a/learning-examples/listen_new_direct.py +++ b/learning-examples/listen-new-tokens/listen_logsubscribe.py @@ -1,27 +1,25 @@ +""" +Listens for new Pump.fun token creations via Solana WebSocket. +Monitors logs for 'Create' instructions, decodes and prints token details (name, symbol, mint, etc.). + +It is usually faster than blockSubscribe, but slower than Geyser. +""" + import asyncio import base64 import json import os import struct -import sys import base58 import websockets +from dotenv import load_dotenv +from solders.pubkey import Pubkey -sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))) -from core.pubkeys import PumpAddresses +load_dotenv() WSS_ENDPOINT = os.environ.get("SOLANA_NODE_WSS_ENDPOINT") - - -# Load the IDL JSON file -with open("idl/pump_fun_idl.json") as f: - idl = json.load(f) - -# Extract the "create" instruction definition -create_instruction = next( - instr for instr in idl["instructions"] if instr["name"] == "create" -) +PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") def parse_create_instruction(data): @@ -58,18 +56,6 @@ def parse_create_instruction(data): return None -def print_transaction_details(log_data): - print(f"Signature: {log_data.get('signature')}") - - for log in log_data.get("logs", []): - if log.startswith("Program data:"): - try: - data = base58.b58decode(log.split(": ")[1]).decode("utf-8") - print(f"Data: {data}") - except: - pass - - async def listen_for_new_tokens(): while True: try: @@ -80,14 +66,14 @@ async def listen_for_new_tokens(): "id": 1, "method": "logsSubscribe", "params": [ - {"mentions": [str(PumpAddresses.PROGRAM)]}, + {"mentions": [str(PUMP_PROGRAM_ID)]}, {"commitment": "processed"}, ], } ) await websocket.send(subscription_message) print( - f"Listening for new token creations from program: {PumpAddresses.PROGRAM}" + f"Listening for new token creations from program: {PUMP_PROGRAM_ID}" ) # Wait for subscription confirmation diff --git a/learning-examples/listen_new_portal.py b/learning-examples/listen-new-tokens/listen_pumpportal.py similarity index 97% rename from learning-examples/listen_new_portal.py rename to learning-examples/listen-new-tokens/listen_pumpportal.py index 3817a8d..0bb5c72 100644 --- a/learning-examples/listen_new_portal.py +++ b/learning-examples/listen-new-tokens/listen_pumpportal.py @@ -1,3 +1,7 @@ +""" +Listens for new Pump.fun token creations via PumpPortal WebSocket. +""" + import asyncio import json from datetime import datetime diff --git a/learning-examples/proto/geyser.proto b/learning-examples/listen-new-tokens/proto/geyser.proto similarity index 100% rename from learning-examples/proto/geyser.proto rename to learning-examples/listen-new-tokens/proto/geyser.proto diff --git a/learning-examples/proto/solana-storage.proto b/learning-examples/listen-new-tokens/proto/solana-storage.proto similarity index 100% rename from learning-examples/proto/solana-storage.proto rename to learning-examples/listen-new-tokens/proto/solana-storage.proto diff --git a/learning-examples/get_pumpswap_pools.py b/learning-examples/pumpswap/get_pumpswap_pools.py similarity index 94% rename from learning-examples/get_pumpswap_pools.py rename to learning-examples/pumpswap/get_pumpswap_pools.py index 8b511d1..6578367 100644 --- a/learning-examples/get_pumpswap_pools.py +++ b/learning-examples/pumpswap/get_pumpswap_pools.py @@ -1,3 +1,9 @@ +""" +This module provides functionality to: +1. Find market addresses by base mint +2. Fetch and parse market data (including pool addresses) from Pump AMM program accounts +""" + import asyncio import os import struct diff --git a/learning-examples/manual_sell_pumpswap.py b/learning-examples/pumpswap/manual_sell_pumpswap.py similarity index 97% rename from learning-examples/manual_sell_pumpswap.py rename to learning-examples/pumpswap/manual_sell_pumpswap.py index 566cc56..0340264 100644 --- a/learning-examples/manual_sell_pumpswap.py +++ b/learning-examples/pumpswap/manual_sell_pumpswap.py @@ -1,3 +1,12 @@ +""" +This module provides functionality to: +- Find market addresses by token mint. +- Fetch and parse market data from PUMP AMM pools. +- Calculate token prices in AMM pools. +- Create associated token accounts (ATAs) idempotently. +- Sell tokens on the PUMP AMM with slippage protection. +""" + import asyncio import os import struct diff --git a/learning-examples/track_bonding_curve_progress.py b/learning-examples/track_bonding_curve_progress.py deleted file mode 100644 index 11933c1..0000000 --- a/learning-examples/track_bonding_curve_progress.py +++ /dev/null @@ -1,96 +0,0 @@ -""" -Track bonding curve progress for a pump.fun token by mint address. -""" - -import asyncio -import os -import struct - -from dotenv import load_dotenv -from solana.rpc.async_api import AsyncClient -from solders.pubkey import Pubkey - -# Import pump.fun program address -from core.pubkeys import PumpAddresses - -load_dotenv() - -RPC_URL = os.getenv("SOLANA_NODE_RPC_ENDPOINT") -TOKEN_MINT = "xWrzYY4c1LnbSkLrd2LDUg9vw7YtVyJhGmw7MABpump" - -LAMPORTS_PER_SOL = 1_000_000_000 -TOKEN_DECIMALS = 6 -EXPECTED_DISCRIMINATOR = struct.pack(" Pubkey: - """Derive the bonding curve PDA address from mint.""" - return Pubkey.find_program_address([b"bonding-curve", bytes(mint)], program_id)[0] - - -async def get_account_data(client: AsyncClient, pubkey: Pubkey) -> bytes: - """Fetch raw account data for a given public key.""" - resp = await client.get_account_info(pubkey, encoding="base64") - if not resp.value or not resp.value.data: - raise ValueError(f"Account {pubkey} not found or has no data") - - return resp.value.data - - -def parse_curve_state(data: bytes) -> dict: - """Decode bonding curve account data.""" - if data[:8] != EXPECTED_DISCRIMINATOR: - raise ValueError("Invalid discriminator for bonding curve") - - fields = struct.unpack_from("