1. 金融数据服务从零搭建的完整思路1.1 为什么我要自己动手做一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合与分发服务。核心能力就三件事把外部行情、财报、宏观经济等数据抓回来做清洗和标准化再通过统一的 API 接口吐给前端或者策略脚本用。市面上现成的金融数据 API 不少但用下来总有几个绕不开的痛点。免费额度小得可怜稍微高频一点的请求就被限流付费方案按调用次数计费做回测的时候动辄几十万次请求账单直接起飞更麻烦的是数据格式各家不一样今天用 A 家的行情、B 家的财报字段名、时间戳格式、复权方式全对不上光写适配层就能耗掉大半精力。我最初只是想给自己的一套小策略脚本找个稳定的数据源结果越做越大最后干脆把它整理成一个独立的服务。这套东西适合谁如果你是会写点代码、想自己掌控数据管道的独立开发者或者小团队里需要一套内部数据中台但预算有限那这套思路可以直接抄。它不追求机构级的低延迟和高可用但胜在可控、可改、成本低一台普通云主机就能跑起来。1.2 整体架构怎么拆三层分离的设计取舍我把整个服务拆成了三层采集层、存储层、服务层。这个拆法不是拍脑袋定的是被现实逼出来的。最早我是把抓取和 API 写在一个进程里结果发现两个问题。第一抓取任务经常因为网络抖动或者目标站点改版而卡住一卡就把整个 API 服务拖死了第二行情数据是高频写入的而 API 查询是高频读取的两者混在一起数据库锁竞争严重查询延迟忽高忽低。拆成三层之后采集层只管把数据弄回来写进库服务层只管从库里读数据往外吐中间用存储层解耦。采集层挂了不影响查询查询压力大也不影响采集节奏。这个设计的好处是每一层都可以独立扩容和重启坏处是数据从采集到可查询之间会有几秒到几分钟的延迟——对于做日线级别策略的我来说完全可以接受但如果你要做分钟级甚至秒级的实时交易这套架构就不合适了得换成消息队列加流处理的方案。具体到技术选型采集层我用 Python 写因为处理各种数据格式、做字段映射实在太方便了存储层用 PostgreSQL 加 TimescaleDB 扩展时序数据用 hypertable 存普通的关系型数据用普通表存一个库搞定两种需求服务层用 FastAPI自带异步和文档生成写起来快。这套组合的核心理念是用成熟工具解决成熟问题不为了炫技引入一堆中间件。1.3 数据源的选择逻辑与合规边界数据源这块要特别小心。我的原则是只使用公开的、允许程序化访问的接口。具体来说分三类。第一类是官方开放 API比如一些交易所和统计机构提供的公开数据接口这类最稳有文档、有配额说明按规矩调用就行。第二类是公开网页上的结构化数据比如财报页面这类要注意目标站点的 robots 协议和使用条款控制请求频率别给人添麻烦。第三类是自己积累的历史数据比如平时手动导出备份的行情文件。这里必须强调一点任何数据服务都要把合规放在第一位。不要试图绕过网站的访问限制不要高频轰炸别人的服务器不要抓取明确声明禁止抓取的内容。我给自己定的规矩是单域名请求间隔不低于 1 秒遇到 429 状态码就指数退避连续失败三次就暂停该源一小时。这些限制看起来保守但能保证服务长期稳定运行不会哪天突然被封。2. 核心模块的细节拆解与实操要点2.1 采集层如何写出抗造的抓取任务采集层是整个服务里最容易出问题的部分因为外部环境不可控。我踩过的坑包括目标站点突然改版导致解析失败、网络超时导致任务卡死、重复抓取导致数据重复写入。针对这些问题我总结了一套固定的写法。每个采集任务都是一个独立的函数接受参数、返回标准化的数据结构中间的任何异常都在函数内部捕获并记录绝不往外抛。任务本身不关心调度调度交给外部的定时器。这样做的好处是单个任务失败不会影响其他任务排查问题也简单看日志就知道是哪个源出了问题。import time import requests from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min2, max30)) def fetch_quote(symbol: str) - dict: url fhttps://api.example.com/quote/{symbol} resp requests.get(url, timeout10) if resp.status_code 429: raise RuntimeError(rate limited) resp.raise_for_status() raw resp.json() return normalize_quote(raw)上面这段代码里tenacity负责重试指数退避的等待时间从 2 秒开始翻倍最多等 30 秒。timeout10是必须的没有超时设置的请求就是定时炸弹。normalize_quote是标准化函数把不同来源的字段统一成内部格式。注意重试次数不要设太多三次足够。重试太多次反而会加重目标站点的负担也拖慢自己的任务队列。遇到持续失败就让它失败记录下来人工排查。标准化这块我定义了一套内部 schema所有数据源抓回来之后都要转成这个格式。以行情为例核心字段就六个symbol标的代码、ts时间戳统一用 UTC 毫秒、open、high、low、close成交量单独一个字段。时间戳统一是重中之重我见过太多因为时区没对齐导致数据错位的案例。2.2 存储层时序数据表设计的几个关键决策存储层的设计直接决定了查询性能。我用 TimescaleDB 的 hypertable 来存行情数据按时间分区。这里有几个参数需要仔细调。第一个是分区间隔chunk_time_interval。默认是 7 天但对于日线数据来说太细了会产生大量小分区反而拖慢查询。我的经验是日线数据用 1 年一个分区分钟线用 1 个月tick 级数据用 1 天。判断标准是每个分区的大小控制在 100MB 到 1GB 之间比较合适。SELECT create_hypertable( quotes_daily, ts, chunk_time_interval INTERVAL 1 year );第二个是索引。时序数据最常用的查询模式是某个标的在某段时间范围内的数据所以复合索引(symbol, ts DESC)是必须的。注意顺序不能反symbol在前是因为它的区分度高先按标的过滤再按时间范围扫描效率最高。CREATE INDEX idx_quotes_daily_symbol_ts ON quotes_daily (symbol, ts DESC);第三个是数据保留策略。不是所有数据都需要永久保存tick 级数据保留三个月就够了分钟线保留两年日线永久保留。TimescaleDB 自带保留策略可以自动删除过期分区。SELECT add_retention_policy(quotes_tick, INTERVAL 3 months);提示保留策略一定要在数据量还小的时候就配好。等库里有几亿行数据再想清理删除操作本身就会锁表很久影响线上查询。普通的关系型数据比如标的元信息、财报数据就用普通表存。财报数据有个特点是一行记录字段特别多而且不同行业的财报科目差异很大。我的做法是核心字段营收、净利润、总资产等用固定列存扩展字段用一个 JSONB 列存这样既保证了常用查询的性能又保留了灵活性。2.3 服务层API 接口设计与缓存策略服务层用 FastAPI接口设计遵循 RESTful 风格但针对金融数据的特点做了优化。核心接口就四个查行情、查财报、查标的列表、查数据更新时间。每个接口都支持批量查询因为前端画图或者策略回测往往一次要拿多个标的的数据一个一个请求太慢。from fastapi import FastAPI, Query from typing import List app FastAPI() app.get(/quotes) async def get_quotes( symbols: List[str] Query(...), start: int Query(...), end: int Query(...), freq: str Query(1d) ): # 从数据库批量查询并返回 ...缓存策略是服务层性能的关键。金融数据有个特点历史数据几乎不变最新数据变化频繁。所以我的缓存分两级。历史数据比如一个月前的缓存时间设得很长一天都行最新数据缓存时间设得很短几秒到几十秒。判断标准是数据的时间戳如果查询的时间范围完全在过去这个区间内就走长缓存。这里有个细节缓存 key 的构造要把所有查询参数都包含进去包括 symbols 列表的顺序。我一开始没注意这点导致[A, B]和[B, A]命中了不同的缓存浪费了不少内存。后来统一在构造 key 之前对 symbols 排序问题就解决了。3. 完整实操流程从零到跑通第一个接口3.1 环境准备与依赖安装我假设你用的是一台 Ubuntu 22.04 的云主机2 核 4G 起步数据量大了再升配。整个环境准备分四步。第一步装 PostgreSQL 和 TimescaleDB。TimescaleDB 有官方的 apt 源按官方文档加源安装就行。装完之后在 PostgreSQL 里执行CREATE EXTENSION timescaledb;启用扩展。第二步建数据库和用户。不要用默认的 postgres 超级用户跑应用单独建一个应用用户只给它必要的权限。CREATE USER finsvc WITH PASSWORD your_strong_password; CREATE DATABASE financial_services OWNER finsvc;第三步装 Python 依赖。我用的是 Python 3.11依赖管理用 pip 加 requirements.txt简单直接。pip install fastapi uvicorn psycopg2-binary requests tenacity pandas第四步建表。把前面说的行情表、财报表、标的元信息表都建好索引和保留策略一起配上。这一步建议写成一个 SQL 脚本方便以后重建环境。3.2 采集任务的调度与运行采集任务的调度我用的是最朴素的方案crontab 加一个 Python 入口脚本。为什么不用 Celery 或者 Airflow因为我的任务量不大日线数据一天跑一次分钟线数据五分钟跑一次crontab 完全够用而且没有额外的中间件要维护。入口脚本的逻辑是读取配置遍历所有需要采集的标的逐个调用采集函数把结果批量写入数据库。写入用INSERT ... ON CONFLICT DO UPDATE这样重复采集不会产生重复数据也能自动更新最新值。def upsert_quotes(conn, rows): with conn.cursor() as cur: cur.executemany( INSERT INTO quotes_daily (symbol, ts, open, high, low, close, volume) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (symbol, ts) DO UPDATE SET open EXCLUDED.open, high EXCLUDED.high, low EXCLUDED.low, close EXCLUDED.close, volume EXCLUDED.volume , rows) conn.commit()注意ON CONFLICT依赖唯一约束建表的时候一定要在(symbol, ts)上建唯一索引否则冲突检测不生效会插入重复数据。调度频率的设置有个经验日线数据在收盘后一小时跑避开数据源更新最频繁的时段分钟线数据在整点后两分钟跑给数据源留出更新时间。跑得太早数据还没更新跑得太晚又影响使用这个时间窗口需要根据实际数据源的更新规律来调。3.3 接口联调与性能验证服务跑起来之后用 curl 或者浏览器直接测接口。我习惯先测单个标的的小时间范围确认数据正确再测批量和大范围看性能。curl http://localhost:8000/quotes?symbolsAAPLstart1700000000000end1700100000000freq1d性能验证主要看两个指标响应时间和数据库查询时间。FastAPI 自带日志数据库查询时间可以在 SQL 里用EXPLAIN ANALYZE看。如果发现某个查询慢先看有没有走索引再看是不是扫描了太多分区。我实测下来单标的查一年的日线数据走索引的情况下响应时间在 50 毫秒以内批量查 100 个标的同一时间段响应时间在 300 毫秒左右。这个性能对于个人使用完全够用。如果要做更复杂的聚合查询比如计算移动平均建议在数据库里预计算好存成物化视图查询的时候直接读比实时算快得多。4. 常见问题与排查技巧实录4.1 数据采集失败的排查路径采集失败是最常见的问题排查要按顺序来。先看日志里的错误类型是网络超时、HTTP 错误码还是解析失败。网络超时通常是目标站点响应慢或者本地网络问题重试一般能解决HTTP 错误码要看具体是什么401/403 是权限问题404 是接口地址变了429 是限流解析失败说明目标站点的数据结构变了需要更新解析逻辑。我整理了一个速查表遇到问题按这个顺序排查基本能覆盖九成以上的情况。错误类型可能原因排查方法解决方式连接超时目标站点慢或本地网络问题ping 目标域名测其他站点重试检查本地网络401/403认证信息失效或权限不足检查 API key 和请求头更新认证信息404接口地址变更对比文档和实际请求 URL更新接口地址429请求频率超限检查请求间隔和并发数降低频率加退避解析失败数据结构变更打印原始响应对比更新解析逻辑数据重复唯一约束缺失检查表结构和索引补建唯一索引提示日志里一定要记录原始响应内容哪怕只记录前 500 个字符。解析失败的时候原始响应是排查问题的唯一线索。我一开始只记录错误信息不记录原始数据结果每次解析失败都要重新手动请求一遍效率极低。4.2 查询变慢的优化思路查询变慢通常有三个原因索引没走对、分区太多、数据量太大。排查的时候先用EXPLAIN ANALYZE看执行计划确认是否走了索引。如果没走索引检查查询条件是否和索引列匹配特别是时间范围查询条件要写成ts X AND ts Y的形式不要用函数包裹ts否则索引失效。分区太多的问题在数据积累到一定量之后才会显现。判断标准是查询计划里显示扫描了大量分区。解决办法是调整分区间隔把多个小分区合并成大分区。TimescaleDB 支持在线调整分区间隔但已经生成的分区不会自动合并需要手动处理。数据量太大的问题除了用保留策略清理过期数据还可以考虑冷热分离。把最近三个月的数据放在高性能存储上更早的数据归档到便宜的存储查询的时候根据时间范围路由到不同的存储。这个方案实现起来复杂一些数据量没到千万级之前不用考虑。4.3 数据质量校验的实用技巧数据质量是金融数据的生命线错误的数据比没有数据更可怕。我在采集之后加了一道校验检查几个关键点时间戳是否连续日线数据不应该有跳空、价格是否在合理范围内比如涨跌幅超过 50% 要标记、成交量是否非负。校验不通过的数据不直接丢弃而是标记出来存到一张异常表里人工确认后再决定是修正还是丢弃。这样做的好处是不会因为校验规则太严而误杀正常数据也不会让异常数据污染主表。def validate_quote(row, prev_row): if row[high] row[low]: return False, high low if row[volume] 0: return False, negative volume if prev_row and abs(row[close] / prev_row[close] - 1) 0.5: return False, price jump too large return True, None这套校验规则是我根据实际数据踩坑总结出来的。比如high low这种明显错误通常是数据源本身的问题遇到过几次之后我就加上了这条校验。价格跳空超过 50% 的规则帮我抓到过好几次数据源返回错误数据的情况特别是除权除息日附近数据源如果没有正确处理复权价格会出现巨大跳空。5. 服务扩展与长期维护的个人体会5.1 从单机到多实例的平滑演进服务跑了一段时间之后如果查询压力变大可以考虑加实例。因为服务层是无状态的加实例很简单在前面挂个 Nginx 做负载均衡就行。但要注意缓存的问题如果用的是进程内缓存多实例之间缓存不共享会导致缓存命中率下降。解决办法是把缓存挪到 Redis 里所有实例共享一份缓存。数据库这块读压力大的话可以加只读副本把查询请求路由到副本上。TimescaleDB 支持流复制配置起来和普通 PostgreSQL 一样。写入还是走主库读取走副本这样读写分离之后主库的压力会小很多。不过我要提醒一句不要过早优化。我见过太多项目在只有几百个请求每天的时候就搞了一套复杂的分布式架构结果维护成本高得吓人实际收益却微乎其微。单机 PostgreSQL 在数据量千万级、日请求十万级以下的时候完全扛得住先把单机跑稳遇到瓶颈再扩展。5.2 数据源变更的应对策略数据源变更是这个项目长期维护中最大的不确定性。目标站点改版、接口下线、字段调整这些都会发生而且往往没有提前通知。我的应对策略是抽象采集接口隔离变更影响。每个数据源的采集逻辑封装在一个独立的模块里对外暴露统一的函数签名。数据源变更的时候只需要改对应的模块其他部分不受影响。同时每个模块都有独立的测试用例用固定的样本数据验证解析逻辑改完之后跑一遍测试确认没有引入回归问题。另外我建议至少维护两个数据源。主源出问题的时候可以切到备源虽然数据可能有细微差异但总比服务完全不可用好。切换逻辑可以做成配置项改个配置就能切不用改代码。5.3 我踩过的几个印象深刻的坑第一个坑是时区。早期我没注意时间戳的时区问题采集的时候用的是本地时间存储的时候也没转换结果查询的时候发现数据对不上。后来统一规定所有时间戳在进入系统的那一刻就转成 UTC存储和传输全程用 UTC只在展示的时候转成本地时间。这条规则看起来简单但能避免大量因为时区导致的诡异问题。第二个坑是浮点数精度。金融数据对精度要求高用 float 存储价格会出现精度丢失比如 0.1 0.2 不等于 0.3 这种经典问题。解决办法是用NUMERIC类型存储价格Python 侧用Decimal处理。虽然性能比 float 稍差但精度有保障对于金融数据来说这个取舍是值得的。第三个坑是数据库连接管理。早期我每个请求都新建一个数据库连接请求量一上来连接数就爆了。后来改成连接池用psycopg2的ThreadedConnectionPool连接复用之后性能和稳定性都好了很多。连接池的大小要根据实际并发量调太小了请求排队太大了数据库扛不住一般设成最大并发数的 1.5 倍左右比较合适。这套financial-services我从最初的一个脚本慢慢迭代到现在中间重构过两次踩的坑基本都记录在上面了。它不是什么高大上的架构但胜在每一处设计都有明确的理由每一个参数都有实际的依据。如果你也在做类似的事情希望这些经验能帮你少走点弯路。数据服务这东西稳定比先进重要可控比功能多重要先把核心链路跑通跑稳再考虑锦上添花的事情。