如何使用web3.py替换待处理交易 提高Gas优先转出钱包资产

现有代码核心问题 #

• Pending交易过滤器依赖单节点连接，遇到节点同步延迟、连接闪断会直接漏抓黑客发起的交易，且没有异常重连逻辑，连接断开后脚本直接崩溃退出
• 单位换算、余额计算存在逻辑错误：Gas成本计算重复做单位换算，导致算出的最大转账额经常超过账户实际可用余额，交易被节点以「余额不足」为由直接丢弃，不会进入交易池
• Gas溢价计算逻辑存在精度偏差，实际设置的Gas价格经常达不到比黑客交易高20%的预期，无法实现抢跑
• 仅向单个节点发送一次交易，没有冗余广播、重试机制：遇到单个节点交易池满、本地校验规则拦截时，交易根本不会传播到全网
• 固定20秒的运行窗口过短，且没有交易上链状态校验，要么提前退出错过交易时机，要么发送一次失败后没有后续动作

修复后可稳定运行的代码 #

import time
from decimal import Decimal
from web3 import Web3, exceptions

# -------------------------- 配置项 --------------------------
# 多节点RPC列表，建议混合搭配自有节点、不同服务商的节点，提升广播覆盖率
RPC_LIST = [
    "你的RPC节点地址1",
    "你的RPC节点地址2",
    "你的RPC节点地址3"
]
w3_list = [Web3(Web3.HTTPProvider(rpc)) for rpc in RPC_LIST]
# 默认用第一个节点做链上数据查询
w3 = w3_list[0]

address_from = "存在被盗风险的钱包地址"
recipient = "资产接收地址".replace(' ', '')
private_key = "风险钱包私钥"
chain_id = 1 # 对应链ID，以太坊主网为1，BSC为56，根据实际链修改
GAS_PREMIUM = 1.2  # Gas比黑客交易高20%，可根据需要调高到1.5/2提升抢跑成功率
GAS_LIMIT = 21000
# -----------------------------------------------------------

def broadcast_transaction(signed_tx):
    """多节点冗余广播交易"""
    for node_w3 in w3_list:
        try:
            node_w3.eth.sendRawTransaction(signed_tx.rawTransaction)
        except Exception as e:
            # 忽略不影响广播的已知错误
            if "already known" not in str(e) and "nonce too low" not in str(e):
                continue

def get_confirmed_nonce():
    """获取当前地址已上链确认的最新nonce"""
    return w3.eth.getTransactionCount(address_from, "latest")

if __name__ == "__main__":
    current_nonce = get_confirmed_nonce()
    print(f"当前已确认nonce: {current_nonce}, 监控异常交易中...")
    target_nonce = None
    last_send_time = 0
    signed_tx = None

    while True:
        # 自动重建pending过滤器，避免连接断开后过滤器失效
        try:
            pending_filter = w3.eth.filter("pending")
            while True:
                # 已捕获黑客交易后，持续检测交易上链状态+重发抢跑交易
                if target_nonce is not None:
                    onchain_nonce = get_confirmed_nonce()
                    if onchain_nonce > target_nonce:
                        print(f"nonce {target_nonce} 已上链，资产转出完成")
                        exit(0)
                    # 每隔100ms重发一次交易，提升被矿工打包的概率
                    if time.time() - last_send_time > 0.1:
                        broadcast_transaction(signed_tx)
                        last_send_time = time.time()

                # 拉取最新pending交易
                try:
                    pending_tx_hashes = pending_filter.get_new_entries()
                except exceptions.ConnectionError:
                    # 连接断开时跳出内层循环，重建过滤器
                    break

                for tx_hash in pending_tx_hashes:
                    try:
                        pending_tx = w3.eth.getTransaction(tx_hash)
                    except Exception:
                        continue
                    # 检测到黑客从目标地址发起的、非我方接收地址的交易
                    if pending_tx.get("from") == address_from and pending_tx.get("to") != recipient:
                        print(f"检测到黑客可疑交易: {w3.toHex(tx_hash)}")
                        target_nonce = pending_tx.get("nonce")
                        hacker_gas_price = pending_tx.get("gasPrice")
                        # 修正Gas计算：直接在黑客Gas基础上加溢价，单位保留wei避免换算误差
                        target_gas_price = int(hacker_gas_price * GAS_PREMIUM)
                        # 修正余额计算：全链路用wei单位计算，避免精度丢失
                        balance_wei = w3.eth.getBalance(address_from)
                        gas_cost_wei = target_gas_price * GAS_LIMIT
                        # 留1000wei余量，避免因精度问题触发余额不足报错
                        transfer_value_wei = balance_wei - gas_cost_wei - 1000
                        if transfer_value_wei <=0:
                            print("余额不足以覆盖Gas，转出失败")
                            exit(1)
                        # 构造抢跑交易
                        tx = {
                            "chainId": chain_id,
                            "nonce": target_nonce,
                            "to": recipient,
                            "value": transfer_value_wei,
                            "gas": GAS_LIMIT,
                            "gasPrice": target_gas_price,
                        }
                        signed_tx = w3.eth.account.signTransaction(tx, private_key)
                        print(f"开始广播抢跑交易, nonce: {target_nonce}, GasPrice: {w3.fromWei(target_gas_price, 'gwei')} gwei")
                        broadcast_transaction(signed_tx)
                        last_send_time = time.time()
                time.sleep(0.05)
        except Exception as e:
            print(f"过滤器连接异常，1秒后重连: {str(e)}")
            time.sleep(1)
            continue

额外优化建议 #

• 节点尽量选择和矿池直连的低延迟节点，比公共节点能更早收到pending交易，广播的交易也能更快进入矿池
• 如果目标链支持EIP1559，可以把Gas费设置改成maxFeePerGas和maxPriorityFeePerGas模式，抢跑成功率更高
• 私钥不要硬编码在脚本里，运行前临时输入即可，避免私钥泄露

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