1. 金融数据服务从零搭建的完整思路1.1 为什么我要自己搭一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合与分发服务。它要解决的核心问题是当你需要股票行情、汇率、宏观经济指标、公司财报这些数据时要么去各个网站手动扒要么用商业API但免费额度少得可怜要么找到的开源方案年久失修跑不起来。这套服务就是把数据采集、清洗、存储、缓存、对外接口这几个环节串成一条完整的流水线让你在自己的机器上就能跑起来一个稳定的数据源。适合谁来参考如果你是会一点 Python 或 Node.js 的开发者想给自己的量化策略、记账工具、投资看板或者数据分析项目接一个靠谱的数据后端这套东西就是为你准备的。哪怕你只是想定时抓一些汇率数据存到自己的数据库里做趋势分析也能从这里找到可直接复用的模块。我踩过的坑是一开始贪大求全想着一口气把A股、美股、港股、加密货币、外汇全接进来结果每个数据源的字段格式、更新频率、限流策略都不一样代码写到后面变成了一团乱麻。后来我换了个思路先定义一套统一的数据模型再针对每个数据源写适配器整个架构才清爽起来。这个教训在后面会反复提到因为它直接决定了你的服务能不能长期维护下去。1.2 整体架构是怎么设计的整套服务我采用的是经典的分层架构从上到下依次是接口层、缓存层、业务逻辑层、数据采集层、存储层。这个分层不是拍脑袋定的每一层都有它存在的理由。接口层负责对外暴露 RESTful API 和 WebSocket 两种协议。RESTful 用于查询历史数据、获取公司信息这类一次性请求WebSocket 用于推送实时行情因为轮询的方式在数据量大时对服务端压力太大。我试过只用 RESTful 做实时推送客户端每秒发一次请求十个客户端就把服务打挂了换成 WebSocket 之后同样的负载下 CPU 占用降了七成。缓存层用的是 Redis。金融数据有个特点读多写少而且很多数据在短时间内不会变化。比如某只股票昨天的收盘价你查一百次结果都一样没必要每次都去数据库里捞。我把热点数据放在 Redis 里设置合理的过期时间数据库的压力瞬间就下来了。这里有个细节不同数据的过期时间要区别对待。实时行情可能 3 秒就过期日线数据可以缓存到当天收盘后公司基本信息甚至可以缓存 24 小时。业务逻辑层是整套服务的大脑负责数据格式转换、指标计算、异常检测。比如把不同数据源返回的字段名统一成标准命名计算移动平均线、波动率这些衍生指标检测数据中明显的异常值并标记出来。这一层我建议尽量保持纯粹不要掺入任何跟具体数据源相关的代码这样以后换数据源时只需要改采集层。数据采集层是适配器模式的重灾区。每个数据源一个适配器类实现统一的接口方法比如fetch_quote、fetch_history、fetch_fundamentals。适配器内部处理该数据源特有的认证、分页、限流、重试逻辑。我目前接了四个数据源两两互为备份当一个源挂掉或者限流时自动切换到另一个。存储层用的是 PostgreSQL 加 TimescaleDB 扩展。PostgreSQL 负责存公司信息、财报这类关系型数据TimescaleDB 负责存时间序列的行情数据。为什么不用 MySQL因为 TimescaleDB 对时间序列的压缩和查询优化做得确实好同样一亿条行情数据PostgreSQL 加 TimescaleDB 的查询速度比我之前用 MySQL 快了三到五倍。为什么不用专门的时序数据库比如 InfluxDB因为我的关系型数据也需要存储不想维护两套数据库系统TimescaleDB 正好能兼顾。1.3 技术选型背后的取舍逻辑语言层面我选了 Python 作为主力原因很实际金融数据处理生态里 Python 的库最全pandas、numpy、pandas-datareader 这些工具能省掉大量造轮子的时间。有人会说 Python 性能不行但在这个场景下瓶颈在网络 IO 和数据库查询上不在计算上。真到了计算密集型的部分可以把那部分单独抽出来用 Go 或 Rust 写通过 gRPC 调用。Web 框架用的是 FastAPI。选它不是因为性能比 Flask 好多少而是因为它原生支持异步、自动生成 OpenAPI 文档、内置数据校验。异步这一点很关键采集层要同时向多个数据源发请求用异步能大幅缩短采集时间。自动生成文档这一点在团队协作时特别省事前端同事直接看/docs就能知道接口怎么调。任务调度用的是 Celery 加 Redis 作为 broker。为什么不用 cron因为 cron 只能定时触发没法做任务依赖、失败重试、优先级队列这些事。比如财报数据必须在行情数据采集完成之后再处理这种依赖关系用 Celery 的 chain 和 group 很容易表达。Redis 作为 broker 是因为它足够轻量而且我本来就用它做缓存不用额外维护一套消息队列。容器化用的是 Docker Compose。没有上 Kubernetes 是因为这套服务的规模还没到那个程度一台 4 核 8G 的云服务器就能跑得很好。Docker Compose 的好处是配置文件一目了然新人接手时看一遍docker-compose.yml就知道整个系统有哪些组件、怎么互相通信。注意技术选型没有绝对的对错只有适不适合你当前的场景。如果你只是一个人维护一个小项目别上来就搞微服务、消息队列、服务网格那一套运维成本会把你压垮。先用单体架构跑起来等真的遇到瓶颈了再拆。2. 核心模块的细节拆解与实操要点2.1 数据模型设计统一字段是长期维护的基石数据模型这块我返工了三次前两次都是因为没想清楚不同数据源之间的字段差异。举个具体的例子同样是股票行情有的数据源返回的字段叫open、high、low、close有的叫open_price、highest、lowest、closing_price还有的用中文拼音首字母kp、zg、zd、sp。如果你在每个用到行情的地方都去处理这些差异代码会变得极其恶心。我的做法是定义一套标准数据模型所有数据源的数据在进入业务逻辑层之前必须转换成这个标准格式。以行情数据为例标准模型包含这些字段字段名类型说明symbolstring标准化后的标的代码如 AAPL、000001.SZtimestampdatetimeUTC 时间戳精确到秒opendecimal开盘价highdecimal最高价lowdecimal最低价closedecimal收盘价volumebigint成交量turnoverdecimal成交额sourcestring数据来源标识这里有几个设计决策值得展开说。第一价格用decimal而不是float。金融计算里浮点数精度问题会要命0.1 0.2 不等于 0.3 这种事在记账场景下是灾难。PostgreSQL 的numeric类型和 Python 的Decimal类型能精确表示十进制小数虽然计算速度慢一点但准确性优先。第二时间戳统一用 UTC。不同数据源返回的时间可能是北京时间、美东时间、伦敦时间如果不统一跨市场分析时你会疯掉。我在采集层就把所有时间转成 UTC 存储在接口层根据客户端请求的时区再转回去。第三source字段不能省。当两个数据源对同一只股票的收盘价有微小差异时你需要知道哪条数据来自哪里方便排查问题。我遇到过某数据源把除权后的价格当成未除权价格返回的情况就是靠source字段定位到问题源头的。2.2 采集适配器的编写规范与限流处理每个数据源写一个适配器类继承同一个基类。基类定义了几个必须实现的方法和几个可选覆盖的方法。必须实现的是fetch_quote、fetch_history、fetch_fundamentals可选覆盖的是health_check和rate_limit_config。限流处理是适配器里最需要花心思的地方。不同数据源的限流策略差异很大有的限制每分钟请求数有的限制每天请求数有的对不同的接口有不同的限制。我的做法是在适配器内部维护一个令牌桶每次发请求前先获取令牌获取不到就等待或者抛出限流异常由上层决定是否切换数据源。令牌桶的实现我用的是aiolimiter这个库它提供了异步的限流器跟 FastAPI 的异步体系配合得很好。配置大概是这样的from aiolimiter import AsyncLimiter class BaseAdapter: def __init__(self): self.limiter AsyncLimiter( max_rateself.rate_limit_config[requests], time_periodself.rate_limit_config[period] ) async def _request(self, url, params): async with self.limiter: # 实际的 HTTP 请求逻辑 pass重试逻辑我采用的是指数退避加抖动。第一次失败等 1 秒第二次等 2 秒第三次等 4 秒以此类推但每次等待时间加上一个随机的小幅抖动避免多个请求同时重试造成惊群效应。最大重试次数设为 3 次超过就标记该数据源暂时不可用切换到备用源。实操心得写适配器时一定要把原始响应保存下来至少保存最近 100 条。当数据出现异常时你可以回放原始响应来确认是数据源的问题还是你解析代码的问题。我就在这个环节省下了大量跟数据源客服扯皮的时间。2.3 缓存策略什么该缓存、缓存多久、怎么失效缓存这块我吃过亏一开始把所有数据都往 Redis 里塞结果内存爆了而且很多数据其实根本不需要缓存。后来我总结了一个判断标准如果一份数据在它的有效期内被查询超过 3 次就值得缓存否则直接查数据库更省事。具体到各类数据的缓存策略实时行情数据缓存 3 到 5 秒。这个时间窗口足够应对突发的查询高峰又不会让用户看到太陈旧的价格。实现方式是用 Redis 的SETEX命令键名格式是quote:{symbol}。日线及更长时间粒度的历史数据缓存到当天结束。因为日线数据在收盘后就固定了盘中查询时当天的那条数据可能还在变化所以缓存到当天 23:59:59 过期。键名格式是history:{symbol}:{interval}:{date}。公司基本信息、财报数据缓存 24 小时。这类数据更新频率低缓存久一点没问题。但要注意财报发布日前后要主动清除缓存否则用户可能看到的是上一季度的数据。我的做法是在采集任务完成后根据采集到的财报日期主动删除相关缓存键。缓存失效我采用的是主动失效加被动过期双保险。主动失效是在数据更新时删除对应的缓存键被动过期是设置合理的 TTL。只靠主动失效的话万一更新逻辑有 bug 没删掉缓存用户就会一直看到旧数据只靠被动过期的话数据更新后要等 TTL 到期才能生效实时性不够。2.4 接口设计RESTful 与 WebSocket 的分工RESTful 接口我遵循的是资源导向的设计风格。获取某只股票的最新行情是GET /api/v1/quotes/{symbol}获取历史行情是GET /api/v1/quotes/{symbol}/history?start...end...interval...获取公司信息是GET /api/v1/companies/{symbol}。这种设计的好处是直观前端同事看一眼就知道怎么调。分页我采用的是游标分页而不是偏移分页。偏移分页在数据量大时性能很差因为数据库需要扫描并跳过前面的记录。游标分页用上一页最后一条记录的时间戳或 ID 作为游标查询效率稳定。返回格式里包含next_cursor字段客户端拿着这个游标请求下一页。WebSocket 接口用于实时行情推送。客户端连接时通过查询参数指定要订阅的标的列表服务端维护一个订阅关系表。当采集层获取到新行情时通过 Redis 的发布订阅机制通知 WebSocket 服务后者根据订阅关系推送给对应的客户端。这里有个细节WebSocket 推送要做节流。如果某个标的每秒更新十次而客户端只关心每秒一次的价格你推十次就是浪费带宽和客户端资源。我的做法是在服务端做聚合每 500 毫秒推送一次推送的是这 500 毫秒内的最新价格。客户端如果对实时性要求更高可以在连接参数里指定推送频率。3. 完整实操流程与关键环节实现3.1 环境准备与依赖安装我假设你用的是 Ubuntu 22.04 或者 macOSWindows 的话建议用 WSL2因为后面有些工具在原生 Windows 上配置起来比较折腾。第一步是安装 Docker 和 Docker Compose。Ubuntu 上执行sudo apt update sudo apt install -y docker.io docker-compose-plugin sudo systemctl enable --now docker sudo usermod -aG docker $USER执行完最后一条命令后需要重新登录终端让用户组变更生效。验证安装是否成功docker --version docker compose version第二步是克隆项目代码并创建虚拟环境。我习惯用venv而不是 conda因为 venv 更轻量跟 Docker 配合也更简单。git clone 你的仓库地址 financial-services cd financial-services python3 -m venv venv source venv/bin/activate pip install -r requirements.txtrequirements.txt里的核心依赖包括fastapi、uvicorn、sqlalchemy、asyncpg、redis、celery、httpx、pandas、aiolimiter、pydantic。版本号我建议都锁定避免某天自动升级后出现不兼容。第三步是配置环境变量。项目根目录下有个.env.example文件复制成.env然后填入你的配置cp .env.example .env需要填的配置项包括数据库连接串、Redis 连接串、各个数据源的 API Key如果有的话、服务监听端口。API Key 这类敏感信息千万不要提交到代码仓库.env文件要加到.gitignore里。3.2 数据库初始化与迁移数据库我用的是 PostgreSQL 15 加 TimescaleDB 2.x。用 Docker Compose 启动的话docker-compose.yml里已经配置好了services: db: image: timescale/timescaledb:latest-pg15 environment: POSTGRES_USER: finserv POSTGRES_PASSWORD: your_password POSTGRES_DB: financial_services ports: - 5432:5432 volumes: - pgdata:/var/lib/postgresql/data启动数据库后需要执行初始化脚本创建表结构和 TimescaleDB 的超表。我用的迁移工具是 Alembic它能跟踪数据库 schema 的变更历史多人协作时不会出现你改了表结构没告诉我的情况。alembic upgrade head这个命令会依次执行migrations/versions/目录下的迁移脚本。第一个迁移脚本创建基础表第二个把行情表转换成 TimescaleDB 的超表并设置压缩策略。关于压缩策略我的配置是按 7 天为一个 chunk超过 30 天的数据自动压缩。压缩后存储空间能省 80% 以上查询速度反而更快因为扫描的数据块更少。配置语句大概是SELECT add_compression_policy(quotes, INTERVAL 30 days); SELECT add_retention_policy(quotes, INTERVAL 5 years);保留策略设了 5 年超过 5 年的数据自动删除。这个根据你的实际需求调整如果是做长期回测保留时间要设长一点。3.3 采集任务的配置与调度采集任务用 Celery 调度。celery_config.py里定义了各个任务的执行频率CELERYBEAT_SCHEDULE { fetch-realtime-quotes: { task: tasks.fetch_realtime_quotes, schedule: 5.0, # 每 5 秒执行一次 }, fetch-daily-history: { task: tasks.fetch_daily_history, schedule: crontab(hour18, minute0), # 每天 18:00 执行 }, fetch-fundamentals: { task: tasks.fetch_fundamentals, schedule: crontab(hour6, minute0, day_of_week1), # 每周一 6:00 }, }实时行情每 5 秒采集一次这个频率对大多数场景够用了。如果你做高频交易那这套架构不适合你你需要的是 colocation 和专线不是自建数据服务。日线历史数据在每天 18:00 采集因为 A 股 15:00 收盘美股 16:00 收盘北京时间凌晨 4 点或 5 点18:00 这个时间点两边的数据都出来了。财报数据每周一早上采集一次因为财报发布集中在工作日盘后。启动 Celery worker 和 beat 的命令celery -A tasks worker --loglevelinfo --concurrency4 celery -A tasks beat --loglevelinfo--concurrency4表示启动 4 个 worker 进程。这个数字根据你的 CPU 核心数调整一般是核心数加一。但要注意如果采集任务主要是网络 IO 等待可以设大一点如果是计算密集型设成核心数就行。3.4 服务启动与验证所有组件都配置好后用 Docker Compose 一键启动docker compose up -d这个命令会启动数据库、Redis、API 服务、Celery worker、Celery beat 这五个容器。启动完成后用docker compose ps查看各容器状态确保都是running而不是restarting。验证服务是否正常工作的步骤第一步检查 API 文档是否能访问。浏览器打开http://localhost:8000/docs应该能看到自动生成的 Swagger 文档页面。第二步调用一个简单的接口测试数据是否正常返回curl http://localhost:8000/api/v1/quotes/AAPL如果返回了包含最新价格的 JSON 数据说明采集、存储、接口这条链路是通的。第三步检查 WebSocket 推送是否正常。我用websocat这个工具测试websocat ws://localhost:8000/ws/quotes?symbolsAAPL,MSFT连接成功后应该能看到不断推送过来的行情数据。注意第一次启动时数据库是空的采集任务需要运行一段时间才有数据。你可以手动触发一次全量采集来加速这个过程celery -A tasks call tasks.fetch_daily_history。4. 常见问题排查与避坑经验实录4.1 数据源限流与封禁的应对策略这是自建金融数据服务最常见的问题。免费数据源通常有严格的限流稍微频繁一点的请求就会被封 IP。我遇到过某数据源在连续请求 100 次后返回 429 状态码封禁时长从几分钟到几小时不等。应对策略分几个层次。第一层是严格遵守数据源公布的限流规则在适配器里配置好令牌桶参数宁可慢一点也不要触发封禁。第二层是准备多个数据源互为备份当一个源返回 429 时自动切换到下一个。第三层是如果所有源都被封了降级到使用缓存数据并在接口返回里标记数据可能不是最新的。这里有个细节切换数据源时要注意数据的一致性。不同数据源对同一只股票的代码格式可能不同比如有的用AAPL有的用AAPL.US有的用US.AAPL。我在标准数据模型里维护了一个代码映射表切换数据源时自动转换代码格式。4.2 数据质量问题的检测与处理金融数据里脏数据比你想象的多。我遇到过收盘价为 0 的记录、成交量突然放大一万倍的异常值、时间戳错乱的记录。如果不做检测这些脏数据会污染你的分析和策略。我的做法是在业务逻辑层加一个数据质量检查环节对每条采集到的数据做几个基本检查价格是否在合理范围内比如跟上一笔价格相比波动不超过 20%、成交量是否为非负数、时间戳是否在合理区间内。检查不通过的数据标记为可疑存入单独的表中供人工审核不进入正常的数据流。对于缺失数据我的处理策略是如果缺失的是最近几天的数据尝试从备用数据源补采如果补不到在数据库中标记为缺失查询时返回 null 而不是用前值填充。用前值填充虽然看起来数据完整了但会引入偏差做回测时会导致虚高的收益。4.3 性能瓶颈的定位与优化服务跑起来之后随着数据量增长你可能会遇到查询变慢的问题。定位性能瓶颈的工具我用的是pg_stat_statements扩展它能记录所有 SQL 语句的执行统计按总耗时排序一眼就能看出哪条查询最耗资源。常见的性能问题和对策问题现象可能原因解决方案历史数据查询慢缺少合适的索引在 symbol 和 timestamp 上建复合索引实时接口响应慢缓存未命中率高调整缓存 TTL预热热点数据采集任务堆积worker 数量不足增加 concurrency 或拆分任务数据库连接耗尽连接池配置不当调整 SQLAlchemy 连接池大小内存占用持续增长内存泄漏或缓存过大检查代码设置 Redis 内存上限索引这块我要多说一句。TimescaleDB 的超表建索引跟普通表不太一样它支持按 chunk 建索引。我的做法是在symbol和timestamp上建一个复合索引查询时先按 symbol 过滤再按时间范围过滤能充分利用索引。另外对于经常查询的字段比如close可以考虑建覆盖索引把查询需要的字段都包含在索引里避免回表。4.4 常见问题速查表问题排查步骤解决方法服务启动后接口 500查看 API 容器日志通常是数据库连接失败检查 .env 配置采集任务不执行检查 Celery beat 日志确认 beat 容器在运行任务已注册WebSocket 连不上检查反向代理配置确保代理支持 WebSocket 升级数据更新不及时查看采集任务执行记录可能是数据源限流检查适配器日志内存占用过高查看 Redis 内存使用设置 maxmemory 和淘汰策略数据库磁盘满检查数据保留策略调整 retention policy 或扩容磁盘实操心得日志是你的第一排查工具。我在每个关键环节都加了结构化日志包含时间戳、模块名、请求 ID、耗时、结果状态。出问题时用grep按请求 ID 过滤整条链路的执行情况一目了然。日志级别我设的是 INFODEBUG 级别只在排查特定问题时临时开启否则日志量太大会拖慢服务。4.5 安全加固与访问控制虽然是个自托管服务但安全不能马虎。我做了这几件事API 认证用的是 JWT。客户端先用 API Key 换取 JWT后续请求带上 JWT。JWT 里包含用户 ID 和权限范围接口层根据权限范围决定是否放行。API Key 存在数据库里支持随时吊销。速率限制在 API 层也做了一层。即使数据源那边没限流你自己的服务也要防止被恶意刷接口。我用的是slowapi这个库基于客户端 IP 和用户 ID 做限流超过阈值返回 429。数据库和 Redis 都不直接暴露到公网。Docker Compose 里只把 API 服务的端口映射出来数据库和 Redis 只在内部网络通信。如果确实需要从外部访问数据库通过 SSH 隧道而不是直接开放端口。敏感配置全部通过环境变量注入不写在代码里。.env文件权限设为 600只有当前用户可读。生产环境的密钥管理可以考虑用 Docker secrets 或者外部的密钥管理服务。这套东西我从零开始搭前后花了大概三周时间其中一半时间是在处理各种数据源的适配和异常情况。现在它每天稳定采集几十万条数据支撑着我的几个小项目和朋友的量化策略回测。如果你也在做类似的事情我的建议是先把核心链路跑通哪怕只接一个数据源、只支持一种资产类型然后再逐步扩展。一开始就追求大而全大概率会烂尾。