名师讲堂｜使用 Python 计算专利颠覆性创新指数及筛选颠覆性专利（同小类内计算）

2719 字 约 10 分钟 名师讲堂 专利创新与知识Python

今天给大家分享使用 Python 计算专利颠覆性创新指数及筛选颠覆性专利的方法，该方法只在 IPC 小类内计算相似度。该方法参考自冉征等《技术集群结构与颠覆式创新——兼论关联性”陷阱”的突破路径》，通过文本相似度法来综合测度专利的颠覆性程度。

附件中提供了该参考文献的 PDF 文件，感兴趣的小伙伴可以阅读原文。

指标来源与计算原理

颠覆性创新指数（Disruptive Innovation Index）

改进方法：IPC 小类内计算

传统方法在全样本范围内计算相似度，运算量巨大。本文采用改进方法：只在 IPC 小类内计算相似度。

IPC（International Patent Classification）代码格式如 “F04D13/02”，其小类为前4位（部+大类+小类），即 “F04D”。每个专利可能有多个 IPC 代码，计算时在每个 IPC 小类内分别计算相似度，然后取均值聚合到专利级别。

计算步骤概述

  1. 数据读取：读取专利数据样本（标题、摘要、IPC 等信息）
  2. 数据预处理：专利去重、文本清洗
  3. 中文分词：使用 jieba 进行中文分词，过滤停用词
  4. IPC 小类提取：从 IPC 代码提取前4位作为小类标识
  5. TFBIDF 矩阵构建：在 IPC 小类 + 时间窗口内构建 TF-IDF 矩阵并 L2 归一化
  6. 相似度计算：计算后向相似度（BPS）和前向相似度（FPS）
  7. 颠覆性指数计算：radical = FPS / BPS
  8. 聚合到专利级别：多 IPC 小类取均值
  9. 筛选颠覆性专利：radical 指数前 5% 的专利标记为颠覆式创新
  10. 结果保存：输出 CSV 文件

使用 reticulate 创建与管理 Python 虚拟环境

在 R 中通过 reticulate 包来调用 Python，最好的实践是为项目创建一个专属的 Python 虚拟环境，将所需依赖隔离到独立空间，避免与系统 Python（如 Anaconda）发生版本冲突。

重要说明（避免”已初始化”报错）：reticulate 在 R 会话中只能绑定一次 Python——一旦某个 {python} 代码块运行，Python 解释器就被锁定，之后再调用 use_virtualenv() 会报错：

ERROR: The requested version of Python cannot be used, as another version has already been initialized.

因此，虚拟环境的激活必须在所有 {python} 代码块之前完成。本文档的解决方案是在 setup chunk 中通过 Sys.setenv(RETICULATE_PYTHON = ...) 提前锁定 Python 路径，这是 reticulate 选取 Python 的最高优先级入口。

安装 reticulate（仅首次）

# 设置 CRAN 镜像（knit 时 R 处于非交互模式，不会自动选择镜像）
options(repos = c(CRAN = "https://mirrors.tuna.tsinghua.edu.cn/CRAN/"))

# 仅在尚未安装时才安装，避免每次 knit 都重装
if (!requireNamespace("reticulate", quietly = TRUE)) {
  install.packages("reticulate")
  message("reticulate 安装完成！")
} else {
  message("reticulate 已安装，版本：", packageVersion("reticulate"))
}

虚拟环境初始化原理（已在 setup chunk 中完成）

本文档的 setup chunk（隐藏运行）包含如下逻辑：

library(reticulate)

.venv_name   <- ".venv"
.venv_python <- virtualenv_python(.venv_name)

# 虚拟环境不存在时自动创建
if (!file.exists(.venv_python)) {
  virtualenv_create(.venv_name)
  .venv_python <- virtualenv_python(.venv_name)
}

# 通过环境变量抢先锁定 Python（优先级最高，早于任何 {python} chunk）
Sys.setenv(RETICULATE_PYTHON = .venv_python)
use_virtualenv(.venv_name, required = TRUE)

在虚拟环境中安装 Python 包（仅首次）

# 检查关键包是否已安装，缺失的才安装
# 数据处理：微信公众号 RStata
py_pkgs <- c(
  "numpy", "pandas", "scipy", "scikit-learn",
  "jieba"
)
installed <- py_list_packages(".venv")$package
need_install <- setdiff(py_pkgs, installed)

if (length(need_install) > 0) {
  virtualenv_install(".venv", packages = need_install)
  message("已安装缺失的包：", paste(need_install, collapse = ", "))
} else {
  message("所有 Python 包已就绪，无需安装")
}

验证激活状态

# 验证当前绑定的 Python 路径（应指向 .venv 目录）
py_config()

查看已安装的包

# 列出虚拟环境中已安装的包
pkgs <- py_list_packages(".venv")
# 只显示我们关心的包
key_pkgs <- c("numpy", "pandas", "scipy", "scikit-learn", "jieba")
pkgs[pkgs$package %in% key_pkgs, c("package", "version")]

虚拟环境管理常用命令

# 查看所有已创建的虚拟环境
virtualenv_list()

# 删除虚拟环境（当不再需要时）
# virtualenv_remove(".venv")

# 升级某个包
# virtualenv_install(".venv", packages = "jieba", ignore_installed = TRUE)

详细计算代码

加载停用词和用户词典

import os
import re
import numpy as np
import pandas as pd
from scipy.sparse import csr_matrix, diags
from sklearn.feature_extraction.text import TfidfTransformer
import jieba

# ---- 配置参数 ----
# 数据处理：微信公众号 RStata
data_dir = "专利数据样本"
stopwords_file = "stopwords.txt"
dict_file = "dictionary.txt"
target_years = range(2003, 2008)  # 2003~2007
tau = 3
output_file = "patent_disruptive_index_ipc_subclass.csv"

# 加载停用词
stopwords = set()
if os.path.exists(stopwords_file):
    with open(stopwords_file, "r", encoding="utf-8") as f:
        for line in f:
            word = line.strip()
            if word:
                stopwords.add(word)

# 加载用户词典到 jieba
if os.path.exists(dict_file):
    jieba.load_userdict(dict_file)

print(f"停用词数量: {len(stopwords)}")

数据读取

csv_files = sorted([
    os.path.join(data_dir, f)
    for f in os.listdir(data_dir)
    if f.endswith(".csv")
])

print("===== 读取专利数据 =====")
df_list = []
for f in csv_files:
    df_year = pd.read_csv(f, dtype=str, encoding="utf-8")
    df_list.append(df_year)

df_raw = pd.concat(df_list, ignore_index=True)
print(f"原始数据总行数: {len(df_raw)}")
print(f"年份文件: {', '.join(os.path.basename(f) for f in csv_files)}")

数据预处理——专利去重

# 数据处理：微信公众号 RStata
df = df_raw.copy()
df["年份"] = df["newipzlid"].str[:4].astype(int)

print(f"年份范围: {df['年份'].min()} - {df['年份'].max()}")

# 去重步骤一：清洗公开公告号（去掉末尾字母），按 年份 + 公开公告号 去重
# 去重步骤二：按 年份 + 申请号 去重
print("\n===== 专利去重 =====")
print(f"去重前: {len(df)} 行")

df["公开公告号_clean"] = df["公开公告号"].str.replace(r"[A-Z]$", "", regex=True)
df = df.drop_duplicates(subset=["年份", "公开公告号_clean"], keep="first")
df = df.drop(columns=["公开公告号_clean"])
df = df.drop_duplicates(subset=["年份", "申请号"], keep="first")

print(f"去重后: {len(df)} 行")

文本清洗与中文分词

# 数据处理：微信公众号 RStata
# 此处代码需下载讲义材料查看～

IPC 小类提取与长格式展开

# 数据处理：微信公众号 RStata
# IPC 代码格式如 "F04D13/02"，小类为前4位（部+大类+小类），即 "F04D"
print("\n===== 提取 IPC 小类 =====")

def extract_ipc_subclasses(ipc_str):
    """提取 IPC 小类代码（前4位），去重"""
    if pd.isna(ipc_str) or not ipc_str:
        return []
    ipcs = re.split(r";\s*", str(ipc_str))
    ipcs = [ipc.strip() for ipc in ipcs]
    ipcs = [ipc for ipc in ipcs if len(ipc) >= 4]
    subclasses = [ipc[:4] for ipc in ipcs]
    return list(set(subclasses))

df["ipc_subclass"] = df["IPC"].apply(extract_ipc_subclasses)
n_with_ipc = sum(df["ipc_subclass"].apply(len) > 0)
print(f"有 IPC 小类的专利数: {n_with_ipc} / {len(df)}")

# 展开为 (专利, IPC小类) 长格式
df_long = df.explode("ipc_subclass").rename(columns={"ipc_subclass": "ipc_sub"})
df_long = df_long[df_long["ipc_sub"].notna()].copy()

print(f"展开为 (专利, IPC小类) 长格式后: {len(df_long)} 行")
print(f"IPC小类总数: {df_long['ipc_sub'].nunique()}")

TFBIDF 矩阵构建与颠覆性指数计算

# 数据处理：微信公众号 RStata

# 此处代码需下载讲义材料查看～

逐年计算颠覆性指数

# 数据处理：微信公众号 RStata
all_subclasses = sorted(df_long["ipc_sub"].unique())

print(f"\n===== 计算颠覆性创新指数（IPC小类内）=====")
print(f"目标年份: {', '.join(str(y) for y in target_years)}")
print(f"计算窗口: tau = {tau} 年")
print(f"IPC小类总数: {len(all_subclasses)}")

all_results = []

for yr in target_years:
    print(f"\n--- 正在计算 {yr} 年 ---")

    year_results = []
    for sc in all_subclasses:
        result = compute_radical_by_subclass(df_long, sc, yr, tau=tau)
        if len(result) > 0:
            year_results.append(result)

    if year_results:
        year_df = pd.concat(year_results, ignore_index=True)
        print(f"  专利-IPC小类对数: {len(year_df)}")
        print(f"  涉及专利数: {year_df['newipzlid'].nunique()}")
        all_results.append(year_df)
    else:
        print(f"  {yr} 年无有效结果")

# 合并所有年份结果
df_radical_long = pd.concat(all_results, ignore_index=True)
print(f"\n总计 专利-IPC小类对数: {len(df_radical_long)}")

计算 radical 指数并聚合到专利级别

# 数据处理：微信公众号 RStata
# 方案一（求和法）：radical_sum = FPS_sum / BPS_sum
# 方案二（均值法）：radical_avg = FPS_avg / BPS_avg
# 此处代码需下载讲义材料查看～

标记颠覆性专利（全样本前 5%）

# 数据处理：微信公众号 RStata
threshold_sum = df_radical["radical_sum"].quantile(0.95)
threshold_avg = df_radical["radical_avg"].quantile(0.95)

print(f"\n===== 颠覆性创新阈值（top 5%）=====")
print(f"  方案一（求和法）: radical_sum >= {threshold_sum:.4f}")
print(f"  方案二（均值法）: radical_avg >= {threshold_avg:.4f}")

df_radical["颠覆性_sum"] = df_radical["radical_sum"].apply(
    lambda x: 1 if pd.notna(x) and x >= threshold_sum else (
        0 if pd.notna(x) else np.nan
    )
)
df_radical["颠覆性_avg"] = df_radical["radical_avg"].apply(
    lambda x: 1 if pd.notna(x) and x >= threshold_avg else (
        0 if pd.notna(x) else np.nan
    )
)

合并原始信息并保存结果

# 数据处理：微信公众号 RStata
df_info = df[["newipzlid", "年份", "标题", "摘要", "公开公告号", "申请号"]].copy()
df_info = df_info[df_info["年份"].isin(target_years)]
df_final = df_info.merge(df_radical, on=["newipzlid", "年份"], how="left")

# 保存为 CSV 文件
df_output = df_final[[
    "newipzlid", "年份", "标题", "摘要", "公开公告号", "申请号",
    "BPS_sum", "FPS_sum", "radical_sum", "颠覆性_sum",
    "BPS_avg", "FPS_avg", "radical_avg", "颠覆性_avg", "n_ipc_sub"
]].copy()

df_output["年份"] = df_output["年份"].astype(int)
df_output["颠覆性_sum"] = df_output["颠覆性_sum"].astype("Int64")
df_output["颠覆性_avg"] = df_output["颠覆性_avg"].astype("Int64")
df_output["n_ipc_sub"] = df_output["n_ipc_sub"].astype("Int64")

# 保存为 CSV 文件
# 数据处理：微信公众号 RStata
df_output.to_csv(output_file, index=False, encoding="utf-8-sig")

print(f"\n结果已保存到: {output_file}")
print("数据处理：微信公众号 RStata")

结果概览

# 数据处理：微信公众号 RStata
print("===== 结果摘要 =====")
print(f"目标年份: {min(target_years)} - {max(target_years)}")
print(f"去重后专利总数（含窗口期）: {len(df)}")
print(f"目标年份专利数: {len(df_final)}")
print(f"计算了 radical 指数的专利数: {df_final['radical_avg'].notna().sum()}")

print("\n--- 各年颠覆性专利分布（方案一：求和法）---")
for yr in target_years:
    yr_data = df_final[df_final["年份"] == yr]
    total = len(yr_data)
    valid = yr_data["radical_sum"].notna().sum()
    disrupt = int(yr_data["颠覆性_sum"].sum()) if yr_data["颠覆性_sum"].notna().any() else 0
    mean_val = yr_data["radical_sum"].mean()
    median_val = yr_data["radical_sum"].median()
    print(f"  {yr}年: 总={total}, 有效={valid}, 颠覆性={disrupt}, "
          f"均值={mean_val:.4f}, 中位数={median_val:.4f}")

print("\n--- 各年颠覆性专利分布（方案二：均值法）---")
for yr in target_years:
    yr_data = df_final[df_final["年份"] == yr]
    total = len(yr_data)
    valid = yr_data["radical_avg"].notna().sum()
    disrupt = int(yr_data["颠覆性_avg"].sum()) if yr_data["颠覆性_avg"].notna().any() else 0
    mean_val = yr_data["radical_avg"].mean()
    median_val = yr_data["radical_avg"].median()
    print(f"  {yr}年: 总={total}, 有效={valid}, 颠覆性={disrupt}, "
          f"均值={mean_val:.4f}, 中位数={median_val:.4f}")

# 两种方案一致性比较
print("\n--- 两种方案一致性比较 ---")
both = int(((df_final["颠覆性_sum"] == 1) & (df_final["颠覆性_avg"] == 1)).sum())
only_sum = int(((df_final["颠覆性_sum"] == 1) & (df_final["颠覆性_avg"] == 0)).sum())
only_avg = int(((df_final["颠覆性_sum"] == 0) & (df_final["颠覆性_avg"] == 1)).sum())
print(f"  两种方案均标记为颠覆性: {both}")
print(f"  仅求和法标记: {only_sum}")
print(f"  仅均值法标记: {only_avg}")
if both + only_sum + only_avg > 0:
    print(f"  一致率: {both / (both + only_sum + only_avg) * 100:.1f}%")

print("\n===== 全部计算完成！=====")
print("数据处理：微信公众号 RStata")

R 与 Python 函数对照表

R 函数/包 Python 函数/包 说明
dplyr::filter() df[df[...] == ...] 数据筛选
dplyr::mutate() df["col"] = ... 添加/修改列
dplyr::distinct() df.drop_duplicates() 去重
dplyr::group_by() + summarise() df.groupby().agg() 分组汇总
tidyr::unnest() df.explode() 展开列表列
stringr::str_replace_all() df.str.replace() 字符串替换
jiebaR::segment() jieba.lcut() 中文分词
tidytext::bind_tf_idf() sklearn TfidfTransformer TF-IDF 计算
tidytext::cast_sparse() scipy.sparse.csr_matrix 稀疏矩阵构建
Matrix::rowSums() matrix.sum(axis=1).A1 行求和
haven::write_dta() pandas.to_csv() 保存结果文件

点击这里跳转到 RStata 短书平台获取附件：名师讲堂｜使用 Python 计算专利颠覆性创新指数及筛选颠覆性专利（同小类内计算）

相关文章

评论