基于Web3.py实现BSC网络多钱包代币交易跟踪的问题咨询

问题1：钱包地址列表正确传入filter函数的方法 #

ERC20标准的Transfer(address,address,uint256)事件符合以太坊日志的topic规则，按以下逻辑构造过滤参数即可实现精准筛选：

• 事件签名哈希固定为0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef，必须作为topics数组的第一个元素
• 该事件的from（转出地址）和to（转入地址）为indexed参数，对应topics数组的第2、第3位，未indexed的value（转账金额）存于日志的data字段
• 以太坊过滤器规则：同一位的多个topic值为或关系，不同位为且关系，将转码后的监控地址同时放在第2、3位，即可同时匹配转入或转出监控地址的交易

普通42位以太坊地址需要补零为32字节长度的hex字符串才能作为topic参数，转码方法如下：

def address_to_topic(addr: str) -> str:
    # 统一转小写保证匹配，补零到32字节长度
    return Web3.to_hex(Web3.to_bytes(hexstr=addr.lower()).rjust(32, b'\x00'))

filter参数构造示例：

# 预处理监控地址为topic格式
processed_addrs = [address_to_topic(addr) for addr in address_list]
# Transfer事件签名哈希
TRANSFER_EVENT_SIG = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
# 构造过滤规则：Transfer事件 AND (转出是监控地址 OR 转入是监控地址)
filter_params = {
    "topics": [
        TRANSFER_EVENT_SIG,
        processed_addrs,
        processed_addrs
    ]
}
web3_filter = web3.eth.filter(filter_params)

额外说明：如果需要仅监控指定代币的Transfer事件，可在filter_params中新增address字段，值为对应代币合约地址的列表，进一步过滤冗余数据

问题2：HexBytes类型数据与Transfer事件解码方法 #

无需手动处理HexBytes类型，直接用Web3.py内置的ABI事件解析能力即可快速提取日志数据，步骤如下：

1. 定义Transfer事件的最小ABI，初始化事件对象：

ERC20_TRANSFER_ABI = [
    {
        "anonymous": False,
        "inputs": [
            {"indexed": True, "name": "from", "type": "address"},
            {"indexed": True, "name": "to", "type": "address"},
            {"indexed": False, "name": "value", "type": "uint256"},
        ],
        "name": "Transfer",
        "type": "event",
    }
]
# 初始化事件解析对象
transfer_event = web3.eth.contract(abi=ERC20_TRANSFER_ABI).events.Transfer()

2. 遍历日志时直接调用process_log方法完成解析：

while True:
    for log in web3_filter.get_new_entries():
        parsed = transfer_event.process_log(log)
        print(f"代币地址: {parsed['address']}")
        print(f"转出地址: {parsed['args']['from']}")
        print(f"转入地址: {parsed['args']['to']}")
        print(f"转账金额: {parsed['args']['value']}")
        print(f"交易哈希: {parsed['transactionHash'].hex()}")
        print("---")

如果需要手动处理HexBytes，直接调用.hex()方法即可转为十六进制字符串，也可直接作为bytes类型传入ABI解码函数。

完整修正后的代码 #

from web3 import Web3

def address_to_topic(addr: str) -> str:
    return Web3.to_hex(Web3.to_bytes(hexstr=addr.lower()).rjust(32, b'\x00'))

# 配置参数
address_list = ['0x67fdE6D04a82689a59E5188f9B572CBeF53D4763']
BSC_WSS_ENDPOINT = 'wss://bsc.getblock.io/mainnet/?api_key=your_api_key'
TRANSFER_EVENT_SIG = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef"
ERC20_TRANSFER_ABI = [
    {
        "anonymous": False,
        "inputs": [
            {"indexed": True, "name": "from", "type": "address"},
            {"indexed": True, "name": "to", "type": "address"},
            {"indexed": False, "name": "value", "type": "uint256"},
        ],
        "name": "Transfer",
        "type": "event",
    }
]

# 初始化web3连接
web3 = Web3(Web3.WebsocketProvider(BSC_WSS_ENDPOINT))
transfer_event = web3.eth.contract(abi=ERC20_TRANSFER_ABI).events.Transfer()

# 构造过滤器
processed_addrs = [address_to_topic(addr) for addr in address_list]
filter_params = {
    "topics": [
        TRANSFER_EVENT_SIG,
        processed_addrs,
        processed_addrs
    ]
}
web3_filter = web3.eth.filter(filter_params)

# 监听事件
while True:
    for log in web3_filter.get_new_entries():
        try:
            parsed = transfer_event.process_log(log)
            print(f"代币地址: {parsed['address']}")
            print(f"转出: {parsed['args']['from']}")
            print(f"转入: {parsed['args']['to']}")
            print(f"金额: {parsed['args']['value']}")
            print(f"交易哈希: {parsed['transactionHash'].hex()}\n---")
        except Exception:
            # 跳过解析异常的日志
            continue

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