基于web3.py在指定DEX上获取代币购买事件的技术问询

需求背景 #

作为重新学习的资深开发者，开发机器人用于在指定DEX上获取并返回特定交易对的代币购买事件，目标是获取并打印以下变量：

• 买家地址（代币接收方）
• 接收代币数量
• 花费金额（通常为WBNB/WETH）
  要求逻辑适配任意EVM兼容链及对应节点，当前难以明确如何获取将WBNB（WETH）转入LP合约的单笔Transfer事件，现有代码会一次性返回多个事件条目，而非单次单个输出。

现有代码 #

#Library Imports
from web3 import Web3
import web3
import json

#.env
node_url_eth = '#myNode'
web3 = Web3(Web3.HTTPProvider(node_url_eth))

#TokenToFetchPair
token_a = '0x0e09fabb73bd3ade0a17ecc321fd13a19e81ce82' #token address
token_b = '0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c' #WBNB address
#toChecksum
checksum_token_a = Web3.toChecksumAddress(token_a)
checksum_token_b = Web3.toChecksumAddress(token_b)

#PCS/Token Variables
PCS_FACTORY_ABI = '#very long abi'
PCS_FACTORY_ADDRESS="0xca143ce32fe78f1f7019d7d551a6402fc5350c73"
PCS_FACTORY_ADDRESS = Web3.toChecksumAddress(PCS_FACTORY_ADDRESS)
standard_abi = '#commonAbiForTokens'
cakePair_abi = '#AbiCakePair'

#Fetching
TokenContract = web3.eth.contract(address=checksum_token_a, abi=standard_abi)
PCS_FACTORY_CONTRACT = web3.eth.contract(address=PCS_FACTORY_ADDRESS, abi=PCS_FACTORY_ABI)

getTokenName = TokenContract.functions.name().call()
getTokenSymbol = TokenContract.functions.symbol().call()
fetchedPair = PCS_FACTORY_CONTRACT.functions.getPair(checksum_token_a, checksum_token_b).call()

print(getTokenName, getTokenSymbol, fetchedPair)


#FilterEvents(Code I found)
filter = web3.eth.filter({
        "topic0": '0xd78ad95fa46c994b6551d0da85fc275fe613ce37657fb8d5e3d130840159d822', #topic for Swap inside LPAddress
    "fromBlock": 26218413,
    "toBlock": "latest",
    "address": fetchedPair,

    })

print(filter.get_all_entries())

疑问解答 #

1. 是否可实现实时单次获取单个事件输出？ #

可以实现。核心思路是放弃一次性拉取历史事件的方式，改用实时新增事件监听：

• 使用web3.eth.get_filter_changes()方法，该方法只会返回自上次调用以来新增的事件，而非全量历史事件
• 结合循环逻辑，每次仅处理新增的单个/少量事件，实现单次输出单个事件的效果

2. 如何高效获取DEX上的购买事件，避免每分钟数千次调用？ #

可通过以下方式优化：

• 按区块周期轮询：根据目标链的区块生成周期设置休眠时间（比如BSC约3秒、ETH约12秒），避免高频请求
• 精准事件过滤：除了topic0匹配Swap事件，还可通过交易对的token0/token1顺序，过滤出仅属于「购买目标代币」的事件，减少无效数据返回
• 缓存已处理区块：记录已处理的区块号，异常重启后从该区块开始监听，避免重复查询
• 精简ABI：仅加载必要的函数和事件ABI，减少数据传输和解析开销

3. 带解释的参考代码 #

from web3 import Web3
import time

# 初始化Web3连接（替换为你的EVM节点URL）
node_url = "你的EVM节点URL"
web3 = Web3(Web3.HTTPProvider(node_url))

# 配置目标交易对
TARGET_TOKEN = Web3.toChecksumAddress("0x0e09fabb73bd3ade0a17ecc321fd13a19e81ce82")  # 目标代币地址
WRAPPED_NATIVE = Web3.toChecksumAddress("0xbb4CdB9CBd36B01bD1cBaEBF2De08d9173bc095c")  # WBNB地址
DEX_FACTORY_ADDRESS = Web3.toChecksumAddress("0xca143ce32fe78f1f7019d7d551a6402fc5350c73")  # PancakeSwap Factory地址

# 精简ABI：只保留所需的函数和事件
FACTORY_ABI = [
    {"constant":True,"inputs":[{"internalType":"address","name":"tokenA","type":"address"},{"internalType":"address","name":"tokenB","type":"address"}],"name":"getPair","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":False,"stateMutability":"view","type":"function"}
]
PAIR_ABI = [
    {"anonymous":False,"inputs":[{"indexed":True,"internalType":"address","name":"sender","type":"address"},{"indexed":True,"internalType":"address","name":"to","type":"address"},{"indexed":False,"internalType":"uint256","name":"amount0In","type":"uint256"},{"indexed":False,"internalType":"uint256","name":"amount1In","type":"uint256"},{"indexed":False,"internalType":"uint256","name":"amount0Out","type":"uint256"},{"indexed":False,"internalType":"uint256","name":"amount1Out","type":"uint256"}],"name":"Swap","type":"event"}
]
TOKEN_ABI = [
    {"constant":True,"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"payable":False,"stateMutability":"view","type":"function"}
]

# 获取交易对LP合约地址
factory_contract = web3.eth.contract(address=DEX_FACTORY_ADDRESS, abi=FACTORY_ABI)
lp_address = factory_contract.functions.getPair(TARGET_TOKEN, WRAPPED_NATIVE).call()
if lp_address == "0x0000000000000000000000000000000000000000":
    print("交易对不存在")
    exit()

# 获取代币小数位数，用于单位转换（从wei转为可读单位）
target_token_contract = web3.eth.contract(address=TARGET_TOKEN, abi=TOKEN_ABI)
wrapped_native_contract = web3.eth.contract(address=WRAPPED_NATIVE, abi=TOKEN_ABI)
target_decimals = target_token_contract.functions.decimals().call()
wrapped_decimals = wrapped_native_contract.functions.decimals().call()

# 创建Swap事件过滤器：默认从当前区块开始监听，不指定历史区块范围
swap_topic = web3.keccak(text="Swap(address,address,uint256,uint256,uint256,uint256)").hex()
swap_filter = web3.eth.filter({
    "address": lp_address,
    "topic0": swap_topic
})

print(f"开始实时监听 {TARGET_TOKEN} 购买事件...")

# 实时监听循环
while True:
    try:
        # 获取自上次调用以来新增的Swap事件
        new_events = web3.eth.get_filter_changes(swap_filter)
        for event in new_events:
            # 解析事件数据
            decoded_event = web3.eth.contract(address=lp_address, abi=PAIR_ABI).events.Swap().processLog(event)
            args = decoded_event['args']

            # 获取交易对的token0/token1，判断购买逻辑方向
            pair_contract = web3.eth.contract(address=lp_address, abi=[
                {"constant":True,"inputs":[],"name":"token0","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":False,"stateMutability":"view","type":"function"},
                {"constant":True,"inputs":[],"name":"token1","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":False,"stateMutability":"view","type":"function"}
            ])
            token0 = pair_contract.functions.token0().call()
            token1 = pair_contract.functions.token1().call()

            # 判断是否为购买目标代币的操作，并解析数据
            buyer_address = ""
            token_received = 0.0
            wrapped_spent = 0.0
            is_buy = False

            if token0 == TARGET_TOKEN:
                # 目标代币是token0：用户转WBNB（token1）进LP，获得token0
                if args['amount1In'] > 0 and args['amount0Out'] > 0:
                    is_buy = True
                    buyer_address = args['to']
                    token_received = args['amount0Out'] / (10 ** target_decimals)
                    wrapped_spent = args['amount1In'] / (10 ** wrapped_decimals)
            elif token1 == TARGET_TOKEN:
                # 目标代币是token1：用户转WBNB（token0）进LP，获得token1
                if args['amount0In'] > 0 and args['amount1Out'] > 0:
                    is_buy = True
                    buyer_address = args['to']
                    token_received = args['amount1Out'] / (10 ** target_decimals)
                    wrapped_spent = args['amount0In'] / (10 ** wrapped_decimals)

            # 输出购买事件信息
            if is_buy:
                print(f"=== 新购买事件 ===")
                print(f"买家地址: {buyer_address}")
                print(f"接收代币数量: {token_received:.6f}")
                print(f"花费WBNB金额: {wrapped_spent:.6f}")
                print(f"交易哈希: {event['transactionHash'].hex()}")
                print("------------------")

        # 按区块周期休眠，避免高频请求
        time.sleep(3)  # BSC区块周期约3秒，ETH可改为12秒

    except Exception as e:
        print(f"监听出错: {e}")
        time.sleep(10)  # 出错后休眠10秒再重试

代码关键解释：

• 精简ABI：只加载必要的函数和事件，减少解析开销
• 事件过滤：通过事件签名生成topic0，精准匹配Swap事件，默认监听当前及之后的区块
• 实时监听：使用get_filter_changes()仅获取新增事件，避免拉取历史数据
• 购买逻辑判断：根据交易对的token0/token1顺序，精准识别购买操作，解析所需变量
• 稳定机制：添加异常捕获和休眠重试，提升长时间运行的稳定性

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