从零搭建金融数据服务：分层架构、缓存与数据源适配实战

1. 金融数据服务从零搭建的核心思路1.1 为什么我要自己动手做一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我把它定位成一个面向个人开发者和小型团队的自建金融数据聚合与分发服务。它要解决的问题很具体当你需要做行情看板、记账工具、投资组合追踪、量化策略回测或者只是想给自己的副业项目接一点金融数据时你会发现市面上的数据接口要么贵得离谱要么免费额度少得可怜要么文档写得像天书要么稳定性堪忧。我最初是被逼上梁山的。去年帮朋友做一个小的资产配置工具需要拉取股票、基金、汇率这几类数据。试了几个公开接口要么一天只能调几十次要么返回格式三天两头变要么干脆某天就下线了。折腾了两周之后我意识到与其到处求人不如自己搭一套可控的数据服务把数据源、缓存、清洗、分发这几层都握在自己手里。这套服务的核心价值在于三点数据源可替换某个源挂了随时换、缓存层扛住高频读取前端随便刷不心疼、统一输出格式不管底层是哪个源吐给业务层的结构永远一致。适合谁来参考有一定后端基础、想给自己的项目接金融数据的独立开发者想理解数据服务分层设计的学生以及被第三方接口坑过、想自己掌控数据链路的同行。1.2 整体架构的分层设计我把整个服务拆成四层从上到下依次是接入层、业务逻辑层、缓存层、数据源适配层。这个分层不是拍脑袋定的而是踩坑之后总结出来的。一开始我是把取数据和返回数据写在一个函数里的结果每次换数据源都要改业务代码缓存也没法加因为不知道哪段该缓存。后来痛定思痛做了分层每一层只干一件事。接入层负责 HTTP 路由、参数校验、限流和鉴权。业务逻辑层负责把请求翻译成我需要什么数据然后决定走缓存还是走源。缓存层用 Redis 做热点数据存储带 TTL 和主动刷新。数据源适配层是最关键的一层每个数据源都封装成一个独立的 adapter对外暴露统一的接口。这样设计的好处是换数据源只需要新增一个 adapter业务层完全无感缓存策略可以独立调整不影响数据获取逻辑限流和鉴权集中在接入层改一处全局生效。1.3 技术选型的取舍逻辑技术栈我选的是Python FastAPI Redis PostgreSQL。为什么是这套组合我一个个说。选 Python 是因为金融数据处理生态成熟pandas、numpy 这些库处理时间序列数据太顺手了而且写 adapter 的时候各种 HTTP 库、解析库都很全。选 FastAPI 而不是 Flask 或 Django是因为它原生支持异步金融数据服务大量时间花在等外部接口返回上异步能显著提升吞吐而且 FastAPI 自带 Pydantic 校验和自动生成的接口文档省了我写文档的功夫。Redis 做缓存是标配它的过期策略和数据结构尤其是 sorted set 和 hash特别适合存行情快照。PostgreSQL 用来存历史数据和元数据比如数据源配置、用户订阅关系、调用日志。为什么不用 MySQL因为 PostgreSQL 对 JSON 字段和时序查询的支持更好处理半结构化的金融数据更舒服。提示如果你只是做个小工具PostgreSQL 可以先用 SQLite 顶替但缓存层强烈建议保留 Redis哪怕用本地内存缓存也要有否则高频请求会直接把你的数据源打爆。2. 数据源适配层的核心细节与实操要点2.1 统一数据模型的设计适配层最关键的是定义一套统一数据模型让所有数据源都往这个模型上靠。我定义的核心实体有三个Quote行情快照、CandleK线、Instrument标的元信息。Quote包含字段symbol标的代码、price最新价、change涨跌额、change_pct涨跌幅、volume成交量、timestamp时间戳、source数据来源标识。Candle在此基础上多了open、high、low、close和interval周期。Instrument则包含symbol、name、type股票/基金/外汇/加密货币、exchange、currency。为什么要定义这么细因为不同数据源返回的字段名千奇百怪。有的叫last有的叫latest_price有的叫close。如果不在适配层做归一化业务层就要写一堆 if-else 判断数据源那分层就白做了。from pydantic import BaseModel from datetime import datetime from typing import Optional class Quote(BaseModel): symbol: str price: float change: Optional[float] None change_pct: Optional[float] None volume: Optional[float] None timestamp: datetime source: str class Candle(BaseModel): symbol: str interval: str open: float high: float low: float close: float volume: float timestamp: datetime source: str用 Pydantic 定义模型的好处是adapter 返回的数据会自动做类型校验字段缺失或类型不对会立刻报错而不是等到业务层用的时候才炸。2.2 适配器接口的抽象每个数据源适配器都要实现同一个抽象基类我把它叫BaseAdapter。它定义了三个必须实现的方法fetch_quote(symbol)、fetch_candles(symbol, interval, limit)、search_instrument(keyword)。from abc import ABC, abstractmethod class BaseAdapter(ABC): name: str abstractmethod async def fetch_quote(self, symbol: str) - Quote: ... abstractmethod async def fetch_candles(self, symbol: str, interval: str, limit: int) - list[Candle]: ... abstractmethod async def search_instrument(self, keyword: str) - list[Instrument]: ...这样设计之后新增一个数据源就是新建一个文件、继承BaseAdapter、实现三个方法然后在配置里注册一下。业务层通过一个AdapterRegistry来获取适配器完全不知道底层是谁。2.3 数据源优先级与降级策略实际运行中单一数据源不可靠是常态。我设计了一套优先级 降级机制。每个数据源在配置里有一个priority值数字越小优先级越高。业务层请求数据时按优先级依次尝试第一个成功的就返回同时记录哪个源失败了。降级策略要配合熔断。如果某个源连续失败超过阈值我设的是 5 次就把它临时标记为不可用冷却 60 秒后再试。这样避免了一个挂掉的源拖慢所有请求。class AdapterRegistry: def __init__(self): self.adapters: list[BaseAdapter] [] self.failure_count: dict[str, int] {} self.cooldown_until: dict[str, float] {} async def fetch_quote_with_fallback(self, symbol: str) - Quote: now time.time() for adapter in sorted(self.adapters, keylambda a: a.priority): if self.cooldown_until.get(adapter.name, 0) now: continue try: quote await adapter.fetch_quote(symbol) self.failure_count[adapter.name] 0 return quote except Exception: self.failure_count[adapter.name] self.failure_count.get(adapter.name, 0) 1 if self.failure_count[adapter.name] 5: self.cooldown_until[adapter.name] now 60 raise AllSourcesFailedError(symbol)注意熔断阈值和冷却时间要根据数据源的实际稳定性调。我一开始设的是 3 次失败冷却 30 秒结果某个源偶尔抽风就被熔断了反而增加了其他源的压力。后来改成 5 次 / 60 秒才稳定下来。3. 缓存层与请求链路的完整实现3.1 缓存键设计与 TTL 策略缓存做得好不好一半看键设计一半看 TTL。我的缓存键规则是fs:{type}:{symbol}:{extra}比如fs:quote:AAPL、fs:candle:AAPL:1d:100。前缀fs是服务标识方便在共享 Redis 里区分。TTL 不能一刀切。行情快照变化快我设 5 秒K线数据变化慢设 60 秒标的元信息基本不变设 1 小时。这个策略是权衡了数据新鲜度和源压力之后定的。5 秒意味着前端每秒刷一次实际只有五分之一会穿透到源。TTL_CONFIG { quote: 5, candle: 60, instrument: 3600, } async def get_quote_cached(symbol: str) - Quote: key ffs:quote:{symbol} cached await redis.get(key) if cached: return Quote.model_validate_json(cached) quote await registry.fetch_quote_with_fallback(symbol) await redis.setex(key, TTL_CONFIG[quote], quote.model_dump_json()) return quote3.2 缓存穿透与雪崩的防护缓存穿透指的是请求一个根本不存在的标的每次都穿透到源。防护办法是空值缓存查不到的数据也缓存一个短 TTL 的空标记比如 30 秒避免恶意或错误的 symbol 反复打源。缓存雪崩指的是大量键同时过期请求瞬间全压到源上。防护办法是给 TTL 加随机抖动比如 5 秒的 TTL 实际设成 4 到 6 秒之间的随机值。import random def jittered_ttl(base: int) - int: return base random.randint(-base // 5, base // 5)这个抖动看起来不起眼但在高并发下能救命。我之前压测的时候没加抖动整点刷新时源接口直接被瞬时流量打挂加了抖动之后曲线就平滑了。3.3 主动刷新与被动过期结合纯被动过期有个问题第一个请求永远要等源返回体验差。我加了一层主动刷新对于热门标的访问频率高的后台有个定时任务提前刷新缓存让用户请求永远命中热数据。判断热门的方式是统计每个 symbol 的访问次数超过阈值的加入热榜定时任务只刷热榜里的。这样既保证了热门数据的实时性又不浪费资源去刷没人看的标的。async def refresh_hot_symbols(): hot await redis.zrange(fs:hot_symbols, 0, 49, descTrue) for symbol in hot: try: quote await registry.fetch_quote_with_fallback(symbol) await redis.setex(ffs:quote:{symbol}, jittered_ttl(5), quote.model_dump_json()) except Exception as e: logger.warning(frefresh failed for {symbol}: {e})3.4 完整请求链路的串联把上面这些串起来一个完整的行情请求链路是这样的接入层收到GET /quote/AAPL校验参数和鉴权调用业务层的get_quote_cached。业务层先查 Redis命中就返回没命中就走 registry 的降级逻辑取源取到后写缓存再返回。同时异步更新该 symbol 的访问计数用于热榜统计。这条链路我实测下来热门标的的 P99 延迟在 10 毫秒以内冷门标的因为要穿透到源延迟取决于源一般在 200 到 800 毫秒。这个差距是合理的因为冷门数据本来就不该占用缓存资源。4. 常见问题与排查技巧实录4.1 数据源返回格式突变怎么办这是最恶心的问题。某个源某天突然把price字段改成了字符串或者把时间戳从秒改成了毫秒你的解析直接崩。我的应对是在 adapter 里做防御性解析所有字段都做类型转换和默认值处理解析失败时记录原始响应到日志方便事后排查。def safe_float(value, defaultNone): try: return float(value) except (TypeError, ValueError): return default同时我加了一个响应快照机制每个源每天第一次成功请求的原始响应存一份到对象存储格式变了能对比出来。这个习惯帮我定位过好几次昨天还好好的今天就不对了的问题。4.2 时间戳与时区混乱的坑金融数据对时间极其敏感。不同源返回的时间戳有的是 UTC有的是本地时间有的是毫秒有的是秒。我踩过的坑是某次回测发现 K 线对不上查了半天发现是某个源返回的是交易所本地时间我当成 UTC 处理了差了 8 小时。解决办法是在 adapter 层统一转成 UTC 带时区的 datetime业务层只认 UTC。所有涉及展示的地方再按用户时区转换。这个规则定死之后再没出过时间错乱的问题。问题现象可能原因排查方法K线时间对不上时区未统一检查 adapter 是否转 UTC时间戳差 1000 倍秒/毫秒混用看数值量级13 位是毫秒数据延迟一小时源本身延迟对比多个源的时间戳4.3 限流被源封禁的处理免费数据源基本都有限流超了要么返回 429要么直接封 IP。我的做法是在 adapter 层做本地限流用令牌桶控制每个源的请求速率留 20% 余量。比如源限制 60 次/分钟我就限到 48 次/分钟。from asyncio import Semaphore import time class RateLimiter: def __init__(self, rate_per_minute: int): self.interval 60.0 / rate_per_minute self.last_call 0.0 async def acquire(self): now time.time() wait self.last_call self.interval - now if wait 0: await asyncio.sleep(wait) self.last_call time.time()提示限流一定要做在调用源之前而不是收到 429 之后才退避。主动限流比被动退避稳定得多因为被动退避时你已经浪费了一次请求配额。4.4 缓存与源数据不一致的排查有时候用户反馈价格不对其实是缓存还没过期。排查这类问题的顺序是先看 Redis 里该键的 TTL 和值再看源的实际返回最后对比时间戳。如果缓存值的时间戳比源旧很多说明主动刷新没生效或者 TTL 设太长了。我整理了一个速查表遇到数据不一致按这个顺序查步骤检查项命令/方法1缓存值和时间戳redis-cli get fs:quote:XXX2缓存 TTLredis-cli ttl fs:quote:XXX3源实际返回直接调 adapter 的 fetch4刷新任务日志查后台任务是否报错5熔断状态看该源是否在冷却期4.5 高并发下的连接池耗尽FastAPI 异步 外部 HTTP 请求如果不控制连接池高并发下会耗尽文件描述符。我给每个 adapter 配了独立的httpx.AsyncClient设置limitsmax_connections20并且全局复用而不是每次请求新建。这个改动让服务在压测下的稳定性提升了一个档次。另外 Redis 连接也要用连接池redis.asyncio的ConnectionPool默认就够用但记得设置max_connections别让它无限增长。5. 部署与运维的实战经验5.1 容器化与配置分离整个服务我打包成一个 Docker 镜像配置全部走环境变量。数据源配置哪些源启用、优先级、限流参数放在一个 YAML 文件里通过挂载卷注入。这样同一份镜像可以在开发、测试、生产环境跑只换配置。sources: - name: source_a enabled: true priority: 1 rate_limit: 48 - name: source_b enabled: true priority: 2 rate_limit: 30配置分离的好处是某个源挂了要临时禁用改配置重启即可不用改代码重新构建镜像。5.2 监控指标与告警我埋了几个关键指标每个源的请求成功率、平均延迟、熔断次数缓存的命中率接口的 P95/P99 延迟。这些指标推到 Prometheus配了 Grafana 看板。告警规则设了两条某源成功率 5 分钟内低于 80% 告警缓存命中率低于 60% 告警。第二条特别有用命中率掉了说明要么 TTL 设太短要么有异常请求在穿透。5.3 数据源健康检查除了被动熔断我还加了一个主动健康检查任务每 5 分钟对每个启用的源发一个轻量请求比如查一个固定标的的行情验证它是否还活着。健康检查结果也进监控这样能在用户感知之前发现源的问题。这个健康检查的标的要选那种永远存在的比如大盘指数或者主流货币对别选可能退市的个股否则源没问题但标的没了会误报。5.4 灰度切换数据源新增或替换数据源时我不用一刀切而是灰度。新源先以低优先级接入只承接一小部分流量通过权重控制观察一段时间成功率稳定后再逐步提高优先级。这样即使新源有问题影响面也可控。权重控制我是在 registry 里做的每个 adapter 有个weight请求时按权重随机选择而不是严格按优先级。等新源稳定了再把权重调满、优先级调高。6. 后续可扩展的方向这套服务跑了大半年基本满足了我的需求。如果继续往下做我会考虑几个方向。一是加一层数据质量校验对拿到的行情做异常检测比如价格突然偏离均值 10% 就标记可疑避免脏数据污染下游。二是支持 WebSocket 推送现在都是轮询对于实时性要求高的场景推送更省资源。三是把 adapter 做成插件化支持动态加载这样新增数据源连重启都不用。不过说实话对于个人项目现在这套已经够用了。过度设计反而增加维护成本。我的原则是先把核心链路跑通跑稳再考虑锦上添花。当初我要是上来就追求插件化和推送估计到现在还没上线。最后分享一个我踩过的坑别把数据源配置写死在代码里。我第一版就是把 API 地址和密钥硬编码的结果要换源的时候改得满屏都是。现在全部走配置改一个 YAML 文件搞定这个习惯能省你无数时间。