3、数据获取：如何使用CCXT库获取交易所实时行情与历史数据
做量化交易，数据就是你的弹药。没有数据，策略再好也是纸上谈兵。
我个人习惯用CCXT这个库。它统一了上百家交易所的API接口。你写一套代码，就能对接币安、OKX、Bybit等主流平台。省心。
3.1 CCXT是什么？为什么选它？
CCXT是一个开源的JavaScript/Python库。它把各家交易所的API封装成了统一的方法。
说白了，你不需要去读币安的API文档，也不用管OKX的签名算法。CCXT帮你搞定了这些脏活累活。
核心优势：
- 支持120+交易所
- 统一接口：getTicker、fetchOHLCV、createOrder
- 自动处理限频、重连、签名
- 社区活跃，更新及时
我在项目中遇到过一个问题：某交易所突然改了API版本，导致所有订单程序报错。还好CCXT在两天内就发布了适配版本。要是自己手写API调用，那得改到崩溃。
3.2 安装与初始化
安装很简单，一行命令搞定：
pip install ccxt然后初始化交易所对象。以币安为例：
import ccxt
# 创建交易所对象
exchange = ccxt.binance({
'apiKey': 'YOUR_API_KEY',
'secret': 'YOUR_SECRET',
'enableRateLimit': True, # 自动限频，防止被封
'options': {
'defaultType': 'spot', # 现货交易
}
})
# 打印支持的交易对
markets = exchange.load_markets()
print(f"币安支持 {len(markets)} 个交易对")嗯，这里要注意：enableRateLimit一定要设为True。我曾经忘了开这个，结果程序跑了10分钟就被交易所封了IP。惨痛教训。
3.3 获取实时行情
实时行情是策略的"眼睛"。CCXT提供了几种常用接口：
3.3.1 Ticker（最新行情）
# 获取BTC/USDT的最新行情
ticker = exchange.fetch_ticker('BTC/USDT')
print(f"最新价: {ticker['last']}")
print(f"24h涨跌: {ticker['percentage']}%")
print(f"24h成交量: {ticker['baseVolume']} BTC")返回的数据结构很清晰。你直接拿ticker['last']就能用。不用自己拼JSON。
3.3.2 Order Book（深度数据）
# 获取买卖盘口
orderbook = exchange.fetch_order_book('BTC/USDT', limit=10)
print(f"买一价: {orderbook['bids'][0][0]}, 数量: {orderbook['bids'][0][1]}")
print(f"卖一价: {orderbook['asks'][0][0]}, 数量: {orderbook['asks'][0][1]}")做套利策略时，深度数据特别重要。你想想看，如果买卖价差太小，套利空间就没了。我一般会看前10档的深度，判断流动性是否充足。
3.3.3 Trades（实时成交）
# 获取最近成交记录
trades = exchange.fetch_trades('BTC/USDT', limit=50)
for trade in trades[:5]:
print(f"时间: {trade['datetime']}, 价格: {trade['price']}, 数量: {trade['amount']}")小技巧：如果你需要实时推送，可以用WebSocket。CCXT的Pro版本支持。但回测阶段用REST接口就够了，别过度设计。
3.4 获取历史K线数据
回测的核心就是历史数据。CCXT用fetchOHLCV方法获取K线：
# 获取BTC/USDT的1小时K线，最近100根
ohlcv = exchange.fetch_ohlcv('BTC/USDT', timeframe='1h', limit=100)
# 转换成DataFrame方便分析
import pandas as pd
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
print(df.head())返回的数据格式是固定的：[时间戳, 开盘价, 最高价, 最低价, 收盘价, 成交量]。你直接塞进pandas就能用。
3.4.1 支持的时间周期
| 参数值 | 含义 | 常见用途 |
|---|---|---|
| 1m | 1分钟 | 高频策略 |
| 5m | 5分钟 | 短线交易 |
| 1h | 1小时 | 日内策略 |
| 1d | 1天 | 趋势跟踪 |
3.4.2 获取更早的历史数据
交易所API一般只返回最近500-1000根K线。要获取更早的数据，需要分页：
def fetch_historical_ohlcv(exchange, symbol, timeframe, since):
"""获取历史K线，支持分页"""
all_candles = []
while True:
candles = exchange.fetch_ohlcv(symbol, timeframe, since=since, limit=1000)
if len(candles) == 0:
break
all_candles.extend(candles)
since = candles[-1][0] + 1 # 从最后一根K线的时间戳继续
print(f"已获取 {len(all_candles)} 根K线...")
return all_candles
# 获取2023年1月1日以来的1小时K线
since = exchange.parse8601('2023-01-01T00:00:00Z')
data = fetch_historical_ohlcv(exchange, 'BTC/USDT', '1h', since)注意：分页获取时要注意限频。我建议每次请求间隔200ms以上。另外，有些交易所对历史数据的深度有限制，比如币安最多只能拉取500天前的数据。
3.5 数据清洗与存储
拿到的原始数据不能直接用。你想想看，交易所偶尔会返回空值或者异常值。我一般会做这几步处理：
def clean_ohlcv_data(df):
"""清洗K线数据"""
# 1. 删除空值
df = df.dropna()
# 2. 检查时间戳是否连续
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.set_index('timestamp')
df = df.asfreq('1H') # 按小时对齐
# 3. 填充缺失值（用前值填充）
df = df.fillna(method='ffill')
# 4. 删除异常值（比如价格为0）
df = df[df['close'] > 0]
return df
# 存储到CSV
df.to_csv('btc_usdt_1h.csv', index=False)
print("数据已保存")我的习惯：把清洗后的数据存到本地CSV文件。回测时直接读文件，不用每次都去拉交易所数据。既快又省流量。
3.6 核心逻辑流程图
下面这张图展示了数据获取的完整流程：
这个流程看着简单，但每一步都有坑。比如初始化时忘了设限频，或者分页时没处理好时间戳。我刚开始做的时候，光调试数据获取就花了两天。
3.7 常见问题与避坑
- 限频问题：每个交易所的限频规则不同。币安是1200次/分钟，OKX是600次/分钟。我建议统一设成200ms间隔，安全。
- 数据缺失：交易所偶尔会返回空数据。比如某个交易对刚上线，历史数据不全。记得加异常处理。
- 时间戳对齐：不同交易所的时间戳可能有几秒偏差。做套利时，时间同步很重要。我一般会用NTP校准本地时间。
避坑指南：我曾经在回测时发现策略收益特别高，兴奋了半天。后来才发现是数据没清洗，包含了未来数据。从那以后，我每次回测前都会检查数据的时间顺序，确保没有"穿越"。
好了，数据获取这块就讲到这里。CCXT用起来其实不复杂，关键是处理好细节。你动手试试，拉点数据下来看看，比光看文档强多了。
公众号:蓝海资料掘金营，微信deep3321