3、数据获取：使用CCXT库获取交易所实时订单簿数据、WebSocket流式数据接入
做量化交易，数据就是你的眼睛。没有数据，再牛逼的策略也是瞎猜。
我个人习惯把数据获取分成两个层面：快照数据和流式数据。快照数据就是某一时刻的订单簿全貌，流式数据则是实时推送的增量变化。两者缺一不可。
3.1 为什么选择CCXT？
市面上有很多交易所的API，每个交易所的接口风格都不一样。你想想看，如果每个交易所都单独写一套对接代码，那维护成本得多高？
CCXT（CryptoCurrency eXchange Trading Library）就是来解决这个问题的。它封装了100多家交易所的API，提供统一的调用接口。说白了，你学会一个库，就能对接所有主流交易所。
我在项目中遇到过最头疼的事，就是交易所突然改API版本。用CCXT之后，这种痛苦大大减轻了——库的作者会帮你处理这些兼容性问题。
核心优势：
- 统一接口：同样的代码逻辑，切换交易所只需改一个参数
- 自动处理：签名、频率限制、错误重试都帮你封装好了
- 活跃社区：GitHub上1万+星，遇到问题基本都能搜到答案
3.2 安装与初始化
安装很简单，一行命令搞定：
pip install ccxt初始化交易所对象时，我建议把API密钥放在环境变量里，别硬编码在代码中。嗯，这里要注意：测试阶段可以用沙盒环境，别一上来就上真金白银。
import ccxt
import os
# 初始化交易所
exchange = ccxt.binance({
'apiKey': os.getenv('BINANCE_API_KEY'),
'secret': os.getenv('BINANCE_SECRET'),
'enableRateLimit': True, # 自动控制请求频率
'options': {
'defaultType': 'spot' # 现货市场
}
})
# 测试连接
markets = exchange.load_markets()
print(f"已加载 {len(markets)} 个交易对")
我的小技巧：设置
enableRateLimit=True 可以避免被交易所封IP。我曾经因为忘记设置这个，在回测时疯狂请求，结果被Binance封了半小时。
3.3 获取订单簿快照
订单簿数据是量化交易的核心输入。它包含了所有买单和卖单的深度信息。
def get_order_book(symbol='BTC/USDT', limit=100):
"""
获取订单簿数据
:param symbol: 交易对
:param limit: 深度级别，一般取10/50/100
"""
try:
orderbook = exchange.fetch_order_book(symbol, limit)
# 订单簿结构
bids = orderbook['bids'] # 买单 [[价格, 数量], ...]
asks = orderbook['asks'] # 卖单 [[价格, 数量], ...]
timestamp = orderbook['timestamp']
# 计算订单簿不平衡度
bid_volume = sum([b[1] for b in bids[:5]]) # 前5档买单总量
ask_volume = sum([a[1] for a in asks[:5]]) # 前5档卖单总量
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
return {
'symbol': symbol,
'timestamp': timestamp,
'bids': bids,
'asks': asks,
'imbalance': imbalance
}
except Exception as e:
print(f"获取订单簿失败: {e}")
return None
# 使用示例
data = get_order_book('BTC/USDT', 50)
if data:
print(f"BTC/USDT 订单簿不平衡度: {data['imbalance']:.4f}")这里有个关键点：limit参数。取10档还是100档？我个人经验是，做高频交易用5-10档就够了，做日内趋势可以取50-100档。取太多反而引入噪音。
3.4 WebSocket流式数据接入
REST API有个致命问题：每次请求都有延迟。对于订单簿不平衡度策略来说，毫秒级的延迟可能就意味着盈亏。
WebSocket是解决这个问题的标准方案。它建立长连接，交易所主动推送数据给你。说白了，就是「订阅-推送」模式。
import asyncio
import ccxt.pro as ccxtpro # CCXT的WebSocket版本
async def watch_order_book(symbol='BTC/USDT'):
"""
实时监控订单簿变化
"""
exchange = ccxtpro.binance({
'apiKey': os.getenv('BINANCE_API_KEY'),
'secret': os.getenv('BINANCE_SECRET'),
})
try:
while True:
# 等待订单簿更新
orderbook = await exchange.watch_order_book(symbol, limit=10)
# 计算实时不平衡度
bid_volume = sum([b[1] for b in orderbook['bids'][:5]])
ask_volume = sum([a[1] for a in orderbook['asks'][:5]])
imbalance = (bid_volume - ask_volume) / (bid_volume + ask_volume)
# 打印实时数据
print(f"[{orderbook['timestamp']}] 不平衡度: {imbalance:.4f}")
# 这里可以加入你的交易逻辑
if imbalance > 0.6:
print("⚠️ 买单严重失衡，可能上涨")
elif imbalance < -0.6:
print("⚠️ 卖单严重失衡，可能下跌")
except Exception as e:
print(f"WebSocket连接异常: {e}")
finally:
await exchange.close()
# 运行
asyncio.run(watch_order_book())
避坑指南：我曾经在生产环境中遇到WebSocket断连的问题。交易所的WebSocket服务偶尔会重启，你的代码必须要有重连机制。建议加上指数退避重试策略。
3.5 数据存储与缓存
实时数据来了之后，不能只放在内存里。万一程序崩溃，数据就全丢了。
我建议采用两级存储策略：
- 内存缓存：用Redis或Python的deque保存最近N条数据，用于实时计算
- 持久化存储：用InfluxDB或CSV文件保存历史数据，用于回测和分析
import csv
from collections import deque
from datetime import datetime
class OrderBookRecorder:
def __init__(self, max_records=1000):
self.buffer = deque(maxlen=max_records) # 内存缓存
self.csv_file = f"orderbook_{datetime.now().strftime('%Y%m%d')}.csv"
# 初始化CSV文件头
with open(self.csv_file, 'w', newline='') as f:
writer = csv.writer(f)
writer.writerow(['timestamp', 'symbol', 'bid_volume', 'ask_volume', 'imbalance'])
def record(self, data):
# 写入内存
self.buffer.append(data)
# 写入CSV
with open(self.csv_file, 'a', newline='') as f:
writer = csv.writer(f)
writer.writerow([
data['timestamp'],
data['symbol'],
sum([b[1] for b in data['bids'][:5]]),
sum([a[1] for a in data['asks'][:5]]),
data['imbalance']
])
def get_recent(self, n=10):
"""获取最近N条记录"""
return list(self.buffer)[-n:]3.6 本章知识体系
下面这张图展示了数据获取的整体架构，我建议你保存下来对照着看：
我的建议：刚开始做的时候，先用REST API跑通逻辑，再切换到WebSocket。别一上来就搞流式，调试起来很麻烦。等策略稳定了，再追求低延迟。
数据获取是整个量化交易系统的地基。地基不稳，上层建筑再漂亮也是白搭。我个人习惯在数据层多花30%的时间做测试和容错，这会在后续开发中省下大量调试时间。