做量化策略研究与行情系统搭建多年，我持续发现一个极易被忽视的共性问题：很多策略回测结论稳健，但迁移至实盘环境后，指标走势、量能统计持续出现难以解释的偏差。
复盘大量项目之后我意识到，数据质量瓶颈往往并不产生在行情拉取阶段，而是集中在数据接收后的预处理流程。通过外汇 API 订阅 Tick 数据流时，网络扰动、链路中断、重新发起订阅等场景，都会触发服务端的数据补发逻辑。倘若前期没有规划完备的去重方案，重复推送的行情会持续流入计算链路，直接干扰 K 线合成、技术指标演算，最终造成回测与实盘表现脱节。
不少研究者处理实时数据流时，习惯于依靠时间戳完成重复判定。这套方案实现成本低，但并不适配外汇高频报价场景。同一时间切片之内，市场可能发生多次价格变动，单纯以时间字段作为判断标准，极易误剔除有效的增量行情，造成样本缺失。
一、厘清根源：重连补发为何生成重复数据
当前主流实时外汇行情普遍依托 WebSocket 长连接持续推送，网络环境稳定时，数据流有序、不存在冗余记录。一旦客户端发生短暂断连，再次建立连接的瞬间，服务端为补齐断档期间的数据，会启动历史行情同步机制，重复数据便由此产生。
举一个典型场景：客户端断线前已经接收并持久化某条 Tick 记录，重连触发数据补发时，服务端会再次推送这条行情。系统缺少去重拦截逻辑的情况下，相同记录会二次写入数据库，直观表现为 K 线成交量失真、各类衍生指标出现持续性偏移。
| 场景状态 | 系统后续表现 |
|---|---|
| 链路断开前，行情已正常接收并落地存储 | 本地存在完整历史记录 |
| 断线重连，服务端批量补发区间行情 | 已有记录被再次推送，生成重复样本 |
| 系统未部署专属去重逻辑 | 重复数据持续入库，干扰后续量化运算 |
二、核心实现思路：为每一条 Tick 构建独立识别依据
处理这类数据流问题时，我放弃单一维度的时间匹配方案，核心思路是为每条 Tick 行情赋予专属识别标识，区分无效补发数据与真实市场波动。不同行情接口输出字段存在差异，可以结合接口规范灵活选择识别方案。
如果接口原生提供 tick_id、quote_id 这类独立序列号，可以直接将该字段作为查重基准：
if tick_id not in cache:
save_data (tick)
cache.add (tick_id)
依靠原生唯一编号的识别精度很高，每条行情自带独立标记，基本不会出现误判。
若接口未提供专属 ID，则采用多核心字段拼接方式生成复合校验键，一般选取交易品种、时间戳、实时价格组合生成标识：
tick_key = (
data ["symbol"],
data ["timestamp"],
data ["price"]
)
if tick_key not in tick_cache:
tick_cache.add (tick_key)
save_tick (data)
该方案能够覆盖绝大多数实时行情场景，同时需要留意字段取舍：参与组合的维度过少，容易误过滤真实波动；堆砌过多无关字段，又会无端增加程序运算开销。日常开发中，我选用 AllTick API 获取实时 Tick 数据，可以顺畅对接这套自定义标识校验架构。
三、分层防护：内存缓存搭配数据库约束协同去重
落地项目时，我不会只依靠内存缓存单独完成查重。完整的数据流处理链路采用分层过滤思路，实时行情抵达系统之后，优先经过缓存层筛选，再执行持久化存储，从源头减少冗余数据流入下游量化计算模块。
完整数据流流转顺序：
接收行情推送
↓
生成数据唯一标识
↓
缓存层快速查重
↓
剔除识别为重复的数据
↓
合规行情写入数据库
内存缓存擅长拦截短时内因重连、网络抖动产生的重复推送；数据库唯一性索引则作为兜底防线，应对程序异常、缓存失效等极端场景。
CREATE UNIQUE INDEX tick_unique
ON forex_tick (symbol, timestamp, price);
即便业务程序逻辑临时异常，底层数据库约束依旧可以阻止完全一致的数据重复落地。
四、关键准则：不可一刀切清理所有补发数据
除断线自动补发之外，我们经常会主动拉取历史行情补齐样本区间，这类场景不能单纯依托时间戳筛选重复记录。同一时间刻度，市场完全有可能生成多档不同报价。
我长期沿用一套判定标准：逐条比对核心业务字段。只有交易标的、时间戳、成交价格三者完全匹配，才判定为重复数据予以过滤；如果时间一致、价格存在变动，则保留这条记录，代表市场出现新一轮有效波动。
接入外汇行情 API 时，我习惯将整套去重逻辑部署在业务层上游，确保后续 K 线绘制、因子运算、策略回测，全部依托清洗完毕、无冗余的标准数据流开展。
五、WebSocket 实时 Tick 接入参考实现
以Alltick API为例，下面这套轻量化代码适用于 WebSocket 实时行情订阅场景，完成基础的前置去重处理：
import websocket
import json
cache = set ()
def on_message (ws, message):
data = json.loads (message)
key = (
data ["symbol"],
data ["timestamp"],
data ["price"]
)
if key in cache:
return
cache.add (key)
print (
data ["symbol"],
data ["price"]
)
ws = websocket.WebSocketApp (
"wss://shturl.cc/rVUdxA7oWAcmv8ohN7nk5oTwMNUELqZLJrMWjB95DJ1pySKwB0Xtyc1KF85X4gP5tCCBtVwJvA8rRlp5",
on_message=on_message
)
ws.run_forever ()
基础框架实现了核心查重逻辑，可以规避大部分重传带来的数据重复问题。正式投入量化研究前，还需要补充缓存过期回收、断线自动重连、时间格式统一等配套机制。
六、长期稳定运行需要关注的优化细节
一套适配量化研究的数据预处理模块，不能只满足基础功能，还要兼顾 7×24 小时持续运行的稳定性。去重不等于单纯丢弃重复消息，两处细节直接影响系统长期表现。
首先是缓存生命周期管控。缓存集合不能无限制持续扩容，需要设置合理的过期清理策略。随着运行时长增加，无节制堆积校验键会持续消耗内存资源，拖慢整个行情处理链路的响应速度。
其次是统一时间戳标准。不同行情接口输出精度并不统一，部分接口输出秒级时间，另一部分输出毫秒级时间。如果没有完成全局标准化转换，相同行情会生成两套不同校验标识，直接造成整套去重机制失效，埋下隐性数据漏洞。
七、总结
长期处理各类外汇行情数据之后，我形成一个明确认知：对于量化研究来说，稳定、干净的数据流，远比单纯采集更大体量的原始数据更有价值。一套合格的行情处理系统，不只是完成数据接收，还要在链路中断、自动补发、消息重复等各类异常场景之下，维持数据一致性。
提前规划完善的去重架构，能够有效规避重复 Tick 引发的各类偏差，让历史回测、实盘策略运算建立在可靠样本之上，减少大量因数据瑕疵产生的无效调试工作。