第1章：Web3.py实战：连接节点、监听Pending交易、解析交易数据、Gas Price动态调整

各位同学，欢迎来到MEV实战的第一课。

说实话，很多新手一上来就想着怎么抢跑、怎么三明治攻击，结果连节点都没连明白。我见过太多人花大价钱买了节点服务，结果代码里连个连接池都没配，跑两天就被封了IP。

这一章，咱们先把地基打牢。你想想看，连Pending交易都抓不到，还谈什么MEV？

1.1 连接节点：选对Provider是关键

Web3.py连接节点，说白了就是找个中间人帮你跟链上通信。我个人习惯用HTTPProvider，简单直接。但如果你要做高频监听，我建议用WebsocketProvider——延迟更低，连接更稳定。

from web3 import Web3

# 连接本地节点（我推荐用Geth）
w3 = Web3(Web3.HTTPProvider('http://127.0.0.1:8545'))

# 或者用Infura（适合测试，别上主网）
# w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR_KEY'))

# 检查连接
print(f"已连接: {w3.is_connected()}")
print(f"当前区块: {w3.eth.block_number}")

嗯，这里要注意。如果你用Infura，记得配个备用节点。我在项目中遇到过Infura突然宕机，整整5分钟没抓到任何交易，那感觉...你懂的。

1.2 监听Pending交易：别用轮询，用订阅

很多教程教你怎么轮询pending交易，但说实话，那效率太低了。Web3.py支持异步订阅，这才是正确姿势。

import asyncio
from web3 import Web3
from web3.middleware import geth_poa_middleware

async def listen_pending():
    # 用Websocket连接，延迟更低
    w3 = Web3(Web3.WebsocketProvider('wss://mainnet.infura.io/ws/v3/YOUR_KEY'))
    
    # 如果是BSC等PoA链，需要加这行
    w3.middleware_onion.inject(geth_poa_middleware, layer=0)
    
    print("开始监听Pending交易...")
    
    async with w3.listen_on_pending() as subscription:
        async for tx_hash in subscription:
            # 这里拿到的是交易哈希，需要进一步解析
            print(f"捕获到Pending交易: {tx_hash.hex()}")
            
            # 获取完整交易数据
            tx = await w3.eth.get_transaction(tx_hash)
            if tx and tx['to']:
                print(f"目标地址: {tx['to']}")
                print(f"Gas Price: {tx['gasPrice']}")

# 运行
asyncio.run(listen_pending())

1.3 解析交易数据：从哈希到完整信息

拿到交易哈希只是第一步。真正的MEV玩家，需要解析出交易里的每一比特信息。

我一般会解析这几个关键字段：

字段	说明	MEV用途
from	发送方地址	识别大户/机器人
to	接收方地址	判断目标合约
value	转账金额(wei)	评估套利空间
gasPrice	Gas价格	判断竞争激烈程度
input	调用数据	解析具体操作

def parse_transaction(tx):
    """解析交易数据，返回结构化信息"""
    if not tx:
        return None
    
    parsed = {
        'hash': tx['hash'].hex(),
        'from': tx['from'],
        'to': tx['to'],
        'value': tx['value'],
        'gas': tx['gas'],
        'gasPrice': tx['gasPrice'],
        'nonce': tx['nonce'],
        'input': tx['input'].hex() if tx['input'] else '0x',
        'is_contract_creation': tx['to'] is None
    }
    
    # 解析input数据（如果是ERC20转账）
    if len(parsed['input']) >= 138:  # 0x + 4字节方法ID + 32*4参数
        method_id = parsed['input'][:10]
        if method_id == '0xa9059cbb':  # transfer方法
            parsed['token_transfer'] = {
                'to': '0x' + parsed['input'][34:74],
                'amount': int(parsed['input'][74:138], 16)
            }
    
    return parsed

1.4 Gas Price动态调整：抢跑的核心竞争力

为什么你的交易总是抢不过别人？说白了，就是Gas Price没调好。

静态Gas Price在MEV里根本行不通。你需要实时监控mempool里的竞争情况，动态调整你的出价。

class GasPriceManager:
    """动态Gas Price管理器"""
    
    def __init__(self, w3, base_multiplier=1.1):
        self.w3 = w3
        self.base_multiplier = base_multiplier
        self.recent_prices = []  # 记录最近100笔交易的Gas Price
    
    def get_optimal_gas_price(self):
        """获取最优Gas Price"""
        # 获取当前网络建议价格
        base_gas = self.w3.eth.gas_price
        
        # 如果mempool里有竞争，提高出价
        if self.recent_prices:
            avg_competitor_price = sum(self.recent_prices) / len(self.recent_prices)
            # 比平均价高10%
            optimal = max(base_gas, int(avg_competitor_price * 1.1))
        else:
            optimal = int(base_gas * self.base_multiplier)
        
        return optimal
    
    def update_competitor_prices(self, pending_tx):
        """更新竞争对手的Gas Price"""
        if pending_tx and pending_tx['gasPrice']:
            self.recent_prices.append(pending_tx['gasPrice'])
            # 只保留最近100条
            if len(self.recent_prices) > 100:
                self.recent_prices.pop(0)

1.5 本章知识体系

下面这张图，是我自己总结的Web3.py实战核心流程。你照着这个思路走，基本不会跑偏。

嗯，这张图基本涵盖了本章的核心内容。你照着这个流程走一遍，至少能跑通一个基础的MEV监听系统。

1.6 实战建议

最后，给你几个我踩坑踩出来的建议：

• 先跑测试网——别一上来就上主网，Gas费烧不起。我刚开始时一天烧了0.5个ETH，心疼得不行。
• 日志要打全——每个关键步骤都打日志。出了问题，日志就是你最好的debug工具。
• 做好异常处理——节点会断、网络会卡、交易会失败。你的代码得扛得住这些。
• 别贪心——不是每笔Pending交易都值得抢。学会放弃，反而能赚更多。

好了，这一章的内容就到这里。代码都给你了，剩下的就是动手实践。记住，MEV不是看会的，是跑出来的。