1. 项目概述为什么要在Linux下用C处理CTP行情做量化交易或者高频策略的朋友对“CTP”和“tick行情”这两个词一定不陌生。CTP是国内期货市场的主流交易接口而tick行情则是市场最细粒度的数据每一笔成交、每一次报价变动都会形成一个tick。对于策略研发尤其是对延迟极其敏感的高频策略如何高效、稳定地接收并存储这些海量的tick数据是搭建整个策略基础设施的第一步也是最关键的一步。我选择在Linux系统下用C来实现这个功能不是赶时髦而是经过实战检验的必然选择。首先延迟是生命线。Linux内核在进程调度、网络I/O处理上相比其他系统有天然优势配合C这种“零成本抽象”的语言我们能最大程度地控制从网络接收到数据落盘的每一个环节将系统延迟压到最低。其次稳定性至关重要。行情接收服务一旦启动往往需要7x24小时不间断运行Linux系统在长时间高负载下的稳定性有目共睹。最后可控性。从网络连接到内存管理再到磁盘I/O用C在Linux上你可以实现从应用到系统层面的深度定制和优化这是使用现成高级语言框架或Windows平台难以比拟的。简单说这个项目就是打造一个在Linux环境下用纯C编写的、专门“吃”CTP行情并“消化”存储到本地的核心引擎。它不依赖任何大型中间件追求极致的性能和可靠性为后续的策略分析、回测提供最原始、最完整的数据源。2. 核心组件与架构设计思路要完成这个任务我们不能一上来就埋头写代码得先理清整个数据流的脉络和各个组件的职责。一个健壮的行情接收存储系统远不止调用几个API那么简单。2.1 CTP API与行情流解析上期技术CTP提供的API是典型的C语言动态库.so文件。它采用异步回调Callback机制。这意味着我们的程序不是主动去“拉”数据而是向API注册一系列回调函数。当有行情事件如连接成功、收到行情时API的底层线程会主动调用我们注册的函数。对于行情接收最核心的两个回调是OnRtnDepthMarketData: 这是主力军。每当有新的tick数据深度行情产生时这个函数会被触发。我们收到的CThostFtdcDepthMarketDataField结构体里包含了合约代码、最新价、成交量、买卖盘五档价量等所有信息。OnFrontConnected/OnFrontDisconnected: 处理与行情前置机的网络连接事件。我们的程序主体就是一个事件循环初始化API、设置回调、订阅合约然后等待回调函数被触发并处理数据。这里的关键在于回调函数执行速度必须足够快。如果我们在OnRtnDepthMarketData里做复杂的计算或者同步的磁盘写入会阻塞API的内部线程导致后续数据被积压甚至丢失。因此架构设计的核心原则是回调函数只做最必要的工作解析、打包然后迅速将数据转移出去。2.2 生产者-消费者模型与无锁队列为了解决上述阻塞问题生产者-消费者模型是最佳实践。在这个模型里生产者OnRtnDepthMarketData回调函数。它的任务是将收到的CThostFtdcDepthMarketDataField结构体快速复制或移动到一个内存数据结构中。缓冲区一个高效的、线程安全的队列。生产者放入数据消费者取出数据。这里我强烈推荐使用无锁队列Lock-free Queue。传统的带锁队列如std::queue加互斥锁在生产者、消费者竞争激烈时锁的开销会成为性能瓶颈并引入不确定的延迟。无锁队列通过原子操作CAS实现并发访问避免了线程挂起和调度延迟更低、吞吐量更高。像moodycamel::ConcurrentQueue这样的第三方库或者自己用std::atomic实现一个简单的单生产者单消费者队列都是不错的选择。消费者一个或多个独立的线程。它从无锁队列中批量取出积攒的tick数据负责后续的加工和持久化存储。这种设计实现了接收与处理/存储的解耦。网络回调线程几乎不受影响而存储线程可以按照自己的节奏比如攒够1000条或者定时100毫秒进行批量写入效率更高。2.3 存储方案选型二进制文件 vs 数据库数据来了存到哪里这是另一个关键决策。主要考量点是写入速度、查询便利性和存储空间。直接写二进制文件优点速度最快极致简单。将CThostFtdcDepthMarketDataField结构体直接序列化后写入文件几乎没有额外开销。可以使用内存映射文件mmap进一步提升I/O性能。缺点查询分析麻烦。要读取特定合约某段时间的数据需要自己写程序解析文件缺乏灵活性。数据管理如压缩、清理也需要自己实现。适用场景对延迟要求极端苛刻且数据主要用于存档或由特定程序批量导入数据库进行后续分析。写入关系型数据库如MySQL/PostgreSQL优点查询能力强大支持复杂的SQL分析。数据管理方便。缺点写入吞吐量是瓶颈。即使进行批量插入面对每秒可能数千甚至上万的tick数据数据库很容易成为系统瓶颈且I/O延迟较高。适用场景tick数据量不大或对实时查询有强需求。写入时序数据库如InfluxDB, TDengine优点为时间序列数据优化写入吞吐量极高压缩比好自带时间窗口查询等高级功能。缺点引入外部依赖部署和运维复杂度增加。适用场景需要实时监控和查询近期行情数据。混合架构本项目推荐核心思路双写策略。消费者线程同时做两件事高速路径将tick数据以二进制格式追加写入本地日志文件作为主存储和原始备份。追求最高写入速度和数据安全。低速路径将数据同时发送到一个内存缓存或另一个队列由另一个线程异步写入时序数据库如果安装了。用于实时监控和快速查询近期数据。优势兼顾了性能和灵活性。原始二进制文件保证了数据的完整性和最高效的存储时序数据库则提供了便捷的查询入口。对于纯粹追求性能和可靠性的场景我个人的选择是方案1直接写二进制文件并辅以良好的文件滚动和命名策略。数据库的查询需求可以通过后续的离线数据导入服务来满足。3. 核心实现细节与避坑指南有了架构蓝图我们来看看具体实现时有哪些魔鬼细节。这里分享的都是我踩过坑之后总结的经验。3.1 Linux环境准备与CTP API集成首先你需要从期货公司或上期技术官网获取CTP的API包通常是一个.zip文件。里面会包含*.so动态库文件如libthostmduserapi_se.so,libthosttraderapi_se.so我们只需要行情库。*.h头文件。*.xml或*.dtd通讯协议定义文件。在Linux下编译链接的要点库文件放置将.so文件放到系统库路径如/usr/local/lib或者你的项目可执行文件同级目录。建议后者便于部署。编译命令使用-l链接库注意库名要去掉lib前缀和.so后缀。例如如果库文件是libthostmduserapi_se.so编译时应加-lthostmduserapi_se。g -stdc17 -O2 -pthread main.cpp -o tick_recorder -L. -lthostmduserapi_se -lthosttraderapi_se-L.指定在当前目录查找库。运行时依赖确保程序运行时能找到.so库。可以通过设置环境变量LD_LIBRARY_PATH.来实现。注意CTP API内部会创建自己的网络线程。你的主程序在调用Init()之后需要保持运行例如一个while循环或事件等待否则程序会直接退出。常见的做法是使用条件变量或简单的sleep循环。3.2 高效内存管理与数据序列化在OnRtnDepthMarketData回调中参数pDepthMarketData是一个指针。切记这个指针指向的内存是由API内部管理的回调函数结束后可能失效或被复用。因此我们必须立即进行深拷贝。void CTickHandler::OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData) { if (pDepthMarketData) { // 错误做法直接存储指针或引用 // m_queue.push(pDepthMarketData); // 灾难 // 正确做法复制数据到自己的结构体 TickData tick; std::strncpy(tick.InstrumentID, pDepthMarketData-InstrumentID, sizeof(tick.InstrumentID)); tick.LastPrice pDepthMarketData-LastPrice; tick.Volume pDepthMarketData-Volume; // ... 复制其他字段 tick.UpdateTime std::chrono::system_clock::now(); // 添加本地接收时间戳 // 推入无锁队列 m_queue.enqueue(tick); } }这里我定义了一个自己的TickData结构体除了拷贝API的字段强烈建议添加一个本地接收时间戳UpdateTime。因为pDepthMarketData-UpdateTime是交易所时间而网络传输有延迟。本地时间戳对于评估系统延迟、进行精确的时间对齐至关重要。序列化写入文件时为了便于后续读取我通常会在文件开头写入一个魔数Magic Number和版本号在每条记录前写入记录长度。// 文件结构示例 // [文件头: 魔数(4字节) | 版本号(2字节) | 预留(58字节)] 共64字节对齐 // [记录1长度(4字节) | 记录1数据(N字节)] // [记录2长度(4字节) | 记录2数据(N字节)] // ... struct FileHeader { uint32_t magic 0x4B434954; // TICK的十六进制 uint16_t version 1; char reserved[58]; };这样读取程序可以先检查魔数确认文件格式然后根据版本号解析数据通过记录长度可以快速跳转到下一条记录支持流式读取。3.3 文件IO优化与滚动策略直接写文件也有大学问。缓冲写入不要每条tick都调用write系统调用这太慢了。使用std::ofstream并设置合适的缓冲区或者自己维护一个内存缓冲区比如std::vectorchar攒够一定大小如4KB、16KB再一次性写入。std::ofstream的rdbuf()-pubsetbuf()可以设置缓冲区。内存映射文件mmap对于追求极致性能的场景可以考虑mmap。它将文件直接映射到进程的虚拟内存空间对内存的读写即是对文件的读写由操作系统负责页缓存和回写效率极高。但管理起来稍复杂需要注意同步msync和文件大小调整。文件滚动Rolling一个文件不能无限大。需要制定滚动策略。按时间滚动例如每小时或每天生成一个新文件。文件名可以包含日期时间如tick_20231027_0900.dat。按大小滚动当文件超过一定大小如1GB后关闭当前文件创建新文件。混合策略同时考虑时间和大小。这需要消费者线程在写入时定期检查。同步与fsync默认情况下数据写入操作系统页缓存后就返回了并非真正落盘。对于行情数据这种关键信息需要定期强制刷盘。可以使用std::ofstream::flush()刷新流缓冲区和fsync系统调用强制内核将缓存写入磁盘。但fsync很慢不能每条数据都调。一个折中方案是每秒或每写入若干条数据后调用一次flush每分钟或每滚动一个文件时调用一次fsync。4. 完整实现流程与代码框架下面勾勒一个最简化的、但包含核心要素的实现框架。4.1 主程序结构与初始化#include atomic #include thread #include iostream #include “ThostFtdcMdApi.h” // 前向声明 class CTickHandler; class TickStorage; int main() { // 1. 创建API实例 CThostFtdcMdApi* pMdApi CThostFtdcMdApi::CreateFtdcMdApi(./flow/, false, false); if (!pMdApi) { std::cerr “创建API实例失败” std::endl; return -1; } // 2. 创建事件处理与存储对象 auto tickStorage std::make_sharedTickStorage(“./tick_data/”); CTickHandler mdSpi(pMdApi, tickStorage); // 3. 注册事件处理对象 pMdApi-RegisterSpi(mdSpi); // 4. 设置行情前置机地址 pMdApi-RegisterFront(“tcp://180.168.146.187:10131”); // 使用实盘或模拟地址 // 5. 初始化API连接前置机 pMdApi-Init(); // 6. 等待登录成功在OnFrontConnected回调中执行登录 // 登录逻辑应在CTickHandler::OnFrontConnected中实现 // 包括ReqUserLogin并在OnRspUserLogin回调中订阅合约 // 7. 主循环保持程序运行 std::atomicbool running{true}; while (running) { std::this_thread::sleep_for(std::chrono::seconds(1)); // 可以在这里添加一些状态监控或控制命令 } // 8. 退出 pMdApi-Release(); return 0; }4.2 行情回调处理器核心class CTickHandler : public CThostFtdcMdSpi { public: CTickHandler(CThostFtdcMdApi* pApi, std::shared_ptrTickStorage storage) : m_pApi(pApi), m_storage(storage) {} // 连接成功回调 virtual void OnFrontConnected() override { std::cout “行情前置机连接成功开始登录...” std::endl; CThostFtdcReqUserLoginField req{}; std::strcpy(req.BrokerID, “9999”); // 你的经纪商代码 std::strcpy(req.UserID, “000001”); // 你的用户ID std::strcpy(req.Password, “your_password”); int ret m_pApi-ReqUserLogin(req, m_requestId); // 错误处理... } // 登录响应回调 virtual void OnRspUserLogin(CThostFtdcRspUserLoginField *pRspUserLogin, CThostFtdcRspInfoField *pRspInfo, int nRequestID, bool bIsLast) override { if (pRspInfo pRspInfo-ErrorID ! 0) { std::cerr “登录失败: ” pRspInfo-ErrorMsg std::endl; return; } std::cout “登录成功开始订阅合约...” std::endl; // 订阅合约 char* ppInstruments[] {“ag2406”, “rb2405”}; // 合约列表 int ret m_pApi-SubscribeMarketData(ppInstruments, 2); // 错误处理... } // 核心行情回调 virtual void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField *pDepthMarketData) override { if (!pDepthMarketData || !m_storage) return; TickData tick; // 拷贝基础字段 std::strncpy(tick.instrument, pDepthMarketData-InstrumentID, 31); tick.instrument[31] ‘\0’; tick.last_price pDepthMarketData-LastPrice; tick.volume pDepthMarketData-Volume; tick.turnover pDepthMarketData-Turnover; tick.open_interest pDepthMarketData-OpenInterest; // 拷贝时间 std::strncpy(tick.update_time, pDepthMarketData-UpdateTime, 9); std::strncpy(tick.update_millisec, pDepthMarketData-UpdateMillisec, 3); // 添加本地时间戳微秒精度 auto now std::chrono::system_clock::now(); auto us std::chrono::duration_caststd::chrono::microseconds( now.time_since_epoch() ); tick.local_timestamp us.count(); // 将tick数据交给存储管理器异步 m_storage-Push(tick); } // 其他必要的回调如错误通知 OnRspError... private: CThostFtdcMdApi* m_pApi; std::shared_ptrTickStorage m_storage; int m_requestId{0}; };4.3 异步存储管理器实现这是系统的核心负责缓冲和持久化。#include concurrentqueue.h // moodycamel的无锁队列库需单独引入 #include fstream #include chrono struct TickData { char instrument[32]{0}; double last_price{0.0}; int volume{0}; double turnover{0.0}; double open_interest{0.0}; char update_time[9]{0}; // HH:MM:SS char update_millisec[3]{0}; // SSS long long local_timestamp{0}; // 本地接收时间戳微秒 // 其他字段... }; class TickStorage { public: TickStorage(const std::string base_dir) : m_base_dir(base_dir), m_running(true) { // 确保目录存在 std::filesystem::create_directories(base_dir); // 启动消费者线程 m_storage_thread std::thread(TickStorage::StorageWorker, this); OpenNewFile(); } ~TickStorage() { m_running false; if (m_storage_thread.joinable()) { m_storage_thread.join(); } FlushAndClose(); } void Push(const TickData tick) { m_queue.enqueue(tick); } private: void StorageWorker() { std::vectorTickData batch; batch.reserve(1000); // 预分配空间 auto last_flush_time std::chrono::steady_clock::now(); const auto flush_interval std::chrono::seconds(1); while (m_running) { // 尝试从队列中取出最多1000条数据 size_t count m_queue.try_dequeue_bulk(std::back_inserter(batch), 1000); if (count 0) { WriteBatch(batch); batch.clear(); } else { // 队列为空短暂休眠避免空转 std::this_thread::sleep_for(std::chrono::milliseconds(1)); } // 定期刷新缓冲区 auto now std::chrono::steady_clock::now(); if (now - last_flush_time flush_interval) { if (m_ofs) { m_ofs.flush(); } last_flush_time now; CheckFileRolling(); } } } void WriteBatch(const std::vectorTickData batch) { if (!m_ofs.is_open()) return; for (const auto tick : batch) { // 先写入记录长度 uint32_t record_size sizeof(TickData); m_ofs.write(reinterpret_castconst char*(record_size), sizeof(record_size)); // 再写入记录本身 m_ofs.write(reinterpret_castconst char*(tick), record_size); } m_current_size batch.size() * (sizeof(uint32_t) sizeof(TickData)); } void OpenNewFile() { FlushAndClose(); auto now std::chrono::system_clock::now(); auto tt std::chrono::system_clock::to_time_t(now); std::tm tm *std::localtime(tt); char filename[128]; std::strftime(filename, sizeof(filename), “tick_%Y%m%d_%H%M%S.dat”, tm); m_current_file m_base_dir “/” filename; m_ofs.open(m_current_file, std::ios::binary | std::ios::out | std::ios::app); if (!m_ofs) { std::cerr “无法打开文件: ” m_current_file std::endl; return; } // 写入文件头 FileHeader header{}; m_ofs.write(reinterpret_castconst char*(header), sizeof(header)); m_current_size sizeof(header); std::cout “已创建新数据文件: ” m_current_file std::endl; } void CheckFileRolling() { const size_t MAX_FILE_SIZE 1024 * 1024 * 1024; // 1GB if (m_current_size MAX_FILE_SIZE) { std::cout “文件大小超过 ” MAX_FILE_SIZE “ 字节开始滚动...” std::endl; OpenNewFile(); } } void FlushAndClose() { if (m_ofs.is_open()) { m_ofs.flush(); // 可选调用 fsync 确保数据落盘 // int fd fileno(m_ofs); // fsync(fd); m_ofs.close(); } } std::string m_base_dir; std::string m_current_file; std::ofstream m_ofs; size_t m_current_size{0}; moodycamel::ConcurrentQueueTickData m_queue; // 无锁队列 std::thread m_storage_thread; std::atomicbool m_running; };5. 常见问题排查与性能调优即使代码写完了在实际运行中还是会遇到各种问题。这里列几个典型的坑和解决办法。5.1 连接与登录失败问题程序启动后OnFrontConnected没触发或者登录后OnRspUserLogin返回错误。排查网络可达性先用telnet或nc命令测试前置机地址和端口是否能通。防火墙检查本地防火墙或云服务商安全组是否放行了对应端口。经纪商参数仔细核对BrokerID、UserID、Password模拟盘和实盘是不同的。密码可能是需要期货公司提供的认证码。API版本确认使用的API版本如v6.6.9与期货公司柜台版本匹配。流文件目录CreateFtdcMdApi的第一个参数是流文件存储目录确保该目录存在且有写权限。5.2 收不到行情数据问题登录成功了但OnRtnDepthMarketData一直没有被调用。排查合约代码确认订阅的合约代码格式正确且是当前有效的合约非主力合约可能无行情。合约代码通常像rb2410螺纹钢2410合约。交易所确认你订阅的合约所属交易所与你登录的行情前置机是否支持。有的前置机只支持特定交易所。订阅时机必须在登录成功回调 (OnRspUserLogin) 之后才能订阅合约。在连接成功 (OnFrontConnected) 时订阅会失败。流量控制部分API有订阅合约数量的限制检查是否超限。5.3 程序运行缓慢或内存增长问题程序运行一段时间后变卡或者内存占用持续上升。排查与优化生产者过快消费者过慢这是最常见原因。检查存储线程的写入性能。是否每条数据都调fsync文件缓冲区是否太小可以尝试增大消费者线程的批量处理大小。使用更快的存储介质如SSD。将存储逻辑移到单独的进程通过共享内存或本地Socket传递数据避免存储I/O阻塞主线程。内存泄漏确保没有在回调函数中动态分配内存却忘记释放。使用valgrind工具检测。valgrind --leak-checkfull ./tick_recorder锁竞争如果使用了带锁的队列在高频行情下锁竞争会非常激烈。务必使用无锁队列。日志输出在OnRtnDepthMarketData中打印日志到控制台 (std::cout) 是性能杀手会极大拖慢速度。生产环境应关闭或使用异步日志库。5.4 数据文件损坏或读取错误问题存储的文件无法被后续程序正确读取。解决写入原子性确保每次write操作的数据是完整的。我们采用“长度数据”的格式即使程序崩溃读取时也可以通过长度字段跳过损坏的记录。定期同步如前所述定期调用flush()和fsync()但平衡好性能和数据安全。文件结尾程序正常退出时应在文件末尾写入一个特殊的结束标记如长度为0的记录方便读取程序识别文件结束。校验和对于数据准确性要求极高的场景可以在每条记录后增加一个CRC32校验和。写入时计算并写入读取时验证。5.5 性能监控指标一个健壮的系统需要可观测性。建议增加简单的监控队列深度定期输出无锁队列的size_approx()如果持续增长说明消费者跟不上生产者。处理延迟在TickData中记录本地接收时间戳在存储线程中计算当前时间与接收时间戳的差值可以统计延迟分布。吞吐量统计每秒处理的tick数量。文件大小监控当前数据文件的大小和滚动频率。这些指标可以定期打印到日志或者通过简单的HTTP服务暴露出来方便监控系统状态。