手把手教你用 web3.py 按区块快速抓取 BSC/ETH 全部交易

为什么区块维度值得深挖？

在以太坊（ETH）和币安智能链（BSC）上，每一秒都会产生数十笔、甚至上千笔链上交易。与其全网扫交易哈希，不如按区块号精确锁定时间窗口，批量获取该区块全部交易列表，不仅省时省钱，也大幅降低漏报率。
核心关键词：web3.py、BSC、ETH、区块交易、批量抓取、input 数据、交易解析、getBlock、geth_poa_middleware。

环境准备

只需三步：

Python 3.8+
安装最新版 web3：pip install web3==6.*
一个稳定的 RPC 节点，自建或公共节点均可（文末 FAQ 教你挑节点）。

👉 不想折腾节点？点这里一键获取高可用公共 RPC 列表。

核心代码拆解

下面给出一段经过圆润升级的示例，功能比原文更丰富：

支持 ETH & BSC 双链；
引入常见 DEX Router 交易识别；
捕获异常并带重试。

from web3 import Web3
from web3.middleware import geth_poa_middleware
import time
import traceback

ETH_RPC = "https://eth.llamarpc.com"
BSC_RPC = "https://bsc-dataseed.binance.org/"

# 主流 DEX：swapExactTokensForTokens、swapETHForExactTokens...
BUY_METHODS = {
    "0x38ed1739",  # swapExactTokensForTokens
    "0x7ff36ab5",  # swapExactETHForTokens
    "0x8803dbee",  # swapTokensForExactETH
    "0xfb3bdb41"   # swapExactETHForTokensSupportingFeeOnTransferTokens
}

def create_client(rpc: str, poa: bool = False) -> Web3:
    w3 = Web3(Web3.HTTPProvider(rpc))
    if poa:  # BSC、Polygon 这类 PoA/Clique 链必须加
        w3.middleware_onion.inject(geth_poa_middleware, layer=0)
    assert w3.isConnected(), "RPC 连接失败"
    return w3

def scan_block(w3: Web3, block_number: int):
    """返回指定区块所有交易对象（含 input）"""
    try:
        block = w3.eth.get_block(block_number, full_transactions=True)
        return block.transactions
    except Exception:
        traceback.print_exc()
        return []

def is_buy(tx) -> bool:
    """简易判定是否为常见 DEX Router 买币方法"""
    return tx.input[:10] in BUY_METHODS

def analyze_transactions(w3: Web3, block_number: int):
    for tx in scan_block(w3, block_number):
        if is_buy(tx):
            print(f"发现买币 tx: {tx.hash.hex()} 区块={block_number} From={tx['from']}")

if __name__ == "__main__":
    w3 = create_client(BSC_RPC, poa=True)  # 换 ETH_RPC 亦可
    assert w3.isConnected()

    # 随意选个历史区块测试
    analyze_transactions(w3, block_number=28_000_000)

区块交易的 5 种妙用

做市商高频监控：统计滑点、构建实时深度。
套利机器人：比全网 mempool 扫描的抢跑率更高。
NDR（Next-Day Reversal）策略：整理日化资金流向，筛强势代币。
审计溯源：定位闪电贷、跨桥攻击的入口交易。
Gas 行情追踪：通过字段 gasPrice 快速评估网络拥堵。

👉 学习如何把解析结果自动同步到数据库，实现可视化追踪。

FAQ：最常被问到的 6 个问题

1. 为什么有时 `w3.eth.get_block` 会报错？

八成是因为没在 BSC（或 Polygon、Avalanche 等 PoA 链）上加上 geth_poa_middleware。千万记得在调用第一秒前 inject，否则 extraData 字段检验直接崩溃。

2. `full_transactions=True` 会不会撑爆内存？

以 BSC 为例，一个区块极限 3,000 笔交易左右，单条交易对象约 0.8 KB，实际占用 < 3 MB，只要不在极端高并发下都可稳吃。

3. 如何确认自己识别的方法签名是完整的？

官方仓库见 Ethereum Signature Database，输入前 8 位即可校验。

4. 我只想扫描过去 10000 个区块，怎么并行最快？

可使用 Python 的 asyncio+httpx 移步 HTTP 请求层做并发；或直接用 eth_getBlockReceipts（服务器需支持）。

5. 节点响应烂、重试老是超时怎么办？

建议自建或购买云节点；公共节点有速率限制。轮询时添加指数回退：
time.sleep(2 ** attempt)
既能缓压，也防被 Ban。

6. 除了 `input`，还有哪些字段值得监控？

to：收款合约可过滤池子。
value：直接看出 ETH、BNB 原生资产迁移。
gasPrice：推测对手优先级。
logs：事件解析，得到实际兑换量。

小结

通过本文，你已掌握快速按区块抓取 ETH、BSC 全部交易的核心思路与实操代码。记得根据业务需求，灵活调整识别的 methodId、交易方向，以及缓存机制。祝你在下一轮「链上淘金」中先人一步！