1. 金融数据服务从零搭建的完整思路1.1 为什么我要自己搭一套金融数据服务先说清楚这个项目到底在干什么。financial-services这个名字听起来很泛实际上我做的是一套面向个人开发者和小型团队的自托管金融数据聚合与分发服务。核心目标很明确把分散在不同渠道的行情数据、财报数据、宏观经济指标统一拉取、清洗、存储再通过一套标准接口对外输出供量化策略回测、看板展示、报表生成等场景调用。市面上现成的金融数据 API 不少但用下来总有几个绕不开的痛点。免费额度小得可怜稍微高频一点的请求就被限流付费方案按调用次数计费做一次全市场回测账单能吓死人更麻烦的是数据格式各家不一样今天用 A 家的行情明天想换成 B 家代码得重写一遍。我试过同时接三个数据源做交叉验证结果光是字段对齐就写了一整天。所以这套服务的定位就是做一层“数据中间层”。上游对接多个数据源下游只暴露一套统一的接口和数据结构。你换数据源的时候下游代码一行不用改。这个思路跟很多团队做的数据网关是一个道理只不过我把它做成了可以单机跑起来的轻量版本一台 2 核 4G 的云主机就能撑住日常使用。适合谁来参考如果你正在做量化研究、金融类 side project或者公司内部需要一个不依赖第三方 SaaS 的小型数据服务这套东西可以直接抄。不需要大数据集群不需要 K8s会写 Python、会用 Docker 就能跑起来。1.2 整体架构是怎么拆的我把整个服务拆成了四层从下往上分别是采集层、存储层、服务层、接入层。这么拆的原因很简单每一层的职责边界清晰出问题的时候能快速定位是哪一层的锅。采集层负责跟外部数据源打交道。每个数据源写一个独立的 adapteradapter 只做三件事发请求、解析原始响应、把数据转成内部统一格式。adapter 之间互不依赖新增一个数据源就是新增一个文件不影响其他任何代码。这个设计参考了插件模式好处是数据源挂了或者要替换改动范围极小。存储层用 PostgreSQL 做主力时序性特别强的行情数据额外走一层 Redis 缓存。为什么不全用 PostgreSQL因为行情数据写入频率高、查询模式单一全塞进关系库会让写入成为瓶颈。Redis 做最近 N 分钟的行情缓存历史数据落 PostgreSQL这个组合实测下来写入吞吐能提升三到五倍。服务层是核心业务逻辑包括数据清洗、字段标准化、去重、复权计算、指标衍生等。这一层不直接碰数据库连接而是通过 repository 模式封装数据访问方便后面换存储引擎。接入层就是对外暴露的 HTTP 接口用 FastAPI 写的。选 FastAPI 而不是 Flask主要看中它的异步支持和自动生成的接口文档。金融数据查询经常要并发拉多个数据源异步能省不少等待时间。1.3 技术选型背后的取舍逻辑技术栈这块我列个表把选型和理由说清楚方便你按自己的情况调整。组件选型选择理由替代方案语言Python 3.11金融数据处理生态成熟pandas/numpy 无可替代Go性能更好但生态弱Web 框架FastAPI异步原生支持自动文档类型校验强Flask gevent主数据库PostgreSQL 14JSON 字段支持好窗口函数强免费MySQL 8缓存Redis 7行情热点数据缓存降低数据库压力Memcached任务调度APScheduler轻量进程内调度无需额外中间件Celery RabbitMQ容器化Docker Compose单机编排够用学习成本低裸机部署这里重点说两个取舍。第一个是任务调度为什么不用 Celery。Celery 功能确实强但要额外维护 RabbitMQ 或 Redis 作为 broker对于一个单机服务来说太重了。APScheduler 直接在进程内跑定时任务配置简单出问题看日志就行。当然如果你的采集任务量很大需要分布式调度那还是得上 Celery。第二个是为什么行情数据要单独走 Redis。我做过压测纯 PostgreSQL 写入行情数据单表每秒大概能扛 2000 到 3000 条插入。但行情数据在开盘时段是爆发式写入的瞬时峰值能到每秒上万条。加一层 Redis 做缓冲先把数据写进 Redis 的 list再由后台任务批量落库写入峰值就被削平了。这个思路跟消息队列削峰是一个道理只是用 Redis 实现更轻。2. 核心模块的细节拆解与实操要点2.1 数据采集 adapter 的标准化写法adapter 是整个服务的地基写得好不好直接决定后面维护成本。我定的规范是每个 adapter 必须实现三个方法fetch_raw、parse、normalize。fetch_raw负责发请求拿原始数据parse负责把原始响应解析成 Python 字典normalize负责把字典转成内部统一的数据模型。为什么要拆成三步而不是一步到位因为调试的时候你会发现问题往往出在某一环。拆开之后你可以单独测试parse的逻辑不用真的发网络请求。我踩过的坑是早期把三步揉在一起结果数据源改了个字段名我花了两个小时才定位到是解析环节的问题。内部统一数据模型我用 Pydantic 定义核心字段包括symbol、timestamp、open、high、low、close、volume、source。source字段很重要记录数据来自哪个源方便后面做数据溯源和交叉验证。字段类型全部强制校验时间统一用 UTC 时间戳价格统一用 Decimal 而不是 float。为什么用 Decimal因为浮点数在做金融计算时会有精度丢失0.1 0.2 不等于 0.3 这种事在回测里会累积成大误差。from pydantic import BaseModel, Field from decimal import Decimal from datetime import datetime class BarData(BaseModel): symbol: str timestamp: datetime open: Decimal high: Decimal low: Decimal close: Decimal volume: Decimal source: stradapter 的注册我用了一个简单的装饰器模式。每个 adapter 文件顶部加一个register_adapter(source_name)服务启动时自动扫描并注册。这样新增数据源不用改任何配置文件符合开闭原则。注意adapter 里千万不要写业务逻辑比如复权计算、指标衍生这些。adapter 只负责“搬运”数据业务逻辑全部放在服务层。我见过有人把复权逻辑写进 adapter结果换数据源的时候复权算法也得跟着改非常痛苦。2.2 数据清洗与字段对齐的实操细节数据从不同源拉回来格式差异大到超乎想象。有的源用vol表示成交量有的用volume有的时间戳是毫秒有的是秒有的价格是字符串有的是数字。清洗层的任务就是把这些差异全部抹平。我的做法是维护一张字段映射表每个数据源对应一套映射规则。映射表用 YAML 配置改起来不用动代码。source_a: volume: vol timestamp: ts time_unit: ms source_b: volume: volume timestamp: time time_unit: s清洗流程分四步走。第一步做字段重命名按映射表把源字段名改成内部标准名。第二步做类型转换时间戳统一转成 UTC datetime价格统一转 Decimal。第三步做异常值过滤比如价格为负、成交量为负、时间戳超出合理范围的记录直接丢弃。第四步做去重同一个 symbol 同一个 timestamp 的记录只保留一条如果多条记录来自不同源按优先级保留。去重这块有个细节值得说。我用的是 PostgreSQL 的ON CONFLICT DO UPDATE语法配合唯一索引(symbol, timestamp, source)。这样重复插入的时候会自动更新而不是报错。但如果你希望保留多源数据做交叉验证唯一索引就不要带source改成(symbol, timestamp)冲突时按源优先级决定保留哪条。异常值过滤的阈值我建议不要写死做成可配置的。比如价格上限不同标的差异很大比特币和国债的价格上限显然不能一样。我一开始写死了 100000结果遇到一个高价股直接给过滤掉了排查半天才发现是阈值问题。2.3 存储层的表结构设计与索引优化PostgreSQL 这边我建了三张核心表bars存行情 K 线fundamentals存财报数据macro存宏观经济指标。三张表结构类似都是symbol timestamp做联合主键外加一堆指标字段。bars表的字段设计有个讲究。我把open/high/low/close/volume都设成NUMERIC(20, 8)而不是FLOAT。前面说过 Decimal 的精度问题数据库层面也要对应上。NUMERIC(20, 8)表示总共 20 位数字其中 8 位小数足够覆盖绝大多数金融标的的精度需求。索引方面联合主键(symbol, timestamp)本身就是一个 B-tree 索引查询单个标的的时间范围数据走这个索引就够了。但如果你的查询模式是“查某个时间点所有标的的数据”那还需要在timestamp上单独建一个索引。我实测过不加这个索引跨标的查询会走全表扫描数据量上百万行的时候慢到无法接受。分区表这块我犹豫过要不要上。PostgreSQL 原生支持按范围分区按月份分区能显著提升查询性能尤其是历史数据查询。但分区表的管理成本也上去了每个月要提前建好下个月的分区。对于数据量在千万行以内的场景我建议先不上分区等真的遇到性能瓶颈再加。过早优化是万恶之源这话在数据库设计上同样适用。Redis 缓存这块我用的是ZADD存行情数据score 用时间戳member 用序列化后的 bar 数据。查询最近 N 条行情的时候用ZREVRANGE按 score 倒序取效率很高。缓存过期时间设成 24 小时因为历史行情基本不变当天行情在收盘后也不再更新。提示Redis 存 Decimal 类型的时候要先转成字符串取出来再转回 Decimal。直接存 float 会丢精度这个坑我踩过回测结果对不上就是因为缓存里的价格精度丢了。2.4 服务层的数据复权与指标计算复权是金融数据处理里绕不开的一环。前复权、后复权、不复权三种模式对应不同的使用场景。回测一般用前复权看历史真实价格用不复权计算收益率用后复权。我的做法是在服务层提供adjust参数调用方指定用哪种复权模式。复权计算的核心是复权因子。复权因子来自分红送股数据每次除权除息事件都会产生一个新的因子。计算逻辑是后复权价格 原始价格 × 累计复权因子前复权价格 原始价格 × (当前因子 / 历史因子)。这个公式看起来简单但累计因子的计算容易出错尤其是遇到多次除权的情况。我踩过的坑是复权因子没有做精度控制连续乘了几十次之后误差累积到肉眼可见。后来改成每次乘法都用 Decimal 的quantize方法保留 8 位小数误差就控制住了。指标计算这块我实现了常用的 MA、EMA、MACD、RSI、布林带。这些指标的计算逻辑不复杂但要注意计算窗口的边界处理。比如 MA20 需要至少 20 个数据点才能算出第一个值前 19 个点应该返回空而不是用不完整的数据算。我见过有人用pandas.rolling(20).mean()直接算前 19 个是 NaN但如果不做处理直接参与后续计算结果会全错。指标计算我建议用 pandas 的向量化操作不要写 for 循环。同样计算 10 万条数据的 MA20向量化比循环快两个数量级。如果数据量特别大可以考虑用 numpy 的sliding_window_view内存效率更高。3. 完整实操流程与关键环节实现3.1 环境准备与依赖安装先把环境搭起来。我假设你用的是 Ubuntu 22.04其他 Linux 发行版步骤类似。Windows 用户建议用 WSL2原生 Windows 跑 Docker 会有一些路径和权限的坑。第一步装 Docker 和 Docker Compose。官方一键脚本最省事curl -fsSL https://get.docker.com | sh sudo usermod -aG docker $USER装完之后重新登录一下让用户组生效。然后验证docker --version docker compose version第二步创建项目目录结构。我习惯这样组织financial-services/ ├── docker-compose.yml ├── .env ├── app/ │ ├── main.py │ ├── adapters/ │ ├── services/ │ ├── repositories/ │ └── models/ ├── config/ │ └── mappings/ └── scripts/ └── init_db.sql第三步写docker-compose.yml。核心是三个服务PostgreSQL、Redis、应用本身。version: 3.8 services: db: image: postgres:14 environment: POSTGRES_USER: finuser POSTGRES_PASSWORD: finpass POSTGRES_DB: findb volumes: - pgdata:/var/lib/postgresql/data ports: - 5432:5432 redis: image: redis:7-alpine ports: - 6379:6379 app: build: . depends_on: - db - redis env_file: .env ports: - 8000:8000 volumes: pgdata:.env文件里放数据库连接串、Redis 地址、各数据源的 API Key。API Key 千万不要硬编码在代码里也不要提交到 Git。我一般用.env加.gitignore的组合简单有效。3.2 数据库初始化与表结构创建数据库初始化脚本我放在scripts/init_db.sqlDocker Compose 启动 PostgreSQL 的时候会自动执行docker-entrypoint-initdb.d目录下的脚本。把脚本挂载进去就行。CREATE TABLE IF NOT EXISTS bars ( symbol VARCHAR(32) NOT NULL, timestamp TIMESTAMPTZ NOT NULL, open NUMERIC(20, 8) NOT NULL, high NUMERIC(20, 8) NOT NULL, low NUMERIC(20, 8) NOT NULL, close NUMERIC(20, 8) NOT NULL, volume NUMERIC(30, 8) NOT NULL, source VARCHAR(32) NOT NULL, PRIMARY KEY (symbol, timestamp, source) ); CREATE INDEX idx_bars_timestamp ON bars (timestamp); CREATE INDEX idx_bars_symbol_ts ON bars (symbol, timestamp DESC);idx_bars_symbol_ts这个索引用了DESC因为查询最近行情的时候是按时间倒序取的倒序索引能直接命中不用额外排序。这个细节在数据量大之后能省不少时间。财报表和宏观指标表结构类似只是字段不同这里不展开。建完表之后用psql连进去验证一下docker compose exec db psql -U finuser -d findb -c \dt看到三张表就说明初始化成功了。3.3 采集任务的调度与执行采集任务我用 APScheduler 的BackgroundScheduler在 FastAPI 启动的时候一起拉起来。任务分两类定时任务和手动触发任务。定时任务按 cron 表达式跑比如行情数据每 5 分钟拉一次财报数据每天凌晨拉一次。手动触发任务通过 HTTP 接口调用用于补数据或者调试。from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger scheduler BackgroundScheduler() scheduler.add_job( fetch_daily_bars, CronTrigger(minute*/5, hour9-15, day_of_weekmon-fri), idfetch_bars, max_instances1, coalesceTrue )max_instances1保证同一个任务不会并发执行避免重复拉数据。coalesceTrue表示如果任务积压了只执行最近一次跳过中间那些。这两个参数在任务执行时间超过调度间隔的时候特别重要不加的话会出现任务堆积。采集任务的执行流程是adapter 拉数据 → 清洗 → 写 Redis → 批量落库。批量落库我用的是execute_values比逐条 insert 快很多。一次批量插入 1000 条实测比逐条插入快 20 倍以上。from psycopg2.extras import execute_values def bulk_insert_bars(conn, bars): sql INSERT INTO bars (symbol, timestamp, open, high, low, close, volume, source) VALUES %s ON CONFLICT (symbol, timestamp, source) DO UPDATE SET open EXCLUDED.open, high EXCLUDED.high, low EXCLUDED.low, close EXCLUDED.close, volume EXCLUDED.volume execute_values(conn.cursor(), sql, bars, page_size1000)ON CONFLICT DO UPDATE保证重复数据不会报错而是更新为最新值。这个在处理数据源修正历史数据的时候特别有用。3.4 接口设计与查询性能优化对外接口我设计了四个核心端点/bars查行情/fundamentals查财报/macro查宏观指标/indicators查技术指标。每个端点都支持symbol、start、end、limit参数。/bars接口的查询逻辑是先查 Redis 缓存缓存命中直接返回缓存未命中查 PostgreSQL查完写回缓存。缓存 key 的设计是bars:{symbol}:{start}:{end}:{limit}这样不同查询参数对应不同缓存避免缓存污染。分页这块我用的是游标分页而不是 offset 分页。offset 分页在数据量大之后性能急剧下降因为数据库要扫描并跳过前面所有行。游标分页用WHERE timestamp last_timestamp的方式每次查询都走索引性能稳定。def get_bars(symbol, start, end, limit1000, cursorNone): query SELECT * FROM bars WHERE symbol %s AND timestamp BETWEEN %s AND %s params [symbol, start, end] if cursor: query AND timestamp %s params.append(cursor) query ORDER BY timestamp DESC LIMIT %s params.append(limit) return db.execute(query, params).fetchall()接口返回格式统一用 JSON时间戳统一用 ISO 8601 格式价格统一用字符串表示 Decimal。为什么价格用字符串因为 JSON 标准里没有 Decimal 类型用 float 会丢精度用字符串最安全。前端拿到字符串再自己转数字。注意接口一定要加限流。我用的是slowapi基于 IP 做限流默认每分钟 60 次。不加限流的话一个死循环的客户端就能把你的服务打挂。这个教训来自我早期的一次事故一个测试脚本写错了循环条件十分钟发了 50 万次请求数据库直接被打爆。4. 常见问题与排查技巧实录4.1 数据源不稳定导致的采集失败数据源不稳定是常态尤其是免费数据源。我遇到过的情况包括接口超时、返回 502、返回空数据、返回格式变了、突然要加验证码。应对策略分三层重试、降级、告警。重试我用的是指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒最多重试 3 次。为什么用指数退避而不是固定间隔因为数据源挂掉往往是瞬时的固定间隔重试可能连续撞上故障窗口指数退避能错开时间。重试代码用tenacity库几行就能搞定。from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min1, max10)) def fetch_with_retry(url): return requests.get(url, timeout10)降级策略是主数据源挂了自动切到备用数据源。我在 adapter 层做了一个优先级列表按顺序尝试第一个成功的就用。备用源的数据质量可能差一点但总比没有强。告警我用的是简单的邮件通知采集任务连续失败 3 次就发邮件。邮件里带上失败的任务名、错误信息、时间戳。不要搞太复杂的告警系统个人项目用邮件足够了。我试过接 Slack webhook后来发现邮件更可靠而且有存档方便回溯。4.2 数据精度丢失的排查与修复精度问题是最隐蔽的 bug因为它在小数据量下看不出来数据量大了才暴露。我遇到过一次回测结果和实盘对不上排查了两天才发现是复权因子计算时用了 float累积误差导致价格偏差了 0.5%。排查精度问题的思路是先定位是哪一环节引入的误差。我的做法是在关键节点打印 Decimal 的完整精度对比上下游数据。如果上游是精确的下游变了那问题就在中间环节。修复方案前面提过全链路用 Decimal。但要注意Decimal 和 float 混用的时候Python 会隐式转换精度还是会丢。所以从数据库读取、计算、序列化、反序列化每一步都要确保是 Decimal。pandas 的read_sql默认会把 NUMERIC 转成 float需要显式指定dtype或者读完之后转回 Decimal。df pd.read_sql(query, conn) df[close] df[close].apply(Decimal)这个转换会慢一点但精度有保障。如果性能实在扛不住可以考虑用pandas的objectdtype 存 Decimal避免自动转 float。4.3 数据库连接池耗尽的处理连接池耗尽通常发生在并发请求量大的时候。症状是接口响应变慢日志里出现connection pool exhausted或者timeout。根本原因是连接用完没释放或者连接数配置太小。排查第一步是看当前活跃连接数SELECT count(*) FROM pg_stat_activity WHERE state active;如果活跃连接数接近max_connections那就是连接泄漏了。常见原因是代码里开了连接没关或者异常路径下没走到close。修复方法是用上下文管理器确保连接一定会释放。from contextlib import contextmanager contextmanager def get_conn(): conn pool.getconn() try: yield conn finally: pool.putconn(conn)连接池大小我建议设成max_connections的 70% 左右留一些余量给管理连接。PostgreSQL 默认max_connections是 100那连接池设 70 比较合适。设太大反而会因为上下文切换拖慢性能。4.4 常见问题速查表问题现象可能原因排查方法解决方案采集任务失败数据源超时/限流看日志错误码加重试降级回测结果偏差精度丢失对比上下游数据全链路 Decimal接口响应慢缓存未命中/索引缺失看慢查询日志加缓存/加索引连接池耗尽连接泄漏查 pg_stat_activity上下文管理器数据重复去重逻辑失效查唯一索引加 ON CONFLICT内存暴涨全量加载数据看内存监控改分页查询这张表是我踩坑踩出来的遇到问题先对照查一遍能省不少时间。4.5 几个独家避坑技巧第一个技巧采集任务加一个“数据新鲜度”检查。每次采集前先查一下最新数据的时间戳如果距离现在不到 5 分钟就跳过这次采集。这个能避免重复拉取相同数据节省 API 额度。我早期没加这个一天多消耗了 30% 的额度。第二个技巧数据库写入用COPY而不是INSERT。PostgreSQL 的COPY命令是批量导入的最快方式比INSERT快一个数量级。数据量大的时候先把数据写成 CSV再用COPY导入速度提升非常明显。不过COPY不支持ON CONFLICT需要先导入临时表再合并。第三个技巧日志里记录每个采集任务的耗时和条数。这个数据积累下来能帮你发现很多问题。比如某个数据源突然变慢或者某个时段数据量异常日志里一眼就能看出来。我用的是结构化日志JSON 格式方便后面用脚本分析。第四个技巧定期做数据一致性校验。每周跑一次脚本对比主数据源和备用数据源的数据差异超过阈值就告警。这个能提前发现数据源悄悄改了数据格式或者计算逻辑的情况。我遇到过数据源偷偷改了复权算法导致历史数据全变了如果没有校验回测结果会莫名其妙地漂移。这套financial-services我从最初的一个脚本慢慢迭代到现在这个结构前后大概花了三个月。中间重构过两次第一次是把所有逻辑塞在一个文件里后来拆成了模块第二次是把同步代码改成了异步接口响应时间从平均 200ms 降到了 50ms。如果你刚开始做建议先跑通最小闭环再逐步加功能不要一上来就追求完美架构。