1. 背景2024年10月中旬公司打算切换美股行情数据源在此之前使用的是Polygon(已改名为Massive)Polygon的优点有两个1.价格便宜每月2000$ 2.接入简单通过ws和rest接口即可获取所有数据无需接入方自己计算和存储数据。缺点是数据不准经常出现和业内主流券商对不上的情况以AAPL为例 其它券商展示价格是298.55我们平台展示298.43。在金融领域数据是核心基础数据不准其它一切搞的再好都是白搭因为用户不会信任长此以往用户自然流失。在此背景下公司决定从业内顶级数据供应商伦敦交易所集团(LSEG)采购数据LSEG数据的优点有两个1.数据实时性强 2.数据准确性高缺点也有两个1. 成本高几倍于Polygan。2. 接入复杂因为只买了实时行情数据没有全量历史数据需要业务侧自己基于实时行情进行计算与存储。2. 任务接入lseg实时行情数据并全量替换现有接口。3. 难点与解决方案3.1 难点1·海量数据实时接收与清理平均每个交易日接收80M消息(极端行情100M)平均200K/min, 平均3.5K/s峰值1.2M/min, 20K/s。接收速度慢会导致ws连接断开数据丢失。3.2 难点2·海量数据实时计算与存储美股Ticker总数10K需要实时计算min1, min5, min10, min30, min60, day等等。日均新增K线5M平均每分钟50K因为K线窗口都是固定的峰值K线生成速度也是50K。这里有两个要求1. 计算性能要快处理慢会导致数据丢失或者数据延迟进而影响所有依赖的业务。2.存储要快因为存在时间热点问题存储系统要支持高并发写入。3.3 难点3·数据准确性保证行情数据不准的来源有两个1. K线生成公式没有统一的标准各家处理规则不一致成交数据包含不同的类型比如普通成交、大额成交、碎股成交。有些券商统计这些数据有些券商没有。因此光自己计算还不够还需要和外界比对。2. 随着时间的迁移历史价格会因为公司分红、拆并股等事件变更需要及时更新否则误差会越来越大。3.4 难点1的解决方案初版方案是单线程串行处理流程见图1图1这个版本存在的问题是经常因为接收速度慢导致ws连接中断。在此基础上优化的方案是读写分离和Redis 多线程类似将数据解析、处理与数据接收分离出来读协程只负责数据接收收到数据之后写入本地队列即可。由本地worker 异步消费队列。如图2图2这一版的问题大幅减少了ws连接中断的问题但还是因为本地队列发送慢导致数据丢失问题。因此进一步优化了数据发送流程改为多协程异步批量发送。如图3图3除了上述的处理还有其他一些优化1. 开启ws链路压缩减少数据量。2. 调优套接字接收缓冲区减少系统调用次数。3. 开启kafka压缩, 调整ACK参数3.5 难点2的解决方案针对难点2需要实时生成不同时间粒度K线数据如min1, min5, min10, min30, min60, day等等。Ticker总数10K因为K线窗口固定因此同一粒度K线同一时刻生成存在写入热点问题。另外这里要求计算K线要性能快计算慢会导致本地队列移除数据包丢失从而影响K线的准确度。每日新增5M记录K线包含(OHLCVTickerTimestampID), 单条记录占用字节20*58*232 148字节。存储空间近 700MB一个月15GB先说一下存储的选型典型的存储包含以下几种内存数据库、磁盘数据库、时序数据库内存数据库典型的如Redis写入速度最快但是成本最高且无法缓存全量数据需要业务侧做聚合处理比较麻烦。磁盘数据库如MySQL顺序写入速度还行不支持高并发写入存储成本低。需要分表。时序数据库如AWS timestream, 写入速度很快存储成本低IO费用高。在我们的业务场景中K线是基础数据各种各样的业务都依赖它如回测系统、策略系统等等很多业务场景都需要读取大量的行情数据如果基于IO次数收费成本过高。基于性能、成本综合考虑我们选用了MySQL 作为存储。针对数据量特别大的粒度如min1, min5使用一致性哈希分表。初版方案是创建10K协程每个ticker一个本地维护一个行情消息队列收到消息后串行计算不同粒度K线当K线窗口结束之后写入到数据库并创建新的K线流程如图4图4这一版可以用但是存在两个问题一个是计算慢导致本地队列溢出丢数据另一个是时间热点不同粒度的K线在同一个时间点结束导致瞬时写入负载大造成数据库抖动。对于计算慢的问题一种思路是并行计算开多个协程每个K线粒度对一个协程N个粒度创建N个协程。这种方案会导致协程数量爆炸因为ticker总数已经是10K再乘以N虽然golang 协程创建开销小但是一旦协程数量多cpu调度开销就会急剧上升。因此不可行。后来想到了一个新颖的方案聚合计算。K线原本的计算逻辑是这样将当前窗口的第一个成交价作为open, 最后一个成交作为close中间的成交数据用来更新high,low累加每一个事件的volume。如果收到N个成交事件K个粒度则总的计算量是NK,但如果采用聚合计算的方式只计算min1, 仅当min1结束之后再使用这根已结束的min1更新其它粒度如此一来计算量就减少K倍只需计算一次大幅减少计算量。另外针对时间热点问题比如在T60min结束时刻所有日内粒度都结束当前这个窗口期需要写入db每个粒度10KN个粒度写入量为N*10KMySQL扛不住太高的并发写入。这里采用的优化方向有三个一个是对K线做优先级排序业务层面使用最多的粒度写入尽可能实时对于优先级不高的可以延迟写入针对不同粒度引入了随机的写入延迟避免时间热点问题。另一个优化是并发写入。比如min1原本写10K次如果改为每次写入100条则只需要写入100次。随机延迟和解决缓存雪崩问题方案一样都是避免同一时刻触发。除此之外还有一些其它方案可以解决时间延迟问题比如写入kafka削峰限流不过代价是数据更新延迟。还有一个优化方案是减少写入的粒度由业务层做聚合处理比如取消写入min10,min30,改为业务侧基于min5聚合处理这样也极大缓解写入量代价是读取的时候稍微慢一点。优化后的方案如图5:图53.6 难点3的解决方案实时K线准确度问题虽然lseg不提供全量历史数据但提供2年最新数据。对于实时生成的K线准确度问题盘中及盘后会定时异步从lseg拉取最新的K线增量覆盖数据库 这里不是无脑覆盖而是先检查之前有没有更新过如果有更新过则跳过本次更新减少数据库写入量代价是需要缓存已更新数据。历史K线准确度问题市面上没有哪一家数据提供商能够提供完全准确无误数据我们这里采用的是多源比对取共识。每周进行一次全量检查。优先和Polygon对比,如果发现存在不一致则进一步检查Yahoo, Nasdaq, Stooq。取这三家有共识的数据比如如果有两家数据一样则取一家数据覆盖更新如果三家都不一致则告警人工处理。4. 结果1. lseg ws平均接收延迟1ms, max 25ms。2. kafka平均发送延迟10msmax50ms。5. 总结关于高性能可以围绕数据传输、计算、存储三个角度来进行优化。数据传输层面考虑编码方式(文本或者二进制如json vs protobuf)传输协议选择(udp, tcp), 数据压缩。计算层面优化的方式包括并发、异步、缓存、预处理、批量等等。存储层面包括存储引擎选择(B树 vs LSM-Tree), 批量、缓存等等。另外上述结果还可以进一步优化1. 创建多个ws链接提升并发处理量。代价是成本增加lseg按照连接数收费。2. 采用pb编码数据量估计可以减少一半。我们没有使用是因为当前lseg仅在java sdk提供pb编码和我们技术栈不一样维护困难。6. 收获这个项目除了技术难点之外还存在不少非技术难点1. 时间紧任务重开发测试上线只有20天一旦延期就需要同时订阅两个数据源损失2000$.2. 调试时间短因为交易所和LSEG双方都需要合规审核之后才能够拿到真正的行情数据而审核结束距离上线只有不到一周时间在一周的时间内有各种各样未考虑到场景需要处理。3. 无参考资料无论是内部还是外部都没啥经验参考只能一步一个脚印摸索。在那段开发时期起早贪黑早7点到晚24点每周工作6.5天最终顺利交付。给我的收获就是软件开发领域没有啥是完不成的如果有那就再加点时间。