在医药研发与数据智能交叉领域工作时我们常常会讨论一个问题为什么有的药企管线丰富却依然难逃临床失败、监管受挫、市场退出的命运以大药企药物管线的真实走向为引子本文不讨论具体药物的成败而是把“管线命运”抽象成一个数据工程问题如何采集、清洗、建模、可视化药物管线数据并构建早期风险预警系统。这是一篇面向数据分析师、算法工程师和医药数字化从业者的系统教程读完可以直接用 Python 复现整套分析流程。1. 药物管线为什么值得做数据化复盘1.1 什么是药物管线Drug Pipeline药物管线英文通常称为 Drug Pipeline指一家制药企业从药物发现、临床前研究、I/II/III 期临床试验到注册上市全流程中所有在研项目的集合。一个典型的管线项目会经历以下阶段阶段英文缩写主要任务失败风险靶点发现Target Discovery确认疾病相关生物靶点较高先导化合物优化Lead Optimization筛选并优化候选分子较高临床前研究Preclinical动物实验、毒理、药代较高I 期临床Phase I安全性、耐受性、剂量中等II 期临床Phase II初步有效性、副作用最高III 期临床Phase III大规模确证疗效较高注册审批NDA/BLA监管机构审评中等上市后监测Phase IV真实世界安全性较低不同阶段的“命运”截然不同。有的项目在临床前就因为毒性被终止有的在 III 期才发现疗效不显著还有的在审批阶段被要求补充数据。每一类失败背后都有数据特征可循。1.2 为什么要对管线进行数字化分析传统药企对管线项目的管理往往依赖专家经验和手工汇总的 Excel 表格。这种方式在项目数量少时可行但当管线达到几十甚至上百个项目时就会出现几个典型问题历史失败原因散落在不同文档里难以及时复盘。项目阶段、适应症、靶点、试验规模等关键信息没有统一结构。无法用数据回答“当前管线里哪些项目风险最高”。新产品立项时缺少基于历史数据的风险参考。数字化分析的意义在于把“这个项目感觉不太顺”变成“该项目风险评分为 0.82接近历史失败项目的典型特征”。即使不能完全替代医学和临床专家的判断也能为决策提供量化依据。1.3 本文要解决的三个核心问题本文将围绕“管线命运”这一主题落地三个技术目标设计一套标准的药物管线数据模型至少包含阶段、适应症、靶点、试验人数、终止原因等字段。构建一个基于机器学习的管线风险评分模型用历史数据预测项目进入下一阶段的概率。开发一个可视化风险看板让管理者快速定位高风险项目。我们用模拟数据完成整套流程不涉及任何真实药企的未公开数据。整个项目使用 Python 实现适合在本地 Jupyter Notebook 或脚本环境中复现。2. 环境准备与数据说明2.1 运行环境本文示例在以下环境中验证通过操作系统Windows 10 / macOS / Linux 均可Python 版本3.9 及以上开发工具Jupyter Notebook 或 VS Code Python 插件包管理工具pip 或 conda如果你使用 conda可以通过以下命令创建独立环境conda create -n pipeline_risk python3.9 -y conda activate pipeline_risk2.2 安装依赖库我们需要安装以下 Python 库pandas数据处理numpy数值计算scikit-learn机器学习建模与评估matplotlib可视化seaborn统计图表美化执行安装pip install pandas numpy scikit-learn matplotlib seaborn各库版本不需要完全一致本文以常见稳定版本为例。如果安装时遇到网络问题可以改用国内镜像源pip install pandas numpy scikit-learn matplotlib seaborn -i https://pypi.tuna.tsinghua.edu.cn/simple2.3 数据来源与模拟数据说明真实药物管线数据属于企业商业机密或监管保密信息本文使用模拟数据。模拟数据的生成逻辑参考了公开研究中对药物研发成功率的统计II 期临床到 III 期临床的成功率普遍偏低通常在 30% 左右。肿瘤领域项目数量多但竞争激烈失败率也居高不下。有生物标志物筛选的患者人群临床试验成功率通常更稳定。我们生成 500 条模拟管线项目记录覆盖 2010 年到 2024 年进入管线的项目。每条记录包含 10 个字段具体说明见下一节。3. 核心原理如何量化“管线命运”3.1 管线数据的关键字段设计要分析一个药物管线的“命运”首先要把定性信息结构化为表格数据。我们设计以下数据字典字段名类型说明project_id字符串项目唯一编号start_year整数项目进入临床阶段年份therapeutic_area字符串治疗领域如肿瘤、心血管、神经target_type字符串靶点类型如小分子、抗体、核酸phase字符串当前/最近阶段is_biomarker_selected布尔是否采用生物标志物筛选患者trial_patients整数关键试验入组人数prior_failed_count整数同类靶点历史失败项目数partnership布尔是否合作开发final_status字符串项目结果成功、失败、进行中这里的关键在于 final_status 字段。它是我们的目标变量即项目最终走向。做模型时我们把“失败”记为 1其他记为 0用于训练分类模型。3.2 特征工程的基本思路原始数据字段不能直接全部塞给模型需要做特征工程。第一把字符串字段转成数值特征。治疗领域可以用 one-hot 编码也可以用频率编码。这里我们选择 one-hot 编码因为领域类别不多。第二构建一个新特征 “阶段前进概率”。我们可以统计历史上从当前阶段进入下一阶段的比例把这个统计值作为当前项目的一个先验特征。这个特征很能说明问题一个目前处于 II 期的项目天然就比处于 I 期的项目更接近上市但失败风险也不一样。第三构建“竞争强度”特征。把 prior_failed_count 与 target_type 组合形成“同靶点历史失败密度”。例如某靶点已经有 5 个同类项目失败说明该靶点成药性可能存在挑战。代码实现import pandas as pd import numpy as np def build_features(df): df df.copy() # 阶段前进概率 phase_transition { Phase I: 0.65, Phase II: 0.40, Phase III: 0.58, NDA/BLA: 0.90, } df[phase_transition_prob] df[phase].map(phase_transition) # 同靶点历史失败密度 df[failure_density] df[prior_failed_count] / (df[trial_patients] 1) # 使用生物标志物标记 df[biomarker_flag] df[is_biomarker_selected].astype(int) return df3.3 为什么选择可解释模型药物研发是高风险、高监管、高专业壁垒的领域。即使我们能用深度学习提高一点准确率业务方也很难接受一个黑盒模型给出的结论——尤其是在决定是否终止一个临床项目时他们必须向管理层甚至监管机构解释理由。因此本文选择逻辑回归和随机森林作为主要模型。逻辑回归的系数可以直观解释“哪个特征推高了失败概率”随机森林的特征重要性则能告诉我们“历史同类失败数量比入组人数更重要”。3.4 风险评分的设计逻辑我们不只是输出一个“失败/成功”标签而是输出一个 0 到 1 之间的风险分数。分数设计逻辑0 到 0.3低风险项目特征接近历史成功项目。0.3 到 0.6中风险需要关注关键节点。0.6 到 0.8高风险建议管理层专项审查。0.8 以上极高风险考虑终止或重新设计。风险分数来自模型的预测概率。实际项目中阈值需要结合团队的容错偏好来设置。4. 完整实战构建药物管线风险分析系统4.1 创建项目结构与模拟数据我们先创建项目目录pipeline_risk_analysis/ ├── data/ │ └── pipeline_data.csv ├── src/ │ ├── data_preprocess.py │ ├── train_model.py │ ├── predict_risk.py │ └── visualization.py ├── output/ │ └── risk_report.csv └── main.py接下来生成模拟数据。为了便于演示我们把数据生成逻辑写成一个脚本# 文件路径src/data_preprocess.py import pandas as pd import numpy as np np.random.seed(42) n 500 project_ids [fPJ{str(i).zfill(4)} for i in range(1, n 1)] start_years np.random.randint(2010, 2025, n) therapeutic_areas np.random.choice([Oncology, Cardiovascular, CNS, Immunology, Metabolic], n, p[0.35, 0.15, 0.15, 0.2, 0.15]) target_types np.random.choice([Small Molecule, Antibody, Nucleic Acid, Cell Therapy], n, p[0.4, 0.35, 0.15, 0.1]) phases np.random.choice([Phase I, Phase II, Phase III, NDA/BLA], n, p[0.3, 0.35, 0.25, 0.1]) biomarker_flags np.random.choice([True, False], n, p[0.4, 0.6]) trial_patients np.random.randint(20, 2000, n) prior_failed_counts np.random.randint(0, 8, n) partnership_flags np.random.choice([True, False], n, p[0.3, 0.7]) # 生成目标的概率由多个因素决定 logits ( -2 0.8 * (therapeutic_areas Oncology) 0.6 * (target_types Cell Therapy) 0.9 * (phases Phase II) 0.5 * (phases Phase III) - 0.7 * biomarker_flags - 0.3 * partnership_flags 0.15 * prior_failed_counts ) prob_fail 1 / (1 np.exp(-logits)) final_statuses np.where(np.random.random(n) prob_fail, Failed, Phase Completed) df pd.DataFrame({ project_id: project_ids, start_year: start_years, therapeutic_area: therapeutic_areas, target_type: target_types, phase: phases, is_biomarker_selected: biomarker_flags, trial_patients: trial_patients, prior_failed_count: prior_failed_counts, partnership: partnership_flags, final_status: final_statuses, }) df.to_csv(../data/pipeline_data.csv, indexFalse) print(df.head()) print(df[final_status].value_counts())这段代码的逻辑是用人工定义的 logits 公式生成失败概率然后通过 sigmoid 转换成 0 到 1 的概率。这样模拟出的数据具有可解释的统计规律。4.2 数据清洗与特征构造数据集生成后我们需要执行数据清洗检查缺失值。处理异常值例如入组人数为 0。把布尔字段转换为 0/1。对分类字段进行 one-hot 编码。# 文件路径src/data_preprocess.py继续追加 from sklearn.model_selection import train_test_split from sklearn.preprocessing import OneHotEncoder def load_and_clean_data(filepath): df pd.read_csv(filepath) # 去除关键字段为空的行 df df.dropna(subset[phase, final_status, trial_patients]) # 入组人数异常处理 df[trial_patients] df[trial_patients].clip(lower1) # 目标变量失败为1其余为0 df[target] (df[final_status] Failed).astype(int) return df def encode_features(df): df df.copy() # 构建风险先验特征 phase_prior df.groupby(phase)[target].mean().to_dict() df[phase_prior_risk] df[phase].map(phase_prior) # 治疗领域 one-hot area_dummies pd.get_dummies(df[therapeutic_area], prefixarea) target_dummies pd.get_dummies(df[target_type], prefixtarget) df pd.concat([ df, area_dummies, target_dummies ], axis1) feature_cols [ phase_prior_risk, is_biomarker_selected, trial_patients, prior_failed_count, partnership, ] list(area_dummies.columns) list(target_dummies.columns) # 布尔转整数 for col in [is_biomarker_selected, partnership]: df[col] df[col].astype(int) return df, feature_cols这里的一个关键设计是phase_prior_risk特征。它用历史数据中每个阶段的平均失败率作为先验。这样做的好处是模型可以捕捉“阶段本身的难度”而不是让模型从头学习阶段与失败的关系。4.3 训练风险预测模型我们选择逻辑回归和随机森林两个模型先做一个基础对比。# 文件路径src/train_model.py import pandas as pd from sklearn.linear_model import LogisticRegression from sklearn.ensemble import RandomForestClassifier from sklearn.metrics import classification_report, roc_auc_score from sklearn.model_selection import train_test_split from sklearn.preprocessing import StandardScaler def train_models(df, feature_cols): X df[feature_cols] y df[target] X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42, stratifyy ) # 逻辑回归 scaler StandardScaler() X_train_scaled scaler.fit_transform(X_train) X_test_scaled scaler.transform(X_test) lr LogisticRegression(max_iter1000, C1.0) lr.fit(X_train_scaled, y_train) y_pred_lr lr.predict(X_test_scaled) y_prob_lr lr.predict_proba(X_test_scaled)[:, 1] print( Logistic Regression ) print(classification_report(y_test, y_pred_lr)) print(AUC:, roc_auc_score(y_test, y_prob_lr)) # 随机森林 rf RandomForestClassifier( n_estimators200, max_depth6, min_samples_leaf10, random_state42 ) rf.fit(X_train, y_train) y_pred_rf rf.predict(X_test) y_prob_rf rf.predict_proba(X_test)[:, 1] print( Random Forest ) print(classification_report(y_test, y_pred_rf)) print(AUC:, roc_auc_score(y_test, y_prob_rf)) # 特征重要性 importance_df pd.DataFrame({ feature: feature_cols, importance: rf.feature_importances_ }).sort_values(importance, ascendingFalse) print(\n Feature Importance ) print(importance_df) return lr, rf, scaler, X_test, y_test这里有个细节逻辑回归对特征尺度敏感所以我们用StandardScaler做标准化。随机森林基于树模型不需要标准化。4.4 生成风险评分与预警规则模型训练后我们把风险预测应用到整个数据集并生成一版风险报告。# 文件路径src/predict_risk.py def generate_risk_report(df, feature_cols, model, scaler, output_path): X df[feature_cols] X_scaled scaler.transform(X) df[risk_score] model.predict_proba(X_scaled)[:, 1] # 风险等级划分 def risk_level(score): if score 0.3: return Low elif score 0.6: return Medium elif score 0.8: return High else: return Critical df[risk_level] df[risk_score].apply(risk_level) report_cols [ project_id, start_year, therapeutic_area, target_type, phase, is_biomarker_selected, trial_patients, prior_failed_count, partnership, risk_score, risk_level ] report df[report_cols].sort_values(risk_score, ascendingFalse) report.to_csv(output_path, indexFalse) print(f风险报告已保存: {output_path}) print(report.head(20))这个风险等级规则是可配置的。企业落地时可以根据医学团队和管理层的风险偏好调整阈值。比如某个团队只想看到最危险的 5% 项目可以把 Critical 阈值从 0.8 上调到 0.9。4.5 可视化风险看板我们用 matplotlib 和 seaborn 生成三张图各阶段项目的风险分布箱线图。风险等级的分布柱状图。风险分值与历史同类失败数的散点图。# 文件路径src/visualization.py import matplotlib.pyplot as plt import seaborn as sns def plot_risk_report(df, output_dir../output): plt.rcParams[font.sans-serif] [SimHei, Arial Unicode MS, DejaVu Sans] plt.rcParams[axes.unicode_minus] False # 图1各阶段风险分布 plt.figure(figsize(10, 6)) sns.boxplot(datadf, xphase, yrisk_score, order[Phase I, Phase II, Phase III, NDA/BLA]) plt.title(不同临床阶段的项目风险分布) plt.xlabel(阶段) plt.ylabel(风险评分) plt.savefig(f{output_dir}/risk_by_phase.png, dpi150, bbox_inchestight) plt.show() # 图2风险等级分布 plt.figure(figsize(8, 5)) order [Low, Medium, High, Critical] sns.countplot(datadf, xrisk_level, orderorder) plt.title(管线项目风险等级分布) plt.xlabel(风险等级) plt.ylabel(项目数量) plt.savefig(f{output_dir}/risk_level_dist.png, dpi150, bbox_inchestight) plt.show() # 图3风险与历史失败数 plt.figure(figsize(10, 6)) sns.scatterplot(datadf, xprior_failed_count, yrisk_score, huerisk_level, alpha0.6) plt.title(历史同类失败项目数与风险评分的关系) plt.xlabel(同类靶点历史失败项目数) plt.ylabel(风险评分) plt.savefig(f{output_dir}/risk_vs_failures.png, dpi150, bbox_inchestight) plt.show()三张图分别回答不同问题哪个阶段最危险当前管线整体是否健康历史失败数据对当前项目是否有警示作用4.6 主流程串联最后写一个main.py把整个流程串起来# 文件路径main.py import sys sys.path.append(src) from data_preprocess import load_and_clean_data, encode_features from train_model import train_models from predict_risk import generate_risk_report from visualization import plot_risk_report if __name__ __main__: raw_df load_and_clean_data(data/pipeline_data.csv) df, feature_cols encode_features(raw_df) lr, rf, scaler, X_test, y_test train_models(df, feature_cols) generate_risk_report( df, feature_cols, modellr, scalerscaler, output_pathoutput/risk_report.csv ) plot_risk_report(df, output_diroutput)运行命令python main.py预期输出包含两个模型的评估指标、特征重要性排序、风险报告 CSV以及三张 PNG 图片。5. 常见问题与排查思路在实际运行过程中最容易遇到以下几类问题问题现象常见原因解决思路导入 sklaern 失败包名拼写错误应为 scikit-learn检查安装命令重装依赖运行后中文显示为方框matplotlib 缺少中文字体设置 font.sans-serif 并安装中文字体模型 AUC 过高或过低模拟数据分布与真实场景差异明显用真实数据做充分探索不要直接套用阈值报错 KeyError: target数据清洗时没有执行目标列构造检查 load_and_clean_data 是否先于 encode_features 运行特征列训练和预测不一致训练集与预测集分类取值不同用同一套 one-hot 编码器处理数据一个比较隐蔽的问题是phase_prior_risk是利用全量数据的标签计算得到的如果直接用于训练会造成轻微的数据泄露。在实际项目中这个特征应该只从训练集统计然后映射到验证集和测试集。这里为了演示简化了流程真实业务中必须用交叉验证或时间切分。排查顺序一般建议先看原始数据确认字段类型和缺失值。查看特征矩阵的 shape 是否一致。查看测试集评估指标是否稳定。打印几条预测结果结合业务经验判断是否合理。6. 最佳实践与工程建议6.1 数据治理统一字段标准是第一步药物管线数据来自多个系统包括研发管理系统、临床试验运营系统、医学事务部门甚至外部数据库。做任何分析之前先定义统一的数据字典。建议至少做到每个项目有唯一 ID。明确阶段字段的取值范围不允许出现“Phase2”“phase II”之类的不规范写法。所有日期字段使用统一格式。失败原因用代码表管理例如 F001 表示安全性问题F002 表示疗效不足。字段标准化听起来简单实际项目中 80% 的时间都花在这儿。6.2 模型可解释性比精度更重要在药物研发场景中业务方不会因为 AUC 提升了 0.02 就接受模型建议。他们更关心“为什么这个项目被评为高危”。推荐做法用 SHAP 库输出每个样本的特征贡献。在风险报告中附上 top3 风险因子。人工复核规则风险等级达到 High 的项目必须由临床专家结合数据做二次确认。例如一个项目被评为 Critical可能是因为它处于 II 期、没有生物标志物筛选、且同类靶点已有 4 个项目失败。这些解释比单纯给出一个分数更有决策价值。6.3 警惕数据泄露与时间穿越药物研发数据天然带有时间属性。2015 年的项目不可能用到 2019 年才积累的先验信息。因此在构建特征时要注意历史失败数要按“当前时间点之前”进行统计而不是用全量数据。先验概率特征必须在训练集上计算后应用。时间序列数据建议按年份划分训练集和测试集而不是随机划分。用时间切分比随机切分更接近真实预测场景。6.4 技术栈选型与部署本文示例适合分析探索。如果要在企业内落地为常态化分析系统建议升级方向数据存储使用 MySQL/PostgreSQL 存储管线项目表。调度用 Airflow 或 APScheduler 定时执行 ETL 和模型更新。可视化接入 PowerBI 或 Superset避免每次生成静态图片。模型管理用 MLflow 保存训练参数、指标和模型文件方便回溯。6.5 合规与伦理边界药物管线数据中可能存在商业敏感信息和患者隐私信息在处理时需要注意不把真实试验患者级数据导出到非授权环境。分析结果只做内部研发管理参考不用于对外宣传。数据库操作遵循最小权限原则删除数据前备份到隔离环境。这个原则在医药行业尤其重要因为一旦出现数据安全事故不仅影响企业声誉还可能面临监管处罚。7. 总结与学习路线本文以阿斯利康药物管线的命运为引子但重点不在复盘某一家公司的具体决策而是把“管线命运”转化为一个可量化、可建模、可复现的数据分析问题。我们走完了从数据设计、特征工程、模型训练到风险报告生成的全流程。回顾关键知识点药物管线的阶段划分和失败风险差异。如何用 Python 生成符合业务逻辑的模拟数据。如何构造先验风险特征和分类编码特征。如何用逻辑回归和随机森林训练风险预测模型。如何输出可解释的风险报告和可视化图表。实际落地时需要注意的数据泄露、合规和部署问题。如果继续深入学习可以按以下路线推进掌握基础学习 pandas 数据清洗与 sklearn 建模流程。进阶统计学习生存分析Survival Analysis因为药物试验数据天然适合用 Cox 回归等生存模型处理。业务结合阅读医药行业公开的研发成功率报告理解不同适应症、靶点类型的基线风险。工程化研究如何把本文脚本改造成定时运行的数据管道并接入企业级 BI 系统。药物研发是一个高风险、长周期、强监管的行业。没有任何一个数据模型能替代医学判断但一个好设计的数据系统至少能让决策者少踩一些前面已经踩过的坑。如果你也想把历史数据变成决策参考不妨从建立一个简单的管线风险评分表开始。
尧图网络