Latent-class conditional logit

The headline estimator: a finite mixture of conditional logits, fit by expectation-maximization, with optional class-membership demographic regression. Class-specific taste vectors are recovered via maximum likelihood at each M-step; class probabilities are updated either as aggregate shares or, when demographics are present, using a fractional-response multinomial logit model.

Most users reach this estimator through the declarative LCLSpec + lcl.fit workflow; the class below is what lcl.fit constructs and fits under the hood, and it can also be driven directly.

Model

lcl.latent_class_conditional_logit.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 | LCLSpec = 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 isinstance(num_classes, LCLSpec):
        if spec is not None:
            raise ValueError(
                "Pass either LatentClassConditionalLogit(spec) or spec=..., not both."
            )
        spec = num_classes
        num_classes = spec.classes

    if spec is not None:
        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, formula=None, utility_formula=None, membership_formula=None, choice_col=None, case_varnames=None, dem_varnames=None, dems_data=None, em_alg_config=None, mle_config=None, error_config=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
formula	str | None	Backward-compatible combined Formulaic string, for example "choice ~ price + time | income + C(segment)". Prefer utility_formula and membership_formula in new code.	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 if formula is not provided.	None
case_varnames	Sequence[str] | None	A list of alternative-specific variables to include in the utility specification. Required if formula is not provided.	None
dem_varnames	Sequence[str] | None	A list of demographic variables used to predict latent class membership.	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
em_alg_config	:class:`~lcl._struct.EMAlgConfig`	A dataclass (PyTree) containing configuration options for the overall EM algorithm (e.g., maximum iterations, tolerance, hardware distribution).	EMAlgConfig()
mle_config	:class:`~lcl._struct.MleConfig`	A dataclass (PyTree) containing optimization settings for the M-step's internal L-BFGS solver.	MleConfig()

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,
    formula: 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,
    dems_data: Any | None = None,
    em_alg_config: EMAlgConfig | None = None,
    mle_config: MleConfig | None = None,
    error_config: ErrorConfig | 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).
    formula : str | None, default=None
        Backward-compatible combined Formulaic string, for example
        ``"choice ~ price + time | income + C(segment)"``.  Prefer
        ``utility_formula`` and ``membership_formula`` in new code.
    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 if `formula` is not provided.
    case_varnames : Sequence[str] | None, default=None
        A list of alternative-specific variables to include in the utility
        specification. Required if `formula` is not provided.
    dem_varnames : Sequence[str] | None, default=None
        A list of demographic variables used to predict latent class membership.
    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`.
    em_alg_config : :class:`~lcl._struct.EMAlgConfig`, default=EMAlgConfig()
        A dataclass (PyTree) containing configuration options for the overall EM
        algorithm (e.g., maximum iterations, tolerance, hardware distribution).
    mle_config : :class:`~lcl._struct.MleConfig`, default=MleConfig()
        A dataclass (PyTree) containing optimization settings for the M-step's
        internal L-BFGS solver.

    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.
    """
    if self.spec is not None:
        alts_col = alts_col or self.spec.ids.alt
        cases_col = cases_col or self.spec.ids.case
        panels_col = panels_col or self.spec.ids.panel
        choice_col = choice_col or self.spec.ids.choice
        formula = formula if formula is not None else self.spec.formula
        utility_formula = (
            utility_formula
            if utility_formula is not None
            else self.spec.utility_formula
        )
        membership_formula = (
            membership_formula
            if membership_formula is not None
            else self.spec.membership_formula
        )
        if formula is None and utility_formula is None:
            case_varnames = (
                case_varnames if case_varnames is not None else self.spec.utility
            )
        if formula is None and membership_formula is None:
            dem_varnames = (
                dem_varnames if dem_varnames is not None else self.spec.membership
            )
        self.num_classes = self.spec.classes
        if self.spec.numeraire is not None:
            self.numeraire = self.spec.numeraire
            self.numeraire_min_abs = self.spec.numeraire_min_abs

    if alts_col is None or cases_col is None or panels_col is None:
        raise ValueError(
            "alts_col, cases_col, and panels_col are required unless an "
            "LCLSpec is attached to the model."
        )

    if self.num_classes < 2:
        raise ValueError("num_classes must be at least 2.")
    if em_alg_config is None:
        em_alg_config = fit_options.to_em_config() if fit_options else EMAlgConfig()
    if mle_config is None:
        mle_config = optimization_options if optimization_options else MleConfig()
    if error_config is None:
        error_config = inference if inference is not None else ErrorConfig()
    if diagnostics is None:
        diagnostics = DiagnosticsOptions()

    parsed_data = self._ingest_data(
        data=data,
        alts_col=alts_col,
        cases_col=cases_col,
        panels_col=panels_col,
        formula=formula,
        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
    )
    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, weights, init_beta = 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)

    em_vars = _get_starting_vals(
        diff_unchosen_chosen,
        data_struct,
        self.num_classes,
        em_alg_config,
        mle_config,
        self.numeraire_idx,
        self.numeraire_min_abs,
    )

    num_devices = em_alg_config.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})

    logliks_list, em_recursion = [], 0
    em_history_rows: list[dict[str, Any]] = []
    while em_recursion < em_alg_config.maxiter:
        logger.info("EM recursion: %s", em_recursion)
        if progress_callback is not None:
            progress_callback({"event": "em_step", "iteration": em_recursion})

        em_vars = _em_alg(
            em_vars,
            diff_unchosen_chosen,
            data_struct,
            self.num_classes,
            mle_config,
            em_alg_config,
            self.numeraire_idx,
            self.numeraire_min_abs,
        )

        logliks_list.append(em_vars.unconditional_loglik)
        em_history_rows.append(self._em_history_row(em_recursion, em_vars))
        em_recursion += 1

        # Only force a host sync every `check_interval` steps
        if em_recursion >= 5 and (em_recursion % em_alg_config.check_interval == 0):
            # jax.block_until_ready() forces the sync explicitly here
            current_ll = float(em_vars.unconditional_loglik)
            past_ll = float(logliks_list[-5])

            rel_change = abs(current_ll - past_ll) / abs(past_ll)
            if rel_change <= em_alg_config.loglik_tol:
                break

    strict_mle_config = replace(mle_config, ftol=1e-8, maxiter=500)
    em_vars = _em_alg(
        em_vars,
        diff_unchosen_chosen,
        data_struct,
        self.num_classes,
        strict_mle_config,
        em_alg_config,
        self.numeraire_idx,
        self.numeraire_min_abs,
    )
    em_history_rows.append(self._em_history_row(em_recursion, em_vars))
    optimization_history_rows = self._optimizer_snapshot(
        em_vars, diff_unchosen_chosen, data_struct, em_recursion
    )

    estim_time_sec = time() - self._fit_start_time

    log_or_print(logger, "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,
        em_alg_config=em_alg_config,
        error_config=error_config,
        diagnostics_config=diagnostics,
        estim_time_sec=estim_time_sec,
        em_history=em_history_rows,
        optimization_history=optimization_history_rows,
    )

Results

lcl._results.LCLResults(model_spec, em_vars, estimation_data, em_recursion, em_alg_config, error_config, estim_time_sec, diagnostics_config=None, em_history=None, optimization_history=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']	Robust cluster-adjusted covariance matrix, strictly aligned with the Stata finite-sample correction multiplier :math:(G / (G - 1)).
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
em_alg_config	:class:`~lcl._struct.EMAlgConfig`	EM convergence and iteration configuration.	required
error_config	:class:`~lcl._struct.ErrorConfig` | None	Covariance and standard-error configuration.	required
estim_time_sec	float	Wall-clock estimation time in seconds.	required
diagnostics_config	:class:`~lcl._struct.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

Source code in src/lcl/_results.py

def __init__(
    self,
    model_spec: Any,
    em_vars: EMVars,
    estimation_data: Data,
    em_recursion: int,
    em_alg_config: EMAlgConfig,
    error_config: ErrorConfig | 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,
) -> 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.
    em_alg_config : :class:`~lcl._struct.EMAlgConfig`
        EM convergence and iteration configuration.
    error_config : :class:`~lcl._struct.ErrorConfig` | None
        Covariance and standard-error configuration.
    estim_time_sec : float
        Wall-clock estimation time in seconds.
    diagnostics_config : :class:`~lcl._struct.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.
    """
    self.model = model_spec
    self.em_res = em_vars
    self.data = estimation_data
    self.total_recursions = em_recursion
    self.converged = em_recursion < (em_alg_config.maxiter - 1)
    self.estim_time_sec = estim_time_sec
    self.error_config = error_config if error_config is not None else ErrorConfig()
    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.flat_params = self._pack_params()

    # Calculate degrees of freedom
    latent_betas = self.em_res.latent_betas
    num_beta_params = latent_betas.size
    if self.em_res.thetas is not None:
        num_theta_params = self.em_res.thetas.size
    else:
        num_theta_params = self.model.num_classes - 1

    self.num_params = num_beta_params + num_theta_params
    self.cov_matrix = self._compute_covariance()

    # Compute information criteria
    num_panels = self.data.num_panels
    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
    )
    n_star = (self.num_params + 2) / 24
    self.adjusted_bic = (
        jnp.log(num_panels) * n_star - 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,
        )

__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"
    return " | ".join(
        [
            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}>",
        ]
    )

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	Variables, 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
        Variables, 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,
    )
    stds, se_stds = self._apply_delta_method(
        self._calc_population_std_betas,
        self.flat_params,
        dems=self.data.dems,
        num_panels=self.data.num_panels,
    )
    structural = onp.asarray(self.em_res.structural_betas)
    rows = []
    for idx, variable in enumerate(self.model.case_varnames):
        rows.append(
            {
                "variable": 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.

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.
    """
    structural_betas = self.em_res.structural_betas
    if structural_betas is None:
        raise ValueError("Structural betas are required.")
    rows = []
    beta_array = onp.asarray(structural_betas)
    for var_idx, variable in enumerate(self.model.case_varnames):
        for class_idx in range(self.model.num_classes):
            rows.append(
                {
                    "variable": variable,
                    "class": class_idx,
                    "coefficient": float(beta_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.em_res.shares is None:
        raise ValueError("Class shares are required.")
    shares = onp.asarray(self.em_res.shares)
    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):
        row = {"class": class_idx, "share": float(share)}
        if posterior_arr is not None:
            row["effective_panels"] = float(posterior_arr[:, class_idx].sum())
        rows.append(row)
    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"Final log likelihood: {float(self.em_res.unconditional_loglik):.6g}",
        f"Warnings: {warnings.height}",
    ]
    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": "EM convergence flag.",
        },
        {
            "section": "fit",
            "check": "log_likelihood",
            "value": float(self.em_res.unconditional_loglik),
            "status": "ok",
            "message": "Final unconditional log likelihood.",
        },
        {
            "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.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))

predict(X=None, alts=None, cases=None, panels=None, dems=None, past_choices=None, data=None, dems_data=None, past_choices_dems_data=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.	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,
    X: ArrayLike | None = None,
    alts: ArrayLike | None = None,
    cases: ArrayLike | None = None,
    panels: ArrayLike | None = None,
    dems: ArrayLike | None = None,
    past_choices: object | None = None,
    data: object | None = None,
    dems_data: object | None = None,
    past_choices_dems_data: object | 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.
    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
    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:
            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,
            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 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,
        )
        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, :]
    ).squeeze()

    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])
    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)
    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,
    )

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 = (
                " [negative, "
                f"min_abs={getattr(self.model, 'numeraire_min_abs', 1e-5):g}]"
            )
        lines.append(f"  {variable}{suffix}")
    lines.append("")
    lines.append("Class-membership variables:")
    if self.model.dem_varnames:
        lines.extend(f"  {variable}" for variable in self.model.dem_varnames)
    else:
        lines.append("  none")
    return "\n".join(lines)

summarize(num_decimals=3)

Alias for :meth:summarize_betas.

Source code in src/lcl/_results.py

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

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

Print and return population-level moments with Delta-method SEs.

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,
) -> pl.DataFrame:
    """Print and return population-level moments with Delta-method SEs."""
    summary_df = self.beta_summary()
    body_rows, data_clean = [], []
    converter = LatexNodes2Text(math_mode="text")
    header_clean = [converter.latex_to_text(col) for col in header]

    for row in summary_df.iter_rows(named=True):
        coeff_nm = str(row["variable"])
        body_rows.append(
            f"{coeff_nm} & {float(row['mean']):.{num_decimals}f} & {float(row['sd']):.{num_decimals}f} \\\\"
        )
        body_rows.append(
            f" & ({float(row['mean_se']):.{num_decimals}f}) & ({float(row['sd_se']):.{num_decimals}f}) \\\\"
        )
        var_clean = converter.latex_to_text(coeff_nm)
        data_clean.append(
            (
                var_clean,
                f"{float(row['mean']):.{num_decimals}f}",
                f"{float(row['sd']):.{num_decimals}f}",
            )
        )
        data_clean.append(
            (
                "",
                f"({float(row['mean_se']):.{num_decimals}f})",
                f"({float(row['sd_se']):.{num_decimals}f})",
            )
        )

    latex_string = "\n".join(
        [r"\toprule", " & ".join(header) + r" \\", r"\midrule", "%"]
        + body_rows
        + ["%", r"\bottomrule "]
    )
    table_preview = tabulate(
        data_clean,
        headers=header_clean,
        tablefmt="simple_outline",
        floatfmt=f".{num_decimals}f",
    )
    log_or_print(
        logger,
        "\n--- LaTeX Output ---\n\n%s\n\n--- Table preview ---\n\n%s",
        latex_string,
        table_preview,
    )
    return summary_df

Diagnostics

lcl._diagnostics.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

lcl._prediction.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)

Container for counterfactual inference, consumer surplus, and willingness-to-pay (WTP).

Provides methods to analyze decision-maker behavior under new choice sets or policy changes. Utilizes the Delta Method to compute rigorous analytical standard errors for non-linear combinations of parameters (e.g., marginal WTP) across dynamically defined demographic partitions.

Attributes:

Name	Type	Description
predicted_probs	DataFrame	DataFrame of out-of-sample choice probabilities for each alternative.
surplus	DataFrame	DataFrame of expected consumer surplus (inclusive value) per choice situation.
wtp_alt_vars_by_panel	DataFrame	DataFrame of expected marginal WTP for each alternative-specific characteristic, calculated at the individual decision-maker level.
predict_data	:class:`~lcl._struct.Data`	The parsed design matrices corresponding to the counterfactual scenarios.
results	:class:`~lcl._results.LCLResults`	Reference to the parent estimation results, required for Delta Method covariance calculations and parameter unpacking.
class_probs_by_panel	Array | None	Posterior (or prior) probabilities of latent class membership used to generate these predictions. If historical choices were provided during prediction, these represent the Bayesian-updated posteriors.
partition_data	DataFrame | None	Panel-level columns from raw prediction data that are constant within panel and can be used for WTP partitions.

Store prediction outputs and references needed for post-processing.

Parameters:

Name	Type	Description	Default
predicted_probs_df	DataFrame	Long-format alternative choice probabilities.	required
surplus_df	DataFrame	Case-level consumer surplus estimates.	required
wtp_alt_vars_by_panel_df	DataFrame	Panel-level marginal WTP values for non-numeraire variables.	required
predict_data	:class:`~lcl._struct.Data`	Encoded data used to generate the predictions.	required
results	Any	Parent results object. Kept broad to support both latent-class and conditional-logit result containers without a circular import.	required
class_probs_by_panel	Float64[Array, 'panels classes'] | None	Class probabilities used to marginalize class-specific predictions.	None
class_probabilities_source	str	"posterior" when prediction used historical choices, otherwise "prior".	"prior"
partition_data_df	DataFrame | None	Panel-level raw prediction columns available for WTP partitions.	None

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,
) -> 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 = surplus_df
    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

compute_wtp(*wtp_requests, partition_data=None, panel_col='panels', num_decimals=4, class_probabilities='stored', se='delta')

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, none)	Standard-error method. Delta-method standard errors are available for prior class probabilities. Posterior-updated WTP through past_choices requires differentiating through the Bayesian class update and is therefore refused unless se="none".	"delta"

Returns:

Type	Description
dict[str, DataFrame]	Summary tables keyed by their printed titles.

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", "none"] = "delta",
) -> 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", "none"}, default="delta"
        Standard-error method.  Delta-method standard errors are available for
        prior class probabilities.  Posterior-updated WTP through
        ``past_choices`` requires differentiating through the Bayesian class
        update and is therefore refused unless ``se="none"``.

    Returns
    -------
    dict[str, pl.DataFrame]
        Summary tables keyed by their printed titles.

    Raises
    ------
    ValueError
        If the parent model was not estimated with a specified numeraire constraint.
    """
    if se not in {"delta", "none"}:
        raise ValueError("se must be either 'delta' 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=...)."
        )
    if (
        se == "delta"
        and class_probabilities in {"stored", "posterior"}
        and self.class_probabilities_source == "posterior"
    ):
        raise NotImplementedError(
            "Delta-method WTP after past_choices requires differentiating "
            "through the posterior class update. Use se='none' or "
            "class_probabilities='prior'."
        )

    # 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."
            )
        selected_class_probs = None
        if se == "none":
            selected_class_probs = self._class_probs_for_wtp(class_probabilities)
        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
            )

            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,
                    dems=self.predict_data.dems,
                    num_panels=self.predict_data.num_panels,
                )
                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,
                    class_probs=selected_class_probs,
                )
                se_float = float("nan")

            summary_rows.append(
                {
                    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,
                }
            )

        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 {req.alt_var} by "
            f"{req.demographic_var} ({partition_desc})"
        )
        summary_tables[title] = res_df
        log_or_print(
            logger,
            "%s",
            _format_wtp_table(title, res_df, req.demographic_var, 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_value": onp.asarray(denominator),
            "abs_denominator": onp.asarray(jnp.abs(denominator)),
            "min_abs_floor": [
                getattr(self.results.model, "numeraire_min_abs", 1e-5)
            ]
            * self.results.model.num_classes,
        }
    )

elasticities(vars)

Compute full matrices of own- and cross-elasticities for continuous features.

Analytically calculates the percentage change in the probability of choosing alternative J given a one-percent change in a continuous attribute of alternative K. The method handles both conditional (latent class) and unconditional (standard conditional logit) probability matrices via a vectorized cartesian expansion across choice situations.

Parameters:

Name	Type	Description	Default
vars	str | Iterable[str]	The name(s) of the continuous variable(s) for which to compute the elasticities (e.g., "price", ["price", "travel_time"]).	required

Returns:

Type	Description
DataFrame	A DataFrame in long format mapping the target alternative (target_alts, whose attribute is changing) to the affected alternative (alts, whose probability is changing). Includes choice situation IDs, decision-maker panel IDs (if applicable), and the computed point elasticities.

Raises:

Type	Description
ValueError	If latent class probabilities are missing, or if a requested variable is not found in the estimated model specification.

Source code in src/lcl/_prediction.py

def elasticities(self, vars: str | Iterable[str]) -> pl.DataFrame:
    """Compute full matrices of own- and cross-elasticities for continuous features.

    Analytically calculates the percentage change in the probability of choosing
    alternative J given a one-percent change in a continuous attribute of
    alternative K. The method handles both conditional (latent class) and
    unconditional (standard conditional logit) probability matrices via a vectorized
    cartesian expansion across choice situations.

    Parameters
    ----------
    vars : str | Iterable[str]
        The name(s) of the continuous variable(s) for which to compute the
        elasticities (e.g., "price", ["price", "travel_time"]).

    Returns
    -------
    pl.DataFrame
        A DataFrame in long format mapping the target alternative (`target_alts`,
        whose attribute is changing) to the affected alternative (`alts`, whose
        probability is changing). Includes choice situation IDs, decision-maker
        panel IDs (if applicable), and the computed point elasticities.

    Raises
    ------
    ValueError
        If latent class probabilities are missing, or if a requested variable
        is not found in the estimated model specification.
    """
    if isinstance(vars, str):
        vars = [vars]

    data = self.predict_data
    is_lc = hasattr(self.results, "em_res")

    if is_lc:
        betas = self.results.em_res.structural_betas  # (K, C)
        if self.class_probs_by_panel is None:
            raise ValueError(
                "class_probs_by_panel must be available to compute LC elasticities."
            )
        if data.panels is None:
            raise ValueError("Panel identifiers are required for LC elasticities.")
        S_ic = self.class_probs_by_panel[data.panels]  # (N, C)
    else:
        betas = self.results.coeff_[:, None]  # (K, 1)
        S_ic = jnp.ones((data.X.shape[0], 1))  # (N, 1)

    num_classes = betas.shape[1]

    P_ij_c, _ = _choice_probabilities_and_logsum(
        data.X, betas, data.cases, data.num_cases
    )
    P_ij = jnp.sum(S_ic * P_ij_c, axis=1)  # (N,)

    base_dict_j = {
        "cases": onp.array(data.cases),
        "alts": onp.array(data.alts),
        "P_j": onp.array(jnp.maximum(P_ij, 1e-250)),
    }
    base_dict_k = {
        "cases": onp.array(data.cases),
        "target_alts": onp.array(data.alts),
    }

    for c in range(num_classes):
        base_dict_j[f"P_jc_{c}"] = onp.array(P_ij_c[:, c])
        base_dict_k[f"P_kc_{c}"] = onp.array(P_ij_c[:, c])

    df_j = pl.DataFrame(base_dict_j)
    df_k_base = pl.DataFrame(base_dict_k)

    res_dfs = []

    for var in vars:
        try:
            var_idx = self.results.model.case_varnames.index(var)
        except ValueError:
            raise ValueError(f"Variable '{var}' not found in model specification.")

        X_v = data.X[:, var_idx]
        beta_v = betas[var_idx, :]

        # A_{jc} = S_{ic} * beta_{vc} * P_{ij|c}
        A_jc = S_ic * beta_v[None, :] * P_ij_c
        U_j = jnp.sum(A_jc, axis=1)

        df_j_var = df_j.with_columns(pl.Series("U_j", onp.array(U_j)))
        for c in range(num_classes):
            df_j_var = df_j_var.with_columns(
                pl.Series(f"A_jc_{c}", onp.array(A_jc[:, c]))
            )

        df_k_var = df_k_base.with_columns(pl.Series("X_k", onp.array(X_v)))
        cross_df = df_j_var.join(df_k_var, on="cases", how="inner")

        # V_{jk} = sum_c A_{jc} P_{kc}
        cross_df = cross_df.with_columns(
            V_jk=pl.sum_horizontal(
                [
                    pl.col(f"A_jc_{c}") * pl.col(f"P_kc_{c}")
                    for c in range(num_classes)
                ]
            )
        )

        # D_{jk} = U_j * (j == k) - V_{jk}
        cross_df = cross_df.with_columns(
            is_own=pl.col("alts") == pl.col("target_alts")
        ).with_columns(
            D_jk=pl.when(pl.col("is_own"))
            .then(pl.col("U_j") - pl.col("V_jk"))
            .otherwise(-pl.col("V_jk"))
        )

        # E_{jk} = D_{jk} * X_k / P_j
        elas_name = f"elasticity_{var}"
        cross_df = cross_df.with_columns(
            (pl.col("D_jk") * pl.col("X_k") / pl.col("P_j")).alias(elas_name)
        )

        res_dfs.append(cross_df.select(["cases", "alts", "target_alts", elas_name]))

    final_df = res_dfs[0]
    for i in range(1, len(res_dfs)):
        final_df = final_df.join(res_dfs[i], on=["cases", "alts", "target_alts"])

    if data.panels is not None:
        panel_df = pl.DataFrame(
            {"cases": onp.array(data.cases), "panels": onp.array(data.panels)}
        ).unique()
        final_df = panel_df.join(final_df, on="cases")
        final_cols = ["panels", "cases", "alts", "target_alts"] + [
            f"elasticity_{v}" for v in vars
        ]
    else:
        final_cols = ["cases", "alts", "target_alts"] + [
            f"elasticity_{v}" for v in vars
        ]

    return final_df.select(final_cols).sort(["cases", "alts", "target_alts"])

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 and the denominator used for each class.

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`` and the
        denominator used for each class.
    """
    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,
                    "denominator": self.results.model.numeraire,
                    "class": class_idx,
                    "tradeoff": float(ratios[class_idx]),
                    "denominator_value": float(denominator[class_idx]),
                }
            )
    return pl.DataFrame(rows)

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

Configuration object for calculating Marginal Willingness-to-Pay (WTP).

Parameters:

Name	Type	Description	Default
alt_var	str	The target alternative-specific variable for the WTP numerator.	required
demographic_var	str	The demographic variable used to partition the decision-makers. When dummy_vars is supplied, this is the semantic factor name used in the output table.	required
partition_type	PartitionType | str	The strategy for grouping demographic_var. Dummy-coded factors must use PartitionType.CATEGORICAL.	required
bins	list[float] | None	Custom breakpoints if partition_type is "custom_breaks".	None
dummy_vars	list[str] | None	One-hot dummy columns that jointly represent a categorical variable. The all-zero row is treated as the base category.	None
dummy_labels	list[str] | None	Display labels for dummy_vars in the same order. Defaults to the dummy column names.	None
base_category	str	Display label for the all-zero base category when dummy_vars is supplied.	"base"

__post_init__()

Validate and normalize WTP request options.

Source code in src/lcl/_struct.py

def __post_init__(self) -> None:
    """Validate and normalize WTP request options."""
    # Attempt to coerce partition into PartitionType
    if not isinstance(self.partition_type, PartitionType):
        try:
            self.partition_type = PartitionType(self.partition_type)
        except ValueError:
            valid_options = [e.value for e in PartitionType]
            raise ValueError(
                "\n".join(
                    [
                        f"Invalid partition type: {self.partition_type}",
                        f"Must be one of {valid_options}",
                    ]
                )
            )
    # When using custom breaks, ensure bins are a list of breakpoints
    if self.partition_type == PartitionType.CUSTOM_BREAKS and not isinstance(
        self.bins, list
    ):
        raise ValueError(
            "When partition_type is 'custom_breaks', 'bins' must be a list of breakpoints."
        )
    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 for each dummy column."
            )

lcl._struct.PartitionType

Bases: StrEnum

Supported binning strategies for marginal Willingness-To-Pay (WTP) analysis.

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

Array-style container for historical choices used during prediction.

Pass to LCLResults.predict(past_choices=...) to update latent-class membership probabilities with observed choices before scoring counterfactual choice sets. Users with tabular historical-choice data can pass that DataFrame directly to past_choices; this wrapper is intended for callers that already manage design matrices and ID arrays.

Attributes:

Name	Type	Description
X	ArrayLike	Alternative-specific design matrix in long format.
y	ArrayLike	Boolean or binary choice indicators aligned to rows of X.
alts	ArrayLike	Alternative identifiers aligned to rows of X.
cases	ArrayLike	Choice-situation identifiers aligned to rows of X.
panels	ArrayLike	Decision-maker identifiers aligned to rows of X.
dems	(ArrayLike | None, optional)	Panel-level demographic matrix, one row per unique panel, when the fitted latent-class membership model includes demographics.