Refactoring, solana-py only

[Artemyin]

Your script was great point to start. I figured it out how solana API works thanks to you.

I did:

• refactor code
• break into functions
• add type hinting
• edit naming for some constants, variables and functions:

Diff

```diff - wallet_address = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8" + RaydiumLPV4 = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8" ... - search = "initialize2" + log_instruction = "initialize2" ... - def getTokens(...) + def get_tokens_info(...) ```

• replaced standard websockets with SolanaWsClientProtocol from solana.rpc.websocket_api

  - import websockets
  + from solana.rpc.websocket_api import connect
  + from solana.rpc.websocket_api import SolanaWsClientProtocol

... async with connect(WSS) as websocket: subscription_id = await subscribe_to_logs( websocket, RpcTransactionLogsFilterMentions(RaydiumLPV4), Finalized ) ...

<details><summary>Diff</summary>
<p>
lines 41-79

```diff
- 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(f"True, https://solscan.io/tx/{signature}")
-                      getTokens(signature)
-
-                 
-           else:
-               pass

lines 32-74

+ async def main():
+     """The client as an infinite asynchronous iterator:"""
+     async with connect(WSS) as websocket:
+         subscription_id = await subscribe_to_logs(
+             websocket,
+             RpcTransactionLogsFilterMentions(RaydiumLPV4),
+             Finalized
+         )  # type: ignore
+         async for signature in process_messages(websocket, log_instruction):  # type: ignore
+             get_tokens(signature, RaydiumLPV4)
+             break
+         await websocket.logs_unsubscribe(subscription_id)
+ 
+ 
+ async def subscribe_to_logs(websocket: SolanaWsClientProtocol, 
+                             mentions: RpcTransactionLogsFilterMentions,
+                             commitment: Commitment) -> int:
+     await websocket.logs_subscribe(
+         filter_=mentions,
+         commitment=commitment
+     )
+     first_resp = await websocket.recv()
+     return get_subscription_id(first_resp)  # type: ignore
+ 
+ 
+ def get_subscription_id(response: SubscriptionResult) -> int:
+     return response[0].result
+ 
+ 
+ async def process_messages(websocket: SolanaWsClientProtocol,
+                            instruction: str) -> AsyncIterator[Signature]:
+     """Async generator, main websocket's loop"""
+     async for idx, msg in enumerate(websocket):
+         value = get_msg_value(msg)
+         print(idx)
+         if [log for log in value.logs if instruction in log]:
+             print(f"{value.signature=}")
+             pprint(value.logs)
+             yield value.signature
+ 
+ 
+ def get_msg_value(msg: List[LogsNotification]) -> RpcLogsResponse:
+     return msg[0].result.value

• • it uses method .logs_subscribe, instead of websocket.send(json.dumps(parametrs))

    async def subscribe_to_logs(websocket: SolanaWsClientProtocol, 
                        mentions: RpcTransactionLogsFilterMentions,
                        commitment: Commitment) -> int:
    await websocket.logs_subscribe(
    filter_=mentions,
    commitment=commitment
    )
    first_resp = await websocket.recv()
    return get_subscription_id(first_resp) 

• • log filtering happend inside .process_messages in list comprehension. That function is generator function, indeed (yield instead return). its yield signature value if its logs align with required log_instruction.

    async def process_messages(websocket: SolanaWsClientProtocol,
                        instruction: str) -> AsyncIterator[Signature]:
    """Async generator, main websocket's loop"""
    async for idx, msg in enumerate(websocket):
    value = get_msg_value(msg)
    print(idx)
    if [log for log in value.logs if instruction in log]:
        print(f"{value.signature=}")
        pprint(value.logs)
        yield value.signature

• function get_tokens call inside for loop which iterate over process_messages generator, so its more obvious entries points and place to future modifications
• substitute pandas data frame to pure python string formatting in print_table function

There is still bunch place for optimization. But at this point its become more consistent due to using one lib for all purpose. also its more easy for debugging and possible testing

P.S. script stops after finding first LP, remove break for infinite loop

        async for signature in process_messages(websocket, log_instruction):  # type: ignore
            get_tokens(signature, RaydiumLPV4)
            break

[henrytirla]

Hey, Thanks for the contribution. I will like to add there is an abrupt stop of the script after a few minutes due to. Too many calls did you take this into consideration?

[Artemyin]

you catch httpx.HTTPStatusError: Client error '429 Too Many Requests' for url 'https://api.mainnet-beta.solana.com' in get_tokens ?

The 429 issue you are running into is a RPC rate limit. Testnet has the following rate limits at the time of writing:

Maximum number of requests per 10 seconds per IP: 100 Maximum number of requests per 10 seconds per IP for a single RPC: 40 Maximum concurrent connections per IP: 40 Maximum connection rate per 10 seconds per IP: 40 Maximum amount of data per 30 second: 100 MB

gonna try to work around to omit this error