当前位置：首页>python>名师讲堂|使用 Python 计算专利颠覆性创新指数及筛选颠覆性专利(同小类内计算)

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

  • 2026-10-11 06:09:02
名师讲堂|使用 Python 计算专利颠覆性创新指数及筛选颠覆性专利(同小类内计算)

由于借助 AI 工具学习编程已经变得非常容易了，因此之后的课程就不再默认进行视频讲解了，如果特别需要视频讲解也可以联系李老师预约讲解～讲义材料学习过程中遇到的问题也可以及时与李老师联系。

购买 RStata 名师讲堂会员即可参加该课程啦（之前的和未来的都可以参加）！

价格：2800/年 或者 4800/长期

购买会员可以从这里下单：https://rstata.duanshu.com/#/card/list/

名师讲堂会员权益：

  1. 参加每个月 3～4 次的名师讲堂课程；
  2. 参加平台上的其他 R 语言和 Stata 的课程；
  3. 以会员折扣价购买我们分享的数据资料（10 元/份）；
  4. 课程内外的提问解答服务（课程外的尽量帮忙解决）。
* 如果发票可添加小编微信 r_stata2 （RStata 李老师）开具。如需数据资料，购买后可添加小编微信免费领取数据折扣卡。

更多关于 RStata 会员的更多信息可添加微信号 r_stata2 咨询：

课程主页（点击文末的阅读原文即可跳转）：https://rstata.duanshu.com/#/brief/course/e97101ff8c5449b398e10a8a028c34a0

今天给大家分享使用 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() 会报错：

安装 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):
withopen(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 小类 =====")
defextract_ipc_subclasses(ipc_str):
"""提取 IPC 小类代码（前4位），去重"""
if pd.isna(ipc_str) ornot ipc_str:
return []
    ipcs = re.split(r";\s*", str(ipc_str))
    ipcs = [ipc.strip() for ipc in ipcs]
    ipcs = [ipc for ipc in ipcs iflen(ipc) >= 4]
    subclasses = [ipc[:4] for ipc in ipcs]
returnlist(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)
iflen(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: 1if pd.notna(x) and x >= threshold_sum else (
0if pd.notna(x) else np.nan
    )
)
df_radical["颠覆性_avg"] = df_radical["radical_avg"].apply(
lambda x: 1if pd.notna(x) and x >= threshold_avg else (
0if 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() else0
    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() else0
    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 名师讲堂会员即可参加该课程啦（之前的和未来的都可以参加）！

价格：2800/年 或者 4800/长期

购买会员可以从这里下单：https://rstata.duanshu.com/#/card/list/

名师讲堂会员权益：

  1. 参加每个月 3～4 次的名师讲堂课程；
  2. 参加平台上的其他 R 语言和 Stata 的课程；
  3. 以会员折扣价购买我们分享的数据资料（10 元/份）；
  4. 课程内外的提问解答服务（课程外的尽量帮忙解决）。
* 如果发票可添加小编微信 r_stata2 （RStata 李老师）开具。如需数据资料，购买后可添加小编微信免费领取数据折扣卡。

更多关于 RStata 会员的更多信息可添加微信号 r_stata2 咨询：

课程主页（点击文末的阅读原文即可跳转）：https://rstata.duanshu.com/#/brief/course/e97101ff8c5449b398e10a8a028c34a0
本文来自网友投稿或网络内容，如有侵犯您的权益请联系我们删除，联系邮箱：596563534@qq.com 。

如何Python批量爬取项目所在区域的工业企业呢?——给排水设计人员的工作笔记

写Python就能手搓Tensor-Core GEMM!TileLang让CUDA调优原地失业(附保姆级实操)

最新文章

随机文章