1. 金融数据服务从零搭建的完整思路1.1 这个项目到底在做什么第一次看到financial-services这个命名的时候我脑子里蹦出来的第一个念头就是这又是一个把行情、财报、宏观数据揉在一起往外吐的聚合服务。事实也确实如此。这类项目的核心目标非常明确——把散落在不同源头、格式五花八门的金融数据经过清洗、对齐、缓存之后用一套统一的接口暴露出去让上层的量化策略、看板、风控系统或者个人研究脚本都能直接消费。它解决的问题其实很朴素。做过金融数据的人都知道原始数据有多脏时间戳时区不统一、字段命名一会儿驼峰一会儿下划线、停牌日缺失、复权因子对不上、财报口径隔年就变。如果每个消费方都自己去对接、自己去清洗那就是重复造轮子而且每个人造的轮子还都不一样。financial-services这类项目的价值就在于做那个中间层把脏活累活收敛到一处。适合谁来参考呢我的判断是三类人一是想给自己量化策略搭一套稳定数据源的个人开发者二是在小团队里负责数据基建、需要快速交付一个可用数据服务的后端工程师三是做金融类产品、需要理解数据链路全貌的产品或分析师。哪怕你只是想搞清楚行情数据从交易所到我的屏幕中间经历了什么跟着这个思路走一遍也会有收获。1.2 为什么选择服务化而不是脚本化很多人一开始是写脚本的一个fetch_data.py跑定时任务把数据存进 CSV 或者 SQLite用的时候直接读文件。这个方案在数据量小、只有自己用的时候完全没问题我自己早期也是这么干的。但一旦出现下面几种情况脚本方案就会迅速崩掉多个消费方同时要数据文件锁、并发读写问题开始冒头数据源接口有频率限制脚本各自为战导致被限流需要做权限控制不同的人只能看不同的数据数据要实时更新脚本的批处理延迟满足不了。服务化的本质是把数据获取和数据消费解耦。消费方不再关心数据从哪来、怎么清洗只关心接口稳不稳定、字段全不全。这个解耦带来的好处是巨大的数据源换了消费方无感知清洗逻辑改了只改一处要加缓存、加限流、加监控都有统一的切入点。从架构选型上讲我倾向于分层设计而不是把所有逻辑塞进一个进程。典型的分层是这样的层级职责常见实现接入层对外暴露 HTTP/RPC 接口做鉴权、限流Web 框架 网关服务层业务逻辑字段组装、口径转换业务代码数据层统一读写屏蔽底层存储差异ORM / 数据访问对象采集层对接外部数据源定时拉取调度器 采集任务存储层持久化热数据缓存关系库 缓存 时序库这么分的好处是每一层都能独立演进。比如某天你发现某个数据源不稳定只需要在采集层加个备用源服务层和接入层完全不用动。这种改动局部化的能力是服务化相比脚本化最实在的收益。1.3 数据源选型的取舍逻辑金融数据源大致分几类公开免费接口、商业数据供应商、自建采集。选型的时候不能只看免费不免费要综合看稳定性、字段完整度、历史深度、更新频率和合规性。免费公开接口的优点是零成本、上手快缺点是稳定性没保证、字段经常缺、历史数据浅、还有调用频率限制。商业供应商字段全、历史深、有 SLA但成本高而且不同供应商的口径不一样切换成本不低。自建采集最灵活但维护成本高还要处理反爬、数据校验等一堆事。我的经验是分层组合核心的、高频的、对稳定性要求高的数据比如日线行情用商业源或者自建边缘的、低频的、锦上添花的数据比如某些宏观指标用免费源并且做好降级预案。关键是所有数据源都要经过同一套清洗和校验流程不能让上游的差异渗透到下游。提示无论用哪种数据源一定要在采集层做原始数据落地。也就是拉回来的原始报文先原样存一份再做清洗。这样一旦清洗逻辑有 bug可以重放原始数据修复而不用重新去拉。这个习惯帮我省过无数次事。2. 核心模块拆解与关键实现细节2.1 数据模型设计字段口径统一是命门金融数据服务最容易翻车的地方不是技术而是口径。同一个收盘价不同数据源可能一个是未复权、一个是前复权、一个是后复权同一个净利润可能是归母、可能是扣非、可能是单季、可能是累计。如果数据模型不把这些维度显式建模出来下游用的时候必然踩坑。我的做法是在数据模型里强制带上口径标签。以行情为例一条 K 线记录至少包含这些字段symbol标的代码统一成带市场前缀的格式比如SH600000、SZ000001trade_date交易日统一成YYYY-MM-DD不带时区歧义open/high/low/close价格明确标注是否复权volume/amount成交量和成交额单位统一股/手、元/千元要写清楚adjust_type复权类型none/qfq/hfqsource数据来源标识方便追溯updated_at这条记录的更新时间。把adjust_type和source显式建模看起来是小事但它让数据可追溯成为可能。下游发现某个价格不对能立刻定位到是哪个源、哪种复权口径出的问题。财报数据更复杂我一般会拆成三张表利润表、资产负债表、现金流量表每张表用report_date报告期和announce_date公告日双时间戳。为什么要两个时间因为做回测的时候必须用当时能看到的数据也就是按announce_date对齐否则就是未来函数回测结果全是假的。这个坑我见过太多人踩。2.2 采集调度别让定时任务变成定时炸弹采集层最常见的实现是定时任务但定时任务有几个经典陷阱任务重叠、失败无重试、单点故障、时间漂移。任务重叠是指上一次还没跑完下一次又启动了导致同一份数据被重复写入或者互相覆盖。解决办法是加分布式锁或者任务状态标记确保同一任务同一时刻只有一个实例在跑。我用过最简单的方案是基于数据库的唯一约束任务开始前插入一条带唯一键的运行记录插入失败就说明有实例在跑直接退出。失败重试要区分可重试和不可重试。网络超时、限流这种是暂时性的可以指数退避重试字段解析失败、数据校验不通过这种是逻辑性的重试多少次都没用应该直接告警。把这两类分开处理能避免无意义的重复劳动。时间漂移是指任务实际执行时间和预期时间越来越偏。比如你设了每天 8 点跑但任务本身耗时波动累积下来可能跑到 9 点。对于金融数据很多源在特定时间点更新错过窗口就得等第二天。我的做法是任务只负责触发具体拉取逻辑根据当前时间动态决定拉哪个区间而不是写死。# 采集任务的核心骨架示意 def run_collection(task_name, fetch_fn, store_fn, lock): if not lock.acquire(task_name): return # 已有实例在跑直接退出 try: raw fetch_fn() # 拉取原始数据 persist_raw(raw) # 原始数据先落地 cleaned clean(raw) # 清洗 validate(cleaned) # 校验不通过就抛异常 store_fn(cleaned) # 入库 lock.release(task_name, successTrue) except TransientError as e: lock.release(task_name, successFalse) schedule_retry(task_name, backoffTrue) # 暂时性错误重试 except FatalError as e: lock.release(task_name, successFalse) alert(task_name, e) # 逻辑性错误告警2.3 缓存策略热数据放内存冷数据落盘金融数据服务的读请求往往远多于写请求而且有明显的冷热分布。当天的行情、最近的财报被反复查询几年前的日线可能一个月才被翻一次。针对这种分布缓存策略要分层。热数据放内存缓存设置合理的过期时间。行情类数据在交易时段更新频繁过期时间可以短一些比如几十秒到几分钟财报类数据更新频率低过期时间可以长一些几小时甚至一天。冷数据直接落盘查询时走数据库必要时加一层磁盘缓存。这里有个容易忽略的点缓存键的设计。如果缓存键只包含symbol那不同复权类型、不同时间区间的请求会互相污染。缓存键必须包含所有影响结果的参数比如symbol adjust_type start_date end_date fields。键太长会影响性能可以用哈希压缩但要保证唯一性。注意缓存和数据库的一致性是永恒的话题。我的原则是缓存只做加速不做真相来源。任何写操作先落库再失效缓存绝不先写缓存。这样即使缓存出问题数据也不会丢。2.4 接口设计让消费方用得舒服接口设计的好坏直接决定这个服务好不好用。我见过太多数据服务字段命名随心所欲错误码含义不明分页参数一会儿是page一会儿是offset用起来极其痛苦。我的接口设计原则有这么几条。第一字段命名统一风格要么全下划线要么全驼峰别混着来。第二时间参数统一格式我一般用YYYY-MM-DD表示日期用 ISO8601 表示时间戳。第三分页统一用limit offset或者游标别两套并存。第四错误响应结构化包含错误码、错误信息、可能的解决建议。一个典型的行情查询接口大概长这样GET /api/v1/quotes?symbolSH600000start2024-01-01end2024-03-31adjustqfqfieldsopen,high,low,close,volume响应体里除了数据我还会带上元信息比如数据来源、最后更新时间、是否命中缓存。这些元信息对排查问题特别有用消费方一看就知道拿到的数据新不新、从哪来的。3. 完整实操流程与关键环节落地3.1 环境准备与依赖安装假设我们从零开始搭一个最小可用的版本。技术栈我选 Python 生态因为金融数据处理相关的库最全上手也快。Web 框架用 FastAPI异步性能好自带文档数据库用 PostgreSQL对时间序列和复杂查询支持都不错缓存用 Redis调度用 APScheduler 或者 Celery Beat。先建虚拟环境把依赖装好python -m venv venv source venv/bin/activate pip install fastapi uvicorn sqlalchemy psycopg2-binary redis apscheduler pandas numpy requests pydantic数据库和 Redis 我建议用容器起省得污染本机环境docker run -d --name fin-pg -e POSTGRES_PASSWORDsecret -p 5432:5432 postgres:16 docker run -d --name fin-redis -p 6379:6379 redis:7这套环境跑起来一台普通开发机绰绰有余。生产环境的话数据库要做主从Redis 要做持久化调度器要做高可用这些后面再说。3.2 数据库表结构设计行情表我一般这么建CREATE TABLE quotes ( id BIGSERIAL PRIMARY KEY, symbol VARCHAR(16) NOT NULL, trade_date DATE NOT NULL, adjust_type VARCHAR(8) NOT NULL DEFAULT none, open NUMERIC(18, 4), high NUMERIC(18, 4), low NUMERIC(18, 4), close NUMERIC(18, 4), volume BIGINT, amount NUMERIC(24, 4), source VARCHAR(32), updated_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE (symbol, trade_date, adjust_type) ); CREATE INDEX idx_quotes_symbol_date ON quotes (symbol, trade_date);那个唯一约束(symbol, trade_date, adjust_type)很关键它保证了同一标的同一天同一种复权口径只有一条记录重复采集时用ON CONFLICT DO UPDATE就能实现幂等写入不用先删后插。财报表我拆成三张以利润表为例CREATE TABLE income_statement ( id BIGSERIAL PRIMARY KEY, symbol VARCHAR(16) NOT NULL, report_date DATE NOT NULL, announce_date DATE NOT NULL, revenue NUMERIC(24, 4), net_profit NUMERIC(24, 4), net_profit_deducted NUMERIC(24, 4), source VARCHAR(32), updated_at TIMESTAMPTZ DEFAULT NOW(), UNIQUE (symbol, report_date, source) ); CREATE INDEX idx_income_announce ON income_statement (symbol, announce_date);注意announce_date上的索引回测时按公告日查询是高频操作这个索引能省很多时间。3.3 采集任务的实现采集任务的核心是拉取-清洗-校验-入库四步。我以日线行情为例写一个可复用的骨架import pandas as pd from datetime import date, timedelta def fetch_daily_quotes(symbol, start, end): 从数据源拉取原始数据返回 DataFrame # 这里对接具体数据源示意用 raw data_source.get_kline(symbol, start, end) return pd.DataFrame(raw) def clean_quotes(df): 清洗统一字段名、类型、去重 df df.rename(columns{ date: trade_date, vol: volume, turnover: amount }) df[trade_date] pd.to_datetime(df[trade_date]).dt.date df[symbol] df[symbol].str.upper() df df.drop_duplicates(subset[symbol, trade_date]) return df def validate_quotes(df): 校验价格非负、高低价关系合理、成交量非负 assert (df[close] 0).all(), 存在负价格 assert (df[high] df[low]).all(), 最高价低于最低价 assert (df[volume] 0).all(), 存在负成交量 return True def store_quotes(df, adjust_typenone, sourcedemo): 幂等写入 for _, row in df.iterrows(): upsert_quote(row, adjust_type, source)这里有几个实操细节值得说。第一validate_quotes里的断言不是摆设是真正的防线。我遇到过数据源返回的high小于low的情况如果不校验直接入库下游算出来的技术指标全是错的。第二清洗和校验要分开清洗负责格式校验负责逻辑职责清晰才好排查。第三入库用 upsert 而不是 insert保证重复跑任务不会产生脏数据。3.4 服务接口的实现接口层用 FastAPI 写起来很清爽from fastapi import FastAPI, Query, HTTPException from datetime import date app FastAPI(titleFinancial Services API) app.get(/api/v1/quotes) async def get_quotes( symbol: str Query(..., description标的代码如 SH600000), start: date Query(...), end: date Query(...), adjust: str Query(none, regex^(none|qfq|hfq)$), fields: str Query(open,high,low,close,volume), ): cache_key fq:{symbol}:{start}:{end}:{adjust}:{fields} cached redis_client.get(cache_key) if cached: return {data: json.loads(cached), meta: {cached: True}} rows query_quotes(symbol, start, end, adjust, fields.split(,)) if not rows: raise HTTPException(status_code404, detail无数据) redis_client.setex(cache_key, 60, json.dumps(rows)) return {data: rows, meta: {cached: False, source: db}}这个接口有几个设计点。参数校验用Query的regex约束复权类型非法值直接 400不用进业务逻辑。缓存键包含所有影响结果的参数避免污染。返回体带meta字段标明是否命中缓存方便调试。缓存过期时间设 60 秒交易时段够用非交易时段其实可以更长但统一设短一点更省心。3.5 调度配置调度我用 APScheduler配置简单适合中小规模from apscheduler.schedulers.blocking import BlockingScheduler from apscheduler.triggers.cron import CronTrigger scheduler BlockingScheduler() # 每个交易日收盘后拉取日线 scheduler.add_job( collect_daily_quotes, CronTrigger(day_of_weekmon-fri, hour16, minute30), iddaily_quotes, max_instances1, # 防止任务重叠 misfire_grace_time3600, # 错过窗口1小时内仍执行 ) scheduler.start()max_instances1是防重叠的关键参数misfire_grace_time处理任务错过执行窗口的情况。这两个参数不设定时任务迟早出问题。4. 常见问题排查与避坑经验实录4.1 数据不一致的排查思路数据不一致是金融数据服务最头疼的问题。表现可能是同一个标的同一天的价格接口返回的和数据库里的对不上或者两个接口返回的同一字段值不同。排查我一般按这个顺序走。第一步确认是不是缓存问题直接查数据库对比。第二步确认是不是复权口径问题检查请求参数里的adjust。第三步确认是不是数据源本身的问题看原始落地数据。第四步确认是不是并发写入导致的看updated_at时间戳。我整理了一个速查表现象可能原因排查方法接口值与库值不符缓存未失效清缓存后重查同标的不同复权值混淆缓存键缺 adjust检查缓存键构造历史数据被覆盖采集任务重叠查任务运行记录财报数据对不上报告期/公告期混用确认查询用的是哪个时间价格明显异常数据源脏数据查原始落地数据提示我强烈建议在服务里加一个数据质量巡检任务每天定时跑一批校验规则比如价格连续性、成交量突变、财报勾稽关系。发现问题主动告警比等用户反馈强得多。4.2 性能瓶颈的定位与优化数据服务跑一段时间后性能问题会逐渐显现。常见的瓶颈有这么几个数据库慢查询、缓存命中率低、接口序列化开销大、采集任务拖垮数据库。数据库慢查询最常见的原因是缺索引或者索引没走对。用EXPLAIN ANALYZE看执行计划如果出现全表扫描基本就是索引问题。行情表按(symbol, trade_date)建索引后单标的区间查询应该走索引扫描如果没走检查查询条件里的类型是否匹配比如trade_date是DATE类型传字符串可能导致隐式转换而放弃索引。缓存命中率低往往是缓存键设计不合理或者过期时间太短。我一般会统计缓存命中率低于 70% 就要审视缓存策略。有时候是查询参数太发散比如用户传的fields顺序不同导致缓存键不同这种情况可以在构造缓存键前对参数做规范化。采集任务拖垮数据库通常是因为批量写入没控制好。我的做法是分批写入每批几百到几千条批间加短暂 sleep给数据库喘息的机会。另外采集任务尽量避开业务高峰比如放在凌晨。4.3 数据源切换的平滑过渡数据源不是一成不变的供应商涨价、接口下线、质量下降都可能需要切换。切换最怕的是一刀切新源一上历史数据对不上下游全乱。平滑过渡的做法是双写双读对比。新源接入后先不切换消费方而是让新旧两个源同时采集写进不同的表或者带不同的source标识。然后跑一个对比任务逐条比对两个源的数据差异差异在可接受范围内的字段才切换。切换时也不是全量切而是按标的、按时间段逐步切观察一段时间没问题再全量。这个过程中source字段的价值就体现出来了。消费方如果发现数据异常能立刻知道是哪个源的数据方便定位。4.4 几个我踩过的坑第一个坑是时区。早期我没在意时区数据源返回的是 UTC 时间我当成北京时间存了结果所有时间戳差了 8 小时。行情数据还好财报的公告时间差 8 小时可能就跨天了回测直接错位。后来我统一规定所有入库时间戳一律用带时区的TIMESTAMPTZ展示时再转本地时区。第二个坑是浮点数精度。价格用FLOAT存累加计算后出现0.30000000000000004这种结果。金融数据对精度敏感必须用NUMERIC或者DECIMAL别图省事用浮点。第三个坑是停牌处理。停牌日数据源可能返回空也可能返回前收盘价填充。如果不区分算出来的收益率会失真。我的做法是显式记录停牌状态停牌日的行情标记出来下游计算时自行决定怎么处理。第四个坑是复权因子的更新。前复权价格会随着新的除权除息事件而变化也就是说历史的前复权价格不是一成不变的。如果缓存了前复权数据又没及时失效用户拿到的就是过时的价格。解决办法是除权除息事件发生后主动失效相关标的的缓存。4.5 监控与告警的落地服务上线只是开始能不能稳定运行靠的是监控。我一般监控这几个维度接口的 QPS、延迟、错误率采集任务的成功率、耗时、数据量数据质量指标比如每日新增记录数、异常值比例缓存命中率、内存使用。告警要分级别什么都告警否则告警疲劳真出事了反而没人看。我的分级是P0 是服务不可用或者数据大面积错误立即电话告警P1 是部分功能异常或者数据质量下降工作时间告警P2 是性能指标偏离记录到日报。监控数据本身也要存起来方便事后复盘。我见过太多团队出问题时才发现监控数据只保留了几小时根本没法追溯。5. 服务扩展与长期维护的几点思考5.1 从单机到分布式的演进路径一开始单机跑完全没问题一台配置好点的机器日线行情加财报数据撑几年数据量也就几十 GB。但业务增长后单机会遇到瓶颈磁盘、内存、CPU、还有单点故障。演进路径我建议是渐进的别一上来就上微服务。第一步数据库读写分离主库写从库读缓解读压力。第二步采集和服务拆开部署采集任务重的时候不影响接口响应。第三步接口层水平扩展多个实例前面挂负载均衡。第四步数据分片按标的或者按时间把数据分散到多个库。每一步都要有明确的触发条件比如数据库 CPU 持续超过 70%才做读写分离接口 P99 延迟超过 500ms才做水平扩展。没有明确指标就盲目扩展只会增加复杂度。5.2 数据版本管理金融数据有个特殊需求可重现。同样的查询今天和半年前应该返回同样的结果除非数据源本身修正了历史数据。这就要求数据有版本管理。我的做法是给每条记录加version和valid_from、valid_to字段数据修正时不覆盖旧记录而是把旧记录标记失效插入新记录。查询时默认查当前有效版本需要历史版本时按时间点查。这样既保证了可重现又保留了修正历史。这个方案会增加存储和查询复杂度但对于严肃的金融应用是值得的。如果只是个人研究可以简化但至少要保留原始落地数据以便必要时重建。5.3 合规与数据安全金融数据涉及合规问题尤其是涉及个人交易数据或者未公开信息时。我的原则是只采集和使用公开数据不碰任何可能涉及内幕的信息用户数据脱敏存储敏感字段加密接口做鉴权和审计谁在什么时候查了什么数据要有记录。数据安全方面数据库不要暴露公网接口必须走鉴权敏感配置用环境变量或者密钥管理服务别硬编码在代码里。这些是老生常谈但每年都有因为硬编码密钥泄露的事故。5.4 文档与交接数据服务最怕的是只有作者懂。字段含义、口径定义、数据源特性、已知问题这些如果不写下来作者一走接手的人得从头摸索。我一般会维护三份文档一份是接口文档用 OpenAPI 自动生成保证和代码同步一份是数据字典逐字段说明含义、单位、口径、来源一份是运维手册写清楚部署、监控、常见问题处理。文档不用写得多漂亮但要准确、及时更新。提示数据字典尤其重要。我见过太多项目字段名看着一样实际口径天差地别就是因为没有数据字典。花半天时间把数据字典写清楚能省下后面无数次的扯皮。这套东西搭下来一个最小可用的金融数据服务就成型了。从采集到清洗到接口到监控每个环节都有它的门道没有哪一步是可以随便糊弄的。我自己是从写脚本一步步踩坑踩过来的回头看最值钱的不是代码而是那些踩过的坑和总结出的规矩。数据这东西脏起来超乎想象但只要流程立住了后面就是体力活。