如何用web3.py监测以太坊主网新部署的ERC20智能合约

一、区分合约部署交易与普通交易 #

你之前通过tx["to"]是否为空判断部署交易的方法有局限，因为CREATE2方式部署合约的交易是调用工厂合约的方法，此时tx["to"]是工厂合约地址，并非空值。

正确的判断逻辑是：检查交易收据（transaction receipt）中是否存在contractAddress字段且该字段不为空。无论是CREATE（to为空）还是CREATE2（to为工厂合约）部署的合约，只要部署成功，交易收据里都会生成contractAddress字段，指向新创建的合约地址。

二、区分ERC20合约与其他类型合约 #

ERC20合约遵循统一的接口标准，我们可以通过以下两种方式验证：

• 检查标准接口方法：尝试调用ERC20的核心方法（如name()、symbol()、decimals()），如果合约能正常返回结果，大概率是ERC20合约。
• 验证接口ID（ERC165）：如果合约实现了ERC165标准，可以查询其是否支持ERC20的接口ID（0x36372b07）。不过部分早期ERC20合约未实现ERC165，所以第一种方法更通用。

改进后的代码示例 #

from web3 import Web3
import time

# 初始化Web3连接
provider_url = 'https://mainnet.infura.io/v3/c4c4fa6e6baf4d02b165d0a0ee4d2485'
w3 = Web3(Web3.HTTPProvider(provider_url))
print("连接状态:", w3.is_connected())

# ERC20接口方法签名
ERC20_INTERFACE = [
    {
        "constant": True,
        "inputs": [],
        "name": "name",
        "outputs": [{"name": "", "type": "string"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [],
        "name": "symbol",
        "outputs": [{"name": "", "type": "string"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function"
    },
    {
        "constant": True,
        "inputs": [],
        "name": "decimals",
        "outputs": [{"name": "", "type": "uint8"}],
        "payable": False,
        "stateMutability": "view",
        "type": "function"
    }
]

def is_erc20_contract(contract_address):
    """验证地址是否为ERC20合约"""
    try:
        contract = w3.eth.contract(address=contract_address, abi=ERC20_INTERFACE)
        # 尝试调用三个核心方法，若都能正常返回则判定为ERC20
        contract.functions.name().call()
        contract.functions.symbol().call()
        contract.functions.decimals().call()
        return True
    except Exception as e:
        # 调用失败说明不是标准ERC20合约
        return False

def monitor_new_erc20():
    last_block = w3.eth.block_number
    print(f"开始监测，当前区块高度: {last_block}")
    
    while True:
        current_block = w3.eth.block_number
        if current_block > last_block:
            # 遍历从last_block+1到current_block的所有区块
            for block_num in range(last_block + 1, current_block + 1):
                block = w3.eth.get_block(block_num, full_transactions=True)
                print(f"处理区块: {block_num}")
                
                for tx in block.transactions:
                    tx_receipt = w3.eth.get_transaction_receipt(tx.hash)
                    # 判断是否是合约部署交易
                    if tx_receipt.contractAddress:
                        contract_addr = tx_receipt.contractAddress
                        print(f"发现新部署合约: {contract_addr}")
                        # 验证是否为ERC20
                        if is_erc20_contract(contract_addr):
                            print(f"✅ 检测到ERC20合约: {contract_addr}")
                            # 这里可以添加后续逻辑，比如保存地址到数据库等
        last_block = current_block
        # 每5秒轮询一次，避免请求过于频繁
        time.sleep(5)

if __name__ == "__main__":
    monitor_new_erc20()

代码说明 #

• 区块监测逻辑：改为跟踪区块高度，避免重复处理同一区块的交易，同时支持处理遗漏的区块（如果程序短暂中断）。
• 合约部署判断：通过tx_receipt.contractAddress字段判断，覆盖CREATE和CREATE2两种部署方式。
• ERC20验证：通过调用标准接口方法验证，排除非ERC20合约。
• 请求频率控制：添加time.sleep(5)避免对节点发起过多请求，防止被限流。

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