工具

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_tieds_x = s_y 的对数,而 admissable_pairs 是所有可能的对数。细微之处 在于如何处理被截断的观测(例如:由于截断,并非所有对数都能被评估)。

Parameters:
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:

数据框