mixle.stats.processes.temporal module

Time and date modelling on raw timestamps: periodic (cyclic) distributions and seasonal time series.

Real temporal data arrives as raw timestamps – Python datetime/date, numpy.datetime64, ISO strings, or POSIX seconds. This module consumes any of those directly. Two capabilities:

• PeriodicTimeDistribution – a distribution over where in a recurring cycle events fall (time-of-day, day-of-week, season), a von Mises on the cycle phase. Captures recurring timing: “events cluster around 9am”, “activity peaks on weekends”, “blooms in spring”.

• SeasonalTimeSeries – a conditional distribution value | time on (timestamp, value) data: a Gaussian whose mean is a linear trend plus Fourier seasonal harmonics at one or more periods. Like any distribution it has conditional (returns the predictive distribution at a time), mean, log_density and a sampler – not a predict. decompose splits the mean into trend + seasonal parts.

Part of the earth-science/multiphysics/UQ work and generally useful (paleo records are time series; event catalogues – earthquakes, blooms, drilling – have strong calendar/seasonal structure).

to_unix_seconds(x)

Convert raw date/time data to POSIX seconds (float), accepting datetime/date, datetime64, ISO strings, or numbers (already seconds). Returns a 1-D float array; NaT/None become NaN.

Parameters:

x (Any)

Return type:

ndarray

cyclic_phase(times, period)

Map timestamps to a cycle phase in [0, 2pi) for the given period (named or seconds).

Parameters:

• times (Any)

• period (float | str)

Return type:

ndarray

class PeriodicTimeDistribution(period='day', loc=0.0, conc=0.0, name=None, keys=None)

Bases: SequenceEncodableProbabilityDistribution

A distribution over raw timestamps whose recurring-cycle phase is von Mises.

The cycle phase phi = 2*pi*(t mod P)/P is distributed von Mises around a peak; this is a proper density over time within one period (a constant Jacobian log(2*pi/P) makes it integrate to 1). It follows the mixle Distribution / Sampler / Estimator / Accumulator / DataEncoder contract by delegating the circular density to VonMisesDistribution, so recurring time-of-day / day-of-week / seasonal patterns compose with the rest of the library.

period is the cycle ('day', 'week', 'year', … or seconds); loc is the peak phase (radians) and conc the concentration (0 = uniform over the cycle, large = sharply peaked).

Parameters:

• period (float | str)

• loc (float)

• conc (float)

property loc: float

property conc: float

density(t)

Return the probability density or mass at a single observation.

Concrete default: exponentiate log_density (the abstract method subclasses must provide). Leaves with a cheaper closed form may override this.

Parameters:

t (Any)

Return type:

float

log_density(t)

Log-density at one timestamp (the von Mises phase density plus the time Jacobian).

Parameters:

t (Any)

Return type:

float

seq_log_density(x)

Return vectorized log-density values for sequence-encoded observations.

Return type:

ndarray

peak_phase_fraction()

The peak location as a fraction of the cycle (e.g. 0.375 of a day = 09:00).

Return type:

float

sampler(seed=None)

Return a sampler for drawing observations from this distribution.

Parameters:

seed (int | None)

Return type:

PeriodicTimeSampler

estimator(pseudo_count=None)

Return an estimator for fitting this distribution from data.

Parameters:

pseudo_count (float | None)

Return type:

PeriodicTimeEstimator

dist_to_encoder()

Return the data encoder used by this distribution for vectorized methods.

Return type:

PeriodicTimeDataEncoder

class PeriodicTimeEstimator(period='day', name=None, keys=None)

Bases: ParameterEstimator

Maximum-likelihood estimator: the von Mises MLE of the cycle phase (delegated to VonMises).

Parameters:

period (float | str)

accumulator_factory()

Return type:

StatisticAccumulatorFactory

estimate(nobs, suff_stat)

Return type:

PeriodicTimeDistribution

class SeasonalTimeSeries(periods=('year',), harmonics=3, trend=True)

Bases: object

A conditional distribution value | time for raw (timestamp, value) series: a Gaussian whose mean is a linear trend plus Fourier seasonal harmonics.

Models value | time ~ N(mu(time), s^2) with mu(t) = b0 + b1 t + sum over periods/harmonics of [a sin(2pi k t / P) + b cos(...)], fit by least squares (t in days from the first timestamp). Being a distribution, it has no predict – you ask for the conditional distribution at a time and read its mean, sample it, or score data: conditional(t) returns a GaussianDistribution (the posterior-predictive at t, parameter uncertainty + noise), mean(times) is E[value | time], log_density(times, values) scores observations, and sampler(seed).sample(times) draws values. Captures several seasonalities at once (daily + weekly + yearly); decompose splits the mean into trend + per-period parts.

Parameters:

• periods (Sequence[float | str])

• harmonics (int)

• trend (bool)

fit(times, values)

Parameters:

• times (Any)

• values (Any)

Return type:

SeasonalTimeSeries

mean(times)

The conditional expectation E[value | time] – the fitted trend + seasonality.

Parameters:

times (Any)

Return type:

ndarray

conditional(time)

The conditional distribution p(value | time) – a GaussianDistribution (or a list, for an array of times). This is how you ‘predict’ the mixle way: you get a distribution to sample, score, or read .mu / .sigma2 from, not a bare point estimate.

Parameters:

time (Any)

log_density(times, values)

Conditional log-density of (time, value) observations under the model.

Parameters:

• times (Any)

• values (Any)

Return type:

ndarray | float

sampler(seed=None)

Parameters:

seed (int | None)

Return type:

SeasonalTimeSeriesSampler

decompose(times)

Split the prediction into trend and one component per period (the seasonal contributions).

Parameters:

times (Any)

Return type:

dict[str, ndarray]