1. 金融数据服务从零搭建的完整思路1.1 为什么我要自己搭一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合服务。核心目标很明确把分散在不同公开渠道的行情数据、财报数据、宏观经济指标抓取回来做清洗、对齐、存储最后通过一套统一的接口对外提供查询能力。市面上现成的金融数据 API 不少但用下来总有几个绕不开的痛点。免费额度小得可怜稍微跑个回测就把配额烧光了付费方案对个人开发者又不太友好动辄每月几百上千更麻烦的是数据格式各家不一样今天用 A 家的行情、B 家的财报字段名、时间戳格式、复权方式全对不上光写适配层就够喝一壶。我最初只是想给自己的量化小策略找个稳定的数据源结果发现与其到处拼凑不如自己搭一套可控的服务。这套东西适合谁我认为有三类人值得参考。第一类是做量化研究但预算有限的个人开发者你需要一个能长期稳定跑、不被配额卡脖子的数据底座。第二类是想学习数据工程全链路的工程师从采集、清洗、存储到 API 服务这是一个麻雀虽小五脏俱全的实战项目。第三类是做金融类应用原型的团队早期不想在数据采购上投入太多先用自建服务把产品逻辑跑通。需要提前说明的是我全程只使用公开、合规的数据来源所有采集行为都遵守目标站点的访问规则控制请求频率不做任何越界的事情。这一点在后面讲采集策略时还会反复强调。1.2 整体架构是怎么拆的我把整个服务拆成了四层从下往上分别是采集层、处理层、存储层和服务层。这个分层不是拍脑袋定的而是踩过坑之后总结出来的。最早我把采集和清洗写在一个脚本里结果每次数据源改版整个脚本都得重写牵一发动全身。分层之后采集层只负责“把原始数据拿回来”处理层只负责“把脏数据洗干净”职责清晰了维护成本直线下降。采集层用 Python 写核心是几个针对不同数据源的适配器。每个适配器对外暴露统一的接口输入是标的代码和时间范围输出是标准化的原始记录。这样做的好处是新增一个数据源只需要写一个新的适配器不用动其他任何代码。处理层负责三件事字段标准化、时间对齐、异常值处理。字段标准化是把不同来源的字段名映射到统一的内部 schema时间对齐是解决不同数据源时间戳精度不一致的问题有的精确到秒有的只到天异常值处理则是识别并标记那些明显错误的数据点比如价格突然变成负数或者成交量出现天文数字。存储层我选了 PostgreSQL 加 TimescaleDB 扩展的组合。纯时序数据库比如 InfluxDB 在行情数据上确实快但财报、宏观指标这类关系型特征明显的数据用起来别扭。TimescaleDB 的好处是既有 PostgreSQL 的完整 SQL 能力和关系模型又有时序数据的分区和压缩优化一套数据库搞定所有类型的数据运维成本低很多。服务层用 FastAPI 搭对外提供 REST 接口。选 FastAPI 的理由很实际自带 OpenAPI 文档省了写接口文档的功夫异步支持好查询并发上来之后不会轻易被阻塞类型校验基于 Pydantic参数传错了直接返回清晰的错误信息不用自己写一堆校验逻辑。1.3 技术选型背后的取舍逻辑说说几个关键选型的取舍。语言层面选 Python 而不是 Go 或 Java主要考虑的是生态。金融数据处理绕不开 pandas、numpy 这些库用 Python 能直接复用换成其他语言要么找不到对应工具要么得自己造轮子。性能方面这套服务的瓶颈在数据采集和数据库查询不在计算逻辑本身Python 的性能完全够用。数据库选型前面提了这里补充一个细节。我最初试过用 SQLite 做存储开发阶段确实方便单文件、零配置。但数据量上来之后问题就暴露了并发写入性能差采集任务一多就锁表没有原生分区支持历史数据查询越来越慢。迁移到 PostgreSQL 之后这些问题一次性解决。TimescaleDB 的超表功能让行情数据按时间自动分区查询最近一个月的数据和查询三年前的数据响应时间几乎没差别。API 框架选 FastAPI 而不是 Flask 或 Django除了前面说的异步和文档优势还有一个考虑是依赖注入系统。金融数据服务里很多接口需要复用相同的数据库连接、缓存客户端、配置对象FastAPI 的依赖注入让这些资源的生命周期管理变得很干净测试的时候也容易替换成 mock 对象。缓存层我用了 Redis主要缓存两类东西一是热点查询结果比如最近几天的行情数据很多策略会反复读取二是采集任务的去重标记避免同一个数据在短时间内被重复抓取。Redis 的过期策略天然适合这种场景设个 TTL 就行不用自己写清理逻辑。2. 核心模块的细节拆解与实操要点2.1 数据采集适配器怎么写才稳采集是整个服务的地基地基不稳后面全白搭。我写适配器遵循几个原则。第一每个适配器是独立的类继承一个基类基类定义fetch方法的标准签名。第二适配器内部只做“请求加解析”不做任何数据转换原始数据原样返回。第三所有网络请求必须带重试和退避策略。重试策略这块我踩过坑。最早用的是简单的固定间隔重试结果遇到目标站点限流时越重试越被封。后来改成指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒以此类推同时加上随机抖动避免多个采集任务同时重试造成脉冲式压力。实测下来这套策略能把因限流导致的采集失败率降低八成以上。请求频率控制是另一个关键点。我的做法是在适配器基类里内置一个令牌桶限流器每个数据源配置独立的速率参数。比如某个数据源限制每分钟最多 60 次请求令牌桶就按这个速率补充令牌采集任务取不到令牌就等待。这样即使上层调度逻辑写得再激进实际发出的请求也不会超过限制。注意采集频率一定要保守设置宁可慢一点也不要触发目标站点的防护机制。我一般会把官方限制的 70% 作为自己的上限留出安全余量。解析环节要特别注意容错。网页结构或接口返回格式随时可能变解析代码不能假设字段一定存在。我的做法是用.get()而不是直接索引对关键字段做类型检查解析失败时记录原始响应内容到日志方便事后排查。有一次某个数据源把日期字段从2024-01-15改成了15/01/2024因为解析代码做了格式校验并记录了原始数据我十分钟就定位并修复了问题。2.2 数据清洗与标准化的关键细节原始数据拿回来只是第一步清洗才是真正花时间的活。我把清洗流程拆成几个独立的步骤每个步骤是一个纯函数输入输出都是 DataFrame方便单独测试和组合。字段映射是第一步。我定义了一套内部标准 schema比如行情数据统一用symbol、trade_date、open、high、low、close、volume这些字段名。每个数据源的适配器配一个映射字典把原始字段名翻译成标准字段名。映射字典放在配置文件里而不是硬编码在代码中数据源改字段名时改配置就行不用动代码。时间对齐比想象中麻烦。不同数据源的时间戳时区不一样有的用 UTC有的用交易所当地时间还有的用毫秒时间戳。我的处理方式是全部先转成 UTC 的datetime对象存储时统一用 UTC查询时再按需转换。日期字段则统一成date类型避免时间部分带来的干扰。这里有个细节夏令时切换那几天的数据特别容易出错我专门写了一个校验函数检查时间序列是否存在重复或缺失的小时。异常值处理我用的是组合策略。先做统计检测计算滚动窗口内的均值和标准差超出三倍标准差的点标记为可疑再做业务规则校验比如价格不能为负、最高价不能低于最低价、成交量不能为负。可疑点不会直接删除而是打上标记存入数据库查询时可以按需过滤。这样做的好处是保留了原始信息万一统计检测误判了还能追溯。缺失值处理要分情况。行情数据里偶尔缺一两天我一般用前值填充因为金融时间序列的连续性很重要。但如果是财报数据缺失填充就不合适了宁可留空也不要编造数据。这个判断逻辑我写在清洗配置里每种数据类型配不同的缺失值策略。2.3 存储方案的设计与优化数据库表结构的设计直接决定了后续查询的性能。行情数据我用的是 TimescaleDB 的超表按trade_date做时间分区每个分区存一个月的数据。分区的好处是查询时只需要扫描相关分区不用全表扫描。另外我对symbol字段建了索引因为大部分查询都会指定标的代码。财报数据用的是普通关系表因为它的访问模式跟行情完全不同。财报按季度更新数据量小但字段多查询时经常需要跨多个字段过滤。这种场景下普通 B-tree 索引比时序分区更合适。我给常用的过滤字段都建了索引比如报告期、标的代码、指标名称。数据写入我用的是批量插入加ON CONFLICT处理。采集任务一次可能拿回几百上千条记录逐条插入效率太低。批量插入配合事务几千条记录一两秒就能写完。ON CONFLICT用来处理重复数据同一个标的同一天的数据如果重复采集直接更新而不是报错保证幂等性。实操心得批量插入的批次大小要控制好太大容易撑爆内存太小又体现不出批量优势。我实测下来每批 500 到 1000 条是比较舒服的区间具体数值可以根据单条记录的大小调整。数据保留策略也要提前想好。行情数据我保留全部历史因为量化回测经常需要长周期数据。但采集日志、临时中间结果这些我设置了自动清理超过 30 天就删除。TimescaleDB 自带的数据保留策略功能可以自动完成这件事配一条规则就行不用自己写定时任务。3. 完整实操流程与核心环节实现3.1 环境搭建与依赖安装先把环境搭起来。我用的 Python 3.11这个版本在性能和稳定性上比较平衡。数据库用 PostgreSQL 15 加 TimescaleDB 2.x 扩展Redis 用 7.x 版本。操作系统我用的是 Ubuntu 22.04其他 Linux 发行版步骤类似macOS 也能跑Windows 建议用 WSL2。数据库安装完成后需要启用 TimescaleDB 扩展。登录 PostgreSQL 后执行CREATE EXTENSION IF NOT EXISTS timescaledb;然后创建业务数据库和用户CREATE DATABASE financial_services; CREATE USER fin_user WITH PASSWORD your_secure_password; GRANT ALL PRIVILEGES ON DATABASE financial_services TO fin_user;Python 依赖我整理了一个requirements.txt核心的几个是fastapi做 API 框架uvicorn做 ASGI 服务器sqlalchemy做 ORMpsycopg2-binary做 PostgreSQL 驱动pandas和numpy做数据处理httpx做异步 HTTP 请求redis做缓存客户端pydantic做数据校验apscheduler做定时任务调度。pip install -r requirements.txt项目目录结构我按功能划分financial-services/ ├── adapters/ # 数据源适配器 ├── cleaners/ # 数据清洗模块 ├── models/ # 数据库模型 ├── api/ # API 路由 ├── scheduler/ # 定时任务 ├── config/ # 配置文件 └── tests/ # 测试用例这个结构的好处是每个模块职责单一新人接手时能快速定位代码。配置文件我用 YAML 格式数据库连接、数据源参数、采集频率这些都放在里面不同环境用不同的配置文件代码里不出现任何硬编码的配置值。3.2 数据库表结构的创建行情数据表的建表语句是这样的CREATE TABLE market_data ( time TIMESTAMPTZ NOT NULL, symbol VARCHAR(20) NOT NULL, open NUMERIC(18, 6), high NUMERIC(18, 6), low NUMERIC(18, 6), close NUMERIC(18, 6), volume BIGINT, is_suspect BOOLEAN DEFAULT FALSE, source VARCHAR(50), PRIMARY KEY (time, symbol) ); SELECT create_hypertable(market_data, time, chunk_time_interval INTERVAL 1 month);价格字段用NUMERIC(18, 6)而不是FLOAT因为浮点数在金融计算中会有精度问题NUMERIC是精确十进制类型不会出现0.1 0.2 ! 0.3这种糟心事。is_suspect字段就是前面说的异常值标记查询时可以过滤掉。财报数据表CREATE TABLE financial_reports ( id SERIAL PRIMARY KEY, symbol VARCHAR(20) NOT NULL, report_date DATE NOT NULL, metric_name VARCHAR(100) NOT NULL, metric_value NUMERIC(24, 6), unit VARCHAR(20), source VARCHAR(50), created_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE (symbol, report_date, metric_name) );财报数据用“长表”结构每个指标一行而不是每个报告期一行宽表。这样做的好处是新增指标不用改表结构查询时用WHERE metric_name xxx过滤就行。代价是查询多个指标时需要做透视但 PostgreSQL 的crosstab或者应用层的 pandas 都能轻松处理。3.3 采集任务的调度与执行调度我用 APScheduler配置在scheduler模块里。行情数据每个交易日收盘后采集一次财报数据按季度采集宏观指标按月采集。每个任务配置独立的执行时间和重试策略。from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.triggers.cron import CronTrigger scheduler AsyncIOScheduler() scheduler.add_job( collect_market_data, CronTrigger(day_of_weekmon-fri, hour18, minute0), idmarket_data_daily, max_instances1, misfire_grace_time3600 )max_instances1保证同一个任务不会并发执行避免重复采集。misfire_grace_time设置容错窗口如果任务因为服务重启错过了执行时间一小时内还会补执行。采集任务的执行流程是这样的先从数据库读取需要采集的标的列表和最后采集日期确定本次需要采集的时间范围然后调用对应的适配器获取数据拿到原始数据后走清洗流程最后批量写入数据库。整个过程用日志记录关键节点方便排查问题。async def collect_market_data(): symbols get_active_symbols() for symbol in symbols: last_date get_last_collected_date(symbol) start_date last_date timedelta(days1) end_date date.today() if start_date end_date: continue raw_data await adapter.fetch(symbol, start_date, end_date) cleaned clean_market_data(raw_data) bulk_upsert(cleaned) logger.info(fCollected {len(cleaned)} records for {symbol})3.4 API 接口的设计与实现API 层用 FastAPI路由按数据类型分组。行情查询接口支持按标的、时间范围、复权方式过滤返回 JSON 格式的数据。财报查询接口支持按标的、报告期、指标名称过滤。from fastapi import FastAPI, Query, Depends from typing import Optional from datetime import date app FastAPI(titleFinancial Data Service, version1.0.0) app.get(/api/v1/market/{symbol}) async def get_market_data( symbol: str, start: date Query(...), end: date Query(...), adjust: Optional[str] Query(none, regex^(none|forward|backward)$), db Depends(get_db) ): data query_market_data(db, symbol, start, end, adjust) return {symbol: symbol, count: len(data), data: data}接口设计有几个细节值得说。第一路径参数用标的代码查询参数用时间范围和复权方式符合 RESTful 风格。第二复权方式用正则校验只允许none、forward、backward三个值传错了直接返回 422 错误。第三返回结构统一包含symbol、count、data三个字段前端处理起来方便。缓存策略我加在查询接口上。对于最近 30 天内的行情查询结果缓存到 RedisTTL 设 10 分钟。因为这段时间的数据基本不会变缓存命中率很高。历史数据不缓存因为访问频率低缓存反而浪费内存。async def get_cached_market_data(symbol, start, end): cache_key fmarket:{symbol}:{start}:{end} cached await redis.get(cache_key) if cached: return json.loads(cached) data query_from_db(symbol, start, end) if (date.today() - start).days 30: await redis.setex(cache_key, 600, json.dumps(data)) return data4. 常见问题与排查技巧实录4.1 采集环节的高频问题采集环节最常见的问题是请求被拒绝。表现是返回 403 或 429 状态码或者返回一个验证页面而不是数据。遇到这种情况先检查请求头是否完整很多站点会校验User-Agent、Referer这些字段。我一般会模拟正常浏览器的请求头但不会做任何伪装成特定用户或绕过验证的事情。如果请求头没问题还是被拒大概率是频率太高触发了限流。这时候要降低采集频率增大请求间隔。我前面提到的令牌桶限流器就是干这个的。另外可以错峰采集把不同数据源的采集任务分散到不同时间段避免同一时间集中请求。数据解析失败是另一个高频问题。表现是返回的数据结构跟预期不符解析代码抛异常。排查方法是先把原始响应保存下来人工看一眼结构到底长什么样。我习惯在解析失败时自动把原始响应写入一个debug目录文件名带上时间戳和数据源标识方便事后分析。避坑技巧解析代码里对每个字段的访问都做防御性检查用.get()加默认值对关键字段做类型转换和范围校验。宁可返回空值也不要让整个采集任务因为一个字段的问题而崩溃。4.2 数据质量问题的排查思路数据质量问题往往不会立刻暴露而是在使用过程中才被发现。我遇到过几次典型情况。一次是某天的收盘价明显偏离正常范围查下来发现是数据源当天返回了错误数据但格式完全正常解析环节没发现问题。后来我加了一个跨数据源交叉验证的逻辑同一个标的的价格如果跟另一个数据源差异超过 5%就标记为可疑。另一次是时间戳错位。某数据源返回的时间戳是交易所当地时间但我按 UTC 处理了导致数据整体偏移了几个小时。这个问题在日线数据上不明显但在分钟线数据上就是灾难。排查方法是把采集的数据跟已知正确的数据做时间对齐检查发现偏移后修正时区转换逻辑。数据重复也是常见问题。同一个标的同一天的数据被采集了两次如果表结构没有唯一约束就会产生重复记录。我的做法是在表上建唯一约束写入时用ON CONFLICT处理从数据库层面杜绝重复。另外在采集任务里记录最后采集日期避免重复采集相同时间段。4.3 性能问题的定位与优化服务跑起来之后性能问题主要集中在数据库查询上。表现是接口响应慢尤其是查询大时间范围或多标的的时候。定位方法是开启 PostgreSQL 的慢查询日志找出执行时间超过阈值的 SQL用EXPLAIN ANALYZE分析执行计划。常见的优化手段有几个。第一是加索引对查询条件里频繁出现的字段建索引。第二是优化查询语句避免SELECT *只取需要的字段避免在WHERE子句里对字段做函数运算这会导致索引失效。第三是用物化视图预计算常用的聚合结果比如日线数据的月度统计查询时直接读物化视图不用每次现算。缓存命中率低也会导致性能问题。排查方法是监控 Redis 的命中率指标如果低于 80% 就要分析原因。常见原因是缓存键设计不合理或者 TTL 设得太短。我一般会把热点数据的 TTL 设长一些同时用 LRU 淘汰策略保证内存不会无限增长。问题现象可能原因排查方法解决方案接口响应超过 2 秒缺少索引或查询未走索引EXPLAIN ANALYZE 分析执行计划补充索引或重写查询缓存命中率低于 80%缓存键设计不合理或 TTL 太短监控 Redis 命中率指标优化键设计延长热点数据 TTL采集任务频繁失败触发限流或请求头不完整检查返回状态码和响应内容降低频率补全请求头数据出现重复记录缺少唯一约束或采集范围重叠查询重复记录统计加唯一约束用 ON CONFLICT 处理时间序列出现偏移时区处理不一致对比已知正确数据的时间戳统一转 UTC 存储4.4 服务稳定性保障经验服务要长期稳定运行光靠功能正确还不够得有监控和告警。我用了几个轻量级的方案。采集任务的成功率、耗时、数据量这些指标写入日志用logrotate做日志轮转避免磁盘被写满。数据库连接池的状态通过 PostgreSQL 的pg_stat_activity视图监控连接数接近上限时告警。定时任务我用了一个简单的健康检查机制。每个任务执行完成后往 Redis 写一个心跳键键的 TTL 设为任务执行周期的两倍。如果某个任务的心跳键过期了说明它超过两个周期没执行触发告警。这个机制帮我及时发现过几次任务卡死的问题。实操心得服务刚上线时不要追求大而全先把核心的采集和查询跑通稳定运行一两周后再逐步加监控和优化。一上来就搞复杂了出了问题反而不好定位。数据备份也不能忽视。PostgreSQL 自带的pg_dump配合定时任务每天凌晨做一次全量备份保留最近 7 天。备份文件存到独立的磁盘或对象存储不要跟数据库放同一块盘。我吃过亏有一次磁盘故障数据库和备份一起没了只能重新采集花了两天才恢复。5. 后续可以扩展的方向这套服务跑了大半年基本满足了我自己的需求。如果后续要继续扩展我觉得有几个方向值得做。一是增加更多数据源尤其是另类数据比如舆情、供应链这些能给策略提供额外的信息维度。二是做数据版本管理每次数据更新都记录版本支持回溯到任意历史时点这对回测的严谨性很有帮助。三是把 API 层做成 GraphQL让前端可以按需查询字段减少不必要的数据传输。另外如果团队规模扩大可以考虑把采集和处理拆成独立的微服务用消息队列解耦。采集服务只管往队列里扔原始数据处理服务从队列消费并写入数据库。这样采集和处理可以独立扩缩容某个数据源出问题也不会影响其他数据源。不过对个人开发者来说当前的单体架构已经够用过早拆分会增加运维复杂度。我在实际使用中最大的体会是金融数据服务的核心难点不在技术而在数据的准确性和一致性。技术方案再漂亮数据错了就是白搭。所以我在数据校验上投入的精力比写采集和 API 的代码多得多。每次新增数据源我都会花时间做交叉验证确保数据跟权威来源对得上。这个习惯帮我避免了很多后续的麻烦。