1. 金融数据服务从零搭建的完整思路1.1 这个项目到底在做什么“financial-services”这个标题看起来很大实际上它指向的是一个非常具体的工程问题如何把分散在各处的金融数据源整合成一套稳定、可查询、可扩展的后端服务。我在过去几年里参与过三个类似的项目从最早的“一个脚本跑天下”到后来的微服务架构踩过的坑足够写一本小册子。这篇文章不讲空泛的架构理论只讲我在实际搭建金融数据服务时用到的方案、遇到的真实问题和解决办法。金融数据服务的核心需求通常包括几块行情数据的实时获取与缓存、历史数据的批量导入与查询、用户自选股的增删改查、以及围绕这些数据衍生出的简单计算比如涨跌幅、均线。听起来不复杂但真正做起来数据源的稳定性、接口的限流、时区处理、复权计算这些细节会把你折磨得够呛。这套内容适合有一定后端基础、想进入金融科技领域的开发者也适合正在做个人量化项目、需要一套靠谱数据服务的朋友。1.2 为什么选择这样的技术组合我在技术选型上遵循一个原则能用简单方案解决的绝不引入复杂中间件。早期我试过用 Kafka 做行情数据的消息队列结果发现对于日频和分钟频数据来说完全是杀鸡用牛刀运维成本还高得离谱。后来调整为 PostgreSQL 加 Redis 的组合配合一个轻量的任务调度器反而稳定跑了两年多。具体来说PostgreSQL 负责持久化存储它的 JSONB 字段类型特别适合存放结构不固定的行情快照Redis 负责热点数据的缓存和临时队列比如最新报价、当日涨跌幅排行这类高频读取的数据。任务调度用 APScheduler 就够了没必要上 Celery 加 RabbitMQ 那一套除非你的数据源有几十个并且需要分布式抓取。注意金融数据对时间精度要求极高所有时间字段统一用 UTC 存储展示时再转本地时区。我见过太多项目因为时区问题导致K线对不上排查起来非常痛苦。1.3 整体架构的分层设计我把整个服务分成四层数据采集层、数据清洗层、存储层、接口层。数据采集层负责从各个数据源拉取原始数据这一层要处理重试、限流、断点续传数据清洗层做格式统一、异常值过滤、复权计算存储层管理数据库表结构和缓存策略接口层对外提供 RESTful API 和 WebSocket 推送。这样分层的好处是每一层可以独立测试和替换。比如某天某个数据源挂了我只需要在采集层加一个新的适配器上层完全不用动。清洗层的复权算法如果发现有问题也只影响这一层不会波及接口的稳定性。2. 数据采集与清洗的核心细节2.1 数据源的选择与适配策略金融数据源大致分三类交易所官方接口、第三方数据服务商、公开网页数据。官方接口最可靠但通常有权限门槛第三方服务商数据全但需要付费且有限流公开网页数据免费但结构不稳定随时可能改版。我的做法是至少接入两个数据源做交叉验证。主数据源用第三方服务商的API备用数据源用公开接口。每天收盘后跑一次对账任务如果两个源的数据差异超过阈值就触发告警。这个机制帮我发现过好几次数据源单方面出问题的情况。适配器模式在这里特别有用。我定义了一个统一的DataSourceAdapter抽象类包含fetch_quotes、fetch_history、fetch_fundamentals三个核心方法。每个具体的数据源实现这个接口上层调用时完全不关心底层是哪个源。这样新增数据源的成本极低基本上半天就能接完一个。class DataSourceAdapter(ABC): abstractmethod def fetch_quotes(self, symbols: list[str]) - list[Quote]: pass abstractmethod def fetch_history(self, symbol: str, start: date, end: date) - list[Bar]: pass2.2 数据清洗中的复权处理复权是金融数据清洗里最容易出错的地方。简单说股票除权除息后价格会跳空如果不做复权处理计算出来的收益率完全是错的。前复权是以当前价格为基准调整历史价格后复权是以历史某点为基准调整后续价格。做回测通常用前复权做长期收益分析用后复权。我的实现方式是从数据源获取原始的除权除息事件表然后根据事件表逐日计算复权因子。这里有个细节不同数据源的除权事件字段格式差异很大有的用“每10股送X股派Y元”有的直接给复权因子。我统一转换成“每股送转比例”和“每股派息金额”两个标准字段再计算复权因子。实操心得复权计算一定要用 Decimal 类型不要用 float。我早期用 float 计算遇到大比例送转时精度丢失导致复权后的价格和主流软件对不上排查了一整天才发现是浮点精度问题。2.3 异常值检测与处理规则金融数据里的异常值主要有几种价格为零或负数、单日涨跌幅超过合理范围、成交量突然放大百倍、时间戳重复或乱序。这些异常如果不处理会直接污染下游的计算结果。我的处理规则是这样的价格为零或负数的记录直接丢弃并记录日志单日涨跌幅超过30%的标记为可疑但不自动删除而是进入人工审核队列成交量超过过去20日均值50倍的同样标记可疑时间戳重复的保留最新一条乱序的按时间重新排序。这里要特别说一下不要轻易自动修正数据。我见过有项目把涨跌幅超过10%的数据自动截断到10%结果把真实的涨停板数据给改坏了。标记可疑、人工确认虽然麻烦一点但数据质量有保障。3. 存储方案与接口实现3.1 数据库表结构设计要点行情数据的表结构设计有个经典难题是按股票分表还是按时间分表。按股票分表查询单只股票的历史数据很快但跨股票查询某一天的所有数据就很慢按时间分表则相反。我的选择是按时间范围分区同时在股票代码上建索引这样两种查询都能兼顾。具体来说日线数据按年分区分钟线数据按月分区。PostgreSQL 的分区表功能在这里非常好用查询时分区裁剪能大幅减少扫描的数据量。另外我单独建了一张symbols表存放股票的基本信息包括代码、名称、上市日期、退市日期、所属行业等行情表通过外键关联。表名用途分区策略预估数据量daily_bars日线行情按年分区每年约500万行minute_bars分钟线行情按月分区每月约2000万行symbols股票基础信息不分区约1万行adjustments复权事件不分区约10万行3.2 缓存策略与热点数据识别Redis 在金融数据服务里主要承担三个角色最新报价缓存、排行榜缓存、接口限流计数器。最新报价的过期时间设置得很短通常5到10秒因为行情变化快排行榜数据每分钟更新一次过期时间设60秒限流计数器按用户ID和接口路径组合键用滑动窗口算法。热点数据的识别我用了简单的LFU思路在Redis里记录每个股票代码的访问次数每小时统计一次访问频率最高的前100只股票自动加入预热列表服务启动时主动加载到缓存。这个策略让缓存命中率从最初的60%提升到了92%左右。注意Redis 缓存一定要设置合理的过期时间并且做好缓存穿透的防护。我遇到过有人恶意请求不存在的股票代码导致每次请求都打到数据库差点把数据库拖垮。后来加了布隆过滤器才解决。3.3 RESTful 接口设计与版本管理接口设计我遵循几个原则URL 用名词复数、版本号放在路径里、分页参数统一、错误码规范。比如获取日线数据的接口是GET /api/v1/bars/daily?symbol000001start2024-01-01end2024-12-31。版本管理用路径版本/api/v1/、/api/v2/这样。新版本上线后旧版本至少保留半年给调用方足够的迁移时间。错误码我定义了一套简单的规则400开头是客户端错误500开头是服务端错误具体错误信息放在响应体的message字段里。{ code: 200, message: success, data: { symbol: 000001, bars: [ {date: 2024-01-02, open: 10.5, high: 10.8, low: 10.3, close: 10.6, volume: 1200000} ] } }4. 常见问题与排查技巧实录4.1 数据源限流与重试机制数据源限流是最常见的问题。大部分免费接口都有每分钟调用次数限制超了就直接封IP。我的应对策略是令牌桶限流加指数退避重试。令牌桶控制请求速率确保不超过数据源的限制重试时每次等待时间翻倍最多重试5次。具体实现上我用 Redis 的INCR和EXPIRE做一个简单的分布式令牌桶。每个数据源一个桶桶的容量和补充速率根据数据源的限制配置。请求前先尝试获取令牌获取不到就等待或降级到备用数据源。def acquire_token(source: str, capacity: int, refill_rate: float) - bool: key fratelimit:{source} now time.time() pipe redis.pipeline() pipe.get(key) pipe.ttl(key) current, ttl pipe.execute() # 省略具体的令牌桶计算逻辑 return True4.2 时区与交易日历的坑时区问题我在前面提过这里再展开说一个更隐蔽的坑交易日历。不同市场的交易日不同A股有春节、国庆长假美股有感恩节、圣诞节。如果你的服务支持多市场必须为每个市场维护独立的交易日历。我的做法是维护一张trading_calendar表字段包括市场代码、日期、是否交易日。每年初从各交易所官网获取下一年度的交易日历并导入。计算均线、涨跌幅等指标时严格按交易日历取前N个交易日而不是简单地减N天。踩坑记录有一次做港股回测没注意港股在台风天会临时休市结果那天的数据缺失导致整个回测结果偏差很大。后来在交易日历里增加了“临时休市”字段支持手动调整。4.3 常见问题速查表问题现象可能原因排查方法解决方案接口响应慢缓存失效或数据库慢查询查看Redis命中率和慢查询日志预热缓存、加索引数据对不上复权处理错误或时区问题对比原始数据和复权后数据检查复权因子和时区设置采集任务失败数据源限流或网络问题查看采集日志和重试记录调整限流参数、切换备用源内存占用高缓存数据过多或内存泄漏监控Redis内存和进程内存调整过期时间、排查代码数据重复任务重复执行或幂等性缺失检查任务调度记录加唯一约束、实现幂等写入4.4 监控与告警的实践经验监控这块我走过弯路。早期只监控了服务是否存活结果数据采集停了三天都没发现。后来完善了监控体系主要监控几个指标数据采集任务的执行状态和耗时、各数据源的成功率、数据库连接数和慢查询数量、Redis内存使用率和命中率、接口的P99响应时间。告警渠道用邮件加即时通讯工具不同级别的告警走不同渠道。数据采集失败这种影响大的直接发即时通讯工具接口响应时间轻微上升这种发邮件就行。告警阈值不要设得太敏感否则会被大量误报淹没最后反而不看了。我在实际运维中发现每周做一次数据质量报告非常有必要。统计各数据源的可用率、数据缺失率、异常值比例形成趋势图。这样能在问题变大之前发现苗头比如某个数据源的可用率连续下降就可以提前准备切换方案。4.5 性能优化的几个实用技巧数据库层面批量插入比单条插入快一个数量级。我用COPY命令导入历史数据100万行数据大概30秒就能搞定用INSERT逐条插的话要十几分钟。查询层面覆盖索引能显著减少回表比如查询日线数据时只取日期和收盘价就在(symbol, date, close)上建联合索引。接口层面分页查询一定要加最大页大小限制。我见过有人请求page_size100000直接把数据库连接占满。现在我的接口默认每页100条最大1000条超过就返回错误。另外Gzip压缩对JSON响应效果很好行情数据的压缩率通常能到80%以上能省不少带宽。最后分享一个我用了很久的小技巧给每个数据源的健康状态打分。每次采集成功后加分失败后减分分数低于阈值就自动切换到备用源。这个简单的机制让我的服务在过去两年里几乎没有因为数据源问题中断过。分数每天重置一次避免某个源因为历史问题一直被冷落。
