1. 金融数据服务从零搭建的完整思路拆解1.1 为什么我要自己动手做一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合与分发服务。它的核心职责就三件事从多个公开数据源拉取行情、财报、宏观经济指标等金融数据做清洗和标准化然后通过统一的 API 接口对外提供服务。说白了就是把散落在各处的金融数据整合成一个自己能掌控的“数据中台”。为什么不用现成的商业数据接口我踩过的坑是这样的免费接口有调用频率限制付费接口按调用量计费一旦你的应用稍微有点规模成本就直线上升。更关键的是数据格式各家不同今天用 A 家的行情明天想换成 B 家的代码得大改。自建服务最大的好处是数据主权在自己手里上游换源对下游透明接口格式自己定想怎么扩展就怎么扩展。这套东西适合谁我认为有三类人值得参考一是做量化回测的个人开发者需要稳定的历史数据和实时行情二是做金融类小工具、小程序的独立开发者不想被第三方 API 的配额卡脖子三是对数据管道感兴趣、想拿一个真实项目练手数据工程的后端工程师。前提是你得会一点 Python 或者 Node.js懂基本的数据库操作剩下的跟着做就行。1.2 整体架构选型为什么是“采集-清洗-存储-服务”四层我把整个系统拆成四层这个分层不是拍脑袋定的而是根据数据流动的自然顺序来的。采集层负责跟外部数据源打交道清洗层做格式统一和异常处理存储层解决数据怎么存、存多久的问题服务层对外暴露 API。每一层之间用消息队列或者数据库做解耦这样任何一层出问题都不会把整条链路拖死。为什么不用一个大脚本从头跑到尾我试过刚开始确实快一个main.py里写几百行跑起来也能用。但问题是数据源一多某个源挂了整个脚本就卡住想加一个新数据源得在几百行里找地方插代码想换存储方案牵一发动全身。分层之后每个数据源是一个独立的采集器互不干扰加新源就是加一个新文件的事。技术栈方面我选的是Python FastAPI PostgreSQL Redis APScheduler。Python 的生态在金融数据处理上最成熟pandas、numpy 这些库处理时间序列数据非常顺手。FastAPI 自带异步支持和自动文档写 API 效率极高。PostgreSQL 存结构化的历史数据Redis 做实时行情的缓存和去重。APScheduler 负责定时任务调度。这套组合的优点是全部可以本地跑不依赖任何云服务一台 4 核 8G 的机器就能撑起一个中等规模的服务。注意选型时不要盲目追求“最新最热”。我见过有人用 Kafka Flink 做个人项目结果光维护这套基础设施就耗掉大半精力。个人项目的第一原则是能跑起来、好维护不是技术炫技。1.3 数据源的选择逻辑与风险评估金融数据源大致分三类交易所官方接口、第三方聚合接口、网页抓取。我的策略是优先用官方接口其次用有免费额度的第三方最后才考虑抓取。官方接口最稳定但通常只覆盖自家市场第三方聚合接口方便但有配额和停服风险网页抓取最灵活但最脆弱页面一改就失效。具体到数据源我一般会同时接两到三个来源做冗余。比如行情数据主源用一家备源用另一家主源连续失败三次就自动切到备源。这个切换逻辑写在采集层的基类里所有采集器继承同一个基类自动获得重试和切换能力。这里有个经验不要等主源彻底挂了才切要设置一个失败阈值比如连续 3 次请求超时或返回异常就触发切换同时发告警通知。风险评估方面最大的坑是数据源悄悄改了字段格式。我遇到过某接口把volume字段从整数改成字符串结果入库时报错整个采集任务中断。后来我加了一层 schema 校验每个数据源定义一个期望的字段类型采集后先校验再入库不符合就丢弃并记录日志。这个校验层救了我很多次。2. 核心模块的细节解析与实操要点2.1 采集层如何写出“打不死”的数据采集器采集器的核心要求是健壮不是快。我见过太多采集脚本网络一抖动就崩或者被限流了还在死命请求。我的做法是给每个采集器套三层保护超时控制、重试退避、限流遵守。超时控制很简单所有 HTTP 请求必须设timeout我一般设 10 秒。别小看这一条很多脚本卡死就是因为没设超时一个请求挂住整个线程就废了。重试退避是指失败后不要立刻重试而是等一段时间而且等待时间要递增。我用的是指数退避第一次等 1 秒第二次 2 秒第三次 4 秒最多重试 3 次。这样既能应对临时故障又不会给上游造成压力。限流遵守是很多人忽略的。每个数据源都有隐含的频率限制你不遵守轻则被封 IP重则账号被禁。我的做法是在采集器里维护一个令牌桶根据数据源文档里写的限制来配置。比如某接口限制每分钟 60 次我就把令牌桶速率设为每秒 1 个采集任务从桶里拿令牌拿不到就等。这样即使我开了多个采集任务也不会超限。import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(retries3, backoff1.0): session requests.Session() retry Retry( totalretries, backoff_factorbackoff, status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET] ) adapter HTTPAdapter(max_retriesretry) session.mount(https://, adapter) session.mount(http://, adapter) return session session build_session() resp session.get(https://api.example.com/quote, timeout10)这段代码是我每个采集器的标配。Retry对象自动处理重试和退避status_forcelist里列出需要重试的状态码429 是限流5xx 是服务端错误。这样写出来的采集器面对临时故障基本能自愈。实操心得重试不是万能的。如果某个接口连续返回 401 或 403说明是认证问题重试没用应该直接告警。我在重试逻辑里加了一个判断4xx 里除了 429 之外都不重试直接抛异常。2.2 清洗层把“脏数据”变成“干净数据”的关键步骤金融数据的脏超出很多人的想象。同一个指标不同来源的单位可能不一样有的用“元”有的用“万元”时间戳有的用秒有的用毫秒有的干脆是字符串缺失值有的用null有的用-有的用0。清洗层的任务就是把这些统一成一套标准格式。我的清洗流程分四步字段映射、类型转换、缺失值处理、异常值检测。字段映射是第一步每个数据源定义一个映射表把源字段名映射到标准字段名。比如 A 源的last_price和 B 源的close都映射到标准字段close_price。类型转换是第二步所有数值字段统一转成float时间字段统一转成 UTC 时间戳。缺失值处理是第三步我一般用前值填充因为金融时间序列里缺失通常意味着“没有交易”用前一个有效值填充最合理。异常值检测是第四步用简单的 3-sigma 规则超过均值 3 倍标准差的标记为异常记录但不入库。import pandas as pd import numpy as np def clean_quote(df: pd.DataFrame) - pd.DataFrame: df df.rename(columns{last_price: close_price, vol: volume}) df[close_price] pd.to_numeric(df[close_price], errorscoerce) df[volume] pd.to_numeric(df[volume], errorscoerce) df[timestamp] pd.to_datetime(df[timestamp], units, utcTrue) df df.sort_values(timestamp) df[close_price] df[close_price].ffill() df[volume] df[volume].fillna(0) mean, std df[close_price].mean(), df[close_price].std() df[is_outlier] np.abs(df[close_price] - mean) 3 * std return df这段清洗逻辑我用了很久基本能覆盖 90% 的场景。errorscoerce是关键遇到无法转换的值会变成NaN而不是报错后面统一处理。ffill是前向填充fillna(0)用于成交量因为成交量缺失通常意味着零成交。注意异常值不要直接删掉要标记出来。金融数据里暴涨暴跌可能是真实行情直接删掉会丢失重要信息。我的做法是标记is_outlier字段入库时保留但 API 返回时可以过滤。2.3 存储层历史数据与实时数据的分治策略存储层的核心矛盾是历史数据量大但查询频率低实时数据量小但查询频率高。用同一套存储方案会两头不讨好。我的策略是分治历史数据存 PostgreSQL实时数据存 Redis。PostgreSQL 存历史数据表结构按“品种 时间”做分区。比如行情表按月份分区每个月一个分区表。这样查询某个月的数据时PostgreSQL 只扫描对应分区速度很快。索引建在(symbol, timestamp)上覆盖最常见的查询模式。数据保留策略是日线数据永久保留分钟线保留 2 年tick 数据保留 3 个月。过期数据用定时任务归档到冷存储或者直接删除。Redis 存实时数据用 Hash 结构key 是quote:{symbol}field 是各个字段。设置 TTL 为 60 秒超过 60 秒没更新的数据自动过期。这样 API 查询实时行情时直接从 Redis 读响应时间在毫秒级。Redis 还用来做去重采集器每次拿到新数据先检查 Redis 里是否已存在相同时间戳的记录存在就跳过避免重复入库。CREATE TABLE quote_history ( symbol VARCHAR(20) NOT NULL, timestamp TIMESTAMPTZ NOT NULL, close_price NUMERIC(18, 6), volume BIGINT, is_outlier BOOLEAN DEFAULT FALSE, PRIMARY KEY (symbol, timestamp) ) PARTITION BY RANGE (timestamp); CREATE TABLE quote_history_2024_01 PARTITION OF quote_history FOR VALUES FROM (2024-01-01) TO (2024-02-01);分区表的写法如上每个月手动或自动建一个分区。我写了一个定时任务每月 25 号自动创建下个月的分区避免月底忘记。实操心得PostgreSQL 的NUMERIC类型比FLOAT更适合存价格因为浮点数有精度问题。我一开始用FLOAT结果发现 0.1 0.2 不等于 0.3回测时出现莫名其妙的偏差。换成NUMERIC(18, 6)后问题消失。2.4 服务层API 设计的三个核心原则服务层的 API 设计我遵循三个原则统一入口、版本控制、限流保护。统一入口是指所有数据通过一个 base URL 访问比如/api/v1/quote/{symbol}、/api/v1/fundamental/{symbol}。版本控制是指 URL 里带版本号将来接口不兼容升级时老版本继续保留一段时间给调用方迁移的时间。限流保护是指每个 API key 有调用配额超过就返回 429。FastAPI 写这些非常顺手。我用依赖注入做认证和限流每个请求先过认证中间件检查 API key 是否有效然后过限流中间件检查配额是否用完。配额信息存在 Redis 里用滑动窗口算法精确到秒。from fastapi import FastAPI, Depends, HTTPException from redis import Redis app FastAPI() redis Redis() def rate_limit(api_key: str, limit: int 100): key frate:{api_key}:{int(time.time())} current redis.incr(key) if current 1: redis.expire(key, 1) if current limit: raise HTTPException(status_code429, detailRate limit exceeded) app.get(/api/v1/quote/{symbol}) def get_quote(symbol: str, api_key: str Depends(verify_key)): rate_limit(api_key) data redis.hgetall(fquote:{symbol}) if not data: raise HTTPException(status_code404, detailSymbol not found) return data这段代码展示了限流和查询的核心逻辑。incr是原子操作多个请求同时来也不会超限。expire设置 1 秒过期实现每秒重置的滑动窗口。注意API key 不要明文存数据库存哈希值。我用hashlib.sha256对 key 做哈希验证时比对哈希值。这样即使数据库泄露攻击者也无法还原出原始 key。3. 完整实操流程从零到跑通一条数据链路3.1 环境准备与依赖安装我假设你用的是 Ubuntu 22.04 或者 macOSPython 3.10 以上。第一步是建虚拟环境别嫌麻烦这是好习惯。python -m venv venv source venv/bin/activate pip install fastapi uvicorn pandas numpy requests redis psycopg2-binary apschedulerPostgreSQL 和 Redis 的安装我就不展开了网上教程很多。装完后建一个数据库financial建一个用户fin_user密码自己设。Redis 默认配置就行注意设置maxmemory和maxmemory-policy allkeys-lru避免内存打满。配置文件我用.env管理不硬编码在代码里。DATABASE_URLpostgresql://fin_user:passwordlocalhost:5432/financial REDIS_URLredis://localhost:6379/0 API_KEY_SALTyour_random_salt_here3.2 数据库初始化与分区表创建建表 SQL 我放在migrations/目录下用简单的脚本按顺序执行。核心表有三张quote_history存行情fundamental_history存财报macro_history存宏观指标。每张表都按时间分区。import psycopg2 from datetime import datetime, timedelta def create_monthly_partition(conn, table: str, year: int, month: int): start datetime(year, month, 1) if month 12: end datetime(year 1, 1, 1) else: end datetime(year, month 1, 1) partition_name f{table}_{year}_{month:02d} sql f CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF {table} FOR VALUES FROM ({start}) TO ({end}); with conn.cursor() as cur: cur.execute(sql) conn.commit()这个函数我封装成一个定时任务每月 25 号跑一次创建下个月的分区。提前创建是为了避免月初数据写入时找不到分区。3.3 采集任务的调度与执行调度用 APScheduler我配了三个任务行情采集每 5 秒一次财报采集每天一次宏观数据每周一次。每个任务独立运行互不阻塞。from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.scheduled_job(interval, seconds5, max_instances1) def collect_quotes(): for source in quote_sources: try: data source.fetch() cleaned clean_quote(data) save_to_db(cleaned) save_to_redis(cleaned) except Exception as e: logger.error(fSource {source.name} failed: {e}) scheduler.start()max_instances1很重要防止上一次任务还没跑完下一次又启动导致重复采集。异常捕获也是必须的一个源挂了不能影响其他源。3.4 API 服务的启动与验证启动 FastAPI 用 uvicorn生产环境建议用 gunicorn 加 uvicorn worker。uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4启动后访问http://localhost:8000/docs能看到自动生成的 API 文档。验证流程先调/api/v1/quote/AAPL看有没有数据再调/api/v1/history/AAPL?start2024-01-01end2024-01-31看历史数据。如果返回空检查采集任务是否在跑数据库里有没有数据。实操心得我习惯在服务启动时加一个健康检查接口/health返回数据库和 Redis 的连接状态。这样用监控工具一探就知道服务是否正常不用登录服务器看日志。4. 常见问题与排查技巧实录4.1 数据采集失败的五大原因与排查路径采集失败是最常见的问题我整理了一个排查表按出现频率排序。问题现象可能原因排查方法解决方案请求超时网络问题或上游限流用 curl 手动请求同一接口增加超时时间检查限流配置返回 401/403API key 失效或权限不足检查 key 是否过期更新 key检查权限范围返回 429请求频率超限查看采集频率是否过高降低频率增加令牌桶等待数据字段缺失上游改了返回格式对比文档和实际返回更新字段映射加 schema 校验入库报错数据类型不匹配查看数据库错误日志检查清洗逻辑加类型转换这个表我贴在显示器旁边出问题先对照排查能省很多时间。4.2 数据不一致的根源分析与解决数据不一致是指同一指标在不同地方的值不一样。我遇到过三种情况一是 Redis 和 PostgreSQL 里的值不同步原因是 Redis 写入成功但数据库写入失败二是不同数据源的值有差异比如 A 源和 B 源的收盘价差了几分钱三是时区问题有的源用本地时间有的用 UTC。第一种情况的解决方法是先写数据库再写 Redis。数据库是持久化的Redis 是缓存数据库写成功才算成功。如果 Redis 写失败下次采集会补上。第二种情况无解只能选一个主源其他源做参考。我在配置里明确指定主源API 返回主源的值。第三种情况统一用 UTC所有时间戳入库前都转成 UTCAPI 返回时再转成用户指定的时区。4.3 性能瓶颈的定位与优化服务跑久了会变慢我遇到过两个瓶颈数据库查询慢和 Redis 内存满。数据库查询慢通常是索引没建对用EXPLAIN ANALYZE看查询计划发现全表扫描就加索引。我有个查询是SELECT * FROM quote_history WHERE symbol AAPL AND timestamp 2024-01-01一开始只建了symbol索引后来改成(symbol, timestamp)复合索引查询时间从 2 秒降到 20 毫秒。Redis 内存满是因为实时数据没设 TTL越积越多。解决方法是给所有 key 设 TTL并且配置maxmemory-policy allkeys-lru内存满时自动淘汰最久未使用的 key。我还加了一个监控Redis 内存超过 80% 就告警。注意优化之前先测量别凭感觉。我用pg_stat_statements扩展找出最慢的查询用redis-cli --bigkeys找出最大的 key有针对性地优化比盲目调参有效得多。4.4 安全加固的四个必做项金融数据服务的安全不能马虎。我做了四件事API key 认证、HTTPS 加密、请求日志审计、数据库最小权限。API key 认证前面说了每个调用方一个 key可以单独吊销。HTTPS 用 Nginx 做反向代理Lets Encrypt 免费证书。请求日志记录每个请求的 key、IP、时间、路径存 30 天方便追溯。数据库用户只给必要的权限采集用户只能 INSERTAPI 用户只能 SELECT不给 DELETE 和 DROP。这四条做完基本的安全底线就有了。别觉得个人项目没人攻击我搭的服务上线第三天就有人扫端口第七天就有异常请求。安全措施不是防君子是防自动化扫描工具。5. 后续扩展方向与个人经验分享这套服务跑稳定之后我陆续加了几个扩展。一个是数据质量监控面板用 Grafana 展示每个数据源的采集成功率、延迟、数据量一眼就能看出哪个源有问题。另一个是回测数据导出提供一个接口输入品种和时间范围直接返回 CSV 文件方便导入到回测框架。还有一个是Webhook 推送当某个指标超过阈值时主动推送到指定的 URL不用轮询。我个人在实际操作中的体会是金融数据服务最难的不是技术是数据的稳定性和一致性。技术方案可以照抄但数据源的变化、格式的调整、异常的处理这些都需要长期观察和积累。我建议刚开始做的时候先跑通一条最简单的链路一个数据源、一张表、一个接口跑上一周观察数据质量再逐步扩展。别一上来就搞大而全那样很容易在细节里迷失。最后再分享一个小技巧给每个数据源建一个“健康档案”记录它的历史表现比如平均响应时间、失败率、最近一次格式变更时间。这个档案用简单的 JSON 文件存就行每次采集后更新。时间长了你就能预判哪个源可能要出问题提前做准备。这个习惯帮我避免了好几次半夜被报警叫醒的情况。
从零搭建自托管金融数据服务：采集、清洗、存储与API全链路实战
1. 金融数据服务从零搭建的完整思路拆解1.1 为什么我要自己动手做一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合与分发服务。它的核心职责就三件事从多个公开数据源拉取行情、财报、宏观经济指标等金融数据做清洗和标准化然后通过统一的 API 接口对外提供服务。说白了就是把散落在各处的金融数据整合成一个自己能掌控的“数据中台”。为什么不用现成的商业数据接口我踩过的坑是这样的免费接口有调用频率限制付费接口按调用量计费一旦你的应用稍微有点规模成本就直线上升。更关键的是数据格式各家不同今天用 A 家的行情明天想换成 B 家的代码得大改。自建服务最大的好处是数据主权在自己手里上游换源对下游透明接口格式自己定想怎么扩展就怎么扩展。这套东西适合谁我认为有三类人值得参考一是做量化回测的个人开发者需要稳定的历史数据和实时行情二是做金融类小工具、小程序的独立开发者不想被第三方 API 的配额卡脖子三是对数据管道感兴趣、想拿一个真实项目练手数据工程的后端工程师。前提是你得会一点 Python 或者 Node.js懂基本的数据库操作剩下的跟着做就行。1.2 整体架构选型为什么是“采集-清洗-存储-服务”四层我把整个系统拆成四层这个分层不是拍脑袋定的而是根据数据流动的自然顺序来的。采集层负责跟外部数据源打交道清洗层做格式统一和异常处理存储层解决数据怎么存、存多久的问题服务层对外暴露 API。每一层之间用消息队列或者数据库做解耦这样任何一层出问题都不会把整条链路拖死。为什么不用一个大脚本从头跑到尾我试过刚开始确实快一个main.py里写几百行跑起来也能用。但问题是数据源一多某个源挂了整个脚本就卡住想加一个新数据源得在几百行里找地方插代码想换存储方案牵一发动全身。分层之后每个数据源是一个独立的采集器互不干扰加新源就是加一个新文件的事。技术栈方面我选的是Python FastAPI PostgreSQL Redis APScheduler。Python 的生态在金融数据处理上最成熟pandas、numpy 这些库处理时间序列数据非常顺手。FastAPI 自带异步支持和自动文档写 API 效率极高。PostgreSQL 存结构化的历史数据Redis 做实时行情的缓存和去重。APScheduler 负责定时任务调度。这套组合的优点是全部可以本地跑不依赖任何云服务一台 4 核 8G 的机器就能撑起一个中等规模的服务。注意选型时不要盲目追求“最新最热”。我见过有人用 Kafka Flink 做个人项目结果光维护这套基础设施就耗掉大半精力。个人项目的第一原则是能跑起来、好维护不是技术炫技。1.3 数据源的选择逻辑与风险评估金融数据源大致分三类交易所官方接口、第三方聚合接口、网页抓取。我的策略是优先用官方接口其次用有免费额度的第三方最后才考虑抓取。官方接口最稳定但通常只覆盖自家市场第三方聚合接口方便但有配额和停服风险网页抓取最灵活但最脆弱页面一改就失效。具体到数据源我一般会同时接两到三个来源做冗余。比如行情数据主源用一家备源用另一家主源连续失败三次就自动切到备源。这个切换逻辑写在采集层的基类里所有采集器继承同一个基类自动获得重试和切换能力。这里有个经验不要等主源彻底挂了才切要设置一个失败阈值比如连续 3 次请求超时或返回异常就触发切换同时发告警通知。风险评估方面最大的坑是数据源悄悄改了字段格式。我遇到过某接口把volume字段从整数改成字符串结果入库时报错整个采集任务中断。后来我加了一层 schema 校验每个数据源定义一个期望的字段类型采集后先校验再入库不符合就丢弃并记录日志。这个校验层救了我很多次。2. 核心模块的细节解析与实操要点2.1 采集层如何写出“打不死”的数据采集器采集器的核心要求是健壮不是快。我见过太多采集脚本网络一抖动就崩或者被限流了还在死命请求。我的做法是给每个采集器套三层保护超时控制、重试退避、限流遵守。超时控制很简单所有 HTTP 请求必须设timeout我一般设 10 秒。别小看这一条很多脚本卡死就是因为没设超时一个请求挂住整个线程就废了。重试退避是指失败后不要立刻重试而是等一段时间而且等待时间要递增。我用的是指数退避第一次等 1 秒第二次 2 秒第三次 4 秒最多重试 3 次。这样既能应对临时故障又不会给上游造成压力。限流遵守是很多人忽略的。每个数据源都有隐含的频率限制你不遵守轻则被封 IP重则账号被禁。我的做法是在采集器里维护一个令牌桶根据数据源文档里写的限制来配置。比如某接口限制每分钟 60 次我就把令牌桶速率设为每秒 1 个采集任务从桶里拿令牌拿不到就等。这样即使我开了多个采集任务也不会超限。import time import requests from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry def build_session(retries3, backoff1.0): session requests.Session() retry Retry( totalretries, backoff_factorbackoff, status_forcelist[429, 500, 502, 503, 504], allowed_methods[GET] ) adapter HTTPAdapter(max_retriesretry) session.mount(https://, adapter) session.mount(http://, adapter) return session session build_session() resp session.get(https://api.example.com/quote, timeout10)这段代码是我每个采集器的标配。Retry对象自动处理重试和退避status_forcelist里列出需要重试的状态码429 是限流5xx 是服务端错误。这样写出来的采集器面对临时故障基本能自愈。实操心得重试不是万能的。如果某个接口连续返回 401 或 403说明是认证问题重试没用应该直接告警。我在重试逻辑里加了一个判断4xx 里除了 429 之外都不重试直接抛异常。2.2 清洗层把“脏数据”变成“干净数据”的关键步骤金融数据的脏超出很多人的想象。同一个指标不同来源的单位可能不一样有的用“元”有的用“万元”时间戳有的用秒有的用毫秒有的干脆是字符串缺失值有的用null有的用-有的用0。清洗层的任务就是把这些统一成一套标准格式。我的清洗流程分四步字段映射、类型转换、缺失值处理、异常值检测。字段映射是第一步每个数据源定义一个映射表把源字段名映射到标准字段名。比如 A 源的last_price和 B 源的close都映射到标准字段close_price。类型转换是第二步所有数值字段统一转成float时间字段统一转成 UTC 时间戳。缺失值处理是第三步我一般用前值填充因为金融时间序列里缺失通常意味着“没有交易”用前一个有效值填充最合理。异常值检测是第四步用简单的 3-sigma 规则超过均值 3 倍标准差的标记为异常记录但不入库。import pandas as pd import numpy as np def clean_quote(df: pd.DataFrame) - pd.DataFrame: df df.rename(columns{last_price: close_price, vol: volume}) df[close_price] pd.to_numeric(df[close_price], errorscoerce) df[volume] pd.to_numeric(df[volume], errorscoerce) df[timestamp] pd.to_datetime(df[timestamp], units, utcTrue) df df.sort_values(timestamp) df[close_price] df[close_price].ffill() df[volume] df[volume].fillna(0) mean, std df[close_price].mean(), df[close_price].std() df[is_outlier] np.abs(df[close_price] - mean) 3 * std return df这段清洗逻辑我用了很久基本能覆盖 90% 的场景。errorscoerce是关键遇到无法转换的值会变成NaN而不是报错后面统一处理。ffill是前向填充fillna(0)用于成交量因为成交量缺失通常意味着零成交。注意异常值不要直接删掉要标记出来。金融数据里暴涨暴跌可能是真实行情直接删掉会丢失重要信息。我的做法是标记is_outlier字段入库时保留但 API 返回时可以过滤。2.3 存储层历史数据与实时数据的分治策略存储层的核心矛盾是历史数据量大但查询频率低实时数据量小但查询频率高。用同一套存储方案会两头不讨好。我的策略是分治历史数据存 PostgreSQL实时数据存 Redis。PostgreSQL 存历史数据表结构按“品种 时间”做分区。比如行情表按月份分区每个月一个分区表。这样查询某个月的数据时PostgreSQL 只扫描对应分区速度很快。索引建在(symbol, timestamp)上覆盖最常见的查询模式。数据保留策略是日线数据永久保留分钟线保留 2 年tick 数据保留 3 个月。过期数据用定时任务归档到冷存储或者直接删除。Redis 存实时数据用 Hash 结构key 是quote:{symbol}field 是各个字段。设置 TTL 为 60 秒超过 60 秒没更新的数据自动过期。这样 API 查询实时行情时直接从 Redis 读响应时间在毫秒级。Redis 还用来做去重采集器每次拿到新数据先检查 Redis 里是否已存在相同时间戳的记录存在就跳过避免重复入库。CREATE TABLE quote_history ( symbol VARCHAR(20) NOT NULL, timestamp TIMESTAMPTZ NOT NULL, close_price NUMERIC(18, 6), volume BIGINT, is_outlier BOOLEAN DEFAULT FALSE, PRIMARY KEY (symbol, timestamp) ) PARTITION BY RANGE (timestamp); CREATE TABLE quote_history_2024_01 PARTITION OF quote_history FOR VALUES FROM (2024-01-01) TO (2024-02-01);分区表的写法如上每个月手动或自动建一个分区。我写了一个定时任务每月 25 号自动创建下个月的分区避免月底忘记。实操心得PostgreSQL 的NUMERIC类型比FLOAT更适合存价格因为浮点数有精度问题。我一开始用FLOAT结果发现 0.1 0.2 不等于 0.3回测时出现莫名其妙的偏差。换成NUMERIC(18, 6)后问题消失。2.4 服务层API 设计的三个核心原则服务层的 API 设计我遵循三个原则统一入口、版本控制、限流保护。统一入口是指所有数据通过一个 base URL 访问比如/api/v1/quote/{symbol}、/api/v1/fundamental/{symbol}。版本控制是指 URL 里带版本号将来接口不兼容升级时老版本继续保留一段时间给调用方迁移的时间。限流保护是指每个 API key 有调用配额超过就返回 429。FastAPI 写这些非常顺手。我用依赖注入做认证和限流每个请求先过认证中间件检查 API key 是否有效然后过限流中间件检查配额是否用完。配额信息存在 Redis 里用滑动窗口算法精确到秒。from fastapi import FastAPI, Depends, HTTPException from redis import Redis app FastAPI() redis Redis() def rate_limit(api_key: str, limit: int 100): key frate:{api_key}:{int(time.time())} current redis.incr(key) if current 1: redis.expire(key, 1) if current limit: raise HTTPException(status_code429, detailRate limit exceeded) app.get(/api/v1/quote/{symbol}) def get_quote(symbol: str, api_key: str Depends(verify_key)): rate_limit(api_key) data redis.hgetall(fquote:{symbol}) if not data: raise HTTPException(status_code404, detailSymbol not found) return data这段代码展示了限流和查询的核心逻辑。incr是原子操作多个请求同时来也不会超限。expire设置 1 秒过期实现每秒重置的滑动窗口。注意API key 不要明文存数据库存哈希值。我用hashlib.sha256对 key 做哈希验证时比对哈希值。这样即使数据库泄露攻击者也无法还原出原始 key。3. 完整实操流程从零到跑通一条数据链路3.1 环境准备与依赖安装我假设你用的是 Ubuntu 22.04 或者 macOSPython 3.10 以上。第一步是建虚拟环境别嫌麻烦这是好习惯。python -m venv venv source venv/bin/activate pip install fastapi uvicorn pandas numpy requests redis psycopg2-binary apschedulerPostgreSQL 和 Redis 的安装我就不展开了网上教程很多。装完后建一个数据库financial建一个用户fin_user密码自己设。Redis 默认配置就行注意设置maxmemory和maxmemory-policy allkeys-lru避免内存打满。配置文件我用.env管理不硬编码在代码里。DATABASE_URLpostgresql://fin_user:passwordlocalhost:5432/financial REDIS_URLredis://localhost:6379/0 API_KEY_SALTyour_random_salt_here3.2 数据库初始化与分区表创建建表 SQL 我放在migrations/目录下用简单的脚本按顺序执行。核心表有三张quote_history存行情fundamental_history存财报macro_history存宏观指标。每张表都按时间分区。import psycopg2 from datetime import datetime, timedelta def create_monthly_partition(conn, table: str, year: int, month: int): start datetime(year, month, 1) if month 12: end datetime(year 1, 1, 1) else: end datetime(year, month 1, 1) partition_name f{table}_{year}_{month:02d} sql f CREATE TABLE IF NOT EXISTS {partition_name} PARTITION OF {table} FOR VALUES FROM ({start}) TO ({end}); with conn.cursor() as cur: cur.execute(sql) conn.commit()这个函数我封装成一个定时任务每月 25 号跑一次创建下个月的分区。提前创建是为了避免月初数据写入时找不到分区。3.3 采集任务的调度与执行调度用 APScheduler我配了三个任务行情采集每 5 秒一次财报采集每天一次宏观数据每周一次。每个任务独立运行互不阻塞。from apscheduler.schedulers.blocking import BlockingScheduler scheduler BlockingScheduler() scheduler.scheduled_job(interval, seconds5, max_instances1) def collect_quotes(): for source in quote_sources: try: data source.fetch() cleaned clean_quote(data) save_to_db(cleaned) save_to_redis(cleaned) except Exception as e: logger.error(fSource {source.name} failed: {e}) scheduler.start()max_instances1很重要防止上一次任务还没跑完下一次又启动导致重复采集。异常捕获也是必须的一个源挂了不能影响其他源。3.4 API 服务的启动与验证启动 FastAPI 用 uvicorn生产环境建议用 gunicorn 加 uvicorn worker。uvicorn main:app --host 0.0.0.0 --port 8000 --workers 4启动后访问http://localhost:8000/docs能看到自动生成的 API 文档。验证流程先调/api/v1/quote/AAPL看有没有数据再调/api/v1/history/AAPL?start2024-01-01end2024-01-31看历史数据。如果返回空检查采集任务是否在跑数据库里有没有数据。实操心得我习惯在服务启动时加一个健康检查接口/health返回数据库和 Redis 的连接状态。这样用监控工具一探就知道服务是否正常不用登录服务器看日志。4. 常见问题与排查技巧实录4.1 数据采集失败的五大原因与排查路径采集失败是最常见的问题我整理了一个排查表按出现频率排序。问题现象可能原因排查方法解决方案请求超时网络问题或上游限流用 curl 手动请求同一接口增加超时时间检查限流配置返回 401/403API key 失效或权限不足检查 key 是否过期更新 key检查权限范围返回 429请求频率超限查看采集频率是否过高降低频率增加令牌桶等待数据字段缺失上游改了返回格式对比文档和实际返回更新字段映射加 schema 校验入库报错数据类型不匹配查看数据库错误日志检查清洗逻辑加类型转换这个表我贴在显示器旁边出问题先对照排查能省很多时间。4.2 数据不一致的根源分析与解决数据不一致是指同一指标在不同地方的值不一样。我遇到过三种情况一是 Redis 和 PostgreSQL 里的值不同步原因是 Redis 写入成功但数据库写入失败二是不同数据源的值有差异比如 A 源和 B 源的收盘价差了几分钱三是时区问题有的源用本地时间有的用 UTC。第一种情况的解决方法是先写数据库再写 Redis。数据库是持久化的Redis 是缓存数据库写成功才算成功。如果 Redis 写失败下次采集会补上。第二种情况无解只能选一个主源其他源做参考。我在配置里明确指定主源API 返回主源的值。第三种情况统一用 UTC所有时间戳入库前都转成 UTCAPI 返回时再转成用户指定的时区。4.3 性能瓶颈的定位与优化服务跑久了会变慢我遇到过两个瓶颈数据库查询慢和 Redis 内存满。数据库查询慢通常是索引没建对用EXPLAIN ANALYZE看查询计划发现全表扫描就加索引。我有个查询是SELECT * FROM quote_history WHERE symbol AAPL AND timestamp 2024-01-01一开始只建了symbol索引后来改成(symbol, timestamp)复合索引查询时间从 2 秒降到 20 毫秒。Redis 内存满是因为实时数据没设 TTL越积越多。解决方法是给所有 key 设 TTL并且配置maxmemory-policy allkeys-lru内存满时自动淘汰最久未使用的 key。我还加了一个监控Redis 内存超过 80% 就告警。注意优化之前先测量别凭感觉。我用pg_stat_statements扩展找出最慢的查询用redis-cli --bigkeys找出最大的 key有针对性地优化比盲目调参有效得多。4.4 安全加固的四个必做项金融数据服务的安全不能马虎。我做了四件事API key 认证、HTTPS 加密、请求日志审计、数据库最小权限。API key 认证前面说了每个调用方一个 key可以单独吊销。HTTPS 用 Nginx 做反向代理Lets Encrypt 免费证书。请求日志记录每个请求的 key、IP、时间、路径存 30 天方便追溯。数据库用户只给必要的权限采集用户只能 INSERTAPI 用户只能 SELECT不给 DELETE 和 DROP。这四条做完基本的安全底线就有了。别觉得个人项目没人攻击我搭的服务上线第三天就有人扫端口第七天就有异常请求。安全措施不是防君子是防自动化扫描工具。5. 后续扩展方向与个人经验分享这套服务跑稳定之后我陆续加了几个扩展。一个是数据质量监控面板用 Grafana 展示每个数据源的采集成功率、延迟、数据量一眼就能看出哪个源有问题。另一个是回测数据导出提供一个接口输入品种和时间范围直接返回 CSV 文件方便导入到回测框架。还有一个是Webhook 推送当某个指标超过阈值时主动推送到指定的 URL不用轮询。我个人在实际操作中的体会是金融数据服务最难的不是技术是数据的稳定性和一致性。技术方案可以照抄但数据源的变化、格式的调整、异常的处理这些都需要长期观察和积累。我建议刚开始做的时候先跑通一条最简单的链路一个数据源、一张表、一个接口跑上一周观察数据质量再逐步扩展。别一上来就搞大而全那样很容易在细节里迷失。最后再分享一个小技巧给每个数据源建一个“健康档案”记录它的历史表现比如平均响应时间、失败率、最近一次格式变更时间。这个档案用简单的 JSON 文件存就行每次采集后更新。时间长了你就能预判哪个源可能要出问题提前做准备。这个习惯帮我避免了好几次半夜被报警叫醒的情况。
