Does anyone here know how to check if the given token mint address is already tradable using WebSocket? My current code polls the token holders and checks if Raydium already exists as a holder. But it is very slow for sniping, I don't think that is the best way to do it.
I know how to check for newly added liquidity pools by subscribing to the RAYDIUM_V4_ADDRESS using WebSocket. However, I am not looking to subscribe to newly added liquidity pools for all tokens. I want to subscribe to a specific token mint address, but I don't know how to determine if the logs are for Raydium trading allowance.
Could someone advise me on how to do that using WebSocket? Thanks!
My current code that uses polling:
@async_timed
async def check_raydium_authority_for_token_retry(
mint_address: Pubkey,
duration_hours: float = CHECK_RAYDIUM_AUTHORITY_DURATION_HOURS
) -> bool or None:
"""
Check for the presence of the Raydium Authority V4 address for a token mint
and keep checking for up to 'duration_hours'.
"""
opts = TokenAccountOpts(mint=mint_address)
end_time = datetime.datetime.now() + datetime.timedelta(hours=duration_hours)
while datetime.datetime.now() < end_time:
response = await solana_async_client.get_token_accounts_by_owner_json_parsed(
owner=RAYDIUM_AUTHORITY_V4_ADDRESS,
opts=opts,
commitment=Finalized,
)
# Safely access 'value' attribute, defaulting to None if it's not present
response_value = getattr(response, 'value', None)
if response_value:
for account_info in response_value:
account_data = account_info.account.data
if isinstance(account_data, ParsedAccount):
parsed_info = account_data.parsed
if "info" in parsed_info and parsed_info["info"]["mint"] == str(mint_address):
LOGGER.success(f"Raydium Authority V4 address found for token")
return True
await asyncio.sleep(CHECK_RAYDIUM_AUTHORITY_DELAY_SECONDS)
LOGGER.error(f"Raydium Authority V4 address not found for token within {duration_hours} hour(s).")
return None
