3、数据获取实战：使用Web3.py连接以太坊节点、解析ERC20转账事件、获取Uniswap V3池子的状态

好，咱们直接进入正题。

前面聊了那么多理论，现在该动手了。这一章，我会带你走一遍完整的流程：怎么连上链、怎么拿到数据、怎么解析成我们能用的东西。说白了，就是让你亲手摸到链上的“活数据”。

3.1 准备工作：环境与节点

先说说环境。我个人习惯用 Python 3.9+，配合 web3.py 这个库。安装很简单：

pip install web3

然后你需要一个节点。你可以跑自己的节点（比如 Geth），也可以用第三方服务。我早期做项目时图省事，直接用了 Infura 的免费节点。嗯，这里要注意：免费节点有速率限制，高频请求会被卡住。后来我换成了 Alchemy，稳定性好不少。

连接节点的代码长这样：

from web3 import Web3

# 用你的 Infura 或 Alchemy 的 URL
RPC_URL = "https://mainnet.infura.io/v3/YOUR_PROJECT_ID"
w3 = Web3(Web3.HTTPProvider(RPC_URL))

# 检查是否连上
print(w3.is_connected())  # 返回 True 就对了

你想想看，这一步其实就相当于你打开了一条通往区块链的“隧道”。后面所有的数据，都从这条隧道里流过来。

3.2 解析ERC20转账事件

ERC20 的转账事件，是链上最基础、也最重要的数据之一。每次有人转账 USDT、UNI 这些代币，都会触发 Transfer 事件。

事件的结构很简单：

• from：发送方地址
• to：接收方地址
• value：转账金额（注意：通常带小数位，比如 USDT 是 6 位小数）

怎么拿到这些事件？用 get_logs 方法。我举个例子，获取 USDT 最近 100 个区块内的转账记录：

USDT_ADDRESS = "0xdAC17F958D2ee523a2206206994597C13D831ec7"
TRANSFER_EVENT_SIGNATURE = w3.keccak(text="Transfer(address,address,uint256)").hex()

from_block = w3.eth.block_number - 100
to_block = w3.eth.block_number

logs = w3.eth.get_logs({
    "address": USDT_ADDRESS,
    "topics": [TRANSFER_EVENT_SIGNATURE],
    "fromBlock": from_block,
    "toBlock": to_block
})

for log in logs[:5]:  # 只打印前5条
    print(log)

这里有个坑，我曾经踩过：事件签名必须用 keccak256 哈希。如果你写错了字符串，比如少了个空格，就什么都查不到。我当时调试了整整一下午，最后发现是 Transfer(address,address,uint256) 里多了一个空格。嗯，细节决定成败。

拿到原始 log 后，还需要解码。用 web3.py 的 decode_log 方法可以搞定：

from web3._utils.events import get_event_abi

# 获取 USDT 合约的 ABI（这里简化了，实际需要完整 ABI）
usdt_contract = w3.eth.contract(address=USDT_ADDRESS, abi=USDT_ABI)
event_abi = get_event_abi(usdt_contract, "Transfer")

decoded_log = usdt_contract.events.Transfer().process_log(log)
print(decoded_log["args"])

输出结果类似：

AttributeDict({
    'from': '0x...',
    'to': '0x...',
    'value': 1000000000  # 这是 1000 USDT（因为 USDT 是 6 位小数）
})

你看，数据就出来了。说白了，这就是链上最原始的“转账流水”。

3.3 获取Uniswap V3池子的状态

Uniswap V3 的池子，比 V2 复杂一些。每个池子有多个“流动性区间”，但核心数据还是那几样：价格、流动性、手续费。

我以 ETH/USDC 的 0.3% 手续费池子为例。它的合约地址是 0x8ad599c3A0ff1De082011EFDDc00f3aF8c8f8b8f。

获取池子状态，主要调用这几个方法：

• slot0()：当前价格（sqrtPriceX96）、当前 tick、观察期等
• liquidity()：当前池子总流动性
• fee()：手续费率（比如 3000 代表 0.3%）

代码示例：

POOL_ADDRESS = "0x8ad599c3A0ff1De082011EFDDc00f3aF8c8f8b8f"
POOL_ABI = [...]  # 需要完整的 Uniswap V3 Pool ABI

pool_contract = w3.eth.contract(address=POOL_ADDRESS, abi=POOL_ABI)

# 获取 slot0
slot0 = pool_contract.functions.slot0().call()
print(f"当前 sqrtPriceX96: {slot0[0]}")
print(f"当前 tick: {slot0[1]}")

# 获取流动性
liquidity = pool_contract.functions.liquidity().call()
print(f"当前流动性: {liquidity}")

# 获取手续费
fee = pool_contract.functions.fee().call()
print(f"手续费率: {fee}")  # 3000 表示 0.3%

这里有个关键点：sqrtPriceX96 怎么转成真实价格？ 公式是：

price = (sqrtPriceX96 / 2**96) ** 2

但要注意，这个价格是 token1/token0 的比值。在 ETH/USDC 池子里，token0 是 USDC，token1 是 ETH。所以算出来的是 1 ETH 值多少 USDC。

我写个转换函数：

def sqrt_price_to_price(sqrt_price_x96, token0_decimals, token1_decimals):
    price = (sqrt_price_x96 / 2**96) ** 2
    # 调整小数位
    price = price * (10 ** (token0_decimals - token1_decimals))
    return price

# 假设 USDC 是 6 位小数，ETH 是 18 位小数
price = sqrt_price_to_price(slot0[0], 6, 18)
print(f"当前 ETH 价格: {price} USDC")

你想想看，这个价格就是链上实时的“预言机价格”。很多 DeFi 协议就是靠这个来做清算、做套利的。

3.4 实战中的避坑指南

最后，分享几个我踩过的坑：

• 节点请求超时：免费节点经常超时。我建议用 request_kwargs 设置超时时间：w3 = Web3(Web3.HTTPProvider(RPC_URL, request_kwargs={'timeout': 60}))
• 区块范围太大：一次查几千个区块的 log，节点会报错。我一般一次查 500-1000 个区块，分批处理。
• ABI 不完整：Uniswap V3 的 ABI 很长，但你可以只保留你需要的函数。我习惯从 Etherscan 上直接复制。
• 小数位搞错：USDT 是 6 位，UNI 是 18 位，搞错了金额就全乱了。我每次都会先查一下合约的 decimals() 方法。

好了，这一章的内容就到这里。你学会了怎么连节点、怎么拿转账事件、怎么读 Uniswap V3 池子的状态。这些是链上数据分析的“基本功”，后面我们会用这些数据做更复杂的分析。