Model API

Core road-Hawkes API

Model APIs: parametric Hawkes, marked Hawkes, neural-kernel adapters.

class motac.models.RoadHawkesConfig(kernel, family='poisson', init_alpha=0.1, init_beta=0.001, init_dispersion=10.0, kernel_fn=None, validate_kernel=True, maxiter=600, max_travel_time_s=None, speed_gate_smoothness_s=300.0, mu_ridge=0.0, mu_laplacian=0.0, stability_mode='warn', stability_penalty=100.0)[source]

Bases: object

Configuration for the parametric road-constrained Hawkes model.

kernel: ndarray
family: str
init_alpha: float
init_beta: float
init_dispersion: float
kernel_fn: KernelFn | None
validate_kernel: bool
maxiter: int
max_travel_time_s: float | None
speed_gate_smoothness_s: float
mu_ridge: float
mu_laplacian: float
stability_mode: str
stability_penalty: float
__init__(kernel, family='poisson', init_alpha=0.1, init_beta=0.001, init_dispersion=10.0, kernel_fn=None, validate_kernel=True, maxiter=600, max_travel_time_s=None, speed_gate_smoothness_s=300.0, mu_ridge=0.0, mu_laplacian=0.0, stability_mode='warn', stability_penalty=100.0)
motac.models.fit(*, dataset, substrate, config, aggregate_marks=True)[source]

Fit the parametric road-constrained Hawkes model.

Return type:

RoadHawkesFit

motac.models.predict(*, fitted, horizon, n_samples=None, seed=0)[source]

Predict intensities over a horizon; optionally sample counts.

Return type:

dict[str, object]

motac.models.loglik(*, fitted, dataset=None, substrate=None)[source]

Compute log-likelihood for the fitted model on a dataset.

Return type:

float

motac.models.simulate(*, params, substrate, kernel, T, seed=0, family='poisson', kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Simulate counts using fitted parameters and substrate.

Return type:

ndarray

motac.models.fit_road_hawkes_mle(*, travel_time_s, kernel, y, family='poisson', init_mu=None, init_alpha=0.1, init_beta=0.001, init_dispersion=10.0, kernel_fn=None, validate_kernel=True, maxiter=600, max_travel_time_s=None, speed_gate_smoothness_s=300.0, mu_ridge=0.0, mu_laplacian=0.0, stability_mode='warn', stability_penalty=100.0)[source]

Fit (mu, alpha, beta) (and optionally dispersion) for road-constrained model.

This fitter uses MLE and a sparse W(d_travel) kernel.

Parameters:

  • travel_time_s (csr_matrix) – CSR travel-time neighbourhood matrix.

  • kernel (ndarray) – Discrete lag kernel.

  • y (ndarray) – Count matrix (n_cells, n_steps).

  • family (str) – “poisson” or “negbin”.

  • kernel_fn (KernelFn | None) – Optional travel-time kernel function W(d_travel) overriding exp(-beta*d). If provided, it is validated via validate_kernel_fn by default.

Return type:

dict with fitted parameters and optimisation result.

motac.models.forecast_intensity_horizon(*, travel_time_s, mu, alpha, beta, kernel, y_history, horizon, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Deterministic intensity forecast for multiple steps ahead.

This iterates next-step intensity prediction using a mean-field update: the predicted intensity at each step is appended to the history as the expected count for subsequent steps.

Parameters:

  • y_history (ndarray) – Past counts (n_cells, n_steps_history).

  • horizon (int) – Number of steps to forecast (>= 1).

Returns:

Array of shape (n_cells, horizon).

Return type:

lam_forecast

motac.models.forecast_count_paths_horizon(*, travel_time_s, mu, alpha, beta, kernel, y_history, horizon, n_paths=200, seed=0, family='poisson', dispersion=None, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Sample probabilistic forecast paths for counts and intensities.

Return type:

tuple[ndarray, ndarray]

Returns:

  • count_paths – Integer count samples, shape (n_paths, n_cells, horizon).

  • lam_paths – Forecast mean paths, shape (n_paths, n_cells, horizon).

motac.models.forecast_probabilistic_horizon(*, travel_time_s, mu, alpha, beta, kernel, y_history, horizon, n_paths=200, seed=0, family='poisson', dispersion=None, quantiles=(0.05, 0.5, 0.95), kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Convenience probabilistic forecast wrapper.

Returns dictionary with path samples and summaries.

Return type:

dict[str, ndarray]

motac.models.summarize_forecast_paths(*, count_paths, quantiles=(0.05, 0.5, 0.95))[source]

Summarize Monte Carlo count paths with means and quantiles.

Return type:

dict[str, ndarray]

motac.models.road_loglik(*, travel_time_s, mu, alpha, beta, kernel, y, family='poisson', dispersion=None, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Log-likelihood for road-constrained count model under Poisson or NegBin.

Return type:

float

motac.models.validate_categorical_marks_matrix(marks, *, y_obs, n_marks=None)[source]

Validate a categorical mark matrix aligned to a binned observation matrix.

Contract (v1): marks are integer-coded categorical labels with the same shape as y_obs.

Parameters:

  • marks (ndarray) – Candidate mark array.

  • y_obs (ndarray) – Observation count matrix to align against.

  • n_marks (int | None) – Optional upper bound on the number of categories. When provided, all mark values must satisfy 0 <= marks < n_marks.

Returns:

The validated marks as a NumPy array (possibly a view/copy from input).

Return type:

numpy.ndarray

motac.models.encode_categorical_marks_onehot(marks, *, y_obs, n_marks, dtype=<class 'numpy.float32'>)[source]

Encode categorical mark labels as a one-hot tensor.

Parameters:

  • marks (ndarray) – Integer-coded categorical marks of shape (n_cells, n_steps).

  • y_obs (ndarray) – Observation count matrix to align against.

  • n_marks (int) – Number of mark categories.

  • dtype (dtype | type) – Output dtype for the one-hot tensor.

Returns:

One-hot encoding of shape (n_cells, n_steps, n_marks).

Return type:

numpy.ndarray

motac.models.mean_negative_log_likelihood(*, y, mean, family='poisson', dispersion=None)[source]

Mean negative log-likelihood for count observations given predicted means.

Return type:

float

motac.models.predict_intensity_one_step_road(*, travel_time_s, mu, alpha, beta, kernel, y_history, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

One-step-ahead intensity forecast using sparse road-constrained neighbours.

Model:

h(t) = sum_l kernel[l-1] * y(t-l) W = exp(-beta * d_travel) (or W = kernel_fn(d_travel)) lambda(t) = mu + alpha * (W @ h(t))

Parameters:

  • travel_time_s (csr_matrix) – CSR travel-time matrix (seconds) between cells.

  • mu (ndarray) – Baseline per cell (n_cells,).

  • alpha (float) – Non-negative excitation scale.

  • beta (float) – Positive travel-time decay rate. Ignored when kernel_fn is provided.

  • kernel (ndarray) – Discrete lag kernel.

  • y_history (ndarray) – Past counts (n_cells, n_steps_history).

  • kernel_fn (KernelFn | None) – Optional neural-kernel callable mapping nonnegative travel times to nonnegative weights. If provided, the kernel is validated via validate_kernel_fn and used to build W with travel_time_kernel_from_fn.

  • validate_kernel (bool) – If True (default), validate kernel_fn via validate_kernel_fn.

Returns:

Intensities for next step (n_cells,).

Return type:

lam_next

motac.models.predict_intensity_in_sample(*, travel_time_s, mu, alpha, beta, kernel, y, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Predict in-sample intensities lambda[:, t] for an observed count series.

Return type:

ndarray

motac.models.predict_intensity_next_step(*, travel_time_s, mu, alpha, beta, kernel, y_history, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Predict next-step intensities given a history y[:, :T].

Return type:

ndarray

motac.models.fit_forecast_road_hawkes_mle(*, travel_time_s, kernel, y, horizon, family='poisson', init_mu=None, init_alpha=0.1, init_beta=0.001, init_dispersion=10.0, kernel_fn=None, validate_kernel=True, maxiter=600, max_travel_time_s=None, speed_gate_smoothness_s=300.0, mu_ridge=0.0, mu_laplacian=0.0, stability_mode='warn', stability_penalty=100.0)[source]

Convenience workflow: fit -> deterministic intensity forecast.

This is a glue-layer for the parametric road-kernel Hawkes model.

Steps

  1. Fit (mu, alpha, beta) (and optionally dispersion) by MLE via motac.models.fit.fit_road_hawkes_mle().

  2. Forecast the next horizon intensities via motac.models.forecast.forecast_intensity_horizon().

type travel_time_s:

csr_matrix

param travel_time_s:

CSR travel-time neighbourhood matrix.

type kernel:

ndarray

param kernel:

Discrete lag kernel.

type y:

ndarray

param y:

Observed/simulated count matrix (n_cells, n_steps).

type horizon:

int

param horizon:

Forecast horizon (>= 1).

returns:

  • fit: dict returned by fit_road_hawkes_mle

  • lam_forecast: ndarray (n_cells, horizon)

rtype:

dict with keys

class motac.models.KernelFn(*args, **kwargs)[source]

Bases: Protocol

A minimal kernel function interface.

Contract (v1 scaffold): kernels map a nonnegative distance / travel-time tensor to a same-shaped, nonnegative weight tensor.

Implementations should be pure and deterministic.

__init__(*args, **kwargs)
motac.models.validate_kernel_fn(kernel, *, name='kernel')[source]

Validate that a kernel satisfies the minimal (v1) shape/value contract.

This is intentionally small and opinionated: it exists to catch accidental contract drift early (e.g. returning negative weights or wrong shapes).

Parameters:

  • kernel (KernelFn) – Callable to validate.

  • name (str) – Used in raised error messages.

Return type:

None

class motac.models.ExpDecayKernel(lengthscale=1.0)[source]

Bases: object

A tiny, deterministic kernel implementation for unit tests.

Computes w = exp(-d / lengthscale) elementwise.

Parameters:

lengthscale (float) – Positive scale parameter controlling the rate of decay.

lengthscale: float = 1.0
__init__(lengthscale=1.0)
class motac.models.SoftplusQuadraticKernel(a0=0.0, a1=-0.001, a2=0.0)[source]

Bases: object

A tiny nonlinear kernel for ablation studies.

Computes w = softplus(a0 + a1 * d + a2 * d^2) elementwise to allow departures from a pure exponential decay while preserving nonnegativity.

a0: float = 0.0
a1: float = -0.001
a2: float = 0.0
__init__(a0=0.0, a1=-0.001, a2=0.0)
class motac.models.StabilityDiagnostics(kernel_mass, max_row_sum, branching_bound, stable_sufficient)[source]

Bases: object

Coarse stability diagnostics for discrete-time road Hawkes counts.

We use a sufficient-condition bound based on the max row sum of travel-time weights and the lag-kernel mass:

b = alpha * sum(kernel) * max_i sum_j W_ij

Values < 1 indicate a conservative subcritical regime.

kernel_mass: float
max_row_sum: float
branching_bound: float
stable_sufficient: bool
__init__(kernel_mass, max_row_sum, branching_bound, stable_sufficient)
motac.models.compute_stability_diagnostics(*, travel_time_s, alpha, beta, kernel, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Return conservative subcriticality diagnostics for current parameters.

Return type:

StabilityDiagnostics

class motac.models.ParameterRecoverySummary(seeds, mu_true, alpha_true, beta_true, mu_hat, alpha_hat, beta_hat, loglik, loglik_init)[source]

Bases: object

Summary statistics for a multi-seed parameter recovery run.

seeds: tuple[int, ...]
mu_true: ndarray
alpha_true: float
beta_true: float
mu_hat: ndarray
alpha_hat: ndarray
beta_hat: ndarray
loglik: ndarray
loglik_init: ndarray
property n_seeds: int
mu_mae_per_seed()[source]
Return type:

ndarray

alpha_abs_err()[source]
Return type:

ndarray

beta_abs_err()[source]
Return type:

ndarray

__init__(seeds, mu_true, alpha_true, beta_true, mu_hat, alpha_hat, beta_hat, loglik, loglik_init)
motac.models.run_parameter_recovery_road_hawkes_poisson(*, travel_time_s, kernel, mu_true, alpha_true, beta_true, n_steps, seeds, maxiter=600)[source]

Run a small multi-seed parameter recovery harness for the road Hawkes fitter.

This utility is intentionally CI-safe and lightweight: - tiny synthetic substrate - Poisson family - multi-seed to reduce flakiness

Returns a summary object with per-seed fitted parameters and simple error metrics.

Return type:

ParameterRecoverySummary

class motac.models.NeuralIntensityBaseline(bias, weight=1.0, window=7)[source]

Bases: object

Deterministic baseline intensity model over recent counts.

bias: ndarray
weight: float
window: int
predict_intensity(*, y_history)[source]
Return type:

ndarray

__init__(bias, weight=1.0, window=7)

Stability diagnostics

class motac.models.stability.StabilityDiagnostics(kernel_mass, max_row_sum, branching_bound, stable_sufficient)[source]

Bases: object

Coarse stability diagnostics for discrete-time road Hawkes counts.

We use a sufficient-condition bound based on the max row sum of travel-time weights and the lag-kernel mass:

b = alpha * sum(kernel) * max_i sum_j W_ij

Values < 1 indicate a conservative subcritical regime.

kernel_mass: float
max_row_sum: float
branching_bound: float
stable_sufficient: bool
__init__(kernel_mass, max_row_sum, branching_bound, stable_sufficient)
motac.models.stability.compute_stability_diagnostics(*, travel_time_s, alpha, beta, kernel, kernel_fn=None, validate_kernel=True, max_travel_time_s=None, speed_gate_smoothness_s=300.0)[source]

Return conservative subcriticality diagnostics for current parameters.

Return type:

StabilityDiagnostics

Marked Hawkes scaffolding

The marked-Hawkes scaffold lives in motac.models.marked_hawkes.

Categorical marks contract (v1)

For CI-safe unit tests and a stable, minimal API, marks are represented as an integer-coded categorical label matrix aligned to the binned observation matrix y_obs:

  • marks.shape == y_obs.shape == (n_cells, n_steps)

  • marks has an integer dtype

  • marks is non-negative

  • optionally, a category bound can be enforced via n_marks such that 0 <= marks < n_marks

Use validate_categorical_marks_matrix(marks, y_obs=..., n_marks=...) to enforce this contract.

One-hot encoding helper

encode_categorical_marks_onehot(marks, y_obs=..., n_marks=...) produces a one-hot tensor of shape (n_cells, n_steps, n_marks).

Marked Hawkes model scaffolding.

This module is intentionally minimal: it establishes a stable import path and a small set of types/helpers for future marked Hawkes variants (e.g. event type, severity, or other per-event marks).

motac.models.marked_hawkes.validate_categorical_marks_matrix(marks, *, y_obs, n_marks=None)[source]

Validate a categorical mark matrix aligned to a binned observation matrix.

Contract (v1): marks are integer-coded categorical labels with the same shape as y_obs.

Parameters:

  • marks (ndarray) – Candidate mark array.

  • y_obs (ndarray) – Observation count matrix to align against.

  • n_marks (int | None) – Optional upper bound on the number of categories. When provided, all mark values must satisfy 0 <= marks < n_marks.

Returns:

The validated marks as a NumPy array (possibly a view/copy from input).

Return type:

numpy.ndarray

motac.models.marked_hawkes.encode_categorical_marks_onehot(marks, *, y_obs, n_marks, dtype=<class 'numpy.float32'>)[source]

Encode categorical mark labels as a one-hot tensor.

Parameters:

  • marks (ndarray) – Integer-coded categorical marks of shape (n_cells, n_steps).

  • y_obs (ndarray) – Observation count matrix to align against.

  • n_marks (int) – Number of mark categories.

  • dtype (dtype | type) – Output dtype for the one-hot tensor.

Returns:

One-hot encoding of shape (n_cells, n_steps, n_marks).

Return type:

numpy.ndarray