如何用Web3py获取指定合约ERC20代币交易记录并实时监控

无返回结果的排查&修复方法 #

• 确认节点网络匹配：你查询的是Rinkeby测试网合约，首先验证web3连接的链ID是否为4，可执行print(web3.eth.chain_id)校验，若不匹配替换为Rinkeby测试网的RPC节点即可。
• 修正事件签名：标准ERC20的Transfer事件包含3个参数（转出地址、转入地址、转账金额），你当前的签名少了一个地址参数，导致哈希计算错误。正确的签名计算方式为：

eventSignatureHash = web3.keccak(text='Transfer(address,address,uint256)').hex()

• 补充日志获取逻辑：你当前代码仅创建了过滤器，未调用获取日志的方法，添加日志获取代码即可拿到结果：

# 获取历史所有匹配日志
logs = filter.get_all_entries()
# 若查询范围过大节点报错，可分批指定fromBlock、toBlock参数分段查询

• 公共RPC节点限制规避：多数公共RPC仅支持查询最近1000~10000个块的历史日志，若需要查询全量历史，可使用归档节点或者按块高分段循环查询。

实时监控交易的实现方法 #

• 基于websocket节点的推送监听：使用支持websocket的RPC节点，通过watch方法绑定回调函数，有新交易产生时自动触发处理逻辑：

def process_new_transfer(event):
    # 自定义交易处理逻辑，比如解析参数、落库、告警
    print(f"收到新转账交易：{event.transactionHash.hex()}")

filter.watch(process_new_transfer)

• 通用轮询方案：若没有websocket节点，可定期轮询过滤器的新条目实现监控：

import time
while True:
    new_logs = filter.get_new_entries()
    for log in new_logs:
        process_new_transfer(log)
    time.sleep(3) # 可根据需求调整轮询间隔，避免触发节点限流

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