第3章订单流数据采集

3、订单流数据采集：外汇ECN数据源介绍、WebSocket实时订阅、数据清洗与对齐

做量化交易的朋友都知道一句话：数据是交易的血液。尤其是订单流数据，它记录了每一笔成交的细节——谁在买、谁在卖、成交了多少、在什么价位。没有这些，你做的所有分析都是空中楼阁。

这一章，我们就来聊聊怎么拿到这些数据，并且把它洗干净、对齐好。我踩过的坑不少，希望能帮你绕过去。

3.1 外汇ECN数据源：到底选哪个？

ECN，全称是Electronic Communication Network。说白了，就是一个电子撮合系统。它把全球的银行、对冲基金、零售经纪商的订单都拉进来，按价格优先、时间优先的原则撮合。

市面上主流的ECN数据源，我列个表给你看：

数据源	特点	延迟	费用
Dukascopy	历史数据全，Tick级精度，免费	中等	免费（需注册）
TrueFX	机构级数据，无做市商干预	低	付费
LMAX	交易所模式，数据最干净	极低	付费
FXCM	零售端数据，有做市商成分	中等	部分免费

我个人习惯用Dukascopy做回测，因为它的历史数据免费且完整。但实盘的话，我建议用LMAX或TrueFX。为什么？因为Dukascopy的数据虽然全，但它的撮合逻辑和真实ECN有细微差别。我在项目中遇到过，用Dukascopy数据回测的策略，实盘跑起来效果差了一截。后来才发现，是数据源的问题。

  核心要点：回测用Dukascopy，实盘用LMAX或TrueFX。别混用，否则你会怀疑人生。

3.2 WebSocket实时订阅：别用轮询了

很多新手喜欢用REST API轮询拿数据。每秒钟发一次请求，问服务器“有没有新数据？” 这种做法，说白了就是浪费带宽，还拿不到实时数据。

WebSocket就不一样了。它是一条长连接，服务器有新数据，直接推给你。延迟低，效率高。

下面是一个用Python订阅Dukascopy实时数据的例子：

import websocket
import json

def on_message(ws, message):
    data = json.loads(message)
    # 这里处理收到的订单流数据
    print(f"收到Tick: {data['symbol']} @ {data['price']}")

def on_error(ws, error):
    print(f"连接出错: {error}")

def on_close(ws, close_status_code, close_msg):
    print("连接关闭")

def on_open(ws):
    # 订阅EUR/USD的订单流
    subscribe_msg = {
        "type": "subscribe",
        "symbol": "EUR/USD",
        "channels": ["tick", "depth"]
    }
    ws.send(json.dumps(subscribe_msg))
    print("订阅成功")

# 启动连接
ws_url = "wss://data.dukascopy.com/stream"
ws = websocket.WebSocketApp(ws_url,
                            on_open=on_open,
                            on_message=on_message,
                            on_error=on_error,
                            on_close=on_close)
ws.run_forever()

这段代码看起来简单，但有几个坑要注意：

心跳机制：很多ECN数据源要求客户端定期发送心跳包，否则会断开连接。我建议每30秒发一次。
重连逻辑：网络波动是常态。我曾经在凌晨3点被运维电话叫醒，说数据断了3个小时。从那以后，我写的所有WebSocket客户端都带自动重连。
数据积压：如果处理速度跟不上接收速度，数据会积压在内存里。轻则OOM，重则程序崩溃。

我的小技巧：用队列来缓冲数据。WebSocket收到数据后，先丢进队列，然后由另一个线程慢慢处理。这样即使处理慢，也不会丢数据。

3.3 数据清洗：脏数据比没数据更可怕

你想想看，如果数据里有几个异常值，你的策略可能就会做出错误的交易决策。我见过有人因为一个错误的Tick数据，开了一手不该开的单，亏了好几万。

常见的脏数据有这几类：

重复数据：同一个Tick被推送了两次。
乱序数据：后一个Tick的时间戳比前一个还早。
异常价格：价格突然跳到一个离谱的值，比如EUR/USD突然变成0.5。
缺失数据：某段时间内完全没有数据。

清洗的逻辑，我一般这样写：

def clean_tick_data(tick):
    # 1. 去重：检查时间戳和价格是否完全一致
    if (tick['timestamp'], tick['price']) in seen_set:
        return None  # 丢弃

    # 2. 检查乱序
    if tick['timestamp'] <= last_timestamp:
        return None  # 丢弃

    # 3. 检查异常价格（以EUR/USD为例）
    if tick['price'] < 0.8 or tick['price'] > 1.5:
        return None  # 丢弃

    # 4. 更新状态
    seen_set.add((tick['timestamp'], tick['price']))
    last_timestamp = tick['timestamp']

    return tick

注意：清洗逻辑不能太激进。有些看似异常的数据，其实是真实的极端行情。比如2015年瑞郎黑天鹅事件，EUR/CHF瞬间暴跌30%。如果你把这种数据也洗掉了，那你的策略永远学不会应对极端行情。

3.4 数据对齐：不同数据源的时间戳战争

做订单流分析，你往往需要同时订阅多个数据源。比如，用Dukascopy拿Tick数据，用LMAX拿深度数据。问题来了——这两个数据源的时间戳可能不一致。

为什么会这样？因为每个数据源的服务器时间不同，网络延迟也不同。同一个成交事件，在Dukascopy上记录为10:00:00.123，在LMAX上可能记录为10:00:00.456。

对齐的方法，我推荐两种：

时间戳对齐：以其中一个数据源的时间为准，把另一个数据源的数据按时间戳插值对齐。
事件ID对齐：如果数据源提供了唯一的成交ID，直接用ID匹配。这是最准确的方法。

下面是一个简单的时间戳对齐代码：

def align_data(primary_data, secondary_data):
    aligned = []
    j = 0
    for i, p_tick in enumerate(primary_data):
        # 找到secondary_data中时间最接近的tick
        while j < len(secondary_data) - 1 and \
              secondary_data[j+1]['timestamp'] <= p_tick['timestamp']:
            j += 1
        # 取前后两个tick做线性插值
        if j < len(secondary_data) - 1:
            s_tick = interpolate(secondary_data[j], secondary_data[j+1], p_tick['timestamp'])
            aligned.append((p_tick, s_tick))
    return aligned

  经验之谈：我曾经在回测中忽略了数据对齐，结果策略的夏普比率算出来是3.5，实盘只有0.8。后来才发现，是数据没对齐导致的虚假信号。从那以后，数据对齐成了我流水线上最严格的一环。

3.5 本章知识体系

下面这张图，帮你理清本章的核心逻辑：

嗯，到这里，数据采集的核心内容就讲完了。你可能会问，这些代码能直接用吗？我的建议是，先理解逻辑，再根据你自己的数据源做适配。每个数据源的API都有细微差别，直接复制粘贴大概率跑不通。

最后提醒一句：数据采集是整个交易系统的地基。地基没打好，上面盖的房子再漂亮，早晚也得塌。别急着跑策略，先把数据流跑通、跑稳。

公众号:蓝海资料掘金营，微信deep3321

📚 课程目录

3、订单流数据采集：外汇ECN数据源介绍、WebSocket实时订阅、数据清洗与对齐

3.1 外汇ECN数据源：到底选哪个？

3.2 WebSocket实时订阅：别用轮询了

3.3 数据清洗：脏数据比没数据更可怕

3.4 数据对齐：不同数据源的时间戳战争

3.5 本章知识体系