1. 金融数据服务从零搭建的完整思路

1.1 这个项目到底在做什么

“financial-services”这个标题看起来很大，实际上落到具体项目里，它通常指的是一套面向金融场景的数据服务层——把行情、账户、交易、风控、报表这些模块的数据统一收口，对外提供稳定的接口和计算能力。我做过的几个类似项目，核心诉求基本一致：数据源多且杂、实时性要求高、下游调用方五花八门，如果没有一层专门的服务来兜底，整个系统会迅速变成一团乱麻。

这个项目适合谁参考？如果你正在做一个跟金融数据打交道的系统，不管是个人记账工具、量化回测平台、企业内部的资金管理系统，还是给外部客户提供数据接口的中间层，这套思路都能直接套用。它解决的问题不是“怎么写一个接口”，而是“怎么让金融数据在多个系统之间流动时不丢、不乱、不慢”。

我先把结论放在前面：金融数据服务的核心难点从来不是技术选型，而是数据一致性和边界划分。技术栈可以换，但这两件事没想清楚，后面全是坑。

1.2 为什么金融场景不能照搬普通CRUD

普通业务系统的数据服务，读写分离加缓存基本就能撑住。金融场景不一样，它有四个硬性约束：

• 精度不能丢：金额计算不能用浮点数，这是铁律。我见过有人用float存余额，对账时差了0.01，排查了整整两天。
• 时序不能乱：交易流水、行情快照都有严格的时间顺序，乱序写入会导致下游计算出完全错误的结果。
• 幂等必须做：网络抖动导致的重试在金融场景里是常态，同一个请求进来两次，不能扣两次钱。
• 审计要留痕：每一笔数据变更都要能追溯到“谁、什么时候、改了什么、为什么改”。

这四点决定了金融数据服务的架构跟普通服务有本质区别。下面我按实际搭建顺序，把每个环节拆开讲。

1.3 整体架构的分层逻辑

我习惯把这类服务分成四层，从下往上依次是：

分层的意义在于：任何一层的改动不会波及全局。比如行情源从A换成B，只需要改接入层，核心服务层完全无感。这个边界如果一开始不划清楚，后面每加一个数据源都要动全身。

注意：分层不是越多越好。我见过把接入层拆成五个微服务的，结果一个简单的行情推送要经过七次网络跳转，延迟直接飙到秒级。金融场景对延迟敏感，能本地完成的计算不要跨网络。

2. 核心模块的细节拆解与实操要点

2.1 金额与精度的处理方案

金额处理是金融数据服务的地基。我的做法是：所有金额在系统内部统一用最小货币单位的整数表示。比如人民币用“分”，美元用“美分”。这样加减法永远是整数运算，不存在精度问题。

展示的时候再除以100转成元。这个转换只发生在最外层，内部流转全程是整数。

# 金额存储与转换的推荐做法 class Money: def __init__(self, amount_in_cents: int, currency: str = "CNY"): # 内部始终用整数分存储 self._cents = amount_in_cents self._currency = currency @classmethod def from_yuan(cls, yuan_str: str): # 从字符串解析，避免浮点误差 from decimal import Decimal d = Decimal(yuan_str) return cls(int(d * 100)) def to_yuan(self) -> str: from decimal import Decimal return str(Decimal(self._cents) / 100) def __add__(self, other): assert self._currency == other._currency return Money(self._cents + other._cents, self._currency)

为什么要用字符串解析而不是直接传float？因为0.1 + 0.2在浮点里不等于0.3，这个误差在单笔交易里看不出来，但一天百万笔累加之后，对账必然出问题。

利率、汇率这类需要小数运算的场景，用Decimal类型，并且显式指定精度和舍入模式。Python的Decimal默认舍入是ROUND_HALF_EVEN，金融场景通常要求ROUND_HALF_UP，这个要手动设置。

2.2 数据接入层的设计要点

数据接入层要解决三个问题：连得上、收得全、分得清。

连得上是指跟各种数据源的对接。行情源可能是推送式的，账户数据可能是拉取式的，文件导入可能是批量式的。我的经验是统一抽象成一个“采集任务”的概念，每个任务有自己的调度策略和解析逻辑，但输出格式统一。

收得全是指不能丢数据。推送式数据源一定要有断线重连和补拉机制。我踩过的坑是：某次行情推送断了三分钟，重连后只收到了新数据，中间那段缺失导致回测结果完全失真。后来加了序列号校验，发现缺口就主动补拉。

分得清是指数据要带来源标记。同一个字段可能来自不同数据源，值还不一样。比如某只股票的价格，A源报10.01，B源报10.02。系统要能记录“这条数据来自哪里”，下游才能决定信谁。

# 采集任务的基础结构 class CollectTask: def __init__(self, source_name, fetch_interval, parser): self.source_name = source_name self.fetch_interval = fetch_interval self.parser = parser self.last_sequence = None def run(self): raw = self.fetch() # 序列号连续性校验 if not self.check_sequence(raw): self.backfill() parsed = self.parser(raw) # 打上来源标记 for record in parsed: record["_source"] = self.source_name record["_ingest_time"] = now_millis() return parsed

2.3 幂等设计的落地方法

幂等是金融服务的生命线。实现方式有很多种，我推荐的是基于业务唯一键的去重表。

具体做法：每笔写操作进来，先根据业务字段拼出一个唯一键，比如账户ID + 交易日期 + 流水号。然后去重表里查这个键存不存在，不存在就写入并记录，存在就直接返回上次的结果。

CREATE TABLE idempotent_record ( biz_key VARCHAR(128) PRIMARY KEY, result_json TEXT, created_at BIGINT, expire_at BIGINT );

这里有个细节：去重表要设过期时间。不能无限增长，一般保留7到30天足够覆盖所有重试窗口。清理任务每天跑一次，删掉过期的记录。

实操心得：去重表的写入和业务写入必须在同一个事务里。我见过分开写的，结果业务写成功、去重记录写失败，重试时又执行了一遍业务逻辑，直接造成重复扣款。这个坑代价很大。

2.4 接口层的鉴权与限流

对外提供的接口，鉴权用签名机制比token更合适。金融场景的调用方通常是系统对系统，没有用户交互，签名能同时解决身份认证和防篡改。

签名算法一般是：把请求参数按key排序，拼成字符串，加上时间戳和密钥，做HMAC-SHA256。服务端用同样的方式算一遍，比对结果。时间戳用来防重放，一般允许5分钟的偏差。

限流用令牌桶算法，按调用方维度限。不同客户给不同的配额，这个配额存在配置中心里，可以动态调整。我建议限流要做在网关层，不要做在业务代码里，否则每个接口都要写一遍限流逻辑，维护成本太高。

# 签名验证的核心逻辑 import hmac, hashlib, time def verify_signature(params: dict, secret: str, signature: str, timestamp: int): # 时间戳偏差检查，防重放 if abs(time.time() * 1000 - timestamp) > 5 * 60 * 1000: return False # 参数排序拼接 sorted_items = sorted(params.items()) raw = "&".join(f"{k}={v}" for k, v in sorted_items) raw += f"&timestamp={timestamp}" expected = hmac.new(secret.encode(), raw.encode(), hashlib.sha256).hexdigest() return hmac.compare_digest(expected, signature)

hmac.compare_digest是恒定时间比较，防止时序攻击。虽然金融场景里时序攻击不是主要威胁，但养成习惯没坏处。

3. 完整实操流程与关键环节实现

3.1 环境准备与依赖选型

先说存储选型，这是整个项目里最重要的决定之一。

为什么事务数据选PostgreSQL而不是MySQL？因为PG的NUMERIC类型在精度处理上更严谨，而且支持更丰富的约束和索引类型。当然MySQL也能用，但如果团队没有历史包袱，PG是更稳妥的选择。

时序数据用TimescaleDB而不是InfluxDB，主要考虑是运维统一。TimescaleDB是PG的扩展，同一套运维体系就能覆盖，不用额外维护一个数据库集群。对于中小规模的数据量，这个选择能省很多事。

依赖管理用poetry或者pip-tools，锁定版本号。金融系统最怕的就是“昨天还好好的，今天跑不起来了”，依赖版本必须锁死。

# 用poetry初始化项目 poetry init --name financial-services --python "^3.11" poetry add fastapi uvicorn sqlalchemy psycopg2-binary redis poetry add --dev pytest pytest-asyncio

3.2 数据库表结构设计

核心表我一般分这么几张：账户表、流水表、持仓表、行情表、配置表。

账户表存静态信息加当前余额，流水表存每一笔变动，持仓表存当前持仓快照。这里的关键设计是：余额不直接改，而是通过流水累加得出。账户表里的余额字段只是一个缓存，定期跟流水汇总结果对账。

-- 账户表 CREATE TABLE account ( id BIGSERIAL PRIMARY KEY, account_no VARCHAR(32) UNIQUE NOT NULL, currency VARCHAR(8) NOT NULL, balance_cents BIGINT NOT NULL DEFAULT 0, frozen_cents BIGINT NOT NULL DEFAULT 0, version INT NOT NULL DEFAULT 0, created_at BIGINT NOT NULL, updated_at BIGINT NOT NULL ); -- 流水表 CREATE TABLE transaction_log ( id BIGSERIAL PRIMARY KEY, account_no VARCHAR(32) NOT NULL, biz_type VARCHAR(32) NOT NULL, amount_cents BIGINT NOT NULL, balance_after_cents BIGINT NOT NULL, biz_key VARCHAR(128) UNIQUE NOT NULL, created_at BIGINT NOT NULL ); CREATE INDEX idx_tx_account_time ON transaction_log(account_no, created_at DESC);

version字段是乐观锁用的。并发更新余额时，UPDATE account SET balance_cents = ?, version = version + 1 WHERE id = ? AND version = ?，影响行数为0就说明有并发冲突，重试即可。

biz_key上的唯一索引是幂等的最后一道防线。即使应用层的去重逻辑出了问题，数据库层面也会拦住重复写入。

3.3 核心交易流程的实现

一笔交易从请求到落库，完整流程是这样的：

1. 接收请求，验证签名和时间戳
2. 检查幂等键，已存在则直接返回上次结果
3. 开启数据库事务
4. 查询账户，加行锁（SELECT ... FOR UPDATE）
5. 校验余额是否充足
6. 更新账户余额
7. 写入流水记录
8. 写入幂等记录
9. 提交事务
10. 返回结果

def execute_transaction(session, account_no, amount_cents, biz_key, biz_type): # 幂等检查 existing = session.query(IdempotentRecord).filter_by(biz_key=biz_key).first() if existing: return json.loads(existing.result_json) # 行锁查询账户 account = session.query(Account).filter_by( account_no=account_no ).with_for_update().one() if account.balance_cents + amount_cents < 0: raise InsufficientBalanceError() account.balance_cents += amount_cents account.version += 1 account.updated_at = now_millis() tx = TransactionLog( account_no=account_no, biz_type=biz_type, amount_cents=amount_cents, balance_after_cents=account.balance_cents, biz_key=biz_key, created_at=now_millis() ) session.add(tx) result = {"balance": account.balance_cents, "tx_id": tx.id} session.add(IdempotentRecord( biz_key=biz_key, result_json=json.dumps(result), created_at=now_millis(), expire_at=now_millis() + 7 * 24 * 3600 * 1000 )) return result

行锁的选择有讲究。FOR UPDATE会阻塞其他事务的写操作，但不会阻塞读。如果并发量很大，可以考虑用乐观锁替代，减少锁等待。具体选哪种，看你的读写比例。写多读少用悲观锁，读多写少用乐观锁。

3.4 对账机制的建立

对账是金融系统的安全网。我的做法是每天凌晨跑一次全量对账，逻辑是：

• 从流水表汇总每个账户的净变动
• 跟账户表的当前余额比对
• 不一致的记录写入差异表，人工介入

-- 对账查询 SELECT a.account_no, a.balance_cents AS recorded_balance, COALESCE(SUM(t.amount_cents), 0) AS calculated_balance FROM account a LEFT JOIN transaction_log t ON a.account_no = t.account_no GROUP BY a.account_no, a.balance_cents HAVING a.balance_cents != COALESCE(SUM(t.amount_cents), 0);

这个查询在数据量大时会很慢，所以流水表的时间索引很重要。如果数据量到了亿级，就要考虑按时间分区，只对账最近的数据，历史数据归档后单独处理。

实操心得：对账不要等到出问题才做。我建议从项目第一天就建立对账机制，哪怕数据量很小。因为对账逻辑本身也需要调试，等数据量大了再补，排查成本会高很多。

4. 常见问题与排查技巧实录

4.1 数据不一致的排查思路

数据不一致是金融系统最头疼的问题。排查时按这个顺序走：

1. 确认差异范围：是个别账户还是大面积？是某个时间段还是全量？
2. 检查幂等记录：有没有重复的biz_key？有的话说明幂等逻辑失效了。
3. 检查事务边界：业务写入和幂等记录是不是在同一个事务里？
4. 检查并发控制：有没有可能出现两个事务同时读到旧余额？
5. 检查数据源：接入层有没有丢数据或重复数据？

我遇到过一次典型问题：某账户余额比流水汇总多了100块。排查发现是接入层重连时补拉了数据，但补拉的数据没有做幂等，导致同一笔入账被记了两次。修复方式是在接入层也加幂等键，用数据源的原生ID做去重。

4.2 性能问题的定位方法

金融数据服务的性能瓶颈通常出现在三个地方：数据库锁等待、网络延迟、序列化开销。

锁等待用pg_stat_activity查，看有没有长时间运行的事务。网络延迟用链路追踪工具定位，看是哪一跳慢。序列化开销用压测工具测，对比不同序列化方案的吞吐量。

4.3 幂等失效的几种场景

幂等失效我总结了几种常见场景：

• 幂等键设计不合理：用了自增ID做幂等键，每次请求ID都不同，等于没做幂等。
• 去重表过期时间太短：重试窗口是24小时，去重记录只保留1小时，后面的重试就拦不住了。
• 事务未覆盖：业务写入和幂等记录分开提交，中间失败导致状态不一致。
• 多实例并发：两个实例同时收到同一请求，都查了去重表发现不存在，然后都执行了写入。

最后一种情况需要用数据库唯一索引兜底，或者用分布式锁。我倾向于唯一索引，简单可靠，不引入额外组件。

4.4 上线前的检查清单

每次上线前我都会过一遍这个清单：

• [ ] 金额字段是否全部用整数分存储
• [ ] 所有写接口是否都有幂等键
• [ ] 幂等记录是否和业务写入在同一事务
• [ ] 账户更新是否用了行锁或乐观锁
• [ ] 对账任务是否已配置并测试通过
• [ ] 接口签名和时间戳校验是否生效
• [ ] 限流配置是否按调用方维度设置
• [ ] 日志是否包含足够的排查信息
• [ ] 数据库连接池大小是否合理
• [ ] 缓存过期策略是否配置

这个清单看起来简单，但每次上线前认真过一遍，能避免80%的线上问题。我见过太多团队因为赶进度跳过检查，结果上线当晚就出故障。

4.5 日志与审计的实操建议

日志要分两类：技术日志和业务审计日志。

技术日志记录请求链路、耗时、异常堆栈，用于排查技术问题。业务审计日志记录每一笔数据变更的前后值、操作人、操作时间、操作原因，用于合规审计。

审计日志我建议单独存一张表，不要跟技术日志混在一起。字段至少包含：表名、记录ID、变更前值、变更后值、操作人、操作时间、操作来源IP、请求ID。

CREATE TABLE audit_log ( id BIGSERIAL PRIMARY KEY, table_name VARCHAR(64) NOT NULL, record_id BIGINT NOT NULL, before_value JSONB, after_value JSONB, operator VARCHAR(64), source_ip VARCHAR(64), request_id VARCHAR(64), created_at BIGINT NOT NULL ); CREATE INDEX idx_audit_record ON audit_log(table_name, record_id, created_at DESC);

审计日志只增不改不删，保留期限根据合规要求定，一般至少保留5年。存储成本会比较高，可以考虑冷热分离，最近3个月的放热存储，更早的归档到对象存储。

5. 扩展方向与个人经验补充

5.1 从单库到分库的演进路径

项目初期单库完全够用，但数据量到千万级以后，查询会明显变慢。这时候要考虑分库分表。

分片键的选择很关键。金融场景一般按账户ID分片，因为大部分查询都是围绕账户进行的。分片算法用一致性哈希，方便扩容。

分片之后，跨片查询会变复杂。比如要查某个时间段所有账户的流水汇总，就需要并行查询所有分片再合并。这个逻辑最好封装在数据访问层，业务代码不用感知分片细节。

5.2 实时计算能力的引入

如果业务需要实时风控或实时指标，就要引入流式计算。我一般用Kafka做数据管道，Flink做计算引擎。

流程是：业务写入数据库的同时，发一条消息到Kafka。Flink消费消息，做实时聚合和规则匹配，结果写回Redis或数据库。这样业务写入和实时计算解耦，互不影响。

注意：流式计算的结果是近似实时的，不能作为最终对账依据。对账还是要走离线全量计算。两者定位不同，不要混用。

5.3 我踩过的几个印象深刻的坑

第一个坑是时区问题。服务器用UTC，业务用北京时间，结果某天的交易被算到了前一天。后来统一规定：存储用UTC毫秒时间戳，展示时再转时区。

第二个坑是批量导入的幂等。单笔交易的幂等好做，批量导入的幂等就复杂了。一批1000条，第500条失败，重试时前499条不能重复写入。解决方案是给每条记录单独设幂等键，而不是给整批设一个。

第三个坑是缓存与数据库不一致。更新数据库后删缓存，但删缓存失败，导致缓存里是旧数据。后来改成先删缓存再更新数据库，并且加了一个短过期时间兜底。极端情况下最多脏读几秒钟，可以接受。

5.4 给后来者的几点实在建议

如果你正准备做类似的项目，我的建议是：先把幂等和对账做扎实，再考虑性能优化。我见过太多团队一上来就追求高并发，结果数据一致性一塌糊涂，后面花几倍的时间去补。

另外，不要过度设计。项目初期用单体架构完全没问题，等业务量真的上来了再拆。过早拆分成微服务，只会增加调试和部署的复杂度。

最后，测试要覆盖边界情况：余额刚好等于交易金额、并发同时扣款、网络超时后重试、数据源返回空值。这些边界情况在金融场景里都是真实会发生的，测试时覆盖到了，线上就少出事。

数据库连接池的大小设置也有讲究。经验公式是：连接数 = CPU核数 * 2 + 磁盘数。但金融场景事务较长，可以适当放大。我一般从20开始，根据监控数据调整。连接池太小会导致请求排队，太大又会拖垮数据库，需要实际压测找到平衡点。