第二章 数据获取：API接口选择、实时行情订阅、历史数据下载、数据清洗与对齐
做外汇套利策略，第一步不是写策略逻辑，而是搞定数据。
我见过太多人，策略写得天花乱坠，结果一跑回测，数据全是坑。嗯，数据这关过不了，后面全是白费功夫。
2.1 API接口选择：别盲目追求大而全
市面上能拿外汇数据的API不少，但真正适合套利策略的，其实就那么几个。我个人习惯把API分成三类：
| 类型 | 代表 | 适合场景 | 注意点 |
|---|---|---|---|
| 零售经纪商API | OANDA、FXCM、IB | 个人交易、中小资金 | 点差数据真实，但历史数据有限 |
| 专业数据商API | Bloomberg、Refinitiv | 机构级、高频套利 | 贵，但数据质量最高 |
| 开源/免费API | ForexPython、Alpha Vantage | 学习、原型验证 | 延迟高，不适合实盘 |
我建议初学者先从OANDA的API入手。为什么？因为它的文档写得清楚，而且有模拟账户可以随便折腾。我曾经在项目里用OANDA的v20接口，跑了半年套利回测，数据一致性做得很好。
核心原则：API的选择取决于你的套利频率。做1分钟级别的三角套利，延迟必须低于100ms；做日线级别的套利，免费API也够用。
2.2 实时行情订阅：Tick级数据的陷阱
实时行情订阅，说白了就是让数据自己送上门来。WebSocket是主流方案。
这里有个坑——你以为拿到的是实时数据，其实可能是延迟数据。我遇到过一家经纪商，它的Tick数据比真实市场慢了300毫秒。你想想看，300毫秒在套利世界里，黄花菜都凉了。
正确的做法是：订阅多个数据源，做交叉验证。代码示例如下：
import asyncio
import websockets
import json
# OANDA实时行情订阅示例
async def subscribe_forex_prices():
uri = "wss://stream-fxpractice.oanda.com/v3"
async with websockets.connect(uri) as websocket:
# 订阅EUR/USD和USD/JPY
request = {
"type": "subscribe",
"instruments": ["EUR_USD", "USD_JPY"]
}
await websocket.send(json.dumps(request))
while True:
response = await websocket.recv()
data = json.loads(response)
# 这里做数据对齐处理
process_tick_data(data)
asyncio.run(subscribe_forex_prices())避坑指南：我曾经在实盘中发现，某个数据源的EUR/USD报价总是比另一个慢一拍。后来查出来是它的服务器在新加坡，而我的服务器在美东。所以，订阅实时行情时，尽量选离你服务器近的数据中心。
2.3 历史数据下载：别只盯着1分钟线
做套利策略回测，历史数据是命根子。但很多人只下载1分钟K线，这其实不够。
为什么？因为套利策略的入场点往往在秒级甚至毫秒级。1分钟K线会把很多细节抹掉。我个人习惯的做法是：
- Tick数据：用于高频套利策略的精确回测
- 1分钟数据：用于中频套利策略的初步筛选
- 5分钟数据：用于统计套利的协整检验
下载历史数据时，要注意数据的时间戳时区问题。很多API默认返回UTC时间，但你的策略可能跑在美东时间或北京时间。不统一时区，回测结果全是错的。
import pandas as pd
import oandapyV20
from oandapyV20.endpoints import instruments
# 下载EUR/USD历史数据
client = oandapyV20.API(access_token="your_token")
params = {
"granularity": "M1", # 1分钟线
"from": "2024-01-01T00:00:00Z",
"to": "2024-01-31T23:59:59Z"
}
r = instruments.InstrumentsCandles(instrument="EUR_USD", params=params)
client.request(r)
# 转为DataFrame并统一时区
df = pd.DataFrame(r.response['candles'])
df['time'] = pd.to_datetime(df['time'])
df = df.set_index('time').tz_convert('America/New_York')警告：历史数据下载时，一定要检查数据完整性。我曾经下载了3年的数据，跑回测时发现2019年11月的数据全是NaN。原因是那天API服务器维护，但没返回任何错误码。所以，下载完数据后，第一件事就是检查缺失值比例。
2.4 数据清洗与对齐：套利策略的命门
数据清洗，说白了就是把脏数据扔掉。但怎么扔、扔多少，这里头有讲究。
外汇市场的数据脏在哪？主要有三个地方：
- 跳空：周末开盘或重大新闻时，价格直接跳空
- 重复数据：同一个Tick被推送了两次
- 异常点：比如EUR/USD突然从1.10跳到1.20，明显是数据错误
我的清洗流程是这样的：
def clean_forex_data(df):
# 1. 删除重复行
df = df.drop_duplicates()
# 2. 处理缺失值
df = df.fillna(method='ffill') # 前向填充
# 3. 检测异常点（3倍标准差法）
for col in ['bid', 'ask']:
mean = df[col].mean()
std = df[col].std()
df = df[(df[col] > mean - 3*std) & (df[col] < mean + 3*std)]
# 4. 对齐时间戳（统一到毫秒级）
df = df.resample('1ms').last().dropna()
return df数据对齐是套利策略里最容易被忽视的环节。你想想看，如果你做EUR/USD和USD/JPY的套利，两个数据源的时间戳差了1秒，算出来的套利信号全是假的。
我常用的对齐方法是：以最慢的数据源为基准，其他数据源做线性插值对齐到同一时间点。
核心要点：数据对齐的精度决定了套利策略的成败。做1分钟级别的套利，对齐到秒级就够了；做Tick级别的套利，必须对齐到毫秒级。
2.5 本章知识体系总览
下面这张图，是我自己总结的数据获取全流程。每次做新策略，我都会按这个流程走一遍，确保数据不出问题。
这张图把整个数据获取流程串起来了。从选择数据源开始，到最终拿到对齐后的多币种DataFrame，每一步都不能省。
个人经验：我刚开始做套利策略时，总觉得数据清洗浪费时间。后来有一次，因为没做异常检测，回测结果年化收益30%，实盘直接亏了5%。从那以后，我再也不敢跳过数据清洗这一步了。
数据获取这件事，说难不难，说简单也不简单。关键是要形成自己的流程和规范。你按照我上面说的这套方法去做，至少能避开80%的数据坑。