由于借助 AI 工具学习编程已经变得非常容易了，因此之后的课程就不再默认进行视频讲解了，如果特别需要视频讲解也可以联系李老师预约讲解～讲义材料学习过程中遇到的问题也可以及时与李老师联系。
购买 RStata 名师讲堂会员即可参加该课程啦（之前的和未来的都可以参加）！
购买会员可以从这里下单：https://rstata.duanshu.com/#/card/list/
名师讲堂会员权益：
- 参加平台上的其他 R 语言和 Stata 的课程；
- 以会员折扣价购买我们分享的数据资料（10 元/份）；
* 如果发票可添加小编微信 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 小类内分别计算相似度，然后取均值聚合到专利级别。
计算步骤概述
- 数据读取：读取专利数据样本（标题、摘要、IPC 等信息）
- 中文分词：使用
jieba 进行中文分词，过滤停用词 - IPC 小类提取：从 IPC 代码提取前4位作为小类标识
- TFBIDF 矩阵构建：在 IPC 小类 + 时间窗口内构建 TF-IDF 矩阵并 L2 归一化
- 相似度计算：计算后向相似度（BPS）和前向相似度（FPS）
- 颠覆性指数计算：radical = FPS / BPS
- 筛选颠覆性专利：radical 指数前 5% 的专利标记为颠覆式创新
使用 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 安装完成！") message("reticulate 已安装，版本：", packageVersion("reticulate"))虚拟环境初始化原理（已在 setup chunk 中完成）
本文档的 setup chunk（隐藏运行）包含如下逻辑：
.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 包（仅首次）
"numpy", "pandas", "scipy", "scikit-learn",installed <- py_list_packages(".venv")$packageneed_install <- setdiff(py_pkgs, installed)if (length(need_install) > 0) { virtualenv_install(".venv", packages = need_install) message("已安装缺失的包：", paste(need_install, collapse = ", ")) message("所有 Python 包已就绪，无需安装")验证激活状态
# 验证当前绑定的 Python 路径（应指向 .venv 目录）查看已安装的包
pkgs <- py_list_packages(".venv")key_pkgs <- c("numpy", "pandas", "scipy", "scikit-learn", "jieba")pkgs[pkgs$package %in% key_pkgs, c("package", "version")]虚拟环境管理常用命令
# virtualenv_remove(".venv")# virtualenv_install(".venv", packages = "jieba", ignore_installed = TRUE)
详细计算代码
加载停用词和用户词典
from scipy.sparse import csr_matrix, diagsfrom sklearn.feature_extraction.text import TfidfTransformerstopwords_file = "stopwords.txt"dict_file = "dictionary.txt"target_years = range(2003, 2008) # 2003~2007output_file = "patent_disruptive_index_ipc_subclass.csv"if os.path.exists(stopwords_file):withopen(stopwords_file, "r", encoding="utf-8") as f:if os.path.exists(dict_file): jieba.load_userdict(dict_file)print(f"停用词数量: {len(stopwords)}")数据读取
os.path.join(data_dir, f)for f in os.listdir(data_dir)print("===== 读取专利数据 =====") df_year = pd.read_csv(f, dtype=str, encoding="utf-8")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)}")数据预处理——专利去重
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)} 行")文本清洗与中文分词
IPC 小类提取与长格式展开
# IPC 代码格式如 "F04D13/02"，小类为前4位（部+大类+小类），即 "F04D"print("\n===== 提取 IPC 小类 =====")defextract_ipc_subclasses(ipc_str):"""提取 IPC 小类代码（前4位），去重"""if pd.isna(ipc_str) ornot ipc_str: 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)}")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 矩阵构建与颠覆性指数计算
逐年计算颠覆性指数
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)}")print(f"\n--- 正在计算 {yr} 年 ---")for sc in all_subclasses: result = compute_radical_by_subclass(df_long, sc, yr, tau=tau) year_results.append(result) 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)df_radical_long = pd.concat(all_results, ignore_index=True)print(f"\n总计 专利-IPC小类对数: {len(df_radical_long)}")计算 radical 指数并聚合到专利级别
# 方案一（求和法）：radical_sum = FPS_sum / BPS_sum# 方案二（均值法）：radical_avg = FPS_avg / BPS_avg标记颠覆性专利（全样本前 5%）
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.nandf_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合并原始信息并保存结果
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")"newipzlid", "年份", "标题", "摘要", "公开公告号", "申请号","BPS_sum", "FPS_sum", "radical_sum", "颠覆性_sum","BPS_avg", "FPS_avg", "radical_avg", "颠覆性_avg", "n_ipc_sub"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")df_output.to_csv(output_file, index=False, encoding="utf-8-sig")print(f"\n结果已保存到: {output_file}")print("数据处理：微信公众号 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--- 各年颠覆性专利分布（方案一：求和法）---") yr_data = df_final[df_final["年份"] == yr] 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--- 各年颠覆性专利分布（方案二：均值法）---") yr_data = df_final[df_final["年份"] == yr] 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 函数对照表
| | |
|---|
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 | |
tidytext::cast_sparse() | scipy.sparse.csr_matrix | |
Matrix::rowSums() | matrix.sum(axis=1).A1 | |
haven::write_dta() | pandas.to_csv() | |
如何参加课程？
购买 RStata 名师讲堂会员即可参加该课程啦（之前的和未来的都可以参加）！
购买会员可以从这里下单：https://rstata.duanshu.com/#/card/list/
名师讲堂会员权益：
- 参加平台上的其他 R 语言和 Stata 的课程；
- 以会员折扣价购买我们分享的数据资料（10 元/份）；
* 如果发票可添加小编微信 r_stata2 （RStata 李老师）开具。如需数据资料，购买后可添加小编微信免费领取数据折扣卡。
更多关于 RStata 会员的更多信息可添加微信号 r_stata2 咨询：
课程主页（点击文末的阅读原文即可跳转）：https://rstata.duanshu.com/#/brief/course/e97101ff8c5449b398e10a8a028c34a0
10个月宝宝每天需要喝多少奶粉？
