mixle.stats.directional.projected_normal module

Projected normal distribution – a flexible circular law from projecting a 2-D Gaussian.

The angle theta = atan2(Z2, Z1) of a bivariate normal Z ~ N(mu, I_2) follows the (isotropic) projected normal PN(mu) on the circle. With mu = (mu_x, mu_y) and a(theta) = mu_x cos theta + mu_y sin theta = mu . u(theta), its density is

f(theta; mu) = (1 / 2 pi) exp(-||mu||^2 / 2) [1 + a Phi(a) / phi(a)],

where phi/Phi are the standard normal pdf/cdf. It is uniform at mu = 0 and concentrates in the direction of mu as ||mu|| grows; unlike the von Mises it can be asymmetric/peaked depending on mu. It samples exactly (draw N(mu, I_2), take the angle). Parameters are fit by EM with the latent radius (Nunez-Antonio & Gutierrez-Pena 2005): given the current mu, E[r | theta] has a closed form and the M-step is mu = mean(E[r | theta] u(theta)).

The ratio a Phi(a) / phi(a) is evaluated stably for all a via the scaled complementary error function: Phi(a)/phi(a) = sqrt(pi/2) erfcx(-a/sqrt2), so M(a) := sqrt(pi/2) erfcx(-a/sqrt2) and a Phi(a)/phi(a) = a M(a).

References

• Mardia & Jupp, Directional Statistics (2000), sec. 3.5.6 (projected/offset normal).

• Nunez-Antonio & Gutierrez-Pena, “A Bayesian analysis of directional data using the projected normal distribution”, J. Applied Statistics (2005).

class ProjectedNormalDistribution(mu_x, mu_y, name=None, keys=None)

Bases: SequenceEncodableProbabilityDistribution

Isotropic projected normal PN(mu) on the circle, mu = (mu_x, mu_y).

Parameters:

• mu_x (float)

• mu_y (float)

• name (str | None)

• keys (str | None)

density(x)

Return the probability density at a single angle (radians).

Parameters:

x (float)

Return type:

float

log_density(x)

Return the log-density at a single angle (radians).

Parameters:

x (float)

Return type:

float

seq_log_density(x)

Return vectorized log-density for sequence-encoded (cos, sin) observations.

Parameters:

x (tuple[ndarray, ndarray])

Return type:

ndarray

classmethod compute_capabilities()

classmethod compute_declaration()

classmethod backend_legacy_sufficient_statistics(x, params, engine)

Per-row E-step stats (E[r] cos, E[r] sin, 1) under the CURRENT estimate’s parameters.

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, …]

classmethod backend_log_density_from_params(cos_t, sin_t, mu_x, mu_y, engine)

Engine-neutral projected-normal log-density from (cos, sin) and the mean vector.

Parameters:

• cos_t (Any)

• sin_t (Any)

• mu_x (Any)

• mu_y (Any)

• engine (Any)

Return type:

Any

backend_seq_log_density(x, engine)

Engine-neutral vectorized log-density for encoded (cos, sin) data.

Parameters:

• x (Any)

• engine (Any)

Return type:

Any

classmethod backend_stacked_params(dists, engine)

Stacked projected-normal mean vectors for a homogeneous mixture kernel.

Parameters:

• dists (Sequence[ProjectedNormalDistribution])

• engine (Any)

Return type:

dict[str, Any]

classmethod backend_stacked_log_density(x, params, engine)

Return an (n, k) matrix of projected-normal log densities.

Parameters:

• x (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

Any

classmethod backend_stacked_sufficient_statistics(x, weights, params, engine)

Stacked E-step resultants (sum r cos, sum r sin, count) under per-component parameters.

Parameters:

• x (Any)

• weights (Any)

• params (dict[str, Any])

• engine (Any)

Return type:

tuple[Any, Any, Any]

sampler(seed=None)

Return a sampler that draws N(mu, I_2) and returns the angle.

Parameters:

seed (int | None)

Return type:

ProjectedNormalSampler

estimator(pseudo_count=None)

Return an EM (latent-radius) estimator for mu.

Parameters:

pseudo_count (float | None)

Return type:

ProjectedNormalEstimator

dist_to_encoder()

Return the data encoder used by this distribution (cos/sin of the angle).

Return type:

ProjectedNormalDataEncoder

class ProjectedNormalSampler(dist, seed=None)

Bases: DistributionSampler

Draw angles as atan2(Z2, Z1) for Z ~ N(mu, I_2).

Parameters:

• dist (ProjectedNormalDistribution)

• seed (int | None)

sample(size=None)

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)

Return type:

float | ndarray

class ProjectedNormalAccumulator(name=None, keys=None)

Bases: SequenceEncodableStatisticAccumulator

Accumulate the E[r|theta]-weighted resultant (sum r*cos, sum r*sin, count) (EM E-step).

Parameters:

• name (str | None)

• keys (str | None)

update(x, weight, estimate)

Parameters:

• x (float)

• weight (float)

• estimate (ProjectedNormalDistribution | None)

Return type:

None

initialize(x, weight, rng)

Parameters:

• x (float)

• weight (float)

• rng (RandomState | None)

Return type:

None

seq_update(x, weights, estimate)

Parameters:

• x (tuple[ndarray, ndarray])

• weights (ndarray)

• estimate (Any)

Return type:

None

seq_initialize(x, weights, rng)

Parameters:

• x (tuple[ndarray, ndarray])

• weights (ndarray)

• rng (RandomState | None)

Return type:

None

combine(suff_stat)

Parameters:

suff_stat (tuple[float, float, float])

Return type:

ProjectedNormalAccumulator

value()

Return type:

tuple[float, float, float]

from_value(x)

Parameters:

x (tuple[float, float, float])

Return type:

ProjectedNormalAccumulator

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:

ProjectedNormalDataEncoder

class ProjectedNormalAccumulatorFactory(name=None, keys=None)

Bases: StatisticAccumulatorFactory

Factory for ProjectedNormalAccumulator.

Parameters:

• name (str | None)

• keys (str | None)

make()

Return type:

ProjectedNormalAccumulator

class ProjectedNormalEstimator(name=None, keys=None)

Bases: ParameterEstimator

EM estimator: mu = mean(E[r|theta] u(theta)) (one M-step per accumulated E-step).

Parameters:

• name (str | None)

• keys (str | None)

accumulator_factory()

Return type:

ProjectedNormalAccumulatorFactory

estimate(nobs, suff_stat)

Parameters:

• nobs (float | None)

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

Return type:

ProjectedNormalDistribution

class ProjectedNormalDataEncoder

Bases: DataSequenceEncoder

Encode angles as their cosine and sine.

seq_encode(x)

Encode the iid observation sequence x for vectorized evaluation.

Parameters:

x (Sequence[float])

Return type:

tuple[ndarray, ndarray]