Web3py获取事件列表遇ABI不匹配警告及数据提取等问题

一、解决MismatchedABI警告问题 #

警告根源是当前过滤的是代理合约的所有日志，其中包含ABI未覆盖的无关事件（比如代理合约的管理变更事件），导致web3.py尝试解码时失败。直接通过事件签名过滤，只获取StakeConfirmed事件的日志，就能彻底规避该警告。

二、实现事件字段提取与批量查询所有历史事件 #

核心优化点： #

• 按StakeConfirmed事件的topic创建过滤器，精准监听目标事件
• 循环分批次查询区块（规避10000块查询限制）
• 解码事件后提取staker、time，并将amount转换为ether单位

完整优化代码 #

import json
from web3 import Web3
import time

# 连接以太坊主网
infura_url = 'https://mainnet.infura.io/v3/e212e98e02a74730a1c6382e7ae6f95d'
w3 = Web3(Web3.HTTPProvider(infura_url))

# 代理合约地址与ABI（替换为正确的ABI内容）
ankr_pool = '0x84db6eE82b7Cf3b47E8F19270abdE5718B936670'
# 注意：请将下方的[真实ABI内容]替换为合约0xD201a7dF1D0F7E066EFdD448CBc8433F0b88C3e9的真实ABI字符串
ankr_pool_abi = json.loads("[真实ABI内容]")

contract = w3.eth.contract(address=ankr_pool, abi=ankr_pool_abi)
stake_confirmed_event = contract.events.StakeConfirmed

# 合约部署区块（根据实际情况调整，这里沿用原代码起始块）
START_BLOCK = 16128812
# 每次查询的区块数量（以太坊节点通常限制最大10000，可按需调整）
BATCH_SIZE = 10000

def process_stake_event(event_data):
    """处理单个StakeConfirmed事件，提取目标字段"""
    staker = event_data['args']['staker']
    timestamp = event_data['args']['time']
    # 将wei转换为ether单位
    amount_ether = w3.fromWei(event_data['args']['amount'], 'ether')
    print(f"Staker地址: {staker}")
    print(f"质押时间戳: {timestamp}")
    print(f"质押金额(ETH): {amount_ether}\n")

def fetch_all_stake_events():
    """循环分批次获取所有历史StakeConfirmed事件"""
    current_end_block = w3.eth.block_number
    start_block = START_BLOCK

    while start_block <= current_end_block:
        end_block = min(start_block + BATCH_SIZE - 1, current_end_block)
        print(f"正在查询区块范围: {start_block} - {end_block}")

        # 按事件topic创建过滤器，仅获取目标事件
        event_filter = stake_confirmed_event.create_filter(fromBlock=start_block, toBlock=end_block)
        events = event_filter.get_all_entries()

        for event in events:
            process_stake_event(event)

        # 更新起始区块，进入下一批次
        start_block = end_block + 1
        # 避免请求过于频繁触发节点限制
        time.sleep(1)

# 执行批量历史事件查询
fetch_all_stake_events()

# 可选：实时监听新的StakeConfirmed事件
def listen_new_events(poll_interval=2):
    print("\n开始监听新的质押事件...")
    event_filter = stake_confirmed_event.create_filter(fromBlock='latest')
    while True:
        new_events = event_filter.get_new_entries()
        for event in new_events:
            process_stake_event(event)
        time.sleep(poll_interval)

# 若需要实时监听，取消下方注释
# listen_new_events()

关键说明 #

1. ABI替换：务必将代码中的[真实ABI内容]替换为合约0xD201a7dF1D0F7E066EFdD448CBc8433F0b88C3e9的完整ABI字符串，确保StakeConfirmed事件的定义正确。
2. 区块范围调整：START_BLOCK可设置为合约实际部署区块，可通过区块链浏览器查询确认。
3. 延迟设置：批量查询时的time.sleep(1)用于避免触发节点请求频率限制，可根据节点规则调整时长。
4. 实时监听：代码末尾提供了实时监听新事件的函数，按需启用即可。

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