小 T 导读:如果我们要做行情 tick 数据的存储,怎样的数据结构查找起来才会比较快?在加入 TDengine 之前,本文作者丁博在弘源泰平量化投资做量化工程师,曾经遇到过这一类存储行情 tick 数据的问题,本文会就此问题进行详细的技术解读。
本文将以标准 CTP 行情接口(http://www.sfit.com.cn/5_1_DocumentDown.htm)为例,假设行情结构为 CThostFtdcDepthMarketDataField(https://mckelv.in/python-ctp-deps/struct_c_thost_ftdc_depth_market_data_field.html),展开说明。
内存存储方案
如果你的需求仅仅是盘中实时分析,且监控的 Instrument(CTP 接口对现货、期货、期权等合约的统称, 以下简称【合约】) 总数不多,则可以直接使用内存存储。通常只有超高频交易系统才必须这么做。内存存储也有很多可选方案,其中有两大方案较为通用。
两级 map 方案
第一级 map 的类型为 std::unordered_map,键为 InstrumentID, 值为第二级 map 的指针。第二级 map 的类型为 std::map,键为行情时间戳,值为行情结构体。(注:行情时间戳需要根据 UpdateTime 和 UpdateMillisec 两个字段构造一个类型为 long 的毫秒值)。 std::unordered_map 底层依赖的数据结构是哈希表,按 key 索引速度是最快的。std::map 底层的数据结构是二叉树搜索树,可以严格按照 key 的大小顺序迭代全部或某一段数据。 总体而言这个数据结构的优势是: 快速查找某个合约某个时间点或某个时间段返回的行情。这是后续做交易信号计算的基础。
#include "ThostFtdcUserApiStruct.h"
#include "ThostFtdcUserApiDataType.h"
#include <map>
#include <unordered_map>
using namespace std;
int main()
{
unordered_map<TThostFtdcInstrumentIDType, map<long, CThostFtdcDepthMarketDataField>*> tickData;
}
map + array
由于每种合约每天的标准行情 tick 总数都是固定的(个别交易所除外),因此我们可以提前初始化好一个数组来存行情。按每秒 2 个 tick 算(500 毫秒一个点),标准行情的长度可能是 28800。当收到行情通知时,行情时间距离哪个标准 tick 点最近就归为哪个 tick。比如行情时间是 9 点 50 分 20 秒 133 毫秒,那么可以当作 9 点 50 分 20 秒 0 毫秒的行情。如果出现前后两个 tick 时间大于 500 毫秒的情况,那就还需要补全中间空缺的行情,相当于边收行情边做标准化操作。这样做的优势是:
- 交易策略通常会依赖标准化的行情计算交易信号,收行情和标准化并作一步会更节省时间。
- 可以直接用数组下标索引对应时间的行情,查找的时间复杂度为 O(1)。
#include "ThostFtdcUserApiStruct.h"
#include "ThostFtdcUserApiDataType.h"
#include <unordered_map>
#include <array>
using namespace std;
int main()
{
unordered_map<TThostFtdcInstrumentIDType, array<CThostFtdcDepthMarketDataField, 28800>> tickData;
}
持久化存储方案
无论是否做超高频交易,持久化存储行情都是有必要的。通常持久化存储为的是进行盘后复盘分析, 因为在大数据量下,传统的存储方案(MongoDB、MySQL、直接存文件等等)很快就会遇到性能瓶颈(无论是读还是写),不适合做盘中的计算。近年来,时序数据库(Time Series Database)异军突起,使得盘中盘后使用一种存储方案成为可能。特别是像 TDengine 这样带有缓存功能、消息队列功能和集群功能的时序数据库,用来存行情是非常合适。下面我将以 TDengine Database 为例为大家介绍持久化存储方案。
下载 TDengine Database Server
在下载阶段,不同的系统使用的安装包也有所不同,Ubuntu 系统用 deb 包, CentOS 系统用 RPM 包。下载地址为: All Downloads – TDengine。
安装并启动
Ubuntu
sudo dpkg -i TDengine-server-2.4.0.7-Linux-x64.deb
CentOS
sudo rpm -ivh TDengine-server-2.4.0.7-Linux-x64.rpm
安装成功后,如何启动 TDengine Database 的提示信息就会自动弹出,照着操作就可以。
建行情表
由于所有行情的结构都是一样的,因此只需要一张超级表进行行情建表即可,其中每个合约对应一张子表,InstrumentID 作为子表名,交易所代码作为一个行情标签。为了方便演示,下面的示例只包含了 4 个行情字段:
- 进入 taos 命令行
bo@RDBB:~$ taos
Welcome to the TDengine shell from Linux, Client Version:2.4.0.12
Copyright (c) 2020 by TAOS Data, Inc. All rights reserved.
- 执行下面的语句
create database marketdata;
use marketdata;
create stable tick(
ts timestamp,
updatetime binary(9),
updatemillisec int,
askprice1 double,
bidprice1 double,
askvolume1 int,
bidvolume1 int
) tags (exchangeid binary(9));
- 查看表结构
taos> desc tick;
Field | Type | Length | Note |
=================================================================================
ts | TIMESTAMP | 8 | |
updatetime | BINARY | 9 | |
updatemillisec | INT | 4 | |
askprice1 | DOUBLE | 8 | |
bidprice1 | DOUBLE | 8 | |
askvolume1 | INT | 4 | |
bidvolume1 | INT | 4 | |
exchangeid | BINARY | 9 | TAG |
Query OK, 8 row(s) in set (0.000378s)
写入行情
#include "ThostFtdcUserApiStruct.h"
#include "ThostFtdcUserApiDataType.h"
#include "taos.h"
#include "taoserror.h"
#include <iostream>
#include <sstream>
using namespace std;
void insertTickData(TAOS* taos, CThostFtdcDepthMarketDataField &tick) {
stringstream sql;
// 会自动创建子表tick.InstrumentID
sql << "insert into " << tick.InstrumentID << " using tick tags("
<< tick.ExchangeID << ") values(now, '" << tick.UpdateTime << "', "
<< tick.UpdateMillisec << "," << tick.AskPrice1 << "," << tick.BidPrice1
<< "," << tick.AskVolume1 << "," << tick.BidVolume1 << ")";
TAOS_RES *res = taos_query(taos, sql.str().c_str());
if (res == nullptr || taos_errno(res) != 0) {
cerr << "insertTitckData failed," << taos_errno(res) << ", " << taos_errstr(res) << endl;
}
}
int main()
{
TAOS *taos = taos_connect("localhost", "root", "taosdata", "marketdata", 6030);
// 构造测试数据
CThostFtdcDepthMarketDataField tick;
strcpy_s(tick.InstrumentID, "IH2209");
strcpy_s(tick.UpdateTime, "14:10:32");
strcpy_s(tick.ExchangeID, "DEC");
tick.UpdateMillisec = 500;
tick.AskPrice1 = 123.8;
tick.BidPrice1 = 123.4;
tick.AskVolume1 = 10;
tick.BidVolume1 = 9;
// 写入测试数据
insertTickData(taos, tick);
taos_close(taos);
}
查询最新的行情
TDengine 对每个表的最新数据都有缓存功能,无需再读磁盘,使用 last 函数就能快速获取。
#include "ThostFtdcUserApiStruct.h"
#include "ThostFtdcUserApiDataType.h"
#include "taos.h"
#include "taoserror.h"
#include <string>
#include <iostream>
using namespace std;
CThostFtdcDepthMarketDataField* getLastTick(TAOS* taos, const char* instrumentID) {
string sql("select last(*) from ");
sql += instrumentID;
TAOS_RES* res = taos_query(taos, sql.c_str());
if (res == nullptr || taos_errno(res) != 0) {
cerr << "getLastTick failed," << taos_errno(res) << ", " << taos_errstr(res) << endl;
return nullptr;
}
TAOS_ROW row = taos_fetch_row(res);
if (row == nullptr) {
return nullptr;
}
CThostFtdcDepthMarketDataField* tick = new CThostFtdcDepthMarketDataField();
//int64_t ts = *((int64_t*)row[0]);
memcpy(tick->UpdateTime, row[1], 9);
tick->UpdateMillisec = *(int*)row[2];
tick->AskPrice1 = *((double *)row[3]);
tick->BidPrice1 = *((double*)row[4]);
taos_free_result(res);
return tick;
}
int main() {
TAOS* taos = taos_connect("localhost", "root", "taosdata", "marketdata", 6030);
CThostFtdcDepthMarketDataField* tick = getLastTick(taos, "IH2209");
cout << "askPrice1=" << tick->AskPrice1 << " bidPrice1=" << tick->BidPrice1 << endl;
delete tick;
taos_close(taos);
}
以上两个示例程序,展示了写入和查询的方法。结合 TDengine 内置的查询函数和按窗口聚合功能,可实现更多功能,比如:
- 使用 MAX、 FIRST、 MIN、 LAST 四个 SQL 函数计算 K 线上高、开、低、收四个价位。
- 使用 INTERVAL 和 SLIDING 查询子句和 AVG 函数计算移动均价。此处不再给出具体示例,可参考官方文档。
从实际业务出发的实践经验分享
除了上述内容外,TDengine Database 还有非常丰富的分析函数,如果你感兴趣的话建议参考官方文档。此外,在 TDengine 的实际应用中,也有很多客户的实践是关于量化投资场景中的数据处理。
以同花顺为例,其每天都需要接收海量交易所行情数据,以确保行情数据的数据准确,但由于该部分数据过于庞大,而且使用场景颇多,因此每天会产生很多的加工数据,在组合管理(PMS)上还会使用到历史行情数据。之前他们采用的是 Postgres+LevelDB 作为数据的存储方案,但仍旧痛点频发,随后通过对数据流、行情获取模块的分析,发现目前主要存在以下两个亟需解决的问题:
- 依赖多,稳定性较差:PMS作为多品种的投后分析服务, 需要使用到各种日线数据、当天实时行情数据、当天分钟数据等,在数据获取方面需要依赖Http以及Postgres、LevelDB等数据库。过于多的数据获取链路会导致平台可靠性降低,同时依赖于其他各个服务,导致查询问题过于复杂。
- 性能不能满足需求: PMS作为多品种投后分析,在算法分析层面需要大量的行情获取,而且对行情获取的性能也有较大的要求,当前所有行情会占据大量分析的性能。
从业务发展的角度来讲,存储方案的改造迫在眉睫,之后同花顺开始对 ClickHouse、InfluxDB、TDengine 等数据存储方案进行调研。由于行情数据是绑定时间戳的形式,所以显然时序数据库更适用于这个业务场景,在 InfluxDB 和 TDengine 之间,由于 TDengine 的写入速度远高于 InfluxDB,且集群版开源,同时还支持包含 C/C++、Java、Python、Go 和 RESTful 在内的多种数据接口,因此成为同花顺的最终选用方案。
改造之后的性能效果提升还是非常明显的,下图是同花顺做的一张改造前后性能对比图,可以更为直观地感受到效果提升:
同时改造后,稳定性也显著增强,改造前调用数据情况共 40W 次,共出现 0.01% 的异常,改造后出现异常降低至 0.001%。
在 TDengine Database 官网的 Case 合集中,还有弘源泰平量化、同心源基金等几篇聚焦投资量化场景下数据处理难题的客户案例,由于篇幅所限,便不在此一一列举了,有需要的朋友可以去官网查找文章进行参考。如果还有投资量化场景下其他的数据处理难题,也欢迎在文章下方进行留言,我们后续可以加微信进行详细讨论和沟通。