如何用Python高效获取单对单原生SOL转账并避免滞后？

1. 切换到批量交易查询+异步处理 #

get_transaction单条请求慢的核心原因是单请求单处理，换成批量接口get_multiple_transactions可大幅减少RPC请求次数，结合异步IO进一步压缩处理时间。

• 操作步骤：
  • 用get_signatures_for_address批量拉取目标发送（或接收）钱包的交易签名，每次拉取50-100条（根据RPC限额调整），通过before参数实现增量查询。
  • 将签名列表传入get_multiple_transactions，一次性获取多条交易数据。
  • 用异步方式执行上述流程，避免同步等待的时间浪费。

示例代码（异步版本）：

import asyncio
from solana.rpc.async_api import AsyncClient

async def fetch_batch_transactions(rpc_url, target_from, target_to, last_sig=None):
    client = AsyncClient(rpc_url)
    # 批量获取发送钱包的交易签名
    sigs_resp = await client.get_signatures_for_address(
        target_from,
        limit=100,
        before=last_sig
    )
    sig_list = [sig["signature"] for sig in sigs_resp.value]
    if not sig_list:
        await client.close()
        return [], None
    
    # 批量获取交易详情
    txs_resp = await client.get_multiple_transactions(sig_list)
    filtered_txs = []
    
    for tx_item in txs_resp.value:
        tx = tx_item.transaction
        # 快速过滤：交易有错误直接跳过
        if tx_item.meta.err is not None:
            continue
        
        # 检查涉及的钱包，超过两个直接跳过
        account_keys = [key.pubkey for key in tx.message.account_keys]
        involved_wallets = set(account_keys)
        if len(involved_wallets) != 2 or target_from not in involved_wallets or target_to not in involved_wallets:
            continue
        
        # 验证是否为原生SOL转账（SystemProgram的Transfer指令）
        is_valid_transfer = False
        for instr in tx.instructions:
            # SystemProgram固定地址：11111111111111111111111111111111
            if instr.program_id != "11111111111111111111111111111111":
                break
            # Transfer指令数据长度固定为68字节（1字节指令码+32字节from+32字节to+8字节金额）
            if len(instr.data) != 68:
                break
            # 解析from和to地址
            from_addr = instr.data[1:33].hex()
            to_addr = instr.data[33:65].hex()
            if (from_addr == target_from and to_addr == target_to) or (from_addr == target_to and to_addr == target_from):
                is_valid_transfer = True
                break
        
        if is_valid_transfer:
            filtered_txs.append(tx_item)
    
    await client.close()
    return filtered_txs, sig_list[-1] if sig_list else None

# 调用示例
async def main():
    RPC_URL = "https://api.mainnet-beta.solana.com"
    FROM_WALLET = "你的发送钱包地址"
    TO_WALLET = "你的接收钱包地址"
    last_processed_sig = None
    while True:
        valid_txs, last_processed_sig = await fetch_batch_transactions(RPC_URL, FROM_WALLET, TO_WALLET, last_processed_sig)
        for tx in valid_txs:
            print(f"找到有效转账: {tx.transaction.signatures[0]}")
        await asyncio.sleep(0.5)  # 短周期轮询，可根据需求调整

asyncio.run(main())

2. 使用WebSocket实时订阅交易签名 #

Solana RPC支持WebSocket订阅，能实时推送目标地址的新交易签名，无需主动轮询，彻底解决滞后问题。

• 操作步骤：
  • 建立WebSocket连接，订阅目标发送/接收钱包的交易签名事件。
  • 收到新签名后，异步获取并处理交易数据，过滤符合条件的转账。

示例代码：

import asyncio
from solana.rpc.websocket_api import connect
from solana.rpc.async_api import AsyncClient

async def subscribe_real_time_transactions(rpc_ws_url, target_from, target_to):
    async with connect(rpc_ws_url) as websocket:
        # 订阅发送钱包的已确认交易签名
        await websocket.logs_subscribe(target_from, commitment="confirmed")
        # 接收订阅确认消息
        await websocket.recv()
        print("订阅成功，等待新交易...")
        
        while True:
            msg = await websocket.recv()
            if "result" not in msg or "value" not in msg["result"]:
                continue
            
            sig = msg["result"]["value"]["signature"]
            # 异步获取单条交易（实时单条请求开销可控）
            async with AsyncClient(rpc_ws_url.replace("wss://", "https://")) as client:
                tx_resp = await client.get_transaction(sig)
                tx_item = tx_resp.value
                tx = tx_item.transaction
                
                # 复用之前的过滤逻辑
                if tx_item.meta.err is not None:
                    continue
                
                account_keys = [key.pubkey for key in tx.message.account_keys]
                involved_wallets = set(account_keys)
                if len(involved_wallets) != 2 or target_from not in involved_wallets or target_to not in involved_wallets:
                    continue
                
                is_valid_transfer = False
                for instr in tx.instructions:
                    if instr.program_id != "11111111111111111111111111111111":
                        break
                    if len(instr.data) != 68:
                        break
                    from_addr = instr.data[1:33].hex()
                    to_addr = instr.data[33:65].hex()
                    if (from_addr == target_from and to_addr == target_to) or (from_addr == target_to and to_addr == target_from):
                        is_valid_transfer = True
                        break
                
                if is_valid_transfer:
                    print(f"实时捕获有效转账: {sig}")

# 调用示例
asyncio.run(subscribe_real_time_transactions("wss://api.mainnet-beta.solana.com", "你的发送钱包地址", "你的接收钱包地址"))

3. 过滤逻辑优化技巧 #

• 提前通过交易的meta.err字段过滤失败交易，直接跳过无需解析。
• 先检查交易的message.account_keys，如果包含两个目标钱包之外的地址，直接跳过指令解析。
• 只关注confirmed或finalized状态的交易，避免处理未确认的无效交易。

额外建议 #

• 使用付费RPC节点（如QuickNode、Alchemy），相比公共节点，付费节点请求限额更高、延迟更低，能提升批量查询和WebSocket订阅的稳定性。
• 缓存已处理的交易签名，避免重复处理相同交易。

内容的提问来源于stack exchange，提问作者FairyLiquid