3. 数据获取与处理：交易所API介绍（以Binance为例）、获取K线数据、数据清洗与存储
做量化交易，第一步就是搞定数据。没有数据，策略就是空中楼阁。今天咱们就聊聊怎么从交易所拿数据，怎么洗干净，怎么存起来。
我个人习惯用Binance，它API文档清晰，社区活跃，坑也相对少。你想想看，如果API文档都写不明白，你敢把真金白银放进去吗？
3.1 交易所API是个啥？
说白了，API就是交易所给你开的一个“后门”。你写代码，通过这个门跟交易所服务器说话。你说“我要查BTC价格”，它就把价格给你。你说“我要买0.1个ETH”，它就帮你下单。
Binance的API分两种：
- REST API：一问一答。你发请求，它回数据。适合拿历史数据、查账户信息。
- WebSocket：长连接。交易所主动推数据给你。适合实时行情、订单簿更新。
嗯，这里要注意：REST API有频率限制。Binance对普通API Key的限制是每分钟1200次请求。我曾经有个同事，写了个死循环去拉数据，结果IP直接被封了24小时。所以，写代码时一定要加限速。
3.2 获取K线数据
K线数据是量化策略的“米”。没米怎么做饭？
Binance的K线接口长这样：
GET /api/v3/klines
参数：
symbol: 交易对，比如"BTCUSDT"
interval: K线周期，"1m","5m","15m","1h","4h","1d"等
limit: 返回条数，最大1000
startTime: 开始时间戳（毫秒）
endTime: 结束时间戳（毫秒）
返回的数据是一个二维数组，每根K线包含12个字段。我一般只取前6个：
| 索引 | 字段 | 说明 |
|---|---|---|
| 0 | openTime | 开盘时间戳 |
| 1 | open | 开盘价 |
| 2 | high | 最高价 |
| 3 | low | 最低价 |
| 4 | close | 收盘价 |
| 5 | volume | 成交量 |
写个Python函数来拉数据：
import requests
import pandas as pd
def fetch_klines(symbol, interval, limit=1000):
url = "https://api.binance.com/api/v3/klines"
params = {
"symbol": symbol,
"interval": interval,
"limit": limit
}
resp = requests.get(url, params=params)
data = resp.json()
# 只取前6个字段
rows = []
for k in data:
rows.append([
pd.to_datetime(k[0], unit='ms'), # 时间戳转datetime
float(k[1]), float(k[2]), float(k[3]), float(k[4]), float(k[5])
])
df = pd.DataFrame(rows, columns=['time','open','high','low','close','volume'])
return df
# 测试一下
df = fetch_klines("BTCUSDT", "1h", 100)
print(df.head())
我在项目中遇到过一个问题：Binance返回的时间戳是毫秒级的，但Pandas默认用纳秒。直接塞进去会溢出。所以上面代码里我用了pd.to_datetime(k[0], unit='ms')，这个坑我踩过，你记住就行。
3.3 数据清洗
拿到的数据能直接用吗？不能。交易所偶尔会返回脏数据。比如：
- 某根K线的价格是0（交易所维护期间）
- 时间戳乱跳（网络波动导致数据错位）
- 成交量异常大（可能是交易所的测试交易）
我一般做三步清洗：
- 去重：按时间戳去重，保留第一条。
- 去空：删除open/high/low/close/volume为0或NaN的行。
- 排序：按时间戳升序排列。
核心原则：宁可少一根K线，也不要一根脏K线。脏数据会让策略回测结果完全失真。
def clean_klines(df):
# 去重
df = df.drop_duplicates(subset=['time'])
# 去空
df = df.dropna()
# 过滤价格和成交量为0
df = df[(df['close'] > 0) & (df['volume'] > 0)]
# 排序
df = df.sort_values('time').reset_index(drop=True)
return df
df_clean = clean_klines(df)
小技巧：清洗完后，可以检查一下时间间隔是否均匀。比如1小时K线，相邻两根的时间差应该是3600秒。如果出现跳变，说明中间有缺失数据，需要补全或标记。
3.4 数据存储
数据洗干净了，存哪儿？我个人推荐用Parquet格式。原因有三：
- 压缩率高，比CSV小3-5倍
- 读取速度快，支持列式读取
- 保留数据类型，不用每次读都重新解析
当然，如果你只是做小规模研究，CSV也够用。但如果你要存几年数据，Parquet是更好的选择。
import pyarrow.parquet as pq
import pyarrow as pa
def save_klines(df, filepath):
table = pa.Table.from_pandas(df)
pq.write_table(table, filepath)
def load_klines(filepath):
table = pq.read_table(filepath)
return table.to_pandas()
# 使用示例
save_klines(df_clean, "BTCUSDT_1h.parquet")
df_loaded = load_klines("BTCUSDT_1h.parquet")
注意：Parquet文件不能追加写入。如果你要增量更新数据，建议每天生成一个新文件，或者用数据库（比如InfluxDB）来存。
3.5 知识体系总览
下面这张图，把整个数据获取与处理的流程串起来了。你照着这个思路走，不会迷路。
这张图里，我特意加了一个循环箭头。为什么？因为数据是持续产生的。你不可能只拉一次就完事。我习惯写个定时任务，每小时拉一次最新数据，增量更新到本地。这样既保证数据完整，又不会触发API限频。
好了，数据获取与处理这块就聊到这儿。记住：数据质量决定策略上限。花80%的时间在数据上，剩下20%写策略，这是值得的。