使用Web3和Python检测Polygon链PairCreated事件失效问题求助

核心调整点 #

• 替换节点配置：你代码中仍在使用BSC的节点变量bscNode，需要替换为Polygon主网的Websocket节点地址，注意事件订阅必须使用WS协议，不能用HTTP节点。
• 放弃createFilter接口：大部分Polygon公共节点禁用了eth_newFilter系列接口（该接口资源消耗高，Polygon节点提供商普遍限制），改用直接拉取区块日志的方式实现事件检测，这是最核心的适配点。
• 补全异常日志输出：你当前代码用裸except吞掉了所有错误，无法定位节点返回的权限、接口不支持等报错，需要补充错误打印逻辑方便排查。
• 替换递归重启逻辑：finally块中直接递归调用listenForTokens会随着重启次数增加出现栈溢出，改用循环结构实现重连更稳定。
• 确认ABI匹配：虽然QuickSwap是Uniswap V2 fork，仍建议直接从QuickSwap Factory合约（地址0x5757371414417b8C6CAad45bAeF941aBc7d3Ab32）拉取原生ABI，避免BSC ABI的隐式不兼容问题。

修改后可运行核心代码 #

import web3, json, asyncio
from web3 import Web3

# 替换为你自己的Polygon WS节点地址
POLYGON_WS_NODE = "wss://你的Polygon节点WS地址"
QUICKSWAP_FACTORY = "0x5757371414417b8C6CAad45bAeF941aBc7d3Ab32"
# 替换为QuickSwap Factory的原生ABI
LISTENING_ABI = [你对应的ABI内容]

web3 = Web3(Web3.WebsocketProvider(POLYGON_WS_NODE))
contract = web3.eth.contract(address=QUICKSWAP_FACTORY, abi=LISTENING_ABI)

def foundToken(event):
    jsonEventContents = json.loads(Web3.toJSON(event))
    # 原有代币处理逻辑
    print("新Pair检测到:", jsonEventContents)

async def token_loop(poll_interval=2):
    # 直接通过日志匹配检测事件，无需调用createFilter
    event_topic = web3.keccak(text="PairCreated(address,address,address,uint256)").hex()
    from_block = web3.eth.block_number
    while True:
        try:
            current_block = web3.eth.block_number
            if current_block > from_block:
                logs = web3.eth.get_logs({
                    "fromBlock": from_block,
                    "toBlock": current_block,
                    "address": QUICKSWAP_FACTORY,
                    "topics": [event_topic]
                })
                for log in logs:
                    parsed_event = contract.events.PairCreated().process_log(log)
                    foundToken(parsed_event)
                from_block = current_block + 1
            await asyncio.sleep(poll_interval)
        except Exception as e:
            print("轮询出错:", str(e))
            await asyncio.sleep(1)
            # 出错后重置起始区块避免漏单
            from_block = web3.eth.block_number

def listen_for_tokens():
    # 改用循环代替递归避免栈溢出
    while True:
        try:
            loop = asyncio.get_event_loop()
            loop.run_until_complete(token_loop(2))
        except Exception as e:
            print("主进程出错，1秒后重启:", str(e))
            asyncio.sleep(1)

if __name__ == "__main__":
    listen_for_tokens()

额外优化建议 #

• 优先使用付费Polygon节点的WS接口，免费公共节点的可用性和延迟都无法满足实时检测新币的需求
• 可以调整轮询间隔到1秒，进一步降低检测延迟，Polygon出块时间约为2秒，1秒轮询足够覆盖所有新区块

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