从零搭建金融数据服务：架构分层、数据清洗与缓存策略实战

1. 金融数据服务从零搭建的完整思路

1.1 为什么我要自己搭一套金融数据服务

先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛，但落到实际工程里，它指的是一套面向金融场景的数据服务层——把行情、财报、宏观经济指标、汇率、利率这些散落在各处的数据，统一采集、清洗、存储，再通过标准接口对外提供查询能力。它解决的核心问题是：数据源太多太杂，格式不统一，直接对接业务代码会让整个系统变成一团乱麻。

我做这套东西的起因很直接。之前手上有个投研分析的小工具，需要同时拉A股行情、美股指数、国内宏观经济数据，还要算一些简单的财务比率。一开始图省事，每个数据源写一个脚本直接调，结果三个月后代码里全是各种 API 的密钥、不同的时间格式、五花八门的字段命名，改一个数据源要翻五个文件。后来痛定思痛，决定抽一层出来，把所有数据源统一收口，这就是financial-services的雏形。

这套服务适合谁参考？如果你正在做量化回测、投研工具、财务分析系统，或者任何需要稳定获取金融数据的应用，这套架构都能直接拿去用。哪怕你只是想定期抓一些财经数据做个人看板，里面的采集调度、数据清洗、缓存策略这些模块也能拆出来单独用。技术栈上我用的是 Python 为主，FastAPI 做接口层，PostgreSQL 存结构化数据，Redis 做缓存，整体不依赖什么冷门组件，一台普通云主机就能跑起来。

1.2 整体架构怎么分层才不乱

金融数据服务和普通业务服务最大的区别在于：数据源极不稳定，且对时效性和准确性要求极高。行情数据可能每秒都在变，财报数据一个季度才更新一次，宏观数据按月发布。如果用一个统一的采集频率去处理所有数据，要么浪费资源，要么错过关键更新。

所以我的分层思路是这样的：最底层是数据源适配层，每个数据源一个适配器，负责处理该数据源特有的认证、分页、限流、字段映射；往上一层是采集调度层，用不同的调度策略驱动不同的适配器，行情类高频、财报类低频、宏观类按发布时间触发；再往上是清洗与标准化层，把所有数据统一成内部标准格式，比如时间一律用 UTC 时间戳，金额一律用最小货币单位，字段命名统一用蛇形命名法；最上面是服务接口层，对外提供 RESTful 接口和批量查询能力。

这样分层的好处是，新增一个数据源只需要写一个适配器，不用动其他任何代码。我后来加港股数据的时候，从写适配器到上线测试，半天就搞定了。如果当初把所有逻辑揉在一起，加一个源至少得改一周。

提示：分层的时候一定要把“数据源特有的逻辑”和“通用逻辑”严格分开。我见过太多项目把某个数据源的字段名直接透传到接口层，结果换个数据源整个前端都得跟着改。

1.3 技术选型背后的取舍逻辑

选 PostgreSQL 而不是 MySQL，主要考虑的是金融数据里大量涉及时间序列查询和复杂聚合。PostgreSQL 的窗口函数、CTE、以及BRIN索引在处理按时间范围扫描的场景下表现更稳。而且JSONB字段类型让我可以在标准化表结构之外，保留原始数据的完整快照，方便排查问题。

缓存用 Redis 是常规操作，但这里有个细节：不同数据的缓存策略完全不同。实时行情缓存 3 到 5 秒，日线数据缓存到当天收盘，财报数据缓存 24 小时，宏观数据缓存 7 天。我一开始用统一的 60 秒过期，结果财报数据被反复拉取，白白浪费了 API 配额。后来改成按数据类型配置不同的 TTL，API 调用量直接降了七成。

接口层选 FastAPI，一是异步支持好，二是自动生成 OpenAPI 文档，前端同事对接起来不用我反复解释字段含义。至于调度，我没上 Celery 这种重家伙，而是用 APScheduler 加一个简单的任务队列，对于中小规模的数据服务来说完全够用，运维成本也低得多。

2. 核心模块的细节拆解与实操要点

2.1 数据源适配器怎么写才通用

适配器是整个服务的基石，写得好不好直接决定了后续扩展的成本。我的做法是定义一个抽象基类，把所有数据源共有的行为抽象出来：

from abc import ABC, abstractmethod from typing import Any class BaseAdapter(ABC): source_name: str rate_limit: int # 每分钟最大请求数 @abstractmethod async def fetch_raw(self, **kwargs) -> Any: """从数据源拉取原始数据""" pass @abstractmethod def normalize(self, raw: Any) -> list[dict]: """将原始数据转换为内部标准格式""" pass async def fetch(self, **kwargs) -> list[dict]: raw = await self.fetch_raw(**kwargs) return self.normalize(raw)

这个基类里，fetch_raw负责处理认证、请求、重试这些脏活，normalize负责字段映射和类型转换。每个具体的数据源适配器只需要实现这两个方法。我还在基类里加了限流装饰器，用令牌桶算法控制请求频率，避免触发数据源的封禁策略。

字段映射这块有个坑要特别注意：不同数据源对同一概念的定义可能完全不同。比如“成交量”，有的源给的是股数，有的给的是手数，有的甚至是金额。我在normalize里强制要求所有适配器把成交量统一成股数，如果源数据是手数就乘以 100。这个转换逻辑必须写在适配器里，不能留到上层去猜。

注意：适配器里绝对不要做业务逻辑判断。我见过有人在适配器里根据数据值决定要不要报警，结果换个数据源报警逻辑就失效了。适配器只负责“取数据”和“转格式”，其他一概不管。

2.2 数据清洗与标准化的关键规则

数据清洗是金融数据服务里最耗时间但也最不能省的一步。原始数据里常见的脏东西包括：缺失值用各种奇怪符号表示（--、N/A、null、空字符串）、时间格式五花八门（有的带时区有的不带）、数字里混着千分位逗号和货币符号、字段名大小写不一致。

我的清洗规则分三步走。第一步是类型强制转换，所有数值字段先转成Decimal类型，避免浮点精度问题。金融计算里用float是自找麻烦，0.1 加 0.2 不等于 0.3 这种事在财务对账时能让人崩溃。第二步是缺失值统一处理，所有缺失值一律转成None，在入库时存为NULL，查询时由业务层决定怎么展示。第三步是时间标准化，所有时间字段统一转成 UTC 时间戳存储，展示时再按用户时区转换。

这里有个实操心得：清洗规则一定要写成可配置的。我一开始把规则硬编码在代码里，后来发现某个数据源突然改了字段格式，不得不改代码重新部署。现在我把字段映射和清洗规则放在 YAML 配置文件里，改规则只需要改配置重启服务，不用动代码。

# config/adapters/source_a.yaml fields: trade_date: target: trade_date type: date format: "%Y%m%d" volume: target: volume type: decimal multiplier: 100 # 手转股 amount: target: amount type: decimal strip_chars: ",¥"

2.3 缓存策略的精细化配置

缓存这块我踩过的坑最多，值得单独拿出来说。最开始我用的是最简单的“查缓存，没有就查库，然后写缓存”模式，结果遇到两个问题：一是缓存击穿，某个热点数据过期瞬间大量请求打到数据库；二是缓存和数据库不一致，数据更新后缓存还是旧的。

解决缓存击穿用的是互斥锁加空值缓存。当缓存未命中时，不是所有请求都去查库，而是先抢一个分布式锁，抢到的去查库并写缓存，没抢到的等一小会儿再查缓存。对于确实不存在的数据，也缓存一个空值标记，避免反复查询。

缓存一致性方面，我采用的是写时更新加过期兜底。数据更新时主动删除对应缓存，同时设置一个较短的过期时间作为兜底。金融数据对一致性要求高，但也不是所有数据都需要强一致，行情数据差几秒可以接受，财报数据差几分钟问题也不大。

2.4 接口设计中的分页与批量查询

对外接口设计直接影响到调用方的体验。金融数据查询有两个典型场景：一是查单只股票的某段时间数据，二是批量查多只股票的某个时点数据。这两种场景对接口的要求完全不同。

单标的时序查询用时间范围加游标分页。不要用offset/limit，因为金融数据在持续写入，用 offset 分页会导致数据重复或遗漏。我的做法是返回一个游标，下次查询带上这个游标继续往后取。

批量查询用标的列表加字段过滤。调用方传一个标的代码列表和需要的字段列表，服务端只返回这些字段，避免传输大量无用数据。这里有个细节：批量查询一定要限制单次请求的标的数量上限，我设的是 200 个，超过就分批处理。不设上限的话，有人传几千个标的进来，数据库直接被打爆。

# 批量查询接口示例 @app.post("/api/v1/quotes/batch") async def batch_quotes( symbols: list[str], fields: list[str] = ["close", "volume"], trade_date: str = None ): if len(symbols) > 200: raise HTTPException(400, "单次查询标的数不能超过200") # ... 查询逻辑

3. 完整实操流程与核心环节实现

3.1 环境搭建与依赖安装

先把基础环境跑起来。我用的 Python 3.11，数据库 PostgreSQL 15，缓存 Redis 7。操作系统不限，Linux 和 macOS 都行，Windows 建议用 WSL2。

# 创建虚拟环境 python -m venv venv source venv/bin/activate # Windows 用 venv\Scripts\activate # 安装核心依赖 pip install fastapi uvicorn[standard] sqlalchemy asyncpg redis apscheduler pyyaml httpx pydantic

数据库初始化这块，我建议用 Alembic 做迁移管理，不要手动建表。金融数据的表结构后期调整频率很高，手动改表迟早会出乱子。

pip install alembic alembic init migrations # 修改 alembic.ini 中的 sqlalchemy.url alembic revision --autogenerate -m "init tables" alembic upgrade head

核心表结构我设计了四张主表：instruments存标的的基础信息，quotes存行情数据，financials存财报数据，macro_indicators存宏观指标。每张表都按时间做了分区，行情表按月分区，财报表按季度分区。分区的好处是查询时能自动裁剪掉不相关的分区，速度提升非常明显。

提示：PostgreSQL 的分区表在主键设计上有个坑，分区键必须包含在主键里。我一开始用自增 ID 做主键，后来改分区时不得不重建表。建议一开始就用(id, trade_date)这样的复合主键。

3.2 数据采集调度的配置与启动

调度这块我用 APScheduler 的AsyncIOScheduler，和 FastAPI 跑在同一个事件循环里，省得再维护一个独立的调度进程。

from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.triggers.interval import IntervalTrigger scheduler = AsyncIOScheduler() # 实时行情，交易时段每5秒采集一次 scheduler.add_job( collect_realtime_quotes, IntervalTrigger(seconds=5), id="realtime_quotes", max_instances=1 ) # 日线行情，每天收盘后采集 scheduler.add_job( collect_daily_quotes, CronTrigger(hour=15, minute=30, day_of_week="mon-fri"), id="daily_quotes" ) # 财报数据，每季度发布期密集采集 scheduler.add_job( collect_financials, CronTrigger(hour="*/2", day="1-30", month="1,4,7,10"), id="financials" )

这里的关键是max_instances=1，防止上一次采集还没跑完下一次又启动了。金融数据采集有时候会因为网络问题变慢，不加这个限制会出现任务堆积。

调度启动后，我加了一个简单的监控面板，用 Redis 记录每个任务的最后执行时间和执行状态。如果某个任务超过预期时间没执行，就发告警。这个监控不复杂，但非常有用，我有次就是因为数据源接口变更导致采集任务静默失败，靠这个监控才发现。

3.3 数据入库的批量写入优化

数据入库的性能直接决定了整个服务的吞吐量。我一开始用 ORM 逐条插入，采集 5000 条行情数据要花将近 30 秒，后来改成批量插入，同样的数据量 2 秒搞定。

from sqlalchemy.dialects.postgresql import insert async def bulk_upsert_quotes(session, quotes: list[dict]): stmt = insert(Quote).values(quotes) stmt = stmt.on_conflict_do_update( index_elements=["symbol", "trade_date"], set_={ "close": stmt.excluded.close, "volume": stmt.excluded.volume, "amount": stmt.excluded.amount, "updated_at": func.now() } ) await session.execute(stmt) await session.commit()

用ON CONFLICT DO UPDATE实现 upsert，这样重复采集不会产生重复数据，还能自动更新最新值。批量大小我设的是 1000 条一批，太大容易导致单次事务超时，太小又体现不出批量优势。这个值可以根据实际数据库配置调整，一般 500 到 2000 之间都合理。

还有个细节：入库前一定要做数据校验。我遇到过数据源返回的行情数据里混着停牌股票的空值，如果不校验直接入库，后续查询会报错。校验规则包括：价格必须大于零、成交量不能为负、交易日期不能是未来时间。校验不通过的数据记录到日志里，人工排查。

3.4 接口服务的部署与压测

服务用 Uvicorn 启动，生产环境建议用 Gunicorn 加 Uvicorn worker 的方式，充分利用多核 CPU。

gunicorn app.main:app \ --workers 4 \ --worker-class uvicorn.workers.UvicornWorker \ --bind 0.0.0.0:8000 \ --timeout 120 \ --access-logfile -

部署完成后一定要做压测。我用 Locust 写了个简单的压测脚本，模拟 100 个并发用户查询行情数据。第一次压测结果很不理想，QPS 只有 200 左右，排查发现是数据库连接池太小。SQLAlchemy 默认连接池是 5 个，改成 20 个之后 QPS 直接上到 1500。

# 数据库连接池配置 engine = create_async_engine( DATABASE_URL, pool_size=20, max_overflow=10, pool_pre_ping=True, pool_recycle=3600 )

pool_pre_ping这个参数建议打开，它会在每次从连接池取连接时先 ping 一下，避免使用到已经断开的连接。金融数据服务经常长时间运行，数据库连接被中间件断开是常有的事，不加这个参数会时不时报连接错误。

4. 常见问题排查与避坑经验实录

4.1 数据源接口变更的应对策略

数据源接口变更是这个项目里最让人头疼的问题，没有之一。我统计了一下，平均每个月都会遇到至少一次数据源字段调整或接口地址变更。应对策略的核心是快速发现、快速定位、快速修复。

快速发现靠的是数据校验和监控告警。我在入库前加了一层校验，如果某次采集的数据里关键字段缺失率超过 10%，就触发告警。这样能在数据源变更的第一时间收到通知，而不是等业务方反馈数据不对才发现。

快速定位靠的是原始数据快照。每次采集的原始响应我都会存一份到对象存储里，保留 7 天。出问题时直接对比原始数据和标准化后的数据，一眼就能看出是哪个字段的映射出了问题。

快速修复靠的是配置化。前面提到的 YAML 配置在这里发挥了关键作用，大部分字段变更只需要改配置，不用改代码。只有接口地址或认证方式变更才需要动代码，这种情况相对较少。

4.2 数据延迟与不一致的处理

金融数据对时效性要求高，但数据源本身可能有延迟。比如某个数据源宣称实时行情延迟 3 秒，实际高峰期可能延迟 30 秒。如果业务方按 3 秒的预期来用数据，就会出问题。

我的处理方式是在数据里带上数据时间戳和采集时间戳两个字段。数据时间戳是数据源标注的时间，采集时间戳是我们实际拿到数据的时间。业务方可以根据这两个时间戳的差值判断数据新鲜度，自行决定是否使用。

对于多数据源的数据不一致问题，我的策略是主源优先加交叉校验。每个数据类型指定一个主数据源，其他源作为备份和校验。如果主源和备源的数据差异超过阈值，记录异常并告警，但默认仍然使用主源数据。这样既保证了服务可用性，又能及时发现数据质量问题。

注意：不要试图自动合并多个数据源的数据，除非你非常清楚每个源的质量特征。我试过用加权平均合并两个源的行情数据，结果因为一个源在特定时段有系统性偏差，合并后的数据反而更不准。后来改成主备模式，简单可靠。

4.3 数据库性能瓶颈的排查思路

服务跑了一段时间后，查询越来越慢，这是必然会遇到的问题。排查数据库性能瓶颈我一般按这个顺序来：先看慢查询日志，再看索引使用情况，最后看表膨胀和统计信息。

慢查询日志是第一步。PostgreSQL 开启log_min_duration_statement后，所有超过阈值的查询都会记录下来。我设的阈值是 200 毫秒，大部分正常查询都在 50 毫秒以内，超过 200 毫秒的基本都有优化空间。

索引这块，金融数据最常用的查询条件是“某个标的在某段时间范围内”。所以(symbol, trade_date)的复合索引是必须的，而且trade_date要放在后面，因为范围查询只能用在索引的最后一列。如果写成(trade_date, symbol)，按标的查询时索引效果会差很多。

表膨胀是 PostgreSQL 特有的问题。频繁的更新和删除会导致表和索引膨胀，查询时需要扫描更多数据页。定期执行VACUUM ANALYZE能缓解这个问题，但更根本的解决办法是分区。我按月分区后，历史分区的数据基本不变，膨胀问题自然就消失了。

-- 查看表膨胀情况 SELECT schemaname, tablename, pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS total_size, pg_size_pretty(pg_relation_size(schemaname||'.'||tablename)) AS table_size FROM pg_tables WHERE schemaname = 'public' ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC;

4.4 服务高可用与故障恢复

金融数据服务一旦挂掉，依赖它的业务都会受影响。所以高可用设计是必须的。我的方案是多实例加健康检查加自动重启。

多实例部署在至少两台机器上，前面挂一个负载均衡。每个实例都暴露一个/health接口，返回数据库连接状态、缓存连接状态、最近一次采集时间等信息。负载均衡定期检查这个接口，发现异常就把流量切到其他实例。

自动重启用 systemd 或 supervisor 都行，我用的 systemd，配置简单且和系统集成好。关键是重启策略要合理，我设的是失败后 5 秒重启，最多重启 5 次，超过就停止并告警。不加限制的话，如果服务因为配置错误反复崩溃，会陷入无限重启循环。

# /etc/systemd/system/financial-services.service [Unit] Description=Financial Data Service After=network.target postgresql.service redis.service [Service] Type=exec User=appuser WorkingDirectory=/opt/financial-services ExecStart=/opt/financial-services/venv/bin/gunicorn app.main:app \ --workers 4 \ --worker-class uvicorn.workers.UvicornWorker \ --bind 0.0.0.0:8000 Restart=on-failure RestartSec=5 StartLimitBurst=5 [Install] WantedBy=multi-user.target

故障恢复方面，最重要的是数据可重建。所有原始数据都有快照，标准化数据可以从快照重新生成。所以即使数据库完全损坏，也能在几个小时内恢复。我建议定期做恢复演练，确保备份和恢复流程真的可用，而不是等到出事才发现备份是坏的。

4.5 常见问题速查表

这套financial-services我从最初的一个脚本慢慢迭代到现在，中间踩的坑基本都写在上面的内容里了。如果你刚开始搭类似的服务，我的建议是先把适配器层和清洗层做扎实，这两块做好了，后面加数据源、改接口都会轻松很多。至于调度和缓存，可以先用最简单的方案跑起来，等遇到性能问题再优化，不要一开始就过度设计。