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

简介：一份面向金融领域的知识图谱构建项目源码包，基于Neo4j图数据库、Python与Cypher查询语言完成。项目代码完整、结构清晰，包含从数据采集到知识存储的完整链路，适合高校计算机、人工智能、金融科技等相关专业学生用于期末大作业、课程设计或毕业设计，也适合想学习知识图谱与自然语言处理结合的开发者参考。资源共105个文件，包含7个Python脚本、7个Jupyter Notebook、多个LSTM-CRF命名实体识别模型权重文件、股票新闻CSV数据、Neo4j配置、PDF/PPT说明文档及24张运行效果截图等，整体约41.3MB，便于直接复现与二次改造。目前已有82人学习下载。包内提供从数据爬取（Scrapy）、实体识别（LSTM+CRF）到图数据构建与查询的完整流程讲解，并附带模型训练产物、checkpoint及目录说明，可按文档逐步操作，快速跑通金融知识图谱构建链路；同时预留扩展空间，可在此基础上增加实体关系、丰富可视化交互，用于毕业设计或项目展示。

1. 金融知识图谱期末大作业：为什么 Neo4j+Python+Cypher 这套组合值得做

期末大作业里放着金融知识图谱构建项目，技术栈是 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 UNIQUE

Neo4j 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 很难让老师一眼看懂业务逻辑。

对比下来更直观：

Cypher 的路径匹配是原生图遍历，写出来的语句和业务问题的描述几乎一一对应。这就是金融知识图谱选 Neo4j 的核心原因：不是为了存数据，而是为了在关系上做推导。源码工程里真正值钱的部分也是这一层——数据只是被塞进图里的原料，查询能力才是交付物。

3. 用 Python 连接 Neo4j：驱动选型与 Cypher 参数化查询

图模型设计好之后，接下来就是让 Python 和 Neo4j 对话。这个章节要解决的是：本地环境怎么搭、用哪个驱动、连接代码怎么写才不容易在期末答辩时翻车。

3.1 环境准备：本地 Neo4j 与 Python 驱动安装

Neo4j 的安装方式主要有两种：桌面版和社区版 zip 包。桌面版自带数据库管理界面，适合课程演示；社区版适合脚本自动化。无论哪种，安装完启动后要确认默认端口7687（Bolt 协议）和7474（HTTP 控制台）是通的。浏览器能打开7474页面说明服务起来了。

Python 这边只需要装官方驱动：

pip install neo4j

如果源码工程里还依赖pandas做数据清洗，一起装：

pip install pandas neo4j

安装完成后建议把依赖写进requirements.txt，方便对方复现环境。这里有个小提醒：pip 安装的 neo4j 驱动是客户端，不是数据库本身，很多人装完驱动却连不上，是因为本地根本没启动 Neo4j 服务，先分清这两件事。

3.2 官方驱动还是 py2neo：期末作业的选型参考

网上很多教程用的是py2neo，因为它提供一个更“Pythonic”的图对象操作接口。但我在实际使用中更推荐官方neo4j-driver，原因直接看对比：

期末大作业的源码如果用的是老版本 py2neo，在 Neo4j 4.4 以上版本很容易出现握手失败之类的报错。我一般会建议直接改为官方驱动，代码改动量其实不大，就是把session.run()的调用方式统一一下，换来的是兼容性稳定。这不是玄学，是驱动维护节奏决定的。

3.3 连接与第一条 Cypher：session、事务与参数化查询

用官方驱动建立连接的核心代码很短：

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", code=company_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", code=code, name=name, ) with driver.session() as session: session.execute_write(create_company, "000001", "平安银行")

execute_write会把事务的提交和回滚都处理好，函数内部抛出异常时事务自动回滚，不会留下半截数据。这种写法在批量导入时尤其重要，因为一个批次里可能只需几条坏数据，整个批次就会干净地回滚。

4. 从源码工程到完整图谱：清洗、批写入与执行顺序

这一章讲的是“构建流程详解”的核心。拿到一套知识图谱源码，正确的阅读顺序是先搞懂数据从哪来、脚本分几步、最终写进 Neo4j 的是什么。很多期末大作业的源码工程其实并不复杂，但不按顺序跑，或者漏掉中间一步，后面查询全是空的。

4.1 源码工程的目录结构与执行顺序

我一般会把期末大作业的源码工程拆成五个部分，方便答辩时讲清楚：

对应的执行顺序是固定的：先清洗，再建节点，再建关系，最后验证。如果你拿到别人的源码，先按这个顺序跑一遍，看中间每个脚本的日志输出是否正常。最容易犯的错是拿到源码就直接python build_relations.py，结果关系里的端点节点根本不存在，一个关系都写不进去。

4.2 数据清洗：从原始表到干净的实体清单

清洗这一步决定了图谱的准确率。最常见的坑是股票代码被 Excel 或 pandas 读成浮点数，000001变成1.0。所以读取时强制指定为字符串：

import pandas as pd df = pd.read_csv("data/company.csv", dtype=str).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", index=False)

dtype=str让所有列以字符串读入，避免代码、手机号这类前导零字段失真。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 """, batch=batch, ) clean_df = pd.read_csv("data/clean_company.csv", dtype=str).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 """, batch=batch, )

这里的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和Person：

MATCH (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 临时启动不了，也有一份可展示的成果。构建知识图谱的价值从来不在于“把数据放进图里”，而在于放进去之后能查出新信息。把这个逻辑想清楚，你的项目就从“作业”变成了“作品”。希望帮到你。

本文还有配套的精品资源，点击获取