金融知识图谱构建实战：Neo4j+Python+Cypher完整指南

简介一份面向金融领域的知识图谱构建项目源码包基于Neo4j图数据库、Python与Cypher查询语言完成。项目代码完整、结构清晰包含从数据采集到知识存储的完整链路适合高校计算机、人工智能、金融科技等相关专业学生用于期末大作业、课程设计或毕业设计也适合想学习知识图谱与自然语言处理结合的开发者参考。资源共105个文件包含7个Python脚本、7个Jupyter Notebook、多个LSTM-CRF命名实体识别模型权重文件、股票新闻CSV数据、Neo4j配置、PDF/PPT说明文档及24张运行效果截图等整体约41.3MB便于直接复现与二次改造。目前已有82人学习下载。包内提供从数据爬取Scrapy、实体识别LSTMCRF到图数据构建与查询的完整流程讲解并附带模型训练产物、checkpoint及目录说明可按文档逐步操作快速跑通金融知识图谱构建链路同时预留扩展空间可在此基础上增加实体关系、丰富可视化交互用于毕业设计或项目展示。1. 金融知识图谱期末大作业为什么 Neo4jPythonCypher 这套组合值得做期末大作业里放着金融知识图谱构建项目技术栈是 Neo4j、Python 和 Cypher还附带源码工程时很多人第一反应是“又是数据库课设”。我反而觉得这是最有性价比的一类选题金融数据天生就是图结构公司、股东、行业、证券之间的关系一旦超过两层关系型数据库的 SQL 就开始绕而知识图谱能把“谁间接控制谁”这类问题变成一条 Cypher 路径查询。这套源码工程的核心就三件事把金融实体建模成节点和关系用 Python 清洗数据并写进 Neo4j最后用 Cypher 做穿透查询验证构建结果。适合想认真把图数据库跑通、并让期末大作业拿得出手的同学。2. 构建流程的第一道工序实体、关系与唯一约束怎么设计拿到一个源码工程先不要急着装环境。很多期末大作业的演示环节出问题不是因为代码跑不通而是图模型本身设计得含糊。金融知识图谱不是把 Excel 表变成一堆点和线就完事它要能回答具体问题比如“某自然人通过哪些中间公司间接控制目标公司”“两家上市公司之间是否存在循环持股”。这些问题决定了实体和关系怎么定。2.1 先画本体再写代码实体、关系、属性怎么定常见做法是先定义四类实体公司、人员、行业、证券。公司是核心节点人员包括高管、股东、法人行业用于聚合统计证券用于对接行情或代码。实体不必贪多期末大作业里四类已经完全够用。关系按照业务语义拆我一般会先列出金融场景里最常问的几类问题再反推关系持股对应holds_share任职对应serves_as投资对应invests_in公司归属行业对应belongs_to。关系要带方向比如(a:Company)-[:holds_share]-(b:Company)表示 a 持有 b 的股份方向一旦统一后面 Cypher 写起来就不会混淆。属性不要一股脑全塞在节点上。像持股比例、投资金额、任职起止日期这些描述的是“这段关系”而不是“这个公司”应该放在关系上(p:Person)-[r:serves_as {start_date: 2020-01-01, end_date: 2023-12-31}]-(c:Company)这样设计的好处是后续做时间维度分析时可以直接查r.start_date不用回源表重新解析。金融场景里关系属性尤其重要因为股权和任职都有时效性一个没有时间戳的holds_share关系在风控场景里几乎没有说服力。2.2 业务主键、唯一约束与标签命名图谱的数据卫生知识图谱构建过程中最容易犯的错误是拿节点的“名称”当“主键”。公司叫“华信投资有限公司”可能在好几个省份都有注册自然人同名更是常见。如果直接用MERGE (n:Company {name: row.name})数据一进去就把两家不同公司合并成一个节点后面所有穿透查询都会给出错误路径。正确的做法是给每类实体分配一个业务主键。公司用统一社会信用代码或公司代码人员用脱敏后的身份证号证券用证券代码。这些主键要在 Neo4j 里建唯一约束从数据库层面兜底CREATE CONSTRAINT company_code_unique FOR (n:Company) REQUIRE n.company_code IS UNIQUENeo4j 5.x 用的是REQUIRE语法老版本 4.x 是ASSERT如果你的源码工程是从旧教程里扒的注意替换。人员表同理主键建议用person_code不要把姓名作为唯一依据。属性命名也要提前统一。我用小写下划线风格避免触碰 Cypher 保留字。name、code、start_date这类是安全的但如果你非要用type、order这种字段名查询时必须加反引号比如n.order否则直接报语法错误。这个细节会在后面写入脚本时反复遇到前期定好规则能省很多事。2.3 关系型建模与图建模的差别为什么两层以上就翻车很多同学会问这些数据用 MySQL 也能存为什么非要用 Neo4j我举个实际场景。查“某公司直接股东”用 JOIN 两次就能搞定但查“某自然人对某上市公司的间接持股路径自然人 → 中间公司 → 目标公司”在关系型数据库里需要递归 CTE而且层数越深 SQL 越长。课程作业里演示“三层股权穿透”用 MySQL 写出来的 SQL 很难让老师一眼看懂业务逻辑。对比下来更直观查询需求关系型写法Neo4j Cypher 写法一层持股JOIN 两次MATCH (a)-[:holds_share]-(b)三层穿透递归 CTESQL 明显膨胀MATCH (a)-[:holds_share*1..3]-(b)找循环持股自连接加层级判断非常别扭MATCH p(a)-[:holds_share*1..5]-(a)Cypher 的路径匹配是原生图遍历写出来的语句和业务问题的描述几乎一一对应。这就是金融知识图谱选 Neo4j 的核心原因不是为了存数据而是为了在关系上做推导。源码工程里真正值钱的部分也是这一层——数据只是被塞进图里的原料查询能力才是交付物。3. 用 Python 连接 Neo4j驱动选型与 Cypher 参数化查询图模型设计好之后接下来就是让 Python 和 Neo4j 对话。这个章节要解决的是本地环境怎么搭、用哪个驱动、连接代码怎么写才不容易在期末答辩时翻车。3.1 环境准备本地 Neo4j 与 Python 驱动安装Neo4j 的安装方式主要有两种桌面版和社区版 zip 包。桌面版自带数据库管理界面适合课程演示社区版适合脚本自动化。无论哪种安装完启动后要确认默认端口7687Bolt 协议和7474HTTP 控制台是通的。浏览器能打开7474页面说明服务起来了。Python 这边只需要装官方驱动pip install neo4j如果源码工程里还依赖pandas做数据清洗一起装pip install pandas neo4j安装完成后建议把依赖写进requirements.txt方便对方复现环境。这里有个小提醒pip 安装的 neo4j 驱动是客户端不是数据库本身很多人装完驱动却连不上是因为本地根本没启动 Neo4j 服务先分清这两件事。3.2 官方驱动还是 py2neo期末作业的选型参考网上很多教程用的是py2neo因为它提供一个更“Pythonic”的图对象操作接口。但我在实际使用中更推荐官方neo4j-driver原因直接看对比对比维度官方 neo4j-driverpy2neo维护状态与 Neo4j 版本同步更新更新节奏偏慢易出现协议不兼容事务支持完整支持读写事务和回滚事务接口相对弱异步操作官方支持 Async API生态较弱学习成本需要掌握 Cypher上手容易但封装较重期末大作业的源码如果用的是老版本 py2neo在 Neo4j 4.4 以上版本很容易出现握手失败之类的报错。我一般会建议直接改为官方驱动代码改动量其实不大就是把session.run()的调用方式统一一下换来的是兼容性稳定。这不是玄学是驱动维护节奏决定的。3.3 连接与第一条 Cyphersession、事务与参数化查询用官方驱动建立连接的核心代码很短from neo4j import GraphDatabase uri bolt://localhost:7687 username neo4j password your_password driver GraphDatabase.driver(uri, auth(username, password)) def find_company_by_code(company_code: str): with driver.session() as session: result session.run( MATCH (n:Company) WHERE n.company_code $code RETURN n.name AS name, codecompany_code, ) for record in result: print(record[name]) find_company_by_code(000001)driver是整个进程里共享的单例对象不要每次查询都重新创建否则会有连接开销。session用上下文管理器管理用完自动关闭。真正的查询语句里用了$code占位符实际参数通过第二个参数传入这是 Cypher 的官方参数化写法。这里一定要养成参数化的习惯。字符串拼接虽然也能跑但一旦数据里出现引号、特殊字符要么查询直接报语法错误要么结果悄悄变空值排查起来非常费时间。Cypher 语法里像$这种占位符是内置支持的参数类型由驱动自动映射字符串、数字、列表都可以直接传。写入场景建议用事务函数官方推荐的方式是这样的def create_company(tx, code: str, name: str): tx.run( MERGE (c:Company {company_code: $code}) SET c.name $name, codecode, namename, ) with driver.session() as session: session.execute_write(create_company, 000001, 平安银行)execute_write会把事务的提交和回滚都处理好函数内部抛出异常时事务自动回滚不会留下半截数据。这种写法在批量导入时尤其重要因为一个批次里可能只需几条坏数据整个批次就会干净地回滚。4. 从源码工程到完整图谱清洗、批写入与执行顺序这一章讲的是“构建流程详解”的核心。拿到一套知识图谱源码正确的阅读顺序是先搞懂数据从哪来、脚本分几步、最终写进 Neo4j 的是什么。很多期末大作业的源码工程其实并不复杂但不按顺序跑或者漏掉中间一步后面查询全是空的。4.1 源码工程的目录结构与执行顺序我一般会把期末大作业的源码工程拆成五个部分方便答辩时讲清楚文件/目录职责产出data/存放原始 CSV 与清洗后清单clean_*.csvscripts/clean_data.py统一主键、去空、去重三张实体清单scripts/build_nodes.py写入 Company/Person 等节点图节点scripts/build_relations.py写入 holds_share/serves_as 等关系图关系scripts/check_graph.py统计节点边、抽样路径验证报告对应的执行顺序是固定的先清洗再建节点再建关系最后验证。如果你拿到别人的源码先按这个顺序跑一遍看中间每个脚本的日志输出是否正常。最容易犯的错是拿到源码就直接python build_relations.py结果关系里的端点节点根本不存在一个关系都写不进去。4.2 数据清洗从原始表到干净的实体清单清洗这一步决定了图谱的准确率。最常见的坑是股票代码被 Excel 或 pandas 读成浮点数000001变成1.0。所以读取时强制指定为字符串import pandas as pd df pd.read_csv(data/company.csv, dtypestr).fillna() df[company_code] df[company_code].str.strip().str.upper() df[name] df[name].str.strip() df df.drop_duplicates(subset[company_code]) df.to_csv(data/clean_company.csv, indexFalse)dtypestr让所有列以字符串读入避免代码、手机号这类前导零字段失真。strip()去掉空格upper()统一大写是为了让主键在后续 Cypher 匹配时一致——000001和 000001在 Neo4j 里是两个完全不同的值。drop_duplicates按主键去重保留第一条记录。这里有一个取舍去重时如果同一主键对应多行说明原始数据本身有问题。我一般会在去重前先groupby看重复次数确认是数据冗余还是主键设计错误而不是闷头去重。清洗脚本的产出不一定只有一份公司清单人员和关系表同样要做同样的处理只是主键字段换成person_code。4.3 批写入节点用 UNWIND 代替逐条 INSERT把清洗后的 DataFrame 写进 Neo4j最容易想到的是用 for 循环逐条执行 MERGE。数据量只有几百条时确实能跑但上万条时速度会急剧下降因为每条数据都要单独开启一次事务往返。正确的姿势是用 Cypher 的UNWIND把 Python 列表批量展开from neo4j import GraphDatabase driver GraphDatabase.driver(bolt://localhost:7687, auth(neo4j, password)) def write_companies(tx, batch): tx.run( UNWIND $batch AS row MERGE (c:Company {company_code: row.code}) SET c.name row.name, c.industry row.industry , batchbatch, ) clean_df pd.read_csv(data/clean_company.csv, dtypestr).fillna() records clean_df.to_dict(records) batch_size 500 with driver.session() as session: for i in range(0, len(records), batch_size): session.execute_write(write_companies, records[i:i batch_size])UNWIND $batch AS row把传入的字典列表拆成一行行数据row.code直接取字典的键。MERGE按业务主键company_code匹配存在就更新属性不存在就创建因此脚本可以重复执行不会产生重复节点。batch_size是事务大小我一般取 200 到 500 条太小事务次数多太大会导致单个事务过大内存和锁竞争都会增加。在写入节点之前先执行 2.2 里的唯一约束。先建约束再写入MERGE会走索引匹配速度从全表扫描变成索引查找几万条数据也能在秒级完成。如果顺序反了后续中途发现重复数据清理起来非常痛苦。4.4 批写入关系MERGE 的三种正确姿势节点写完关系写入的逻辑多了一步先定位两个端点。不能假设端点一定存在要在同一条 Cypher 里用MATCH找到两端再MERGE关系def write_shareholdings(tx, batch): tx.run( UNWIND $batch AS row MATCH (src:Company {company_code: row.from_code}) MATCH (dst:Company {company_code: row.to_code}) MERGE (src)-[r:holds_share]-(dst) SET r.ratio row.ratio, r.start_date row.start_date , batchbatch, )这里的MATCH是精确匹配如果from_code或to_code在节点表里不存在这一行会被直接跳过不会报错。这不是 bug是 Cypher 的默认行为。所以脚本跑完后一定要做校验对比原始关系表里有多少条记录实际写入的关系有多少条。差值就是脏数据。如果同一个(src, dst)对可能出现多条关系比如股东在不同时间多次增持MERGE会把它们合并成一条。此时如果想保留多次记录需要给关系加上唯一标识或把时间戳放进关系的主键里MERGE (src)-[r:holds_share {start_date: row.start_date}]-(dst)这是把关系属性纳入匹配条件的写法每次增持都会生成独立的关系。但要注意Neo4j 的关系不像节点那样有唯一约束语法重复执行会导致关系翻倍所以关系写入脚本最好设计成“只跑一次”而不是像节点那样随便重跑。这是源码工程里最容易埋雷的地方。完整跑一遍的命令通常是pip install -r requirements.txt python scripts/clean_data.py python scripts/build_nodes.py python scripts/build_relations.py python scripts/check_graph.py每一步都出了明确提示再往下走不要跳步。5. 构建流程必看的避坑清单Neo4j 与 Cypher 的 5 个翻车现场期末大作业答辩翻车通常不是知识没掌握而是掉进了 Neo4j 的边角坑。下面这几条是我带过的项目里出现频率最高的按“现象 → 原因 → 解决”的方式记录。5.1 查询卡死主键没建索引现象MATCH (c:Company {company_code: 000001})返回结果要十几秒甚至会卡住控制台。原因Neo4j 不会自动为节点的自定义属性建索引查询变成全库扫描。解决在写入前建好唯一约束它会自动附带索引已经写了一半数据也没关系中途补建索引也能加速后续查询。提示索引在建之前先用SHOW INDEXES确认现状避免重复建索引。5.2 MERGE 把两家不同公司合并成一个节点现象图谱里“华信投资”只要注册地不同应该是两个节点但查询出来只有一条且部分公司信息混在一起。原因写入时用了MERGE (n:Company {name: row.name})公司名称被当成了主键。解决一律用业务主键company_code做MERGE匹配条件公司名称只作为属性SET进去。如果节点已经合并错了需要先把错误节点拆开手工修改关系没有捷径。5.3 Python 拼接 Cypher 造成的引号地狱现象用 f-string 拼 Cypher 查询公司名里带一个单引号时查询报错或者返回空结果更严重的是拼错了关键词导致删错数据。原因字符串拼接让引号嵌套失控而且无法利用 Cypher 的参数缓存。解决全部改成参数化查询用$code、$name这样的占位符Python 驱动会自动处理转义。这是源码工程里最该移植的经验没有之一。5.4 批量写入越跑越慢事务与批量大小没配合好现象写入前几千条数据很快到后面速度明显下降甚至到 1 万条以后基本卡住。原因常见两种情况一是事务里混入了先MATCH再CREATE的长路径操作锁竞争积累二是batch_size设置过大单事务处理时间过长。解决把所有重活拆成节点和关系两阶段每批控制在 500 以内并且保证每个批次的 Cypher 只做一类操作。索引缺失也会放大这个问题先补索引再调批量大小。5.5 DETACH DELETE 清空整库的惊魂时刻现象想清掉测试数据执行MATCH (n) DETACH DELETE n结果整个图库全部空了包括后面辛辛苦苦补好的数据。原因MATCH (n)匹配了图里所有节点没有任何标签过滤。解决清库前先确认范围限定标签比如只清Company和PersonMATCH (n:Company) DETACH DELETE n如果真的要整库清空先确认这绝不是你的主库。这类操作没有后悔药我吃过亏之后清库前永远先跑一条MATCH (n:Company) RETURN count(n)确认数量。6. 图谱验证与加分技巧用 Cypher 做股权穿透和环检测图谱构建完成不等于作业交付。最后一步是验证以及把最有价值的两类查询展示出来。评审老师不在乎你写了多少行 Python他在乎的是这个“知识图谱”能不能回答关系型数据回答不了的问题。6.1 用 Cypher 验证图谱构建结果先做总量统计确认节点和边的数量与清洗后的源数据一致MATCH (n:Company) RETURN count(n) AS company_cnt; MATCH ()-[r:holds_share]-() RETURN count(r) AS share_relation_cnt;再抽一条真实路径做业务验证比如查某家公司三层内的股权穿透MATCH p(a:Company)-[:holds_share*1..3]-(b:Company) WHERE a.company_code 000001 RETURN p LIMIT 10;*1..3表示路径长度 1 到 3 跳。如果返回的路径里出现明显不符合业务常识的中间节点说明关系写入方向或节点匹配有问题需要回到构建脚本里查。验证这一步不能省它相当于给上一章所有步骤做一次体检。6.2 两个值得加分的进阶查询第一个是找循环持股这是股权结构里的异常信号也是图数据库比关系型数据库有明显优势的场景MATCH p(a:Company)-[:holds_share*1..5]-(a) RETURN p LIMIT 10;第二个是“自然人通过中间公司间接控制目标公司”的典型风控查询MATCH (p:Person)-[:holds_share]-(mid:Company)-[:holds_share]-(target:Company) WHERE target.company_code 000001 RETURN p.name AS person, mid.name AS intermediate, target.name AS target这两个查询可以直接写进答辩演示脚本里比单纯展示“图谱有多漂亮”更有说服力。给我的教训是构建流程做得再完整如果最后拿不出一两个能讲清楚业务价值的查询期末大作业的分数会被拉低一档。我自己的习惯是图谱导入完先截图保存实体概览再把上面这两个查询跑通把结果导出成 CSV 放到output/目录里。这样即使现场 Neo4j 临时启动不了也有一份可展示的成果。构建知识图谱的价值从来不在于“把数据放进图里”而在于放进去之后能查出新信息。把这个逻辑想清楚你的项目就从“作业”变成了“作品”。希望帮到你。本文还有配套的精品资源点击获取