3. 数据源准备（上）：获取比特币/以太坊价格数据
做量化交易，第一步就是搞定数据。没有数据，再好的策略也是空中楼阁。
我个人习惯用 CCXT 这个库来拉数据。为什么？因为它统一了上百家交易所的 API 接口。你写一套代码，就能对接币安、OKX、火币……省心不少。
3.1 为什么选 CCXT？
说白了，CCXT 就是一个「翻译官」。每家交易所的 API 格式都不一样，有的用 REST，有的用 WebSocket，参数命名也千奇百怪。CCXT 帮你把这些差异抹平了。
我在项目中遇到过最头疼的事：某交易所突然改了接口返回格式，我的数据管道直接崩了。后来换了 CCXT，它社区活跃，更新快，这种坑基本不用我操心。
核心优势：
- 支持 100+ 交易所，包括主流和去中心化交易所
- 统一的方法名：fetch_ohlcv()、fetch_ticker()、fetch_order_book()
- 自动处理频率限制、重试、错误码
- 纯 Python 实现，安装简单：pip install ccxt
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)} 个交易对")
小技巧： 如果你只是拉历史 K 线，不需要 API Key。但建议加上 enableRateLimit=True，我曾经因为没开这个，一分钟请求了 2000 次，直接被币安封了 IP 半小时……
3.3 获取 K 线数据（OHLCV）
恐慌贪婪指数需要价格数据来计算。我们最常用的是日线数据。
def fetch_ohlcv(symbol='BTC/USDT', timeframe='1d', limit=500):
"""
获取历史K线数据
:param symbol: 交易对，如 'BTC/USDT'
:param timeframe: 时间周期，'1d' 日线，'1h' 小时线
:param limit: 返回条数，最大1000
:return: list of [timestamp, open, high, low, close, volume]
"""
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
return ohlcv
# 获取比特币最近500天日线数据
btc_data = fetch_ohlcv('BTC/USDT', '1d', 500)
print(f"获取到 {len(btc_data)} 条数据")
print(f"第一条: {btc_data[0]}")
print(f"最后一条: {btc_data[-1]}")返回的数据格式是这样的：
| 索引 | 字段 | 说明 |
|---|---|---|
| 0 | timestamp | 时间戳（毫秒） |
| 1 | open | 开盘价 |
| 2 | high | 最高价 |
| 3 | low | 最低价 |
| 4 | close | 收盘价 |
| 5 | volume | 成交量 |
3.4 同时获取比特币和以太坊数据
恐慌贪婪指数通常需要多个币种的数据。我习惯写个函数批量拉取：
def fetch_multiple_symbols(symbols=['BTC/USDT', 'ETH/USDT'], timeframe='1d', limit=500):
"""
批量获取多个交易对数据
"""
data = {}
for symbol in symbols:
print(f"正在获取 {symbol} ...")
ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
data[symbol] = ohlcv
print(f" -> 获取到 {len(ohlcv)} 条")
return data
# 拉取数据
all_data = fetch_multiple_symbols(['BTC/USDT', 'ETH/USDT'], '1d', 500)
# 查看比特币收盘价
btc_close = [row[4] for row in all_data['BTC/USDT']]
print(f"比特币最近收盘价: {btc_close[-5:]}")
注意： 免费 API 有频率限制。币安现货接口限制是 1200 次/分钟。如果你同时拉很多币种，建议加个 time.sleep(0.1) 或者直接用 enableRateLimit。
3.5 数据清洗与格式化
原始数据是列表，用起来不方便。我习惯转成 pandas DataFrame：
import pandas as pd
def ohlcv_to_dataframe(ohlcv, symbol='BTC/USDT'):
"""
将OHLCV原始数据转为DataFrame
"""
df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
df['symbol'] = symbol
return df
# 转换比特币数据
btc_df = ohlcv_to_dataframe(all_data['BTC/USDT'], 'BTC/USDT')
eth_df = ohlcv_to_dataframe(all_data['ETH/USDT'], 'ETH/USDT')
print(btc_df.head())嗯，这里要注意：时间戳单位是毫秒，pandas 的 to_datetime 要指定 unit='ms'。我刚开始做的时候忘了这个，结果时间全乱了，回测结果一塌糊涂。
3.6 核心逻辑流程图
下面这张图概括了数据获取的完整流程：
3.7 避坑指南
做数据获取这几年，我踩过的坑不少。分享几个最典型的：
- 时间戳问题： 我曾经直接把毫秒时间戳当 datetime 用，结果回测数据全对不上。记住：CCXT 返回的是毫秒，pandas 要加 unit='ms'。
- 数据缺失： 有些交易所某些币种的历史数据不全。我建议拉完数据后检查一下长度，不够就换个交易所或者用补数据的方法。
- 频率限制： 免费 API 有次数限制。我试过一次性拉 50 个币种，结果被限流了 5 分钟。后来改成逐个拉取，中间加 0.5 秒延迟，问题解决。
- 交易所选择： 不同交易所的同一币种价格有微小差异。做回测时尽量用同一个交易所的数据，否则会有套利偏差。
我的习惯： 每次拉完数据，先画个收盘价折线图看看。如果曲线有断点或者异常跳变，说明数据有问题。肉眼检查比代码检查更直观。
好了，数据源这块我们就讲到这里。比特币和以太坊的日线数据已经到手，下一节我们会把这些数据存到本地，方便后续反复使用。
公众号:蓝海资料掘金营，微信deep3321