A股知识图谱构建实战：从数据采集到关系挖掘的完整指南

1次阅读

共计 2297 个字符，预计需要花费 6 分钟才能阅读完成。

知识图谱能揭示传统量化模型难以捕捉的复杂关联（如实际控制人链条、供应链隐性风险传导）
动态事件关系网络可实时监控风险传染路径（如大股东质押爆仓对关联企业的影响）
非结构化数据转化为结构化关系后，支持更细粒度的因子挖掘（财报文本情绪与股价波动的关联分析）

中文 PDF 存在表格跨页、竖排文本等特殊排版，PyPDF2 提取的原始文本需要二次清洗
关键字段（如 ” 归属于母公司净利润 ”）在不同公司财报中表述差异大（” 归母净利 ”、” 母公司股东净利润 ” 等）
附注中的会计政策变更可能影响财务指标可比性

股权变更需要同时更新股东、被投资企业、关联方三个维度的关系
担保事件涉及担保方、被担保方、反担保方的多跳关系推理
时间维度处理（如限售股解禁日期影响股东减持行为预测）

# Tushare Pro 获取股东信息的标准调用示例
import tushare as ts
pro = ts.pro_api('your_token')

def get_shareholders(code):
    try:
        df = pro.stk_holdernumber(ts_code=code, end_date='20231231')
        # 关键字段校验
        assert not df.empty, "Empty DataFrame returned"
        assert 'holder_name' in df.columns, "Missing holder_name column"
        return df[['holder_name', 'hold_amount', 'hold_ratio']]
    except Exception as e:
        print(f"Error fetching shareholders for {code}: {str(e)}")
        return pd.DataFrame()

使用 PyPDF2 提取 PDF 文本后，针对财报特点设计清洗管道：

import re
from PyPDF2 import PdfReader

def extract_financial_statement(pdf_path):
    reader = PdfReader(pdf_path)
    text = "".join(page.extract_text() for page in reader.pages)

    # 处理财报特殊字符和排版
    text = re.sub(r'(\n\s*)+', '\n', text)  # 合并多余换行
    text = re.sub(r'([0-9]),([0-9])', r'\1\2', text)  # 去除千分位逗号

    # 关键字段正则匹配（考虑多种表述方式）net_profit_pattern = r'(归属于母公司所有者的净利润 | 归母净利润)[：:]+([\d,.]+)'
    match = re.search(net_profit_pattern, text)
    return float(match.group(2).replace(',', '')) if match else None

// 上市公司 - 股东关系建模
CREATE (c:Company {code:"600000", name:"浦发银行"})
CREATE (h:Holder {name:"上海国际集团", type:"法人"})
CREATE (h)-[r:HOLDS {ratio: 29.67, amount: 883000000, update_date: date()}]->(c)

// 多层股权穿透查询
MATCH path=(h:Holder)-[:HOLDS*1..3]->(c:Company {code:"600000"})
RETURN [n in nodes(path) | n.name] AS ownership_chain

构建金融同义词库（如 ” 实控人 ”≈” 实际控制人 ”≈” 最终控制方 ”）
基于上下文的动态消解（” 质押 ” 在股权上下文指质押融资，在会计上下文可能指资产质押）

使用 APOC 插件实现高效批量更新

CALL apoc.periodic.iterate(
  'UNWIND $batch AS row RETURN row',
  'MERGE (h:Holder {name: row.holder_name})
   MERGE (c:Company {code: row.company_code})
   MERGE (h)-[r:HOLDS]->(c)
   SET r.ratio = row.hold_ratio, r.update_date = datetime()',
  {batchSize:1000, parallel:true})

添加时间版本标签实现历史数据追溯

MATCH (c:Company {code:"600000"})-[r:HOLDS]->(h)
SET r.effective_end = date()
CREATE (h)-[r_new:HOLDS {
  ratio: 30.01, 
  effective_start: date(),
  data_source: "2023 年报"}]->(c)

识别隐形关联的潜在方法：