2

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()

2
  • What does this code return? the pair address or what?
    – Ivan Dex
    Commented Dec 11, 2023 at 11:45
  • How do you get the pubkey? I am trying the same code above and i continue to get an error with no exception print out.
    – Guruu
    Commented Dec 21, 2023 at 23:37

3 Answers 3

1

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")
1

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

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.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.