2. 链上数据获取：RPC节点交互、Web3.py库基础、Etherscan API使用

做链上监控，第一步就是拿到数据。这听起来简单，但实际坑不少。我刚开始做安全分析时，以为随便找个公共RPC就能搞定，结果被限流限到怀疑人生。今天咱们就把数据获取这件事彻底聊透。

2.1 RPC节点：你的数据水龙头

RPC（Remote Procedure Call）节点，说白了就是你和区块链之间的桥梁。你发个请求过去，它把链上数据返回给你。没有RPC，你连一个区块都读不到。

公共RPC vs 私有RPC

公共RPC免费但有限制。我记得有一次监控一个热门NFT项目，用公共节点每秒请求超过20次，直接被封了IP。嗯，那次教训挺深刻的。

类型	优点	缺点	适用场景
公共RPC	免费、无需配置	限流严重、不稳定	测试、低频查询
私有RPC	高并发、稳定、专属	需要付费	生产环境、高频监控

我个人习惯用Infura或Alchemy的免费层做开发测试，上线后换成付费节点。你想想看，监控系统如果因为RPC限流而漏掉一笔异常交易，那损失可不是一个节点钱能比的。

常用RPC端点

• Infura：https://mainnet.infura.io/v3/YOUR_PROJECT_ID
• Alchemy：https://eth-mainnet.g.alchemy.com/v2/YOUR_API_KEY
• QuickNode：按需付费，支持多链
• 本地节点：http://localhost:8545（自己跑Geth）

2.2 Web3.py库：Python与链的桥梁

Web3.py是Python生态中最流行的以太坊交互库。说白了，它就是帮你把RPC请求封装成好用的Python函数。你不需要手动拼JSON-RPC了。

安装与连接

pip install web3

连接节点很简单，但要注意选择正确的Provider：

from web3 import Web3

# 连接公共节点
w3 = Web3(Web3.HTTPProvider('https://mainnet.infura.io/v3/YOUR_PROJECT_ID'))

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

# 用WebSocket（推荐用于监控）
w3 = Web3(Web3.WebsocketProvider('wss://mainnet.infura.io/ws/v3/YOUR_PROJECT_ID'))

核心操作：读数据

读数据不需要Gas，也不需要私钥。这是我最常用的几个操作：

# 获取最新区块号
latest_block = w3.eth.block_number
print(f"当前区块: {latest_block}")

# 获取区块详情
block = w3.eth.get_block(latest_block, full_transactions=True)
print(f"区块时间戳: {block['timestamp']}")
print(f"交易数量: {len(block['transactions'])}")

# 获取交易详情
tx_hash = '0x...'  # 替换成实际交易哈希
tx = w3.eth.get_transaction(tx_hash)
print(f"发送方: {tx['from']}")
print(f"接收方: {tx['to']}")
print(f"金额: {w3.from_wei(tx['value'], 'ether')} ETH")

与智能合约交互

监控异常交易，经常需要读合约状态。比如查一个地址的USDT余额：

# USDT合约地址
usdt_address = '0xdAC17F958D2ee523a2206206994597C13D831ec7'

# 合约ABI（简化版，实际需要完整ABI）
usdt_abi = '[{"constant":true,"inputs":[{"name":"_owner","type":"address"}],"name":"balanceOf","outputs":[{"name":"balance","type":"uint256"}],"type":"function"}]'

# 创建合约对象
usdt_contract = w3.eth.contract(address=usdt_address, abi=usdt_abi)

# 查询余额
balance = usdt_contract.functions.balanceOf('0x...').call()
print(f"USDT余额: {balance / 10**6}")

2.3 Etherscan API：链上数据的搜索引擎

Etherscan API相当于给区块链加了个搜索引擎。RPC能拿到原始数据，但Etherscan能给你整理好的数据——比如某个地址的所有交易历史、代币转账记录、内部交易等。

申请API Key

去Etherscan官网注册，创建一个API Key。免费版每秒5次请求，够个人用了。但如果你要跑生产监控，建议付费升级。

常用API端点

功能	API端点	参数示例
获取ETH交易	api.etherscan.io/api?module=account&action=txlist	address, startblock, endblock, sort
获取代币转账	api.etherscan.io/api?module=account&action=tokentx	address, contractaddress, page, offset
获取内部交易	api.etherscan.io/api?module=account&action=txlistinternal	txhash, address
获取合约ABI	api.etherscan.io/api?module=contract&action=getabi	address

Python调用示例

import requests

API_KEY = 'YOUR_ETHERSCAN_API_KEY'
BASE_URL = 'https://api.etherscan.io/api'

def get_normal_txs(address, start_block=0, end_block=99999999):
    """获取地址的普通交易"""
    params = {
        'module': 'account',
        'action': 'txlist',
        'address': address,
        'startblock': start_block,
        'endblock': end_block,
        'sort': 'desc',
        'apikey': API_KEY
    }
    response = requests.get(BASE_URL, params=params)
    data = response.json()
    
    if data['status'] == '1':
        return data['result']
    else:
        print(f"错误: {data['message']}")
        return []

# 使用示例
address = '0x...'
txs = get_normal_txs(address)
for tx in txs[:5]:  # 只看前5笔
    print(f"哈希: {tx['hash']}, 金额: {int(tx['value'])/1e18} ETH")

2.4 知识体系总览

下面这张图帮你理清这三者的关系：

2.5 实战：搭建一个简单的数据监控脚本

把上面学的东西串起来，写一个监控新交易的小脚本：

from web3 import Web3
import time

# 连接节点
w3 = Web3(Web3.WebsocketProvider('wss://mainnet.infura.io/ws/v3/YOUR_KEY'))

# 要监控的地址
watch_addresses = ['0x...', '0x...']

def handle_event(event):
    """处理新交易事件"""
    tx_hash = event['params']['result']
    tx = w3.eth.get_transaction(tx_hash)
    
    # 检查是否涉及监控地址
    if tx['from'] in watch_addresses or tx['to'] in watch_addresses:
        print(f"发现异常交易: {tx_hash.hex()}")
        print(f"金额: {w3.from_wei(tx['value'], 'ether')} ETH")
        # 这里可以加告警逻辑

# 订阅待处理交易
event_filter = w3.eth.filter('pending')
print("开始监控...")

while True:
    try:
        for event in w3.eth.get_filter_changes(event_filter.filter_id):
            handle_event(event)
        time.sleep(0.1)
    except Exception as e:
        print(f"错误: {e}")
        time.sleep(5)

好了，数据获取这块就聊到这儿。记住一个原则：实时监控用RPC+WebSocket，历史分析用Etherscan API。两者配合，基本能覆盖所有链上数据需求。

---