TON区块链Swap交易识别：Python实现技术问询与问题排查

一、Swap识别：核心标识与模式 #

• DEX合约地址匹配：优先匹配交易的收发地址是否属于已知主流DEX（如Ston.fi、DeDust）的合约地址库，这是最基础的筛选条件
• 消息数据的函数标识：Swap交易的消息body中会包含特定函数编码，比如Ston.fi的swap函数对应ASCII编码0x73776170（即"swap"），解析消息body提取这类标识可精准识别Swap操作
• 多消息交易结构：Swap通常包含用户发起的转账/授权消息、DEX内部执行消息、代币返回消息，这类交易的out_msgs数量一般≥2，且涉及代币合约地址交互
• 跨资产流动特征：Swap必然伴随至少两种资产的双向转移（比如TON+FT代币，或两种FT代币），单一TON转账可直接排除

二、代币关联：精准提取与区分 #

• 解析代币转账消息：代币转账的消息body中会包含代币合约地址和转账数量，需遵循TON的FT标准（如JettonWallet合约接口）解析，可调用get_symbol接口验证代币属性
• 明确资产流向：
  • 用户钱包→DEX合约：属于转出资产，若为TON则取交易value字段，若为FT代币则从消息body提取数量和合约地址
  • DEX合约→用户钱包：属于转入资产，同样从消息中提取对应信息
• 匹配交易对：同一Swap组内，转出资产集合与转入资产集合的数量应为1:1（单对Swap）或多对1（多币换一币），通过DEX合约和时间范围关联匹配

三、多交易归组：可靠关联策略 #

• 用query_id作为唯一标识：用户发起Swap时，钱包会生成唯一的query_id并嵌入消息body，同一Swap的所有关联交易都会携带该ID，这是最精准的关联方式
• 缩小时间窗口+DEX枢纽关联：放弃5分钟宽窗口，改用30秒内同一用户与同一DEX合约的双向交易（转出+转入）归组，配合query_id验证
• 交易链路追踪：通过交易的prev_transaction_hash或消息的source_transaction字段，追踪同一Swap的交易链路，确保所有相关交易被纳入同一组

优化后的Python代码示例 #

import asyncio
from TonTools import TonCenterClient, Wallet
from datetime import datetime, timedelta

# 初始化TON中心客户端
client = TonCenterClient(base_url='https://toncenter.com/api/v2/')

# 主流DEX合约地址（请替换为最新有效地址）
KNOWN_DEX_CONTRACTS = {
    'StonFi': 'EQBynBO23ywHy_CgarY9NK9FTz0yDsG82PtcbSTQgGoXwiuA',
    'DeDust': 'EQCD39VS5jcptHL8vMjEXrzGaRcCVYto7HUn4bpAOg8xqB2N'
}

async def parse_jetton_transfer(msg):
    """解析Jetton代币转账消息，返回(代币合约地址, 转账数量)"""
    if not msg.msg_data or 'body' not in msg.msg_data:
        return None
    body = msg.msg_data['body']
    try:
        # 按照Jetton转账消息格式解析：前32字节是op码，接下来32字节是query_id，再48字节是代币合约地址，最后是数量
        op_code = body[:32]
        if op_code != '0xf8a7ea5':  # Jetton转账的标准op码
            return None
        jetton_contract = body[64:112]  # 提取代币合约地址
        amount = int(body[112:144], 16)  # 提取转账数量
        return (jetton_contract, amount)
    except Exception as e:
        return None

async def check_for_swap(wallet_address):
    wallet = Wallet(address=wallet_address, provider=client)
    # 扩大交易获取范围，避免遗漏关联交易
    transactions = await wallet.get_transactions(limit=50)
    # 用query_id作为Swap组的唯一键
    swap_groups = {}

    for tx in transactions:
        tx_time = datetime.utcfromtimestamp(tx.utime)
        query_id = None
        dex_contract = None
        is_dex_tx = False

        # 提取消息中的query_id（用于关联同一Swap）
        if tx.in_msg and 'body' in tx.in_msg.msg_data:
            body = tx.in_msg.msg_data['body']
            # 通常query_id在消息body的第32-64字节（op码之后）
            if len(body) >= 64:
                query_id = body[32:64]

        # 判断是否为DEX相关交易
        if tx.in_msg:
            if tx.in_msg.source in KNOWN_DEX_CONTRACTS.values():
                is_dex_tx = True
                dex_contract = tx.in_msg.source
            if tx.in_msg.destination in KNOWN_DEX_CONTRACTS.values():
                is_dex_tx = True
                dex_contract = tx.in_msg.destination
        # 检查out_msgs是否涉及DEX
        for out_msg in tx.out_msgs:
            if out_msg.destination in KNOWN_DEX_CONTRACTS.values():
                is_dex_tx = True
                dex_contract = out_msg.destination
                break

        if not is_dex_tx or not query_id:
            continue

        # 初始化Swap组
        if query_id not in swap_groups:
            swap_groups[query_id] = {
                'timestamp': tx_time,
                'dex_name': next(k for k, v in KNOWN_DEX_CONTRACTS.items() if v == dex_contract),
                'tokens_sent': [],
                'tokens_received': []
            }

        # 处理用户向DEX转出的资产
        if tx.in_msg.source == wallet_address and tx.in_msg.destination == dex_contract:
            # 检查是否为Jetton转账
            jetton_transfer = parse_jetton_transfer(tx.in_msg)
            if jetton_transfer:
                swap_groups[query_id]['tokens_sent'].append(jetton_transfer)
            else:
                # 否则是TON转账
                swap_groups[query_id]['tokens_sent'].append(('TON', tx.in_msg.value))
        
        # 处理DEX向用户转入的资产
        if tx.in_msg.destination == wallet_address and tx.in_msg.source == dex_contract:
            jetton_transfer = parse_jetton_transfer(tx.in_msg)
            if jetton_transfer:
                swap_groups[query_id]['tokens_received'].append(jetton_transfer)
            else:
                swap_groups[query_id]['tokens_received'].append(('TON', tx.in_msg.value))
        
        # 处理out_msgs中的资产转移
        for out_msg in tx.out_msgs:
            if out_msg.destination == dex_contract:
                jetton_transfer = parse_jetton_transfer(out_msg)
                if jetton_transfer:
                    swap_groups[query_id]['tokens_sent'].append(jetton_transfer)
                else:
                    swap_groups[query_id]['tokens_sent'].append(('TON', out_msg.value))
            elif out_msg.destination == wallet_address:
                jetton_transfer = parse_jetton_transfer(out_msg)
                if jetton_transfer:
                    swap_groups[query_id]['tokens_received'].append(jetton_transfer)
                else:
                    swap_groups[query_id]['tokens_received'].append(('TON', out_msg.value))

    # 输出整理后的Swap信息
    for qid, swap_info in swap_groups.items():
        print(f"\nSwap ID: {qid.hex()}")
        print(f"DEX平台: {swap_info['dex_name']} | 时间: {swap_info['timestamp'].strftime('%Y-%m-%d %H:%M:%S UTC')}")
        print("转出资产:")
        for token, amount in swap_info['tokens_sent']:
            print(f"  - {token}: {amount}")
        print("转入资产:")
        for token, amount in swap_info['tokens_received']:
            print(f"  - {token}: {amount}")

async def main():
    # 替换为目标钱包地址
    target_wallet = '替换为你的钱包地址'
    await check_for_swap(target_wallet)

asyncio.run(main())

内容的提问来源于stack exchange，提问作者erkan ışık bacak