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 observations.

Combinator samplers (mixture/sequence/…) accept batched. With batched=True (the default) each child stream is drawn in one vectorized call instead of a per-draw Python loop – far faster. Because every child sampler owns an independent RandomState, batching consumes each stream in the same order as the loop, so the draws are identical to the legacy path. batched=False forces that legacy per-draw loop as a guaranteed- stable reference. Leaf samplers are already vectorized and ignore the flag.

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]
Parameters:
Return type:

None

initialize(x, weight, rng)[source]
Parameters:
Return type:

None

seq_update(x, weights, estimate)[source]
Parameters:

  • weights (ndarray)

  • estimate (SurvivalDistribution)

Return type:

None

seq_initialize(x, weights, rng)[source]
Parameters:
Return type:

None

scale(c)[source]

Scale linear sufficient statistics in-place by c.

The structural default is correct for ordinary weighted sums, nested tuples/lists/dicts, and numeric arrays. Families whose value() payload includes non-linear metadata such as support bounds must override this method and leave that metadata unscaled.

Parameters:

c (float)

Return type:

SurvivalAccumulator

acc_to_encoder()[source]
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]
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 type:

SurvivalAccumulatorFactory

estimate(nobs, suff_stat)[source]
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 the iid observation sequence x for vectorized evaluation.

Parameters:

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

Return type:

tuple[Any, ndarray, ndarray]