先说结论如果你有超过三张银行卡、两个理财App、一个公积金账户还想知道自己每个月的钱到底花在了哪里那市面上绝大多数记账软件都会让你越记越糊涂。我自己折腾了差不多三个月最后决定绕过所有现成的记账应用自己动手搭了一个私有化的财务管理服务层——这就是我称之为“financial-services”的小项目。它不是又一个客户管理后台而是一套以你个人真实资金流水为核心、以统一查询和自动分类为骨架的轻量级服务。这篇文章就是这套服务从零到能用的完整复盘包含数据建模、多格式流水解析、分类规则引擎、部署和安全策略以及我踩进去又爬出来的坑。1. 为什么我需要一个“自建金融服务层”1.1 记账软件的取舍困境你可能会问市面上不是有那么多记账软件吗随手记、钱迹、鲨鱼记账甚至Excel都有人用为什么还要自己写我的真实感受是它们大多解决的是“记录”问题而不是“分析”问题。手动记账坚持两周就是胜利坚持三个月基本靠意志力硬扛。而自动同步的方案不是要你授权网银密码这放在任何安全评估里都是雷就是导出的账单格式混乱到需要手工二次清洗。到最后软件里躺着的数据和你真实的银行流水对不上分析出来的消费结构自然等于胡说。另一个问题是记账软件是一个封闭的“黑盒”。你辛辛苦苦记了一年想导出数据做一次自定义的趋势分析、想按自己的逻辑给某笔消费打标签发现自己被困在对方的字段设计里。我想做的事情其实很简单把折腾过的每一笔流水统一存在一个自己完全控制的数据库里想怎么查就怎么查想怎么算就怎么算。这就是“financial-services”项目的出发点——它本质上是一套个人财务数据聚合与查询服务从各种数据源把流水抓过来、清洗干净、存起来然后通过一套统一的接口对外提供查询能力。1.2 这个项目到底能做什么这套服务最核心的能力可以拆成四条线多源流水接入支持不同银行、不同平台导出的流水文件CSV、OFX、PDF等统一解析成标准结构。统一数据模型所有账户、交易流水、余额快照都按照一套字段定义存储查询时不用关心数据来自哪个机构。自动分类引擎基于标题关键词和规则优先级给每笔交易打上“餐饮”“交通”“工资”“理财”等分类标签并且规则可随时调整。统一查询接口对外提供REST API按时间范围、账户、分类、金额区间等条件查询流水和汇总统计方便后续接自己的可视化面板或者通知机器人。适合它的人我总结了一下大概是这几类有一定动手能力、有多个银行账户需要统一管理、不满足于现有记账软件的分类逻辑、想把财务数据掌握在自己手里的个人开发者或小型团队。1.3 技术选型的思路整个项目我最终选型如下模块选型理由运行时Python 3.11 FastAPI上手快、生态成熟处理CSV/PDF/OFX解析的库很全数据库SQLite SQLAlchemy单机部署够用零运维成本数据量上来后可平滑迁PostgreSQL任务调度APScheduler支持定时扫描导入文件夹不用额外引消息队列文件解析pandas BeautifulSoup pdfplumber分别应对表格类、结构化XML类、PDF类流水文件部署docker-compose一条命令拉起整个服务数据卷持久化方便备份这套选型不算惊艳但非常稳且每一层解决一个问题。接下来我把每一块的关键实现拆开讲。2. 服务端底层设计账号、流水与聚合的核心模型2.1 数据模型的取舍逻辑数据模型是整个服务的地基。在动手建表之前我先明确了一个原则这个库要服务的是“查询分析”不是“记账录入”。所以整个结构设计得相对扁平核心就是三张表账户表、流水表、分类规则表。账户表保存的是你有哪些资金户口可以理解为一个总的“账本目录”。“账户”不仅仅指银行卡也可以是支付宝余额、微信钱包、公积金、甚至是某个投资平台里面的现金账户。这样分开存的好处是每笔交易都能追溯到具体的来源汇总分析时又能按账户类型自由聚合。流水表是绝对的核心。每一个字段都是熬过几轮需求反推之后的产物我贴一下建表的核心语句用SQLAlchemy描述方便迁移class Transaction(Base): __tablename__ transactions id Column(String(64), primary_keyTrue) # 幂等键见下文 account_id Column(String(32), ForeignKey(accounts.id), indexTrue) transaction_date Column(Date, nullableFalse, indexTrue) posting_date Column(Date, nullableTrue) # 实际入账日和交易日常有1-2天差异 description Column(String(255), nullableFalse) # 交易描述/备注清洗后的文本 raw_description Column(Text, nullableTrue) # 原始描述留底排查用 amount Column(BigInteger, nullableFalse) # 金额统一存为“分”正负代表收支 currency Column(String(8), defaultCNY) category Column(String(32), indexTrue) # 自动分类结果 source_file Column(String(255)) # 来自哪个文件便于回溯 imported_at Column(DateTime, defaultfunc.now())字段设计上有几个细节是反复敲定过的值得记下来金额用BigInteger存“分”浮点数运算的精度问题在财务场景下是不可接受的Python的float在计算大额金额时会出现尾差。直接用整数“分”存储所有计算都是整数运算从源头杜绝精度损失。展示时再除以100格式化。一定要保留raw_description解析规则和分类规则经常需要调整保留原始描述可以在规则变更后重新跑分类而不用重新导入数据。id字段是“幂等键”的产物同一笔交易如果被重复导入比如这个月导了一次下个月又导了一次就会产生重复流水统计直接翻倍。我用“账户ID 交易日 金额 原始描述哈希”拼出一个稳定的id重复导入时直接跳过这是所有数据接入逻辑的底线。2.2 账户与流水的关系处理账户表相对简单我给每个账户设计了几个必要的属性账户名称、账户类型银行卡/电子钱包/信用卡/投资账户/其他、所属机构、币种、初始余额、是否启用。class Account(Base): __tablename__ accounts id Column(String(32), primary_keyTrue) # 业务ID如 cmb_debit_card_1234 name Column(String(64), nullableFalse) account_type Column(String(16), nullableFalse) # debit/credit/e_wallet/invest/other institution Column(String(64)) # 招行、支付宝、微信… currency Column(String(8), defaultCNY) initial_balance Column(BigInteger, default0) # 单位分 extra_meta Column(JSON, default{}) # 存放卡号掩码、尾号等 is_active Column(Boolean, defaultTrue)这里有个实际建议账户的id千万不要用自增整数建议使用“有业务含义的字符串”。因为后续接入新的数据源、做API查询的时候我们经常需要在请求里直接指定账户字符串ID远比数字ID可读性强。而且账户可能被删除、重建字符串ID能保证外部引用不轻易失效。余额的处理我单独说一下。交易流水是“流”的概念余额是“快照”的概念。流水表本身不存余额因为手工导入时某笔流水的余额往往是导出文件里给的“交易后余额”并不一定可追溯。我在账户总览里算的余额是初始余额 该账户所有流水金额之和这样每一笔数据的变化都是可推导的不会有余额信息丢失。如果想追踪历史余额趋势未来可以加一张余额快照表定时记录每日余额这是后话。3. 数据接入的血泪实战解析与标准化3.1 为什么解析是最脏最累的活数据模型建得再好接不进来数据就是白搭。我最初以为解析流水就是“读Excel文件填到数据库”实际上这是整条链路上最折磨人的环节。不同金融机构导出的流水文件风格差异极大基本上可以分为三类格式特征解析难度CSV/Excel带标准表头列名明确但是各家的列名完全不同中需要做列名映射OFX/QIF银行标准化交换格式半结构化XML字段固定低最规范的一种PDF银行网页导出的报表纯展示格式可能分页、多栏高需要版面嗅探我针对每一类都写了解析器但核心目标是统一的把每一条交易记录“翻译”成上面那张Transaction表里定义的字段。这里的经验是先定义目标结构再针对每个数据源写一个字段映射函数绝不边解析边设计字段。3.2 CSV类流水的列名映射实战CSV是我用得最多的格式。招商银行、工商银行、支付宝、微信支付都能导出CSV但各家字段名几乎完全不一样招行导出交易日期, 收支, 交易金额, 交易类型, 交易描述, 余额支付宝导出交易时间, 交易分类, 交易对方, 商品说明, 收/支, 金额, 收付款方式微信支付导出交易时间, 交易类型, 交易对方, 商品, 收/支, 金额, 支付方式表面上是同一个格式实际上每个字段名都要做一次映射。我的做法是定义一张映射配置表每次接入新银行最长半小时就能完成接入CSV_MAPPINGS { cmb: { date: 交易日期, amount_raw: 交易金额, income_expense: 收支, description: 交易描述, account_type_hint: 交易类型, }, alipay: { date: 交易时间, amount_raw: 金额, income_expense: 收/支, description: 商品说明, counterparty: 交易对方, }, wechat: { date: 交易时间, amount_raw: 金额, income_expense: 收/支, description: 商品, counterparty: 交易对方, }, }读取的时候用 pandas 加载然后按映射重命名列再做统一的清洗步骤。清洗步骤里最容易出问题的有两个日期格式和收支判定。日期格式各家都不一样有2024/01/15、有2024-01-15 10:23:45、有01/15/2024。我在解析层做了统一处理不猜格式而是让配置里显式指定DATE_FORMATS { cmb: %Y%m%d, alipay: %Y-%m-%d %H:%M:%S, wechat: %Y-%m-%d %H:%M:%S, }收支判定的坑更多。有的银行CSV直接把金额写成负数代表支出有的给你单独一列“收支”写“支出”两个字还有的既不标正负也没收支列需要你根据“交易类型”猜——比如“消费”必然是支出方向。我在代码里封装了一个normalize_amount函数逻辑很简单先看有无方向列有则按方向列结合金额绝对值确定正负没有方向列则按金额本身的符号处理两者都没有就抛异常提示人工介入绝不默认。3.3 OFX格式的解析细节OFXOpen Financial Exchange是银行标准化交换格式在境外银行里非常常见国内部分银行也支持导出。它的结构是基于SGML/XML的STMTTRN TRNTYPEDEBIT/TRNTYPE DTPOSTED20240115000000/DTPOSTED TRNAMT-128.50/TRNAMT FITID20240115001/FITID NAMESUPERMARKET/NAME /STMTTRN这种格式是最省心的因为字段定义是标准化的。直接用BeautifulSoup解析XML节点即可from bs4 import BeautifulSoup def parse_ofx(content: str): soup BeautifulSoup(content, html.parser) transactions [] for txn in soup.find_all(stmttrn): fitid txn.fitid.text if txn.fitid else None trntype txn.trntype.text if txn.trntype else dtposted txn.dtposted.text if txn.dtposted else trnamt txn.trnamt.text if txn.trnamt else 0 name txn.name.text if txn.name else txn.memo.text if txn.memo else direction 1 amount_float float(trnamt) if amount_float 0: direction -1 transactions.append({ amount: int(round(abs(amount_float) * 100)) * direction, description: name.strip(), transaction_date: parse_ofx_date(dtposted), source_id: fitid, }) return transactionsOFX的时间戳格式是YYYYMMDDHHMMSS要专门写一个解析函数转成Python的date对象。这里有个令我印象深刻的细节OFX文件里同一笔交易在重复导出时FITID保持不变所以它天生适合做幂等键——这比我之前用“日期金额描述”拼id的方案稳得多。如果碰到的格式里自带交易唯一标识优先用它作为幂等键。3.4 PDF文档解析最后的倔强有一些银行渠道尤其是信用卡账单只提供PDF解析起来最麻烦。我的思路是先用pdfplumber把整个PDF的内容提取成文本块然后按页眉/页脚结构定位每一笔交易的起止行。真实账单的文本长这样2024-02-15 商户名称 交易描述 1,280.00 人民币因为PDF是排版产物账单行之间没有结构化的分隔。我实际的做法是提取每一页的文字后先找到信用卡账单里典型的表头行比如“交易日期”“商户名称”“交易金额”然后手工确认每一行的列位置按坐标切分import pdfplumber def extract_table_from_pdf(pdf_path: str, columns_x: dict): rows [] with pdfplumber.open(pdf_path) as pdf: for page in pdf.pages: text page.extract_text() lines text.splitlines() for line in lines: if is_transaction_line(line): # 按预定义的关键词位置截取 date_raw extract_by_x(line, columns_x[date]) desc extract_by_x(line, columns_x[description]) amount_raw extract_by_x(line, columns_x[amount]) rows.append({date: date_raw, description: desc, amount: amount_raw}) return rows这种解析非常脆弱——不同银行的账单排版不同同一家银行改了排版也会失效。我的建议是把PDF解析当作最后手段优先看银行App能不能导出Excel/OFX。如果必须走PDF那就要保证原始PDF文件归档完整一旦解析规则调整还能重新跑。4. 对外能力封装统一查询服务和分类规则引擎4.1 交易自动分类的思路与规则引擎设计数据接入跑通之后下一步是让数据“变聪明”。如果你只是为了存流水那前面建表就够了。但我做这个项目的直接动力是月末想一眼看到钱都花哪了。这时候分类准确率高低直接决定这套服务有没有用。最初的分类方案是直接用交易描述里的关键词做匹配描述里带“美团”“饿了么”“餐厅”就归为“餐饮”带“滴滴”“地铁”归为“交通”。跑下来一周后发现准确率大概在70%左右大量交易被漏判或错判。原因很直接微信支付商户名称五花八门“美团平台商户”和“美团外卖”其实都是餐饮但关键词“美食”可能同时命中外卖和门票。同一类消费在不同平台的描述差异巨大比如“12306”和“南方航空”都是交通但关键词“12306”和“航空”完全没关联。一条描述可能命中多个分类关键词没有一个明确的优先级分类结果随机。于是我改成了规则优先级引擎。每一条规则由三部分组成优先级、关键词列表、分类结果。导入流水时按优先级从高到低匹配关键词命中即归类不再继续匹配低优先级规则。CLASSIFICATION_RULES [ {priority: 100, keywords: [房贷, 公积金, 租金, 房租], category: 住房}, {priority: 90, keywords: [滴滴, 高德打车, 地铁, 公交], category: 交通}, {priority: 90, keywords: [美团, 饿了么, 肯德基, 麦当劳, 星巴克], category: 餐饮}, {priority: 80, keywords: [淘宝, 京东, 拼多多], category: 购物}, {priority: 70, keywords: [工资, 奖金, 报销, 退款], category: 收入}, {priority: 0, keywords: [], category: 未分类}, ]有了优先级就能处理“美团外卖”同时含“美团”和“外卖”的情况因为“餐饮”规则放在了“购物”之前。但这里的代价是规则权重需要持续迭代。我建议把规则做成数据库配置表而不是硬编码在代码里这样调整分类逻辑不用重新发布服务。一开始就硬编码是图省事后面改规则改到怀疑人生。4.2 REST API的设计与使用示例服务对外暴露的是一组REST API核心端点不多但每个都经过实用性检验端点功能GET /api/v1/accounts获取所有账户及当前余额GET /api/v1/transactions按条件查询流水支持时间范围、账户、分类、金额区间GET /api/v1/transactions/stats按分类/时间聚合统计收入支出POST /api/v1/import/{account_id}上传流水文件触发解析导入GET /api/v1/rules查看全部分类规则以最常用的统计接口为例我设计了这样的查询参数curl -X GET http://localhost:8000/api/v1/transactions/stats?start2024-01-01end2024-03-31group_bycategory返回结果大致是{ total_income: 4820000, total_expense: 3120000, by_category: { 住房: 880000, 餐饮: 765000, 交通: 230000, 购物: 455000, 未分类: 790000 } }金额单位仍然是“分”前端展示时再除以100。为什么不在API层直接转成元因为单位转换属于展示逻辑如果将来加货币换算、加多币种底层保持整数“分”才是最小惊讶原则。API层保持原始单位让消费方自己决定怎么展示。统计接口的实现也不复杂核心就是SQL里的条件组合加GROUP BYapp.get(/api/v1/transactions/stats) def transaction_stats(start: date, end: date, group_by: str category): query db.query(Transaction) query query.filter(Transaction.transaction_date start) query query.filter(Transaction.transaction_date end) if group_by category: rows query.with_entities( Transaction.category, func.sum(Transaction.amount).label(total) ).group_by(Transaction.category).all() return {rows: [{category: r[0], total: r[1]} for r in rows]}4.3 导入接口的幂等设计与文件归档导入接口是我做得最谨慎的地方因为数据源文件可能被重复上传。流程是这样的上传文件到服务端临时目录按账户对应的解析器解析出标准交易列表对每一笔交易计算幂等键优先用源文件里的交易ID否则用“账户日期金额描述哈希”插入数据库时用INSERT OR IGNORE重复数据直接跳过解析完成后把源文件移动到归档目录按日期归档返回导入结果总记录数、新增数、跳过数。跳过数这个指标特别重要。如果某次解析规则有问题导致大量记录被跳过你会立刻在返回结果里发现而不是等到月末统计时才察觉数据不对。文件归档目录结构大概是这样data/imports/2024/02/15/cmb_20240215.csv data/imports/2024/02/15/alipay_20240215.csv每个文件都留底一旦后续发现某笔交易解析错了可以直接翻原始文件核对。数据溯源能力是一切分析可信度的前提。5. 部署、安全与日常维护的取舍5.1 数据库安全与备份策略个人的财务数据比代码还要敏感存储安全这块我不敢省。数据落库之后真正需要重点保护的是交易描述和账户信息。我的方案是对账户表中的敏感字段卡号、手机号等做加密存储对流水表的核心字段默认不落明文。加密用cryptography库的Fernet对称加密密钥文件单独存放不放在代码仓库里。from cryptography.fernet import Fernet # 生成一次密钥保存到环境变量或本地密钥文件 # key Fernet.generate_key() def encrypt_text(plaintext: str) - str: f Fernet(os.environ[FINANCE_SECRET_KEY].encode()) return f.encrypt(plaintext.encode()).decode() def decrypt_text(ciphertext: str) - str: f Fernet(os.environ[FINANCE_SECRET_KEY].encode()) return f.decrypt(ciphertext.encode()).decode()生产环境里密钥直接放进环境变量docker-compose里用secrets挂载文件避免密钥出现在进程列表中。备份方案相对朴素但足够可靠每天晚上对SQLite数据库文件做一次快照复制再加上每周一次把整个data目录打包上传到自己的对象存储或NAS。SQLite单文件备份非常简单sqlite3 .backup命令或者直接用文件复制都行关键是备份要带着导入的原始流水文件一起这样即使数据库损坏了也能靠原始文件重建。5.2 docker-compose 部署配置我部署在一台内网服务器上用docker-compose管理。整个服务只有一个容器没有外部依赖部署简单得有些低调version: 3.8 services: financial-services: build: . container_name: financial-services restart: unless-stopped ports: - 8000:8000 volumes: - ./data:/app/data - ./secrets:/app/secrets:ro environment: - FINANCE_SECRET_KEY_FILE/app/secrets/secret.key - DATABASE_PATH/app/data/finance.db - IMPORT_DIR/app/data/imports logging: driver: json-file options: max-size: 10m max-file: 3这里有一个特别容易翻车的点不要在代码里写死数据库路径。用环境变量控制DATABASE_PATH和IMPORT_DIR这样本地调试和服务器部署走同一套代码避免“本地好好的部署就报错”的惨剧。5.3 数据同步的自动化与异常告警流水导入可以手动做但长期用下来还是要自动化。APScheduler在服务启动时注册两个定时任务每小时扫描一遍imports目录下的待处理文件目录把新投放的文件自动导入每天凌晨2点做一次全量分类规则的重新匹配因为规则可能调整过历史流水的分类要重算。再配合一个极简的告警机制导入失败或连续N次同步为空时通过webhook往自己的即时通讯工具推一条消息。这里我不展开具体用什么IM工具只是强烈建议服务必须能在无人值守时给出手腕而不是安静地坏着。我自己经历过连续一周定时任务静默失败、周末打开面板发现数据一片空白的惨况。6. 踩坑复盘与后续想做的事6.1 四大高频坑按痛苦程度排序第一坑金额精度被浮点数吃了。早期用float存金额导入1000笔流水后汇总发现总额比银行账单少了0.0000001元。虽然显示出来无感但Excel对账时会因为浮点余差导致“账平不了”。改成整数“分”后所有金额字段全部用整数再也没有精度问题。第二坑CSV的编码花样太多。有些银行导出的CSV是GBK编码pandas默认用UTF-8读就乱码。后来我在解析器里加了编码自动探测逐字节尝试UTF-8、GBK、GB18030并把解析出的source_file原始编码记到日志里方便后续追溯。第三坑重复导入防不胜防。同一个银行App第一次导出的日期范围和第二次导出的范围重叠一部分重复流水的量看着不大但汇总统计一旦出错你根本不知道错在哪一笔。幂等键设计必须是一等公民不是事后补救方案。第四坑时区与夏令时对日切的影响。信用卡账单的交易日和入账日可能跨天如果按北京时间凌晨0点切日有些发生在23点59分的交易会被记到第二天。我的处理是在账户配置里增加一个“日切时间”字段默认设为凌晨3点模拟银行的实际日切规则。6.2 对项目未来的扩展想法这套服务跑了大半年目前数据量在2万笔左右SQLite查询和统计都是毫秒级没有性能压力。后续如果想继续折腾我大致有三个方向增加自然语言解析分类。现在规则引擎对“商户名称”之外的描述信息利用不足如果引入简单的NLP比如把交易对方的语义向量化再聚类能进一步提高分类覆盖率。增加预算预警。按分类设置月度预算当某类支出超过阈值时自动提醒这本质上只是对统计接口再加一层定时逻辑。接入更标准的金融数据源。比如部分银行开放了开发者API如果证书和网络条件允许就能彻底摆脱“导出-导入”的模式实现准实时同步。如果你也想做类似的东西我的建议是先别急着把架构设计得很大老老实实从一个银行账户的CSV导入开始搭好流水表和分类规则的骨架再把第二个、第三个账户接进来。你会发现数据进来之后需求会自己慢慢长出来——这时候再回头调整规则和模型就有方向了。