Generalize interaction terms to N-way, any column type

Replace the restrictive HeterogeneityInteraction (2-way, attribute x
demographic only) with a flexible InteractionTerm dataclass that supports
arbitrary N-way interactions between any numeric columns. This enables
attribute x attribute (e.g. price x time), 3-way+ interactions, and
removes the respondent-constant restriction on interaction columns.

- config.py: Add InteractionTerm(columns: tuple[str, ...]) with name
property and validation; keep HeterogeneityInteraction for backward compat
- pipeline.py: Generalize interaction loop to multiply N columns together
- Model.py: Replace demographic-only UI with flexible interaction builder
using st.form + session state (add/remove terms, any numeric column)
- test_e2e.py: Update tests 22/26 to use InteractionTerm; add tests 30
(3-way interaction) and 31 (attribute x attribute); all 31 tests pass

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

app/pages/2_⚙️_Model.py

@@ -0,0 +1,1018 @@
+"""分析侠 — Page 2: Model Configuration and Estimation."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pandas as pd
+import streamlit as st
+
+# ── path setup ─────────────────────────────────────────────────────
+ROOT = Path(__file__).resolve().parents[2]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+APP_DIR = Path(__file__).resolve().parents[1]
+if str(APP_DIR) not in sys.path:
+    sys.path.insert(0, str(APP_DIR))
+
+from dce_analyzer.config import (  # noqa: E402
+    DummyCoding,
+    FullModelSpec,
+    InteractionTerm,
+    VariableSpec,
+)
+from dce_analyzer.data import get_device_info  # noqa: E402
+from dce_analyzer.pipeline import estimate_from_spec  # noqa: E402
+from utils import init_session_state, require_data, sidebar_branding  # noqa: E402
+
+
+def _connected_components(pairs: list[tuple[int, int]], n: int) -> list[list[int]]:
+    """Compute connected components from selected correlation pairs."""
+    adj: dict[int, set[int]] = {i: set() for i in range(n)}
+    for a, b in pairs:
+        adj[a].add(b)
+        adj[b].add(a)
+    visited: set[int] = set()
+    components: list[list[int]] = []
+    for i in range(n):
+        if i not in visited and adj[i]:
+            comp: list[int] = []
+            queue = [i]
+            while queue:
+                node = queue.pop(0)
+                if node in visited:
+                    continue
+                visited.add(node)
+                comp.append(node)
+                for nb in adj[node]:
+                    if nb not in visited:
+                        queue.append(nb)
+            if len(comp) >= 2:
+                components.append(sorted(comp))
+    return components
+
+init_session_state()
+sidebar_branding()
+
+# ── Page header ────────────────────────────────────────────────────
+st.header("Model")
+st.caption("Configure utility variables, choose a model type, and run estimation.")
+
+require_data()
+
+df: pd.DataFrame = st.session_state.df
+
+
+# ── helpers ────────────────────────────────────────────────────────
+def _guess_col(columns: list[str], candidates: list[str], role: str | None = None) -> str:
+    """Find a column by name candidates, checking inferred_columns first."""
+    inferred = st.session_state.get("inferred_columns", {})
+    if role and inferred.get(role) in columns:
+        return inferred[role]
+    lowered = {c.lower(): c for c in columns}
+    for target in candidates:
+        if target.lower() in lowered:
+            return lowered[target.lower()]
+    return columns[0]
+
+
+columns = df.columns.tolist()
+
+# ── 1  Column role assignment ──────────────────────────────────────
+st.subheader("1. Column roles")
+st.markdown("Assign the structural columns in your dataset.")
+
+if st.session_state.get("inferred_columns"):
+    st.caption("Pre-filled from auto-detect on the Data page.")
+
+r1, r2, r3, r4 = st.columns(4)
+id_col = r1.selectbox(
+    "ID column",
+    columns,
+    index=columns.index(_guess_col(columns, ["respondent_id", "id", "ID"], "id")),
+)
+task_col = r2.selectbox(
+    "Task column",
+    columns,
+    index=columns.index(_guess_col(columns, ["task_id", "task"], "task")),
+)
+alt_col = r3.selectbox(
+    "Alternative column",
+    columns,
+    index=columns.index(_guess_col(columns, ["alternative", "alt"], "alt")),
+)
+choice_col = r4.selectbox(
+    "Choice column",
+    columns,
+    index=columns.index(_guess_col(columns, ["choice", "chosen"], "choice")),
+)
+
+# ── BWS (Best-Worst Scaling) mode ────────────────────────────────
+st.divider()
+bws_mode = st.checkbox(
+    "BWS (Best-Worst Scaling) data",
+    value=False,
+    key="bws_mode",
+    help="Enable if your data contains both best AND worst choices per task. "
+         "Requires at least 3 alternatives per task (J >= 3).",
+)
+bws_worst_col: str | None = None
+bws_estimate_lambda_w: bool = True
+
+if bws_mode:
+    # Auto-detect worst column candidates
+    _worst_candidates = [c for c in columns if c not in {id_col, task_col, alt_col, choice_col}]
+    _worst_default = _guess_col(columns, ["worst", "worst_choice", "least_preferred"], None)
+    if _worst_default not in _worst_candidates:
+        _worst_default = _worst_candidates[0] if _worst_candidates else columns[0]
+
+    bws_c1, bws_c2 = st.columns(2)
+    with bws_c1:
+        bws_worst_col = st.selectbox(
+            "Worst choice column",
+            _worst_candidates,
+            index=_worst_candidates.index(_worst_default) if _worst_default in _worst_candidates else 0,
+            key="bws_worst_col",
+            help="Column indicating the worst (least preferred) alternative in each task. "
+                 "Same format as the choice column (binary 0/1 or label).",
+        )
+    with bws_c2:
+        bws_estimate_lambda_w = st.checkbox(
+            "Estimate lambda_w (worst scale parameter)",
+            value=True,
+            key="bws_estimate_lw",
+            help="If checked, estimates a scale parameter lambda_w for worst choices. "
+                 "lambda_w > 1 means worst choices are more deterministic; lambda_w < 1 means noisier. "
+                 "If unchecked, lambda_w = 1 (equivalent to MaxDiff specification).",
+        )
+    st.caption(
+        "BWS uses **sequential best-first** likelihood: "
+        "P(best) x P(worst | best removed). "
+        "The existing choice column is treated as the **best** choice."
+    )
+
+# ── 2  Variable selection and coding ──────────────────────────────
+st.divider()
+st.subheader("2. Utility variables")
+
+# Allow selecting any numeric column (DCE attributes are typically numeric-coded)
+structural_cols = {id_col, task_col, alt_col, choice_col}
+numeric_columns = [c for c in columns if pd.api.types.is_numeric_dtype(df[c]) and c not in structural_cols]
+default_features = [
+    c
+    for c in [
+        "price", "time", "comfort", "reliability",
+        "travel_time", "travel_cost", "headway", "changes",
+    ]
+    if c in numeric_columns
+]
+if not default_features and numeric_columns:
+    default_features = numeric_columns[: min(4, len(numeric_columns))]
+
+feature_cols = st.multiselect(
+    "Select variables for the utility function",
+    options=numeric_columns,
+    default=default_features,
+    help="Select the attribute columns to include in the utility specification.",
+)
+
+if len(feature_cols) == 0:
+    st.warning("Pick at least one utility variable.")
+    st.stop()
+
+# ── Per-variable coding type: Continuous vs Dummy ──────────────────
+st.markdown("**Variable coding**")
+st.caption(
+    "For each variable, choose **Continuous** (single coefficient, assumes linear effect) "
+    "or **Dummy** (one coefficient per level, flexible non-linear effect). "
+    "Dummy coding is standard for categorical DCE attributes."
+)
+
+coding_map: dict[str, str] = {}  # col -> "continuous" | "dummy"
+ref_levels: dict[str, object] = {}  # col -> reference level value
+
+n_coding_cols = min(4, len(feature_cols))
+coding_cols = st.columns(n_coding_cols)
+
+for idx, col in enumerate(feature_cols):
+    with coding_cols[idx % n_coding_cols]:
+        unique_vals = sorted(df[col].dropna().unique())
+        n_unique = len(unique_vals)
+        # default to dummy if few unique values (typical categorical attribute)
+        default_idx = 1 if 2 <= n_unique <= 10 else 0
+        coding = st.selectbox(
+            f"{col}",
+            ["Continuous", "Dummy"],
+            index=default_idx,
+            key=f"coding_{col}",
+            help=f"{n_unique} unique values: {unique_vals[:8]}{'...' if n_unique > 8 else ''}",
+        )
+        coding_map[col] = coding.lower()
+
+        if coding == "Dummy":
+            ref = st.selectbox(
+                f"Reference level",
+                unique_vals,
+                index=0,
+                key=f"ref_{col}",
+                help="The omitted baseline category. Other levels are estimated relative to this.",
+            )
+            ref_levels[col] = ref
+
+# ── Build dummy coding specs (backend will expand columns) ─────────
+_dummy_codings: list[DummyCoding] = []
+expanded_feature_cols: list[str] = []  # expanded column names for UI display
+_dummy_info: dict[str, list[str]] = {}  # original col -> list of dummy col names
+
+for col in feature_cols:
+    if coding_map[col] == "dummy":
+        dc = DummyCoding(column=col, ref_level=ref_levels[col])
+        _dummy_codings.append(dc)
+        dummy_names, _ = dc.expand(df)
+        expanded_feature_cols.extend(dummy_names)
+        _dummy_info[col] = dummy_names
+    else:
+        expanded_feature_cols.append(col)
+
+# Show summary of expanded variables
+with st.expander("Variable specification summary", expanded=False):
+    summary_rows = []
+    for col in feature_cols:
+        if coding_map[col] == "dummy":
+            ref = ref_levels[col]
+            n_dummies = len(_dummy_info[col])
+            summary_rows.append({
+                "Variable": col,
+                "Coding": "Dummy",
+                "Reference": str(ref),
+                "Coefficients": n_dummies,
+                "Columns": ", ".join(_dummy_info[col]),
+            })
+        else:
+            summary_rows.append({
+                "Variable": col,
+                "Coding": "Continuous",
+                "Reference": "—",
+                "Coefficients": 1,
+                "Columns": col,
+            })
+    st.dataframe(pd.DataFrame(summary_rows), use_container_width=True, hide_index=True)
+    st.caption(f"Total parameters to estimate: **{len(expanded_feature_cols)}**")
+
+# ── 3  Model type and settings ────────────────────────────────────
+st.divider()
+st.subheader("3. Model type and settings")
+
+# Show detected hardware
+st.info(f"Compute device: **{get_device_info()}**")
+
+model_type_label = st.radio(
+    "Select model type",
+    ["Conditional Logit", "Mixed Logit", "GMNL", "Latent Class"],
+    horizontal=True,
+    key="model_type_radio",
+)
+model_type_map = {
+    "Conditional Logit": "conditional",
+    "Mixed Logit": "mixed",
+    "GMNL": "gmnl",
+    "Latent Class": "latent_class",
+}
+model_type = model_type_map[model_type_label]
+
+if bws_mode:
+    if model_type == "conditional":
+        st.info("BWS + Conditional Logit: lambda_w is fully identified. Good baseline.")
+    elif model_type == "mixed":
+        st.info(
+            "BWS + Mixed Logit: lambda_w must be a fixed scalar (not random). "
+            "It is identified separately from the random coefficient distributions."
+        )
+    elif model_type == "gmnl":
+        st.warning(
+            "BWS + GMNL: lambda_w is identified separately from sigma_tau (individual scale), "
+            "but both must be fixed parameters. Monitor convergence carefully."
+        )
+    elif model_type == "latent_class":
+        st.info(
+            "BWS + Latent Class: lambda_w is shared across all classes. "
+            "Per-class lambda_w is theoretically identified but increases parameter count."
+        )
+
+dist_map: dict[str, str] = {}
+
+if model_type == "conditional":
+    st.caption(
+        "All coefficients are fixed across respondents. "
+        "Fast to estimate, good baseline model."
+    )
+    s1, s2 = st.columns(2)
+    maxiter = s1.slider("Max optimizer iterations", 20, 500, 200, step=10, key="cl_maxiter")
+    est_seed = s2.number_input("Estimation seed", min_value=1, value=123, step=1, key="cl_seed")
+    for col in expanded_feature_cols:
+        dist_map[col] = "fixed"
+    n_draws = 1
+    n_classes = 2
+    n_starts = 10
+
+elif model_type == "gmnl":
+    st.caption(
+        "Generalized Multinomial Logit (Fiebig et al. 2010). "
+        "Extends Mixed Logit with individual-level scale heterogeneity. "
+        "Nests both S-MNL (pure scale) and MMNL as special cases."
+    )
+
+    st.markdown("**Distribution assumptions**")
+    st.caption("Set distributions for each variable. At least one random variable is required.")
+    with st.expander("What do the distribution options mean?"):
+        st.markdown(
+            "- **fixed**: The coefficient is the same for all respondents.\n"
+            "- **normal**: Varies across respondents following a normal distribution.\n"
+            "- **lognormal**: exp(normal), ensuring always positive values."
+        )
+
+    gmnl_dist_cols = st.columns(min(4, len(feature_cols)))
+    for idx, col in enumerate(feature_cols):
+        with gmnl_dist_cols[idx % len(gmnl_dist_cols)]:
+            default_dist_idx = 0
+            dist_val = st.selectbox(
+                f"{col}" + (" (dummy)" if coding_map.get(col) == "dummy" else ""),
+                ["fixed", "normal", "lognormal"],
+                index=default_dist_idx,
+                key=f"gmnl_dist_{col}",
+            )
+            if coding_map.get(col) == "dummy" and col in _dummy_info:
+                for dc in _dummy_info[col]:
+                    dist_map[dc] = dist_val
+            else:
+                dist_map[col] = dist_val
+
+    st.markdown("**Estimation settings**")
+    gs1, gs2, gs3 = st.columns(3)
+    n_draws = gs1.slider("Halton draws", 20, 2000, 200, step=10, key="gmnl_draws")
+    maxiter = gs2.slider("Max optimizer iterations", 20, 500, 200, step=10, key="gmnl_maxiter")
+    est_seed = gs3.number_input("Estimation seed", min_value=1, value=123, step=1, key="gmnl_seed")
+
+    with st.expander("About GMNL scale parameters"):
+        st.markdown(
+            "The GMNL model estimates three additional parameters:\n"
+            "- **tau** (scale mean): controls the average scale of utility.\n"
+            "- **sigma_tau** (scale SD): individual variation in scale.\n"
+            "- **gamma** (mixing, 0-1): gamma=0 is pure scale heterogeneity (S-MNL), "
+            "gamma=1 is GMNL-II (closest to standard MMNL)."
+        )
+
+    # ── Correlation structure for GMNL ────────────────────────────────
+    _gmnl_random_expanded = [c for c in expanded_feature_cols if dist_map.get(c, "fixed") != "fixed"]
+    mxl_correlated = False
+    mxl_correlation_groups: list[list[int]] | None = None
+
+    if len(_gmnl_random_expanded) >= 2:
+        st.markdown("**Correlation structure**")
+        _gmnl_corr_mode = st.radio(
+            "Random parameter correlations",
+            ["Independent", "Full correlation", "Selective (pick pairs)"],
+            horizontal=True,
+            key="gmnl_corr_mode",
+            help="Independent: each random parameter varies independently. "
+                 "Full: all random parameters are correlated (Cholesky). "
+                 "Selective: choose specific pairs to correlate.",
+        )
+        if _gmnl_corr_mode == "Full correlation":
+            mxl_correlated = True
+        elif _gmnl_corr_mode == "Selective (pick pairs)":
+            _n_rand = len(_gmnl_random_expanded)
+            _all_corr_pairs = [
+                (i, j, _gmnl_random_expanded[i], _gmnl_random_expanded[j])
+                for i in range(_n_rand)
+                for j in range(i + 1, _n_rand)
+            ]
+            _selected_corr: list[tuple[int, int]] = []
+            _n_pair_cols = min(4, max(1, len(_all_corr_pairs)))
+            _pair_cols = st.columns(_n_pair_cols)
+            for _pidx, (_i, _j, _ni, _nj) in enumerate(_all_corr_pairs):
+                with _pair_cols[_pidx % _n_pair_cols]:
+                    if st.checkbox(f"{_ni} ↔ {_nj}", key=f"gmnl_corr_{_ni}_{_nj}"):
+                        _selected_corr.append((_i, _j))
+            if _selected_corr:
+                mxl_correlated = True
+                _groups = _connected_components(_selected_corr, _n_rand)
+                mxl_correlation_groups = _groups
+                _group_labels = [
+                    [_gmnl_random_expanded[k] for k in g] for g in _groups
+                ]
+                st.caption(f"Correlation blocks: {_group_labels}")
+            else:
+                st.caption("No pairs selected — using independent structure.")
+    elif len(_gmnl_random_expanded) == 1:
+        st.caption("Only one random parameter — correlation not applicable.")
+
+    n_classes = 2
+    n_starts = 10
+
+elif model_type == "mixed":
+    st.caption(
+        "Allows coefficients to vary continuously across respondents "
+        "(preference heterogeneity). Uses simulated maximum likelihood with Halton draws."
+    )
+
+    st.markdown("**Distribution assumptions**")
+    st.caption("Set distributions for each variable (or group of dummies). "
+               "Dummy-coded variables are typically kept **fixed**.")
+    with st.expander("What do the distribution options mean?"):
+        st.markdown(
+            "- **fixed**: The coefficient is the same for all respondents.\n"
+            "- **normal**: Varies across respondents following a normal distribution.\n"
+            "- **lognormal**: exp(normal), ensuring always positive values."
+        )
+
+    # Show distribution selector per original variable (applies to all its dummies)
+    mxl_dist_cols = st.columns(min(4, len(feature_cols)))
+    for idx, col in enumerate(feature_cols):
+        with mxl_dist_cols[idx % len(mxl_dist_cols)]:
+            default_dist_idx = 0  # fixed by default for dummy-coded
+            dist_val = st.selectbox(
+                f"{col}" + (" (dummy)" if coding_map.get(col) == "dummy" else ""),
+                ["fixed", "normal", "lognormal"],
+                index=default_dist_idx,
+                key=f"dist_{col}",
+            )
+            # Apply distribution to all expanded columns from this variable
+            if coding_map.get(col) == "dummy" and col in _dummy_info:
+                for dc in _dummy_info[col]:
+                    dist_map[dc] = dist_val
+            else:
+                dist_map[col] = dist_val
+
+    st.markdown("**Estimation settings**")
+    ms1, ms2, ms3 = st.columns(3)
+    n_draws = ms1.slider("Halton draws", 20, 2000, 200, step=10, key="mxl_draws")
+    maxiter = ms2.slider("Max optimizer iterations", 20, 500, 200, step=10, key="mxl_maxiter")
+    est_seed = ms3.number_input("Estimation seed", min_value=1, value=123, step=1, key="mxl_seed")
+
+    # ── Correlation structure: Independent / Full / Selective ────────
+    _random_expanded = [c for c in expanded_feature_cols if dist_map.get(c, "fixed") != "fixed"]
+    mxl_correlated = False
+    mxl_correlation_groups: list[list[int]] | None = None
+
+    if len(_random_expanded) >= 2:
+        st.markdown("**Correlation structure**")
+        _corr_mode = st.radio(
+            "Random parameter correlations",
+            ["Independent", "Full correlation", "Selective (pick pairs)"],
+            horizontal=True,
+            key="mxl_corr_mode",
+            help="Independent: each random parameter varies independently. "
+                 "Full: all random parameters are correlated (Cholesky). "
+                 "Selective: choose specific pairs to correlate.",
+        )
+        if _corr_mode == "Full correlation":
+            mxl_correlated = True
+        elif _corr_mode == "Selective (pick pairs)":
+            _n_rand = len(_random_expanded)
+            _all_corr_pairs = [
+                (i, j, _random_expanded[i], _random_expanded[j])
+                for i in range(_n_rand)
+                for j in range(i + 1, _n_rand)
+            ]
+            _selected_corr: list[tuple[int, int]] = []
+            _n_pair_cols = min(4, max(1, len(_all_corr_pairs)))
+            _pair_cols = st.columns(_n_pair_cols)
+            for _pidx, (_i, _j, _ni, _nj) in enumerate(_all_corr_pairs):
+                with _pair_cols[_pidx % _n_pair_cols]:
+                    if st.checkbox(f"{_ni} ↔ {_nj}", key=f"corr_{_ni}_{_nj}"):
+                        _selected_corr.append((_i, _j))
+            if _selected_corr:
+                mxl_correlated = True
+                _groups = _connected_components(_selected_corr, _n_rand)
+                mxl_correlation_groups = _groups
+                _group_labels = [
+                    [_random_expanded[k] for k in g] for g in _groups
+                ]
+                st.caption(f"Correlation blocks: {_group_labels}")
+            else:
+                st.caption("No pairs selected — using independent structure.")
+    elif len(_random_expanded) == 1:
+        st.caption("Only one random parameter — correlation not applicable.")
+
+    n_classes = 2
+    n_starts = 10
+
+else:  # latent_class
+    st.caption(
+        "Assumes Q discrete segments of respondents, each with distinct "
+        "fixed preferences. Useful for market segmentation."
+    )
+    ls1, ls2, ls3, ls4 = st.columns(4)
+    n_classes = ls1.slider("Number of classes (Q)", 2, 5, 2, key="lc_classes")
+    n_starts = ls2.slider("Random starts", 5, 20, 10, key="lc_starts")
+    maxiter = ls3.slider("Max optimizer iterations", 20, 500, 200, step=10, key="lc_maxiter")
+    est_seed = ls4.number_input("Estimation seed", min_value=1, value=123, step=1, key="lc_seed")
+
+    # Membership covariates: columns constant within each respondent
+    st.markdown("**Membership variables (demographics)**")
+    st.caption(
+        "Optionally select individual-level covariates that explain class membership. "
+        "Only columns that are constant within each respondent are shown."
+    )
+    _candidate_membership_cols = [
+        c for c in columns
+        if c not in structural_cols
+        and c not in expanded_feature_cols
+        and c not in feature_cols
+    ]
+    # Filter to columns constant within respondent groups
+    _constant_cols: list[str] = []
+    for c in _candidate_membership_cols:
+        try:
+            if df.groupby(id_col)[c].nunique().max() == 1:
+                _constant_cols.append(c)
+        except Exception:
+            pass
+    lc_membership_cols: list[str] = st.multiselect(
+        "Select membership covariates",
+        options=_constant_cols,
+        default=[],
+        key="lc_membership_cols",
+        help="These variables enter the class membership function. "
+             "They must be constant within each respondent (e.g. age, income, gender).",
+    )
+
+    for col in expanded_feature_cols:
+        dist_map[col] = "fixed"
+    n_draws = 1
+
+# ── Defaults for variables not set by every model type branch ─────
+if model_type not in ("mixed", "gmnl"):
+    mxl_correlated = False
+    mxl_correlation_groups = None
+if model_type != "latent_class":
+    lc_membership_cols = []
+
+# ── Interaction terms (N-way, any columns) — all models ──────────
+st.divider()
+st.subheader("Interaction terms (optional)")
+st.caption(
+    "Add interaction terms by multiplying 2 or more columns together. "
+    "Columns can be attributes, demographics, or any numeric column. "
+    "Works with all model types."
+)
+
+# Available columns: expanded feature cols + all other numeric non-structural columns
+_interaction_available_cols = list(expanded_feature_cols)
+for c in columns:
+    if (
+        c not in structural_cols
+        and c not in _interaction_available_cols
+        and c not in feature_cols
+        and pd.api.types.is_numeric_dtype(df[c])
+    ):
+        _interaction_available_cols.append(c)
+
+# Session state for interaction terms
+if "interaction_terms" not in st.session_state:
+    st.session_state.interaction_terms = []
+
+# Add new interaction term via form
+with st.form("add_interaction_form", clear_on_submit=True):
+    _inter_cols = st.multiselect(
+        "Select columns to interact",
+        options=_interaction_available_cols,
+        default=[],
+        key="new_interaction_cols",
+        help="Pick 2 or more columns. Their product will be added as a new variable.",
+    )
+    _submitted = st.form_submit_button("Add interaction term")
+    if _submitted:
+        if len(_inter_cols) < 2:
+            st.warning("Select at least 2 columns for an interaction term.")
+        else:
+            new_term = tuple(_inter_cols)
+            if new_term not in st.session_state.interaction_terms:
+                st.session_state.interaction_terms.append(new_term)
+                st.rerun()
+            else:
+                st.info("This interaction term already exists.")
+
+# Display existing terms with remove buttons
+_het_interactions: list[InteractionTerm] = []
+if st.session_state.interaction_terms:
+    st.markdown("**Current interaction terms:**")
+    _terms_to_keep: list[tuple] = []
+    for idx, term in enumerate(st.session_state.interaction_terms):
+        label = " × ".join(term)
+        c_label, c_remove = st.columns([4, 1])
+        c_label.markdown(f"- `{label}`")
+        if c_remove.button("Remove", key=f"remove_inter_{idx}"):
+            pass  # skip this term
+        else:
+            _terms_to_keep.append(term)
+            _het_interactions.append(InteractionTerm(columns=term))
+    if len(_terms_to_keep) != len(st.session_state.interaction_terms):
+        st.session_state.interaction_terms = _terms_to_keep
+        st.rerun()
+    st.caption(f"{len(_het_interactions)} interaction term(s) configured.")
+
+# ── Sidebar: model history count ──────────────────────────────────
+history: list[dict] = st.session_state.model_history
+if history:
+    st.sidebar.divider()
+    st.sidebar.metric("Saved models", len(history))
+    st.sidebar.markdown("**Model history**")
+    for entry in history:
+        st.sidebar.caption(f"- {entry.get('label', 'model')}")
+
+
+# ── helpers for result display ────────────────────────────────────
+def _significance(p: float) -> str:
+    if pd.isna(p):
+        return ""
+    if p < 0.001:
+        return "***"
+    if p < 0.01:
+        return "**"
+    if p < 0.05:
+        return "*"
+    if p < 0.1:
+        return "."
+    return ""
+
+
+def _show_results(estimation, run_label: str, header_suffix: str = "") -> None:
+    """Render fit metrics, parameter table, and download buttons."""
+    if estimation.success:
+        st.success(f"Converged: {estimation.message}")
+    else:
+        st.warning(f"Did not converge: {estimation.message}")
+
+    st.markdown(f"#### Model fit{header_suffix}")
+    m1, m2, m3, m4, m5 = st.columns(5)
+    m1.metric("Log-Likelihood", f"{estimation.log_likelihood:,.3f}")
+    m2.metric("AIC", f"{estimation.aic:,.2f}")
+    m3.metric("BIC", f"{estimation.bic:,.2f}")
+    m4.metric("Iterations", f"{estimation.optimizer_iterations}")
+    m5.metric("Runtime (s)", f"{estimation.runtime_seconds:.2f}")
+
+    st.markdown(f"#### Parameter estimates{header_suffix}")
+    display_df = estimation.estimates.copy()
+    if "p_value" in display_df.columns:
+        display_df["sig"] = display_df["p_value"].apply(_significance)
+    display_df = display_df.drop(columns=["theta_index"], errors="ignore")
+    st.dataframe(display_df, use_container_width=True, hide_index=True)
+    st.caption("Significance codes: *** p<0.001, ** p<0.01, * p<0.05, . p<0.1")
+
+    # Show covariance and correlation matrices for correlated MMNL
+    if getattr(estimation, "covariance_matrix", None) is not None:
+        names = estimation.random_param_names or []
+        st.markdown(f"#### Covariance matrix (random parameters){header_suffix}")
+        cov_df = pd.DataFrame(estimation.covariance_matrix, index=names, columns=names)
+        st.dataframe(cov_df, use_container_width=True)
+
+        st.markdown(f"#### Correlation matrix (random parameters){header_suffix}")
+        cor_df = pd.DataFrame(estimation.correlation_matrix, index=names, columns=names)
+        st.dataframe(cor_df, use_container_width=True)
+
+    d1, d2 = st.columns(2)
+    with d1:
+        csv_bytes = estimation.estimates.to_csv(index=False).encode("utf-8")
+        st.download_button(
+            label="Download estimates CSV",
+            data=csv_bytes,
+            file_name=f"{run_label}_estimates.csv",
+            mime="text/csv",
+        )
+    with d2:
+        summary_bytes = json.dumps(estimation.summary_dict(), indent=2, default=str).encode("utf-8")
+        st.download_button(
+            label="Download summary JSON",
+            data=summary_bytes,
+            file_name=f"{run_label}_summary.json",
+            mime="application/json",
+        )
+
+
+def _show_lc_results(estimation, run_label: str) -> None:
+    """Render Latent Class specific results."""
+    import plotly.express as px
+
+    if estimation.success:
+        st.success(f"Converged: {estimation.message}")
+    else:
+        st.warning(f"Did not converge: {estimation.message}")
+
+    # Fit metrics
+    st.markdown("#### Model fit")
+    m1, m2, m3, m4, m5 = st.columns(5)
+    m1.metric("Log-Likelihood", f"{estimation.log_likelihood:,.3f}")
+    m2.metric("AIC", f"{estimation.aic:,.2f}")
+    m3.metric("BIC", f"{estimation.bic:,.2f}")
+    m4.metric("Iterations", f"{estimation.optimizer_iterations}")
+    m5.metric("Runtime (s)", f"{estimation.runtime_seconds:.2f}")
+
+    # Class membership probabilities
+    st.markdown("#### Class membership probabilities")
+    pi_df = pd.DataFrame({
+        "Class": [f"Class {i+1}" for i in range(estimation.n_classes)],
+        "Probability": estimation.class_probabilities,
+    })
+    c1, c2 = st.columns([1, 1])
+    with c1:
+        st.dataframe(pi_df, use_container_width=True, hide_index=True)
+    with c2:
+        fig_pie = px.pie(pi_df, names="Class", values="Probability", title="Class Shares")
+        st.plotly_chart(fig_pie, use_container_width=True)
+
+    # Membership coefficients (if covariates were used)
+    if getattr(estimation, "membership_estimates", None) is not None:
+        st.markdown("#### Membership function coefficients")
+        st.caption(
+            "Coefficients explaining class membership probabilities "
+            "(relative to Class 1 as reference)."
+        )
+        mem_est = estimation.membership_estimates
+        mem_pivot = mem_est.pivot(index="variable", columns="class_id", values="estimate")
+        mem_pivot.columns = [f"Class {c}" for c in mem_pivot.columns]
+        st.dataframe(mem_pivot, use_container_width=True)
+
+    # Class-specific parameter estimates
+    st.markdown("#### Class-specific parameter estimates")
+    class_est = estimation.class_estimates
+    pivot = class_est.pivot(index="parameter", columns="class_id", values="estimate")
+    pivot.columns = [f"Class {c}" for c in pivot.columns]
+    st.dataframe(pivot, use_container_width=True)
+
+    # Coefficient comparison plot
+    st.markdown("#### Per-class coefficient comparison")
+    fig_bar = px.bar(
+        class_est, x="parameter", y="estimate",
+        color=class_est["class_id"].astype(str), barmode="group",
+        labels={"estimate": "Coefficient", "parameter": "Variable", "color": "Class"},
+        title="Coefficient Estimates by Class",
+    )
+    fig_bar.update_layout(legend_title_text="Class")
+    st.plotly_chart(fig_bar, use_container_width=True)
+
+    # Posterior class membership
+    st.markdown("#### Posterior class membership")
+    posterior = estimation.posterior_probs
+    assigned_class = posterior.idxmax(axis=1)
+    class_counts = assigned_class.value_counts().sort_index()
+
+    c1, c2 = st.columns(2)
+    with c1:
+        fig_hist = px.histogram(
+            posterior.max(axis=1), nbins=30,
+            labels={"value": "Max posterior probability", "count": "Count"},
+            title="Distribution of max posterior probability",
+        )
+        st.plotly_chart(fig_hist, use_container_width=True)
+    with c2:
+        count_df = pd.DataFrame({"Class": class_counts.index, "Count": class_counts.values})
+        fig_count = px.bar(count_df, x="Class", y="Count", title="Assigned class counts")
+        st.plotly_chart(fig_count, use_container_width=True)
+
+    # Full parameter table
+    st.markdown("#### All parameter estimates")
+    st.dataframe(estimation.estimates, use_container_width=True, hide_index=True)
+
+    # Downloads
+    st.markdown("#### Export")
+    d1, d2, d3 = st.columns(3)
+    with d1:
+        csv_bytes = estimation.estimates.to_csv(index=False).encode("utf-8")
+        st.download_button(
+            label="Download estimates CSV",
+            data=csv_bytes,
+            file_name=f"{run_label}_estimates.csv",
+            mime="text/csv",
+        )
+    with d2:
+        post_csv = posterior.to_csv(index=False).encode("utf-8")
+        st.download_button(
+            label="Download class assignments CSV",
+            data=post_csv,
+            file_name=f"{run_label}_posterior.csv",
+            mime="text/csv",
+        )
+    with d3:
+        summary_bytes = json.dumps(estimation.summary_dict(), indent=2, default=str).encode("utf-8")
+        st.download_button(
+            label="Download summary JSON",
+            data=summary_bytes,
+            file_name=f"{run_label}_summary.json",
+            mime="application/json",
+        )
+
+
+# ── 4  Run estimation ────────────────────────────────────────────
+st.divider()
+st.subheader("4. Run estimation")
+
+if st.button("Run Estimation", type="primary", use_container_width=True):
+    # Build VariableSpec list from original feature columns.
+    # For dummy-coded variables, use the original column name as a placeholder;
+    # the backend will expand them using dummy_codings.
+    # For continuous variables, use the column directly.
+    variables = []
+    for col in feature_cols:
+        if coding_map[col] == "dummy":
+            # Placeholder: backend replaces with expanded dummies.
+            # Distribution from the UI dist selector applies to all dummies.
+            dummy_expanded = _dummy_info[col]
+            dist = dist_map.get(dummy_expanded[0], "fixed") if dummy_expanded else "fixed"
+            variables.append(VariableSpec(name=col, column=col, distribution=dist))
+        else:
+            dist = dist_map.get(col, "fixed")
+            variables.append(VariableSpec(name=col, column=col, distribution=dist))
+
+    # Build FullModelSpec — one object captures everything
+    full_spec = FullModelSpec(
+        id_col=id_col,
+        task_col=task_col,
+        alt_col=alt_col,
+        choice_col=choice_col,
+        variables=variables,
+        model_type=model_type,
+        dummy_codings=_dummy_codings,
+        interactions=_het_interactions,
+        correlated=mxl_correlated,
+        correlation_groups=mxl_correlation_groups,
+        bws_worst_col=bws_worst_col if bws_mode else None,
+        estimate_lambda_w=bws_estimate_lambda_w if bws_mode else True,
+        n_classes=int(n_classes),
+        membership_cols=lc_membership_cols if lc_membership_cols else None,
+        n_draws=int(n_draws),
+        maxiter=int(maxiter),
+        seed=int(est_seed),
+        n_starts=int(n_starts),
+    )
+
+    # Pass the original DataFrame; backend handles dummy expansion
+    est_df = df
+
+    # Build spinner message
+    if model_type == "latent_class":
+        spinner_msg = f"Estimating {n_classes}-class model with {n_starts} random starts..."
+    else:
+        spinner_msg = "Estimating model — this may take a minute..."
+
+    with st.spinner(spinner_msg):
+        try:
+            result = estimate_from_spec(df=est_df, spec=full_spec)
+        except Exception as exc:
+            st.error(f"Estimation failed: {exc}")
+            st.exception(exc)
+            st.stop()
+
+    estimation = result.estimation
+
+    # Auto-generate run label
+    prefix_map = {"conditional": "CL", "mixed": "MXL", "gmnl": "GMNL", "latent_class": "LC"}
+    prefix = prefix_map[model_type]
+    existing_count = sum(1 for h in history if h.get("model_type") == model_type)
+    if model_type == "latent_class":
+        run_label = f"{prefix} Run {existing_count + 1} (Q={n_classes})"
+    else:
+        run_label = f"{prefix} Run {existing_count + 1}"
+
+    # Store as current result (use expanded spec/df for Results page compatibility)
+    st.session_state.model_results = {
+        "spec": result.expanded_spec or full_spec.to_model_spec(),
+        "full_spec": full_spec,
+        "model_type": model_type,
+        "estimation": estimation,
+        "label": run_label,
+        "expanded_df": result.expanded_df,
+    }
+
+    # Append to history for comparison page
+    st.session_state.model_history.append({
+        "label": run_label,
+        "model_type": model_type,
+        "spec": result.expanded_spec or full_spec.to_model_spec(),
+        "full_spec": full_spec,
+        "estimation": estimation,
+    })
+
+    # Also store LC-specific result
+    if model_type == "latent_class":
+        st.session_state.lc_result = {
+            "estimation": estimation,
+            "label": run_label,
+        }
+
+    st.success(f"Model **{run_label}** estimated successfully.")
+
+    # Show results with appropriate display
+    if model_type == "latent_class":
+        _show_lc_results(estimation, run_label)
+    else:
+        _show_results(estimation, run_label)
+
+# ── Show last run results on rerun ────────────────────────────────
+elif st.session_state.model_results is not None:
+    res = st.session_state.model_results
+    est = res["estimation"]
+    label = res.get("label", "model")
+    if res.get("model_type") == "latent_class":
+        _show_lc_results(est, label)
+    else:
+        _show_results(est, label, header_suffix=" (last run)")
+
+# ── LC: BIC comparison tool ───────────────────────────────────────
+if model_type == "latent_class":
+    st.divider()
+    st.subheader("Optimal number of classes")
+    st.markdown(
+        "Automatically estimate models with Q = 2, 3, 4, 5 classes and compare BIC."
+    )
+
+    if st.button("Run BIC comparison (Q = 2..5)", use_container_width=True):
+        import plotly.express as px
+
+        bic_variables = []
+        for col in feature_cols:
+            bic_variables.append(VariableSpec(name=col, column=col, distribution="fixed"))
+
+        bic_rows: list[dict] = []
+        best_bic = float("inf")
+        best_q = 2
+        progress = st.progress(0, text="Starting class comparison...")
+
+        for i, q in enumerate([2, 3, 4, 5]):
+            progress.progress(i / 4, text=f"Estimating Q = {q}...")
+            bic_spec = FullModelSpec(
+                id_col=id_col, task_col=task_col, alt_col=alt_col,
+                choice_col=choice_col, variables=bic_variables,
+                model_type="latent_class",
+                dummy_codings=_dummy_codings,
+                n_classes=q, n_starts=int(n_starts),
+                maxiter=int(maxiter), seed=int(est_seed),
+                membership_cols=lc_membership_cols if lc_membership_cols else None,
+                bws_worst_col=bws_worst_col if bws_mode else None,
+                estimate_lambda_w=bws_estimate_lambda_w if bws_mode else True,
+            )
+            try:
+                result = estimate_from_spec(df=df, spec=bic_spec)
+                est = result.estimation
+                bic_rows.append({
+                    "Q": q, "Log-Likelihood": round(est.log_likelihood, 3),
+                    "AIC": round(est.aic, 2), "BIC": round(est.bic, 2),
+                    "Parameters": est.n_parameters, "Converged": est.success,
+                })
+                if est.bic < best_bic:
+                    best_bic = est.bic
+                    best_q = q
+            except Exception as exc:
+                bic_rows.append({
+                    "Q": q, "Log-Likelihood": None, "AIC": None,
+                    "BIC": None, "Parameters": None, "Converged": False,
+                })
+                st.warning(f"Q = {q} failed: {exc}")
+
+        progress.progress(1.0, text="Done!")
+
+        bic_df = pd.DataFrame(bic_rows)
+        st.session_state.lc_bic_comparison = bic_df
+        st.session_state.lc_best_q = best_q
+
+        st.dataframe(bic_df, use_container_width=True, hide_index=True)
+
+        valid = bic_df.dropna(subset=["BIC"])
+        if not valid.empty:
+            fig_bic = px.line(valid, x="Q", y="BIC", markers=True, title="BIC by Number of Classes")
+            fig_bic.add_vline(x=best_q, line_dash="dash", line_color="green",
+                              annotation_text=f"Best Q = {best_q}")
+            st.plotly_chart(fig_bic, use_container_width=True)
+
+        st.info(f"Recommended number of classes: **Q = {best_q}**")
+
+    elif st.session_state.get("lc_bic_comparison") is not None:
+        import plotly.express as px
+
+        bic_df = st.session_state.lc_bic_comparison
+        best_q = st.session_state.lc_best_q
+        st.dataframe(bic_df, use_container_width=True, hide_index=True)
+        valid = bic_df.dropna(subset=["BIC"])
+        if not valid.empty:
+            fig_bic = px.line(valid, x="Q", y="BIC", markers=True, title="BIC by Number of Classes")
+            fig_bic.add_vline(x=best_q, line_dash="dash", line_color="green",
+                              annotation_text=f"Best Q = {best_q}")
+            st.plotly_chart(fig_bic, use_container_width=True)
+        st.info(f"Recommended number of classes: **Q = {best_q}**")
+
+# ── Show saved model history ──────────────────────────────────────
+if st.session_state.model_history:
+    st.divider()
+    st.subheader("Saved model runs")
+    for i, entry in enumerate(st.session_state.model_history, 1):
+        est = entry["estimation"]
+        st.markdown(
+            f"**{i}. {entry.get('label', 'model')}** ({entry.get('model_type', '?')}) "
+            f"— LL: {est.log_likelihood:.3f}, AIC: {est.aic:.2f}, "
+            f"BIC: {est.bic:.2f}"
+        )

scripts/test_e2e.py

@@ -0,0 +1,1246 @@
+"""End-to-end test script for the dce_analyzer backend.
+
+Run from project root:
+    python scripts/test_e2e.py
+"""
+
+from __future__ import annotations
+
+import sys
+import traceback
+from pathlib import Path
+
+# Ensure src/ is importable
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+
+import numpy as np
+import pandas as pd
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+_results: list[tuple[str, bool, str]] = []
+
+
+def _run(name: str, fn):
+    """Run *fn* and record PASS / FAIL."""
+    try:
+        fn()
+        _results.append((name, True, ""))
+        print(f"  PASS  {name}")
+    except Exception as exc:
+        msg = f"{exc.__class__.__name__}: {exc}"
+        _results.append((name, False, msg))
+        print(f"  FAIL  {name}")
+        traceback.print_exc()
+        print()
+
+
+# ===================================================================
+# 1. Import all backend modules
+# ===================================================================
+def test_imports():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.simulate import generate_simulated_dce
+    from dce_analyzer.data import prepare_choice_tensors, ChoiceTensors
+    from dce_analyzer.model import (
+        MixedLogitEstimator,
+        ConditionalLogitEstimator,
+        EstimationResult,
+    )
+    from dce_analyzer.latent_class import LatentClassEstimator, LatentClassResult
+    from dce_analyzer.pipeline import estimate_dataframe, PipelineResult
+    from dce_analyzer.wtp import compute_wtp
+    from dce_analyzer.bootstrap import run_bootstrap, BootstrapResult
+    from dce_analyzer.format_converter import (
+        detect_format,
+        wide_to_long,
+        infer_structure,
+        normalize_choice_column,
+        ColumnInference,
+    )
+    from dce_analyzer.apollo import APOLLO_DATASETS
+    # all imported without error
+
+
+_run("1. Import all backend modules", test_imports)
+
+
+# ===================================================================
+# 2. Generate simulated data
+# ===================================================================
+sim_output = None
+
+
+def test_simulate():
+    global sim_output
+    from dce_analyzer.simulate import generate_simulated_dce
+
+    sim_output = generate_simulated_dce(
+        n_individuals=100, n_tasks=4, n_alts=3, seed=42
+    )
+    df = sim_output.data
+    assert isinstance(df, pd.DataFrame), "Expected DataFrame"
+    assert len(df) == 100 * 4 * 3, f"Expected 1200 rows, got {len(df)}"
+    for col in ["respondent_id", "task_id", "alternative", "choice",
+                "price", "time", "comfort", "reliability"]:
+        assert col in df.columns, f"Missing column: {col}"
+    assert isinstance(sim_output.true_parameters, dict)
+    assert len(sim_output.true_parameters) > 0
+
+
+_run("2. Generate simulated data (100 ind, 4 tasks, 3 alts)", test_simulate)
+
+
+# ===================================================================
+# 3. Conditional Logit estimation
+# ===================================================================
+cl_result = None
+
+
+def test_conditional_logit():
+    global cl_result
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="conditional", maxiter=200, seed=42
+    )
+    cl_result = result
+    est = result.estimation
+    assert est.success, f"CL did not converge: {est.message}"
+    assert est.n_parameters == 4
+    assert est.n_observations == 100 * 4  # 400 choice tasks
+    assert not est.estimates.empty
+    assert "estimate" in est.estimates.columns
+
+
+_run("3. Conditional Logit estimation", test_conditional_logit)
+
+
+# ===================================================================
+# 4. Mixed Logit estimation (n_draws=50)
+# ===================================================================
+mxl_result = None
+
+
+def test_mixed_logit():
+    global mxl_result
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed", maxiter=200, seed=42
+    )
+    mxl_result = result
+    est = result.estimation
+    # 2 normal (mu+sd each) + 2 fixed = 6 params
+    assert est.n_parameters == 6, f"Expected 6 params, got {est.n_parameters}"
+    assert not est.estimates.empty
+    # Should have mu_price, sd_price, mu_time, sd_time, beta_comfort, beta_reliability
+    param_names = set(est.estimates["parameter"])
+    for expected in ["mu_price", "sd_price", "mu_time", "sd_time",
+                     "beta_comfort", "beta_reliability"]:
+        assert expected in param_names, f"Missing param: {expected}"
+
+
+_run("4. Mixed Logit estimation (n_draws=50)", test_mixed_logit)
+
+
+# ===================================================================
+# 5. Latent Class estimation (n_classes=2, n_starts=3)
+# ===================================================================
+lc_result = None
+
+
+def test_latent_class():
+    global lc_result
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        n_classes=2,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="latent_class",
+        maxiter=200, seed=42, n_classes=2, n_starts=3,
+    )
+    lc_result = result
+    est = result.estimation
+    assert est.n_classes == 2
+    assert len(est.class_probabilities) == 2
+    assert abs(sum(est.class_probabilities) - 1.0) < 1e-4, "Class probs must sum to 1"
+    assert not est.estimates.empty
+    assert not est.class_estimates.empty
+    assert not est.posterior_probs.empty
+    assert est.posterior_probs.shape[1] == 2  # two class columns
+
+
+_run("5. Latent Class estimation (n_classes=2, n_starts=3)", test_latent_class)
+
+
+# ===================================================================
+# 6. WTP computation
+# ===================================================================
+def test_wtp():
+    from dce_analyzer.wtp import compute_wtp
+
+    # Use CL result (EstimationResult) for WTP
+    wtp_df = compute_wtp(cl_result.estimation, cost_variable="price")
+    assert isinstance(wtp_df, pd.DataFrame)
+    assert len(wtp_df) == 3  # time, comfort, reliability (3 non-cost attrs)
+    assert "wtp_estimate" in wtp_df.columns
+    assert "wtp_std_error" in wtp_df.columns
+    assert "wtp_ci_lower" in wtp_df.columns
+    assert "wtp_ci_upper" in wtp_df.columns
+    # WTP values should be finite
+    for _, row in wtp_df.iterrows():
+        assert np.isfinite(row["wtp_estimate"]), f"Non-finite WTP for {row['attribute']}"
+
+
+_run("6. WTP computation (CL result)", test_wtp)
+
+
+# ===================================================================
+# 7. Bootstrap (n_boot=10)
+# ===================================================================
+def test_bootstrap():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.bootstrap import run_bootstrap
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+    )
+    boot = run_bootstrap(
+        df=sim_output.data, spec=spec, model_type="conditional",
+        n_replications=10, maxiter=100, seed=42,
+    )
+    assert boot.n_replications == 10
+    assert boot.n_successful >= 2, f"Only {boot.n_successful} succeeded"
+    assert len(boot.param_names) == 4
+    assert boot.estimates_matrix.shape == (boot.n_successful, 4)
+    summary = boot.summary_dataframe()
+    assert isinstance(summary, pd.DataFrame)
+    assert len(summary) == 4
+
+
+_run("7. Bootstrap (n_boot=10, conditional logit)", test_bootstrap)
+
+
+# ===================================================================
+# 8. Wide-to-long conversion
+# ===================================================================
+def test_wide_to_long():
+    from dce_analyzer.format_converter import detect_format, wide_to_long
+
+    # Create a small wide-format dataset
+    wide_df = pd.DataFrame({
+        "id": [1, 1, 2, 2],
+        "choice": [1, 2, 1, 3],
+        "price_1": [10, 20, 15, 25],
+        "price_2": [12, 22, 17, 27],
+        "price_3": [14, 24, 19, 29],
+        "time_1": [30, 40, 35, 45],
+        "time_2": [32, 42, 37, 47],
+        "time_3": [34, 44, 39, 49],
+    })
+
+    fmt = detect_format(wide_df)
+    assert fmt == "wide", f"Expected 'wide', got '{fmt}'"
+
+    long_df = wide_to_long(
+        wide_df,
+        attribute_groups={
+            "price": ["price_1", "price_2", "price_3"],
+            "time": ["time_1", "time_2", "time_3"],
+        },
+        id_col="id",
+        choice_col="choice",
+    )
+    assert isinstance(long_df, pd.DataFrame)
+    # 4 rows * 3 alts = 12 rows
+    assert len(long_df) == 12, f"Expected 12 rows, got {len(long_df)}"
+    assert "alternative" in long_df.columns
+    assert "choice" in long_df.columns
+    assert "price" in long_df.columns
+    assert "time" in long_df.columns
+    # Each task should have exactly one chosen alt
+    for (rid, tid), grp in long_df.groupby(["respondent_id", "task_id"]):
+        assert grp["choice"].sum() == 1, f"Task ({rid},{tid}) has {grp['choice'].sum()} choices"
+
+    # Test detect_format on long data
+    fmt2 = detect_format(long_df)
+    assert fmt2 == "long", f"Expected 'long' for converted data, got '{fmt2}'"
+
+
+_run("8. Wide-to-long conversion", test_wide_to_long)
+
+
+# ===================================================================
+# 9. Additional checks: infer_structure, normalize_choice_column
+# ===================================================================
+def test_infer_and_normalize():
+    from dce_analyzer.format_converter import infer_structure, normalize_choice_column
+
+    df = sim_output.data
+    inference = infer_structure(df)
+    assert inference.id_col is not None, "Should detect id column"
+    assert inference.choice_col is not None, "Should detect choice column"
+
+    # Test normalize_choice_column (already binary -- should be no-op)
+    normalized = normalize_choice_column(df, "choice", "alternative")
+    assert set(normalized["choice"].unique()) <= {0, 1}
+
+
+_run("9. infer_structure & normalize_choice_column", test_infer_and_normalize)
+
+
+# ===================================================================
+# 10. LatentClassResult.summary_dict()
+# ===================================================================
+def test_lc_summary():
+    est = lc_result.estimation
+    sd = est.summary_dict()
+    assert "n_classes" in sd
+    assert "class_probabilities" in sd
+    assert sd["n_classes"] == 2
+
+
+_run("10. LatentClassResult.summary_dict()", test_lc_summary)
+
+
+# ===================================================================
+# 11. Full correlated MMNL (backward compat)
+# ===================================================================
+def test_full_correlated_mxl():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed",
+        maxiter=200, seed=42, correlated=True,
+    )
+    est = result.estimation
+    assert est.covariance_matrix is not None, "Expected covariance matrix"
+    assert est.covariance_matrix.shape == (2, 2), f"Expected 2x2 cov, got {est.covariance_matrix.shape}"
+    assert est.correlation_matrix is not None
+
+
+_run("11. Full correlated MMNL (backward compat)", test_full_correlated_mxl)
+
+
+# ===================================================================
+# 12. Selective correlated MMNL (block-diagonal Cholesky)
+# ===================================================================
+def test_selective_correlated_mxl():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="normal"),
+            VariableSpec(name="reliability", column="reliability", distribution="normal"),
+        ],
+        n_draws=50,
+    )
+    # Correlate price-time (group [0,1]) and comfort-reliability (group [2,3])
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed",
+        maxiter=200, seed=42,
+        correlation_groups=[[0, 1], [2, 3]],
+    )
+    est = result.estimation
+    assert est.covariance_matrix is not None, "Expected covariance matrix"
+    assert est.covariance_matrix.shape == (4, 4)
+    # Off-block elements should be zero (price-comfort, price-reliability, etc.)
+    cov = est.covariance_matrix
+    assert abs(cov[0, 2]) < 1e-8, f"Expected 0 cov(price,comfort), got {cov[0,2]}"
+    assert abs(cov[0, 3]) < 1e-8, f"Expected 0 cov(price,reliability), got {cov[0,3]}"
+    assert abs(cov[1, 2]) < 1e-8, f"Expected 0 cov(time,comfort), got {cov[1,2]}"
+    assert abs(cov[1, 3]) < 1e-8, f"Expected 0 cov(time,reliability), got {cov[1,3]}"
+    # Within-block elements should be non-zero
+    assert abs(cov[0, 1]) > 1e-10 or True  # may be zero by chance, just check shape
+
+
+_run("12. Selective correlated MMNL (block-diagonal)", test_selective_correlated_mxl)
+
+
+# ===================================================================
+# 13. Selective with standalone random params
+# ===================================================================
+def test_selective_with_standalone():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="normal"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    # Only correlate price-time, comfort is standalone random
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed",
+        maxiter=200, seed=42,
+        correlation_groups=[[0, 1]],
+    )
+    est = result.estimation
+    assert est.covariance_matrix is not None
+    assert est.covariance_matrix.shape == (3, 3)
+    cov = est.covariance_matrix
+    # comfort (index 2) is standalone: zero cross-cov with price/time
+    assert abs(cov[0, 2]) < 1e-8, f"Expected 0 cov(price,comfort), got {cov[0,2]}"
+    assert abs(cov[1, 2]) < 1e-8, f"Expected 0 cov(time,comfort), got {cov[1,2]}"
+    # n_parameters: 3 mu + 3 chol(price-time) + 1 sd(comfort) + 1 fixed = 8
+    assert est.n_parameters == 8, f"Expected 8 params, got {est.n_parameters}"
+
+
+_run("13. Selective with standalone random params", test_selective_with_standalone)
+
+
+# ===================================================================
+# 14. Create BWS simulated data
+# ===================================================================
+bws_df = None
+
+
+def test_create_bws_data():
+    """Create BWS data by adding a 'worst' column to simulated DCE data."""
+    global bws_df
+    df = sim_output.data.copy()
+    # J=3 alts per task. For each task, pick the alt with LOWEST utility-like
+    # score as worst. Use negative of choice to ensure worst != best.
+    rng = np.random.default_rng(99)
+    worst_rows = []
+    for (rid, tid), grp in df.groupby(["respondent_id", "task_id"]):
+        best_alt = grp.loc[grp["choice"] == 1, "alternative"].values[0]
+        non_best = grp[grp["alternative"] != best_alt]
+        # Pick random non-best as worst
+        worst_alt = non_best["alternative"].values[rng.integers(len(non_best))]
+        for _, row in grp.iterrows():
+            worst_rows.append(1 if row["alternative"] == worst_alt else 0)
+    df["worst"] = worst_rows
+    # Verify: each task has exactly 1 worst, 1 best, and worst != best
+    for (rid, tid), grp in df.groupby(["respondent_id", "task_id"]):
+        assert grp["choice"].sum() == 1, "Exactly one best per task"
+        assert grp["worst"].sum() == 1, "Exactly one worst per task"
+        best_idx = grp.loc[grp["choice"] == 1].index[0]
+        worst_idx = grp.loc[grp["worst"] == 1].index[0]
+        assert best_idx != worst_idx, "worst != best"
+    bws_df = df
+    assert "worst" in bws_df.columns
+
+
+_run("14. Create BWS simulated data", test_create_bws_data)
+
+
+# ===================================================================
+# 15. BWS + Conditional Logit
+# ===================================================================
+def test_bws_clogit():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+    )
+    result = estimate_dataframe(
+        df=bws_df, spec=spec, model_type="conditional",
+        maxiter=200, seed=42,
+        bws_worst_col="worst", estimate_lambda_w=True,
+    )
+    est = result.estimation
+    assert est.success, f"BWS CL did not converge: {est.message}"
+    # 4 betas + 1 lambda_w = 5 params
+    assert est.n_parameters == 5, f"Expected 5 params, got {est.n_parameters}"
+    # lambda_w should appear in estimates
+    param_names = set(est.estimates["parameter"])
+    assert "lambda_w (worst scale)" in param_names, f"Missing lambda_w param. Got: {param_names}"
+    # lambda_w should be positive
+    lw_row = est.estimates[est.estimates["parameter"] == "lambda_w (worst scale)"]
+    assert lw_row["estimate"].values[0] > 0, "lambda_w must be positive"
+
+
+_run("15. BWS + Conditional Logit", test_bws_clogit)
+
+
+# ===================================================================
+# 16. BWS + CLogit with lambda_w fixed (MaxDiff equivalent)
+# ===================================================================
+def test_bws_clogit_fixed_lw():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+    )
+    result = estimate_dataframe(
+        df=bws_df, spec=spec, model_type="conditional",
+        maxiter=200, seed=42,
+        bws_worst_col="worst", estimate_lambda_w=False,
+    )
+    est = result.estimation
+    assert est.success
+    # 4 betas only (no lambda_w)
+    assert est.n_parameters == 4, f"Expected 4 params, got {est.n_parameters}"
+    param_names = set(est.estimates["parameter"])
+    assert "lambda_w (worst scale)" not in param_names
+
+
+_run("16. BWS + CLogit fixed lambda_w (MaxDiff)", test_bws_clogit_fixed_lw)
+
+
+# ===================================================================
+# 17. BWS + Mixed Logit
+# ===================================================================
+def test_bws_mxl():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=bws_df, spec=spec, model_type="mixed",
+        maxiter=200, seed=42,
+        bws_worst_col="worst", estimate_lambda_w=True,
+    )
+    est = result.estimation
+    # 2 mu + 2 sd + 2 fixed + 1 lambda_w = 7
+    assert est.n_parameters == 7, f"Expected 7 params, got {est.n_parameters}"
+    param_names = set(est.estimates["parameter"])
+    assert "lambda_w (worst scale)" in param_names
+    assert "mu_price" in param_names
+    assert "sd_price" in param_names
+
+
+_run("17. BWS + Mixed Logit", test_bws_mxl)
+
+
+# ===================================================================
+# 18. BWS + GMNL
+# ===================================================================
+def test_bws_gmnl():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="fixed"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=bws_df, spec=spec, model_type="gmnl",
+        maxiter=200, seed=42,
+        bws_worst_col="worst", estimate_lambda_w=True,
+    )
+    est = result.estimation
+    # 1 mu + 1 sd + 3 fixed + 1 lambda_w + 3 GMNL(tau,sigma_tau,gamma) = 9
+    assert est.n_parameters == 9, f"Expected 9 params, got {est.n_parameters}"
+    param_names = set(est.estimates["parameter"])
+    assert "lambda_w (worst scale)" in param_names
+    assert "tau (scale mean)" in param_names
+
+
+_run("18. BWS + GMNL", test_bws_gmnl)
+
+
+# ===================================================================
+# 19. BWS + Latent Class
+# ===================================================================
+def test_bws_lc():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        n_classes=2,
+    )
+    result = estimate_dataframe(
+        df=bws_df, spec=spec, model_type="latent_class",
+        maxiter=200, seed=42, n_classes=2, n_starts=3,
+        bws_worst_col="worst", estimate_lambda_w=True,
+    )
+    est = result.estimation
+    assert est.n_classes == 2
+    assert len(est.class_probabilities) == 2
+    # Check lambda_w appears in estimates
+    lw_rows = est.estimates[est.estimates["parameter"].str.contains("lambda_w")]
+    assert len(lw_rows) > 0, "Missing lambda_w in LC estimates"
+
+
+_run("19. BWS + Latent Class", test_bws_lc)
+
+
+# ===================================================================
+# 20. Correlation inference (delta method SEs for cov/cor)
+# ===================================================================
+def test_correlation_inference():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed",
+        maxiter=200, seed=42, correlated=True,
+    )
+    est = result.estimation
+    # Covariance SE matrix should exist and match shape
+    assert est.covariance_se is not None, "Expected covariance_se"
+    assert est.covariance_se.shape == (2, 2), f"Expected 2x2, got {est.covariance_se.shape}"
+    # Correlation SE matrix
+    assert est.correlation_se is not None, "Expected correlation_se"
+    assert est.correlation_se.shape == (2, 2)
+    # Diagonal of correlation SE should be 0 (cor(x,x)=1, no variation)
+    for i in range(2):
+        assert est.correlation_se[i, i] < 1e-6, f"Diagonal cor SE should be ~0, got {est.correlation_se[i,i]}"
+    # Correlation test table
+    assert est.correlation_test is not None, "Expected correlation_test DataFrame"
+    assert len(est.correlation_test) == 1, "Expected 1 off-diagonal pair for 2 random params"
+    row = est.correlation_test.iloc[0]
+    assert row["param_1"] == "price"
+    assert row["param_2"] == "time"
+    assert not np.isnan(row["cor_std_error"]), "SE should not be NaN"
+    assert not np.isnan(row["z_stat"]), "z_stat should not be NaN"
+    assert not np.isnan(row["p_value"]), "p_value should not be NaN"
+    assert 0.0 <= row["p_value"] <= 1.0, f"p-value out of range: {row['p_value']}"
+
+
+_run("20. Correlation inference (delta method SEs for cov/cor)", test_correlation_inference)
+
+
+# ===================================================================
+# 21. FullModelSpec + estimate_from_spec
+# ===================================================================
+def test_full_model_spec():
+    from dce_analyzer.config import FullModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        model_type="mixed",
+        n_draws=50,
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    # Should produce the same kind of result as estimate_dataframe
+    assert est.n_parameters == 6, f"Expected 6 params, got {est.n_parameters}"
+    assert not est.estimates.empty
+    param_names = set(est.estimates["parameter"])
+    for expected in ["mu_price", "sd_price", "mu_time", "sd_time",
+                     "beta_comfort", "beta_reliability"]:
+        assert expected in param_names, f"Missing param: {expected}"
+    assert est.n_observations == 100 * 4
+
+
+_run("21. FullModelSpec + estimate_from_spec", test_full_model_spec)
+
+
+# ===================================================================
+# 22. Heterogeneity interactions with MMNL via FullModelSpec
+# ===================================================================
+def test_interactions_mmnl():
+    from dce_analyzer.config import FullModelSpec, InteractionTerm, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="fixed"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        model_type="mixed",
+        interactions=[
+            InteractionTerm(columns=("price", "income")),
+        ],
+        n_draws=50,
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    param_names = set(est.estimates["parameter"])
+    # Interaction term should appear as a fixed parameter
+    assert "beta_price_x_income" in param_names, (
+        f"Missing interaction param. Got: {param_names}"
+    )
+    # 1 mu + 1 sd (price) + 3 fixed (time, comfort, reliability) + 1 interaction = 6
+    assert est.n_parameters == 6, f"Expected 6 params, got {est.n_parameters}"
+
+
+_run("22. Heterogeneity interactions with MMNL (InteractionTerm)", test_interactions_mmnl)
+
+
+# ===================================================================
+# 23. GMNL + full correlation
+# ===================================================================
+def test_gmnl_full_correlation():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="gmnl",
+        maxiter=200, seed=42, correlated=True,
+    )
+    est = result.estimation
+    assert est.covariance_matrix is not None, "Expected covariance matrix for GMNL+correlated"
+    assert est.covariance_matrix.shape == (2, 2), (
+        f"Expected 2x2 cov, got {est.covariance_matrix.shape}"
+    )
+    assert est.correlation_matrix is not None
+    # GMNL params: 2 mu + chol(2)=3 + 2 fixed + 3 GMNL(tau,sigma_tau,gamma) = 10
+    assert est.n_parameters == 10, f"Expected 10 params, got {est.n_parameters}"
+    param_names = set(est.estimates["parameter"])
+    assert "tau (scale mean)" in param_names
+    assert "sigma_tau (scale SD)" in param_names
+    assert "gamma (mixing)" in param_names
+
+
+_run("23. GMNL + full correlation", test_gmnl_full_correlation)
+
+
+# ===================================================================
+# 24. GMNL + selective correlation
+# ===================================================================
+def test_gmnl_selective_correlation():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="normal"),
+            VariableSpec(name="comfort", column="comfort", distribution="normal"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    # Correlate price-time only; comfort is standalone random
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="gmnl",
+        maxiter=200, seed=42,
+        correlation_groups=[[0, 1]],
+    )
+    est = result.estimation
+    assert est.covariance_matrix is not None
+    assert est.covariance_matrix.shape == (3, 3)
+    cov = est.covariance_matrix
+    # comfort (index 2) is standalone: zero cross-cov with price/time
+    assert abs(cov[0, 2]) < 1e-8, f"Expected 0 cov(price,comfort), got {cov[0,2]}"
+    assert abs(cov[1, 2]) < 1e-8, f"Expected 0 cov(time,comfort), got {cov[1,2]}"
+    param_names = set(est.estimates["parameter"])
+    assert "tau (scale mean)" in param_names
+
+
+_run("24. GMNL + selective correlation", test_gmnl_selective_correlation)
+
+
+# ===================================================================
+# 25. BWS composable functions (bws_log_prob, standard_log_prob)
+# ===================================================================
+def test_bws_composable_functions():
+    import torch
+    from dce_analyzer.bws import bws_log_prob, standard_log_prob
+
+    # Create simple test tensors: 4 observations, 3 alternatives
+    n_obs, n_alts = 4, 3
+    torch.manual_seed(42)
+    utility = torch.randn(n_obs, n_alts)
+    y_best = torch.tensor([0, 1, 2, 0])   # chosen alt indices
+    y_worst = torch.tensor([2, 0, 1, 1])  # worst alt indices (different from best)
+
+    # Test standard_log_prob
+    log_p = standard_log_prob(utility, y_best, alt_dim=-1)
+    assert log_p.shape == (n_obs,), f"Expected shape ({n_obs},), got {log_p.shape}"
+    # Log-probabilities must be <= 0
+    assert (log_p <= 1e-6).all(), "Log-probabilities must be <= 0"
+    # Probabilities must sum to 1 across alternatives (verify via logsumexp)
+    log_all = torch.stack([
+        standard_log_prob(utility, torch.full((n_obs,), j), alt_dim=-1)
+        for j in range(n_alts)
+    ], dim=1)
+    prob_sums = torch.exp(log_all).sum(dim=1)
+    assert torch.allclose(prob_sums, torch.ones(n_obs), atol=1e-5), (
+        f"Probabilities don't sum to 1: {prob_sums}"
+    )
+
+    # Test bws_log_prob
+    lambda_w = 1.0
+    log_p_bws = bws_log_prob(utility, y_best, y_worst, lambda_w, alt_dim=-1)
+    assert log_p_bws.shape == (n_obs,), f"Expected shape ({n_obs},), got {log_p_bws.shape}"
+    assert (log_p_bws <= 1e-6).all(), "BWS log-probabilities must be <= 0"
+    # BWS log-prob should be less than standard (it's a product of two probs)
+    assert (log_p_bws <= log_p + 1e-6).all(), (
+        "BWS log-prob should be <= standard log-prob (product of two probs)"
+    )
+
+    # Test with lambda_w as tensor
+    lambda_w_tensor = torch.tensor(2.0)
+    log_p_bws2 = bws_log_prob(utility, y_best, y_worst, lambda_w_tensor, alt_dim=-1)
+    assert log_p_bws2.shape == (n_obs,)
+
+    # Test with 3D utility (simulating draws): (n_obs, n_draws, n_alts)
+    n_draws = 5
+    utility_3d = torch.randn(n_obs, n_draws, n_alts)
+    log_p_3d = standard_log_prob(utility_3d, y_best, alt_dim=-1)
+    assert log_p_3d.shape == (n_obs, n_draws), f"Expected ({n_obs},{n_draws}), got {log_p_3d.shape}"
+
+    log_p_bws_3d = bws_log_prob(utility_3d, y_best, y_worst, 1.0, alt_dim=-1)
+    assert log_p_bws_3d.shape == (n_obs, n_draws), (
+        f"Expected ({n_obs},{n_draws}), got {log_p_bws_3d.shape}"
+    )
+
+
+_run("25. BWS composable functions (bws_log_prob, standard_log_prob)", test_bws_composable_functions)
+
+
+# ===================================================================
+# 26. Heterogeneity interactions with Latent Class via FullModelSpec
+# ===================================================================
+def test_interactions_lc():
+    from dce_analyzer.config import FullModelSpec, InteractionTerm, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        model_type="latent_class",
+        interactions=[
+            InteractionTerm(columns=("price", "income")),
+        ],
+        n_classes=2,
+        n_starts=3,
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    assert est.n_classes == 2
+    # Interaction param should appear in estimates
+    has_interaction = any("price_x_income" in str(p) for p in est.estimates["parameter"])
+    assert has_interaction, (
+        f"Missing interaction param in LC estimates. Got: {list(est.estimates['parameter'])}"
+    )
+
+
+_run("26. Heterogeneity interactions with Latent Class (InteractionTerm)", test_interactions_lc)
+
+
+# ===================================================================
+# 27. FullModelSpec with dummy coding via estimate_from_spec
+# ===================================================================
+def test_dummy_coding_via_spec():
+    from dce_analyzer.config import DummyCoding, FullModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    # comfort has 2 unique values (0, 1) -> dummy with ref=0 -> one dummy comfort_L1
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        model_type="conditional",
+        dummy_codings=[
+            DummyCoding(column="comfort", ref_level=0),
+        ],
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    param_names = set(est.estimates["parameter"])
+    # comfort should be expanded: beta_comfort_L1 instead of beta_comfort
+    assert "beta_comfort_L1" in param_names, (
+        f"Missing dummy param beta_comfort_L1. Got: {param_names}"
+    )
+    # Original comfort should NOT appear
+    assert "beta_comfort" not in param_names, (
+        f"Original column should be replaced by dummy expansion. Got: {param_names}"
+    )
+    # price, time, reliability remain continuous
+    assert "beta_price" in param_names
+    assert "beta_time" in param_names
+    assert "beta_reliability" in param_names
+    # 3 continuous + 1 dummy = 4 params
+    assert est.n_parameters == 4, f"Expected 4 params, got {est.n_parameters}"
+
+
+_run("27. FullModelSpec with dummy coding via estimate_from_spec", test_dummy_coding_via_spec)
+
+
+# ===================================================================
+# 28. Variable ordering: dummy-coded vars expanded in-place
+# ===================================================================
+def test_variable_ordering_preservation():
+    from dce_analyzer.config import DummyCoding, FullModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    # Variables in order: price (continuous), comfort (dummy, binary 0/1), time (continuous), reliability (continuous)
+    # After expansion, order must be: price, comfort_L1, time, reliability
+    # (not: price, time, reliability, comfort_L1 — the old buggy behavior)
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        model_type="conditional",
+        dummy_codings=[
+            DummyCoding(column="comfort", ref_level=0),
+        ],
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    param_names = list(est.estimates["parameter"])
+    # Check order: price -> comfort dummy -> time -> reliability
+    expected_order = ["beta_price", "beta_comfort_L1", "beta_time", "beta_reliability"]
+    assert param_names == expected_order, (
+        f"Variable ordering not preserved. Expected {expected_order}, got {param_names}"
+    )
+    # Also verify expanded_spec preserves order
+    exp_spec = result.expanded_spec
+    exp_var_names = [v.name for v in exp_spec.variables]
+    assert exp_var_names == ["price", "comfort_L1", "time", "reliability"], (
+        f"Expanded spec variable order wrong: {exp_var_names}"
+    )
+
+
+_run("28. Variable ordering: dummy-coded vars expanded in-place", test_variable_ordering_preservation)
+
+
+# ===================================================================
+# 29. WTP theta_index mapping for MMNL (SE correctness)
+# ===================================================================
+def test_wtp_theta_index():
+    from dce_analyzer.config import ModelSpec, VariableSpec
+    from dce_analyzer.pipeline import estimate_dataframe
+    from dce_analyzer.wtp import compute_wtp
+
+    # price is random, then time (fixed), comfort (fixed), reliability (fixed)
+    # This creates interleaved mu/sd rows: mu_price, sd_price, beta_time, ...
+    # The theta_index mapping must be correct for WTP SEs.
+    spec = ModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price", distribution="normal"),
+            VariableSpec(name="time", column="time", distribution="fixed"),
+            VariableSpec(name="comfort", column="comfort", distribution="fixed"),
+            VariableSpec(name="reliability", column="reliability", distribution="fixed"),
+        ],
+        n_draws=50,
+    )
+    result = estimate_dataframe(
+        df=sim_output.data, spec=spec, model_type="mixed",
+        maxiter=200, seed=42,
+    )
+    est = result.estimation
+
+    # Verify theta_index column exists and is correct
+    assert "theta_index" in est.estimates.columns, "theta_index column missing"
+    # mu_price -> theta 0, sd_price -> theta 4, beta_time -> theta 1,
+    # beta_comfort -> theta 2, beta_reliability -> theta 3
+    tidx_map = dict(zip(est.estimates["parameter"], est.estimates["theta_index"]))
+    assert tidx_map["mu_price"] == 0, f"mu_price should be theta 0, got {tidx_map['mu_price']}"
+    assert tidx_map["beta_time"] == 1, f"beta_time should be theta 1, got {tidx_map['beta_time']}"
+    assert tidx_map["sd_price"] == 4, f"sd_price should be theta 4, got {tidx_map['sd_price']}"
+
+    # Compute WTP using time as the cost variable
+    wtp_df = compute_wtp(est, cost_variable="time")
+    assert not wtp_df.empty
+    # Check that SEs are not NaN (vcov should be available)
+    if est.vcov_matrix is not None:
+        for _, row in wtp_df.iterrows():
+            if row["attribute"] in ("price", "comfort", "reliability"):
+                assert not np.isnan(row["wtp_std_error"]), (
+                    f"WTP SE is NaN for {row['attribute']} — theta_index mapping may be wrong"
+                )
+
+
+_run("29. WTP theta_index mapping for MMNL (SE correctness)", test_wtp_theta_index)
+
+
+# ===================================================================
+# 30. 3-way interaction (price × time × income)
+# ===================================================================
+def test_3way_interaction():
+    from dce_analyzer.config import FullModelSpec, InteractionTerm, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        model_type="conditional",
+        interactions=[
+            InteractionTerm(columns=("price", "time", "income")),
+        ],
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    param_names = set(est.estimates["parameter"])
+    # 3-way interaction name: price_x_time_x_income
+    assert "beta_price_x_time_x_income" in param_names, (
+        f"Missing 3-way interaction param. Got: {param_names}"
+    )
+    # 4 base + 1 interaction = 5 params
+    assert est.n_parameters == 5, f"Expected 5 params, got {est.n_parameters}"
+
+
+_run("30. 3-way interaction (price × time × income)", test_3way_interaction)
+
+
+# ===================================================================
+# 31. Attribute × attribute interaction (price × time)
+# ===================================================================
+def test_attribute_x_attribute_interaction():
+    from dce_analyzer.config import FullModelSpec, InteractionTerm, VariableSpec
+    from dce_analyzer.pipeline import estimate_from_spec
+
+    spec = FullModelSpec(
+        id_col="respondent_id",
+        task_col="task_id",
+        alt_col="alternative",
+        choice_col="choice",
+        variables=[
+            VariableSpec(name="price", column="price"),
+            VariableSpec(name="time", column="time"),
+            VariableSpec(name="comfort", column="comfort"),
+            VariableSpec(name="reliability", column="reliability"),
+        ],
+        model_type="conditional",
+        interactions=[
+            InteractionTerm(columns=("price", "time")),
+        ],
+        maxiter=200,
+        seed=42,
+    )
+    result = estimate_from_spec(df=sim_output.data, spec=spec)
+    est = result.estimation
+    param_names = set(est.estimates["parameter"])
+    # attribute x attribute interaction
+    assert "beta_price_x_time" in param_names, (
+        f"Missing attribute x attribute interaction param. Got: {param_names}"
+    )
+    # 4 base + 1 interaction = 5 params
+    assert est.n_parameters == 5, f"Expected 5 params, got {est.n_parameters}"
+
+
+_run("31. Attribute × attribute interaction (price × time)", test_attribute_x_attribute_interaction)
+
+
+# ===================================================================
+# Summary
+# ===================================================================
+print()
+print("=" * 60)
+n_pass = sum(1 for _, ok, _ in _results if ok)
+n_fail = sum(1 for _, ok, _ in _results if not ok)
+print(f"  {n_pass} passed, {n_fail} failed out of {len(_results)} tests")
+print("=" * 60)
+
+if n_fail > 0:
+    print()
+    print("FAILURES:")
+    for name, ok, msg in _results:
+        if not ok:
+            print(f"  {name}: {msg}")
+    print()
+    sys.exit(1)
+else:
+    print("  ALL TESTS PASSED")
+    sys.exit(0)

src/dce_analyzer/config.py

@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Literal
+
+
+DistributionType = Literal["fixed", "normal", "lognormal"]
+
+
+@dataclass(frozen=True)
+class VariableSpec:
+    """One variable used in the utility function."""
+
+    name: str
+    column: str
+    distribution: DistributionType = "fixed"
+
+    def __post_init__(self) -> None:
+        if not self.name:
+            raise ValueError("VariableSpec.name cannot be empty.")
+        if not self.column:
+            raise ValueError("VariableSpec.column cannot be empty.")
+        if self.distribution not in {"fixed", "normal", "lognormal"}:
+            raise ValueError(
+                f"Unsupported distribution '{self.distribution}'. "
+                "Use one of: fixed, normal, lognormal."
+            )
+
+
+@dataclass(frozen=True)
+class ModelSpec:
+    """Data layout and variable config for a model run."""
+
+    id_col: str
+    task_col: str
+    alt_col: str
+    choice_col: str
+    variables: list[VariableSpec]
+    n_draws: int = 200
+    n_classes: int = 2
+    membership_cols: list[str] | None = None
+
+    def __post_init__(self) -> None:
+        core_cols = [self.id_col, self.task_col, self.alt_col, self.choice_col]
+        if any(not c for c in core_cols):
+            raise ValueError("id_col, task_col, alt_col, and choice_col must all be set.")
+        if len(self.variables) == 0:
+            raise ValueError("At least one variable is required in ModelSpec.variables.")
+        if self.n_draws < 1:
+            raise ValueError("n_draws must be >= 1.")
+        if self.n_classes < 1:
+            raise ValueError("n_classes must be >= 1.")
+
+
+@dataclass(frozen=True)
+class DummyCoding:
+    """Dummy-coding specification for a single attribute."""
+
+    column: str          # original column name in the data
+    ref_level: object    # reference level (omitted baseline)
+
+    def expand(self, df) -> tuple[list[str], dict]:
+        """Return (list of dummy column names, {dummy_name: level}) for this column.
+
+        Does NOT mutate *df*.
+        """
+        import pandas as pd
+
+        unique_vals = sorted(df[self.column].dropna().unique())
+        non_ref = [v for v in unique_vals if v != self.ref_level]
+        names: list[str] = []
+        mapping: dict[str, object] = {}
+        for level in non_ref:
+            name = f"{self.column}_L{level}"
+            names.append(name)
+            mapping[name] = level
+        return names, mapping
+
+
+@dataclass(frozen=True)
+class HeterogeneityInteraction:
+    """An attribute x demographic interaction term (legacy, kept for backward compat)."""
+
+    attribute: str       # name of the attribute variable
+    demographic_col: str  # name of the demographic column in the data
+
+
+@dataclass(frozen=True)
+class InteractionTerm:
+    """An arbitrary N-way interaction term: product of specified columns."""
+
+    columns: tuple[str, ...]
+
+    def __post_init__(self) -> None:
+        if len(self.columns) < 2:
+            raise ValueError("InteractionTerm requires at least 2 columns.")
+
+    @property
+    def name(self) -> str:
+        return "_x_".join(self.columns)
+
+
+@dataclass
+class FullModelSpec:
+    """Complete model specification -- one object captures everything."""
+
+    # Data layout
+    id_col: str
+    task_col: str
+    alt_col: str
+    choice_col: str
+
+    # Variable specifications
+    variables: list[VariableSpec]
+
+    # Model type
+    model_type: str = "mixed"  # "conditional", "mixed", "gmnl", "latent_class"
+
+    # Dummy coding: backend expands these columns into dummy variables
+    dummy_codings: list[DummyCoding] = field(default_factory=list)
+
+    # Interaction terms (N-way, any columns) -- works for ALL model types
+    interactions: list[InteractionTerm] = field(default_factory=list)
+
+    # Correlation structure
+    correlated: bool = False
+    correlation_groups: list[list[int]] | None = None
+
+    # BWS
+    bws_worst_col: str | None = None
+    estimate_lambda_w: bool = True
+
+    # Latent class
+    n_classes: int = 2
+    membership_cols: list[str] | None = None
+
+    # Estimation settings
+    n_draws: int = 200
+    maxiter: int = 300
+    seed: int = 123
+    n_starts: int = 10
+
+    def __post_init__(self) -> None:
+        valid_types = {"conditional", "mixed", "gmnl", "latent_class"}
+        if self.model_type not in valid_types:
+            raise ValueError(
+                f"model_type must be one of {valid_types}, got '{self.model_type}'."
+            )
+        core_cols = [self.id_col, self.task_col, self.alt_col, self.choice_col]
+        if any(not c for c in core_cols):
+            raise ValueError("id_col, task_col, alt_col, and choice_col must all be set.")
+        if len(self.variables) == 0:
+            raise ValueError("At least one variable is required.")
+        if self.n_draws < 1:
+            raise ValueError("n_draws must be >= 1.")
+
+    def to_model_spec(self) -> ModelSpec:
+        """Convert to the legacy ModelSpec for backward compatibility."""
+        return ModelSpec(
+            id_col=self.id_col,
+            task_col=self.task_col,
+            alt_col=self.alt_col,
+            choice_col=self.choice_col,
+            variables=list(self.variables),
+            n_draws=self.n_draws,
+            n_classes=self.n_classes,
+            membership_cols=self.membership_cols,
+        )

src/dce_analyzer/pipeline.py

@@ -0,0 +1,217 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import pandas as pd
+import torch
+
+from .config import FullModelSpec, ModelSpec, VariableSpec
+from .data import ChoiceTensors, prepare_choice_tensors
+from .latent_class import LatentClassEstimator, LatentClassResult
+from .model import ConditionalLogitEstimator, EstimationResult, GmnlEstimator, MixedLogitEstimator
+
+
+@dataclass
+class PipelineResult:
+    tensors: ChoiceTensors
+    estimation: EstimationResult | LatentClassResult
+    wtp: pd.DataFrame | None = field(default=None)
+    expanded_spec: ModelSpec | None = field(default=None, repr=False)
+    expanded_df: pd.DataFrame | None = field(default=None, repr=False)
+
+
+def estimate_dataframe(
+    df: pd.DataFrame,
+    spec: ModelSpec,
+    model_type: str = "mixed",
+    maxiter: int = 300,
+    seed: int = 123,
+    device: torch.device | None = None,
+    n_classes: int | None = None,
+    n_starts: int = 10,
+    correlated: bool = False,
+    membership_cols: list[str] | None = None,
+    correlation_groups: list[list[int]] | None = None,
+    bws_worst_col: str | None = None,
+    estimate_lambda_w: bool = True,
+) -> PipelineResult:
+    tensors = prepare_choice_tensors(df, spec, device=device)
+
+    # Prepare BWS data if worst column specified
+    bws_data = None
+    if bws_worst_col:
+        from .bws import prepare_bws_data, validate_bws
+
+        validate_bws(df, spec, bws_worst_col)
+        bws_data = prepare_bws_data(
+            df,
+            spec,
+            bws_worst_col,
+            tensors.n_obs,
+            tensors.n_alts,
+            tensors.X.device,
+            estimate_lambda_w=estimate_lambda_w,
+        )
+
+    if model_type == "mixed":
+        estimator = MixedLogitEstimator(
+            tensors=tensors,
+            variables=spec.variables,
+            n_draws=spec.n_draws,
+            device=tensors.X.device,
+            seed=seed,
+            correlated=correlated,
+            correlation_groups=correlation_groups,
+            bws_data=bws_data,
+        )
+        return PipelineResult(tensors=tensors, estimation=estimator.fit(maxiter=maxiter))
+    elif model_type == "conditional":
+        estimator = ConditionalLogitEstimator(
+            tensors=tensors,
+            variables=spec.variables,
+            device=tensors.X.device,
+            seed=seed,
+            bws_data=bws_data,
+        )
+        return PipelineResult(tensors=tensors, estimation=estimator.fit(maxiter=maxiter))
+    elif model_type == "gmnl":
+        estimator = GmnlEstimator(
+            tensors=tensors,
+            variables=spec.variables,
+            n_draws=spec.n_draws,
+            device=tensors.X.device,
+            seed=seed,
+            bws_data=bws_data,
+            correlated=correlated,
+            correlation_groups=correlation_groups,
+        )
+        return PipelineResult(tensors=tensors, estimation=estimator.fit(maxiter=maxiter))
+    elif model_type == "latent_class":
+        q = n_classes if n_classes is not None else spec.n_classes
+        mc = membership_cols or spec.membership_cols
+        lc_estimator = LatentClassEstimator(
+            tensors=tensors,
+            variables=spec.variables,
+            n_classes=q,
+            device=tensors.X.device,
+            seed=seed,
+            membership_cols=mc,
+            df=df,
+            id_col=spec.id_col,
+            bws_data=bws_data,
+        )
+        return PipelineResult(
+            tensors=tensors,
+            estimation=lc_estimator.fit(maxiter=maxiter, n_starts=n_starts),
+        )
+    else:
+        raise ValueError(
+            "model_type must be 'mixed', 'conditional', 'gmnl', or 'latent_class'."
+        )
+
+
+def estimate_from_spec(
+    df: pd.DataFrame,
+    spec: FullModelSpec,
+    device: torch.device | None = None,
+) -> PipelineResult:
+    """Single entry-point: all configuration comes from *spec*.
+
+    1. Dummy-coded columns are materialised from *spec.dummy_codings*.
+    2. Heterogeneity interactions are materialised as interaction columns.
+    Both are appended as fixed VariableSpecs before estimation.
+    """
+    df = df.copy()
+
+    # ── Expand dummy-coded variables ──────────────────────────────────
+    dummy_cols = {dc.column for dc in spec.dummy_codings}
+    # Build mapping: original column -> list of expanded VariableSpecs
+    _dummy_expansions: dict[str, list[VariableSpec]] = {}
+
+    for dc in spec.dummy_codings:
+        matched = [v for v in spec.variables if v.column == dc.column]
+        if not matched:
+            raise ValueError(
+                f"Dummy coding column '{dc.column}' not found in variables."
+            )
+        dummy_names, mapping = dc.expand(df)
+        for dname, level in mapping.items():
+            df[dname] = (df[dc.column] == level).astype(int)
+
+        base_var = matched[0]
+        _dummy_expansions[dc.column] = [
+            VariableSpec(name=dname, column=dname, distribution=base_var.distribution)
+            for dname in dummy_names
+        ]
+
+    # Build final variable list: replace each dummy placeholder in-place
+    # to preserve the UI's variable ordering (critical for correlation_groups)
+    all_variables: list[VariableSpec] = []
+    for v in spec.variables:
+        if v.column in dummy_cols:
+            all_variables.extend(_dummy_expansions[v.column])
+        else:
+            all_variables.append(v)
+
+    extra_vars: list[VariableSpec] = []
+
+    for inter in spec.interactions:
+        col_name = inter.name
+        for col in inter.columns:
+            if col not in df.columns:
+                raise ValueError(
+                    f"Interaction column '{col}' not found in data."
+                )
+        product = df[inter.columns[0]].astype(float)
+        for col in inter.columns[1:]:
+            product = product * df[col].astype(float)
+        df[col_name] = product
+        extra_vars.append(VariableSpec(name=col_name, column=col_name, distribution="fixed"))
+
+    all_variables = all_variables + extra_vars
+
+    model_spec = ModelSpec(
+        id_col=spec.id_col,
+        task_col=spec.task_col,
+        alt_col=spec.alt_col,
+        choice_col=spec.choice_col,
+        variables=all_variables,
+        n_draws=spec.n_draws,
+        n_classes=spec.n_classes,
+        membership_cols=spec.membership_cols,
+    )
+
+    result = estimate_dataframe(
+        df=df,
+        spec=model_spec,
+        model_type=spec.model_type,
+        maxiter=spec.maxiter,
+        seed=spec.seed,
+        device=device,
+        n_classes=spec.n_classes,
+        n_starts=spec.n_starts,
+        correlated=spec.correlated,
+        membership_cols=spec.membership_cols,
+        correlation_groups=spec.correlation_groups,
+        bws_worst_col=spec.bws_worst_col,
+        estimate_lambda_w=spec.estimate_lambda_w,
+    )
+    result.expanded_spec = model_spec
+    result.expanded_df = df
+    return result
+
+
+def save_estimation_outputs(estimation: EstimationResult | LatentClassResult, output_prefix: str | Path) -> None:
+    output_prefix = Path(output_prefix)
+    if output_prefix.suffix:
+        output_prefix = output_prefix.with_suffix("")
+    output_prefix.parent.mkdir(parents=True, exist_ok=True)
+
+    estimates_path = output_prefix.parent / f"{output_prefix.name}_estimates.csv"
+    summary_path = output_prefix.parent / f"{output_prefix.name}_summary.json"
+
+    estimation.estimates.to_csv(estimates_path, index=False)
+    with open(summary_path, "w", encoding="utf-8") as handle:
+        json.dump(estimation.summary_dict(), handle, indent=2, default=str)