diff --git a/learning-examples/get_graduating_tokens.py b/learning-examples/get_graduating_tokens.py index 1695c13..df72be3 100644 --- a/learning-examples/get_graduating_tokens.py +++ b/learning-examples/get_graduating_tokens.py @@ -4,24 +4,31 @@ import struct from dotenv import load_dotenv from solana.rpc.async_api import AsyncClient -from solana.rpc.types import MemcmpOpts +from solana.rpc.types import MemcmpOpts, TokenAccountOpts from solders.pubkey import Pubkey load_dotenv() RPC_ENDPOINT = os.environ.get("SOLANA_NODE_RPC_ENDPOINT1") PUMP_PROGRAM_ID = Pubkey.from_string("6EF8rrecthR5Dkzon8Nwu78hRvfCKubJ14M5uBEwF6P") -# Known 8-byte discriminator for bonding curve accounts -# This ensures we're only fetching bonding curve accounts +TOKEN_PROGRAM_ID = 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") -async def get_bonding_curves_by_reserves(): +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). + + Args: + client: Optional AsyncClient instance. If None, a new one will be created. + + 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 @@ -29,16 +36,24 @@ async def get_bonding_curves_by_reserves(): # 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 (bytes 6 and 7 in little-endian) - # This helps us pre-filter for values less than 2^48 (~281T) + # 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:] - async with AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120) as client: + should_close_client = client is None + try: + if should_close_client: + client = AsyncClient(RPC_ENDPOINT, commitment="processed", timeout=120) + await client.is_connected() + # Define on-chain filters for getProgramAccounts filters = [ - MemcmpOpts(offset=0, bytes=BONDING_CURVE_DISCRIMINATOR_BYTES), # only bonding curve accounts - MemcmpOpts(offset=30, bytes=msb_prefix), # real_token_reserves < ~281T - MemcmpOpts(offset=48, bytes=b'\x00'), # complete == False (not migrated) + # 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'), ] # Query accounts matching filters @@ -52,18 +67,20 @@ async def get_bonding_curves_by_reserves(): for acc in response.value: raw = acc.account.data - # Parse the account layout according to known struct (49 bytes): - # [discriminator][virtual_token_reserves][virtual_sol_reserves] - # [real_token_reserves][real_sol_reserves][token_total_supply][complete] - # Example: https://solscan.io/account/ASwuqjAGjWVhojrdot9yrnXTV4hMdTEryAWkGn4UUJma#anchorData - - offset = 8 # skip 8-byte discriminator - offset += 8 # skip virtual_token_reserves - offset += 8 # skip virtual_sol_reserves - + # 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(" 0: + return response.value[0].account + else: + print(f"No token accounts found for {bonding_curve_address}") + return None + except Exception as e: + print(f"Error finding associated token account: {e}") + return None + finally: + if should_close_client and client: + await client.close() + + +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)) async def main(): - bonding_curves_response = await get_bonding_curves_by_reserves() - print(f"Total matches: {len(bonding_curves_response)}") + 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) + + for bonding_curve in bonding_curves: + # Find the SPL token account owned by the bonding curve + associated_token_account = await find_associated_bonding_curve( + str(bonding_curve.pubkey), client + ) + + if associated_token_account: + mint_address = get_mint_address(associated_token_account.data) + print(f"Bonding curve: {bonding_curve.pubkey}") + print(f"Mint address: {mint_address}") + print("="*50) + + # For demonstration, only process the first curve + break if __name__ == "__main__":