Python脚本如何操作区块链数据

wen 实用脚本 2026-07-10 1

Python脚本实战指南

目录导读

为什么用Python操作区块链数据？
核心准备：环境搭建与依赖库
连接区块链节点：从RPC到Web3
读取链上数据：区块、交易与余额
写入数据：签名、发送交易与智能合约交互
数据解析：过滤、存储与分析技巧
常见问题问答（Q&A）

为什么用Python操作区块链数据？

区块链本质是一个分布式账本,其数据公开、不可篡改、可追溯，Python凭借丰富的库生态（如Web3.py、requests、pandas）和简洁语法，成为操作区块链数据的首选语言。

Python脚本如何操作区块链数据

核心优势：

通过RPC接口直接与节点通信,无需运行完整节点
支持主要公链（以太坊、比特币、BSC、Polygon等）
可快速编写自动化脚本进行数据抓取、监控和交易

适用场景：

查询钱包余额
监听链上事件（如NFT铸造、大额转账）
批量发送加密货币
跨链数据对比

核心准备：环境搭建与依赖库

安装基础库

pip install web3 requests pandas json5

获取节点访问端点

公共RPC节点：Binance（https://bsc-dataseed.binance.org/）、以太坊（https://mainnet.infura.io/...）
自行搭建节点：Geth、OpenEthereum（需同步完整数据）

注意：公共节点有速率限制，生产环境建议使用私有RPC或付费节点服务（如Infura、Alchemy）。

连接区块链节点：从RPC到Web3

基础连接（以太坊为例）

from web3 import Web3
# 连接到以太坊主网
rpc_url = "https://mainnet.infura.io/v3/YOUR_API_KEY"
w3 = Web3(Web3.HTTPProvider(rpc_url))
if w3.is_connected():
    print("✅ 连接成功，当前区块高度:", w3.eth.block_number)
else:
    print("❌ 连接失败")

连接其他链（BSC、Polygon）

仅需更换RPC URL和链ID：

bsc_rpc = "https://bsc-dataseed.binance.org/"
w3_bsc = Web3(Web3.HTTPProvider(bsc_rpc))

提示：每次请求会消耗RPC配额，建议使用w3.middleware_onion添加缓存中间件。

读取链上数据：区块、交易与余额

查询账户余额

address = "0x742d35Cc6634C0532925a3b844Bc454e4438f44e"
balance_wei = w3.eth.get_balance(address)
print(f"余额: {w3.from_wei(balance_wei, 'ether')} ETH")

获取历史交易

tx_hash = "0x88df01642990d1a6f7c2e5a1e1b6c9f7e5a1e1b6c9f7e5a1e1b6c9f7e5a1e1"
tx = w3.eth.get_transaction(tx_hash)
print(f"发送方: {tx['from']}, 接收方: {tx['to']}, 金额: {w3.from_wei(tx['value'], 'ether')} ETH")

解析区块内所有交易

block = w3.eth.get_block(18000000, full_transactions=True)
for tx in block['transactions']:
    print(f"Tx: {tx['hash'].hex()} | 金额: {w3.from_wei(tx['value'], 'ether')}")

写入数据：签名、发送交易与智能合约交互

发送ETH转账（需私钥）

from web3.middleware import geth_poa_middleware
# 适用PoA链（BSC、Polygon）
w3.middleware_onion.inject(geth_poa_middleware, layer=0)
private_key = "0xYOUR_PRIVATE_KEY"
account = w3.eth.account.from_key(private_key)
nonce = w3.eth.get_transaction_count(account.address)
tx = {
    'nonce': nonce,
    'to': '0x接收方地址',
    'value': w3.to_wei(0.01, 'ether'),
    'gas': 21000,
    'gasPrice': w3.eth.gas_price
}
signed_tx = w3.eth.account.sign_transaction(tx, private_key)
tx_hash = w3.eth.send_raw_transaction(signed_tx.rawTransaction)
print(f"交易已发送: {tx_hash.hex()}")

调用智能合约（ERC20转账示例）

# 合约ABI（简化版）
abi = [{"constant":False,"inputs":[{"name":"_to","type":"address"},{"name":"_value","type":"uint256"}],"name":"transfer","outputs":[{"name":"","type":"bool"}],"type":"function"}]
contract = w3.eth.contract(address='0x合约地址', abi=abi)
tx = contract.functions.transfer('0x接收方', w3.to_wei(100, 'ether')).build_transaction({
    'from': account.address,
    'nonce': nonce,
    'gas': 60000,
    'gasPrice': w3.eth.gas_price
})
signed_tx = w3.eth.account.sign_transaction(tx, private_key)
w3.eth.send_raw_transaction(signed_tx.rawTransaction)

安全警告：私钥绝不能硬编码在脚本中！建议使用环境变量或硬件钱包签名。

数据解析：过滤、存储与分析技巧

实时监听链上事件

event_filter = w3.eth.filter({'address': '0x合约地址', 'topics': ['0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef']})  
# Transfer事件签名哈希
while True:
    for event in w3.eth.get_filter_changes(event_filter.filter_id):
        print(f"转账事件: {event}")
    time.sleep(2)

数据存储方案

CSV文件：适合小规模记录
SQLite/PostgreSQL：适合历史数据分析
数据库索引优化：对区块号、地址字段建索引

批量数据提取（使用multiprocessing）

from multiprocessing import Pool
def get_balance_in_batch(address_list):
    with Pool(10) as p:
        results = p.map(w3.eth.get_balance, address_list)
    return results

常见问题问答（Q&A）

Q1：为什么使用公共RPC节点时频繁报错“429 Too Many Requests”？

原因：公共节点有请求速率限制（通常每秒5-15次）。
解决：

添加time.sleep(0.2)在循环中
使用w3.middleware_onion.add(web3.middleware.SimpleCacheMiddleware)
升级到付费节点（如Infura Growth计划，月费约$50）

Q2：如何获取特定代币（ERC20）在某个地址上的余额？

def get_token_balance(token_contract, wallet_address):
    contract = w3.eth.contract(address=token_contract, abi=ERC20_ABI)
    balance = contract.functions.balanceOf(wallet_address).call()
    decimals = contract.functions.decimals().call()
    return balance / (10 ** decimals)

Q3：智能合约调用时报错“execution reverted”是什么原因？

排查步骤：

检查燃气限制（gas）是否充足（建议加20%缓冲）
确认地址和ABI完全匹配
调用w3.eth.call()模拟执行（不发送真实交易）
使用Etherscan的“Read Contract”验证参数

Q4：Python脚本能否同时监控多条链？

实现方案：

使用多线程（threading）或异步（asyncio）
每条链创建独立的Web3实例
注意：不要在不同线程共享同一Web3对象

import asyncio
import aiohttp
async def monitor_chain(name, rpc):
    w3 = Web3(Web3.AsyncHTTPProvider(rpc))
    async with w3:
        block = await w3.eth.get_block('latest')
        print(f"{name}最新区块: {block['number']}")

Q5：如何确保脚本长期稳定运行？

最佳实践：

添加连接重试机制（最多3次，间隔2秒）
使用系统服务（systemd）管理Python进程
记录错误日志到文件（logging模块）
设置异常触发告警（如Telegram Bot通知）

通过Python操作区块链数据,你不仅能高效获取链上信息，还能实现自动化交易、实时监控和数据可视化，从基础的余额查询到复杂的智能合约交互，本文提供了一套完整的实战路径，建议初学者先利用公共RPC测试网（如以太坊Sepolia、BSC Testnet）熟悉流程，再逐步转入主网。

注意：操作真实资产前，务必在测试环境中验证全部逻辑；私钥存储采用硬件钱包或多签方案，避免资产损失。