mixle.stats.processes.hawkes_process module

Evaluate, estimate, and sample from a univariate Hawkes process with an exponential kernel.

Defines HawkesProcessDistribution, HawkesProcessSampler, HawkesProcessAccumulatorFactory, HawkesProcessAccumulator, HawkesProcessEstimator, and HawkesProcessDataEncoder.

A Hawkes process is a self-exciting temporal point process: every event transiently raises the intensity of future events. With the exponential triggering kernel g(s) = alpha * exp(-beta s) the conditional intensity given the history is

lambda(t) = mu + sum_{t_j < t} alpha * exp(-beta (t - t_j)),

with background rate mu > 0, excitation jump alpha >= 0, and decay rate beta > 0. The branching ratio alpha / beta is the expected number of direct offspring per event; the process is sub-critical (stationary) when alpha < beta.

Data type: each observation is a 1-D array/list of event times, sorted, lying in the fixed window [0, window]. The exact log-likelihood of one realization t_1 < ... < t_n is the standard point-process log-likelihood sum_i log lambda(t_i) - integral_0^window lambda(s) ds:

sum_i log(mu + alpha R_i) - mu*window - (alpha/beta) sum_i (1 - exp(-beta (window - t_i))),

where R_i = sum_{j<i} exp(-beta (t_i - t_j)) obeys the O(n) recursion R_i = exp(-beta (t_i - t_{i-1})) (R_{i-1} + 1) (R_1 = 0).

Fitting uses the Veen-Schoenberg / Lewis-Mohler EM over the latent branching structure (each event is an immigrant from mu or an offspring of an earlier event). The E-step responsibilities and the offspring-delay statistic are accumulated in O(n) via a companion recursion, giving finite sufficient statistics (S0, G, W, n_events, total_window) and the closed-form M-step mu = S0/total_window, beta = G/W, alpha = beta * G / n_events (the standard edge-effect-free branching estimator; exact as window -> inf).

Reference: Hawkes, ‘Spectra of some self-exciting and mutually exciting point processes’, Biometrika (1971).

class HawkesProcessDistribution(mu, alpha, beta, window, name=None, keys=None)

Bases: SequenceEncodableProbabilityDistribution

Univariate Hawkes process with an exponential excitation kernel on a fixed window.

Parameters:

• mu (float)

• alpha (float)

• beta (float)

• window (float)

• name (str | None)

• keys (str | None)

intensity(t, times, marks=None)

Conditional rate lambda(t) = mu + alpha sum_{t_i < t} exp(-beta (t - t_i)) given the history.

times is the event history; marks is accepted for TemporalPointProcess signature parity (the univariate Hawkes process is unmarked) and is ignored.

Parameters:

• t (float)

• times (Any)

• marks (Any)

Return type:

float

expected_count(t_start, t_end, times, marks=None)

Compensator integral_{t_start}^{t_end} lambda(s) ds of the intensity given the history.

marks is accepted for signature parity and ignored (the univariate process is unmarked).

Parameters:

• t_start (float)

• t_end (float)

• times (Any)

• marks (Any)

Return type:

float

density(x)

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

Parameters:

x (Any)

Return type:

float

log_density(x)

Exact log-likelihood of one realization (a sorted event-time sequence in [0, window]).

Parameters:

x (Any)

Return type:

float

seq_log_density(x)

Vectorized exact log-likelihood over a padded (num_realizations, max_len) time matrix.

Parameters:

x (tuple[ndarray, ndarray, float])

Return type:

ndarray

sampler(seed=None)

Return a HawkesProcessSampler (Ogata thinning) for this distribution.

Parameters:

seed (int | None)

Return type:

HawkesProcessSampler

estimator(pseudo_count=None)

Return a HawkesProcessEstimator over the same observation window.

Parameters:

pseudo_count (float | None)

Return type:

HawkesProcessEstimator

dist_to_encoder()

Returns a HawkesProcessDataEncoder bound to this window.

Return type:

HawkesProcessDataEncoder

class HawkesProcessSampler(dist, seed=None)

Bases: DistributionSampler

Draw realizations on [0, window] by Ogata’s thinning algorithm.

Parameters:

• dist (HawkesProcessDistribution)

• seed (int | None)

sample(size=None)

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

Parameters:

size (int | None)

Return type:

ndarray | list[ndarray]

class HawkesProcessAccumulator(window, name=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

Accumulate the EM branching sufficient statistics (S0, G, W, n_events, total_window).

Parameters:

• window (float)

• name (str | None)

• keys (str | None)

update(x, weight, estimate)

Parameters:

• x (Any)

• weight (float)

• estimate (HawkesProcessDistribution | None)

Return type:

None

initialize(x, weight, rng)

Parameters:

• x (Any)

• weight (float)

• rng (RandomState | None)

Return type:

None

seq_update(x, weights, estimate)

Parameters:

• x (tuple[ndarray, ndarray, float])

• weights (ndarray)

• estimate (HawkesProcessDistribution)

Return type:

None

seq_initialize(x, weights, rng)

Parameters:

• x (tuple[ndarray, ndarray, float])

• weights (ndarray)

• rng (RandomState | None)

Return type:

None

combine(suff_stat)

Parameters:

suff_stat (tuple[float, float, float, float, float])

Return type:

HawkesProcessAccumulator

value()

Return type:

tuple[float, float, float, float, float]

from_value(x)

Parameters:

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

Return type:

HawkesProcessAccumulator

scale(c)

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:

HawkesProcessAccumulator

key_merge(stats_dict)

Pool this accumulator’s statistics into stats_dict under its merge key.

The structural default implements the common single-key pattern: store the accumulator under self.keys the first time the key is seen, else combine into the one already there. Accumulators with several named keys (e.g. an HMM’s init/trans/state keys) or a non-accumulator stats payload override this. A keys of None (the default) is a no-op.

Parameters:

stats_dict (dict[str, Any])

Return type:

None

key_replace(stats_dict)

Replace this accumulator’s statistics from the pooled stats_dict entry (see key_merge).

Parameters:

stats_dict (dict[str, Any])

Return type:

None

acc_to_encoder()

Return type:

HawkesProcessDataEncoder

class HawkesProcessAccumulatorFactory(window, name=None, keys=None)

Bases: StatisticAccumulatorFactory

Factory for HawkesProcessAccumulator.

Parameters:

• window (float)

• name (str | None)

• keys (str | None)

make()

Return type:

HawkesProcessAccumulator

class HawkesProcessEstimator(window, name=None, keys=None)

Bases: ParameterEstimator

EM branching M-step: mu=S0/total_window, beta=G/W, alpha=beta*G/n_events.

Parameters:

• window (float)

• name (str | None)

• keys (str | None)

accumulator_factory()

Return type:

HawkesProcessAccumulatorFactory

estimate(nobs, suff_stat)

Estimate (mu, alpha, beta) from the accumulated branching sufficient statistics.

Parameters:

• nobs (float | None)

• suff_stat (tuple[float, float, float, float, float])

Return type:

HawkesProcessDistribution

class HawkesProcessDataEncoder(window)

Bases: DataSequenceEncoder

Encode realizations (event-time arrays) into a padded (num_realizations, max_len) matrix.

Parameters:

window (float)

seq_encode(x)

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[Any])

Return type:

tuple[ndarray, ndarray, float]