1. 金融数据服务从零搭建的核心思路拆解1.1 为什么我要自己搭一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自用金融数据聚合与分发服务。它要解决的核心问题是行情数据、财报数据、宏观经济指标这些信息散落在不同渠道格式五花八门每次写策略或者做分析都要重新对接一遍重复劳动太多。这套服务做的事情就是把这些数据统一拉取、清洗、存储然后通过一套标准接口对外提供让上层的分析脚本、回测系统、看板工具都能用同一套方式取数。适合谁来参考如果你写过量化策略、做过投资组合分析、或者只是想让自己的记账工具自动同步一些公开的金融指标这套东西都能直接用。它不涉及任何交易执行纯粹是数据层面的基础设施。技术栈上我选的是 Python 加 FastAPI 做服务层PostgreSQL 做存储Redis 做缓存定时任务用 APScheduler。这套组合的好处是轻量、依赖少、本地就能跑起来不需要复杂的集群环境。为什么不用现成的商业数据接口成本是一方面更关键的是灵活度。商业接口的字段和频率是固定的我想加一个自定义的计算指标或者想把两个来源的数据做交叉验证就很难受。自建服务虽然前期投入多一些但后面每加一个数据源、每改一次口径都是改几行配置的事。这个取舍在后面会反复提到因为它决定了整个架构的设计方向。1.2 整体架构的分层逻辑我把整个服务分成四层从上到下依次是接口层、业务层、采集层、存储层。接口层负责对外暴露 RESTful API用 FastAPI 自动生成文档方便调试。业务层处理数据聚合、指标计算、权限校验这些逻辑。采集层是各个数据源的适配器每个源一个模块互不干扰。存储层就是 PostgreSQL 加 Redis 的组合前者存历史数据后者存热点数据。这样分层的原因很直接数据源是最容易变的部分。某个公开接口改了返回格式或者某个页面结构变了我只需要动采集层对应的那个适配器业务层和接口层完全不用碰。反过来如果我要加一个新的 API 端点也只动接口层。层与层之间通过明确的数据结构通信我定义了一个统一的Quote、FinancialReport、MacroIndicator这些数据类所有采集器返回的都是这些类型业务层不关心数据是从哪来的。这里有个经验不要一开始就追求大而全的抽象。我第一版的时候想设计一个万能的数据模型结果发现不同数据源的字段差异太大强行统一反而增加了很多转换代码。后来改成按数据类型分模型行情归行情财报归财报宏观归宏观每个模型只保留最核心的字段扩展字段用 JSONB 存一下子就清爽了。这个思路在后面每个模块里都有体现。1.3 技术选型背后的取舍选 FastAPI 而不是 Flask 或 Django主要看中三点原生异步支持、Pydantic 数据校验、自动生成 OpenAPI 文档。金融数据接口经常需要并发拉取多个源异步能省不少时间。Pydantic 的模型校验让采集层返回的数据在进入业务层之前就被强制检查脏数据很难漏进来。自动文档则省去了手写接口说明的功夫团队里其他人接手时直接看/docs就行。数据库选 PostgreSQL 而不是 MySQL核心原因是 JSONB 类型和窗口函数。金融数据里很多字段是不定长的比如财报的附注、行情的扩展属性用 JSONB 存既灵活又能建索引。窗口函数在计算移动平均、同比环比这些指标时特别顺手不用在应用层写一堆循环。Redis 主要用来缓存最近查询频繁的数据比如最新行情、当日指标减少数据库压力。定时任务用 APScheduler 而不是 Celery是因为这套服务是单机部署不需要分布式任务队列。APScheduler 配置简单和 FastAPI 跑在同一个进程里少维护一个组件。如果后面数据量上来了要拆出去再换 Celery 也不迟。这个“够用就好”的原则贯穿整个项目避免过度设计。2. 核心模块的细节解析与实操要点2.1 数据模型设计的关键决策数据模型是整个服务的地基设计不好后面改起来很痛苦。我拿行情数据举例说明思路。一条行情记录包含标的代码、时间戳、开高低收、成交量、成交额这些基础字段另外还有一个extra字段用 JSONB 存扩展信息。为什么这么设计因为不同市场、不同品种的行情字段不完全一样股票有复权因子期货有持仓量数字货币有买卖盘深度。如果每个都建独立表表数量会爆炸如果全塞进一个宽表又会有大量空字段。我的做法是核心字段固定扩展字段用 JSONB。核心字段建普通索引扩展字段按需建 GIN 索引。查询的时候常用条件走核心字段特殊需求走 JSONB 查询。这样既保证了通用查询的性能又保留了灵活性。实测下来千万级数据量下按标的加时间范围查询响应时间稳定在几十毫秒。财报数据模型稍微复杂一些。我把它拆成两张表一张存报告元信息报告期、公告日期、审计意见等一张存具体科目科目代码、科目名称、数值、单位。科目表用自关联的方式支持树形结构因为财报科目本身是有层级的比如“流动资产”下面有“货币资金”“应收账款”等。这样设计的好处是不管哪个市场的财报只要把科目映射到统一的科目树上就能做跨市场的对比分析。注意时间戳统一用 UTC 存储展示的时候再转本地时区。我一开始图省事存了本地时间后来做跨市场分析时被时区问题坑得很惨数据对不齐排查了半天才发现是存储层的问题。2.2 采集器的编写规范与避坑每个数据源写一个采集器放在collectors目录下文件名就是源的名字。采集器必须实现一个统一的接口包含fetch和parse两个方法。fetch负责发请求拿原始数据parse负责把原始数据转成标准数据类。这样拆分的好处是请求逻辑和解析逻辑分开测试哪一步出问题一目了然。写采集器有几个坑我踩过这里列出来。第一是请求频率控制公开接口一般都有隐性限流请求太密会被临时封禁。我的做法是在采集器基类里加一个令牌桶限流器每个源配置自己的速率比如每秒最多两次。第二是重试策略网络抖动导致的失败要重试但解析失败重试没意义要区分开。第三是数据去重同一个时间点的数据可能被重复采集入库时用唯一约束或者ON CONFLICT处理。解析环节最麻烦的是字段映射。不同源对同一个概念的叫法不一样比如“收盘价”有的叫close有的叫closing_price有的叫last。我的做法是每个采集器维护一个映射字典把源字段映射到标准字段。映射字典单独放一个配置文件改的时候不用动代码。这个设计在接入新源时特别省事复制一份配置改改就行。# 采集器基类的核心逻辑示意 class BaseCollector: rate_limit 2 # 每秒请求数 max_retries 3 def fetch(self, params): # 限流 重试逻辑 pass def parse(self, raw): # 字段映射 类型转换 pass def run(self, params): raw self.fetch(params) return self.parse(raw)2.3 缓存策略与数据一致性缓存用得好能大幅提升性能用不好就是数据不一致的源头。我的策略是分场景处理。最新行情这类数据变化快、查询频繁缓存时间设短一点比如 5 秒过期后重新拉取。历史数据变化少缓存时间可以设长比如 1 小时。财报数据基本不变缓存 24 小时都没问题。关键问题是缓存更新时机。我采用的是“写时更新”加“读时兜底”。采集器写入新数据时主动删除对应的缓存键下次查询时重新加载。同时查询接口里加一个兜底逻辑如果缓存里没有就去数据库查查到后写回缓存。这样即使删除缓存失败也不会导致数据永久不一致。提示缓存键的命名要有规律我用的格式是{数据类型}:{标的}:{时间粒度}比如quote:000001:daily。这样批量清理的时候可以用模式匹配不用一个个删。还有一个细节是缓存穿透问题。如果有人频繁查询一个不存在的标的每次都会打到数据库。我的做法是查不到的数据也缓存一个空值过期时间设短一点比如 30 秒。这样既能挡住恶意查询又不会因为数据后来出现了而长时间返回空。3. 实操过程与核心环节实现3.1 环境搭建与依赖安装先把基础环境跑起来。我用的 Python 3.11数据库 PostgreSQL 15Redis 7。依赖管理用requirements.txt核心依赖就几个fastapi、uvicorn、sqlalchemy、asyncpg、redis、apscheduler、httpx、pydantic。版本我都锁死了避免自动升级引入不兼容。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装依赖 pip install -r requirements.txt # 启动 PostgreSQL 和 Redis假设已安装 # 初始化数据库 python scripts/init_db.py数据库初始化脚本负责建表、建索引、插入基础配置数据。我把建表语句写在 SQL 文件里用脚本执行而不是用 ORM 的自动建表。原因是自动建表对索引和约束的控制不够精细而且生产环境改表结构时不好追踪。SQL 文件纳入版本管理每次变更都有记录。配置文件用.env加pydantic-settings管理。数据库连接串、Redis 地址、各数据源的密钥都放环境变量里代码里只引用配置对象。这样本地开发和部署环境的配置可以完全隔离不会把密钥提交到代码仓库。3.2 采集任务的调度配置调度用 APScheduler 的BackgroundScheduler在 FastAPI 启动时初始化。每个采集任务配置独立的触发器和参数。比如日线行情在每天收盘后跑一次分钟线行情每分钟跑一次财报数据在公告季每天跑两次。任务之间互相独立一个失败不影响其他。from apscheduler.schedulers.background import BackgroundScheduler scheduler BackgroundScheduler() # 日线行情每天 16:00 跑 scheduler.add_job( collect_daily_quotes, cron, hour16, minute0, iddaily_quotes ) # 分钟线行情交易时段每分钟跑 scheduler.add_job( collect_minute_quotes, cron, day_of_weekmon-fri, hour9-15, minute*, idminute_quotes ) scheduler.start()任务执行结果要记录日志成功失败都记方便排查。我用的是标准库的logging按天切分日志文件。关键指标比如采集条数、耗时、失败原因都结构化输出后面可以用脚本做监控告警。注意调度器启动时要加misfire_grace_time参数防止任务因为服务重启错过执行时间。我设的是 300 秒意思是任务错过执行时间后 5 分钟内还会补跑一次。3.3 接口层的实现与参数校验接口层用 FastAPI 的APIRouter分模块组织行情、财报、宏观各一个路由文件。每个端点用 Pydantic 模型定义请求参数和响应结构FastAPI 会自动做校验和文档生成。响应结构统一包一层包含code、message、data三个字段方便前端统一处理。from fastapi import APIRouter, Query from pydantic import BaseModel router APIRouter(prefix/api/v1) class QuoteResponse(BaseModel): code: str timestamp: str open: float high: float low: float close: float volume: int router.get(/quotes, response_modelQuoteResponse) async def get_quotes( symbol: str Query(..., description标的代码), start: str Query(..., description开始日期), end: str Query(..., description结束日期), granularity: str Query(daily, description时间粒度) ): # 业务逻辑 pass参数校验这块除了 Pydantic 自带的类型检查我还加了自定义校验器。比如日期格式必须是YYYY-MM-DD粒度只能是daily、minute、weekly这几个枚举值。校验失败时返回明确的错误信息而不是笼统的 400。这样调用方一看就知道哪里传错了。分页是必须的金融数据动辄几千上万条一次全返回既慢又占内存。我用的是游标分页而不是偏移分页因为偏移分页在数据量大时性能会下降。游标用时间戳加标的代码组合查询时带上游标条件返回结果里带下一页的游标。实测下来千万级数据下分页查询稳定在百毫秒以内。3.4 数据清洗与异常处理采集回来的原始数据不能直接入库要先过一遍清洗。清洗规则包括去除重复记录、填充缺失字段、修正明显错误的值、统一单位。比如成交量有的源用股有的用手统一转成股。价格有的源带复权有的不带统一转成不复权复权因子单独存。异常处理分两类可恢复的和不可恢复的。网络超时、临时限流属于可恢复的重试几次一般能成功。数据格式变了、字段缺失属于不可恢复的重试也没用直接记录错误并告警。我的做法是在采集器里区分这两类异常可恢复的走重试逻辑不可恢复的直接抛到上层由调度器记录并通知。class RecoverableError(Exception): pass class FatalError(Exception): pass def collect_with_retry(collector, params): for attempt in range(collector.max_retries): try: return collector.run(params) except RecoverableError as e: if attempt collector.max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 except FatalError: raise清洗后的数据入库前还要做一次校验确保必填字段不为空、数值在合理范围内。比如价格不能为负成交量不能是小数。校验不通过的数据记录到异常表人工排查后再决定是丢弃还是修正。4. 常见问题与排查技巧实录4.1 采集失败的典型原因与排查路径采集失败是最常见的问题原因五花八门。我整理了一个排查顺序按这个顺序走基本能定位到问题。第一步看日志里的错误类型是网络错误还是解析错误。网络错误的话检查目标地址是否可达、是否被限流、请求头是否完整。解析错误的话把原始响应打出来看结构是不是变了。第二步看是不是所有源都失败还是单个源失败。如果所有源都失败大概率是网络或者服务本身的问题。如果单个源失败那就是这个源特有的问题重点查它的请求参数和解析逻辑。第三步看失败是不是有规律比如每天固定时间失败那可能是目标源在那个时间段维护。提示我习惯在采集器里把原始响应存一份到本地文件保留最近 7 天。排查解析问题时直接看原始响应比重新发请求快得多也不会因为重复请求触发限流。还有一个隐蔽的问题是编码。有些源返回的响应头里声明的编码和实际编码不一致导致解析出来是乱码。我的做法是统一按 UTF-8 解码失败时尝试 GBK再失败就记录原始字节。这个坑我在接入一个老接口时踩过排查了大半天才发现是编码问题。4.2 数据不一致的定位方法数据不一致通常表现为同一个指标在不同接口返回的值不一样或者历史数据和最新数据对不上。定位这类问题我一般从三个方向查。第一是查采集时间看两条数据是不是同一批次采集的不同批次可能因为源更新了导致差异。第二是查清洗规则看两条数据是不是走了不同的清洗路径。第三是查缓存看是不是缓存没更新导致返回了旧数据。我遇到过一次典型的不一致日线行情接口返回的收盘价和分钟线聚合出来的收盘价对不上。排查后发现是分钟线采集有缺失聚合时用了不完整的数据。解决办法是在聚合逻辑里加一个完整性检查分钟线数量不足预期时标记为不完整不参与聚合。这个检查后来成了标配所有聚合计算前都先做完整性校验。问题现象可能原因排查方法同一指标不同接口值不同采集批次不同或清洗路径不同对比采集时间和清洗日志历史数据与最新数据对不上缓存未更新或源数据修正清缓存后重查对比源数据聚合结果偏差大基础数据缺失或重复检查基础数据完整性和唯一性查询返回空但数据库有数据缓存穿透或查询条件错误绕过缓存直查数据库4.3 性能瓶颈的识别与优化服务跑起来后性能瓶颈一般出现在两个地方数据库查询和采集任务。数据库查询慢的话先看执行计划确认索引有没有用上。我遇到过 JSONB 字段查询没走索引的情况原因是查询条件里用了函数导致索引失效。改成表达式索引后查询时间从几秒降到几十毫秒。采集任务慢的话先看是网络慢还是解析慢。网络慢的话考虑并发拉取用asyncio.gather同时请求多个源。解析慢的话看是不是用了低效的循环或者正则。我有个采集器解析 XML 用了ElementTree的迭代方式数据量大时很慢换成lxml后快了十倍。注意并发拉取要控制并发数不是越大越好。我一般设 5 到 10 个并发太多了反而因为上下文切换和限流导致整体变慢。这个值要根据目标源的承受能力和本地网络情况调。还有一个容易忽略的瓶颈是日志写入。如果日志级别设成 DEBUG高频采集任务会产生大量日志磁盘 IO 成为瓶颈。生产环境我把日志级别设成 INFO只在关键节点打日志DEBUG 日志按需开启。4.4 常见问题速查表问题排查步骤解决方案采集任务不执行检查调度器是否启动、任务是否注册重启服务确认任务配置数据入库报唯一约束冲突检查是否有重复采集用 ON CONFLICT 忽略或更新接口响应超时检查数据库查询和缓存命中率加索引、优化查询、增加缓存内存占用持续增长检查是否有未释放的连接或缓存用连接池设置缓存过期时间时区导致数据错位检查存储和展示的时区设置统一 UTC 存储展示时转换5. 扩展方向与个人实操体会5.1 后续可以怎么扩展这套服务目前覆盖了行情、财报、宏观三类数据后面可以往几个方向扩。一是加数据源比如接入更多市场的行情或者加另类数据如舆情、供应链数据。加源的成本很低写个采集器加个配置就行。二是加计算层在业务层里做更多衍生指标的计算比如波动率、相关性、因子暴露。这些计算可以做成插件按需启用。三是加通知能力数据异常或者采集失败时推送到常用工具。我目前用的是邮件通知后面可以加更多渠道。四是加权限控制如果多人使用需要区分数据访问权限。FastAPI 的依赖注入做这个很方便加个中间件校验 token 就行。提示扩展的时候尽量保持接口稳定新增功能用新端点不要改老端点的行为。这样已有的调用方不会因为升级而挂掉。5.2 我踩过的几个印象深刻的坑第一个坑是数据库连接池配置。一开始没设连接池上限采集任务并发高的时候把数据库连接占满了导致接口查询拿不到连接。后来设了pool_size和max_overflow问题解决。这个参数要根据并发量和数据库承受能力调我设的是 10 和 20。第二个坑是时间戳精度。不同源的时间戳精度不一样有的到秒有的到毫秒有的到微秒。入库时没统一导致同一时间点的数据被当成不同记录。后来统一转成毫秒精度问题解决。这个细节在跨源对比时特别重要。第三个坑是异常数据的传播。有个源偶尔会返回价格为 0 的记录清洗时没过滤导致后面计算收益率时出现无穷大。后来在清洗规则里加了价格必须大于 0 的校验并且把异常记录单独存表方便回溯。5.3 给准备动手的朋友几句实在话如果你打算搭一套类似的服务我的建议是先跑通最小闭环再逐步完善。最小闭环就是一个数据源、一个采集任务、一张表、一个查询接口。把这四个环节跑通后面加源加功能都是在这个骨架上扩展。不要一开始就设计复杂的架构很多需求是跑起来之后才发现的。另外数据质量比数据数量重要。与其接十个源但每个都有缺失和错误不如先把两三个源做扎实。清洗和校验的代码值得多花时间写这部分偷懒后面会加倍还回来。我现在的清洗规则有二十多条每一条都是踩坑后加的。最后日志和监控要尽早做。服务跑起来后出问题是必然的有完善的日志和监控排查效率天差地别。我用的方案很简单日志写文件加按天切分关键指标写数据库加定时检查够用了。等规模大了再上专业监控工具也不迟。 
长河君 · 2026/11/4 10:46:59 · 阅读 1.2k 从零搭建金融数据服务：架构设计与采集器避坑指南
1. 金融数据服务从零搭建的核心思路拆解1.1 为什么我要自己搭一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自用金融数据聚合与分发服务。它要解决的核心问题是行情数据、财报数据、宏观经济指标这些信息散落在不同渠道格式五花八门每次写策略或者做分析都要重新对接一遍重复劳动太多。这套服务做的事情就是把这些数据统一拉取、清洗、存储然后通过一套标准接口对外提供让上层的分析脚本、回测系统、看板工具都能用同一套方式取数。适合谁来参考如果你写过量化策略、做过投资组合分析、或者只是想让自己的记账工具自动同步一些公开的金融指标这套东西都能直接用。它不涉及任何交易执行纯粹是数据层面的基础设施。技术栈上我选的是 Python 加 FastAPI 做服务层PostgreSQL 做存储Redis 做缓存定时任务用 APScheduler。这套组合的好处是轻量、依赖少、本地就能跑起来不需要复杂的集群环境。为什么不用现成的商业数据接口成本是一方面更关键的是灵活度。商业接口的字段和频率是固定的我想加一个自定义的计算指标或者想把两个来源的数据做交叉验证就很难受。自建服务虽然前期投入多一些但后面每加一个数据源、每改一次口径都是改几行配置的事。这个取舍在后面会反复提到因为它决定了整个架构的设计方向。1.2 整体架构的分层逻辑我把整个服务分成四层从上到下依次是接口层、业务层、采集层、存储层。接口层负责对外暴露 RESTful API用 FastAPI 自动生成文档方便调试。业务层处理数据聚合、指标计算、权限校验这些逻辑。采集层是各个数据源的适配器每个源一个模块互不干扰。存储层就是 PostgreSQL 加 Redis 的组合前者存历史数据后者存热点数据。这样分层的原因很直接数据源是最容易变的部分。某个公开接口改了返回格式或者某个页面结构变了我只需要动采集层对应的那个适配器业务层和接口层完全不用碰。反过来如果我要加一个新的 API 端点也只动接口层。层与层之间通过明确的数据结构通信我定义了一个统一的Quote、FinancialReport、MacroIndicator这些数据类所有采集器返回的都是这些类型业务层不关心数据是从哪来的。这里有个经验不要一开始就追求大而全的抽象。我第一版的时候想设计一个万能的数据模型结果发现不同数据源的字段差异太大强行统一反而增加了很多转换代码。后来改成按数据类型分模型行情归行情财报归财报宏观归宏观每个模型只保留最核心的字段扩展字段用 JSONB 存一下子就清爽了。这个思路在后面每个模块里都有体现。1.3 技术选型背后的取舍选 FastAPI 而不是 Flask 或 Django主要看中三点原生异步支持、Pydantic 数据校验、自动生成 OpenAPI 文档。金融数据接口经常需要并发拉取多个源异步能省不少时间。Pydantic 的模型校验让采集层返回的数据在进入业务层之前就被强制检查脏数据很难漏进来。自动文档则省去了手写接口说明的功夫团队里其他人接手时直接看/docs就行。数据库选 PostgreSQL 而不是 MySQL核心原因是 JSONB 类型和窗口函数。金融数据里很多字段是不定长的比如财报的附注、行情的扩展属性用 JSONB 存既灵活又能建索引。窗口函数在计算移动平均、同比环比这些指标时特别顺手不用在应用层写一堆循环。Redis 主要用来缓存最近查询频繁的数据比如最新行情、当日指标减少数据库压力。定时任务用 APScheduler 而不是 Celery是因为这套服务是单机部署不需要分布式任务队列。APScheduler 配置简单和 FastAPI 跑在同一个进程里少维护一个组件。如果后面数据量上来了要拆出去再换 Celery 也不迟。这个“够用就好”的原则贯穿整个项目避免过度设计。2. 核心模块的细节解析与实操要点2.1 数据模型设计的关键决策数据模型是整个服务的地基设计不好后面改起来很痛苦。我拿行情数据举例说明思路。一条行情记录包含标的代码、时间戳、开高低收、成交量、成交额这些基础字段另外还有一个extra字段用 JSONB 存扩展信息。为什么这么设计因为不同市场、不同品种的行情字段不完全一样股票有复权因子期货有持仓量数字货币有买卖盘深度。如果每个都建独立表表数量会爆炸如果全塞进一个宽表又会有大量空字段。我的做法是核心字段固定扩展字段用 JSONB。核心字段建普通索引扩展字段按需建 GIN 索引。查询的时候常用条件走核心字段特殊需求走 JSONB 查询。这样既保证了通用查询的性能又保留了灵活性。实测下来千万级数据量下按标的加时间范围查询响应时间稳定在几十毫秒。财报数据模型稍微复杂一些。我把它拆成两张表一张存报告元信息报告期、公告日期、审计意见等一张存具体科目科目代码、科目名称、数值、单位。科目表用自关联的方式支持树形结构因为财报科目本身是有层级的比如“流动资产”下面有“货币资金”“应收账款”等。这样设计的好处是不管哪个市场的财报只要把科目映射到统一的科目树上就能做跨市场的对比分析。注意时间戳统一用 UTC 存储展示的时候再转本地时区。我一开始图省事存了本地时间后来做跨市场分析时被时区问题坑得很惨数据对不齐排查了半天才发现是存储层的问题。2.2 采集器的编写规范与避坑每个数据源写一个采集器放在collectors目录下文件名就是源的名字。采集器必须实现一个统一的接口包含fetch和parse两个方法。fetch负责发请求拿原始数据parse负责把原始数据转成标准数据类。这样拆分的好处是请求逻辑和解析逻辑分开测试哪一步出问题一目了然。写采集器有几个坑我踩过这里列出来。第一是请求频率控制公开接口一般都有隐性限流请求太密会被临时封禁。我的做法是在采集器基类里加一个令牌桶限流器每个源配置自己的速率比如每秒最多两次。第二是重试策略网络抖动导致的失败要重试但解析失败重试没意义要区分开。第三是数据去重同一个时间点的数据可能被重复采集入库时用唯一约束或者ON CONFLICT处理。解析环节最麻烦的是字段映射。不同源对同一个概念的叫法不一样比如“收盘价”有的叫close有的叫closing_price有的叫last。我的做法是每个采集器维护一个映射字典把源字段映射到标准字段。映射字典单独放一个配置文件改的时候不用动代码。这个设计在接入新源时特别省事复制一份配置改改就行。# 采集器基类的核心逻辑示意 class BaseCollector: rate_limit 2 # 每秒请求数 max_retries 3 def fetch(self, params): # 限流 重试逻辑 pass def parse(self, raw): # 字段映射 类型转换 pass def run(self, params): raw self.fetch(params) return self.parse(raw)2.3 缓存策略与数据一致性缓存用得好能大幅提升性能用不好就是数据不一致的源头。我的策略是分场景处理。最新行情这类数据变化快、查询频繁缓存时间设短一点比如 5 秒过期后重新拉取。历史数据变化少缓存时间可以设长比如 1 小时。财报数据基本不变缓存 24 小时都没问题。关键问题是缓存更新时机。我采用的是“写时更新”加“读时兜底”。采集器写入新数据时主动删除对应的缓存键下次查询时重新加载。同时查询接口里加一个兜底逻辑如果缓存里没有就去数据库查查到后写回缓存。这样即使删除缓存失败也不会导致数据永久不一致。提示缓存键的命名要有规律我用的格式是{数据类型}:{标的}:{时间粒度}比如quote:000001:daily。这样批量清理的时候可以用模式匹配不用一个个删。还有一个细节是缓存穿透问题。如果有人频繁查询一个不存在的标的每次都会打到数据库。我的做法是查不到的数据也缓存一个空值过期时间设短一点比如 30 秒。这样既能挡住恶意查询又不会因为数据后来出现了而长时间返回空。3. 实操过程与核心环节实现3.1 环境搭建与依赖安装先把基础环境跑起来。我用的 Python 3.11数据库 PostgreSQL 15Redis 7。依赖管理用requirements.txt核心依赖就几个fastapi、uvicorn、sqlalchemy、asyncpg、redis、apscheduler、httpx、pydantic。版本我都锁死了避免自动升级引入不兼容。# 创建虚拟环境 python -m venv venv source venv/bin/activate # 安装依赖 pip install -r requirements.txt # 启动 PostgreSQL 和 Redis假设已安装 # 初始化数据库 python scripts/init_db.py数据库初始化脚本负责建表、建索引、插入基础配置数据。我把建表语句写在 SQL 文件里用脚本执行而不是用 ORM 的自动建表。原因是自动建表对索引和约束的控制不够精细而且生产环境改表结构时不好追踪。SQL 文件纳入版本管理每次变更都有记录。配置文件用.env加pydantic-settings管理。数据库连接串、Redis 地址、各数据源的密钥都放环境变量里代码里只引用配置对象。这样本地开发和部署环境的配置可以完全隔离不会把密钥提交到代码仓库。3.2 采集任务的调度配置调度用 APScheduler 的BackgroundScheduler在 FastAPI 启动时初始化。每个采集任务配置独立的触发器和参数。比如日线行情在每天收盘后跑一次分钟线行情每分钟跑一次财报数据在公告季每天跑两次。任务之间互相独立一个失败不影响其他。from apscheduler.schedulers.background import BackgroundScheduler scheduler BackgroundScheduler() # 日线行情每天 16:00 跑 scheduler.add_job( collect_daily_quotes, cron, hour16, minute0, iddaily_quotes ) # 分钟线行情交易时段每分钟跑 scheduler.add_job( collect_minute_quotes, cron, day_of_weekmon-fri, hour9-15, minute*, idminute_quotes ) scheduler.start()任务执行结果要记录日志成功失败都记方便排查。我用的是标准库的logging按天切分日志文件。关键指标比如采集条数、耗时、失败原因都结构化输出后面可以用脚本做监控告警。注意调度器启动时要加misfire_grace_time参数防止任务因为服务重启错过执行时间。我设的是 300 秒意思是任务错过执行时间后 5 分钟内还会补跑一次。3.3 接口层的实现与参数校验接口层用 FastAPI 的APIRouter分模块组织行情、财报、宏观各一个路由文件。每个端点用 Pydantic 模型定义请求参数和响应结构FastAPI 会自动做校验和文档生成。响应结构统一包一层包含code、message、data三个字段方便前端统一处理。from fastapi import APIRouter, Query from pydantic import BaseModel router APIRouter(prefix/api/v1) class QuoteResponse(BaseModel): code: str timestamp: str open: float high: float low: float close: float volume: int router.get(/quotes, response_modelQuoteResponse) async def get_quotes( symbol: str Query(..., description标的代码), start: str Query(..., description开始日期), end: str Query(..., description结束日期), granularity: str Query(daily, description时间粒度) ): # 业务逻辑 pass参数校验这块除了 Pydantic 自带的类型检查我还加了自定义校验器。比如日期格式必须是YYYY-MM-DD粒度只能是daily、minute、weekly这几个枚举值。校验失败时返回明确的错误信息而不是笼统的 400。这样调用方一看就知道哪里传错了。分页是必须的金融数据动辄几千上万条一次全返回既慢又占内存。我用的是游标分页而不是偏移分页因为偏移分页在数据量大时性能会下降。游标用时间戳加标的代码组合查询时带上游标条件返回结果里带下一页的游标。实测下来千万级数据下分页查询稳定在百毫秒以内。3.4 数据清洗与异常处理采集回来的原始数据不能直接入库要先过一遍清洗。清洗规则包括去除重复记录、填充缺失字段、修正明显错误的值、统一单位。比如成交量有的源用股有的用手统一转成股。价格有的源带复权有的不带统一转成不复权复权因子单独存。异常处理分两类可恢复的和不可恢复的。网络超时、临时限流属于可恢复的重试几次一般能成功。数据格式变了、字段缺失属于不可恢复的重试也没用直接记录错误并告警。我的做法是在采集器里区分这两类异常可恢复的走重试逻辑不可恢复的直接抛到上层由调度器记录并通知。class RecoverableError(Exception): pass class FatalError(Exception): pass def collect_with_retry(collector, params): for attempt in range(collector.max_retries): try: return collector.run(params) except RecoverableError as e: if attempt collector.max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 except FatalError: raise清洗后的数据入库前还要做一次校验确保必填字段不为空、数值在合理范围内。比如价格不能为负成交量不能是小数。校验不通过的数据记录到异常表人工排查后再决定是丢弃还是修正。4. 常见问题与排查技巧实录4.1 采集失败的典型原因与排查路径采集失败是最常见的问题原因五花八门。我整理了一个排查顺序按这个顺序走基本能定位到问题。第一步看日志里的错误类型是网络错误还是解析错误。网络错误的话检查目标地址是否可达、是否被限流、请求头是否完整。解析错误的话把原始响应打出来看结构是不是变了。第二步看是不是所有源都失败还是单个源失败。如果所有源都失败大概率是网络或者服务本身的问题。如果单个源失败那就是这个源特有的问题重点查它的请求参数和解析逻辑。第三步看失败是不是有规律比如每天固定时间失败那可能是目标源在那个时间段维护。提示我习惯在采集器里把原始响应存一份到本地文件保留最近 7 天。排查解析问题时直接看原始响应比重新发请求快得多也不会因为重复请求触发限流。还有一个隐蔽的问题是编码。有些源返回的响应头里声明的编码和实际编码不一致导致解析出来是乱码。我的做法是统一按 UTF-8 解码失败时尝试 GBK再失败就记录原始字节。这个坑我在接入一个老接口时踩过排查了大半天才发现是编码问题。4.2 数据不一致的定位方法数据不一致通常表现为同一个指标在不同接口返回的值不一样或者历史数据和最新数据对不上。定位这类问题我一般从三个方向查。第一是查采集时间看两条数据是不是同一批次采集的不同批次可能因为源更新了导致差异。第二是查清洗规则看两条数据是不是走了不同的清洗路径。第三是查缓存看是不是缓存没更新导致返回了旧数据。我遇到过一次典型的不一致日线行情接口返回的收盘价和分钟线聚合出来的收盘价对不上。排查后发现是分钟线采集有缺失聚合时用了不完整的数据。解决办法是在聚合逻辑里加一个完整性检查分钟线数量不足预期时标记为不完整不参与聚合。这个检查后来成了标配所有聚合计算前都先做完整性校验。问题现象可能原因排查方法同一指标不同接口值不同采集批次不同或清洗路径不同对比采集时间和清洗日志历史数据与最新数据对不上缓存未更新或源数据修正清缓存后重查对比源数据聚合结果偏差大基础数据缺失或重复检查基础数据完整性和唯一性查询返回空但数据库有数据缓存穿透或查询条件错误绕过缓存直查数据库4.3 性能瓶颈的识别与优化服务跑起来后性能瓶颈一般出现在两个地方数据库查询和采集任务。数据库查询慢的话先看执行计划确认索引有没有用上。我遇到过 JSONB 字段查询没走索引的情况原因是查询条件里用了函数导致索引失效。改成表达式索引后查询时间从几秒降到几十毫秒。采集任务慢的话先看是网络慢还是解析慢。网络慢的话考虑并发拉取用asyncio.gather同时请求多个源。解析慢的话看是不是用了低效的循环或者正则。我有个采集器解析 XML 用了ElementTree的迭代方式数据量大时很慢换成lxml后快了十倍。注意并发拉取要控制并发数不是越大越好。我一般设 5 到 10 个并发太多了反而因为上下文切换和限流导致整体变慢。这个值要根据目标源的承受能力和本地网络情况调。还有一个容易忽略的瓶颈是日志写入。如果日志级别设成 DEBUG高频采集任务会产生大量日志磁盘 IO 成为瓶颈。生产环境我把日志级别设成 INFO只在关键节点打日志DEBUG 日志按需开启。4.4 常见问题速查表问题排查步骤解决方案采集任务不执行检查调度器是否启动、任务是否注册重启服务确认任务配置数据入库报唯一约束冲突检查是否有重复采集用 ON CONFLICT 忽略或更新接口响应超时检查数据库查询和缓存命中率加索引、优化查询、增加缓存内存占用持续增长检查是否有未释放的连接或缓存用连接池设置缓存过期时间时区导致数据错位检查存储和展示的时区设置统一 UTC 存储展示时转换5. 扩展方向与个人实操体会5.1 后续可以怎么扩展这套服务目前覆盖了行情、财报、宏观三类数据后面可以往几个方向扩。一是加数据源比如接入更多市场的行情或者加另类数据如舆情、供应链数据。加源的成本很低写个采集器加个配置就行。二是加计算层在业务层里做更多衍生指标的计算比如波动率、相关性、因子暴露。这些计算可以做成插件按需启用。三是加通知能力数据异常或者采集失败时推送到常用工具。我目前用的是邮件通知后面可以加更多渠道。四是加权限控制如果多人使用需要区分数据访问权限。FastAPI 的依赖注入做这个很方便加个中间件校验 token 就行。提示扩展的时候尽量保持接口稳定新增功能用新端点不要改老端点的行为。这样已有的调用方不会因为升级而挂掉。5.2 我踩过的几个印象深刻的坑第一个坑是数据库连接池配置。一开始没设连接池上限采集任务并发高的时候把数据库连接占满了导致接口查询拿不到连接。后来设了pool_size和max_overflow问题解决。这个参数要根据并发量和数据库承受能力调我设的是 10 和 20。第二个坑是时间戳精度。不同源的时间戳精度不一样有的到秒有的到毫秒有的到微秒。入库时没统一导致同一时间点的数据被当成不同记录。后来统一转成毫秒精度问题解决。这个细节在跨源对比时特别重要。第三个坑是异常数据的传播。有个源偶尔会返回价格为 0 的记录清洗时没过滤导致后面计算收益率时出现无穷大。后来在清洗规则里加了价格必须大于 0 的校验并且把异常记录单独存表方便回溯。5.3 给准备动手的朋友几句实在话如果你打算搭一套类似的服务我的建议是先跑通最小闭环再逐步完善。最小闭环就是一个数据源、一个采集任务、一张表、一个查询接口。把这四个环节跑通后面加源加功能都是在这个骨架上扩展。不要一开始就设计复杂的架构很多需求是跑起来之后才发现的。另外数据质量比数据数量重要。与其接十个源但每个都有缺失和错误不如先把两三个源做扎实。清洗和校验的代码值得多花时间写这部分偷懒后面会加倍还回来。我现在的清洗规则有二十多条每一条都是踩坑后加的。最后日志和监控要尽早做。服务跑起来后出问题是必然的有完善的日志和监控排查效率天差地别。我用的方案很简单日志写文件加按天切分关键指标写数据库加定时检查够用了。等规模大了再上专业监控工具也不迟。