CTP期货接口数据接收：行情回调、成交去重与断线恢复

简介：面向期货行情接入初学者与C++开发者的CTP数据接收示例工程，聚焦上期所等期货行情接收场景，提供一套经过实际编译验证的完整代码，可直接对接期货公司前置获取行情数据，帮助绕开网络上常见示例无法编译、报错频繁的问题。压缩包共76个文件，约19.74MB，以cpp和h源码、dll动态库与lib导入库为核心，辅以sln和vcxproj工程文件、exe与pdb调试产物、con和tlog运行日志及xml配置说明等，目录保留VS2010/2012下的完整构建环境。目前已有1534人学习下载。读者可参考其中的行情与交易API头文件、SPI回调实现和工程配置，理解CTP行情订阅、连接与数据回调流程，并对照附带说明完成账号申请后的联调；适合需要可运行基线工程、排查编译链接问题或搭建期货数据接收程序的中级开发者。

1. 期货CTP接口数据接收：先搞清楚你在收哪几路数据

做期货CTP接口数据接收，最容易翻车的不是连不上前置，而是连上之后把回调里的数直接当成干净成交价用。柜台推过来的行情快照、订单回报、成交回报是三条不同的流，初始化顺序、订阅方式、去重键都不一样。这篇从一个能跑起来的最小例子切入，把行情 MdApi 和交易 TraderApi 的接收链路拆开：前置地址怎么填、登录和结算确认谁先谁后、OnRtnDepthMarketData里哪些字段能直接用、断线之后流文件怎么续。适合刚接手柜台接入的开发者，也适合已经能收到数、但被漏包和异常值折腾过的熟手。

2. 环境与最小骨架：让 OnFrontConnected 先回来

2.1 动态库、位数与目录：三件必须先对齐的事

CTP 接口的接入成本，一半花在环境上。拿到的压缩包通常只有头文件和动态库，没有源码，编译期靠头文件，运行期靠.so或.dll。第一件要对齐的是位数：柜台给的 Linux 库分 32 位和 64 位，你的编译器和 Python 解释器必须跟库一致。用file看一眼库的架构，比运行时报wrong ELF class再回头查要省事。

第二件是流文件目录。CreateFtdcMdApi和CreateFtdcTraderApi的第一个参数都是流文件存放路径，API 会把私有流、公共流的重传数据写进去。目录不存在、没有写权限、磁盘满，表现都是Init之后回调不回来，日志里还不一定有明显报错。我的习惯是启动脚本里先建目录再起进程，目录按flow/md、flow/td分开，避免两个 API 实例抢同一个路径。

第三件是网络出站。前置地址格式是tcp://ip:port，行情前置和交易前置通常是不同端口，别把两个地址填反。连不上时先telnet或nc -vz探端口，再看本机防火墙出站规则，最后才怀疑库版本。下面这段准备脚本把目录、库架构、端口探测一次性做完：

#!/usr/bin/env bash set -euo pipefail BASE=/opt/ctp-demo mkdir -p "$BASE"/{bin,include,lib,flow/md,flow/td,log} # 1. 确认动态库架构与当前解释器/编译器一致 file "$BASE/lib/thostmduserapi_se.so" file "$BASE/lib/thosttraderapi_se.so" getconf LONG_BIT # 2. 确认动态库依赖能被解析，缺 libstdc++ 会在这里暴露 ldd "$BASE/lib/thostmduserapi_se.so" | grep -E "not found" && exit 1 || true # 3. 探测前置端口，行情和交易分开测 timeout 3 bash -c 'cat < /dev/null > /dev/tcp/127.0.0.1/41213' && echo "md front ok" timeout 3 bash -c 'cat < /dev/null > /dev/tcp/127.0.0.1/41205' && echo "td front ok"

脚本里的BASE是自建目录，flow/md和flow/td分别给行情、交易 API 用。ldd那行只筛not found，有缺失直接退出，不要带着半残的依赖去调回调。端口探测用 bash 内建的/dev/tcp，不依赖额外工具，超时 3 秒，避免脚本卡死。地址和端口换成柜台实际给的值，这里只做形态示例。

还有一点容易被忽略：CTP 结构体里的字符串字段大量使用定长char数组，编码常见是 GBK。合约中文名、错误信息里的中文，直接在终端打印可能乱码。接收层只做透传没问题，一旦要落库或展示，就要在边界处统一转 UTF-8，别等到查数据时才发现一半乱码一半正常。

2.2 行情 API 最小初始化序列：Create、RegisterFront、RegisterSpi、Init

行情接收的最小闭环只有五步：创建 API 实例、注册前置、注册回调对象、Init、在OnFrontConnected里登录并订阅。顺序不能乱，RegisterSpi必须早于Init，否则最早的那次连接回调会丢。下面这段 C++ 骨架可以直接编译运行，把地址、账号换成柜台给的值即可：

#include "ThostFtdcMdApi.h" #include <cstdio> #include <cstring> class MdSpi : public CThostFtdcMdSpi { public: explicit MdSpi(CThostFtdcMdApi* api) : api_(api) {} void OnFrontConnected() override { CThostFtdcReqUserLoginField req; std::memset(&req, 0, sizeof(req)); std::strncpy(req.BrokerID, "9999", sizeof(req.BrokerID) - 1); std::strncpy(req.UserID, "user01", sizeof(req.UserID) - 1); std::strncpy(req.Password, "pass01", sizeof(req.Password) - 1); // 返回 0 只代表请求发出，登录结果在 OnRspUserLogin int ret = api_->ReqUserLogin(&req, ++req_id_); if (ret != 0) std::printf("ReqUserLogin send failed ret=%d\n", ret); } void OnRspUserLogin(CThostFtdcRspUserLoginField* rsp, CThostFtdcRspInfoField* info, int req_id, bool last) override { if (info && info->ErrorID != 0) { std::printf("login failed id=%d msg=%s\n", info->ErrorID, info->ErrorMsg); return; } char* list[] = {const_cast<char*>("rb2510"), const_cast<char*>("au2512")}; int ret = api_->SubscribeMarketData(list, 2); if (ret != 0) std::printf("subscribe send failed ret=%d\n", ret); } void OnRspSubMarketData(CThostFtdcSpecificInstrumentField* inst, CThostFtdcRspInfoField* info, int req_id, bool last) override { if (info && info->ErrorID != 0) std::printf("subscribe failed %s id=%d\n", inst ? inst->InstrumentID : "?", info->ErrorID); } void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* d) override { // 回调线程里只做拷贝或入队，不要在这里写盘 std::printf("%s %s.%03d last=%.3f vol=%d bid1=%.3f ask1=%.3f\n", d->InstrumentID, d->UpdateTime, d->UpdateMillisec, d->LastPrice, d->Volume, d->BidPrice1, d->AskPrice1); } private: CThostFtdcMdApi* api_; int req_id_ = 0; }; int main() { CThostFtdcMdApi* api = CThostFtdcMdApi::CreateFtdcMdApi("./flow/md/", false, false); if (!api) return -1; MdSpi spi(api); api->RegisterSpi(&spi); api->RegisterFront(const_cast<char*>("tcp://127.0.0.1:41213")); api->Init(); api->Join(); // 生产环境用事件循环或信号控制退出 api->Release(); return 0; }

CreateFtdcMdApi的三个参数分别是流文件目录、是否使用 UDP 行情、是否使用组播行情。后两个传false就是走 TCP 前置，先保证能收数，别一上来开组播。RegisterFront可以调用多次注册多个前置，API 内部会做切换；只注册一个也能跑，但单点断了就没有备用。Init之后当前线程不要立刻退出，Join会阻塞住，让 API 自己的网络线程工作。

编译命令里要把头文件、库路径和运行时搜索路径都带上，否则运行时会报找不到.so：

g++ -std=c++17 -O2 md_demo.cpp \ -I./include -L./lib -lthostmduserapi_se \ -Wl,-rpath,'$ORIGIN/lib' -o md_demo

-lthostmduserapi_se对应行情库，交易库是thosttraderapi_se。-Wl,-rpath,'$ORIGIN/lib'把运行时库路径钉在可执行文件旁边的lib目录，换机器部署时不用改LD_LIBRARY_PATH。

如果OnFrontConnected一直不回来，排查顺序是：前置地址有没有tcp://前缀、端口有没有探通、Init是不是在RegisterSpi之前被调用、流文件目录是否可写。这四项占了我见过的连接问题的大半，剩下的才是库版本和柜台侧限制。连接回调回来之后，登录失败常见原因是账号密码错、柜台限制登录时段、行情账号没有订阅权限，OnRspUserLogin里的ErrorID和ErrorMsg要原样打出来，别只打一句“登录失败”。

3. 行情回调拆解：OnRtnDepthMarketData 里哪些字段能直接用

3.1 订阅确认与快照字段：先分清累计量和瞬时量

SubscribeMarketData传的是合约代码数组和数量，返回 0 只代表请求发出，真正订阅成功要看OnRspSubMarketData。有些柜台对不存在的合约、没有权限的合约不会在发送时拒绝，只在响应回调里给错误码。订阅数量也不要一个连接硬扛全市场，按板块拆成多个MdApi实例更稳，单连接订阅过多时，部分柜台会限速或直接丢订阅请求。

OnRtnDepthMarketData推的是快照，不是逐笔成交。字段里最容易被误用的是Volume和Turnover：这两个都是当日累计值，不是本次增量。LastPrice是最新价，涨跌停时可能等于停板价；BidPrice1、AskPrice1是买一卖一，但集合竞价、停板、无报价时会出现无效值。CTP 习惯用DBL_MAX（约1.7976931348623157e+308）表示无效浮点数，直接参与计算会把均值、差值全部带偏。

回调里第一件事是把需要的字段拷出来，别把 API 的结构体指针传到别的线程。CTP 的结构体内存在回调返回后可能被复用，异步线程再去读就是悬空指针。下面这段只做字段提取和无效值过滤：

void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* d) override { Tick t; t.instrument = d->InstrumentID; t.trading_day = d->TradingDay; t.action_day = d->ActionDay; t.update_time = d->UpdateTime; t.update_ms = d->UpdateMillisec; t.last = valid_price(d->LastPrice) ? d->LastPrice : 0.0; t.volume = d->Volume; // 累计量，后续差分 t.open_interest = d->OpenInterest; t.bid1 = valid_price(d->BidPrice1) ? d->BidPrice1 : 0.0; t.ask1 = valid_price(d->AskPrice1) ? d->AskPrice1 : 0.0; t.bid_vol1 = t.bid1 > 0 ? d->BidVolume1 : 0; t.ask_vol1 = t.ask1 > 0 ? d->AskVolume1 : 0; t.local_ns = now_ns(); // 本地接收时间，steady clock queue_.push(std::move(t)); // 回调线程只入队 } static bool valid_price(double p) { // CTP 用 DBL_MAX 表示无效值，阈值取 1e100 足够 return p > 0.0 && p < 1e100; }

valid_price里的阈值不要写成p != DBL_MAX，不同字段、不同柜台对无效值的表达不完全一致，用数量级判断更稳。local_ns用单调时钟，别用系统墙钟，墙钟被校时会跳变，延迟统计会算出负数。queue_是回调线程和落盘线程之间的边界，下一节展开。

还有一个细节：UpdateTime只到秒，UpdateMillisec是毫秒部分，拼起来才是完整时间。部分柜台在集合竞价阶段会推UpdateMillisec为 0 的快照，同一秒内多条记录时间戳相同，排序时要带上本地接收序号，不能只按交易所时间排。

3.2 累计成交量差分：夜盘和集合竞价是两个分水岭

把Volume直接当“这一笔成交了多少”是新手最常见的错误。它是当日累计，正确做法是按TradingDay + InstrumentID分组做差分。首条没有前值，增量置 0；差分为负说明交易日切换或中间丢包，也不能当成负成交。下面这段对已落盘的 tick 做差分：

import pandas as pd df = pd.read_parquet("tick.parquet") df = df.sort_values(["TradingDay", "InstrumentID", "UpdateTime", "UpdateMillisec", "local_ns"]) df["volume_delta"] = ( df.groupby(["TradingDay", "InstrumentID"])["Volume"] .diff() ) # 首条没有前值；负值来自交易日切换或缺口，先置 0 并单独标记 df["gap_flag"] = (df["volume_delta"] < 0).astype("int8") df["volume_delta"] = df["volume_delta"].fillna(0).clip(lower=0).astype("int64") print(df.groupby("InstrumentID")["volume_delta"].sum().head())

groupby的键必须带TradingDay，只按合约分组会把夜盘和日盘的累计值串起来。gap_flag单独留一列，后面核对丢包时按它筛。clip(lower=0)只是让下游计算不被负数污染，不代表缺口不存在，真正的缺口要靠行情序列的时间连续性和成交量守恒去查。

夜盘的坑在于TradingDay和ActionDay不一致。晚上 21 点之后的行情，ActionDay是自然日，TradingDay已经是下一个交易日。落盘分区如果按ActionDay切，一个文件里会混两个交易日的夜盘和日盘；按TradingDay切，又跟人的自然日直觉不一致。我的做法是文件目录按TradingDay分，文件内部保留ActionDay字段，查询时两个都能用。集合竞价的第一笔Volume可能包含竞价撮合量，这一条的增量不可靠，做日内量能指标时要么剔除，要么单独标记。

3.3 回调线程不能堵：入队、批量落盘、别在回调里发查询

CTP 的回调跑在 API 内部网络线程上，回调里做同步磁盘 IO、sleep、发查询请求，都会拖慢后续数据的接收。短时间看不出问题，行情一密集就开始丢包，而且丢得没有规律，像玄学。血泪经验是：回调函数只做字段拷贝和入队，落盘、计算、告警全部放到独立线程。

队列可以用有界环形队列，满了就计数丢弃并告警，不要让内存无限涨。下面是一个简化的互斥队列示意，生产环境可以换成无锁队列，但边界逻辑一样：

template <typename T, size_t N> class TickQueue { public: bool push(T&& v) { std::lock_guard<std::mutex> lk(mu_); if ((head_ + 1) % N == tail_) { dropped_++; return false; } buf_[head_] = std::move(v); head_ = (head_ + 1) % N; return true; } bool pop(T& out) { std::lock_guard<std::mutex> lk(mu_); if (tail_ == head_) return false; out = std::move(buf_[tail_]); tail_ = (tail_ + 1) % N; return true; } uint64_t dropped() const { return dropped_; } private: T buf_[N]; size_t head_ = 0, tail_ = 0; mutable std::mutex mu_; uint64_t dropped_ = 0; };

push返回false表示队列满，调用方要把dropped_打到监控里，不能静默丢。落盘线程每次批量取一批，比如 500 条或 100 毫秒凑一批，再写文件，减少系统调用次数。dropped_持续增长说明消费端跟不上，要么加磁盘，要么降落盘频率，要么检查是不是在回调里偷偷做了重计算。

4. 交易数据接收：登录、结算确认与私有流回报

4.1 登录链路的四个必调请求：认证、登录、结算确认、查合约

交易 API 的初始化比行情多几步。创建实例时要指定流文件目录，注册前置和回调之后Init，然后在OnFrontConnected里依次走认证、登录、结算确认、查询。生产环境通常启用了AppID和AuthCode，必须先ReqAuthenticate，认证通过才能登录；不启用认证的环境可以跳过认证直接登录，但代码里最好留开关，别把顺序写死。

class TraderSpi : public CThostFtdcTraderSpi { public: void OnFrontConnected() override { CThostFtdcReqAuthenticateField auth; std::memset(&auth, 0, sizeof(auth)); std::strncpy(auth.BrokerID, broker_, sizeof(auth.BrokerID) - 1); std::strncpy(auth.UserID, user_, sizeof(auth.UserID) - 1); std::strncpy(auth.AppID, appid_, sizeof(auth.AppID) - 1); std::strncpy(auth.AuthCode, authcode_, sizeof(auth.AuthCode) - 1); api_->ReqAuthenticate(&auth, ++req_id_); } void OnRspAuthenticate(CThostFtdcRspAuthenticateField* rsp, CThostFtdcRspInfoField* info, int id, bool last) override { if (info && info->ErrorID != 0) { std::printf("auth failed id=%d msg=%s\n", info->ErrorID, info->ErrorMsg); return; } CThostFtdcReqUserLoginField login; std::memset(&login, 0, sizeof(login)); std::strncpy(login.BrokerID, broker_, sizeof(login.BrokerID) - 1); std::strncpy(login.UserID, user_, sizeof(login.UserID) - 1); std::strncpy(login.Password, pass_, sizeof(login.Password) - 1); api_->ReqUserLogin(&login, ++req_id_); } void OnRspUserLogin(CThostFtdcRspUserLoginField* rsp, CThostFtdcRspInfoField* info, int id, bool last) override { if (info && info->ErrorID != 0) return; session_id_ = rsp->SessionID; front_id_ = rsp->FrontID; CThostFtdcSettlementInfoConfirmField confirm; std::memset(&confirm, 0, sizeof(confirm)); std::strncpy(confirm.BrokerID, broker_, sizeof(confirm.BrokerID) - 1); std::strncpy(confirm.InvestorID, user_, sizeof(confirm.InvestorID) - 1); api_->ReqSettlementInfoConfirm(&confirm, ++req_id_); } void OnRspSettlementInfoConfirm(CThostFtdcSettlementInfoConfirmField* rsp, CThostFtdcRspInfoField* info, int id, bool last) override { if (info && info->ErrorID != 0) return; CThostFtdcQryInstrumentField q; std::memset(&q, 0, sizeof(q)); api_->ReqQryInstrument(&q, ++req_id_); // 空条件查全市场，注意流控 } private: CThostFtdcTraderApi* api_ = nullptr; int req_id_ = 0; int front_id_ = 0, session_id_ = 0; const char* broker_ = "9999"; const char* user_ = "user01"; const char* pass_ = "pass01"; const char* appid_ = "app01"; const char* authcode_ = "auth01"; };

ReqAuthenticate里的AppID、AuthCode由柜台分配，填错会在认证回调里直接报错，别反复重试把账号锁了。OnRspUserLogin里的FrontID和SessionID要存下来，后面订单回报去重会用到。结算确认是每个交易日首次登录的必做动作，不做的话后续查询和交易会被柜台拒绝，错误信息通常跟“未确认结算单”有关。

ReqQryInstrument空条件会返回全市场合约，回调OnRspQryInstrument会分很多批，bIsLast为true才算结束。合约列表一般一天查一次落库缓存，不要每次登录都全量查，查询类请求有每秒一次的限制，发快了会返回流控错误。

4.2 私有流与回报：OnRtnOrder 不是重复包，OnRtnTrade 才按成交去重

交易 API 的回报分私有流和公共流。私有流推自己的订单、成交、资金变化，公共流推合约状态、公告。创建实例后要显式订阅：

CThostFtdcTraderApi* api = CThostFtdcTraderApi::CreateFtdcTraderApi("./flow/td/"); api->SubscribePrivateTopic(THOST_TERT_RESTART); api->SubscribePublicTopic(THOST_TERT_RESTART); api->RegisterSpi(&spi); api->RegisterFront(const_cast<char*>("tcp://127.0.0.1:41205")); api->Init();

订阅模式有三个常用值，选错会直接表现为“重启后漏成交”或“收到重复回报”：

THOST_TERT_RESTART在盘中重启时会把当天历史回报重推一遍，如果不做去重，订单表会重复。THOST_TERT_QUICK最省事，但重启期间的成交就漏了。要求不漏单的场景用RESUME，前提是流文件目录保留着，别每次重启前清空。

订单回报和成交回报的去重键不一样。OnRtnOrder按FrontID + SessionID + OrderRef定位一笔订单，但同一笔订单在“已报、部分成交、全成、已撤”每个状态变化时都会回调一次，这不是重复推送，是状态机在推进，处理时按OrderStatus更新而不是插入新行。OnRtnTrade才是成交明细，唯一键用TradeID，同一笔成交不要按OrderRef去重：

void OnRtnOrder(CThostFtdcOrderField* o) override { // 同一 OrderRef 会多次回调，按状态更新，不要当重复包丢弃 std::printf("order %s ref=%s status=%c traded=%d\n", o->InstrumentID, o->OrderRef, o->OrderStatus, o->VolumeTraded); } void OnRtnTrade(CThostFtdcTradeField* t) override { // 成交唯一键是 TradeID，同一订单可以有多笔成交 std::printf("trade %s trade_id=%s price=%.2f vol=%d\n", t->InstrumentID, t->TradeID, t->Price, t->Volume); }

OrderStatus是字符枚举，常见值包括未报、已报、部分成交、全部成交、已撤，落库时建议同时保留原始字符和可读映射。VolumeTraded是累计成交手数，跟行情里的Volume一样是累计语义，不要再做一次求和。

4.3 查询流控与回报堆积：串行队列比并发更快

交易 API 的查询类请求有频率限制，常见表现是返回-2（未处理请求超过许可数）或-3（每秒发送请求数超过许可数）。查询合约、查资金、查持仓、查成交，全部走同一个限制。并发发查询不会更快，只会让请求在柜台侧排队，回报延迟反而变大。

我的做法是做一个串行查询队列，全局每秒只放一个查询出去，收到OnRspXxx的bIsLast之后再放下一个。查询结果用请求 ID 关联，别靠回调顺序猜。回报线程和查询线程分开，OnRtnOrder、OnRtnTrade不参与查询排队，否则一笔成交回报会被前面的查询堵住。

5. 避坑与排查：断线、漏包、异常值的现场处理

5.1 断线重连与流文件恢复：先分清 API 重连和进程重启

CTP 的断线分两种。一种是网络抖动，API 内部会重连，OnFrontDisconnected先回调，过一会儿OnFrontConnected再回调；另一种是进程挂了重启，这时候靠流文件恢复。两种情况的处理不一样，混在一起就会漏数据。

OnFrontDisconnected的nReason能给出大致方向：网络读失败、网络写失败、接收心跳超时、发送心跳失败、收到错误报文。读失败和心跳超时多半是网络或前置侧问题，写失败可能是本机发送缓冲满，错误报文要看柜台日志。行情连接重连成功后，之前的订阅关系不一定自动恢复，稳妥做法是在OnFrontConnected里重新登录、重新订阅，并把订阅列表做成可重放的配置。

交易连接重连后，如果进程没重启，API 会自己续上私有流；如果进程重启，SubscribePrivateTopic用RESUME，并且保证流文件目录没被清空。流文件目录一旦删除，RESUME没有续传起点，只能退化成RESTART或QUICK，这时候必须用成交查询做对账补漏。对账逻辑不要省，它是漏单的最后一道后悔药。

5.2 五条踩坑记录：现象、原因、解决

第一条：夜盘成交量出现负增量。现象是差分后的volume_delta在 21:00 附近为负，白天正常。原因是夜盘切换交易日，Volume从新交易日重新累计，而分组键只用了合约代码。解决是把TradingDay加进分组键，负值置 0 并打

本文还有配套的精品资源，点击获取