mixle.stats.combinator.survival module

Survival model with right-censoring: a bundled event-time + censoring-indicator likelihood.

Survival/reliability data is (t, event): either the event was observed at time t (event=1) or the unit was still alive when observation stopped, a right-censored time (event=0, the true event is only known to be > t). The likelihood bundles the density for observed events with the survival function for censored ones:

log L = sum_{event} log f(t) + sum_{censored} log S(t), S(t) = 1 - F(t).

SurvivalDistribution wraps any event-time base that exposes cdf and quantile (Weibull, Exponential, GeneralizedPareto, GeneralizedExtremeValue, …). Censoring is exogenous, so the only parameters are the base’s; fitting them under censoring is the standard EM that imputes the unobserved event times: a censored time c contributes the base’s mass beyond c, approximated deterministically by a grid of conditional quantiles F^{-1}(F(c) + q (1 - F(c))). Each mixle EM iteration re-imputes with the improved base, converging to the right-censored MLE – which, unlike fitting on the observed events alone, is unbiased.

class SurvivalDistribution(base, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Right-censored survival likelihood over a base event-time distribution.

Parameters:

  • base (SequenceEncodableProbabilityDistribution)

  • name (str | None)

  • keys (str | None)

density(x)[source]

Return the likelihood contribution of (t, event).

Parameters:

x (tuple[float, int])

Return type:

float

log_density(x)[source]

Return log f(t) for an observed event, else log S(t) for a right-censored time.

Parameters:

x (tuple[float, int])

Return type:

float

seq_log_density(x)[source]

Vectorized likelihood: base density for events, log survival for censored rows.

Parameters:

x (tuple[Any, ndarray, ndarray])

Return type:

ndarray

sampler(seed=None)[source]

Return a sampler (draws uncensored event times; censoring is exogenous).

Parameters:

seed (int | None)

Return type:

SurvivalSampler

estimator(pseudo_count=None)[source]

Return an estimator that fits the base under right-censoring by conditional-quantile EM.

Parameters:

pseudo_count (float | None)

Return type:

SurvivalEstimator

dist_to_encoder()[source]

Return the data encoder (base encoding of the times + the event mask).

Return type:

SurvivalDataEncoder

class SurvivalSampler(dist, seed=None)[source]

Bases: DistributionSampler

Draw uncensored event times (t, 1) from the base (censoring is applied externally).

Parameters:

  • dist (SurvivalDistribution)

  • seed (int | None)

sample(size=None)[source]

Draw one uncensored event-time pair or a list of pairs.

Parameters:

size (int | None)

class SurvivalAccumulator(base_accumulator, base_encoder, n_impute=16, keys=None)[source]

Bases: SingleChildAccumulator

Feed observed events and conditional-quantile imputations of censored times to the base accumulator.

Parameters:

  • base_accumulator (SequenceEncodableStatisticAccumulator)

  • base_encoder (DataSequenceEncoder)

  • n_impute (int)

  • keys (str | None)

update(x, weight, estimate)[source]

Accumulate one observed or right-censored event-time record.

Parameters:
Return type:

None

initialize(x, weight, rng)[source]

Initialize the base sufficient statistics from one survival record.

Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate encoded survival records using conditional-tail imputation.

Parameters:

  • weights (ndarray)

  • estimate (SurvivalDistribution)

Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize from encoded survival records without a current estimate.

Parameters:
Return type:

None

scale(c)[source]

Scale the delegated base sufficient statistics by a constant.

Parameters:

c (float)

Return type:

SurvivalAccumulator

acc_to_encoder()[source]

Return an encoder that records event times and censoring indicators.

Return type:

SurvivalDataEncoder

class SurvivalAccumulatorFactory(base_factory, base_encoder, n_impute, keys)[source]

Bases: StatisticAccumulatorFactory

Factory for SurvivalAccumulator.

Parameters:

  • base_factory (StatisticAccumulatorFactory)

  • base_encoder (DataSequenceEncoder)

  • n_impute (int)

  • keys (str | None)

make()[source]

Create an empty survival accumulator.

Return type:

SurvivalAccumulator

class SurvivalEstimator(base_estimator, n_impute=16, name=None, keys=None)[source]

Bases: ParameterEstimator

Fit the base event-time distribution under right-censoring (conditional-quantile imputation EM).

Parameters:

  • base_estimator (ParameterEstimator)

  • n_impute (int)

  • name (str | None)

  • keys (str | None)

accumulator_factory()[source]

Return a factory for right-censored survival sufficient-statistic accumulators.

Return type:

SurvivalAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate the base event-time distribution from imputed survival statistics.

Parameters:
Return type:

SurvivalDistribution

class SurvivalDataEncoder(base_encoder)[source]

Bases: MaskedBaseEncoder

Encode (t, event) data as the base encoding of the times plus the boolean event mask.

Parameters:

base_encoder (DataSequenceEncoder)

seq_encode(x)[source]

Encode survival records as base times plus a boolean event mask.

Parameters:

x (Sequence[tuple[float, int]])

Return type:

tuple[Any, ndarray, ndarray]