Source code for hawkes_engine.model

"""The result object a fit returns.

A ``FitResult`` is frozen because it is a statement about a finished
computation, not a workbook. Everything in it was measured or derived at the
moment of the fit; nothing is recomputed lazily, so serialising it is exact and
two fits with the same inputs produce equal objects.

Spaces are named on every field that could be misread. ``hessian`` is the
observed information in **natural** space; ``cov`` is the covariance in
**log** space, because that is the space the optimizer stepped in and the space
in which the excitation parameters are comparable; ``se`` is back in **natural**
space, because that is the space the user writes parameters in. The three are
related by delta-method transforms, and a silent mix-up between them is the
most likely defect in an estimator of this kind, so the names are explicit and
the conversions happen in exactly one place (:mod:`hawkes_engine.hessian`).
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any

import numpy as np

from hawkes_engine._typing import FloatArray, JsonObject

# Schema 2 adds ``se_source``, ``se_note`` and ``auto``. A reader accepts schema 1
# and fills the three with their "not recorded" values, because a result written by
# 0.1.0 is still a true statement about a fit; it simply cannot say which method
# produced its standard errors.
SCHEMA_VERSION = 2
_READABLE_SCHEMAS = (1, 2)


[docs] @dataclass(frozen=True) class FitResult: """A maximum-likelihood fit and everything known about its trustworthiness. Attributes ---------- mu, alpha, beta Natural-space estimates. ``alpha`` is ``(M, M)`` row = target, column = source (ADR-002). ``beta`` is ``(M, M)`` or ``(M,)`` when the decay is shared, in which case ``beta_ij = beta_j``. loglik The log-likelihood at the estimate. converged The optimizer finished *and* the recomputed log-space gradient is within the convergence tolerance. ``status`` and ``nit`` are scipy's. gradient_inf_norm ``||d ell / d theta||_inf`` at the returned point, recomputed here rather than read from the optimizer's last call. rho Spectral radius of the branching matrix ``Gamma = alpha / beta``. rho_ci Delta-method 95% interval for ``rho``, or ``None`` when the covariance is unavailable. It is a first-order interval and is not valid at a boundary; it is reported because ``rho`` is the headline number and an interval is more honest than a point even when it is approximate. se, cov ``cov`` is the log-space covariance of the free parameters; ``se`` maps ``"mu"``, ``"alpha"``, ``"beta"`` to natural-space standard errors of the same shape as the estimates. Both are ``None`` when the information matrix is not invertible to the stated tolerance — never a number the library does not believe. hessian Observed information ``-d2 ell / d theta2`` in natural space over the free parameters, in :class:`~hawkes_engine._layout.Layout` order. hessian_condition ``cond(H)`` from the singular values, not the ratio of diagonal entries. reliable One of ``"reliable"``, ``"marginal"``, ``"unreliable"``, ``"unknown"``. This states whether the fit converged and is well conditioned. It does **not** state whether the estimator is unbiased or whether an interval covers at the nominal rate; those are properties of the estimator, not of this fit; the pre-registered study measured them (see :func:`hawkes_engine.trust_lookup`). The value comes from :func:`hawkes_engine.report`; ``"unknown"`` is a placeholder that ``fit`` replaces before returning. notes One line per reason a field is ``None``, non-finite, or not to be trusted. Empty when there is nothing to say. data The inputs and derived summaries: ``n_events``, ``T``, ``M``, ``shared_decay``, ``seed``, ``zero_diagonal``, ``n_starts``, ``se_method``, ``residual_gaps_summary``. n_params Number of free parameters. status, nit scipy's termination status and iteration count for the winning start. bound_hit, bound_coordinates Whether a coordinate stopped on a search bound, and which. A fit that did is a boundary value, not a stationary point. theta The log-space optimum over the free parameters. se_source ``"hessian"``, ``"bootstrap"`` or ``"none"``: which method produced ``se``. ``"none"`` means no interval was computed or the calibration map says the estimator is undefined in this regime. Read it before reading ``se``. se_note One sentence saying why ``se_source`` has the value it has. Empty only for a result read from a 0.1.0 document, which did not record one. auto The full :class:`~hawkes_engine.se_policy.AutoDecision` record when ``se_method="auto"``, else ``None``. It names the map cell, the distance, the ``q`` it read, the threshold and the projected cost, so the choice can be audited from the result alone. """ mu: FloatArray alpha: FloatArray beta: FloatArray loglik: float converged: bool status: int nit: int gradient_inf_norm: float n_params: int rho: float rho_ci: tuple[float, float] | None se: dict[str, FloatArray] | None cov: FloatArray | None hessian: FloatArray | None hessian_condition: float | None reliable: str notes: tuple[str, ...] = () data: dict[str, Any] = field(default_factory=dict) bound_hit: bool = False bound_coordinates: tuple[str, ...] = () theta: FloatArray | None = None se_source: str = "hessian" se_note: str = "" auto: Any = None def __repr__(self) -> str: m = self.data.get("M", self.mu.size) conv = "converged" if self.converged else f"not converged (status {self.status})" return ( f"FitResult(M={m}, n_params={self.n_params}, loglik={self.loglik:.6g}, " f"rho={self.rho:.6g}, {conv}, reliable={self.reliable!r})" )
[docs] def to_dict(self) -> JsonObject: """A JSON-serialisable mapping. Stable: see ``SCHEMA_VERSION``.""" return { "schema": SCHEMA_VERSION, "mu": self.mu.tolist(), "alpha": self.alpha.tolist(), "beta": self.beta.tolist(), "loglik": self.loglik, "converged": self.converged, "status": self.status, "nit": self.nit, "gradient_inf_norm": self.gradient_inf_norm, "n_params": self.n_params, "rho": self.rho, "rho_ci": None if self.rho_ci is None else list(self.rho_ci), "se": None if self.se is None else {k: v.tolist() for k, v in self.se.items()}, "cov": None if self.cov is None else self.cov.tolist(), "hessian": None if self.hessian is None else self.hessian.tolist(), "hessian_condition": self.hessian_condition, "reliable": self.reliable, "notes": list(self.notes), "data": _jsonable(self.data), "bound_hit": self.bound_hit, "bound_coordinates": list(self.bound_coordinates), "theta": None if self.theta is None else self.theta.tolist(), "se_source": self.se_source, "se_note": self.se_note, "auto": None if self.auto is None else self.auto.to_dict(), }
[docs] @classmethod def from_dict(cls, payload: JsonObject) -> FitResult: """Rebuild from :meth:`to_dict`. Rejects a schema it does not know.""" schema = payload.get("schema") if schema not in _READABLE_SCHEMAS: raise ValueError( f"FitResult schema {schema!r} is not supported by this version, " f"which reads schema {SCHEMA_VERSION}" ) se = payload["se"] auto = payload.get("auto") if auto is not None: from hawkes_engine.se_policy import AutoDecision auto = AutoDecision.from_dict(dict(auto)) return cls( mu=np.asarray(payload["mu"], dtype=np.float64), alpha=np.asarray(payload["alpha"], dtype=np.float64), beta=np.asarray(payload["beta"], dtype=np.float64), loglik=float(payload["loglik"]), converged=bool(payload["converged"]), status=int(payload["status"]), nit=int(payload["nit"]), gradient_inf_norm=float(payload["gradient_inf_norm"]), n_params=int(payload["n_params"]), rho=float(payload["rho"]), rho_ci=None if payload["rho_ci"] is None else (float(payload["rho_ci"][0]), float(payload["rho_ci"][1])), se=None if se is None else {k: np.asarray(v, dtype=np.float64) for k, v in se.items()}, cov=None if payload["cov"] is None else np.asarray(payload["cov"], dtype=np.float64), hessian=None if payload["hessian"] is None else np.asarray(payload["hessian"], dtype=np.float64), hessian_condition=None if payload["hessian_condition"] is None else float(payload["hessian_condition"]), reliable=str(payload["reliable"]), notes=tuple(str(n) for n in payload["notes"]), data=dict(payload["data"]), bound_hit=bool(payload["bound_hit"]), bound_coordinates=tuple(str(c) for c in payload["bound_coordinates"]), theta=None if payload.get("theta") is None else np.asarray(payload["theta"], dtype=np.float64), se_source=str(payload.get("se_source", "hessian")), se_note=str(payload.get("se_note", "")), auto=auto, )
def _jsonable(value: Any) -> Any: """Recursively convert NumPy scalars and arrays to plain JSON types.""" if isinstance(value, dict): return {str(k): _jsonable(v) for k, v in value.items()} if isinstance(value, list | tuple): return [_jsonable(v) for v in value] if isinstance(value, np.ndarray): return value.tolist() if isinstance(value, np.generic): return value.item() return value