1. 金融数据服务从零搭建的完整思路1.1 为什么我要自己动手做一套金融数据服务先说清楚这套东西是干什么的。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自建金融数据服务层——把行情数据、基本面数据、汇率、利率这些散落在各处的金融信息通过统一的接口聚合起来对外提供稳定的查询能力。它解决的核心问题是当你需要在自己的应用里接入金融数据时要么用第三方付费API贵、有调用限制、数据格式不统一要么自己爬不稳定、维护成本高、数据质量参差。这套服务就是在两者之间找一个平衡点。适合谁来参考有一定后端基础、懂基本的数据库操作、想在自己的项目里接入金融数据但不想被第三方API绑死的开发者。如果你只是想做个简单的股票查询小工具这套方案可能有点重但如果你要做的是量化回测系统、个人记账工具、投资组合追踪器这类需要频繁读取金融数据的应用那这套东西就非常合适。我踩过的第一个坑就是一开始想得太简单觉得不就是调几个接口存数据库再查出来吗结果真做起来才发现金融数据的坑远比普通业务数据深——数据源不稳定、字段含义模糊、时区处理混乱、复权计算复杂、交易日历不统一每一个都能让你debug到凌晨。1.2 整体架构怎么设计才合理我的设计思路是分层解耦从上到下分成四层接入层对外暴露RESTful API和WebSocket两种接口RESTful用于查询历史数据和基本面数据WebSocket用于推送实时行情。服务层核心业务逻辑包括数据聚合、缓存管理、限流控制、数据校验。采集层负责从各个数据源拉取原始数据做初步清洗和格式统一。存储层根据数据特性选择不同的存储方案——时序数据用PostgreSQLTimescaleDB基本面数据用普通关系表热点数据用Redis缓存。为什么这么分因为金融数据有个特点读多写少但对实时性和准确性要求极高。如果把采集和服务混在一起一旦某个数据源挂了整个服务都会受影响。分层之后采集层可以独立重试、独立扩容服务层只管从存储层读数据稳定性大幅提升。另一个关键决策是不用微服务。很多人一上来就搞微服务觉得架构越复杂越牛逼。但这套系统的数据量级其实不大——个人使用的话每天增量数据也就几十MB用单体应用模块化代码完全够用。微服务带来的网络开销、部署复杂度、调试难度在这个场景下纯属自找麻烦。我用的是模块化的单体架构代码层面通过清晰的模块边界来隔离部署层面就是一个进程简单可靠。1.3 技术选型的取舍逻辑技术栈的选择我纠结了很久最终定下来的是组件选型理由语言Python 3.11金融数据处理生态最成熟pandas、numpy随手可用Web框架FastAPI异步性能好自动生成文档类型提示友好主数据库PostgreSQL 15 TimescaleDB时序数据压缩率高SQL兼容性好缓存Redis 7热点数据缓存限流计数器任务队列Celery Redis定时采集任务异步处理部署Docker Compose一键启动环境隔离为什么不用Go或Rust性能确实更好但金融数据处理阶段大量依赖pandas做计算用Go的话要么调Python脚本增加复杂度要么自己实现一套计算逻辑开发成本太高。Python在这个场景下的性能瓶颈主要在IO等待用异步就能解决大部分问题真正CPU密集的计算可以丢给numpy底层C实现。为什么不用MongoDB金融数据虽然看起来像文档但本质上结构化程度很高——股票代码、日期、开盘价、收盘价这些都是强类型字段。用关系型数据库能获得更好的查询优化、事务保证和数据一致性。而且TimescaleDB对时序数据的支持比MongoDB好太多压缩率能到90%以上。2. 数据采集与清洗的核心细节2.1 数据源的选择与对接策略金融数据源大致分三类官方交易所接口、第三方数据提供商、公开数据接口。我的策略是主备结合多源校验。主数据源我选的是几个公开的金融数据接口优点是免费、数据质量尚可、覆盖A股和港股。备用数据源用另一个公开接口做交叉验证。为什么不只用一家因为实测下来任何单一数据源都有出问题的时候——有时候是接口挂了有时候是数据延迟有时候是字段突然变了。多源校验能在数据入库前发现异常。对接的时候有几个关键点请求频率控制大部分公开接口都有频率限制我设的是每秒最多3次请求用Redis做令牌桶限流。超过限制就排队等待不要硬刚否则IP被封了更麻烦。重试机制网络请求失败是常态我设的是指数退避重试最多3次。第一次等1秒第二次等2秒第三次等4秒。超过3次就标记为失败等下一轮采集周期再试。数据格式适配不同数据源返回的字段名、日期格式、数值单位都不一样。我在采集层做了一个适配器模式每个数据源对应一个适配器类负责把原始数据转换成统一的内部格式。# 适配器基类示例 class BaseAdapter: def fetch(self, symbol: str, start: str, end: str) - list[dict]: raise NotImplementedError def normalize(self, raw_data: list[dict]) - list[dict]: 将原始数据转换为统一格式 raise NotImplementedError class SourceAAdapter(BaseAdapter): def fetch(self, symbol, start, end): # 具体请求逻辑 pass def normalize(self, raw_data): result [] for item in raw_data: result.append({ symbol: item[code], trade_date: parse_date(item[date]), open: float(item[open]), high: float(item[high]), low: float(item[low]), close: float(item[close]), volume: int(item[vol]), amount: float(item[amount]), }) return result2.2 数据清洗的五个关键步骤原始数据拿到手只是开始清洗才是重头戏。我总结了五个必须做的步骤第一步去重。同一交易日同一股票可能有多条记录原因可能是采集重复或数据源本身有问题。去重规则是(symbol, trade_date)唯一保留最新采集的那条。第二步缺失值处理。金融数据缺失很常见尤其是停牌、节假日。处理策略要分情况如果是停牌导致的缺失标记为停牌状态不要用前值填充如果是数据源漏采尝试从备用源补如果补不到标记为缺失查询时返回null而不是0。第三步异常值检测。价格突然变成0或者负数成交量突然放大100倍这些都要标记出来。我用的是简单的3-sigma原则如果某个值偏离过去20个交易日均值超过3倍标准差就标记为可疑人工复核后再决定是否入库。第四步复权处理。这是金融数据特有的坑。股票除权除息后价格会出现跳空如果不做复权处理计算收益率时会得到错误结果。前复权适合看盘后复权适合回测。我的做法是同时存储原始价格和前复权价格查询时通过参数指定用哪种。第五步时区统一。所有时间戳统一存UTC查询时再转成用户指定的时区。A股是UTC8港股是UTC8美股是UTC-5夏令时UTC-4如果不统一跨市场查询时会乱套。注意复权计算需要用到除权除息数据这个数据比行情数据更难获取。我的做法是先用简单的比例复权根据除权日的价格比例调整等积累足够数据后再切换到精确复权。2.3 交易日历的维护交易日历看起来简单实际上是个大坑。不同市场的交易日完全不同——A股春节休市港股春节休市但除夕可能开半天美股感恩节休市。而且还有临时休市比如台风、极端天气。我的做法是维护一张trading_calendar表字段包括market、date、is_trading_day、note。每年初从各交易所官网获取下一年度的交易日历手动录入。临时休市通过公告获取后更新。为什么不用第三方库试过几个要么更新不及时要么覆盖市场不全。自己维护虽然麻烦但可控性最强。而且这张表一旦建好后续维护成本很低每年更新一次就行。查询逻辑上所有涉及日期的操作都要先过交易日历。比如计算过去5个交易日的收益率不能简单用date - 5而是要从交易日历里取最近5个交易日。3. 存储方案与查询优化实操3.1 时序数据表结构设计行情数据是典型的时序数据表结构设计直接影响查询性能。我的表结构是这样的CREATE TABLE daily_quotes ( symbol VARCHAR(20) NOT NULL, trade_date DATE NOT NULL, open NUMERIC(12, 4), high NUMERIC(12, 4), low NUMERIC(12, 4), close NUMERIC(12, 4), volume BIGINT, amount NUMERIC(20, 4), adj_factor NUMERIC(12, 6) DEFAULT 1.0, created_at TIMESTAMPTZ DEFAULT NOW(), PRIMARY KEY (symbol, trade_date) ); SELECT create_hypertable(daily_quotes, trade_date);几个关键决策主键用(symbol, trade_date)复合主键而不是自增ID。因为查询几乎总是按股票代码日期范围来查复合主键能直接命中索引。价格字段用NUMERIC而不是FLOAT。浮点数在金融计算中会有精度问题比如0.10.2不等于0.3。NUMERIC虽然性能稍差但精度有保证。用TimescaleDB的hypertable。它会自动按时间分区查询时只扫描相关分区性能提升明显。实测下来1000万行数据的范围查询从原来的2秒降到200毫秒。adj_factor字段存复权因子查询时用close * adj_factor得到复权价。这样原始价格和复权价格都能算出来不用存两份。3.2 缓存策略与热点数据识别金融数据的查询有明显的二八定律——80%的查询集中在20%的股票上。我的缓存策略是L1缓存Redis。缓存最近30天的日线数据key格式是quote:{symbol}:{date}TTL设为1小时。为什么是1小时因为盘中数据会变但盘后数据不变1小时能覆盖大部分场景。L2缓存应用内存。用LRU Cache缓存最近查询的1000条记录TTL 5分钟。这层缓存主要是为了应对高频重复查询比如用户反复刷新同一个页面。缓存预热每天开盘前把关注列表里的股票数据预加载到Redis。关注列表从用户配置里读一般不超过100只股票。缓存更新策略用的是Cache-Aside查询时先查缓存命中就返回未命中就查数据库然后写回缓存。数据更新时先更新数据库再删除缓存不是更新缓存删除更安全避免并发写导致脏数据。实操心得缓存key一定要加版本号或日期后缀否则数据结构变更时会读到旧格式的数据排查起来很痛苦。我吃过这个亏后来所有key都加了v2前缀。3.3 查询性能优化的实战技巧查询优化我做了几件事第一索引优化。除了主键索引还建了(trade_date)的单列索引用于按日期范围扫描全市场数据。另外建了(symbol, trade_date DESC)的索引用于查单只股票的最新N条记录。第二查询重写。很多查询可以改写成更高效的形式。比如查最新价格不要用ORDER BY trade_date DESC LIMIT 1而是用WHERE trade_date (SELECT MAX(trade_date) FROM ...)后者能直接命中索引。第三分区裁剪。TimescaleDB的hypertable会自动做分区裁剪但前提是查询条件里要有时间范围。如果查询不带时间条件会扫描所有分区。所以我在API层强制要求时间范围参数不传就默认最近30天。第四物化视图。对于一些复杂的聚合查询比如计算某只股票的20日均线我用物化视图预计算。每天收盘后刷新一次查询时直接读视图速度提升10倍以上。CREATE MATERIALIZED VIEW ma_20 AS SELECT symbol, trade_date, AVG(close) OVER ( PARTITION BY symbol ORDER BY trade_date ROWS BETWEEN 19 PRECEDING AND CURRENT ROW ) AS ma20 FROM daily_quotes; CREATE INDEX ON ma_20 (symbol, trade_date);4. API设计与接口实现4.1 RESTful接口的规范设计API设计我遵循几个原则资源导向、版本控制、统一响应格式。资源导向的意思是URL里用名词不用动词。比如查行情是GET /api/v1/quotes/{symbol}不是GET /api/v1/getQuote?symbolxxx。查基本面是GET /api/v1/fundamentals/{symbol}。版本控制用URL前缀/api/v1/不用Header。为什么因为URL版本更直观调试时一眼就能看出用的是哪个版本。Header版本虽然更优雅但实际用起来容易忘。统一响应格式{ code: 0, message: success, data: { ... }, request_id: abc-123 }code为0表示成功非0表示各种错误。request_id用于追踪请求排查问题时非常有用。核心接口列表接口方法说明/api/v1/quotes/{symbol}GET查询历史行情/api/v1/quotes/{symbol}/latestGET查询最新行情/api/v1/fundamentals/{symbol}GET查询基本面数据/api/v1/calendar/{market}GET查询交易日历/api/v1/watchlistGET/POST/DELETE管理关注列表/ws/v1/quotesWebSocket实时行情推送4.2 实时行情推送的实现实时行情用WebSocket推送实现逻辑是客户端连接时带上要订阅的股票代码列表。服务端把连接信息注册到Redis的订阅表里。采集层拿到新数据后发布到Redis的Pub/Sub频道。服务端的推送模块订阅频道收到数据后根据订阅表推送给对应客户端。为什么用Redis Pub/Sub而不是直接推因为采集层和服务层是分离的可能部署在不同进程甚至不同机器上。Redis Pub/Sub提供了一个简单的消息总线解耦了生产者和消费者。推送频率控制A股行情是3秒一笔但没必要每笔都推。我的策略是批量推送——每3秒聚合一次把这段时间内所有变化的数据打包成一个消息推给客户端。这样既保证了实时性又减少了推送次数。# 推送模块核心逻辑 async def push_loop(): pubsub redis.pubsub() await pubsub.subscribe(quote_updates) buffer {} last_push time.time() async for message in pubsub.listen(): if message[type] ! message: continue data json.loads(message[data]) symbol data[symbol] buffer[symbol] data if time.time() - last_push 3.0: await broadcast(buffer) buffer.clear() last_push time.time()4.3 限流与鉴权机制限流用的是滑动窗口算法基于Redis实现。每个API Key每分钟最多60次请求超过就返回429。为什么用滑动窗口而不是固定窗口固定窗口在窗口切换时会有突发流量问题——比如限制每分钟60次用户在00:59发60次01:00又发60次实际2秒内发了120次。滑动窗口能平滑这个问题。鉴权用API Key放在Header的X-API-Key字段里。Key的生成规则是sk_ 32位随机字符串。存储时只存哈希值不存明文防止数据库泄露后Key被滥用。注意API Key一定要支持禁用和轮换。我遇到过Key泄露的情况幸好有禁用功能及时止损。轮换功能让用户可以定期更换Key降低泄露风险。5. 常见问题与排查技巧实录5.1 数据采集失败的排查思路采集失败是最常见的问题排查思路按以下顺序第一步确认网络连通性。先用curl手动请求一次数据源接口看是否能通。如果不通检查DNS、防火墙、代理设置。第二步检查请求参数。有时候是参数格式变了比如日期格式从2024-01-01变成了20240101。对比一下之前成功的请求和现在失败的请求看参数有什么差异。第三步查看返回内容。如果HTTP状态码是200但数据为空可能是数据源改了返回结构。打印原始返回内容对比文档。第四步检查频率限制。如果返回429或403说明请求太频繁被限流了。降低请求频率或者换备用数据源。第五步查看日志。我在采集层打了详细的日志包括请求URL、请求参数、返回状态码、返回内容摘要。排查时直接grep日志很快就能定位问题。常见问题速查表现象可能原因解决方法连接超时网络问题或数据源宕机切换备用源检查网络返回403IP被封或频率超限降低频率更换IP返回数据为空参数错误或数据源变更对比文档检查参数数据字段缺失数据源结构调整更新适配器代码数据值异常数据源质量问题标记异常人工复核5.2 查询性能突然下降的排查查询变慢通常有几个原因原因一数据量增长导致索引失效。随着数据积累原来的索引可能不再高效。解决方法是EXPLAIN ANALYZE分析查询计划看是否走了全表扫描。如果是考虑加索引或调整查询。原因二缓存穿透。大量请求查询不存在的数据缓存永远不命中全部打到数据库。解决方法是缓存空结果TTL设短一点比如1分钟。原因三数据库连接池耗尽。并发请求太多连接池不够用请求排队等待。解决方法是调大连接池或者优化查询减少连接占用时间。原因四TimescaleDB分区过多。如果按天分区几年下来会有上千个分区查询规划时间变长。解决方法是调整分区策略比如按月分区或者合并历史分区。我遇到过一次查询突然从200ms变成5秒的情况排查了半天发现是某个物化视图刷新失败导致查询走了原始表。后来加了物化视图刷新监控刷新失败就告警。5.3 数据一致性问题的处理金融数据对一致性要求极高我遇到过几次数据不一致的情况场景一同一股票同一日期有两条不同价格的记录。原因是两个数据源返回的数据有差异。解决方法是设定优先级主数据源优先备用源只做补充。场景二复权因子计算错误导致复权价不对。原因是除权除息数据漏采。解决方法是增加校验逻辑——复权后的价格不能出现超过涨跌停限制的跳空。场景三时区转换错误导致日期偏移。原因是有些数据源返回的是本地时间但没有时区标记。解决方法是在适配器层强制指定时区统一转UTC。实操心得数据一致性校验一定要自动化。我写了一个校验脚本每天收盘后跑一遍检查价格范围、成交量非负、日期连续性等。发现问题自动发邮件告警比人工检查靠谱得多。6. 部署与运维的实战经验6.1 Docker Compose一键部署方案部署我用的是Docker Compose配置文件大概长这样version: 3.8 services: db: image: timescale/timescaledb:latest-pg15 environment: POSTGRES_DB: financial POSTGRES_USER: app POSTGRES_PASSWORD: ${DB_PASSWORD} volumes: - pgdata:/var/lib/postgresql/data ports: - 5432:5432 redis: image: redis:7-alpine volumes: - redisdata:/data ports: - 6379:6379 app: build: . depends_on: - db - redis environment: DATABASE_URL: postgresql://app:${DB_PASSWORD}db:5432/financial REDIS_URL: redis://redis:6379/0 ports: - 8000:8000 worker: build: . command: celery -A tasks worker --loglevelinfo depends_on: - db - redis environment: DATABASE_URL: postgresql://app:${DB_PASSWORD}db:5432/financial REDIS_URL: redis://redis:6379/0 volumes: pgdata: redisdata:几个关键点密码用环境变量不写在配置文件里。用.env文件管理.env加入.gitignore。数据卷持久化容器删了数据还在。app和worker分开app处理API请求worker处理采集任务互不影响。健康检查给db和redis加healthcheckapp等它们健康后再启动。6.2 监控与告警配置监控我用了三个层面层面一系统监控。用docker stats看CPU、内存、网络。如果CPU持续超过80%说明需要优化或扩容。层面二应用监控。在FastAPI里加了中间件记录每个请求的耗时、状态码。用Prometheus采集指标Grafana展示。层面三业务监控。监控采集任务的成功率、数据延迟、数据量。如果采集成功率低于95%或者数据延迟超过1小时就告警。告警渠道用的是邮件Webhook。邮件用于非紧急告警Webhook推送到即时通讯工具用于紧急告警。关键监控指标指标阈值告警级别采集成功率 95%警告数据延迟 1小时警告API P99延迟 1秒警告数据库连接数 80%严重磁盘使用率 85%严重6.3 备份与恢复策略金融数据丢了是灾难性的备份策略必须可靠数据库备份每天凌晨2点全量备份用pg_dump导出压缩后上传到对象存储。保留最近30天的备份。Redis备份Redis主要做缓存数据丢了可以从数据库重建所以不做定期备份。但开启了AOF持久化防止意外重启丢数据。配置文件备份所有配置文件用Git管理每次变更都有记录。恢复流程我也演练过几次从对象存储下载备份文件解压用pg_restore恢复然后重启服务。整个过程大概15分钟可以接受。注意备份文件一定要定期验证可恢复性。我遇到过备份文件损坏的情况幸好发现得早。现在每月做一次恢复演练确保备份真的能用。7. 后续扩展方向与个人体会这套系统目前跑了大半年整体稳定。后续我打算做几个扩展扩展一增加技术指标计算。目前只存了原始行情数据均线、MACD、RSI这些指标都是查询时实时算的。数据量大了之后实时算会慢打算改成预计算每天收盘后批量算好存起来。扩展二支持更多市场。目前只覆盖A股和港股后续想加美股和期货。不同市场的交易规则、数据格式差异很大需要写新的适配器。扩展三增加数据质量评分。给每条数据打一个质量分综合考虑数据源可靠性、采集时间、校验结果。查询时可以按质量分过滤优先返回高质量数据。扩展四开放API给更多用户。目前是自己用后续想开放给朋友用。需要增加用户管理、配额管理、更细粒度的权限控制。我个人在实际操作中的体会是金融数据服务的核心难点不在技术而在数据质量。技术方案再优雅数据不准就是白搭。所以我在数据校验上花的精力比写代码还多。另外不要追求大而全先把一个市场、一种数据类型做扎实再逐步扩展。一开始就想覆盖所有市场所有数据类型最后往往什么都做不好。最后分享一个小技巧采集任务的时间安排要避开数据源的高峰期。我一开始设在整点采集结果经常超时。后来改到每小时的15分和45分成功率明显提升。数据源也是人维护的也有高峰期错峰采集能省很多事。