Python SDK
Zanbara Python SDK 为 Python 应用提供完整的 REST API 和 WebSocket 客户端封装,支持异步和同步两种模式。
安装
pip install zanbara-sdk快速开始
同步模式
from zanbara import ZanbaraClient
# 创建客户端
client = ZanbaraClient('https://api.zanbarax.com/v1')
# 获取订单簿
orderbook = client.get_orderbook('BTC-PERP', depth=20)
print(f"最优买价: {orderbook['bids'][0][0]}")
print(f"最优卖价: {orderbook['asks'][0][0]}")
# 获取市场统计
stats = client.get_market_stats('BTC-PERP')
print(f"最新价: {stats['lastPrice']}")
print(f"24h涨跌: {stats['priceChangePercent24h']}%")异步模式
import asyncio
from zanbara import AsyncZanbaraClient
async def main():
# 创建异步客户端
client = AsyncZanbaraClient('https://api.zanbarax.com/v1')
# 获取订单簿
orderbook = await client.get_orderbook('BTC-PERP', depth=20)
print(f"订单簿: {orderbook}")
# 获取K线数据
klines = await client.get_klines('BTC-PERP', '1h', limit=100)
print(f"获取了 {len(klines)} 根K线")
asyncio.run(main())REST API 客户端
初始化
from zanbara import ZanbaraClient
# 基础配置
client = ZanbaraClient(
base_url='https://api.zanbarax.com/v1',
timeout=10,
api_key='YOUR_API_KEY' # 可选,用于私有接口
)市场数据
# 获取所有市场
markets = client.get_markets()
for market in markets:
print(f"{market['symbol']}: {market['baseCurrency']}/{market['quoteCurrency']}")
# 获取订单簿
orderbook = client.get_orderbook('BTC-PERP', depth=20)
# 获取最新成交
trades = client.get_trades('BTC-PERP', limit=50)
# 获取K线数据
klines = client.get_klines(
market='BTC-PERP',
interval='1h',
start_time=int((datetime.now() - timedelta(days=1)).timestamp() * 1000),
end_time=int(datetime.now().timestamp() * 1000),
limit=24
)
# 获取24h统计
stats = client.get_market_stats('BTC-PERP')账户操作
# 设置 API Key
client.set_api_key('YOUR_API_KEY')
# 获取持仓
positions = client.get_positions('YOUR_WALLET_ADDRESS')
for pos in positions:
print(f"{pos['market']}: {pos['side']} {pos['size']}")
# 获取订单
orders = client.get_orders('YOUR_WALLET_ADDRESS', status='open')
# 获取成交历史
fills = client.get_fills(
'YOUR_WALLET_ADDRESS',
market='BTC-PERP',
limit=50
)高级订单
# 创建 Iceberg 订单
iceberg_order = client.create_iceberg_order(
symbol='BTC-PERP',
side='buy',
total_quantity='5.0',
visible_quantity='0.5',
price='43000.0'
)
print(f"Iceberg 订单 ID: {iceberg_order['icebergOrderId']}")
# 创建 TWAP 订单
twap_order = client.create_twap_order(
symbol='ETH-PERP',
side='sell',
total_quantity='10.0',
duration_seconds=3600,
num_slices=10,
randomize_intervals=True
)
print(f"TWAP 订单 ID: {twap_order['twapOrderId']}")
# 获取执行统计
iceberg_stats = client.get_iceberg_stats('iceberg_abc123')
print(f"执行进度: {iceberg_stats['executedQuantity']}/{iceberg_stats['totalQuantity']}")
twap_stats = client.get_twap_stats('twap_xyz789')
print(f"平均成交价: {twap_stats['averageExecutionPrice']}")
print(f"总滑点: {twap_stats['totalSlippage']}")
# 暂停/恢复/取消
client.pause_iceberg_order('iceberg_abc123')
client.resume_iceberg_order('iceberg_abc123')
client.cancel_iceberg_order('iceberg_abc123')WebSocket 客户端
基础使用
import asyncio
from zanbara import ZanbaraWebSocket
async def on_ticker(data):
print(f"Ticker: {data['market']} ${data['lastPrice']}")
async def main():
ws = ZanbaraWebSocket('wss://ws.zanbarax.com')
# 连接
await ws.connect()
# 订阅 Ticker
await ws.subscribe_ticker('BTC-PERP', on_ticker)
# 保持运行
await ws.run_forever()
asyncio.run(main())订阅多个频道
import asyncio
from zanbara import ZanbaraWebSocket
class MarketMonitor:
def __init__(self):
self.ws = ZanbaraWebSocket('wss://ws.zanbarax.com')
async def on_orderbook(self, data):
print(f"订单簿更新: {data['market']}")
async def on_trades(self, data):
print(f"新成交: {data['market']} {data['side']} {data['quantity']}@{data['price']}")
async def on_ticker(self, data):
print(f"Ticker: {data['market']} {data['lastPrice']} ({data['priceChangePercent24h']}%)")
async def start(self, markets):
await self.ws.connect()
for market in markets:
await self.ws.subscribe_orderbook(market, self.on_orderbook)
await self.ws.subscribe_trades(market, self.on_trades)
await self.ws.subscribe_ticker(market, self.on_ticker)
await self.ws.run_forever()
# 使用
async def main():
monitor = MarketMonitor()
await monitor.start(['BTC-PERP', 'ETH-PERP', 'SOL-PERP'])
asyncio.run(main())认证连接 (私有频道)
import asyncio
from zanbara import ZanbaraClient, ZanbaraWebSocket
async def main():
# 获取 WebSocket token
client = ZanbaraClient('https://api.zanbarax.com/v1', api_key='YOUR_API_KEY')
token = client.get_websocket_token()
# 使用 token 连接
ws = ZanbaraWebSocket(f'wss://ws.zanbarax.com?token={token}')
await ws.connect()
# 订阅私有频道
await ws.subscribe_orders('YOUR_WALLET_ADDRESS', on_order_update)
await ws.subscribe_positions('YOUR_WALLET_ADDRESS', on_position_update)
await ws.subscribe_account(on_account_update)
await ws.run_forever()
async def on_order_update(data):
print(f"订单更新: {data['orderId']} {data['status']}")
async def on_position_update(data):
print(f"持仓更新: {data['market']} {data['unrealizedPnl']}")
async def on_account_update(data):
print(f"账户更新: {data['totalEquity']}")
asyncio.run(main())Pandas 集成
SDK 提供 Pandas DataFrame 集成,方便数据分析:
import pandas as pd
from zanbara import ZanbaraClient
client = ZanbaraClient('https://api.zanbarax.com/v1')
# 获取K线数据并转换为 DataFrame
klines = client.get_klines('BTC-PERP', '1h', limit=100)
df = pd.DataFrame(klines)
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
# 转换数据类型
for col in ['open', 'high', 'low', 'close', 'volume']:
df[col] = df[col].astype(float)
# 计算技术指标
df['sma_20'] = df['close'].rolling(window=20).mean()
df['ema_12'] = df['close'].ewm(span=12).mean()
df['ema_26'] = df['close'].ewm(span=26).mean()
df['macd'] = df['ema_12'] - df['ema_26']
# 计算 RSI
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['rsi'] = 100 - (100 / (1 + rs))
# 打印统计
print(df.tail())
print(f"\n统计信息:")
print(f"最高价: {df['high'].max()}")
print(f"最低价: {df['low'].min()}")
print(f"平均成交量: {df['volume'].mean():.2f}")异步批量操作
import asyncio
from zanbara import AsyncZanbaraClient
async def fetch_multiple_markets(client, markets):
"""并发获取多个市场数据"""
tasks = [client.get_market_stats(market) for market in markets]
results = await asyncio.gather(*tasks)
return results
async def main():
client = AsyncZanbaraClient('https://api.zanbarax.com/v1')
markets = ['BTC-PERP', 'ETH-PERP', 'SOL-PERP', 'ARB-PERP']
# 并发获取
stats_list = await fetch_multiple_markets(client, markets)
for stats in stats_list:
print(f"{stats['market']}: ${stats['lastPrice']} ({stats['priceChangePercent24h']}%)")
asyncio.run(main())错误处理
from zanbara import ZanbaraClient, ZanbaraError, ApiError, NetworkError
client = ZanbaraClient('https://api.zanbarax.com/v1')
try:
orderbook = client.get_orderbook('BTC-PERP')
except ApiError as e:
print(f"API 错误: {e.message}")
print(f"错误码: {e.code}")
print(f"状态码: {e.status_code}")
except NetworkError as e:
print(f"网络错误: {e}")
except ZanbaraError as e:
print(f"SDK 错误: {e}")完整示例
import asyncio
import pandas as pd
from datetime import datetime, timedelta
from zanbara import AsyncZanbaraClient, ZanbaraWebSocket
class TradingBot:
def __init__(self, api_url, ws_url):
self.api_client = AsyncZanbaraClient(api_url)
self.ws_client = ZanbaraWebSocket(ws_url)
self.current_price = {}
async def initialize(self):
"""初始化"""
# 连接 WebSocket
await self.ws_client.connect()
print("✓ WebSocket 已连接")
# 加载历史数据
await self.load_historical_data()
async def load_historical_data(self):
"""加载历史数据"""
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=1)).timestamp() * 1000)
klines = await self.api_client.get_klines(
'BTC-PERP',
'1h',
start_time=start_time,
end_time=end_time
)
df = pd.DataFrame(klines)
print(f"✓ 加载了 {len(df)} 根K线")
async def on_ticker(self, data):
"""处理 Ticker 更新"""
market = data['market']
price = float(data['lastPrice'])
self.current_price[market] = price
print(f"{market}: ${price}")
async def run(self, markets):
"""运行机器人"""
await self.initialize()
# 订阅市场数据
for market in markets:
await self.ws_client.subscribe_ticker(market, self.on_ticker)
print("✓ 交易机器人已启动")
await self.ws_client.run_forever()
# 运行
async def main():
bot = TradingBot(
'https://api.zanbarax.com/v1',
'wss://ws.zanbarax.com'
)
await bot.run(['BTC-PERP', 'ETH-PERP'])
asyncio.run(main())下一步
Rust SDK - Rust SDK 文档
SDK 示例 - 更多实战示例
TypeScript SDK - TypeScript SDK 文档
Last updated