工具¶
- lifelines.utils.add_covariate_to_timeline(long_form_df, cv, id_col, duration_col, event_col, start_col='start', stop_col='stop', add_enum=False, overwrite=True, cumulative_sum=False, cumulative_sum_prefix='cumsum_', delay=0) DataFrame¶
这是一个实用函数,用于帮助创建一个长格式表格,跟踪受试者的协变量随时间的变化。它旨在随着时间推移添加越来越多的协变量进行跟踪时迭代使用。在使用此函数之前,建议查看文档https://lifelines.readthedocs.io/en/latest/Time%20varying%20survival%20regression.html#dataset-creation-for-time-varying-regression
- Parameters:
long_form_df (DataFrame) – 一个包含初始或中间“长”形式的时间变化观测值的DataFrame。必须包含列id_col、‘start’、‘stop’和event_col。请参阅函数to_long_format以将数据转换为长格式。
cv (DataFrame) – 一个包含(可能不止一个)随时间跟踪的协变量的DataFrame。必须包含列 id_col和duration_col。duration_col表示自受试者生命开始以来的时间。
id_col (string) – 长格式数据框和交叉验证中表示受试者唯一标识符的列。
duration_col (string) – cv中表示观察发生时间自出生以来的列。
event_col (string) – df中表示感兴趣事件是否发生的列
add_enum (bool, 可选) – 一个布尔标志,用于表示是否为每个主题添加一个枚举行的列。这对于指定特定观察非常有用,例如:df[df[‘enum’] == 1] 将获取每个主题的第一个观察。
overwrite (bool, 可选) – 如果为True,当cv和long_form_df中都存在该列且时间戳相同时,long_form_df中的协变量值将被cv中的协变量值覆盖。如果为False,默认行为是将值相加。
cumulative_sum (bool, 可选) – 随时间累加新的协变量。如果协变量是新增的,而不是状态变化(例如:给予更多药物 vs 测量体温),则此操作有意义。
cumulative_sum_prefix (字符串, 可选) – 添加到计算的累积和列的前缀
delay (int, 可选) – 向协变量添加延迟(用于检查分析中的反向因果关系)
- Returns:
long_form_df – 一个DataFrame,包含更新的行以反映从cv添加的新时间切片(如果有的话),以及从cv添加的新协变量的新列(或更新的列)
- Return type:
数据框
- lifelines.utils.concordance_index(event_times, predicted_scores, event_observed=None) float¶
计算一系列事件时间和预测分数之间的一致性指数(C-index)。第一个是来自观察数据的真实生存时间,另一个是来自某种模型的预测分数。
c-index 是模型在观测数据中,当 X 确实大于 Y 时,预测 X 大于 Y 的频率的平均值。c-index 还处理如何处理删失值(显然,如果 Y 是删失的,很难知道 X 是否真的大于 Y)。
一致性指数是一个介于0和1之间的值,其中:
0.5 是随机预测的预期结果,
1.0 表示完全一致,
0.0 表示完全的反一致性(将预测值乘以 -1 得到 1.0)
内部进行的计算是
>>> (pairs_correct + 0.5 * pairs_tied) / admissable_pairs
其中
pairs_correct是满足条件的对数,即如果t_x > t_y,则s_x > s_y的对数,pairs_tied是s_x = s_y的对数,而admissable_pairs是所有可能的对数。细微之处 在于如何处理被截断的观测(例如:由于截断,并非所有对数都能被评估)。- Parameters:
event_times (iterable) – 一个长度为n的可迭代对象,包含观察到的生存时间。
predicted_scores (可迭代对象) – 一个长度为n的可迭代对象,包含预测分数 - 这些可以是生存时间、风险等。参见 https://stats.stackexchange.com/questions/352183/use-median-survival-time-to-calculate-cph-c-statistic/352435#352435
event_observed (iterable, optional) – 一个长度为n的可迭代对象,表示审查标志,1表示已观察,0表示未观察。默认值为None,表示假设所有都已观察。
- Returns:
c-index – 一个介于0和1之间的值。
- Return type:
浮点数
参考文献
Harrell FE, Lee KL, Mark DB. 多变量预后模型:模型开发、评估假设和充分性、测量和减少误差的问题。医学统计 1996;15(4):361-87.
示例
from lifelines.utils import concordance_index cph = CoxPHFitter().fit(df, 'T', 'E') concordance_index(df['T'], -cph.predict_partial_hazard(df), df['E'])
- lifelines.utils.covariates_from_event_matrix(df, id_col) DataFrame¶
这是一个辅助函数,用于处理特定格式的二进制事件数据流,并将其转换为add_covariate_to_timeline可以接受的格式。例如,假设你有一个数据集如下所示:
id promotion movement raise 0 1 1.0 NaN 2.0 1 2 NaN 5.0 NaN 2 3 3.0 5.0 7.0
其中值(除了id列)表示特定用户的事件发生时间,相对于主体的出生/进入时间。这是从SQL表中提取数据的常见格式。我们称之为持续时间矩阵,我们希望将此DataFrame转换为可以包含在长格式DataFrame中的格式(有关此内容的更多详细信息,请参见add_covariate_to_timeline)。
持续时间矩阵应为每个主题有一行(但不一定是所有主题)。
- Parameters:
df (DataFrame) – 我们想要转换的DataFrame
id_col (string) – 长格式数据框和交叉验证中表示受试者唯一标识符的列。
示例
cv = covariates_from_event_matrix(duration_df, 'id') long_form_df = add_covariate_to_timeline(long_form_df, cv, 'id', 'duration', 'e', cumulative_sum=True)
- lifelines.utils.datetimes_to_durations(start_times, end_times, fill_date=datetime.datetime(2024, 10, 29, 12, 0, 58, 377961), freq='D', dayfirst=False, na_values=None, format=None)¶
这是一个非常灵活的函数,用于将start_times和end_times数组转换为适合lifelines的格式:duration和event observation数组。
- Parameters:
start_times (数组、Series 或 DataFrame) – 表示开始时间的可迭代对象。这些可以是字符串或日期时间对象。
end_times (数组、Series 或 DataFrame) – 表示结束时间的可迭代对象。这些可以是字符串或日期时间。这些值可以是 None 或空字符串,对应于审查。
fill_date (一个日期时间、数组、Series 或 DataFrame,可选(默认=datetime.Today())) – 如果 end_times 缺失或为空时使用的日期。这对应于观察的最后日期。此日期之后的任何内容也会被截断。
freq (string, optional (default=’D’)) – 使用的时间单位。参见 Pandas 的 'freq'。默认为 'D' 表示天。
dayfirst (bool, 可选 (默认=False)) – 参见 Pandas to_datetime
na_values (list[str], optional) – 要识别为NA/NaN的值列表。例如:[‘’, ‘NaT’]
format – 请参阅 Pandas 的 to_datetime
- Returns:
T (numpy array) – 表示持续时间的浮点数数组,时间单位由freq给出。
C (numpy array) – 事件观察的布尔数组:如果观察到死亡则为1,否则为0。
示例
from lifelines.utils import datetimes_to_durations start_dates = ['2015-01-01', '2015-04-01', '2014-04-05'] end_dates = ['2016-02-02', None, '2014-05-06'] T, E = datetimes_to_durations(start_dates, end_dates, freq="D") T # array([ 397., 1414., 31.]) E # array([ True, False, True])
- lifelines.utils.find_best_parametric_model(event_times, event_observed=None, scoring_method: str = 'AIC', additional_models=None, censoring_type='right', timeline=None, alpha=None, ci_labels=None, entry=None, weights=None, show_progress=False)¶
为了快速确定最佳的单变量模型,此函数将遍历lifelines中可用的每个参数模型,并选择最小化特定拟合度量的模型。
¹最佳,根据拟合度量。
- Parameters:
event_times (list, np.array, pd.Series) – 一个 (n,) 数组,表示观察到的生存时间。如果是区间截尾数据,则为 (下限, 上限) 的元组。
event_observed (list, np.array, pd.Series) – 一个 (n,) 数组,表示审查标志,1 表示已观察,0 表示未观察。默认值为 None,表示假设所有都已观察。
scoring_method (string) – 其中之一 {“AIC”, “BIC”}
additional_models (list) – 实现lifelines API的其他参数模型的列表。
censoring_type (str) – {“right”, “left”, “interval”}
timeline (list, optional) – 返回模型在时间线中的值(正增长)
alpha (float, optional) – 置信区间中的alpha值。仅在此次调用fit时覆盖初始化的alpha。
ci_labels (list, optional) – 为生成的置信区间添加自定义列名,作为一个长度为2的列表:[<下界名称>, <上界名称>]。默认值:
entry (一个长度为n的数组或pd.Series) – 受试者进入研究时的相对时间。这对于左截断(非左删失)的观察非常有用。如果为None,则所有成员在“出生”时进入研究:时间零点。
weights (一个数组,或长度为n的pd.Series) – 每个观测值的整数权重
注意
由于不稳定性,这里没有测试GeneralizedGammaFitter。
- Return type:
拟合的最佳模型和最佳得分的元组
- lifelines.utils.group_survival_table_from_events(groups, durations, event_observed, birth_times=None, weights=None, limit=-1) Tuple[ndarray, DataFrame, DataFrame, DataFrame]¶
将多个事件序列连接成DataFrames。这是survival_table_from_events对分组数据的泛化。
- Parameters:
groups (a (n,) array) – 个体的组ID。
durations (a (n,) 数组) – 每个个体的持续时间
event_observed (a (n,) array) – 事件观察,如果观察到则为1,否则为0。
birth_times (一个 (n,) 数组) – 当受试者首次被观察到的时间。受试者的死亡事件发生在 [出生时间 + 观察持续时间]。 通常设置为全零,但也可以是正数或负数。
limit
- Returns:
unique_groups (np.array) – 所有存在的唯一组的数组
removed (DataFrame) – 每个组在event_times时的移除计数数据的DataFrame,列名为‘removed:
’ observed (DataFrame) – 每个组在event_times时的观察计数数据的DataFrame,列名为‘observed:
’ censored (DataFrame) – 每个组在event_times时的删失计数数据的DataFrame,列名为‘censored:
’
示例
#input group_survival_table_from_events(waltonG, waltonT, np.ones_like(waltonT)) #data available in test_suite.py #output [ array(['control', 'miR-137'], dtype=object), removed:control removed:miR-137 event_at 6 0 1 7 2 0 9 0 3 13 0 3 15 0 2 , observed:control observed:miR-137 event_at 6 0 1 7 2 0 9 0 3 13 0 3 15 0 2 , censored:control censored:miR-137 event_at 6 0 0 7 0 0 9 0 0 , ]
- lifelines.utils.k_fold_cross_validation(fitters, df, duration_col, event_col=None, k=5, scoring_method='log_likelihood', fitter_kwargs={}, seed=None)¶
对数据集执行交叉验证。如果提供了多个模型,所有模型将在每个k个子集上进行训练。
- Parameters:
fitters (model) – 一个或多个对象,这些对象拥有一个方法:
fit(self, data, duration_col, event_col)请注意,最后两个参数将作为关键字参数给出, 并且event_col是可选的。这些对象还必须定义 下面提到的“predictor”方法。df (DataFrame) – 一个包含必要列 duration_col 和(可选的)event_col 的 Pandas DataFrame,以及其他协变量。duration_col 指的是受试者的生存时间。event_col 指的是是否观察到“死亡”事件:如果观察到则为1,否则为0(删失)。
duration_col (string) – DataFrame中包含受试者寿命的列的名称。
event_col (string, optional) – DataFrame 中包含受试者死亡观察的列的名称。如果留空,则假定所有个体都未被审查。
k (int) – 执行折叠的次数。将保留n/k的数据用于测试。
scoring_method (str) – 其中之一 {‘log_likelihood’, ‘concordance_index’} log_likelihood: 返回平均未惩罚的部分对数似然。 concordance_index: 返回一致性指数
fitter_kwargs – 传递给fitter.fit方法的关键字参数。
seed (在 np.random.seed 中固定一个种子)
- Returns:
results – (k,1) 每个折叠的分数列表。分数可以是任何内容。
- Return type:
列表
- lifelines.utils.median_survival_times(model_or_survival_function) float¶
计算生存函数的中位生存时间。
- Parameters:
model_or_survival_function (lifelines 模型或 DataFrame) – 这可以是一个单变量的 lifelines 模型,或者是一个包含一个或多个生存函数的 DataFrame。
- lifelines.utils.qth_survival_time(q: float, model_or_survival_function) float¶
返回单个生存函数达到第q百分位数的时间,即求解 \(q = S(t)\) 中的 \(t\)。
- Parameters:
q (float) – 值介于0和1之间。
model_or_survival_function (Series, 单列 DataFrame, 或 lifelines 模型)
- lifelines.utils.qth_survival_times(q, survival_functions) DataFrame | float¶
找到一个或多个生存函数达到第q百分位数的时间。
- Parameters:
q (float or array) – 一个介于0和1之间的浮点数,表示生存函数达到第q百分位数的时间。
survival_functions (一个 (n,d) 的 DataFrame、Series 或 NumPy 数组。) – 如果是 DataFrame 或 Series,将返回索引值(实际时间) 如果是 NumPy 数组,将返回索引。
- Returns:
如果 d==1,返回一个浮点数,如果是无穷大则返回 np.inf。 如果 d > 1,返回一个包含首次超过该值的时间的 DataFrame。
- Return type:
浮点数,或数据框
- lifelines.utils.restricted_mean_survival_time(model_or_survival_function, t: float = inf, return_variance=False) float | Tuple[float, float]¶
计算生存函数的限制平均生存时间,RMST。其定义为
\[\text{RMST}(t) = \int_0^t S(\tau) d\tau\]我们使用上限而不是总是使用\(\infty\)的原因是因为生存函数的尾部具有高方差,并且对RMST有强烈影响。
- Parameters:
model_or_survival_function (lifelines 模型或 DataFrame) – 这可以是一个单变量模型,或者一个 pandas DataFrame。不过,前者会提供更准确的估计。
t (float) – RMST中积分的上限。
示例
from lifelines import KaplanMeierFitter, WeibullFitter from lifelines.utils import restricted_mean_survival_time kmf = KaplanMeierFitter().fit(T, E) restricted_mean_survival_time(kmf, t=3.5) restricted_mean_survival_time(kmf.survival_function_, t=3.5) wf = WeibullFitter().fit(T, E) restricted_mean_survival_time(wf) restricted_mean_survival_time(wf.survival_function_)
参考文献
https://bmcmedresmethodol.biomedcentral.com/articles/10.1186/1471-2288-13-152#Sec27
- lifelines.utils.survival_events_from_table(survival_table, observed_deaths_col='observed', censored_col='censored')¶
这是函数
survival_table_from_events的反函数。- Parameters:
survival_table (DataFrame) –
- a pandas DataFrame with index as the durations and columns “observed” and “censored”, referring to
在时间t死亡和被审查的个体数量。
observed_deaths_col (str, 可选 (默认: “observed”)) – 生存表中表示在特定时间观察到死亡的受试者数量的列
censored_col (str, 可选 (默认: “censored”)) – 生存表中表示在特定时间被审查的受试者数量的列
- Returns:
T (数组) – 观察持续时间 – 每个元素对应一个观察时间
E (数组) – 事件观察 – 如果观察到则为1,否则为0。
W (数组) – 权重 – 用于“压缩”数据的整数权重
示例
# Ex: The survival table, as a pandas DataFrame: observed censored index 1 1 0 2 0 1 3 1 0 4 1 1 5 0 1 # would return T = np.array([ 1., 2., 3., 4., 4., 5.]), E = np.array([ 1., 0., 1., 1., 0., 0.]) W = np.array([ 1, 1, 1, 1, 1, 1])
另请参阅
survival_table_from_events
- lifelines.utils.survival_table_from_events(death_times, event_observed, birth_times=None, columns=['removed', 'observed', 'censored', 'entrance', 'at_risk'], weights=None, collapse=False, intervals=None) DataFrame¶
从右删失数据集中创建生存表。
- Parameters:
death_times ((n,) array) – 表示事件时间
event_observed ((n,) 数组) – 1 表示观察到的事件,0 表示删失事件。
birth_times (一个 (n,) 数组,可选) – 表示受试者首次被观察的时间。受试者的死亡事件则发生在 [出生时间 + 观察持续时间]。 如果为 None(默认值),birth_times 将被设置为第一次观察时间或 0,取较小者。
columns (iterable, optional) – 一个长度为3的数组,依次调用被移除的个体、观察到的死亡和审查。
weights ((n,1) 数组, 可选) – 可选的参数,用于为个体使用权重。如果未提供,则假定权重为1。
collapse (bool, 可选 (默认=False)) – 如果为True,将生存表折叠为生命表,以显示间隔区间内的事件
intervals (可迭代对象, 可选) – 默认为 None,否则是一个列表/(n,1) 数组的区间边缘测量值。如果保持为 None 同时 collapse=True,则将使用 Freedman-Diaconis 规则来确定直方图区间的间隔。
- Returns:
Pandas DataFrame,其索引为event_times中的唯一时间或间隔。名为‘removed’的列指的是在期末从人口中移除的个体数量。名为‘observed’的列指的是被观察到死亡的移除个体数量(即未删失)。名为‘censored’的列定义为‘removed’ - ‘observed’(由于event_observed而离开人口的个体数量)。
- Return type:
数据框
示例
#Uncollapsed output removed observed censored entrance at_risk event_at 0 0 0 0 11 11 6 1 1 0 0 11 7 2 2 0 0 10 9 3 3 0 0 8 13 3 3 0 0 5 15 2 2 0 0 2 #Collapsed output removed observed censored at_risk event_at (0, 2] 34 33 1 312 (2, 4] 84 42 42 278 (4, 6] 64 17 47 194 (6, 8] 63 16 47 130 (8, 10] 35 12 23 67 (10, 12] 24 5 19 32
- lifelines.utils.to_episodic_format(df, duration_col, event_col, id_col=None, time_gaps=1) DataFrame¶
此函数接受一个“扁平”数据集(即非时间变化的数据集),并将其转换为具有静态变量的时间变化数据集。
如果你的数据集有不满足比例风险假设的变量,并且你需要创建一个随时间变化的数据集以包含与时间的交互项,这将非常有用。
- Parameters:
df (DataFrame) – 静态数据集的DataFrame。
duration_col (string) – 表示df中代表每个主题持续时间的列的字符串。
event_col (string) – 表示df中代表受试者是否经历事件的列的字符串。
id_col (string, 可选) – 指定表示ID的列,否则lifelines会创建一个自动递增的列。
time_gaps (float or int) – 指定所需的时间间隔。例如,如果时间间隔为2,且一个主题存活了10.5个单位时间, 那么最终的长格式将包含5 + 1行:(0, 2], (2, 4], (4, 6], (6, 8], (8, 10], (10, 10.5] 较小的时间间隔将产生较大的DataFrame,而较大的时间间隔将产生较小的DataFrame。在极限情况下, 长格式的DataFrame将与原始DataFrame相同。
示例
from lifelines.datasets import load_rossi from lifelines.utils import to_episodic_format rossi = load_rossi() long_rossi = to_episodic_format(rossi, 'week', 'arrest', time_gaps=2.) from lifelines import CoxTimeVaryingFitter ctv = CoxTimeVaryingFitter() # age variable violates proportional hazard long_rossi['time * age'] = long_rossi['stop'] * long_rossi['age'] ctv.fit(long_rossi, id_col='id', event_col='arrest', show_progress=True) ctv.print_summary()
- lifelines.utils.to_long_format(df, duration_col) DataFrame¶
此函数将生存分析DataFrame转换为lifelines的“长”格式。lifelines的“长”格式通常用于下一个常见函数
add_covariate_to_timeline。- Parameters:
df (DataFrame) – 一个符合标准生存分析形式的DataFrame(每个观察对象一个,包含协变量、持续时间和事件标志)
duration_col (string) – 表示df中代表每个主题持续时间的列的字符串。
- Returns:
long_form_df – 一个包含新列的DataFrame。可以将其输入到add_covariate_to_timeline
- Return type:
数据框