Fetch newly listed created pairs on Raydium

Question

I am facing following error:- Error finding latest created pair: argument 'program': 'str' object cannot be converted to 'Pubkey' when i try convert string to PublicKey.. It starts giving me import error of :- from solana.publickey import PublicKey

i have tried to import PublicKey with many alternative ways. However, none has seemed to work. Any help in this regard will be appreciated.

#Here is the Code:-

from solana.rpc.api import Client

rpc_url = 'https://api.mainnet-beta.solana.com/' solana_client = Client(rpc_url)

RAYDIUM_PROGRAM_ID = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"

def find_latest_created_pair(): try: raydium_accounts = solana_client.get_program_accounts(RAYDIUM_PROGRAM_ID)

    latest_created_pair = max(
        raydium_accounts,
        key=lambda account: account.get('data', {}).get('created_at', 0)
    )

    print('Latest Created Pair:', latest_created_pair)
except Exception as e:
    print('Error finding latest created pair:', e)

find_latest_created_pair()

Answer 1

Bitquery provides Solana APIs for real-time data, including information on Raydium and Orca. Docs link here.

For newly created pairs, check this graphQL query

The developer plan is free and sufficient for testing. You can sign up here.

Full disclosure, I work at Bitquery.

Answer 2

If Pubkey is not available directly from the solana package, you may be using a newer version of solana-py, which requires types from the solders package. See the note at the README:

Note: This library uses many core types from the Solders package which used to be provided by solana-py itself. If you are upgrading from an old version and you're looking for something that was deleted, it's probably in solders now.

So to fix this, you just need to add solders to your project, then add:

from solders.pubkey import Pubkey

Then you'll be able to convert your string to a Pubkey with:

RAYDIUM_PROGRAM_ID = Pubkey.from_string("675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8")

Answer 3

Here is my solution to this

import asyncio
import sys

import websockets
import json
from solana.rpc.api import Client
from solders.pubkey import Pubkey
from solders.signature import Signature
import pandas as pd
from tabulate import tabulate


wallet_address = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"
seen_signatures = set()
solana_client = Client("https://api.mainnet-beta.solana.com")

def  getTokens(str_signature):
    signature = Signature.from_string(str_signature)
    transaction = solana_client.get_transaction(signature, encoding="jsonParsed",max_supported_transaction_version=0).value
    instruction_list = transaction.transaction.transaction.message.instructions
    if any("initialize2" in message for message in transaction.transaction.meta.log_messages):
        for instructions in instruction_list:
            if instructions.program_id == Pubkey.from_string(wallet_address):
                print("============NEW POOL DETECTED====================")
                Token0= instructions.accounts[8]
                Token1= instructions.accounts[9]
                # Your data
                data = {'Token_Index': ['Token0', 'Token1'],
                        'Account Public Key': [Token0, Token1]}

                df = pd.DataFrame(data)
                table = tabulate(df, headers='keys', tablefmt='fancy_grid')
                print(table)
             


async def run():
   uri = "wss://api.mainnet-beta.solana.com"
   async with websockets.connect(uri) as websocket:
       # Send subscription request
       await websocket.send(json.dumps({
           "jsonrpc": "2.0",
           "id": 1,
           "method": "logsSubscribe",
           "params": [
               {"mentions": [wallet_address]},
               {"commitment": "finalized"}
           ]
       }))

       first_resp = await websocket.recv()
       response_dict = json.loads(first_resp)
       if 'result' in response_dict:
          print("Subscription successful. Subscription ID: ", response_dict['result'])

       # Continuously read from the WebSocket
       async for response in websocket:
          
           response_dict = json.loads(response)

           if response_dict['params']['result']['value']['err'] == None :
               signature = response_dict['params']['result']['value']['signature']

               if signature not in seen_signatures:
                  seen_signatures.add(signature)
                  log_messages_set = set(response_dict['params']['result']['value']['logs'])

                  search="initialize2"
                  if any(search in message for message in log_messages_set):
                      print("True",signature)
                      getTokens(signature)

                 
           else:
               pass


async def main():
    await run()

asyncio.run(main())

Github repo can be found here