数据获取入门：用CCXT连接交易所

做价差捕捉，第一步就是拿到数据。

没有数据，什么策略都是空谈。我个人习惯把数据获取比作「做饭前的买菜」——菜买不好，厨艺再高也白搭。

为什么选CCXT？

市面上有很多交易所API库，但CCXT是我用得最顺手的。它统一了上百家交易所的接口，你写一套代码，就能对接币安、OKX、Bybit等等。

说白了，它就是个「翻译官」。把各家交易所不同的API格式，翻译成统一的Python调用方式。

安装与初始化

安装很简单，一行命令搞定：

pip install ccxt

然后初始化交易所对象。我以币安为例：

import ccxt

# 创建交易所实例
exchange = ccxt.binance({
    'apiKey': '你的API_KEY',
    'secret': '你的SECRET_KEY',
    'enableRateLimit': True,  # 自动限流，防止被封
    'options': {
        'defaultType': 'spot',  # 现货市场
    }
})

# 测试连接
markets = exchange.load_markets()
print(f"成功连接，共加载 {len(markets)} 个交易对")

获取实时Ticker数据

Ticker就是当前市场的快照。包含最新价、买一卖一价、24小时成交量等。

我经常用它来判断「当前市场是否活跃」。比如价差策略需要高流动性，如果某个币的24小时成交量太小，我一般会直接跳过。

# 获取BTC/USDT的Ticker
ticker = exchange.fetch_ticker('BTC/USDT')

print(f"最新价: {ticker['last']}")
print(f"买一价: {ticker['bid']}")
print(f"卖一价: {ticker['ask']}")
print(f"24h成交量: {ticker['baseVolume']}")
print(f"24h涨跌幅: {ticker['percentage']}%")

嗯，这里要注意：Ticker是「瞬时数据」，每次调用都返回当前快照。如果你需要连续监控，建议用WebSocket，而不是轮询。不过那是后面章节的内容了。

获取K线数据

K线才是价差分析的核心。它包含了开盘、最高、最低、收盘、成交量五个维度。

我个人习惯用1分钟K线做高频价差，用15分钟K线做日内趋势。

# 获取BTC/USDT的1分钟K线，最近100根
ohlcv = exchange.fetch_ohlcv('BTC/USDT', timeframe='1m', limit=100)

# 返回格式：[[timestamp, open, high, low, close, volume], ...]
for candle in ohlcv[:5]:  # 只打印前5根
    timestamp = candle[0]
    open_price = candle[1]
    high_price = candle[2]
    low_price = candle[3]
    close_price = candle[4]
    volume = candle[5]
    
    print(f"时间: {timestamp}, 开: {open_price}, 高: {high_price}, 低: {low_price}, 收: {close_price}, 量: {volume}")

数据频率选择

做价差捕捉，频率选择很关键。我总结了一个经验表：

策略类型	推荐频率	适用场景
高频价差	1秒 - 1分钟	期现套利、跨所套利
日内价差	5分钟 - 15分钟	趋势跟踪、均值回归
中低频价差	1小时 - 4小时	统计套利、配对交易

你想想看，如果你做的是期现套利，价差可能在几秒内就消失了。用1小时K线根本抓不住。反过来，如果你做的是跨品种价差，用1秒K线反而会被噪音干扰。

完整示例：获取现货与期货的价差数据

这是我最常用的一个脚本，同时获取现货和永续合约的K线，然后计算价差：

import ccxt
import pandas as pd
from datetime import datetime

# 初始化现货和合约交易所
spot = ccxt.binance({'options': {'defaultType': 'spot'}})
future = ccxt.binance({'options': {'defaultType': 'future'}})

# 获取数据
symbol = 'BTC/USDT'
timeframe = '1m'
limit = 100

spot_ohlcv = spot.fetch_ohlcv(symbol, timeframe, limit=limit)
future_ohlcv = future.fetch_ohlcv(symbol, timeframe, limit=limit)

# 转成DataFrame
spot_df = pd.DataFrame(spot_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
future_df = pd.DataFrame(future_ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])

# 计算价差
spot_df['future_close'] = future_df['close']
spot_df['spread'] = spot_df['future_close'] - spot_df['close']

# 打印最新价差
latest = spot_df.iloc[-1]
print(f"时间: {datetime.fromtimestamp(latest['timestamp']/1000)}")
print(f"现货价: {latest['close']:.2f}")
print(f"期货价: {latest['future_close']:.2f}")
print(f"价差: {latest['spread']:.2f}")

本章知识体系

下面这张图，是我做数据获取时的核心流程：

这张图展示了从初始化到计算价差的完整链路。你跟着这个流程走，基本不会出错。

常见问题

• 连接超时怎么办？——加代理，或者换节点。我一般用交易所的备用域名。
• 数据对不上？——检查时间戳。现货和合约的K线时间可能不同步，需要对齐。
• 频率太高被封？——开启enableRateLimit，或者用多个API Key轮询。