如何用 Python 批量获取 Solana 地址余额及交易时间？

2024-07-12 06:54:38

如何用 Python 脚本获取文本文件中的 SOL 地址余额及最后交易时间

在 Solana 生态系统中，批量获取地址余额和交易信息是开发者经常遇到的需求。手动查询效率低下且容易出错，而利用 Solana API 和 Python 脚本可以轻松实现自动化查询。本文将探讨如何使用 Python 脚本从文本文件中读取 SOL 地址，并获取每个地址的 SOL 余额和最后一次 SPL 转账时间。

从需求出发

假设我们有一个文本文件，其中包含数百个 SOL 地址。我们需要快速获取每个地址的 SOL 余额以及最后一次进行 SPL 转账的时间。

为了解决这个问题，我们需要一个可以自动执行以下任务的程序：

读取文本文件并提取所有 SOL 地址。
利用 Solana API 查询每个地址的 SOL 余额。
同样利用 Solana API 查询每个地址的交易历史记录，并找到最后一次 SPL 转账交易。
将结果以清晰易懂的格式输出。

Python 脚本实现

以下 Python 脚本可以帮助我们实现上述目标：

import aiohttp
import asyncio
import time
import re
from datetime import datetime

async def fetch_sol_balance(session, address):
    solana_api_url = "https://api.mainnet-beta.solana.com"
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getAccountInfo",
        "params": [address, {"encoding": "jsonParsed"}]
    }
    async with session.post(solana_api_url, json=payload) as response:
        if response.status == 200:
            data = await response.json()
            sol_balance = data.get('result', {}).get('value', {}).get('lamports', 0) / 10**9
            return sol_balance
        return None

async def fetch_last_spl_transfer(session, address):
    await asyncio.sleep(0.25)  # 避免 API 限流
    solana_api_url = "https://api.mainnet-beta.solana.com"
    payload_signatures = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getSignaturesForAddress",
        "params": [address, {"limit": 100}] 
    }
    async with session.post(solana_api_url, json=payload_signatures) as response:
        if response.status == 200:
            data = await response.json()
            signatures = data.get('result', [])
            for signature_data in signatures:
                signature = signature_data['signature']
                payload_transaction = {
                    "jsonrpc": "2.0",
                    "id": 1,
                    "method": "getTransaction",
                    "params": [signature, "jsonParsed"]
                }
                await asyncio.sleep(0.25) # 避免 API 限流
                async with session.post(solana_api_url, json=payload_transaction) as trans_response:
                    if trans_response.status == 200:
                        trans_data = await trans_response.json()
                        transaction = trans_data.get('result', {})
                        if transaction:
                            if transaction.get('meta', {}).get('logMessages'):
                              for log in transaction['meta']['logMessages']:
                                  if "spl-token" in log:
                                      block_time = transaction.get('blockTime')
                                      return block_time
    return None

def convert_timestamp_to_utc(timestamp):
    return datetime.utcfromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S UTC') if timestamp else "无最近 SPL 转账记录"

async def main():
    file_path = "sol_addresses.txt"  # 修改为你的文件名
    with open(file_path, 'r', encoding='utf-8') as file:
        html_content = file.read()

    addresses = re.findall(r'https://solscan.io/account/(\w+)', html_content)
    target_balance = float(input("请输入要显示的最低 SOL 余额："))

    start_time = time.time()
    async with aiohttp.ClientSession() as session:
        tasks = []
        for address in addresses:
            tasks.append(asyncio.create_task(fetch_sol_balance(session, address)))
            tasks.append(asyncio.create_task(fetch_last_spl_transfer(session, address)))
        results = await asyncio.gather(*tasks)

    address_data = {}
    for i in range(0, len(results), 2):
        address = addresses[i // 2]
        balance = results[i]
        last_transfer = results[i + 1]
        if balance is not None and balance >= target_balance:
            address_data[address] = {
                'balance': balance,
                'last_transfer': convert_timestamp_to_utc(last_transfer)
            }

    if address_data:
        for address, data in address_data.items():
            print(f"地址: {address}, SOL 余额: {data['balance']}, 最后一次 SPL 转账时间: {data['last_transfer']}")
    else:
        print("未找到 SOL 余额达到或超过指定金额的钱包。")

    end_time = time.time()
    print("\n总耗时：", end_time - start_time, "秒")

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

代码解读

导入必要库 : 代码首先导入了所需的库，包括 aiohttp 用于异步 HTTP 请求，asyncio 用于异步编程，time 用于计算时间，re 用于正则表达式匹配，以及 datetime 用于处理时间戳。
fetch_sol_balance 函数 : 该函数接收一个 session 对象和一个 SOL 地址作为参数，异步查询并返回该地址的 SOL 余额。
fetch_last_spl_transfer 函数 : 该函数接收一个 session 对象和一个 SOL 地址作为参数，异步查询该地址的最新 100 条交易记录，并查找包含 "spl-token" 的交易，返回该交易的时间戳。函数中添加了 await asyncio.sleep(0.25) 语句，用于在每次请求之间暂停 0.25 秒，避免因请求频率过高导致 API 限流。
convert_timestamp_to_utc 函数 : 该函数将时间戳转换为 UTC 时间字符串，如果传入的时间戳为空则返回 "无最近 SPL 转账记录"。
main 函数 :
- 从名为 "sol_addresses.txt" 的文本文件中读取 SOL 地址列表，你需要将文件名修改为你实际使用的文件名。
- 提示用户输入最低 SOL 余额。
- 创建一个异步 HTTP 会话。
- 使用 asyncio.gather 并发执行所有查询任务，提高效率。
- 将查询结果存储在字典中，方便后续处理。
- 遍历结果字典，打印满足条件的地址信息。
程序入口 : if __name__ == "__main__": 语句确保代码只有在作为主程序运行时才会执行。

常见问题解答

问：如何修改脚本以读取不同格式的地址列表？

答：可以修改 main 函数中读取地址列表部分的代码。例如，如果地址列表存储在 CSV 文件中，可以使用 csv 模块读取文件内容。
问：如何更改 Solana API 节点？

答：可以修改 fetch_sol_balance 和 fetch_last_spl_transfer 函数中的 solana_api_url 变量，将其设置为你想使用的 Solana API 节点地址。
问：如何处理脚本运行过程中出现的错误？

答：可以使用 try...except 语句捕获异常，并进行相应的处理。例如，可以打印错误信息，或者将错误记录到日志文件中。
问：如何提高脚本的运行效率？

答：可以使用异步编程技术，例如 asyncio 模块，并发执行查询任务，从而提高脚本的运行效率。
问：如何将查询结果保存到文件中？

答：可以使用文件操作函数，例如 open 和 write，将查询结果写入到文件中。例如，可以将结果保存为 CSV 文件，方便后续分析和处理。