【Web3初识系列】如何连接 Binance Smart Chain通过交易对绘制 k 线？

连接 Binance Smart Chain通过交易对绘制 k 线

安装 web3

pip install web3

连接到 Binance Smart Chain

使用公共的 BSC 节点 URL。

from web3 import Web3# 连接到 BSC 公共节点
bsc_url = "https://bsc-dataseed.binance.org/"
web3 = Web3(Web3.HTTPProvider(bsc_url))# 检查连接是否成功
if web3.is_connected():print("Successfully connected to Binance Smart Chain")
else:print("Failed to connect to Binance Smart Chain")# 获取链上信息，例如当前区块号
block_number = web3.eth.block_number
print(f"Current block number: {block_number}")

获取代币的合约实例

需要代币合约的 ABI 和 地址。

# PPIG 代币合约的 ABI 
token_abi = '''
[{"constant": true,"inputs": [],"name": "name","outputs": [{"name": "", "type": "string"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "symbol","outputs": [{"name": "", "type": "string"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "decimals","outputs": [{"name": "", "type": "uint8"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [{"name": "_owner", "type": "address"}],"name": "balanceOf","outputs": [{"name": "balance", "type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"},{"constant": true,"inputs": [],"name": "totalSupply","outputs": [{"name": "", "type": "uint256"}],"payable": false,"stateMutability": "view","type": "function"}
]
'''# 创建代币合约实例
token_contract = web3.eth.contract(address=token_address, abi=token_abi)# 获取代币信息
token_name = token_contract.functions.name().call()
token_symbol = token_contract.functions.symbol().call()
token_decimals = token_contract.functions.decimals().call()
token_total_supply = token_contract.functions.totalSupply().call()print(f"Token Name: {token_name}")
print(f"Token Symbol: {token_symbol}")
print(f"Token Decimals: {token_decimals}")
print(f"Token Total Supply: {web3.from_wei(token_total_supply, 'ether')} {token_symbol}")

监听特定事件

监听 PPIG 交易对的 Swap 事件。

# 添加 Geth POA 中间件（如果使用的是 PoA 网络）
web3.middleware_onion.inject(geth_poa_middleware, layer=0)# PPIG/WBNB 交易对合约的 ABI
pair_abi = '''
[{"anonymous": false,"inputs": [{"indexed": true, "name": "sender", "type": "address"},{"indexed": false, "name": "amount0In", "type": "uint256"},{"indexed": false, "name": "amount1In", "type": "uint256"},{"indexed": false, "name": "amount0Out", "type": "uint256"},{"indexed": false, "name": "amount1Out", "type": "uint256"},{"indexed": true, "name": "to", "type": "address"}],"name": "Swap","type": "event"}
]
'''# 创建交易对合约实例
pair_contract = web3.eth.contract(address=pair_address, abi=pair_abi)# 创建 Swap 事件过滤器
swap_event_filter = pair_contract.events.Swap.create_filter(fromBlock='latest')# 处理事件的回调函数
def handle_event(event):print(event)# 提取事件数据timestamp = web3.eth.get_block(event['blockNumber'])['timestamp']dt_object = datetime.fromtimestamp(timestamp)amount0In = web3.from_wei(event['args']['amount0In'], 'ether')amount1In = web3.from_wei(event['args']['amount1In'], 'ether')amount0Out = web3.from_wei(event['args']['amount0Out'], 'ether')amount1Out = web3.from_wei(event['args']['amount1Out'], 'ether')# 根据具体需求处理数据，这里只是打印出来print(f"Time: {dt_object}, amount0In: {amount0In}, amount1In: {amount1In}, amount0Out: {amount0Out}, amount1Out: {amount1Out}")# 轮询新事件
def log_loop(event_filter, poll_interval):while True:for event in event_filter.get_new_entries():handle_event(event)time.sleep(poll_interval)

生成 K 线图

收集一段时间后，我们可以使用 Pandas 和 Matplotlib 生成 K 线图。

import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetimedata = {'timestamp': [datetime(2024, 6, 19, 12, 0), datetime(2024, 6, 19, 12, 5), datetime(2024, 6, 19, 12, 10)],'price': [10, 12, 11],'volume': [100, 150, 120]
}df = pd.DataFrame(data)# 设置时间为索引
df.set_index('timestamp', inplace=True)# 生成 K 线图
plt.figure(figsize=(10, 5))
plt.plot(df.index, df['price'], marker='o')
plt.title('PPIG/WBNB Price')
plt.xlabel('Time')
plt.ylabel('Price')
plt.grid()
plt.show()

注意事项

生成k线时，使用到了模拟数据，在实际的环境中注意换成实时数据包装。