Latent-class conditional logit

The latent-class estimator fits a finite mixture of conditional logits by expectation-maximization. Class membership may be represented by aggregate shares or modeled as a function of panel characteristics with fractional multinomial logit.

Most users should start with LCLSpec and lcl.fit. The lower-level class remains available for direct use.

import lcl
from lcl import FitOptions, Options

options = Options(fit=FitOptions(starts=3))
results = lcl.fit(data, spec, options=options)

# Lower-level orchestration when a persistent model object is useful:
model = lcl.LatentClassConditionalLogit(spec=spec)
results = model.fit(data, options=options)

Pass spec by keyword to the lower-level constructor.

Model

lcl.LatentClassConditionalLogit(num_classes=5, numeraire=None, *, spec=None, numeraire_min_abs=DEFAULT_NEGATIVE_MIN_ABS)

Bases: ChoiceModel

Specification and estimation for latent-class conditional logit models.

This class provides the interface for defining and fitting a latent-class conditional logit model using an Expectation-Maximization (EM) algorithm. It inherits from the abstract base class ChoiceModel and manages the data ingestion, initialization, and iterative optimization of latent taste parameters and class membership probabilities.

Parameters:

Name	Type	Description	Default
num_classes	int	The number of discrete latent classes to estimate.	5
numeraire	str | None	The name of the variable to be used as the numeraire (e.g., price or cost). If specified, its taste parameter is mathematically constrained to be strictly negative across all latent classes via a softplus transformation to ensure theoretically consistent willingness-to-pay calculations.	None

Attributes:

Name	Type	Description
num_classes	int	The number of discrete latent classes.
numeraire	str | None	The name of the numeraire variable.
numeraire_idx	int | None	The column index of the numeraire variable in the expanded design matrix, resolved during the fit method.
num_vars	int	The total number of alternative-specific variables (taste parameters), resolved during the fit method.
num_dem_vars	int	The total number of demographic variables, resolved during the fit method.

Create an unfitted latent-class conditional-logit model specification.

Source code in src/lcl/latent_class_conditional_logit.py

def __init__(
    self,
    num_classes: int = 5,
    numeraire: str | None = None,
    *,
    spec: LCLSpec | None = None,
    numeraire_min_abs: float = DEFAULT_NEGATIVE_MIN_ABS,
) -> None:
    """Create an unfitted latent-class conditional-logit model specification."""
    super().__init__()
    if spec is not None:
        num_classes = spec.classes
        if (
            numeraire is not None
            and spec.numeraire is not None
            and numeraire != spec.numeraire
        ):
            raise ValueError(
                "numeraire conflicts with the negative constraint in spec."
            )
        numeraire = numeraire or spec.numeraire
        numeraire_min_abs = spec.numeraire_min_abs

    self.spec = spec
    self.num_classes = num_classes
    self.numeraire = numeraire
    self.numeraire_min_abs = numeraire_min_abs
    self.numeraire_idx: int | None = None

fit(data, alts_col=None, cases_col=None, panels_col=None, utility_formula=None, membership_formula=None, choice_col=None, case_varnames=None, dem_varnames=None, variable_labels=None, dems_data=None, options=None, fit_options=None, optimization_options=None, inference=None, diagnostics=None, progress_callback=None)

Fit the latent-class conditional logit model using an EM algorithm.

This method ingests raw data, translates it into strictly contiguous, zero-indexed JAX arrays (PyTrees), and executes the hardware-accelerated EM optimization routine.

Parameters:

Name	Type	Description	Default
data	Any	The main dataset containing choice situations. Accepts a Polars DataFrame, Pandas DataFrame, or dictionary of arrays.	required
alts_col	str	The name of the column identifying specific alternatives within a choice situation.	None
cases_col	str	The name of the column grouping observations into distinct choice situations.	None
panels_col	str	The name of the column mapping choice situations to specific decision-makers (panels).	None
utility_formula	str | None	Formulaic string for the alternative-specific utility specification. Examples include "choice ~ cost + time + C(mode)" or, when choice_col supplies the outcome, "~ cost + time + C(mode)".	None
membership_formula	str | None	Right-hand-side Formulaic string for class-membership demographics, for example "~ income + C(segment)". A left-hand side is not accepted because latent class labels are unobserved.	None
choice_col	str | None	The name of the boolean or binary column indicating chosen alternatives. Required when utility_formula has no left-hand side.	None
case_varnames	Sequence[str] | None	A list of alternative-specific variables to include in the utility specification. Required if utility_formula is not provided.	None
dem_varnames	Sequence[str] | None	A list of demographic variables used to predict latent class membership.	None
variable_labels	Mapping[str, str] | None	Optional mapping from raw DataFrame/model variable names to human-readable labels used in presentation tables. Labels do not change model specification, constraints, prediction inputs, or WTP request names.	None
dems_data	Any | None	An optional, separate panel-level dataset containing demographics. If provided, it will be merged with the main data on panels_col.	None
fit_options	FitOptions | None	Preferred EM settings, including multi-start orchestration.	None
optimization_options	OptimizationOptions | None	Preferred exact-Newton M-step settings.	None
inference	InferenceOptions | None	Preferred covariance and standard-error settings.	None
diagnostics	DiagnosticsOptions | None	Diagnostic thresholds and switches.	None
progress_callback	callable | None	Receives structured hardware, start, EM-step, and completion events.	None

Notes

Case or panel weights are not supported for latent-class estimation and this method takes no weights argument; passing one is a :class:TypeError rather than a silent no-op. Weighted estimation is available for :class:~lcl.conditional_logit.ConditionalLogit.

Returns:

Type	Description
class:`~lcl._results.LCLResults`	A container holding the estimated parameters, optimization metadata, information criteria, and methods for inference (standard errors, predictions).

Raises:

Type	Description
ValueError	If a numeraire was specified during class instantiation but cannot be found in the expanded design matrix columns.

Source code in src/lcl/latent_class_conditional_logit.py

def fit(
    self,
    data: Any,
    alts_col: str | None = None,
    cases_col: str | None = None,
    panels_col: str | None = None,
    utility_formula: str | None = None,
    membership_formula: str | None = None,
    choice_col: str | None = None,
    case_varnames: Sequence[str] | None = None,
    dem_varnames: Sequence[str] | None = None,
    variable_labels: Mapping[str, str] | None = None,
    dems_data: Any | None = None,
    options: Options | None = None,
    fit_options: FitOptions | None = None,
    optimization_options: OptimizationOptions | None = None,
    inference: InferenceOptions | None = None,
    diagnostics: DiagnosticsOptions | None = None,
    progress_callback: Callable[[dict[str, Any]], None] | None = None,
) -> LCLResults:
    """Fit the latent-class conditional logit model using an EM algorithm.

    This method ingests raw data, translates it into strictly contiguous,
    zero-indexed JAX arrays (PyTrees), and executes the hardware-accelerated
    EM optimization routine.

    Parameters
    ----------
    data : Any
        The main dataset containing choice situations. Accepts a Polars DataFrame,
        Pandas DataFrame, or dictionary of arrays.
    alts_col : str
        The name of the column identifying specific alternatives within a choice
        situation.
    cases_col : str
        The name of the column grouping observations into distinct choice
        situations.
    panels_col : str
        The name of the column mapping choice situations to specific
        decision-makers (panels).
    utility_formula : str | None, default=None
        Formulaic string for the alternative-specific utility specification.
        Examples include ``"choice ~ cost + time + C(mode)"`` or, when
        ``choice_col`` supplies the outcome, ``"~ cost + time + C(mode)"``.
    membership_formula : str | None, default=None
        Right-hand-side Formulaic string for class-membership demographics,
        for example ``"~ income + C(segment)"``.  A left-hand side is not
        accepted because latent class labels are unobserved.
    choice_col : str | None, default=None
        The name of the boolean or binary column indicating chosen alternatives.
        Required when ``utility_formula`` has no left-hand side.
    case_varnames : Sequence[str] | None, default=None
        A list of alternative-specific variables to include in the utility
        specification. Required if ``utility_formula`` is not provided.
    dem_varnames : Sequence[str] | None, default=None
        A list of demographic variables used to predict latent class membership.
    variable_labels : Mapping[str, str] | None, default=None
        Optional mapping from raw DataFrame/model variable names to
        human-readable labels used in presentation tables.  Labels do not
        change model specification, constraints, prediction inputs, or WTP
        request names.
    dems_data : Any | None, default=None
        An optional, separate panel-level dataset containing demographics. If
        provided, it will be merged with the main `data` on `panels_col`.
    fit_options : FitOptions | None, optional
        Preferred EM settings, including multi-start orchestration.
    optimization_options : OptimizationOptions | None, optional
        Preferred exact-Newton M-step settings.
    inference : InferenceOptions | None, optional
        Preferred covariance and standard-error settings.
    diagnostics : DiagnosticsOptions | None, optional
        Diagnostic thresholds and switches.
    progress_callback : callable | None, optional
        Receives structured hardware, start, EM-step, and completion events.

    Notes
    -----
    Case or panel weights are not supported for latent-class estimation and
    this method takes no ``weights`` argument; passing one is a
    :class:`TypeError` rather than a silent no-op.  Weighted estimation is
    available for :class:`~lcl.conditional_logit.ConditionalLogit`.

    Returns
    -------
    :class:`~lcl._results.LCLResults`
        A container holding the estimated parameters, optimization metadata,
        information criteria, and methods for inference (standard errors,
        predictions).

    Raises
    ------
    ValueError
        If a `numeraire` was specified during class instantiation but cannot be
        found in the expanded design matrix columns.
    """
    self.spec = resolve_lcl_spec(
        spec=self.spec,
        alts_col=alts_col,
        cases_col=cases_col,
        panels_col=panels_col,
        choice_col=choice_col,
        case_varnames=case_varnames,
        dem_varnames=dem_varnames,
        utility_formula=utility_formula,
        membership_formula=membership_formula,
        classes=self.num_classes,
        numeraire=self.numeraire,
        numeraire_min_abs=(
            self.numeraire_min_abs if self.numeraire is not None else None
        ),
        variable_labels=variable_labels,
    )
    alts_col = self.spec.ids.alt
    cases_col = self.spec.ids.case
    panels_col = self.spec.ids.panel
    choice_col = self.spec.ids.choice
    utility_formula = self.spec.utility_formula
    membership_formula = self.spec.membership_formula
    case_varnames = self.spec.utility
    dem_varnames = self.spec.membership
    variable_labels = self.spec.variable_labels
    self.num_classes = self.spec.classes
    self.numeraire = self.spec.numeraire
    self.numeraire_min_abs = self.spec.numeraire_min_abs

    resolved_options = _resolve_options(
        options,
        fit_options=fit_options,
        optimization_options=optimization_options,
        inference=inference,
        diagnostics=diagnostics,
    )
    fit_options = resolved_options.fit
    optimization_options = resolved_options.optimization
    inference = resolved_options.inference
    diagnostics = resolved_options.diagnostics

    parsed_data = self._ingest_data(
        data=data,
        alts_col=alts_col,
        cases_col=cases_col,
        panels_col=panels_col,
        utility_formula=utility_formula,
        membership_formula=membership_formula,
        choice_col=choice_col,
        case_varnames=case_varnames,
        dem_varnames=dem_varnames,
        dems_data=dems_data,
    )

    self._pre_fit(
        parsed_data.case_varnames,
        parsed_data.dem_varnames,
        self.numeraire,
        variable_labels=variable_labels,
    )
    self.num_vars = len(self.case_varnames)
    self.num_dem_vars = len(self.dem_varnames) if self.dem_varnames else 0

    if self.numeraire:
        try:
            self.numeraire_idx = self.case_varnames.index(self.numeraire)
        except ValueError:
            raise ValueError(
                f"Numeraire '{self.numeraire}' not found in expanded design matrix."
            )
    else:
        self.numeraire_idx = None

    data_struct, _, _ = self._setup_data(parsed_data)
    if data_struct.num_panels is None:
        raise ValueError("panels_col is required for latent-class models.")
    if self.num_classes > data_struct.num_panels:
        raise ValueError("num_classes cannot exceed the number of panels.")
    diff_unchosen_chosen = _diff_unchosen_chosen(data_struct)
    packing = ParamPacking(
        num_alt_vars=self.num_vars,
        num_classes=self.num_classes,
        num_dem_vars=self.num_dem_vars,
        numeraire_idx=self.numeraire_idx,
        numeraire_min_abs=self.numeraire_min_abs,
    )

    # Resolve a coarser clustering once, in encoded panel order, so every
    # start shares it and the results object never re-reads the raw frame.
    cluster_ids: onp.ndarray | None = None
    num_clusters: int | None = None
    cluster_column = inference.cluster_column
    if cluster_column is not None:
        cluster_ids, num_clusters = self._resolve_panel_cluster_ids(
            data, parsed_data, cluster_column, panels_col=panels_col
        )
        logger.info(
            "Clustering standard errors on %r: %s groups across %s panels.",
            cluster_column,
            num_clusters,
            data_struct.num_panels,
        )

    num_devices = fit_options.num_devices
    if num_devices > 1:
        if self.num_classes % num_devices == 0:
            message = f"Distributing {self.num_classes} classes across {num_devices} devices."
        else:
            message = f"Found {num_devices} devices; padding classes for balanced sharding."
    else:
        message = "Running beta updates on a single device."
    logger.info(message)
    if progress_callback is not None:
        progress_callback({"event": "hardware", "message": message})

    # Independent starts share one ingested dataset and one compiled EM step,
    # and the winner is kept outright rather than refit from its seed.
    best_run: _EMRun | None = None
    failures: list[str] = []
    for start_index in range(fit_options.starts):
        seed = fit_options.seed + start_index
        if progress_callback is not None:
            progress_callback(
                {
                    "event": "start",
                    "start": start_index + 1,
                    "starts": fit_options.starts,
                    "seed": seed,
                }
            )
        start_options = replace(fit_options, seed=seed)
        if fit_options.starts == 1:
            # A single start has nothing to fall back on, so let the original
            # exception and its traceback reach the caller unwrapped.
            run = self._run_em(
                diff_unchosen_chosen=diff_unchosen_chosen,
                data_struct=data_struct,
                fit_options=start_options,
                optimization_options=optimization_options,
                progress_callback=progress_callback,
            )
        else:
            try:
                run = self._run_em(
                    diff_unchosen_chosen=diff_unchosen_chosen,
                    data_struct=data_struct,
                    fit_options=start_options,
                    optimization_options=optimization_options,
                    progress_callback=progress_callback,
                )
            except Exception as exc:  # noqa: BLE001 - reported to the caller
                failures.append(f"seed {seed}: {exc}")
                logger.warning("LCL start with seed %s failed: %s", seed, exc)
                continue
        if best_run is None or run.loglik > best_run.loglik:
            best_run = run

    if best_run is None:
        detail = "; ".join(failures)
        raise RuntimeError(f"All {fit_options.starts} EM starts failed: {detail}")
    if fit_options.starts > 1:
        logger.info(
            "Selected EM start seed %s with log likelihood %.6f.",
            best_run.seed,
            best_run.loglik,
        )

    em_vars = best_run.em_vars
    em_history_rows = best_run.history
    em_recursion = best_run.recursions

    # Observed-data Newton polish.  EM is linearly convergent, so it stops
    # short of a stationary point; the observed information and the sandwich
    # covariance both assume the score vanishes at the reported estimate.
    if em_vars.latent_betas is None or em_vars.shares is None:
        raise RuntimeError("The EM run returned an incomplete parameter state.")
    flat_params = packing.pack(
        em_vars.latent_betas, em_vars.thetas, em_vars.shares
    )
    polish_report: PolishReport | None = None
    if fit_options.polish:
        if progress_callback is not None:
            progress_callback({"event": "polish", "iterations": None})
        polished, polish_report = polish_observed_data(
            flat_params,
            diff_unchosen_chosen,
            data_struct,
            packing,
            maxiter=fit_options.polish_maxiter,
            max_step_norm=optimization_options.max_step_norm,
            line_search_maxiter=optimization_options.line_search_maxiter,
        )
        if polish_report.accepted:
            em_vars = em_vars_from_flat(
                polished, diff_unchosen_chosen, data_struct, packing
            )
        score_max = polish_report.score_after
        if progress_callback is not None:
            progress_callback(
                {
                    "event": "polish",
                    "iterations": polish_report.iterations,
                    "score_before": polish_report.score_before,
                    "score_after": polish_report.score_after,
                }
            )
    else:
        score_max = observed_score_max(
            flat_params, diff_unchosen_chosen, data_struct, packing
        )

    # A fit has converged when the observed-data score has actually vanished.
    # Reporting convergence from a log-likelihood change instead lets a
    # slowly crawling EM claim an optimum it has not reached.
    converged = bool(score_max <= fit_options.score_tol)
    if not converged:
        logger.warning(
            "The maximum absolute observed-data score is %.3e, above the "
            "tolerance %.3g, so the estimate is not a stationary point of the "
            "mixture likelihood. Standard errors assume it is. Consider "
            "raising max_em_iter or polish_maxiter.",
            score_max,
            fit_options.score_tol,
        )

    em_vars, class_permutation = _canonicalize_classes(em_vars)
    em_history_rows = _permute_em_history(em_history_rows, class_permutation)
    final_em_iter = max(em_recursion - 1, 0)
    optimization_history_rows = self._optimizer_snapshot(
        em_vars, diff_unchosen_chosen, data_struct, final_em_iter
    )

    estim_time_sec = time() - self._fit_start_time

    logger.info("Estimation time: %.3f seconds", estim_time_sec)
    if progress_callback is not None:
        progress_callback(
            {"event": "complete", "estimation_time_seconds": estim_time_sec}
        )

    return LCLResults(
        model_spec=self,
        em_vars=em_vars,
        estimation_data=data_struct,
        em_recursion=em_recursion,
        converged=converged,
        inference=inference,
        diagnostics_config=diagnostics,
        estim_time_sec=estim_time_sec,
        em_history=em_history_rows,
        optimization_history=optimization_history_rows,
        observed_score_max=score_max,
        score_tol=fit_options.score_tol,
        em_criterion_met=best_run.criterion_met,
        polish_report=polish_report,
        cluster_ids=cluster_ids,
        num_clusters=num_clusters,
        param_packing=packing,
    )

Results

lcl.results.LCLResults(model_spec, em_vars, estimation_data, em_recursion, converged, inference, estim_time_sec, diagnostics_config=None, em_history=None, optimization_history=None, observed_score_max=float('nan'), score_tol=0.0001, em_criterion_met=False, polish_report=None, cluster_ids=None, num_clusters=None, param_packing=None)

Post-estimation results and inference container.

Computes robust sandwich covariance matrices (clustered at the decision-maker level) and handles the extraction of population-level moments via the Delta Method.

Attributes:

Name	Type	Description
cov_matrix	Float64[Array, 'all_params all_params']	Covariance of the reported (structural) parameters, aligned row for row with :meth:parameter_names. Constrained coefficients appear on the scale printed by :meth:class_coefficients, so sqrt(diag(cov_matrix)) matches the published standard errors. Clustered covariance uses the Stata maximum-likelihood multiplier :math:(G / (G - 1)).
latent_cov_matrix	Float64[Array, 'all_params all_params']	Covariance in the unconstrained parameterization the optimizer works in. This is the matrix the delta method consumes: target functions apply the softplus transform internally, so its Jacobian is differentiated rather than applied twice.
caic	float	Consistent Akaike Information Criterion (Bozdogan, 1987).
bic	float	Bayesian Information Criterion (Schwarz, 1978).
adjusted_bic	float	Sample-size adjusted BIC (Sclove, 1987).

Build a latent-class results object and compute inference artifacts.

Parameters:

Name	Type	Description	Default
model_spec	Any	Fitted model specification. Kept broad to avoid a runtime circular import with :class:~lcl.latent_class_conditional_logit.LatentClassConditionalLogit.	required
em_vars	:class:`~lcl._struct.EMVars`	Final EM state containing parameters, probabilities, and log likelihood.	required
estimation_data	:class:`~lcl._struct.Data`	Encoded estimation data.	required
em_recursion	int	Number of EM recursions completed before termination.	required
converged	bool	Whether the explicit EM stopping criterion was satisfied.	required
inference	:class:`~lcl.options.InferenceOptions` | None	Covariance and standard-error configuration.	required
estim_time_sec	float	Wall-clock estimation time in seconds.	required
diagnostics_config	:class:`~lcl.options.DiagnosticsOptions` | None	Thresholds and switches for public diagnostics.	None
em_history	list[dict[str, Any]] | None	EM log-likelihood and class-share history.	None
optimization_history	list[dict[str, Any]] | None	Final class-level M-step diagnostics.	None
observed_score_max	float	Largest absolute component of the observed-data score at the reported estimate. Recomputed here when covariance estimation runs.	float('nan')
score_tol	float	Stationarity tolerance used for the converged flag, reused by :meth:diagnostics so the two can never disagree.	1e-4
em_criterion_met	bool	Whether the Aitken EM stopping criterion was satisfied before the iteration cap.	False
polish_report	:class:`~lcl._polish.PolishReport` | None	Outcome of the observed-data Newton polish.	None
cluster_ids	ArrayLike | None	Zero-indexed cluster identifier per panel, for clustering coarser than the decision-maker.	None
num_clusters	int | None	Number of distinct clusters implied by cluster_ids.	None
param_packing	:class:`~lcl._params.ParamPacking` | None	Reuse of the packing built during estimation.	None

Source code in src/lcl/_results.py

def __init__(
    self,
    model_spec: Any,
    em_vars: EMVars,
    estimation_data: Data,
    em_recursion: int,
    converged: bool,
    inference: InferenceOptions | None,
    estim_time_sec: float,
    diagnostics_config: DiagnosticsOptions | None = None,
    em_history: list[dict[str, Any]] | None = None,
    optimization_history: list[dict[str, Any]] | None = None,
    observed_score_max: float = float("nan"),
    score_tol: float = 1e-4,
    em_criterion_met: bool = False,
    polish_report: PolishReport | None = None,
    cluster_ids: ArrayLike | None = None,
    num_clusters: int | None = None,
    param_packing: ParamPacking | None = None,
) -> None:
    """Build a latent-class results object and compute inference artifacts.

    Parameters
    ----------
    model_spec : Any
        Fitted model specification. Kept broad to avoid a runtime circular import
        with :class:`~lcl.latent_class_conditional_logit.LatentClassConditionalLogit`.
    em_vars : :class:`~lcl._struct.EMVars`
        Final EM state containing parameters, probabilities, and log likelihood.
    estimation_data : :class:`~lcl._struct.Data`
        Encoded estimation data.
    em_recursion : int
        Number of EM recursions completed before termination.
    converged : bool
        Whether the explicit EM stopping criterion was satisfied.
    inference : :class:`~lcl.options.InferenceOptions` | None
        Covariance and standard-error configuration.
    estim_time_sec : float
        Wall-clock estimation time in seconds.
    diagnostics_config : :class:`~lcl.options.DiagnosticsOptions` | None
        Thresholds and switches for public diagnostics.
    em_history : list[dict[str, Any]] | None
        EM log-likelihood and class-share history.
    optimization_history : list[dict[str, Any]] | None
        Final class-level M-step diagnostics.
    observed_score_max : float, optional
        Largest absolute component of the observed-data score at the reported
        estimate.  Recomputed here when covariance estimation runs.
    score_tol : float, default=1e-4
        Stationarity tolerance used for the ``converged`` flag, reused by
        :meth:`diagnostics` so the two can never disagree.
    em_criterion_met : bool, default=False
        Whether the Aitken EM stopping criterion was satisfied before the
        iteration cap.
    polish_report : :class:`~lcl._polish.PolishReport` | None, optional
        Outcome of the observed-data Newton polish.
    cluster_ids : ArrayLike | None, optional
        Zero-indexed cluster identifier per panel, for clustering coarser than
        the decision-maker.
    num_clusters : int | None, optional
        Number of distinct clusters implied by ``cluster_ids``.
    param_packing : :class:`~lcl._params.ParamPacking` | None, optional
        Reuse of the packing built during estimation.
    """
    self.model = model_spec
    self.em_res = em_vars
    self.data = estimation_data
    self.total_recursions = em_recursion
    self.converged = converged
    self.estim_time_sec = estim_time_sec
    self.inference = inference if inference is not None else InferenceOptions()
    self.diagnostics_config = (
        diagnostics_config
        if diagnostics_config is not None
        else DiagnosticsOptions()
    )
    self.em_history_ = _history_frame(em_history)
    self.optimization_history_ = _history_frame(optimization_history)
    if self.em_res.latent_betas is None:
        raise ValueError("Latent betas are required to construct LCL results.")
    if self.em_res.structural_betas is None:
        raise ValueError("Structural betas are required to construct LCL results.")
    if self.em_res.shares is None:
        raise ValueError("Class shares are required to construct LCL results.")
    if self.data.num_panels is None:
        raise ValueError("Panel identifiers are required for LCL results.")

    self._param_packing = param_packing or ParamPacking(
        num_alt_vars=self.model.num_vars,
        num_classes=self.model.num_classes,
        num_dem_vars=self.model.num_dem_vars,
        numeraire_idx=self.model.numeraire_idx,
        numeraire_min_abs=self.model.numeraire_min_abs,
    )
    self.flat_params = self._pack_params()
    self.num_params = self._param_packing.num_params
    self.score_tol = float(score_tol)
    self.em_criterion_met = bool(em_criterion_met)
    self.polish_report = polish_report
    self._cluster_ids = (
        None if cluster_ids is None else jnp.asarray(cluster_ids, dtype=jnp.int32)
    )
    self._num_clusters = num_clusters
    # Populated by _compute_covariance; stays None when inference is skipped.
    self.information_diagnostics: Any = None
    self.observed_score_max = float(observed_score_max)
    self.latent_cov_matrix = self._compute_covariance()
    self.cov_matrix = self._structural_covariance(self.latent_cov_matrix)

    # Compute information criteria
    num_panels = self.data.num_panels
    self.aic = 2 * self.num_params - 2 * self.em_res.unconditional_loglik
    self.aic3 = 3 * self.num_params - 2 * self.em_res.unconditional_loglik
    self.caic = (
        jnp.log(num_panels) + 1
    ) * self.num_params - 2 * self.em_res.unconditional_loglik
    self.bic = (
        jnp.log(num_panels) * self.num_params - 2 * self.em_res.unconditional_loglik
    )
    self.adjusted_bic = (
        jnp.log((num_panels + 2) / 24) * self.num_params
        - 2 * self.em_res.unconditional_loglik
    )
    logger.info(
        "Information criteria: CAIC=%.1f, BIC=%.1f, adjusted BIC=%.1f",
        self.caic,
        self.bic,
        self.adjusted_bic,
    )

    if not self.converged:
        logger.warning(
            "Optimization did not converge after %s iterations.",
            self.total_recursions,
        )

abic property

Deprecated alias for :attr:adjusted_bic.

convergence property

Deprecated alias for :attr:converged.

covariance property

Deprecated alias for :attr:cov_matrix.

covariance_available property

Report whether a usable covariance matrix was estimated.

__repr__()

Return a compact, human-readable summary of fit quality.

Source code in src/lcl/_results.py

def __repr__(self) -> str:
    """Return a compact, human-readable summary of fit quality."""
    status = "Converged" if self.converged else "Did Not Converge"
    parts = [
        f"<LCLResults: {self.model.num_classes} Classes",
        f"{status}",
        f"Log likelihood: {self.em_res.unconditional_loglik:.1f}",
        f"CAIC: {self.caic:.1f}",
        f"BIC: {self.bic:.1f}",
        f"Adj. BIC: {self.adjusted_bic:.1f}",
    ]
    if not self.inference.skip and not self.covariance_available:
        parts.append("Covariance unavailable")
    return " | ".join(parts) + ">"

audit_report()

Return a text audit report for replication materials.

Source code in src/lcl/_results.py

def audit_report(self) -> str:
    """Return a text audit report for replication materials."""
    diagnostics_table = self.diagnostics().to_frame()
    return "\n\n".join(
        [
            "1. Model Specification\n" + self.spec_summary(),
            "2. Fit Statistics\n"
            + "\n".join(
                [
                    f"Log likelihood: {float(self.em_res.unconditional_loglik):.6g}",
                    f"CAIC: {float(self.caic):.6g}",
                    f"BIC: {float(self.bic):.6g}",
                    f"Adjusted BIC: {float(self.adjusted_bic):.6g}",
                    f"Estimation seconds: {self.estim_time_sec:.3f}",
                ]
            ),
            "3. Class Shares\n" + str(self.class_shares()),
            "4. Diagnostics\n" + str(diagnostics_table),
        ]
    )

beta_summary()

Return population-level coefficient moments with Delta-method SEs.

Returns:

Type	Description
DataFrame	Raw variables, display labels, mean coefficients, standard deviations across classes, Delta-method standard errors, and class-specific extrema.

Source code in src/lcl/_results.py

def beta_summary(self) -> pl.DataFrame:
    """Return population-level coefficient moments with Delta-method SEs.

    Returns
    -------
    pl.DataFrame
        Raw variables, display labels, mean coefficients, standard deviations
        across classes, Delta-method standard errors, and class-specific extrema.
    """
    if self.data.num_panels is None:
        raise ValueError("Panel identifiers are required to summarize LCL results.")

    means, se_means = self._apply_delta_method(
        self._calc_population_mean_betas,
        self.flat_params,
        dems=self.data.dems,
        num_panels=self.data.num_panels,
    )
    variances, se_variances = self._apply_delta_method(
        self._calc_population_var_betas,
        self.flat_params,
        dems=self.data.dems,
        num_panels=self.data.num_panels,
    )
    structural = onp.asarray(self.em_res.structural_betas)

    # sd = sqrt(var), so se(sd) = se(var) / (2 sd) -- but only where the
    # variance is separated from zero.  A variable whose coefficient is
    # common to every class has no identified spread, and the derivative of
    # the square root there is unbounded, so the standard error is NaN rather
    # than the zero a floored square root would silently report.
    variance_array = onp.asarray(variances, dtype=onp.float64)
    se_variance_array = onp.asarray(se_variances, dtype=onp.float64)
    scale = onp.maximum(onp.max(structural**2, axis=1), 1.0)
    identified = variance_array > DEGENERATE_VARIANCE_RTOL * scale
    stds = onp.sqrt(onp.maximum(variance_array, 0.0))
    with onp.errstate(divide="ignore", invalid="ignore"):
        se_stds = onp.where(
            identified, se_variance_array / (2.0 * stds), onp.nan
        )
    if not bool(onp.all(identified)):
        degenerate = [
            variable
            for variable, keep in zip(self.model.case_varnames, identified)
            if not keep
        ]
        logger.warning(
            "Between-class spread is not identified for %s: the coefficient "
            "is common to every class, so its standard deviation has no "
            "standard error.",
            ", ".join(degenerate),
        )
    rows = []
    for idx, variable in enumerate(self.model.case_varnames):
        rows.append(
            {
                "variable": variable,
                "label": _model_variable_label(self.model, variable),
                "mean": float(means[idx]),
                "mean_se": float(se_means[idx]),
                "sd": float(stds[idx]),
                "sd_se": float(se_stds[idx]),
                "min_class": float(onp.min(structural[idx, :])),
                "max_class": float(onp.max(structural[idx, :])),
            }
        )
    return pl.DataFrame(rows)

class_coefficients()

Return class-specific structural coefficients.

Returns:

Type	Description
DataFrame	Long-format table with one row per variable and latent class. The variable column preserves raw model names; label contains human-readable presentation labels.

Source code in src/lcl/_results.py

def class_coefficients(self) -> pl.DataFrame:
    """Return class-specific structural coefficients.

    Returns
    -------
    pl.DataFrame
        Long-format table with one row per variable and latent class.  The
        ``variable`` column preserves raw model names; ``label`` contains
        human-readable presentation labels.
    """
    structural_betas, standard_errors = self._apply_delta_method(
        self._calc_structural_betas, self.flat_params
    )
    rows = []
    beta_array = onp.asarray(structural_betas)
    se_array = onp.asarray(standard_errors)
    for var_idx, variable in enumerate(self.model.case_varnames):
        for class_idx in range(self.model.num_classes):
            rows.append(
                {
                    "variable": variable,
                    "label": _model_variable_label(self.model, variable),
                    "class": class_idx,
                    "coefficient": float(beta_array[var_idx, class_idx]),
                    "std_error": float(se_array[var_idx, class_idx]),
                    "constrained": variable == self.model.numeraire,
                }
            )
    return pl.DataFrame(rows)

class_shares()

Return aggregate latent-class shares.

Returns:

Type	Description
DataFrame	One row per latent class with aggregate class share and effective panel mass.

Source code in src/lcl/_results.py

def class_shares(self) -> pl.DataFrame:
    """Return aggregate latent-class shares.

    Returns
    -------
    pl.DataFrame
        One row per latent class with aggregate class share and effective
        panel mass.
    """
    if self.data.num_panels is None:
        raise ValueError("Panel identifiers are required for class shares.")
    shares, share_se = self._apply_delta_method(
        self._calc_class_shares,
        self.flat_params,
        dems=self.data.dems,
        num_panels=self.data.num_panels,
    )
    shares_array = onp.asarray(shares)
    share_se_array = onp.asarray(share_se)
    rows = []
    posterior = self.em_res.class_probs_by_panel
    posterior_arr = onp.asarray(posterior) if posterior is not None else None
    for class_idx, share in enumerate(shares_array):
        row = {
            "class": class_idx,
            "share": float(share),
            "std_error": float(share_se_array[class_idx]),
        }
        if posterior_arr is not None:
            row["effective_panels"] = float(posterior_arr[:, class_idx].sum())
        rows.append(row)
    return pl.DataFrame(rows)

classification_diagnostics()

Summarize posterior separation and modal classification by class.

Source code in src/lcl/_results.py

def classification_diagnostics(self) -> pl.DataFrame:
    """Summarize posterior separation and modal classification by class."""
    posterior = self.em_res.class_probs_by_panel
    if posterior is None:
        raise ValueError("Posterior class probabilities are required.")
    probabilities = onp.asarray(posterior, dtype=onp.float64)
    modal = onp.argmax(probabilities, axis=1)
    entropy = -onp.sum(probabilities * onp.log(onp.maximum(probabilities, 1e-300)))
    entropy_r2 = 1.0 - entropy / (
        probabilities.shape[0] * onp.log(self.model.num_classes)
    )
    prior_shares = probabilities.mean(axis=0)
    rows = []
    for class_idx in range(self.model.num_classes):
        selected = modal == class_idx
        modal_count = int(selected.sum())
        average_posterior = (
            float(probabilities[selected, class_idx].mean())
            if modal_count
            else float("nan")
        )
        prior = float(prior_shares[class_idx])
        if modal_count and 0.0 < average_posterior < 1.0 and 0.0 < prior < 1.0:
            occ = (average_posterior / (1.0 - average_posterior)) / (
                prior / (1.0 - prior)
            )
        else:
            occ = float("nan")
        rows.append(
            {
                "class": class_idx,
                "modal_panels": modal_count,
                "modal_share": modal_count / probabilities.shape[0],
                "average_posterior": average_posterior,
                "odds_correct_classification": occ,
                "entropy_r2": float(entropy_r2),
            }
        )
    return pl.DataFrame(rows)

convergence_report()

Return a compact convergence and diagnostic report.

Source code in src/lcl/_results.py

def convergence_report(self) -> str:
    """Return a compact convergence and diagnostic report."""
    diagnostics = self.diagnostics().to_frame()
    warnings = diagnostics.filter(pl.col("status") != "ok")
    lines = [
        f"Converged: {self.converged}",
        f"EM recursions: {self.total_recursions}",
        f"EM criterion met: {self.em_criterion_met}",
        f"Final log likelihood: {float(self.em_res.unconditional_loglik):.6g}",
        f"Max observed-data score: {self.observed_score_max:.3e} "
        f"(tolerance {self.score_tol:.3g})",
        f"Warnings: {warnings.height}",
    ]
    if self.polish_report is not None:
        report = self.polish_report
        lines.append(
            f"Observed-data polish: {report.iterations} Newton step(s), "
            f"log likelihood {report.loglik_before:.6f} -> "
            f"{report.loglik_after:.6f}, score {report.score_before:.3e} -> "
            f"{report.score_after:.3e}"
        )
    if self.em_history_.height:
        last = self.em_history_.tail(1).row(0, named=True)
        lines.append(f"Last EM history row: {last}")
    return "\n".join(lines)

diagnose()

Alias for :meth:diagnostics.

Source code in src/lcl/_results.py

def diagnose(self) -> LCLDiagnostics:
    """Alias for :meth:`diagnostics`."""
    return self.diagnostics()

diagnostics()

Return structured model diagnostics.

Source code in src/lcl/_results.py

def diagnostics(self) -> LCLDiagnostics:
    """Return structured model diagnostics."""
    rows: list[dict[str, object]] = [
        {
            "section": "fit",
            "check": "converged",
            "value": bool(self.converged),
            "status": "ok" if self.converged else "warning",
            "message": (
                "Whether the estimate is a stationary point of the mixture "
                "likelihood, judged by observed_score_max against score_tol."
            ),
        },
        {
            "section": "fit",
            "check": "log_likelihood",
            "value": float(self.em_res.unconditional_loglik),
            "status": "ok",
            "message": "Final unconditional log likelihood.",
        },
        {
            "section": "fit",
            "check": "observed_score_max",
            "value": self.observed_score_max,
            "status": (
                "warning"
                if onp.isfinite(self.observed_score_max)
                and self.observed_score_max > self.score_tol
                else "ok"
            ),
            "message": (
                "Maximum absolute component of the final observed-data score, "
                f"against the score_tol of {self.score_tol:.3g}. This is the "
                "same test that sets the converged flag, so the two agree by "
                "construction."
            ),
        },
        {
            "section": "fit",
            "check": "em_criterion_met",
            "value": bool(self.em_criterion_met),
            "status": "ok" if self.em_criterion_met else "warning",
            "message": (
                "Whether the Aitken-extrapolated EM criterion was met before "
                "the iteration cap."
            ),
        },
        {
            "section": "data",
            "check": "panels",
            "value": int(self.data.num_panels or 0),
            "status": "ok",
            "message": "Number of decision-maker panels.",
        },
        {
            "section": "data",
            "check": "cases",
            "value": int(self.data.num_cases),
            "status": "ok",
            "message": "Number of choice situations.",
        },
    ]

    if self.polish_report is not None:
        report = self.polish_report
        rows.append(
            {
                "section": "fit",
                "check": "polish_loglik_gain",
                "value": float(report.loglik_after - report.loglik_before),
                "status": "ok",
                "message": (
                    f"Log likelihood gained by {report.iterations} "
                    "observed-data Newton step(s) after EM. A large value "
                    "means EM stopped well short of the optimum."
                ),
            }
        )
        rows.append(
            {
                "section": "fit",
                "check": "polish_score_reduction",
                "value": float(report.score_before),
                "status": "ok",
                "message": (
                    "Observed-data score before the polish, for comparison "
                    "with observed_score_max after it."
                ),
            }
        )

    rows.append(
        {
            "section": "inference",
            "check": "covariance_available",
            "value": bool(self.inference.skip or self.covariance_available),
            "status": (
                "ok"
                if self.inference.skip or self.covariance_available
                else "warning"
            ),
            "message": (
                "Whether a finite covariance matrix was estimated. When false, "
                "every standard error in this results object is NaN."
            ),
        }
    )

    if self.diagnostics_config.check_separation and self.data.num_panels:
        prior = onp.asarray(
            self._get_class_probs(
                self._unpack_params(self.flat_params)[1],
                self.data.dems,
                self.data.num_panels,
            )
        )
        min_prior = float(prior.min())
        separated = min_prior <= self.diagnostics_config.separation_threshold
        rows.append(
            {
                "section": "latent_class",
                "check": "min_membership_probability",
                "value": min_prior,
                "status": "warning" if separated else "ok",
                "message": (
                    "Smallest prior class-membership probability over panels. "
                    "A value at zero means some demographic cell never belongs "
                    "to that class, so its membership coefficients are "
                    "unbounded and the observed information is singular in "
                    "that direction. Merge the cell, drop the variable, or fit "
                    "fewer classes."
                ),
            }
        )

    if self.information_diagnostics is not None and (
        self.diagnostics_config.check_collinearity
    ):
        info = self.information_diagnostics
        rows.append(
            {
                "section": "inference",
                "check": "information_rank",
                "value": float(info.rank),
                "status": "warning" if info.rank_deficient else "ok",
                "message": (
                    f"Numerical rank of the observed information out of "
                    f"{info.num_params} parameters. A deficient rank means "
                    "some standard errors are not identified."
                ),
            }
        )
        rows.append(
            {
                "section": "inference",
                "check": "information_condition_number",
                "value": float(info.condition_number),
                "status": (
                    "warning"
                    if not info.positive_definite or info.condition_number > 1e12
                    else "ok"
                ),
                "message": (
                    "Ratio of largest to smallest eigenvalue of the observed "
                    "information. Large values indicate weakly identified "
                    "parameter directions."
                ),
            }
        )
        rows.append(
            {
                "section": "inference",
                "check": "information_min_eigenvalue",
                "value": float(info.smallest_eigenvalue),
                "status": "ok" if info.positive_definite else "warning",
                "message": (
                    "Smallest eigenvalue of the observed information. Values "
                    "at or below zero indicate a saddle point rather than a "
                    "maximum."
                ),
            }
        )

    if self.em_res.class_probs_by_panel is not None:
        posterior = onp.asarray(self.em_res.class_probs_by_panel)
        entropy = -onp.sum(
            posterior * onp.log(onp.maximum(posterior, 1e-300)), axis=1
        )
        rows.append(
            {
                "section": "latent_class",
                "check": "posterior_entropy_mean",
                "value": float(entropy.mean()),
                "status": "ok",
                "message": "Mean entropy of posterior class membership.",
            }
        )

    shares_df = self.class_shares()
    min_share = float(cast(float, shares_df["share"].min()))
    rows.append(
        {
            "section": "latent_class",
            "check": "min_class_share",
            "value": min_share,
            "status": "warning" if min_share < 0.01 else "ok",
            "message": "Small classes can indicate weakly identified local optima.",
        }
    )
    if "effective_panels" in shares_df.columns:
        rows.append(
            {
                "section": "latent_class",
                "check": "min_effective_panels",
                "value": float(cast(float, shares_df["effective_panels"].min())),
                "status": "ok",
                "message": "Smallest posterior panel mass across classes.",
            }
        )

    structural = onp.asarray(self.em_res.structural_betas)
    max_abs_beta = float(onp.max(onp.abs(structural)))
    rows.append(
        {
            "section": "coefficients",
            "check": "max_abs_beta",
            "value": max_abs_beta,
            "status": (
                "warning"
                if (
                    self.diagnostics_config.warn_large_coefficients
                    and max_abs_beta
                    > self.diagnostics_config.large_coefficient_threshold
                )
                else "ok"
            ),
            "message": "Largest absolute structural coefficient.",
        }
    )
    numeraire_idx = getattr(self.model, "numeraire_idx", None)
    if numeraire_idx is not None:
        min_abs_numeraire = float(onp.min(onp.abs(structural[numeraire_idx, :])))
        threshold = self.diagnostics_config.near_zero_numeraire_threshold
        rows.append(
            {
                "section": "coefficients",
                "check": "min_abs_numeraire",
                "value": min_abs_numeraire,
                "status": (
                    "warning"
                    if (
                        self.diagnostics_config.warn_near_zero_numeraire
                        and min_abs_numeraire < threshold
                    )
                    else "ok"
                ),
                "message": "Small numeraires can dominate WTP/tradeoff ratios.",
            }
        )

    return LCLDiagnostics(pl.DataFrame(rows))

loglik(data, dems_data=None, *, per_panel=False)

Score observed choices with the fitted empirical specification.

The fitted encoder is reused, so Formulaic categorical levels and expanded columns retain their training-time meaning.

Parameters:

Name	Type	Description	Default
data	object	Long-format data containing one observed choice per case.	required
dems_data	object | None	Optional panel-level demographics joined by the fitted panel ID column.	None
per_panel	bool	Return a panel-level table instead of the total log likelihood.	False

Returns:

Type	Description
float or DataFrame	Total log likelihood when per_panel=False. Otherwise, a table with the original panel IDs and their log-likelihood contributions.

Source code in src/lcl/_results.py

def loglik(
    self,
    data: object,
    dems_data: object | None = None,
    *,
    per_panel: bool = False,
) -> float | pl.DataFrame:
    """Score observed choices with the fitted empirical specification.

    The fitted encoder is reused, so Formulaic categorical levels and expanded
    columns retain their training-time meaning.

    Parameters
    ----------
    data : object
        Long-format data containing one observed choice per case.
    dems_data : object | None, optional
        Optional panel-level demographics joined by the fitted panel ID column.
    per_panel : bool, default=False
        Return a panel-level table instead of the total log likelihood.

    Returns
    -------
    float or pl.DataFrame
        Total log likelihood when ``per_panel=False``. Otherwise, a table with
        the original panel IDs and their log-likelihood contributions.
    """
    parsed = self.model._transform_data(
        data,
        dems_data=dems_data,
        require_choice=True,
    )
    data_struct = cast(Data, self.model._setup_data(parsed)[0])
    if data_struct.num_panels is None or data_struct.panels is None:
        raise ValueError("Panel identifiers are required to score LCL data.")
    diff = _diff_unchosen_chosen(data_struct)
    panel_values = self._panel_loglik_fn(self.flat_params, diff, data_struct)

    if not per_panel:
        return float(jnp.sum(panel_values))

    first_panel_rows = data_struct.panels != jnp.roll(data_struct.panels, shift=1)
    first_panel_rows = first_panel_rows.at[0].set(True)
    return pl.DataFrame(
        {
            "panel": onp.asarray(parsed.original_panels[first_panel_rows]),
            "log_likelihood": onp.asarray(panel_values, dtype=onp.float64),
        }
    )

membership_coefficients()

Return nonbaseline class-membership coefficients with standard errors.

Class 0 is the reference category and therefore has no separately estimated membership coefficients.

Source code in src/lcl/_results.py

def membership_coefficients(self) -> pl.DataFrame:
    """Return nonbaseline class-membership coefficients with standard errors.

    Class 0 is the reference category and therefore has no separately
    estimated membership coefficients.
    """
    coefficients, standard_errors = self._apply_delta_method(
        self._calc_membership_coefficients, self.flat_params
    )
    coefficient_array = onp.asarray(coefficients)
    se_array = onp.asarray(standard_errors)
    variables = ["Intercept", *(self.model.dem_varnames or [])]
    rows = []
    for variable_idx, variable in enumerate(variables):
        for class_idx in range(1, self.model.num_classes):
            rows.append(
                {
                    "variable": variable,
                    "label": _model_variable_label(self.model, variable),
                    "class": class_idx,
                    "reference_class": 0,
                    "coefficient": float(
                        coefficient_array[variable_idx, class_idx - 1]
                    ),
                    "std_error": float(se_array[variable_idx, class_idx - 1]),
                }
            )
    return pl.DataFrame(rows)

parameter_names()

Return names aligned with rows and columns of cov_matrix.

Source code in src/lcl/_results.py

def parameter_names(self) -> list[str]:
    """Return names aligned with rows and columns of ``cov_matrix``."""
    names = [
        f"class_{class_idx}:{variable}"
        for variable in self.model.case_varnames
        for class_idx in range(self.model.num_classes)
    ]
    membership_rows = ["Intercept", *(self.model.dem_varnames or [])]
    names.extend(
        f"membership_class_{class_idx}:{variable}"
        for variable in membership_rows
        for class_idx in range(1, self.model.num_classes)
    )
    if len(names) != self.num_params:
        raise RuntimeError(
            "Parameter-name layout does not match covariance packing."
        )
    return names

predict(data=None, *, X=None, alts=None, cases=None, panels=None, dems=None, dem_panel_ids=None, past_choices=None, dems_data=None, past_choices_dems_data=None, panel_weights=None)

Generate out-of-sample latent-class predictions.

Prediction can be requested either with raw tabular data, which is encoded using the fitted model specification, or with already-constructed arrays. When historical choices are supplied through past_choices, class membership probabilities are updated with Bayes' rule before computing counterfactual choice probabilities, consumer surplus, and willingness to pay.

Parameters:

Name	Type	Description	Default
X	ArrayLike | None	Alternative-specific design matrix for array-style prediction. Ignored when data is provided.	None
alts	ArrayLike | None	Alternative identifiers aligned to rows of X.	None
cases	ArrayLike | None	Choice-situation identifiers aligned to rows of X.	None
panels	ArrayLike | None	Decision-maker identifiers aligned to rows of X.	None
dems	ArrayLike | None	Panel-level demographics for array-style prediction. When dem_panel_ids is omitted, rows must be in sorted unique panel-ID order.	None
dem_panel_ids	ArrayLike | None	Panel IDs aligned with rows of dems. The parser uses these IDs to validate and reorder demographic rows.	None
past_choices	PastChoicesData or tabular data	Historical choices used to condition latent-class membership probabilities. Pass a :class:~lcl._struct.PastChoicesData instance for array-style inputs, or a Polars/Pandas/DataFrame-like object containing the fitted model's alternative, case, panel, choice, alternative-specific, and demographic columns.	None
data	object | None	Long-format prediction data. If provided, the fitted encoder parses this data using the original empirical specification.	None
dems_data	object | None	Optional panel-level demographics to merge into data during prediction.	None
past_choices_dems_data	object | None	Optional panel-level demographics to merge into tabular past_choices. This argument is not used with :class:~lcl._struct.PastChoicesData.	None

Returns:

Type	Description
class:`~lcl._prediction.LCLPrediction`	Prediction results, including choice probabilities, consumer surplus, panel-level WTP values, and the class probabilities used for prediction.

Raises:

Type	Description
ValueError	If required prediction identifiers are missing, if fitted latent-class parameters are unavailable, or if past_choices_dems_data is provided without tabular past_choices.

Source code in src/lcl/_results.py

def predict(
    self,
    data: object | None = None,
    *,
    X: ArrayLike | None = None,
    alts: ArrayLike | None = None,
    cases: ArrayLike | None = None,
    panels: ArrayLike | None = None,
    dems: ArrayLike | None = None,
    dem_panel_ids: ArrayLike | None = None,
    past_choices: object | None = None,
    dems_data: object | None = None,
    past_choices_dems_data: object | None = None,
    panel_weights: str | Mapping[object, float] | Sequence[float] | None = None,
) -> LCLPrediction:
    """Generate out-of-sample latent-class predictions.

    Prediction can be requested either with raw tabular data, which is encoded
    using the fitted model specification, or with already-constructed arrays.
    When historical choices are supplied through ``past_choices``, class
    membership probabilities are updated with Bayes' rule before computing
    counterfactual choice probabilities, consumer surplus, and willingness to pay.

    Parameters
    ----------
    X : ArrayLike | None, optional
        Alternative-specific design matrix for array-style prediction. Ignored
        when ``data`` is provided.
    alts : ArrayLike | None, optional
        Alternative identifiers aligned to rows of ``X``.
    cases : ArrayLike | None, optional
        Choice-situation identifiers aligned to rows of ``X``.
    panels : ArrayLike | None, optional
        Decision-maker identifiers aligned to rows of ``X``.
    dems : ArrayLike | None, optional
        Panel-level demographics for array-style prediction. When
        ``dem_panel_ids`` is omitted, rows must be in sorted unique panel-ID
        order.
    dem_panel_ids : ArrayLike | None, optional
        Panel IDs aligned with rows of ``dems``. The parser uses these IDs to
        validate and reorder demographic rows.
    past_choices : PastChoicesData or tabular data, optional
        Historical choices used to condition latent-class membership probabilities.
        Pass a :class:`~lcl._struct.PastChoicesData` instance for array-style
        inputs, or a Polars/Pandas/DataFrame-like object containing the fitted
        model's alternative, case, panel, choice, alternative-specific, and
        demographic columns.
    data : object | None, optional
        Long-format prediction data. If provided, the fitted encoder parses this
        data using the original empirical specification.
    dems_data : object | None, optional
        Optional panel-level demographics to merge into ``data`` during prediction.
    past_choices_dems_data : object | None, optional
        Optional panel-level demographics to merge into tabular ``past_choices``.
        This argument is not used with :class:`~lcl._struct.PastChoicesData`.

    Returns
    -------
    :class:`~lcl._prediction.LCLPrediction`
        Prediction results, including choice probabilities, consumer surplus,
        panel-level WTP values, and the class probabilities used for prediction.

    Raises
    ------
    ValueError
        If required prediction identifiers are missing, if fitted latent-class
        parameters are unavailable, or if ``past_choices_dems_data`` is provided
        without tabular ``past_choices``.
    """
    if past_choices is None and past_choices_dems_data is not None:
        raise ValueError(
            "past_choices_dems_data can only be used when past_choices is provided."
        )
    partition_data_df = None
    raw_prediction_data = None
    if data is not None:
        parsed_predict = self.model._transform_data(data, dems_data=dems_data)
        encoder = getattr(self.model, "_encoder", None)
        if encoder is not None:
            raw_prediction_data = _coerce_frame(data).sort(
                list(
                    dict.fromkeys(
                        [encoder.panels_col, encoder.cases_col, encoder.alts_col]
                    )
                )
            )
            partition_data_df = _prediction_partition_data(
                data, dems_data, encoder.panels_col
            )
    else:
        if X is None or alts is None or cases is None or panels is None:
            raise ValueError(
                "Provide either data=... or X, alts, cases, and panels."
            )
        parsed_predict = _parsed_prediction_arrays(
            X=X,
            dems=dems,
            alts=alts,
            cases=cases,
            panels=panels,
            dem_panel_ids=dem_panel_ids,
            case_varnames=self.model.case_varnames,
            dem_varnames=self.model.dem_varnames,
        )
    predict_data = cast(Data, self.model._setup_data(parsed_predict)[0])
    if predict_data.num_panels is None or predict_data.panels is None:
        raise ValueError(
            "Panel identifiers are required for latent-class prediction."
        )
    structural_betas = self.em_res.structural_betas
    if structural_betas is None:
        raise ValueError("Structural betas are required for prediction.")
    shares = self.em_res.shares
    if shares is None:
        raise ValueError("Class shares are required for prediction.")
    if self.em_res.thetas is not None and predict_data.dems is None:
        raise ValueError(
            "dems is required for array-style prediction because the fitted "
            "class-membership model uses demographics. Pass dem_panel_ids to "
            "validate their panel alignment."
        )

    # Retained for posterior-updated WTP inference, which differentiates
    # through the Bayes update rather than freezing the posterior.
    data_past: Data | None = None
    diff_unchosen_chosen_past: DiffUnchosenChosen | None = None
    if past_choices is not None:
        parsed_past = _parse_past_choices(
            model=self.model,
            past_choices=past_choices,
            past_choices_dems_data=past_choices_dems_data,
        )
        _validate_past_choice_panels(parsed_past, parsed_predict)
        data_past = cast(Data, self.model._setup_data(parsed_past)[0])
        diff_unchosen_chosen_past = _diff_unchosen_chosen(data_past)
        class_probs_by_panel, _ = _compute_conditional_class_probs(
            structural_betas=structural_betas,
            thetas=self.em_res.thetas,
            shares=shares,
            diff_unchosen_chosen=diff_unchosen_chosen_past,
            data=data_past,
        )
        class_probabilities_source = "posterior"
    elif self.em_res.thetas is not None and predict_data.dems is not None:
        class_probs_by_panel = self._get_class_probs(
            self.em_res.thetas, predict_data.dems, predict_data.num_panels
        )
        class_probabilities_source = "prior"
    else:
        class_probs_by_panel = jnp.repeat(
            shares[None, :], predict_data.num_panels, axis=0
        )
        class_probabilities_source = "prior"

    choice_probs_by_class, log_sum_exp_utility = _choice_probabilities_and_logsum(
        predict_data.X,
        structural_betas,
        predict_data.cases,
        predict_data.num_cases,
    )

    # Ensure alpha (marginal utility of income) is correctly signed
    numeraire_idx = getattr(self.model, "numeraire_idx", None)
    if numeraire_idx is None:
        marginal_utility_income = jnp.ones(self.model.num_classes)
    else:
        marginal_utility_income = -structural_betas[numeraire_idx, :]

    surplus_by_class = log_sum_exp_utility / marginal_utility_income[None, :]

    if numeraire_idx is not None:
        betas_sans_numeraire = jnp.delete(structural_betas, numeraire_idx, axis=0)
        wtp_alt_vars_by_class = betas_sans_numeraire / marginal_utility_income
        wtp_alt_vars_by_panel = class_probs_by_panel @ wtp_alt_vars_by_class.T
        schema = [
            var for var in self.model.case_varnames if var != self.model.numeraire
        ]
    else:
        wtp_alt_vars_by_panel = jnp.empty((predict_data.num_panels, 0))
        schema = []

    panel_first_rows = predict_data.panels != jnp.roll(predict_data.panels, shift=1)
    panel_first_rows = panel_first_rows.at[0].set(True)
    panels_unique = onp.array(parsed_predict.original_panels[panel_first_rows])
    encoder = getattr(self.model, "_encoder", None)
    resolved_panel_weights = resolve_panel_weights(
        panel_weights,
        panels_unique,
        raw_prediction_data,
        encoder.panels_col if encoder is not None else "panels",
    )
    wtp_alt_vars_by_panel_df = pl.DataFrame(
        onp.array(wtp_alt_vars_by_panel), schema=schema
    ).with_columns(pl.Series("panels", panels_unique))

    if (
        predict_data.num_cases_per_panel is None
        or predict_data.panels_of_cases is None
    ):
        raise ValueError(
            "Panel identifiers are required for latent-class prediction."
        )
    conditional_surplus = jnp.einsum(
        "np,np->n",
        class_probs_by_panel[predict_data.panels_of_cases],
        surplus_by_class,
    )

    unconditional_choice_probs = jnp.sum(
        class_probs_by_panel[predict_data.panels] * choice_probs_by_class, axis=1
    )

    predicted_probs_df = pl.DataFrame(
        {
            "panels": parsed_predict.original_panels,
            "cases": parsed_predict.original_cases,
            "alts": parsed_predict.original_alts,
            "choice_probs": onp.array(
                unconditional_choice_probs, dtype=onp.float64
            ),
        }
    )

    first_case_rows = predict_data.cases != jnp.roll(predict_data.cases, shift=1)
    first_case_rows = first_case_rows.at[0].set(True)
    surplus_df = pl.DataFrame(
        {
            "panels": onp.array(parsed_predict.original_panels[first_case_rows]),
            "cases": onp.array(parsed_predict.original_cases[first_case_rows]),
            "surplus": onp.array(conditional_surplus, dtype=onp.float64),
        }
    )

    return LCLPrediction(
        predicted_probs_df=predicted_probs_df,
        surplus_df=surplus_df,
        wtp_alt_vars_by_panel_df=wtp_alt_vars_by_panel_df,
        predict_data=predict_data,
        results=self,
        class_probs_by_panel=class_probs_by_panel,
        class_probabilities_source=class_probabilities_source,
        partition_data_df=partition_data_df,
        original_alts=parsed_predict.original_alts,
        original_cases=parsed_predict.original_cases,
        original_panels=parsed_predict.original_panels,
        raw_prediction_data=raw_prediction_data,
        panel_weights=resolved_panel_weights,
        past_diff_unchosen_chosen=diff_unchosen_chosen_past,
        past_data=data_past,
    )

spec_summary()

Return a human-readable model specification summary.

Source code in src/lcl/_results.py

def spec_summary(self) -> str:
    """Return a human-readable model specification summary."""
    spec = getattr(self.model, "spec", None)
    if spec is not None:
        return "\n".join(spec.summary_lines())

    lines = [
        "Latent-class conditional logit",
        f"Classes: {self.model.num_classes}",
        "",
        "Utility variables:",
    ]
    for variable in self.model.case_varnames:
        suffix = ""
        if variable == self.model.numeraire:
            suffix = (
                f" [negative, min_abs={self._param_packing.numeraire_min_abs:g}]"
            )
        label = _model_variable_label(self.model, variable)
        variable_text = label if label == variable else f"{label} ({variable})"
        lines.append(f"  {variable_text}{suffix}")
    lines.append("")
    lines.append("Class-membership variables:")
    if self.model.dem_varnames:
        for variable in self.model.dem_varnames:
            label = _model_variable_label(self.model, variable)
            variable_text = label if label == variable else f"{label} ({variable})"
            lines.append(f"  {variable_text}")
    else:
        lines.append("  none")
    return "\n".join(lines)

summarize(num_decimals=3, *, show=True)

Alias for :meth:summarize_betas.

Source code in src/lcl/_results.py

def summarize(self, num_decimals: int = 3, *, show: bool = True) -> pl.DataFrame:
    """Alias for :meth:`summarize_betas`."""
    return self.summarize_betas(num_decimals=num_decimals, show=show)

summarize_betas(header=('Variable', "Means (\\beta's)", "Standard deviations (\\sigma's)"), num_decimals=3, *, show=True)

Print and return population-level coefficient moments.

Parameters:

Name	Type	Description	Default
header	tuple[str, str, str]	Column labels used in the printed LaTeX and terminal tables.	("Variable", ...)
num_decimals	int	Number of decimal places used in printed tables.	3
show	bool	Emit LaTeX and terminal renderings. Set to False for computation-only use.	True

Returns:

Type	Description
DataFrame	Tidy coefficient-moment table. The variable column preserves raw model names; label contains presentation labels used for printing.

Source code in src/lcl/_results.py

def summarize_betas(
    self,
    header: tuple[str, str, str] = (
        "Variable",
        r"Means (\beta's)",
        r"Standard deviations (\sigma's)",
    ),
    num_decimals: int = 3,
    *,
    show: bool = True,
) -> pl.DataFrame:
    """Print and return population-level coefficient moments.

    Parameters
    ----------
    header : tuple[str, str, str], default=("Variable", ...)
        Column labels used in the printed LaTeX and terminal tables.
    num_decimals : int, default=3
        Number of decimal places used in printed tables.
    show : bool, default=True
        Emit LaTeX and terminal renderings. Set to ``False`` for
        computation-only use.

    Returns
    -------
    pl.DataFrame
        Tidy coefficient-moment table.  The ``variable`` column preserves raw
        model names; ``label`` contains presentation labels used for printing.
    """
    summary_df = self.beta_summary()
    if show:
        log_or_print(
            logger,
            "%s",
            format_lcl_beta_summary(summary_df, header, num_decimals),
        )
    return summary_df

Held-out scoring

LCLResults.loglik transforms observed choices with the fitted encoder:

total_ll = results.loglik(test_data)
panel_ll = results.loglik(test_data, per_panel=True)

The panel-level form returns original panel IDs and their log-likelihood contributions. It is also the scoring path used by cross-validation.

Summary methods return Polars frames. Pass show=False to suppress their LaTeX and terminal renderings:

summary = results.summarize_betas(show=False)
class_coefficients = results.class_coefficients()  # includes std_error
membership = results.membership_coefficients()    # class 0 is the reference
classification = results.classification_diagnostics()

parameter_names() labels covariance rows and columns exactly. converged, cov_matrix, and adjusted_bic are the canonical names shared with conditional logit; convergence, covariance, and abic are deprecated aliases.

Diagnostics

lcl.results.LCLDiagnostics(frame)

Structured diagnostics for a fitted latent-class model.

Parameters:

Name	Type	Description	Default
frame	DataFrame	Diagnostic checks with at least section, check, value, status, and message columns.	required

Store diagnostic checks.

Source code in src/lcl/_diagnostics.py

def __init__(self, frame: pl.DataFrame) -> None:
    """Store diagnostic checks."""
    self._frame = frame

__repr__()

Return a compact textual representation.

Source code in src/lcl/_diagnostics.py

def __repr__(self) -> str:
    """Return a compact textual representation."""
    n_warn = self._frame.filter(pl.col("status") != "ok").height
    return f"LCLDiagnostics(checks={self._frame.height}, warnings={n_warn})"

print()

Print a compact diagnostics table.

Source code in src/lcl/_diagnostics.py

def print(self) -> None:
    """Print a compact diagnostics table."""
    rows = self._frame.select(["section", "check", "status", "value", "message"])
    print(tabulate(rows.iter_rows(), headers=rows.columns, tablefmt="simple"))

to_frame()

Return diagnostics as a Polars DataFrame.

Source code in src/lcl/_diagnostics.py

def to_frame(self) -> pl.DataFrame:
    """Return diagnostics as a Polars DataFrame."""
    return self._frame

Prediction and counterfactuals

Tabular prediction is preferred because it reuses the fitted encoder:

prediction = results.predict(data=counterfactual_data)
shares = prediction.market_shares()
aggregate = prediction.aggregate_elasticities(["cost", "time"])

Pass panel_weights= to predict as a panel-keyed mapping, a prediction-data column name, or a vector in sorted prediction-panel order. WTP supports se="delta", se="bootstrap" (an asymptotic parametric bootstrap), and se="none". Posterior-conditioned WTP uncertainty is refused because the current implementation does not differentiate through the Bayesian update.

Surplus frames include surplus_units (money with a numeraire, otherwise utils). Use baseline_prediction.surplus_change(counterfactual_prediction) for the identified welfare change rather than comparing unnormalised levels.

For array-style prediction, supply dem_panel_ids with dems so demographic rows can be validated and reordered. Without those IDs, demographic rows must follow sorted unique panel-ID order.

lcl.results.LCLPrediction(predicted_probs_df, surplus_df, wtp_alt_vars_by_panel_df, predict_data, results, class_probs_by_panel=None, class_probabilities_source='prior', partition_data_df=None, original_alts=None, original_cases=None, original_panels=None, raw_prediction_data=None, panel_weights=None, past_diff_unchosen_chosen=None, past_data=None)

Bases: _PredictionBase

Latent-class prediction with partitioned WTP inference.

Source code in src/lcl/_prediction.py

def __init__(
    self,
    predicted_probs_df: pl.DataFrame,
    surplus_df: pl.DataFrame,
    wtp_alt_vars_by_panel_df: pl.DataFrame,
    predict_data: Data,
    results: Any,
    class_probs_by_panel: Float64[Array, "panels classes"] | None = None,
    class_probabilities_source: str = "prior",
    partition_data_df: pl.DataFrame | None = None,
    original_alts: Any | None = None,
    original_cases: Any | None = None,
    original_panels: Any | None = None,
    raw_prediction_data: pl.DataFrame | None = None,
    panel_weights: Sequence[float] | onp.ndarray | None = None,
    past_diff_unchosen_chosen: Any | None = None,
    past_data: Data | None = None,
) -> None:
    """Store prediction outputs and references needed for post-processing.

    Parameters
    ----------
    predicted_probs_df : pl.DataFrame
        Long-format alternative choice probabilities.
    surplus_df : pl.DataFrame
        Case-level consumer surplus estimates.
    wtp_alt_vars_by_panel_df : pl.DataFrame
        Panel-level marginal WTP values for non-numeraire variables.
    predict_data : :class:`~lcl._struct.Data`
        Encoded data used to generate the predictions.
    results : Any
        Parent results object. Kept broad to support both latent-class and
        conditional-logit result containers without a circular import.
    class_probs_by_panel : Float64[Array, "panels classes"] | None, optional
        Class probabilities used to marginalize class-specific predictions.
    class_probabilities_source : str, default="prior"
        ``"posterior"`` when prediction used historical choices, otherwise
        ``"prior"``.
    partition_data_df : pl.DataFrame | None, optional
        Panel-level raw prediction columns available for WTP partitions.
    """
    self.predicted_probs = predicted_probs_df
    self.surplus_units = (
        "money"
        if getattr(results.model, "numeraire_idx", None) is not None
        else "utils"
    )
    self.surplus = surplus_df.with_columns(
        pl.lit(self.surplus_units).alias("surplus_units")
    )
    self.wtp_alt_vars_by_panel = wtp_alt_vars_by_panel_df
    self.predict_data = predict_data
    self.results = results
    self.class_probs_by_panel = class_probs_by_panel
    self.class_probabilities_source = class_probabilities_source
    self.partition_data = partition_data_df
    self.original_alts = (
        onp.asarray(original_alts)
        if original_alts is not None
        else onp.asarray(predict_data.alts)
    )
    self.original_cases = (
        onp.asarray(original_cases)
        if original_cases is not None
        else onp.asarray(predict_data.cases)
    )
    self.original_panels = (
        onp.asarray(original_panels)
        if original_panels is not None
        else (
            None
            if predict_data.panels is None
            else onp.asarray(predict_data.panels)
        )
    )
    self.raw_prediction_data = raw_prediction_data
    # Retained so willingness to pay computed from Bayesian-updated class
    # membership can be differentiated through that update rather than
    # treating the posterior as a fixed constant.
    self.past_diff_unchosen_chosen = past_diff_unchosen_chosen
    self.past_data = past_data
    num_panels = predict_data.num_panels
    if num_panels is None:
        raise ValueError("Panel identifiers are required for prediction.")
    weights = (
        onp.ones(num_panels, dtype=onp.float64)
        if panel_weights is None
        else onp.asarray(panel_weights, dtype=onp.float64)
    )
    if weights.shape != (num_panels,):
        raise ValueError(
            "panel_weights must contain one value per prediction panel."
        )
    if not onp.all(onp.isfinite(weights)) or onp.any(weights < 0.0):
        raise ValueError("panel_weights must be finite and nonnegative.")
    if not onp.any(weights > 0.0):
        raise ValueError("At least one panel weight must be positive.")
    self.panel_weights = weights

compute_wtp(*wtp_requests, partition_data=None, panel_col='panels', num_decimals=4, class_probabilities='stored', se='delta', bootstrap_draws=500, bootstrap_seed=0, show=True)

Compute the Marginal Willingness-to-Pay (WTP) across demographic partitions.

Evaluates the ratio of the target parameter to the negative cost parameter (marginal utility of income) for dynamically defined subsets of decision-makers. Outputs formatted LaTeX and terminal summary tables, including analytical standard errors derived via the Delta Method.

Parameters:

Name	Type	Description	Default
*wtp_requests	WTPRequest | Iterable[WTPRequest]	One or more configuration objects specifying the target variable, the demographic partitioning variable, and the binning strategy (e.g., quintiles, categorical, custom breaks, or a dummy-coded categorical factor).	()
partition_data	object | None	Optional panel-level or long-format tabular data containing partitioning variables that were not included in the fitted class-membership specification. Values must be constant within each panel.	None
panel_col	str	Panel identifier column in partition_data.	"panels"
num_decimals	int	Number of decimal places used in printed WTP tables.	4
class_probabilities	(stored, prior, posterior)	Class-membership probabilities used for WTP/tradeoff point estimates. "stored" uses the probabilities already attached to this prediction object, including Bayesian posterior updates from past_choices. "prior" recomputes demographics-only class probabilities. "posterior" requires that prediction was created with past_choices.	"stored"
se	(delta, bootstrap, none)	Standard-error method. Delta-method and asymptotic parametric-bootstrap standard errors are available for prior class probabilities. When the prediction used past_choices, both methods differentiate through the Bayesian class update, so the reported uncertainty reflects the same posterior as the point estimate. Note that partitions built from the data (quintiles, custom breaks) are treated as fixed, so standard errors are conditional on the realized partition and on the demographic design.	"delta"
bootstrap_draws	int	Number of asymptotic parameter draws for se="bootstrap".	500
bootstrap_seed	int	Reproducible random seed for parametric-bootstrap draws.	0
show	bool	Emit LaTeX and terminal renderings for each request.	True

Returns:

Type	Description
dict[str, DataFrame]	Summary tables keyed by their printed titles. Each table preserves raw variable names in variable and partition_variable and includes presentation labels in label and partition_label.

Raises:

Type	Description
ValueError	If the parent model was not estimated with a specified numeraire constraint.

Source code in src/lcl/_prediction.py

def compute_wtp(
    self,
    *wtp_requests: WTPRequest | Iterable[WTPRequest],
    partition_data: object | None = None,
    panel_col: str = "panels",
    num_decimals: int = 4,
    class_probabilities: Literal["stored", "prior", "posterior"] = "stored",
    se: Literal["delta", "bootstrap", "none"] = "delta",
    bootstrap_draws: int = 500,
    bootstrap_seed: int = 0,
    show: bool = True,
) -> dict[str, pl.DataFrame]:
    """Compute the Marginal Willingness-to-Pay (WTP) across demographic partitions.

    Evaluates the ratio of the target parameter to the negative cost parameter
    (marginal utility of income) for dynamically defined subsets of decision-makers.
    Outputs formatted LaTeX and terminal summary tables, including analytical
    standard errors derived via the Delta Method.

    Parameters
    ----------
    *wtp_requests : WTPRequest | Iterable[WTPRequest]
        One or more configuration objects specifying the target variable,
        the demographic partitioning variable, and the binning strategy (e.g.,
        quintiles, categorical, custom breaks, or a dummy-coded categorical
        factor).
    partition_data : object | None, optional
        Optional panel-level or long-format tabular data containing partitioning
        variables that were not included in the fitted class-membership
        specification. Values must be constant within each panel.
    panel_col : str, default="panels"
        Panel identifier column in ``partition_data``.
    num_decimals : int, default=4
        Number of decimal places used in printed WTP tables.
    class_probabilities : {"stored", "prior", "posterior"}, default="stored"
        Class-membership probabilities used for WTP/tradeoff point estimates.
        ``"stored"`` uses the probabilities already attached to this
        prediction object, including Bayesian posterior updates from
        ``past_choices``. ``"prior"`` recomputes demographics-only class
        probabilities. ``"posterior"`` requires that prediction was created
        with ``past_choices``.
    se : {"delta", "bootstrap", "none"}, default="delta"
        Standard-error method. Delta-method and asymptotic parametric-bootstrap
        standard errors are available for prior class probabilities.
        When the prediction used ``past_choices``, both methods
        differentiate through the Bayesian class update, so the reported
        uncertainty reflects the same posterior as the point estimate.  Note
        that partitions built from the data (quintiles, custom breaks) are
        treated as fixed, so standard errors are conditional on the realized
        partition and on the demographic design.
    bootstrap_draws : int, default=500
        Number of asymptotic parameter draws for ``se="bootstrap"``.
    bootstrap_seed : int, default=0
        Reproducible random seed for parametric-bootstrap draws.
    show : bool, default=True
        Emit LaTeX and terminal renderings for each request.

    Returns
    -------
    dict[str, pl.DataFrame]
        Summary tables keyed by their printed titles.  Each table preserves
        raw variable names in ``variable`` and ``partition_variable`` and
        includes presentation labels in ``label`` and ``partition_label``.

    Raises
    ------
    ValueError
        If the parent model was not estimated with a specified numeraire constraint.
    """
    if se not in {"delta", "bootstrap", "none"}:
        raise ValueError("se must be 'delta', 'bootstrap', or 'none'.")
    if class_probabilities not in {"stored", "prior", "posterior"}:
        raise ValueError(
            "class_probabilities must be 'stored', 'prior', or 'posterior'."
        )
    if (
        class_probabilities == "posterior"
        and self.class_probabilities_source != "posterior"
    ):
        raise ValueError(
            "class_probabilities='posterior' requires predict(..., past_choices=...)."
        )
    use_posterior = (
        class_probabilities in {"stored", "posterior"}
        and self.class_probabilities_source == "posterior"
    )
    if (
        se in {"delta", "bootstrap"}
        and use_posterior
        and (self.past_data is None or self.past_diff_unchosen_chosen is None)
    ):
        raise ValueError(
            "Posterior-updated WTP inference needs the past-choice design "
            "that produced the posterior. Recreate the prediction with "
            "predict(..., past_choices=...)."
        )

    # We rely on the explicitly tracked numeraire index from _pre_fit
    if getattr(self.results.model, "numeraire_idx", None) is None:
        raise ValueError("A numeraire must be defined to compute WTP.")

    cost_idx = self.results.model.numeraire_idx
    if self.predict_data.panels is None or self.predict_data.num_panels is None:
        raise ValueError("Panel identifiers are required to compute WTP.")

    requests = _flatten_wtp_requests(wtp_requests)
    if not requests:
        return {}

    df_with_idx = self.wtp_alt_vars_by_panel.with_row_index("panel_idx")

    if (
        self.predict_data.dems is not None
        and self.results.model.dem_varnames is not None
    ):
        dems_df = pl.DataFrame(
            onp.array(self.predict_data.dems),
            schema=self.results.model.dem_varnames,
        ).with_row_index("panel_idx")

        df_with_idx = df_with_idx.join(dems_df, on="panel_idx")

    partition_cols = _partition_columns(requests)
    missing_partition_cols = [
        col for col in partition_cols if col not in df_with_idx.columns
    ]
    if missing_partition_cols:
        source_partition_data = partition_data
        source_panel_col = panel_col
        if source_partition_data is None and self.partition_data is not None:
            source_partition_data = self.partition_data
            source_panel_col = "panels"

        if source_partition_data is None:
            raise ValueError(
                "WTP partition columns were not found in the fitted/prediction "
                "demographics: "
                f"{missing_partition_cols}. Pass partition_data=... for "
                "panel-level grouping variables outside the model specification."
            )
        external_partitions = _coerce_partition_data(
            source_partition_data, source_panel_col, missing_partition_cols
        )
        df_with_idx = df_with_idx.join(external_partitions, on="panels", how="left")
        has_missing_partition = df_with_idx.select(
            pl.any_horizontal(pl.col(missing_partition_cols).is_null()).any()
        ).item()
        if has_missing_partition:
            raise ValueError(
                "partition_data is missing partition values for one or more "
                "prediction panels."
            )

    summary_tables: dict[str, pl.DataFrame] = {}

    for req in requests:
        partition_type = req.partition_type
        if not isinstance(partition_type, PartitionType):
            partition_type = PartitionType(partition_type)

        partitioned_df = _apply_wtp_partition(df_with_idx, req)
        if "_partition_order" in partitioned_df.columns:
            partitioned_df = partitioned_df.sort("_partition_order")
        try:
            target_idx = self.results.model.case_varnames.index(req.alt_var)
        except ValueError:
            raise ValueError(
                f"Alternative-specific variable '{req.alt_var}' not found in "
                "model specification."
            )
        target_label = self.results.model.variable_label(req.alt_var)
        partition_label = self.results.model.variable_label(req.demographic_var)
        selected_class_probs = None
        if se == "none":
            selected_class_probs = self._class_probs_for_wtp(class_probabilities)
        # Differentiating through the Bayes update keeps the reported
        # uncertainty consistent with the point estimate: the posterior is a
        # smooth function of the same coefficients, not a fixed constant.
        posterior_kwargs = (
            {
                "past_diff_unchosen_chosen": self.past_diff_unchosen_chosen,
                "past_data": self.past_data,
            }
            if use_posterior
            else {"past_diff_unchosen_chosen": None, "past_data": None}
        )
        summary_rows = []

        for partition_name, subset_df in partitioned_df.group_by(
            "Partition", maintain_order=True
        ):
            subset_panel_indices = jnp.array(
                subset_df["panel_idx"].to_numpy(), dtype=jnp.int32
            )
            subset_panel_weights = jnp.asarray(
                self.panel_weights[onp.asarray(subset_panel_indices)]
            )

            if se == "delta":
                mean_wtp, se_val = self.results._apply_delta_method(
                    self._compute_subset_mean_wtp,
                    self.results.flat_params,
                    target_idx=target_idx,
                    cost_idx=cost_idx,
                    subset_panel_indices=subset_panel_indices,
                    subset_panel_weights=subset_panel_weights,
                    dems=self.predict_data.dems,
                    num_panels=self.predict_data.num_panels,
                    **posterior_kwargs,
                )
                se_float = float(se_val)
            elif se == "bootstrap":
                mean_wtp = self._compute_subset_mean_wtp(
                    self.results.flat_params,
                    target_idx=target_idx,
                    cost_idx=cost_idx,
                    subset_panel_indices=subset_panel_indices,
                    subset_panel_weights=subset_panel_weights,
                    dems=self.predict_data.dems,
                    num_panels=self.predict_data.num_panels,
                    **posterior_kwargs,
                )
                se_val = self.results._parametric_bootstrap_se(
                    self._compute_subset_mean_wtp,
                    self.results.flat_params,
                    target_idx=target_idx,
                    cost_idx=cost_idx,
                    subset_panel_indices=subset_panel_indices,
                    subset_panel_weights=subset_panel_weights,
                    dems=self.predict_data.dems,
                    num_panels=self.predict_data.num_panels,
                    draws=bootstrap_draws,
                    seed=bootstrap_seed,
                    **posterior_kwargs,
                )
                se_float = float(se_val)
            else:
                if selected_class_probs is None:
                    raise ValueError("Class probabilities were not available.")
                mean_wtp = self._compute_subset_mean_wtp_from_class_probs(
                    target_idx=target_idx,
                    cost_idx=cost_idx,
                    subset_panel_indices=subset_panel_indices,
                    subset_panel_weights=subset_panel_weights,
                    class_probs=selected_class_probs,
                )
                se_float = float("nan")

            summary_rows.append(
                {
                    "variable": req.alt_var,
                    "label": target_label,
                    "partition_variable": req.demographic_var,
                    "partition_label": partition_label,
                    req.demographic_var: str(_partition_label(partition_name)),
                    "Mean_Marginal_WTP": float(mean_wtp),
                    "Standard_Error": se_float,
                    "Class_Probabilities": class_probabilities,
                    "SE_Method": se,
                    "Panel_Count": subset_df.height,
                    "Effective_Panel_Weight": float(
                        onp.asarray(subset_panel_weights).sum()
                    ),
                }
            )

        res_df = pl.DataFrame(summary_rows)
        partition_desc = (
            "dummy-coded categorical"
            if req.dummy_vars is not None
            else partition_type.value
        )
        title = (
            f"Marginal WTP for {target_label} by "
            f"{partition_label} ({partition_desc})"
        )
        summary_tables[title] = res_df
        if show:
            log_or_print(
                logger,
                "%s",
                format_wtp_table(
                    title,
                    res_df,
                    req.demographic_var,
                    partition_label,
                    num_decimals,
                ),
            )

    return summary_tables

denominator_diagnostics()

Return denominator diagnostics for WTP/tradeoff ratios.

Source code in src/lcl/_prediction.py

def denominator_diagnostics(self) -> pl.DataFrame:
    """Return denominator diagnostics for WTP/tradeoff ratios."""
    numeraire_idx = getattr(self.results.model, "numeraire_idx", None)
    if numeraire_idx is None:
        raise ValueError("A numeraire must be defined to compute diagnostics.")
    structural_betas = self.results.em_res.structural_betas
    if structural_betas is None:
        raise ValueError("Structural betas are required.")
    denominator = -structural_betas[numeraire_idx, :]
    return pl.DataFrame(
        {
            "class": list(range(self.results.model.num_classes)),
            "denominator": [self.results.model.numeraire]
            * self.results.model.num_classes,
            "denominator_label": [
                self.results.model.variable_label(str(self.results.model.numeraire))
            ]
            * self.results.model.num_classes,
            "denominator_value": onp.asarray(denominator),
            "abs_denominator": onp.asarray(jnp.abs(denominator)),
            "min_abs_floor": [self.results._param_packing.numeraire_min_abs]
            * self.results.model.num_classes,
        }
    )

tradeoff(*wtp_requests, **kwargs)

Alias for :meth:compute_wtp with more neutral terminology.

Source code in src/lcl/_prediction.py

def tradeoff(
    self,
    *wtp_requests: WTPRequest | Iterable[WTPRequest],
    **kwargs: Any,
) -> dict[str, pl.DataFrame]:
    """Alias for :meth:`compute_wtp` with more neutral terminology."""
    return self.compute_wtp(*wtp_requests, **kwargs)

wtp_by_class(target=None)

Return class-specific WTP/tradeoff ratios.

Parameters:

Name	Type	Description	Default
target	str | None	Optional target variable to filter. By default, all non-numeraire alternative-specific variables are returned.	None

Returns:

Type	Description
DataFrame	Class-specific ratios beta_target / -beta_numeraire with raw variable names, display labels, and denominator diagnostics.

Source code in src/lcl/_prediction.py

def wtp_by_class(self, target: str | None = None) -> pl.DataFrame:
    """Return class-specific WTP/tradeoff ratios.

    Parameters
    ----------
    target : str | None, optional
        Optional target variable to filter.  By default, all non-numeraire
        alternative-specific variables are returned.

    Returns
    -------
    pl.DataFrame
        Class-specific ratios ``beta_target / -beta_numeraire`` with raw
        variable names, display labels, and denominator diagnostics.
    """
    numeraire_idx = getattr(self.results.model, "numeraire_idx", None)
    if numeraire_idx is None:
        raise ValueError("A numeraire must be defined to compute WTP.")
    structural_betas = self.results.em_res.structural_betas
    if structural_betas is None:
        raise ValueError("Structural betas are required.")

    denominator = -structural_betas[numeraire_idx, :]
    rows = []
    for var_idx, variable in enumerate(self.results.model.case_varnames):
        if var_idx == numeraire_idx:
            continue
        if target is not None and variable != target:
            continue
        ratios = structural_betas[var_idx, :] / denominator
        for class_idx in range(self.results.model.num_classes):
            rows.append(
                {
                    "variable": variable,
                    "label": self.results.model.variable_label(variable),
                    "denominator": self.results.model.numeraire,
                    "denominator_label": self.results.model.variable_label(
                        str(self.results.model.numeraire)
                    ),
                    "class": class_idx,
                    "tradeoff": float(ratios[class_idx]),
                    "denominator_value": float(denominator[class_idx]),
                }
            )
    return pl.DataFrame(rows)

lcl.options.WTPRequest(alt_var, demographic_var, partition_type, bins=None, dummy_vars=None, dummy_labels=None, base_category='base') dataclass

Configuration for a marginal willingness-to-pay summary.

__post_init__()

Normalize and validate the partition request.

Source code in src/lcl/options.py

def __post_init__(self) -> None:
    """Normalize and validate the partition request."""
    if not isinstance(self.partition_type, PartitionType):
        try:
            self.partition_type = PartitionType(self.partition_type)
        except ValueError:
            valid_options = [item.value for item in PartitionType]
            raise ValueError(
                f"Invalid partition type: {self.partition_type}\n"
                f"Must be one of {valid_options}"
            ) from None
    if self.partition_type == PartitionType.CUSTOM_BREAKS and not isinstance(
        self.bins, list
    ):
        raise ValueError(
            "When partition_type is 'custom_breaks', bins must be breakpoints."
        )
    if isinstance(self.bins, list) and any(
        right <= left for left, right in zip(self.bins, self.bins[1:])
    ):
        raise ValueError("Custom WTP breakpoints must be strictly increasing.")
    if self.dummy_vars is not None:
        if not self.dummy_vars:
            raise ValueError("dummy_vars must contain at least one column name.")
        if len(set(self.dummy_vars)) != len(self.dummy_vars):
            raise ValueError("dummy_vars cannot contain duplicate column names.")
        if self.partition_type != PartitionType.CATEGORICAL:
            raise ValueError(
                "Dummy-coded WTP partitions require partition_type='categorical'."
            )
        if self.dummy_labels is not None and len(self.dummy_labels) != len(
            self.dummy_vars
        ):
            raise ValueError("dummy_labels must have one label per dummy column.")

lcl.options.PartitionType

Bases: StrEnum

Supported binning strategies for WTP analysis.

lcl.options.PastChoicesData(X, y, alts, cases, panels, dems=None, dem_panel_ids=None) dataclass

Array-style historical choices used to update class membership.