cofi.utils.ReducedLikelihoodManager

class cofi.utils.ReducedLikelihoodManager(fwd_funcs: List[Tuple[Callable, dict]], d_obs_list: List[ndarray], jacobian_fn: Callable, cases: str | List[str] = 'none', Cd_refs: ndarray | List[ndarray] | None = None, kernels=None, copy_G: bool = False, track_stats: bool = False)

Manager for multiple ReducedLikelihood instances with shared Jacobian computation.

This class manages multiple ReducedLikelihood objects and provides efficient evaluation of the combined objective function, gradient, and Hessian by caching Jacobian matrices to avoid redundant computation.

Parameters:

• fwd_funcs (List[Tuple[Callable, dict]]) – List of (forward_function, fwd_kwargs) tuples for each dataset

• d_obs_list (List[np.ndarray]) – List of observed data arrays, one per forward function

• jacobian_fn (Callable) – Function with signature: jacobian_fn(model, n_data, fwd_func, fwd_kwargs) -> np.ndarray Should return Jacobian matrix of shape (n_data, n_params)

• cases (Union[str, List[str]], default='none') – Covariance case(s). Either a single string applied to all, or a list with one case per dataset. Options: ‘none’, ‘scaled’, ‘kernel’, ‘spherical’, ‘diag’, ‘full’

• Cd_refs (Optional[Union[np.ndarray, List[np.ndarray]]], default=None) – Reference covariance matrix(ces). Either a single matrix applied to all, or a list with one per dataset. Required for ‘scaled’ case. For ‘none’ case: defaults to identity matrix if not provided. Not used for ‘spherical’, ‘diag’, or ‘full’ cases.

• kernels (optional) – Kernel instance(s) for ‘kernel’ case. Either a single kernel applied to all datasets using ‘kernel’, or a list with one per dataset. Required for datasets with case='kernel'.

• copy_G (bool, default=False) – If True, copy Jacobian matrices before assigning to ReducedLikelihood.G. Use if jacobian_fn returns views that may be mutated.

• track_stats (bool, default=False) – If True, track statistics on cache hits and Jacobian evaluations

reduced_likelihoods

The managed ReducedLikelihood instances

Type:

List[ReducedLikelihood]

stats

Statistics dictionary (if track_stats=True) with keys: - ‘n_cache_hits’: Number of cache hits - ‘n_jacobian_evals’: Total Jacobian evaluations across all datasets

Type:

dict

Examples

>>> import numpy as np
>>> from cofi.utils import ReducedLikelihoodManager
>>>
>>> # Define forward functions and data
>>> def fwd1(m): return np.array([m[0], m[1]])
>>> def fwd2(m): return np.array([m[0] + m[1]])
>>> fwd_funcs = [(fwd1, {}), (fwd2, {})]
>>> d_obs_list = [np.array([1.0, 2.0]), np.array([3.0])]
>>>
>>> # Define Jacobian function
>>> def jacobian(m, n_data, fwd, fwd_kwargs):
...     # Compute Jacobian (simplified example)
...     if n_data == 2:
...         return np.array([[1, 0], [0, 1]])
...     else:
...         return np.array([[1, 1]])
>>>
>>> # Create manager
>>> manager = ReducedLikelihoodManager(
...     fwd_funcs=fwd_funcs,
...     d_obs_list=d_obs_list,
...     jacobian_fn=jacobian,
...     cases=['spherical', 'spherical'],
...     track_stats=True
... )
>>>
>>> # Evaluate at a model
>>> model = np.array([1.5, 2.5])
>>> obj = manager.objective(model)
>>> grad = manager.gradient(model)
>>> hess = manager.hessian(model)
>>>
>>> # Check statistics
>>> print(manager.stats)

Reference Details

get_ml_covs(model: ndarray) → List[ndarray | None]

Get maximum likelihood covariance estimates for all datasets.

Parameters:

model (np.ndarray) – Model parameters ([m_phys, eta] when using kernel)

Returns:

List of covariance matrices, one per dataset

Return type:

List[Optional[np.ndarray]]

get_stats() → dict

Get performance statistics (if track_stats=True).

Returns:

Statistics dictionary with keys ‘n_cache_hits’ and ‘n_jacobian_evals’

Return type:

dict

Raises:

ValueError – If track_stats was not enabled at initialization

gradient(model: ndarray) → ndarray

Evaluate the gradient of the objective function.

Parameters:

model (np.ndarray) – Model parameters ([m_phys, eta] when using kernel)

Returns:

Gradient vector

Return type:

np.ndarray

hessian(model: ndarray) → ndarray

Evaluate the Hessian of the objective function.

Parameters:

model (np.ndarray) – Model parameters ([m_phys, eta] when using kernel)

Returns:

Hessian matrix

Return type:

np.ndarray

invalidate_cache()

Manually invalidate the Jacobian cache.

This forces recomputation on the next evaluation. Useful if the forward function or Jacobian computation has changed.

objective(model: ndarray) → float

Evaluate the combined objective function (negative log-likelihood).

Parameters:

model (np.ndarray) – Model parameters ([m_phys, eta] when using kernel)

Returns:

Negative log-likelihood value

Return type:

float

back to top