Solana链指定地址最早5笔购买交易查询脚本优化求助

核心问题解决思路 #

1. 按从旧到新顺序处理交易 #

getSignaturesForAddress 默认返回从新到旧的签名列表，要优先处理最早的交易，只需将获取到的签名列表反转，即可得到从旧到新的顺序，再逐个检查有效性。

2. 简化购买交易过滤逻辑 #

结合以下关键特征快速判定有效购买交易：

• 交易执行无错误（meta.err 为 None）
• 日志中包含 Program log: Instruction: Buy 指令（确认是购买操作）
• 存在 SPL Token 转账指令（代币从曲线合约到用户）
• 存在 SOL 转账指令（用户向合约支付SOL）

同时直接提取所需的所有字段，无需多层嵌套遍历。

优化后的完整代码 #

import asyncio
import aiohttp
import json

# Ankr RPC端点配置
RPC_URL = "https://rpc.ankr.com/solana/xxxxxxx"
BONDING_CURVE_ADDRESS = "BWh5dm8UgSM6vs8KWg1Tvfgj7MhRmW3kZFGF3sfrwYyA"

MAX_TRANSACTIONS = 15  # 最多查询的交易总数
VALID_TRANSACTION_COUNT = 5  # 需要找到的有效交易数量

async def fetch_signatures(session, address, limit):
    """查询指定地址的交易签名，返回从新到旧的列表"""
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getSignaturesForAddress",
        "params": [address, {"limit": limit}]
    }
    async with session.post(RPC_URL, json=payload) as response:
        if response.status == 200:
            data = await response.json()
            return data.get("result", [])
        print(f"获取签名失败: {response.status}")
        return []

async def fetch_transaction_details(session, signature):
    """获取交易详情，包含解析后的指令"""
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "getTransaction",
        "params": [signature, {"encoding": "jsonParsed", "maxSupportedTransactionVersion": 0}]
    }
    async with session.post(RPC_URL, json=payload) as response:
        if response.status == 200:
            data = await response.json()
            return data.get("result")
        print(f"获取交易详情失败: {response.status}")
        return None

def parse_buy_transaction(transaction):
    """解析并验证购买交易，返回所需字段或None"""
    meta = transaction.get("meta", {})
    
    # 1. 跳过失败交易
    if meta.get("err") is not None:
        return None
    
    # 2. 检查是否为Buy指令
    log_messages = meta.get("logMessages", [])
    if not any("Program log: Instruction: Buy" in log for log in log_messages):
        return None
    
    # 3. 初始化结果字段
    result = {
        "timestamp": transaction.get("blockTime", "N/A"),
        "tx_hash": transaction.get("signature", "N/A"),
        "fee": meta.get("fee", "N/A"),
        "token_amount": 0,
        "sol_spent": 0,
        "sender": "",
        "recipient": ""
    }
    
    # 4. 遍历所有指令，提取转账信息
    for instruction in transaction.get("transaction", {}).get("message", {}).get("instructions", []):
        parsed = instruction.get("parsed", {})
        program = instruction.get("program")
        
        # 提取SOL转账（用户→合约）
        if program == "system" and parsed.get("type") == "transfer":
            info = parsed.get("info", {})
            if info.get("destination") == BONDING_CURVE_ADDRESS:
                result["sender"] = info.get("source")
                result["sol_spent"] = int(info.get("amount", 0)) / 10**9  # 转换为SOL单位
        
        # 提取Token转账（合约→用户）
        elif program == "spl-token" and parsed.get("type") == "transfer":
            info = parsed.get("info", {})
            if info.get("source") == BONDING_CURVE_ADDRESS:
                result["recipient"] = info.get("destination")
                result["token_amount"] = int(info.get("amount", 0)) / 10**6  # 假设是6位小数代币，需根据实际调整
    
    # 验证是否获取到完整信息
    if not result["sender"] or not result["recipient"] or result["sol_spent"] == 0 or result["token_amount"] == 0:
        return None
    
    return result

async def filter_and_log_transactions(address):
    async with aiohttp.ClientSession() as session:
        print(f"正在查询地址的交易: {address}")
        
        # 获取最多15笔交易签名，反转后得到从旧到新的顺序
        signatures = await fetch_signatures(session, address, MAX_TRANSACTIONS)
        if not signatures:
            print("未找到任何交易签名。")
            return
        
        # 反转列表，优先处理最早的交易
        signatures.reverse()
        
        valid_transactions = []
        processed = 0
        
        for sig_info in signatures:
            if len(valid_transactions) >= VALID_TRANSACTION_COUNT or processed >= MAX_TRANSACTIONS:
                break
            
            signature = sig_info["signature"]
            transaction = await fetch_transaction_details(session, signature)
            processed += 1
            
            if not transaction:
                print(f"交易未找到: {signature}")
                continue
            
            parsed_tx = parse_buy_transaction(transaction)
            if parsed_tx:
                valid_transactions.append(parsed_tx)
        
        # 输出结果
        print(f"\n找到 {len(valid_transactions)} 笔有效购买交易（从最早到最新）:")
        for idx, tx in enumerate(valid_transactions, 1):
            print(f"\n第 {idx} 笔交易:")
            print(f"  - 时间戳: {tx['timestamp']}")
            print(f"  - 交易哈希: {tx['tx_hash']}")
            print(f"  - 发送方: {tx['sender']}")
            print(f"  - 接收方: {tx['recipient']}")
            print(f"  - 代币数量: {tx['token_amount']}")
            print(f"  - 花费SOL: {tx['sol_spent']}")
            print(f"  - 交易手续费: {tx['fee'] / 10**9} SOL")

if __name__ == "__main__":
    asyncio.run(filter_and_log_transactions(BONDING_CURVE_ADDRESS))

关键优化点说明 #

1. 交易顺序调整：通过反转签名列表，直接实现从旧到新的处理顺序，无需复杂的分页逻辑。
2. 过滤逻辑简化：将验证和字段提取合并到一个函数中，通过遍历顶层指令直接获取SOL和Token转账信息，避免多层嵌套遍历内部指令。
3. 字段完整性验证：确保提取到所有必填字段后才判定为有效交易，避免无效数据。
4. 单位转换：将原始的lamport单位转换为可读的SOL/代币单位，提升结果可读性。

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