基于web3及web3.py如何查询丢弃、替换及pending交易信息？

嘿，我来帮你把这几个问题拆解清楚——不管是通用Web3环境下的核心思路，还是用web3.py实现的具体方法，都给你讲明白：

首先得先搞懂这两类交易的本质：

• 已丢弃交易：指被节点从内存池（mempool）中彻底移除、永远不会被打包上链的交易，通常是因为长时间未被打包、gas价格过低，或者被用户主动取消（比如发送nonce相同但gas为0的交易）。
• 被替换交易：一般是同一个地址发送了相同nonce但gas费更高的交易，节点会优先打包gas更高的版本，原交易被挤出内存池。

通用的查询思路围绕节点内存池和历史数据展开：

• 依赖节点的内存池访问权限：大部分主流节点（Geth、Erigon等）都提供了内存池相关的API，但公共节点通常会限制这类接口，所以优先用自己搭建的节点或支持mempool访问的服务。
• 监听内存池变化：通过节点的订阅接口（如eth_subscribe）跟踪pending交易，一旦发现同一个地址出现重复nonce的交易，就能判断原交易可能被替换。
• 利用Debug类API：部分节点（如Geth）的debug接口可以查询历史内存池中的交易，但需要节点启动时开启debug模式，且权限较高。

下面是针对你的开发场景，用web3.py实现的具体代码方案：

1. 获取Pending（待处理）交易信息 #

方法一：批量获取pending区块的所有交易 #

pending区块包含了当前内存池中的所有待处理交易，你可以直接通过get_block方法获取：

from web3 import Web3

# 连接到节点（这里以本地Geth节点为例，也可以用支持mempool的远程节点）
w3 = Web3(Web3.HTTPProvider("http://localhost:8545"))

# 获取pending区块的完整数据（包含交易详情）
pending_block = w3.eth.get_block("pending", full_transactions=True)

# 遍历输出所有pending交易信息
for tx in pending_block["transactions"]:
    print(f"交易哈希：{tx.hash.hex()}")
    print(f"发送地址：{tx['from']}")
    print(f"Nonce：{tx['nonce']}")
    print(f"Gas价格：{w3.from_wei(tx['gasPrice'], 'gwei')} Gwei")
    print("---")

方法二：实时监听新进入内存池的pending交易 #

如果需要实时接收新的pending交易，推荐用WebSocket连接节点（HTTP轮询效率较低）：

from web3 import Web3

# 用WebSocket连接节点
w3 = Web3(Web3.WebsocketProvider("ws://localhost:8546"))

def handle_new_pending_tx(tx_hash):
    """处理新的pending交易回调"""
    tx = w3.eth.get_transaction(tx_hash)
    print(f"新Pending交易：{tx_hash.hex()} | 地址：{tx['from']} | Nonce：{tx['nonce']}")

# 订阅pending交易
subscription_id = w3.eth.subscribe("newPendingTransactions", handle_new_pending_tx)

# 保持监听（可以根据需求添加停止逻辑）
try:
    while True:
        pass
except KeyboardInterrupt:
    w3.eth.unsubscribe(subscription_id)
    print("已停止监听pending交易")

2. 查询已被替换/丢弃的交易信息 #

这类交易已经不在内存池中了，需要结合地址nonce追踪和节点历史数据判断：

方式一：通过地址Nonce变化判断 #

每个地址的nonce是递增的，已打包交易的最大nonce+1就是当前的下一个可用nonce。如果原pending交易的nonce小于当前地址的可用nonce，且没有找到对应nonce的已打包交易，说明原交易被丢弃；如果找到了相同nonce的已打包交易，说明原交易被替换。

def check_tx_status(address, target_nonce):
    """检查指定地址的目标nonce交易是否被替换/丢弃"""
    # 获取地址当前的可用nonce（已打包交易的最大nonce+1）
    current_nonce = w3.eth.get_transaction_count(address)
    
    # 遍历最近100个区块，查找是否有相同nonce的已打包交易
    latest_block = w3.eth.block_number
    for block_num in range(latest_block, max(latest_block - 100, 0), -1):
        block = w3.eth.get_block(block_num, full_transactions=True)
        for tx in block["transactions"]:
            if tx["from"].lower() == address.lower() and tx["nonce"] == target_nonce:
                print(f"✅ Nonce {target_nonce}的交易已被替换，新交易哈希：{tx.hash.hex()}")
                return "replaced"
    
    # 如果当前nonce大于目标nonce，且未找到对应已打包交易，说明交易被丢弃
    if current_nonce > target_nonce:
        print(f"❌ Nonce {target_nonce}的交易已被丢弃，地址当前可用nonce：{current_nonce}")
        return "dropped"
    
    # 否则交易可能还在pending状态
    print(f"ℹ️ Nonce {target_nonce}的交易仍在pending状态")
    return "pending"

# 示例：检查地址0x123...的nonce为5的交易状态
check_tx_status("0xYourTargetAddress", 5)

方式二：使用节点Debug API（需节点开启debug模式） #

如果你的节点开启了debug功能（比如Geth启动时添加--debug参数），可以调用debug接口查询历史内存池中的交易：

# 查询指定哈希的原始交易（如果还在节点历史内存池中）
raw_tx = w3.debug.get_raw_transaction("0xTargetTxHash")
if raw_tx:
    # 解码原始交易数据
    decoded_tx = w3.eth.account.decode_transaction(raw_tx)
    print(f"找到交易详情：{decoded_tx}")
else:
    print("该交易已被从内存池移除（丢弃或被替换）")

⚠️ 注意：Debug API权限较高，公共节点一般不会开放，建议用自己搭建的节点测试。

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