mixle.stats.processes.inhomogeneous_poisson module¶

Evaluate, estimate, and sample from an inhomogeneous Poisson process with piecewise-constant rate.

Defines InhomogeneousPoissonProcessDistribution, InhomogeneousPoissonProcessSampler, InhomogeneousPoissonProcessAccumulatorFactory, InhomogeneousPoissonProcessAccumulator, InhomogeneousPoissonProcessEstimator, and InhomogeneousPoissonProcessDataEncoder.

Data type: each observation is a 1-D array/list of event times within the window [edges[0], edges[-1]]. The intensity lambda(t) is constant rates[b] on bin b (the bins are given by edges; uniform bins on [0, t_max] by default). The log-likelihood of one realization with per-bin event counts n_b is

sum_b n_b * log(rates[b]) - sum_b rates[b] * width[b],

i.e. the standard Poisson-process log-likelihood sum_i log lambda(t_i) - integral lambda. The MLE is closed-form: rates[b] = (total events in bin b) / (width[b] * n_realizations).

Reference: Daley & Vere-Jones, An Introduction to the Theory of Point Processes (Springer, 2003).

class InhomogeneousPoissonProcessDistribution(rates, t_max=None, edges=None, name=None, keys=None)[source]

Bases: SequenceEncodableProbabilityDistribution

Inhomogeneous Poisson process with piecewise-constant intensity on a fixed window.

Parameters:

rates (Sequence[float] | ndarray)
t_max (float | None)
edges (Sequence[float] | ndarray | None)
name (str | None)
keys (str | None)

intensity(t, times=None, marks=None)[source]

Conditional rate lambda(t) = rates[bin containing t].

The inhomogeneous Poisson process is not self-exciting, so the rate depends only on t. times/marks are accepted for TemporalPointProcess signature parity and ignored. Raises ValueError for t outside the support [edges[0], edges[-1]].

Parameters:

t (float)
times (Any)
marks (Any)

Return type:

float

expected_count(t_start, t_end, times=None, marks=None)[source]

Compensator integral_{t_start}^{t_end} lambda(s) ds – the piecewise-rate integral.

Computed as sum_b rate_b * width(overlap([t_start, t_end], bin_b)). times/marks are accepted for signature parity and ignored. With t_start=edges[0], t_end=edges[-1] this returns the full integral sum_b rate_b width_b used by log_density.

Parameters:

t_start (float)
t_end (float)
times (Any)
marks (Any)

Return type:

float

density(x)[source]

Probability density of one realization x (a sequence of event times).

Parameters:: x (Any)
Return type:: float

log_density(x)[source]

Log-likelihood of one realization: sum_b n_b log rate_b - sum_b rate_b width_b.

Parameters:: x (Any)
Return type:: float

seq_log_density(x)[source]

Vectorized log-likelihood for a (num_realizations, num_bins) matrix of per-bin counts.

Parameters:: x (ndarray)
Return type:: ndarray

sampler(seed=None)[source]

Return an InhomogeneousPoissonProcessSampler for this distribution.

Parameters:: seed (int | None)
Return type:: InhomogeneousPoissonProcessSampler

estimator(pseudo_count=None)[source]

Return an InhomogeneousPoissonProcessEstimator over the same bin edges.

Parameters:: pseudo_count (float | None)
Return type:: InhomogeneousPoissonProcessEstimator

dist_to_encoder()[source]

Returns an InhomogeneousPoissonProcessDataEncoder bound to these bin edges.

Return type:: InhomogeneousPoissonProcessDataEncoder

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

Bases: DistributionSampler

Draw realizations by binwise thinning: n_b ~ Poisson(rate_b * width_b) then uniform times.

Parameters:

dist (InhomogeneousPoissonProcessDistribution)
seed (int | None)

sample(size=None)[source]

Draw one realization (event-time array) or a list of size realizations.

Parameters:: size (int | None)
Return type:: ndarray | list[ndarray]

class InhomogeneousPoissonProcessAccumulator(edges, name=None, keys=None)[source]

Bases: SequenceEncodableStatisticAccumulator

Accumulate weighted per-bin event counts and the weighted number of realizations.

Parameters:

edges (ndarray)
name (str | None)
keys (str | None)

update(x, weight, estimate)[source]

Accumulate weighted per-bin counts for one event-time realization.

Parameters:

x (Any)
weight (float)
estimate (InhomogeneousPoissonProcessDistribution | None)

Return type:

None

initialize(x, weight, rng)[source]

Initialize the sufficient statistics with one weighted realization.

Parameters:

x (Any)
weight (float)
rng (RandomState | None)

Return type:

None

seq_update(x, weights, estimate)[source]

Accumulate weighted per-bin counts from encoded realizations.

Parameters:

x (ndarray)
weights (ndarray)
estimate (Any | None)

Return type:

None

seq_initialize(x, weights, rng)[source]

Initialize the sufficient statistics from encoded realizations.

Parameters:

x (ndarray)
weights (ndarray)
rng (RandomState | None)

Return type:

None

combine(suff_stat)[source]

Merge serialized bin-count statistics into this accumulator.

Parameters:: suff_stat (tuple[ndarray, float])
Return type:: InhomogeneousPoissonProcessAccumulator

value()[source]

Return the weighted bin counts and weighted realization count.

Return type:: tuple[ndarray, float]

from_value(x)[source]

Restore the accumulator from serialized bin-count statistics.

Parameters:: x (tuple[ndarray, float])
Return type:: InhomogeneousPoissonProcessAccumulator

scale(c)[source]

Scale accumulated sufficient statistics by a constant.

Parameters:: c (float)
Return type:: InhomogeneousPoissonProcessAccumulator

key_merge(stats_dict)[source]

Merge this accumulator into a keyed statistics dictionary.

Parameters:: stats_dict (dict[str, Any])
Return type:: None

key_replace(stats_dict)[source]

Replace this accumulator from a keyed statistics dictionary.

Parameters:: stats_dict (dict[str, Any])
Return type:: None

acc_to_encoder()[source]

Return an encoder that bins event times on this accumulator’s edges.

Return type:: InhomogeneousPoissonProcessDataEncoder

class InhomogeneousPoissonProcessAccumulatorFactory(edges, name=None, keys=None)[source]

Bases: StatisticAccumulatorFactory

Factory for InhomogeneousPoissonProcessAccumulator.

Parameters:

edges (ndarray)
name (str | None)
keys (str | None)

make()[source]

Create an empty inhomogeneous Poisson process accumulator.

Return type:: InhomogeneousPoissonProcessAccumulator

class InhomogeneousPoissonProcessEstimator(num_bins=None, t_max=None, edges=None, name=None, keys=None)[source]

Bases: ParameterEstimator

Closed-form MLE: rate_b = (weighted events in bin b) / (width_b * weighted realizations).

Parameters:

num_bins (int | None)
t_max (float | None)
edges (Sequence[float] | ndarray | None)
name (str | None)
keys (str | None)

accumulator_factory()[source]

Return a factory for inhomogeneous Poisson sufficient-statistic accumulators.

Return type:: InhomogeneousPoissonProcessAccumulatorFactory

estimate(nobs, suff_stat)[source]

Estimate per-bin rates from accumulated (bin_counts, n_realizations).

Parameters:

nobs (float | None)
suff_stat (tuple[ndarray, float])

Return type:

InhomogeneousPoissonProcessDistribution

class InhomogeneousPoissonProcessDataEncoder(edges)[source]

Bases: DataSequenceEncoder

Encode a list of realizations (event-time arrays) into a (num_realizations, num_bins) count matrix.

Parameters:: edges (Sequence[float] | ndarray)

seq_encode(x)[source]

Encode event-time realizations as per-bin count rows.

Parameters:: x (Sequence[Any])
Return type:: ndarray