如何使用web3.py实时跟踪代币的最新转账记录
BSC代币转账实时监控实现方案
方案1:Web3.py原生实现(优先推荐)
核心是监听ERC20标准的Transfer事件,所有合规代币的转账操作都会触发该链上事件,无需依赖页面刷新获取数据。
- 前置准备
- 安装web3.py依赖:
pip install web3 - 准备可用的BSC RPC节点地址,优先选择支持WebSocket的节点以获得更好的实时性
- 安装web3.py依赖:
- 代码示例
from web3 import Web3 import asyncio # 初始化web3连接,此处替换为你自己的BSC WS RPC地址 w3 = Web3(Web3.WebsocketProvider("wss://你的BSC-WS-RPC地址")) # 目标代币合约地址 TOKEN_ADDRESS = "0x5774b2fc3e91af89f89141eacf76545e74265982" # ERC20 Transfer事件的最小ABI ERC20_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" } ] # 初始化合约实例 contract = w3.eth.contract(address=w3.to_checksum_address(TOKEN_ADDRESS), abi=ERC20_ABI) # 定义事件处理函数 def handle_transfer(event): tx_hash = event["transactionHash"].hex() from_addr = event["args"]["from"] to_addr = event["args"]["to"] value = event["args"]["value"] # 此处替换为你自己的业务逻辑,比如打印、存储、告警等 print(f"新转账检测到 | 交易哈希: {tx_hash} | 转出方: {from_addr} | 接收方: {to_addr} | 转账数量: {value}") # 实时监听事件 async def log_loop(event_filter, poll_interval): while True: for event in event_filter.get_new_entries(): handle_transfer(event) await asyncio.sleep(poll_interval) if __name__ == "__main__": # 创建Transfer事件过滤器,从最新区块开始监听 transfer_filter = contract.events.Transfer.create_filter(fromBlock="latest") loop = asyncio.get_event_loop() try: loop.run_until_complete(log_loop(transfer_filter, 2)) except KeyboardInterrupt: print("停止监听")
方案2:替代实现方案
如果节点获取有问题,可选择以下更简单的实现方式:
- 公开API轮询:调用区块链浏览器的公开代币转账查询接口,设置3-5秒的轮询间隔,每次拉取最新的10条转账记录,和本地缓存的最新交易哈希对比,出现新哈希即为新转账,该方案实现门槛极低,仅需处理接口请求逻辑即可
- 第三方链上监控服务:直接使用现成的链上事件监控工具,配置好目标合约地址和
Transfer事件后,可通过Webhook、邮件等方式推送新转账通知,无需自行开发监控逻辑
注意事项:
- 使用公共RPC或公开API时要遵守对应的频率限制,避免请求过于频繁被封禁
- 生产环境使用时建议校验交易的区块确认数,避免链上临时分叉导致的无效交易数据
- 解析
value字段时要按代币的小数位做单位换算,才能得到用户常规认知的转账金额
内容的提问来源于stack exchange,提问作者Mesut Karakuş
相关产品推荐
相关产品推荐

