如何高效获取Solana地址的最早交易签名（Python实现）

问题描述 #

需要获取指定Solana地址的最早交易签名，但现有代码从最新签名开始分批遍历，面对数十亿笔交易的地址时请求次数过多，极易触发RPC速率限制，效率极低。

现有代码：

import asyncio
from solana.rpc.async_api import AsyncClient
from typing import Optional
from solders.pubkey import Pubkey

SOLANA_RPC_URL = "https://api.mainnet-beta.solana.com/"  # Example RPC URL, adjust as necessary

async def get_oldest_transaction_signature(address: str) -> Optional[str]:
    async with AsyncClient(SOLANA_RPC_URL) as client:
        # Start with None to get the most recent signatures first
        before = None
        oldest_signature = None

        address = Pubkey.from_string(address)

        while True:
            # Fetch a batch of signatures
            signatures_response = await client.get_signatures_for_address(
                address, before=before, limit=1000
            )
            signatures = signatures_response.value
            if not signatures:
                break  # No more signatures found
            # Update the oldest signature
            oldest_signature = signatures[-1].signature
            # Set the 'before' parameter to the oldest signature in the current batch
            before = oldest_signature
    if oldest_signature:
        return str(oldest_signature)
    else:
        return None

# Usage example
async def main():
    address = "BXbmrFeRNUn13V5YYHnXe2Ws55H48hJuB7LyKqEyXLTh"  # Replace with the actual address
    oldest_signature = await get_oldest_transaction_signature(address)
    print(f"Oldest Transaction Signature: {oldest_signature}")

if __name__ == "__main__":
    asyncio.run(main())

高效解决方案 #

1. 使用升序排序直接获取最早签名 #

Solana RPC的get_signatures_for_address方法支持sort_order参数，设置为SortOrder.ASC（升序，从旧到新排列），同时将limit设为1，即可一次性获取该地址的最早交易签名，无需遍历所有交易。

修改后的代码：

import asyncio
from solana.rpc.async_api import AsyncClient
from solana.rpc.types import SortOrder
from typing import Optional
from solders.pubkey import Pubkey

SOLANA_RPC_URL = "https://api.mainnet-beta.solana.com/"  # 可替换为付费RPC提高稳定性

async def get_oldest_transaction_signature(address: str) -> Optional[str]:
    async with AsyncClient(SOLANA_RPC_URL) as client:
        address = Pubkey.from_string(address)
        
        # 按升序（旧到新）查询，仅取1条结果
        signatures_response = await client.get_signatures_for_address(
            address, sort_order=SortOrder.ASC, limit=1
        )
        signatures = signatures_response.value
        
        if signatures:
            return str(signatures[0].signature)
        else:
            return None

# Usage example
async def main():
    address = "BXbmrFeRNUn13V5YYHnXe2Ws55H48hJuB7LyKqEyXLTh"  # Replace with the actual address
    oldest_signature = await get_oldest_transaction_signature(address)
    print(f"Oldest Transaction Signature: {oldest_signature}")

if __name__ == "__main__":
    asyncio.run(main())

2. 优化建议 #

• 使用付费RPC服务：公共RPC（如主网默认API）速率限制严格，对于高频或大规模查询，建议使用QuickNode、Alchemy等付费RPC，它们支持更高的请求配额和更稳定的服务。
• 添加重试机制：如果遇到速率限制，可实现带指数退避的重试逻辑，避免频繁请求被封禁。
• 确认RPC版本支持：确保使用的RPC节点支持sort_order参数，部分旧版本节点可能不兼容该功能。

3. 备选方案：二分查找区块高度 #

如果sort_order参数不可用，可通过二分查找地址首次出现交易的区块高度来减少查询次数：

1. 获取当前最新区块高度，设置初始查找范围（从0到最新高度）。
2. 每次取中间高度，查询该高度附近的交易签名，判断地址是否有交易。
3. 逐步缩小范围，直到找到最早的区块，再从该区块中提取最早的交易签名。

这种方法的请求次数远少于全量遍历，但实现复杂度高于直接使用升序排序。

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