做一个名为 financial-services 的项目听起来像是要搭建一套银行核心系统实际上我做的是一个面向小团队投研场景的数据服务中间层。过去大半年我把它从一个临时脚本集合磨成了现在每天稳定扛住几十万次调用的小平台也踩了不少只会在真实业务里遇到的坑。这篇文章不聊空洞的金融科技概念就聊聊这个项目从需求分析、架构设计到接入行情源、做指标计算和风控检查的完整过程以及哪些设计决定帮我省了后来大半年的维护时间。如果你正准备在内部搭一套类似的金融数据服务或者你只是想知道一个金融服务项目里真正麻烦的地方在哪这篇应该能给你一些参考。1. 项目背景这个金融服务项目到底要解决什么问题1.1 内部项目里的数据孤岛是怎么一点一点堆出来的最开始的时候团队里其实并没有人提出我们要做一个中台大家只是各做各的项目。有人做一个股票筛选工具有人做持仓分析有人做定投回测。每个项目都有自己的数据获取方式有的是写爬虫去公开网页抓有的是天天从第三方数据平台手动导出 Excel还有的直接调某个行情供应商的接口。结果就是同一家公司三个项目里算出来的市盈率可以完全不一样。我还记得有一次开周会负责筛选工具的同事说某只股票 PE 只有 12 倍很便宜负责持仓分析的同事立刻反驳说他系统里显示的是 18 倍根本不便宜。两个人吵了半天最后发现一个人用的是静态市盈率按最近一期已披露的年报算另一个人用的是 TTM 滚动市盈率按最近四个季度加总算。再往下追连净利润的口径都不一样有人用了归母净利润有人用了包含少数股东权益的合并净利润。这个场景其实非常典型。团队一旦超过两三个人、项目一旦超过两三个数据逻辑就会出现无法收敛的分叉。你很难让每个项目组都自觉维护同一套口径定义也不应该靠人肉同步来保证一致性。更稳妥的做法是把数据获取、数据清洗、指标计算这些公共能力抽出来做成一个独立服务让所有下游项目都从这套服务里拿数据和算好的指标。这就是 financial-services 这个项目最初的形态。1.2 为什么不是继续买数据服务而是自研一个中间层有人可能会问市面上那么多现成的金融数据服务直接订阅不就行了吗我当时也反复评估过这个问题。外部成熟方案确实能解决一部分问题但有两个硬伤是绕不开的。第一个硬伤是定制化能力弱。我们的策略里有很多内部约定比如回撤必须统一用后复权价格计算PE 必须区分静态和 TTM 两种口径规则引擎要允许运营人员随时调整阈值。这些需求在外部产品里往往无法满足要么等厂商排期要么自己导出来二次加工最终还是回到乱糟糟的局面。第二个硬伤是成本叠加。多个项目分别订阅同一家服务账号要按项目开费用也要按项目算。与其每个项目都买一份不如由中间层统一接入下游项目只跟内部 API 打交道数据源订阅成本只发生一次。不过我必须强调自研不是万能的。如果团队只是做一次性分析或者只有一个小工具需要行情数据直接买现成服务更划算。只有当你能看到结构性的复用需求比如至少两个项目都在消费同一类数据并且后续还有第三个、第四个项目要接入自研中间层才值得启动。这个边界我在项目一开始就反复跟团队对齐过否则很容易做一个野心巨大但永远上不了线的大平台最后变成公司里人人都知道、但没人敢用的瓷器活。2. 整体架构与数据模型设计先把地基打稳2.1 四层架构接入、缓存、计算、输出想清楚要解决什么问题之后我画了一个非常朴素的分层方案。整个服务分成四层每一层的职责都非常窄互相之间不能越界。接入层负责从各个行情和财务数据源拉取原始数据并把它们转换成内部统一的数据结构。这一层是唯一允许接触外部数据源的地方其他层一律不许直接发起外部请求。缓存层承担数据中转近期行情快照、K线、已经算好的指标都会放一份在 Redis 里避免每个请求都穿透到外部。计算层消费缓存层和存储层的数据跑指标计算和风控规则引擎把结果再写回缓存。输出层是一个 REST API 网关统一做鉴权、限流和响应格式化其他系统只跟这层打交道。为什么要明确分层最直接的理由是故障边界清楚。行情源抖动很常见某个上游接口出问题的时候接入层的采集任务会失败但只要 Redis 里还有可用的缓存API 层依然能正常给下游返回数据。反过来也一样如果哪一天下游并发量暴涨限流只会在 API 层触发并不会把压力传导到行情源。要是没有这层隔离任何一个环节出问题整个服务都会被拖下水。2.2 统一的 Symbol 与 K线模型做行情类服务第一个要处理的坑就是各个数据源的标的代码不统一。同一家公司不同行情源返回的代码可能完全不同有的叫 AAPL有的叫 AAPL.US有的甚至直接在名称里带上交易所后缀。如果不在接入层做一次彻底映射后面全链路都会跟着乱。我采取的方案是在内部维护一套统一的 symbol 体系。每个标的有且只有一个内部主键接入层适配器负责把外部代码映射成内部代码之后下游只管用这套主键完全不感知外部差异。class Symbol(BaseModel): symbol: str # 内部统一代码如 AAPL.US asset_type: str # stock / fund / bond / crypto exchange: str currency: str source_code: dict # {source_a: AAPL, source_b: AAPL.US}K线的内部结构我也做了强制约束open、high、low、close、volume 五个核心字段必须齐全并且要满足最基本的合法性high 必须大于等于 max(open, close)low 必须小于等于 min(open, close)。一旦在接入层发现违法数据直接丢弃并触发告警而不是带病入库等下游去发现。数据在入口第一次被校验是最便宜的校验。2.3 数据表设计行情、财务、风控事件存储层我用 PostgreSQL 存了三张核心表设计上尽量简单避免过度建模。表名用途关键字段symbols标的信息symbol, asset_type, exchange, currencyprice_bars历史K线symbol, timeframe, ts, open, high, low, close, volumefinancial_reports财务报告symbol, report_date, period, revenue, net_income, pe_ttm, roeprice_bars 的主键是 (symbol, timeframe, ts)这样天然保证了同一条K线重复写入不会产生重复数据。financial_reports 的主键是 (symbol, report_date)配合 INSERT...ON CONFLICT DO UPDATE 做幂等同步任务重跑多少遍都不会把数据翻倍。这里有一个后来才补上的教训字段类型一定要用 numeric不要用 double precision。虽然 numeric 写起来啰嗦但在金融场景里可以避免浮点误差导致金额对不上。我遇到过的最典型的情况是明明同一笔数据用 double 存储的版本在累计求和后和实际值差了 0.01 元而下游系统拿着这个结果做对账怎么都对不平。自从所有金额和指标字段换成 numeric 之后这类问题就绝迹了。3. 核心模块实现行情接入、财务同步与指标计算3.1 适配器模式对接不同行情源接入层最核心的设计是适配器模式。每个行情源的情况都不一样有的提供 WebSocket 实时推送有的只提供 REST 接口有的按次数计费有的按并发量限流。如果把这些差异全部堆在业务代码里那业务代码基本没法维护。我定义了一个统一的抽象接口每个行情源实现一个适配器业务代码只依赖这个接口不关心后端到底连的是哪家。class MarketDataAdapter(ABC): abstractmethod async def get_bars(self, symbol: str, timeframe: str, start: datetime, end: datetime) - list[Bar]: 获取K线数据 abstractmethod async def get_quote(self, symbol: str) - Quote: 获取实时快照每个适配器内部自己处理鉴权、字段映射、限流和重试。新增一个行情源只需要写一个新类后端的数据加工逻辑一行都不用改。我在适配器里统一加了指数退避逻辑第一次失败后等 1 秒重试第二次等 2 秒第三次等 4 秒最多重试 3 次。如果连续失败超过阈值任务自动降级先返回缓存中的旧数据同时把告警发出来。这套逻辑在行情源偶尔抽风的时候救了我很多次与其每次都拼命重试把对方打挂不如放慢节奏保命。3.2 财务数据同步的幂等与校验财务数据和行情数据不一样它不是高频更新的但口径问题比行情严重得多。财报是异步发布的每年 4 月底、8 月底、10 月底会有披露高峰同一份报告可能因为各种原因被数据源更新多次。所以财务同步管线必须支持重跑每一次重新同步都要能覆盖之前的数据。我的做法是围绕 (symbol, report_date) 这个唯一键做幂等写入。不管同步任务跑一遍还是十遍最终库里只有最新的一份。同时我对关键财务字段做了命名层面的口径区分net_income_parent 表示归母净利润net_income_total 表示合并净利润revenue 表示营业收入每个字段的含义在注释里写明。这样做的好处是下游拿到数据时不会再把两个口径混在一起。在数据校验上我加了报告期连续性检查。比如一份年报和一个季报如果覆盖的期间重叠或者相邻报告期的日期出现倒挂说明原始数据有问题我会暂缓入库并告警。这类校验不需要特别复杂的逻辑但对防止脏数据进入下游特别有效。3.3 指标计算统一放在服务层指标计算我全部收拢到服务层而不是让每个下游自己写。这样最直接的好处是口径统一所有项目拿到的 PE、波动率、最大回撤都是同一套代码算出来的。等到以后要调整算法只需要改服务端客户端完全不用动。以波动率计算为例我用日收益率序列的标准差乘以年化因子import numpy as np import pandas as pd def calc_annualized_volatility(returns: pd.Series) - float: # 假设 returns 是日频收益率序列 return float(returns.std() * np.sqrt(252))最大回撤的计算也简单但注意要基于净值序列做滚动峰值比较def calc_max_drawdown(prices: pd.Series) - float: # prices 必须是后复权价格序列 cummax prices.cummax() drawdown (prices - cummax) / cummax return float(drawdown.min())这些函数本身都不复杂但把它们放进服务层之后所有下游拿到的数字都来自同一套实现从根上杜绝了口径分叉。值得一提的是PE-TTM 这类指标不能直接存一个原始值了事我会在服务层显式计算当前总市值除以最近四个季度归母净利润之和。财报还没出齐的时候宁可返回 null 也不返回一个残缺数字避免下游把不完整的数据当成完整数据用。4. 风控规则引擎把人工检查变成自动拦截4.1 规则参数化而不是写死在业务代码里做金融服务尤其涉及交易决策辅助的时候风控检查是绕不开的一环。一开始我们只需要单标的止损线后来要加组合集中度、最大回撤限制再后来还要支持不同的策略组配不同的阈值。如果每个检查都写死在服务代码里需求一变就要改代码、发版这种节奏根本跟不上。我设计了一个非常轻量的规则引擎。每条规则用一个配置对象描述规则名、触发条件、严重级别都写在配置里而不是散落在代码的角落。RULES [ Rule( namesingle_position_limit, conditionposition_value / portfolio_value 0.2, severitywarn, ), Rule( namemax_drawdown_stop, conditionportfolio_drawdown -0.15, severitycritical, ), ]我没有引入复杂的规则表达式语言因为解析表达式本身会带来安全和调试成本。项目里实际用的是 Python 函数加 JSON 配置的组合配置里写规则名和参数实际执行逻辑是注册表里的 Python 函数。这样既保持了灵活性又不会踩到表达式注入的坑排查问题的时候还能直接看函数源码。4.2 风险事件的闭环处理风控不只是判断还要有后续动作。我在风险事件表里记录每个事件触发的时间、规则名、相关标的信息、触发指标值然后按 severity 分发到不同的通知通道。warn 级别只写日志并发给即时通讯机器人critical 级别会额外触发告警推送确保值班人员能第一时间注意到。另外还要考虑告警风暴的问题。行情剧烈波动的时候同一规则可能会在一分钟之内连续触发几十次。如果不做去重通知渠道会被刷屏真正的关键事件反而被淹没。我的做法是以规则和标的为维度做频率限制同一规则同一标的在十分钟内只发一次通知后续触发只更新事件计数。在极端行情下数据源往往会伴随延迟、K线异常等问题。这时候风控引擎尤其重要但也最容易拿到脏数据。我加了数据新鲜度检查如果行情时间戳超过阈值宁可跳过本次判定在结果里标记数据延迟未检查也不能用旧数据给出一个错误的安全结论。这个设计在平时看起来多此一举但真正遇到行情剧烈波动的时候它能避免系统给出完全误导的输出。5. API 网关与调用体验让其他系统接入不费劲5.1 鉴权与限流不要让内部服务裸奔服务对内开放之后接入方会越来越多这时候如果不做访问控制很容易出现某个下游项目写了一个循环拉全量数据直接把行情源打挂的情况。我在 API 网关层必须做两件事鉴权和限流。鉴权用的是最简单的 API Key 方案。每个接入项目分配一个 Key服务端在 Redis 里校验有效性。因为这是纯内部服务不需要复杂的 OAuth 流程只要能把非法调用挡在外面就够了。如果以后有外部合作的接入需求可以再升级为签名认证不影响现有调用方。限流用 Redis 计数器实现固定窗口限流每个 Key 每秒钟最多 N 次超过则返回 429。async def check_rate_limit(key: str, limit: int, window: int) - bool: counter await redis.incr(frate:{key}:{window_open(window)}) if counter 1: await redis.expire(frate:{key}:{window_open(window)}, window) return counter limit固定窗口在突发流量上不如滑动窗口平滑但胜在实现简单、内存占用小内部服务完全够用。后面如果流量增长到需要更平滑的限制可以平滑升级成时间轮或滑动窗口接口契约不用变。我还做了一件事API 层的限流只是第一步适配器层还有一个全局限流器因为后台定时任务同样会消耗行情源配额两边必须放到一起考虑否则 API 限流做得再好定时任务也会把外部接口打满。5.2 响应契约字段、版本、错误码API 设计的核心不是把数据返回去而是让调用方拿到数据后不需要猜。我在响应结构里统一了几个字段code、message、data_version、data。code 代表业务状态0 表示成功data_version 代表指标口径版本data 是实际数据。错误码统一约定错误码含义4001参数错误4002数据源无此标的4003数据暂时不可用4010鉴权失败4290触发限流空值处理也是很容易出歧义的地方。对于财务指标没有数据时我要求返回 null而不是 0 或者空字符串。因为 0 在金融含义里可能被下游误读为没有盈利null 才能明确表达此时无数据。这个约定刚推的时候有两个接入方不理解觉得用 0 更简单我解释了半天后来他们自己踩了一次空数据当 0 处理导致图表异常之后就再也没人反对了。一个真实的响应示例大概长这样{ code: 0, message: ok, data_version: pe_ttm_v3, data: { symbol: 0700.HK, timestamp: 2024-07-15T10:30:0008:00, last_price: 382.4, pe_ttm: 23.15 } }6. 上线后的真实踩坑与优化6.1 开盘瞬间的缓存穿透服务上线后遇到的第一个故障发生在每天早上开盘后的几分钟。行情快照接口的 QPS 突然飙升大量请求同时打到了行情源上结果把行情源短暂限流了。我查日志发现是典型的缓存穿透开盘时第一批请求到达Redis 里还没有对应标的的缓存所有请求同时回源去取数据瞬间把外部接口打挂。这个问题的标准解法有三板斧我全用上了。第一是空值缓存哪怕查不到也缓存一个短 TTL 的空对象防止反复回源第二是互斥锁同一个 symbol 只允许一个请求去重建缓存其他请求短暂等待后直接读缓存第三是预加载在交易日开盘前把当天可能被高频访问的标的快照提前刷入 Redis。三个方案一起上线后快照接口即使在开盘瞬间也保持稳定响应。现在每天开盘前半小时我会让定时任务扫描当天的关注列表把快照预先拉一遍相当于给缓存做了预热。6.2 财务指标口径漂移第二个坑跟外部数据源的选择有关。有一段时间同步任务同时从两个数据源拉财务数据结果下游系统报告里同一家公司的 PE 突然从 12 倍跳到了 18 倍。排查之后才发现数据源 A 返回的是静态市盈率按最近一期已披露报告计算数据源 B 返回的是 TTM 滚动市盈率按最近四个季度滚动计算。这两个口径在业绩稳定的时候差别不大但遇到业绩大幅波动的季度差值可能非常离谱。这个问题的本质是数据口径漂移。解决方案不是只选一个数据源就完事而是服务层必须明确以哪个口径为准。我在 financial_reports 表里增加了 pe_static 和 pe_ttm 两个字段API 响应里用 data_version 标注当前返回的口径。这样即使底层源换了一百遍下游系统也知道自己拿到的是什么版本的数据出了问题可以直接追溯到具体口径。6.3 回撤与涨跌幅的复权陷阱回撤计算看起来是最简单的指标但也是最容易在实现时翻车的地方。一开始我直接拿原始价格序列计算最大回撤结果发现一只长期分红、经常除权的股票回撤被严重高估。原因是原始价格没有包含分红再投资除权之后价格自然下跌系统误以为这是一次巨大的回撤。正确做法是策略回测和回撤分析必须使用后复权价格也就是把历史价格按分红配股向上调整。后复权价格保留了完整的历史涨幅信息适合做收益和回撤计算。前复权价格适合看行情图但不适合做回测因为它的基准会随着最新价格变化而整体移动同一段历史回撤在不同时间点看到的数值可能不一样。现在指标计算层统一用后复权价格生成回撤类指标K线接口里保留原始价字段给需要做除权分析的下游使用。这个问题如果不在服务层统一处理每个下游项目各算一遍迟早还会出现同一个策略在两个系统里回撤结果不一样的尴尬局面。我在这个项目上最深的体会是金融服务类项目里最贵的不是代码而是数据质量与口径管理的成本。一个指标算错可能让下游交易决策产生完全不同的结果一个缓存配置写错可能在行情剧烈波动时自己先挂了。所以如果你想自建一套 financial-services我的建议很直接先从数据模型和口径约定开始再写业务逻辑每一层都要有数据校验机制。代码今天写的丑一点没关系可以迭代但脏数据和混乱的口径会一直留在系统里后面付出的代价是当时的十倍。
金融数据服务中间层实战：从架构设计到风控落地
做一个名为 financial-services 的项目听起来像是要搭建一套银行核心系统实际上我做的是一个面向小团队投研场景的数据服务中间层。过去大半年我把它从一个临时脚本集合磨成了现在每天稳定扛住几十万次调用的小平台也踩了不少只会在真实业务里遇到的坑。这篇文章不聊空洞的金融科技概念就聊聊这个项目从需求分析、架构设计到接入行情源、做指标计算和风控检查的完整过程以及哪些设计决定帮我省了后来大半年的维护时间。如果你正准备在内部搭一套类似的金融数据服务或者你只是想知道一个金融服务项目里真正麻烦的地方在哪这篇应该能给你一些参考。1. 项目背景这个金融服务项目到底要解决什么问题1.1 内部项目里的数据孤岛是怎么一点一点堆出来的最开始的时候团队里其实并没有人提出我们要做一个中台大家只是各做各的项目。有人做一个股票筛选工具有人做持仓分析有人做定投回测。每个项目都有自己的数据获取方式有的是写爬虫去公开网页抓有的是天天从第三方数据平台手动导出 Excel还有的直接调某个行情供应商的接口。结果就是同一家公司三个项目里算出来的市盈率可以完全不一样。我还记得有一次开周会负责筛选工具的同事说某只股票 PE 只有 12 倍很便宜负责持仓分析的同事立刻反驳说他系统里显示的是 18 倍根本不便宜。两个人吵了半天最后发现一个人用的是静态市盈率按最近一期已披露的年报算另一个人用的是 TTM 滚动市盈率按最近四个季度加总算。再往下追连净利润的口径都不一样有人用了归母净利润有人用了包含少数股东权益的合并净利润。这个场景其实非常典型。团队一旦超过两三个人、项目一旦超过两三个数据逻辑就会出现无法收敛的分叉。你很难让每个项目组都自觉维护同一套口径定义也不应该靠人肉同步来保证一致性。更稳妥的做法是把数据获取、数据清洗、指标计算这些公共能力抽出来做成一个独立服务让所有下游项目都从这套服务里拿数据和算好的指标。这就是 financial-services 这个项目最初的形态。1.2 为什么不是继续买数据服务而是自研一个中间层有人可能会问市面上那么多现成的金融数据服务直接订阅不就行了吗我当时也反复评估过这个问题。外部成熟方案确实能解决一部分问题但有两个硬伤是绕不开的。第一个硬伤是定制化能力弱。我们的策略里有很多内部约定比如回撤必须统一用后复权价格计算PE 必须区分静态和 TTM 两种口径规则引擎要允许运营人员随时调整阈值。这些需求在外部产品里往往无法满足要么等厂商排期要么自己导出来二次加工最终还是回到乱糟糟的局面。第二个硬伤是成本叠加。多个项目分别订阅同一家服务账号要按项目开费用也要按项目算。与其每个项目都买一份不如由中间层统一接入下游项目只跟内部 API 打交道数据源订阅成本只发生一次。不过我必须强调自研不是万能的。如果团队只是做一次性分析或者只有一个小工具需要行情数据直接买现成服务更划算。只有当你能看到结构性的复用需求比如至少两个项目都在消费同一类数据并且后续还有第三个、第四个项目要接入自研中间层才值得启动。这个边界我在项目一开始就反复跟团队对齐过否则很容易做一个野心巨大但永远上不了线的大平台最后变成公司里人人都知道、但没人敢用的瓷器活。2. 整体架构与数据模型设计先把地基打稳2.1 四层架构接入、缓存、计算、输出想清楚要解决什么问题之后我画了一个非常朴素的分层方案。整个服务分成四层每一层的职责都非常窄互相之间不能越界。接入层负责从各个行情和财务数据源拉取原始数据并把它们转换成内部统一的数据结构。这一层是唯一允许接触外部数据源的地方其他层一律不许直接发起外部请求。缓存层承担数据中转近期行情快照、K线、已经算好的指标都会放一份在 Redis 里避免每个请求都穿透到外部。计算层消费缓存层和存储层的数据跑指标计算和风控规则引擎把结果再写回缓存。输出层是一个 REST API 网关统一做鉴权、限流和响应格式化其他系统只跟这层打交道。为什么要明确分层最直接的理由是故障边界清楚。行情源抖动很常见某个上游接口出问题的时候接入层的采集任务会失败但只要 Redis 里还有可用的缓存API 层依然能正常给下游返回数据。反过来也一样如果哪一天下游并发量暴涨限流只会在 API 层触发并不会把压力传导到行情源。要是没有这层隔离任何一个环节出问题整个服务都会被拖下水。2.2 统一的 Symbol 与 K线模型做行情类服务第一个要处理的坑就是各个数据源的标的代码不统一。同一家公司不同行情源返回的代码可能完全不同有的叫 AAPL有的叫 AAPL.US有的甚至直接在名称里带上交易所后缀。如果不在接入层做一次彻底映射后面全链路都会跟着乱。我采取的方案是在内部维护一套统一的 symbol 体系。每个标的有且只有一个内部主键接入层适配器负责把外部代码映射成内部代码之后下游只管用这套主键完全不感知外部差异。class Symbol(BaseModel): symbol: str # 内部统一代码如 AAPL.US asset_type: str # stock / fund / bond / crypto exchange: str currency: str source_code: dict # {source_a: AAPL, source_b: AAPL.US}K线的内部结构我也做了强制约束open、high、low、close、volume 五个核心字段必须齐全并且要满足最基本的合法性high 必须大于等于 max(open, close)low 必须小于等于 min(open, close)。一旦在接入层发现违法数据直接丢弃并触发告警而不是带病入库等下游去发现。数据在入口第一次被校验是最便宜的校验。2.3 数据表设计行情、财务、风控事件存储层我用 PostgreSQL 存了三张核心表设计上尽量简单避免过度建模。表名用途关键字段symbols标的信息symbol, asset_type, exchange, currencyprice_bars历史K线symbol, timeframe, ts, open, high, low, close, volumefinancial_reports财务报告symbol, report_date, period, revenue, net_income, pe_ttm, roeprice_bars 的主键是 (symbol, timeframe, ts)这样天然保证了同一条K线重复写入不会产生重复数据。financial_reports 的主键是 (symbol, report_date)配合 INSERT...ON CONFLICT DO UPDATE 做幂等同步任务重跑多少遍都不会把数据翻倍。这里有一个后来才补上的教训字段类型一定要用 numeric不要用 double precision。虽然 numeric 写起来啰嗦但在金融场景里可以避免浮点误差导致金额对不上。我遇到过的最典型的情况是明明同一笔数据用 double 存储的版本在累计求和后和实际值差了 0.01 元而下游系统拿着这个结果做对账怎么都对不平。自从所有金额和指标字段换成 numeric 之后这类问题就绝迹了。3. 核心模块实现行情接入、财务同步与指标计算3.1 适配器模式对接不同行情源接入层最核心的设计是适配器模式。每个行情源的情况都不一样有的提供 WebSocket 实时推送有的只提供 REST 接口有的按次数计费有的按并发量限流。如果把这些差异全部堆在业务代码里那业务代码基本没法维护。我定义了一个统一的抽象接口每个行情源实现一个适配器业务代码只依赖这个接口不关心后端到底连的是哪家。class MarketDataAdapter(ABC): abstractmethod async def get_bars(self, symbol: str, timeframe: str, start: datetime, end: datetime) - list[Bar]: 获取K线数据 abstractmethod async def get_quote(self, symbol: str) - Quote: 获取实时快照每个适配器内部自己处理鉴权、字段映射、限流和重试。新增一个行情源只需要写一个新类后端的数据加工逻辑一行都不用改。我在适配器里统一加了指数退避逻辑第一次失败后等 1 秒重试第二次等 2 秒第三次等 4 秒最多重试 3 次。如果连续失败超过阈值任务自动降级先返回缓存中的旧数据同时把告警发出来。这套逻辑在行情源偶尔抽风的时候救了我很多次与其每次都拼命重试把对方打挂不如放慢节奏保命。3.2 财务数据同步的幂等与校验财务数据和行情数据不一样它不是高频更新的但口径问题比行情严重得多。财报是异步发布的每年 4 月底、8 月底、10 月底会有披露高峰同一份报告可能因为各种原因被数据源更新多次。所以财务同步管线必须支持重跑每一次重新同步都要能覆盖之前的数据。我的做法是围绕 (symbol, report_date) 这个唯一键做幂等写入。不管同步任务跑一遍还是十遍最终库里只有最新的一份。同时我对关键财务字段做了命名层面的口径区分net_income_parent 表示归母净利润net_income_total 表示合并净利润revenue 表示营业收入每个字段的含义在注释里写明。这样做的好处是下游拿到数据时不会再把两个口径混在一起。在数据校验上我加了报告期连续性检查。比如一份年报和一个季报如果覆盖的期间重叠或者相邻报告期的日期出现倒挂说明原始数据有问题我会暂缓入库并告警。这类校验不需要特别复杂的逻辑但对防止脏数据进入下游特别有效。3.3 指标计算统一放在服务层指标计算我全部收拢到服务层而不是让每个下游自己写。这样最直接的好处是口径统一所有项目拿到的 PE、波动率、最大回撤都是同一套代码算出来的。等到以后要调整算法只需要改服务端客户端完全不用动。以波动率计算为例我用日收益率序列的标准差乘以年化因子import numpy as np import pandas as pd def calc_annualized_volatility(returns: pd.Series) - float: # 假设 returns 是日频收益率序列 return float(returns.std() * np.sqrt(252))最大回撤的计算也简单但注意要基于净值序列做滚动峰值比较def calc_max_drawdown(prices: pd.Series) - float: # prices 必须是后复权价格序列 cummax prices.cummax() drawdown (prices - cummax) / cummax return float(drawdown.min())这些函数本身都不复杂但把它们放进服务层之后所有下游拿到的数字都来自同一套实现从根上杜绝了口径分叉。值得一提的是PE-TTM 这类指标不能直接存一个原始值了事我会在服务层显式计算当前总市值除以最近四个季度归母净利润之和。财报还没出齐的时候宁可返回 null 也不返回一个残缺数字避免下游把不完整的数据当成完整数据用。4. 风控规则引擎把人工检查变成自动拦截4.1 规则参数化而不是写死在业务代码里做金融服务尤其涉及交易决策辅助的时候风控检查是绕不开的一环。一开始我们只需要单标的止损线后来要加组合集中度、最大回撤限制再后来还要支持不同的策略组配不同的阈值。如果每个检查都写死在服务代码里需求一变就要改代码、发版这种节奏根本跟不上。我设计了一个非常轻量的规则引擎。每条规则用一个配置对象描述规则名、触发条件、严重级别都写在配置里而不是散落在代码的角落。RULES [ Rule( namesingle_position_limit, conditionposition_value / portfolio_value 0.2, severitywarn, ), Rule( namemax_drawdown_stop, conditionportfolio_drawdown -0.15, severitycritical, ), ]我没有引入复杂的规则表达式语言因为解析表达式本身会带来安全和调试成本。项目里实际用的是 Python 函数加 JSON 配置的组合配置里写规则名和参数实际执行逻辑是注册表里的 Python 函数。这样既保持了灵活性又不会踩到表达式注入的坑排查问题的时候还能直接看函数源码。4.2 风险事件的闭环处理风控不只是判断还要有后续动作。我在风险事件表里记录每个事件触发的时间、规则名、相关标的信息、触发指标值然后按 severity 分发到不同的通知通道。warn 级别只写日志并发给即时通讯机器人critical 级别会额外触发告警推送确保值班人员能第一时间注意到。另外还要考虑告警风暴的问题。行情剧烈波动的时候同一规则可能会在一分钟之内连续触发几十次。如果不做去重通知渠道会被刷屏真正的关键事件反而被淹没。我的做法是以规则和标的为维度做频率限制同一规则同一标的在十分钟内只发一次通知后续触发只更新事件计数。在极端行情下数据源往往会伴随延迟、K线异常等问题。这时候风控引擎尤其重要但也最容易拿到脏数据。我加了数据新鲜度检查如果行情时间戳超过阈值宁可跳过本次判定在结果里标记数据延迟未检查也不能用旧数据给出一个错误的安全结论。这个设计在平时看起来多此一举但真正遇到行情剧烈波动的时候它能避免系统给出完全误导的输出。5. API 网关与调用体验让其他系统接入不费劲5.1 鉴权与限流不要让内部服务裸奔服务对内开放之后接入方会越来越多这时候如果不做访问控制很容易出现某个下游项目写了一个循环拉全量数据直接把行情源打挂的情况。我在 API 网关层必须做两件事鉴权和限流。鉴权用的是最简单的 API Key 方案。每个接入项目分配一个 Key服务端在 Redis 里校验有效性。因为这是纯内部服务不需要复杂的 OAuth 流程只要能把非法调用挡在外面就够了。如果以后有外部合作的接入需求可以再升级为签名认证不影响现有调用方。限流用 Redis 计数器实现固定窗口限流每个 Key 每秒钟最多 N 次超过则返回 429。async def check_rate_limit(key: str, limit: int, window: int) - bool: counter await redis.incr(frate:{key}:{window_open(window)}) if counter 1: await redis.expire(frate:{key}:{window_open(window)}, window) return counter limit固定窗口在突发流量上不如滑动窗口平滑但胜在实现简单、内存占用小内部服务完全够用。后面如果流量增长到需要更平滑的限制可以平滑升级成时间轮或滑动窗口接口契约不用变。我还做了一件事API 层的限流只是第一步适配器层还有一个全局限流器因为后台定时任务同样会消耗行情源配额两边必须放到一起考虑否则 API 限流做得再好定时任务也会把外部接口打满。5.2 响应契约字段、版本、错误码API 设计的核心不是把数据返回去而是让调用方拿到数据后不需要猜。我在响应结构里统一了几个字段code、message、data_version、data。code 代表业务状态0 表示成功data_version 代表指标口径版本data 是实际数据。错误码统一约定错误码含义4001参数错误4002数据源无此标的4003数据暂时不可用4010鉴权失败4290触发限流空值处理也是很容易出歧义的地方。对于财务指标没有数据时我要求返回 null而不是 0 或者空字符串。因为 0 在金融含义里可能被下游误读为没有盈利null 才能明确表达此时无数据。这个约定刚推的时候有两个接入方不理解觉得用 0 更简单我解释了半天后来他们自己踩了一次空数据当 0 处理导致图表异常之后就再也没人反对了。一个真实的响应示例大概长这样{ code: 0, message: ok, data_version: pe_ttm_v3, data: { symbol: 0700.HK, timestamp: 2024-07-15T10:30:0008:00, last_price: 382.4, pe_ttm: 23.15 } }6. 上线后的真实踩坑与优化6.1 开盘瞬间的缓存穿透服务上线后遇到的第一个故障发生在每天早上开盘后的几分钟。行情快照接口的 QPS 突然飙升大量请求同时打到了行情源上结果把行情源短暂限流了。我查日志发现是典型的缓存穿透开盘时第一批请求到达Redis 里还没有对应标的的缓存所有请求同时回源去取数据瞬间把外部接口打挂。这个问题的标准解法有三板斧我全用上了。第一是空值缓存哪怕查不到也缓存一个短 TTL 的空对象防止反复回源第二是互斥锁同一个 symbol 只允许一个请求去重建缓存其他请求短暂等待后直接读缓存第三是预加载在交易日开盘前把当天可能被高频访问的标的快照提前刷入 Redis。三个方案一起上线后快照接口即使在开盘瞬间也保持稳定响应。现在每天开盘前半小时我会让定时任务扫描当天的关注列表把快照预先拉一遍相当于给缓存做了预热。6.2 财务指标口径漂移第二个坑跟外部数据源的选择有关。有一段时间同步任务同时从两个数据源拉财务数据结果下游系统报告里同一家公司的 PE 突然从 12 倍跳到了 18 倍。排查之后才发现数据源 A 返回的是静态市盈率按最近一期已披露报告计算数据源 B 返回的是 TTM 滚动市盈率按最近四个季度滚动计算。这两个口径在业绩稳定的时候差别不大但遇到业绩大幅波动的季度差值可能非常离谱。这个问题的本质是数据口径漂移。解决方案不是只选一个数据源就完事而是服务层必须明确以哪个口径为准。我在 financial_reports 表里增加了 pe_static 和 pe_ttm 两个字段API 响应里用 data_version 标注当前返回的口径。这样即使底层源换了一百遍下游系统也知道自己拿到的是什么版本的数据出了问题可以直接追溯到具体口径。6.3 回撤与涨跌幅的复权陷阱回撤计算看起来是最简单的指标但也是最容易在实现时翻车的地方。一开始我直接拿原始价格序列计算最大回撤结果发现一只长期分红、经常除权的股票回撤被严重高估。原因是原始价格没有包含分红再投资除权之后价格自然下跌系统误以为这是一次巨大的回撤。正确做法是策略回测和回撤分析必须使用后复权价格也就是把历史价格按分红配股向上调整。后复权价格保留了完整的历史涨幅信息适合做收益和回撤计算。前复权价格适合看行情图但不适合做回测因为它的基准会随着最新价格变化而整体移动同一段历史回撤在不同时间点看到的数值可能不一样。现在指标计算层统一用后复权价格生成回撤类指标K线接口里保留原始价字段给需要做除权分析的下游使用。这个问题如果不在服务层统一处理每个下游项目各算一遍迟早还会出现同一个策略在两个系统里回撤结果不一样的尴尬局面。我在这个项目上最深的体会是金融服务类项目里最贵的不是代码而是数据质量与口径管理的成本。一个指标算错可能让下游交易决策产生完全不同的结果一个缓存配置写错可能在行情剧烈波动时自己先挂了。所以如果你想自建一套 financial-services我的建议很直接先从数据模型和口径约定开始再写业务逻辑每一层都要有数据校验机制。代码今天写的丑一点没关系可以迭代但脏数据和混乱的口径会一直留在系统里后面付出的代价是当时的十倍。
