Solana链上地址交易记录跟踪指南

使用Python全面追踪并解析交易记录中的token信息

核心亮点

使用Solana API: 通过官方RPC接口获取交易签名和详情，从而追踪地址上的所有交易。
解析交易信息: 从交易详情中提取token地址、token名称、交易方向（入账或出账）以及交易时间。
完善代码示例: 提供详尽代码及思路，包括错误处理、数据结构优化及展示方法。

前言

随着越来越多项目和用户开始关注Solana区块链，通过Python实时跟踪链上交易记录已成为数据分析和监控的重要工具。本文将详细讲解如何通过Python构建一个脚本，用以跟踪Solana链上某个地址的交易记录，并将包含的token地址、名称、交易方向和时间等信息进行展示。我们将介绍所需库的安装、如何连接Solana网络、获取交易签名、拉取交易详情、解析交易数据以及展示解析结果。

所需依赖及安装配置

安装必备库

为了与Solana区块链交互，我们需要使用官方提供的Python客户端库，并结合数据解析与展示工具。主要依赖如下：

solana：用于与Solana JSON RPC接口交互
requests：发送API请求
pandas：数据整理与展示

可通过pip命令进行安装：


# 安装solana和requests库
pip install solana requests pandas

分步实现方案

1. 连接Solana网络

利用Solana RPC客户端，可以连接到Solana主网或测试网。这里我们以主网为例：


# 导入客户端模块
from solana.rpc.api import Client

# 设定主网RPC网址
solana_url = "https://api.mainnet-beta.solana.com"
client = Client(solana_url)

2. 获取交易签名

通过指定地址，调用客户端方法获取最近若干笔交易的签名列表。每个签名可以为后续获取交易详情做准备。


from solders.pubkey import Pubkey

# 指定要查询的Solana钱包地址（替换为实际地址）
address = "你的Solana钱包地址"
pubkey = Pubkey.from_string(address)

# 获得最新10笔交易签名
response = client.get_signatures_for_address(pubkey, limit=10)
signatures = response.get('result', [])

3. 拉取交易详情

使用获取到的交易签名，通过调用交易详情API，从每笔交易中获取全面数据。详细数据会包含区块时间、交易指令以及余额变化等信息。


from datetime import datetime

def fetch_transaction_details(signature):
    # 可选项增加确认参数，确保返回已确认交易详情
    response = client.get_transaction(signature, opts={'commitment': 'confirmed'})
    return response.get('result', {})

# 示例：获取第一笔交易的详情
if signatures:
    tx_detail = fetch_transaction_details(signatures[0]['signature'])
    print(tx_detail)

4. 解析交易信息

一笔交易可能包含多个指令，解析时需关注具体指令中的token信息。通过解析交易中的指令及后续token余额变化，可以获得下列信息：

Token地址：标识token的独特地址
Token名称：通常需要额外查询映射表或API获取
交易方向：根据发出地址与接收地址判断，例如，如果目标地址为发送方则视为“出账”，若为接收方则视为“入账”。
时间信息：从区块时间转换为人类可阅读的格式

下文代码段展示了如何解析和提取这些信息：


def parse_transaction(tx_detail, target_address):
    parsed_data = []
    if not tx_detail:
        return parsed_data

    # 获取区块时间
    block_time = tx_detail.get("blockTime")
    if block_time:
        transaction_time = datetime.fromtimestamp(block_time).strftime('%Y-%m-%d %H:%M:%S')
    else:
        transaction_time = "未知"
        
    # 提取所有指令
    instructions = tx_detail.get("transaction", {}).get("message", {}).get("instructions", [])
    for ins in instructions:
        # 此处以假设交易指令含有‘parsed’字段，部分交易可能需手动解析
        parsed = ins.get("parsed", {})
        if parsed.get("type") == "transfer":
            info = parsed.get("info", {})
            token_address = info.get("token", "未知")
            # 根据source和destination区分交易方向
            # 注意：具体字段名称及判断方法可能依赖于使用的API的返回格式
            source = info.get("source", "")
            direction = "出账" if source.lower() == target_address.lower() else "入账"
            
            # 通过额外函数获取token名称（需要自行实现）
            token_name = get_token_name(token_address)
            
            parsed_data.append({
                "交易时间": transaction_time,
                "Token地址": token_address,
                "Token名称": token_name,
                "交易方向": direction
            })
    return parsed_data

def get_token_name(token_address):
    # 示例函数：通常需要查询对应平台或具有映射的API返回名称
    # 此处直接返回token_address作为占位符，用户可根据实际需求实现
    return token_address

# 示例：解析第一笔交易
if signatures:
    tx_detail = fetch_transaction_details(signatures[0]['signature'])
    parsed_info = parse_transaction(tx_detail, address)
    for info in parsed_info:
        print(info)

5. 整合数据与展示

将所有解析后的交易数据进行整合，可以组织为列表或DataFrame以便展示。使用Pandas库可以将数据以表格形式直观展示。


import pandas as pd

def display_transactions(parsed_transactions):
    if not parsed_transactions:
        print("没有找到交易记录")
        return
    df = pd.DataFrame(parsed_transactions)
    print(df)

# 获取多笔交易并显示
all_transactions = []
for sig in signatures:
    detail = fetch_transaction_details(sig['signature'])
    data = parse_transaction(detail, address)
    all_transactions.extend(data)

display_transactions(all_transactions)

代码实现概览

下面是一个整合上述各步骤的完整示例脚本，该脚本可用于跟踪指定Solana地址的交易记录，并从中提取出token地址、token名称、交易方向和交易时间等信息。


# -*- coding: utf-8 -*-
# 导入必要库
from solana.rpc.api import Client
from solders.pubkey import Pubkey
from datetime import datetime
import pandas as pd

# 连接Solana主网
solana_url = "https://api.mainnet-beta.solana.com"
client = Client(solana_url)

def get_token_name(token_address):
    # 实现获取token名称的具体方法，例如查询API或维护映射字典
    # 此处仅作为示例返回token_address
    return token_address

def fetch_transaction_details(signature):
    response = client.get_transaction(signature, opts={'commitment': 'confirmed'})
    return response.get('result', {})

def parse_transaction(tx_detail, target_address):
    parsed_data = []
    if not tx_detail:
        return parsed_data
    # 获取交易时间并转换为可读格式
    block_time = tx_detail.get("blockTime")
    if block_time:
        transaction_time = datetime.fromtimestamp(block_time).strftime('%Y-%m-%d %H:%M:%S')
    else:
        transaction_time = "未知"
    instructions = tx_detail.get("transaction", {}).get("message", {}).get("instructions", [])
    for ins in instructions:
        parsed = ins.get("parsed", {})
        # 判断是否为转账交易
        if parsed.get("type") == "transfer":
            info = parsed.get("info", {})
            token_address = info.get("token", "未知")
            source = info.get("source", "")
            direction = "出账" if source.lower() == target_address.lower() else "入账"
            token_name = get_token_name(token_address)
            parsed_data.append({
                "交易时间": transaction_time,
                "Token地址": token_address,
                "Token名称": token_name,
                "交易方向": direction
            })
    return parsed_data

def get_transactions(address, limit=10):
    pubkey = Pubkey.from_string(address)
    response = client.get_signatures_for_address(pubkey, limit=limit)
    return response.get('result', [])

def display_transactions(parsed_transactions):
    if not parsed_transactions:
        print("没有找到交易记录")
        return
    df = pd.DataFrame(parsed_transactions)
    # 将DataFrame打印为表格
    print(df)
    # 使用HTML表格展示（可选：保存为HTML文件）
    html_table = df.to_html(index=False)
    with open("transactions.html", "w", encoding="utf-8") as f:
        f.write(html_table)
    print("交易记录保存为 transactions.html")

def main():
    # 替换为目标Solana钱包地址
    target_address = "你的Solana钱包地址"
    transactions = get_transactions(target_address, limit=10)
    all_parsed = []
    for sig in transactions:
        detail = fetch_transaction_details(sig['signature'])
        tx_data = parse_transaction(detail, target_address)
        all_parsed.extend(tx_data)
    display_transactions(all_parsed)

if __name__ == "__main__":
    main()

如何理解与扩展代码

上述脚本涵盖了从获取签名到解析及展示数据的全流程。若需要进一步扩展，你可以：

在函数get_token_name中增加对Token名称的查询，将token地址映射到对应的友好名称，可能通过调用第三方API（如Solscan或CoinGecko API）来实现。
对于交易方向判定，可根据更多交易字段进行精确判断，并区分内部转账、外部收发等情况。
增加日志与异常捕捉机制，对网络延迟、无效数据以及API调用失败等情况进行处理。
采用WebSocket订阅接口实现实时监控，减少轮询查询的网络资源浪费。

此外，完善的数据存储与可视化设计能提升整体应用的用户体验。你可以将数据写入数据库，结合前端框架呈现实时监控仪表板，构建一个完整的交易追踪平台。

展示交易记录数据的HTML表格

为方便在网页中直观展现追踪数据，可以将交易记录保存为HTML表格格式。以下为生成HTML表格的示例格式：

交易时间	Token地址	Token名称	交易方向
2025-03-01 14:30:55	Token_A_Address	Token_A	入账
2025-03-02 09:15:20	Token_B_Address	Token_B	出账