基于Raydium的Python跨钱包交易PnL分析程序问题排查

一、Solscan API代码返回0结果的修复 #

问题根源 #

代码存在以下导致无结果的问题：

• Solscan Pro API返回的交易数据嵌套在data字段中，原代码直接遍历返回值导致空结果
• Raydium程序ID不全，缺少CLMM等新版本AMM地址
• 交易匹配仅依赖tokenAccount，易因同token不同账户导致匹配失败
• 买入卖出判断逻辑不准确，未结合稳定币/SOL的交易方向

修改后的完整代码 #

import requests
from datetime import datetime

# 更新Raydium AMM程序ID（覆盖经典、V3、CLMM版本）
RAYDIUM_AMM_PROGRAM_ID = [
    "CPMMoo8L3F4NbTegBCKVNunggL7H1ZpdTHKxQB5qKP1C",
    "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8",
    "5quBtoiQqxF9Jv6KYKctB59NT3gtJD2Y65kdnB1Uev3h",
    "CAMMCzo5YL8w4VFF8KVHrK22GGUsp5VTaW7grrKgrWqK",
    "routeUGWgWzqBWFcrCfv8tritsqukccJPu3q5GPP3xS",
    "9xQeWvG816bUx9EPjHmaT23yvVM2ZWbrrpZb9PusVFin",
]

def fetch_transaction_history(account, api_key, limit=100, before=None):
    url = f"https://pro-api.solscan.io/v1.0/account/transactions?account={account}&limit={limit}"
    if before:
        url += f"&before={before}"
    headers = {"accept": "application/json", "token": api_key}
    response = requests.get(url, headers=headers)
    if response.status_code == 200:
        return response.json().get('data', [])  # 提取嵌套的data字段
    else:
        print(f"请求失败 {account}: {response.status_code}")
        return []

def extract_token_info(instructions):
    for instruction in instructions:
        if instruction.get('programId') in RAYDIUM_AMM_PROGRAM_ID:
            # 通过指令名称判断是否为swap交易
            if 'swap' in instruction.get('name', '').lower() or 'trade' in instruction.get('name', '').lower():
                params = instruction.get('params', {})
                return {
                    'mintIn': params.get('mintIn'),
                    'mintOut': params.get('mintOut'),
                    'amountIn': float(params.get('amountIn', 0)),
                    'amountOut': float(params.get('amountOut', 0)),
                }
    return None

def filter_and_normalize_transactions(transactions):
    valid_transactions = []
    STABLE_SOL_MINTS = ['So11111111111111111111111111111111111111112', 'EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v']
    
    for tx in transactions:
        if tx.get('status') == 'Success':
            token_info = extract_token_info(tx.get('parsedInstruction', []))
            if token_info:
                # 用SOL/USDC判断买入卖出：换入非稳定币为买入，反之为卖出
                is_buy = token_info['mintIn'] in STABLE_SOL_MINTS
                valid_transactions.append({
                    'blockTime': tx.get('blockTime'),
                    'amountIn': token_info['amountIn'],
                    'amountOut': token_info['amountOut'],
                    'tokenMint': token_info['mintOut'] if is_buy else token_info['mintIn'],
                    'isBuy': is_buy
                })
    return valid_transactions

def match_buy_transactions(your_trades, copied_trades):
    matched_trades = []
    copied_buys = [t for t in copied_trades if t['isBuy']]
    
    for your_buy in your_trades:
        if your_buy['isBuy']:
            # 匹配相同token，且你的买入时间在复制钱包买入后10分钟内（符合复制交易延迟）
            your_time = datetime.fromtimestamp(your_buy['blockTime'])
            for copied_buy in copied_buys:
                copied_time = datetime.fromtimestamp(copied_buy['blockTime'])
                if your_buy['tokenMint'] == copied_buy['tokenMint'] and (your_time - copied_time).total_seconds() <= 600:
                    matched_trades.append(your_buy)
                    break
    return matched_trades

def calculate_pnl(matched_buys, your_sells):
    pnl = 0.0
    buys_by_token = {}
    
    # 按token分组，用先进先出原则匹配买卖
    for buy in matched_buys:
        mint = buy['tokenMint']
        buys_by_token[mint] = buys_by_token.get(mint, []) + [buy]
    
    for sell in your_sells:
        if not sell['isBuy']:
            mint = sell['tokenMint']
            if mint in buys_by_token and buys_by_token[mint]:
                earliest_buy = buys_by_token[mint].pop(0)
                pnl += sell['amountOut'] - earliest_buy['amountIn']
    return round(pnl, 2)

# 主程序
api_key = "你的API密钥"
your_wallet = "你的钱包地址"
copied_wallets = ["复制的钱包地址"]

your_trades_raw = fetch_transaction_history(your_wallet, api_key, limit=200)
your_trades = filter_and_normalize_transactions(your_trades_raw)
your_buys = [t for t in your_trades if t['isBuy']]
your_sells = [t for t in your_trades if not t['isBuy']]

print(f"你的买入交易: {len(your_buys)} 笔")
print(f"你的卖出交易: {len(your_sells)} 笔")

pnl_breakdown = {}
for wallet in copied_wallets:
    copied_trades_raw = fetch_transaction_history(wallet, api_key, limit=200)
    copied_trades = filter_and_normalize_transactions(copied_trades_raw)
    matched_trades = match_buy_transactions(your_buys, copied_trades)
    pnl = calculate_pnl(matched_trades, your_sells)
    pnl_breakdown[wallet] = pnl

for wallet, pnl in pnl_breakdown.items():
    print(f"复制钱包 {wallet} 触发的交易盈亏: {pnl} USD")

overall_pnl = sum(pnl_breakdown.values())
print(f"所有复制交易总盈亏: {overall_pnl} USD")

关键修改点 #

• 修正Solscan API数据提取逻辑，获取嵌套在data字段中的交易列表
• 补充Raydium CLMM程序ID，覆盖更多交易场景
• 改用token mint地址匹配交易，避免账户差异导致的匹配失败
• 加入时间窗口判断（10分钟内），符合复制交易的实际延迟特性
• 基于SOL/USDC的交易方向判断买入卖出，逻辑更准确

二、Solana库导入报错的修复 #

问题根源 #

报错ModuleNotFoundError: No module named 'solana.publickey'的原因：

• solana库版本不兼容，导入路径随版本变更
• 虚拟环境冲突，运行环境未正确安装solana库
• 原代码使用已弃用的get_confirmed_transaction方法

解决方案 #

1. 安装兼容版本的solana库

pip uninstall solana -y
pip install solana==0.30.0

2. 修正导入与交易获取逻辑
   修改后的完整代码：

import json
import time
import threading
import numpy as np
from urllib.request import Request, urlopen
from solana.rpc.api import Client
from solana.publickey import PublicKey

# 常量定义
EXPLORER_URL_TX = "https://explorer.solana.com/tx/"
ADDRESS_RAYDIUM_AMM = "675kPX9MHTjS2zt1qfr1NYHuzeLXfQM9H24wFSUt1Mp8"
ADDRESS_SOLSCAN = "https://api.solscan.io/account?address="

# Solana客户端初始化
SOLANA_MAINNET = "https://api.mainnet-beta.solana.com"
client = Client(SOLANA_MAINNET)

resultArr = []
maxTxCount = 10000
txCount = 0
lastSignature = None
rounds = 0

def get_tx_detail(tx_signature):
    try:
        # 使用get_transaction替代已弃用的get_confirmed_transaction
        tx = client.get_transaction(tx_signature, max_supported_transaction_version=0)
        if tx['result'] is not None:
            post_token_balances = tx["result"]["meta"]["postTokenBalances"]
            pre_token_balances = tx["result"]["meta"]["preTokenBalances"]
            if post_token_balances != pre_token_balances:
                print(EXPLORER_URL_TX + tx_signature)
                resultArr.append(tx)
    except Exception as e:
        print(f"处理交易 {tx_signature} 出错: {str(e)}")

if __name__ == "__main__":
    RaydiumPubKey = PublicKey(ADDRESS_RAYDIUM_AMM)
    while True:
        print(f"轮次 {rounds + 1}")
        print("获取交易签名...")
        txs = client.get_signatures_for_address(RaydiumPubKey, limit=200, before=lastSignature)['result']
        if not txs:
            break  # 无更多交易时退出
        print(f"处理 {len(txs)} 个交易签名")
        signatures = [tx["signature"] for tx in txs]

        threads = []
        for signature in signatures:
            if txCount >= maxTxCount:
                break
            txCount += 1
            thread = threading.Thread(target=get_tx_detail, args=(signature,))
            threads.append(thread)
            thread.start()

        for thread in threads:
            thread.join()

        if txCount >= maxTxCount:
            break
        rounds += 1
        lastSignature = txs[-1]["signature"]
        time.sleep(3)

# 交易处理逻辑
def process_tx(tx):
    tx_signature = tx["result"]["transaction"]["signatures"][0]
    tx_sender = tx["result"]["transaction"]["message"]["accountKeys"][0]
    block_time = tx["result"]["blockTime"]
    slot = tx["result"]["slot"]
    post_token_balances = tx["result"]["meta"]["postTokenBalances"]
    pre_token_balances = tx["result"]["meta"]["preTokenBalances"]
    token_balances = []

    for pre, post in zip(pre_token_balances, post_token_balances):
        change = 0
        if pre["uiTokenAmount"]["uiAmount"] is not None and post["uiTokenAmount"]["uiAmount"] is not None:
            change = post["uiTokenAmount"]["uiAmount"] - pre["uiTokenAmount"]["uiAmount"]
        owner = pre["owner"]
        token = pre["mint"]
        if change != 0:
            token_balances.append({"owner": owner, "token": token, "change": change})

    return {
        "txSignature": tx_signature,
        "sender": tx_sender,
        "blockTime": block_time,
        "slot": slot,
        "tokenBalances": token_balances
    }

processed_tx_arr = [process_tx(tx) for tx in resultArr]

# 映射token mint到名称
mint_dic = {}
for tx in processed_tx_arr:
    for token_balance in tx["tokenBalances"]:
        token = token_balance["token"]
        if token not in mint_dic:
            try:
                req = Request(ADDRESS_SOLSCAN + token, headers={'User-Agent': 'Mozilla/5.0'})
                data = json.loads(urlopen(req).read())
                mint_dic[token] = data["data"]["tokenInfo"]["name"]
                time.sleep(1)
            except Exception as e:
                print(f"获取token信息失败 {token}: {str(e)}")
                mint_dic[token] = token

for tx in processed_tx_arr:
    for token_balance in tx["tokenBalances"]:
        token_balance["token"] = mint_dic.get(token_balance["token"], token_balance["token"])

# 标准化交易
normalized_tx_arr = []
for tx in processed_tx_arr:
    for token_change in tx["tokenBalances"]:
        normalized_tx_arr.append({
            'txSignature': tx["txSignature"],
            'sender': tx["sender"],
            'blockTime': tx["blockTime"],
            'slot': tx["slot"],
            'owner': token_change["owner"],
            'token': token_change["token"],
            'change': token_change["change"]
        })

# 打印结果
for tx in normalized_tx_arr:
    print(tx)

关键修改点 #

• 安装指定版本solana库，确保导入路径兼容
• 替换已弃用的get_confirmed_transaction为get_transaction
• 添加异常捕获，避免单个交易处理失败导致程序崩溃
• 增加无交易时的退出逻辑，避免无限循环

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