Spaces:
Sleeping
Sleeping
| # Copyright (C) 2026 Hengzhe Zhao. All rights reserved. | |
| # Licensed under dual license: AGPL-3.0 (open-source) or commercial. See LICENSE. | |
| from __future__ import annotations | |
| from dataclasses import dataclass | |
| import numpy as np | |
| import pandas as pd | |
| class SimulationOutput: | |
| data: pd.DataFrame | |
| true_parameters: dict[str, float] | |
| def generate_simulated_dce( | |
| n_individuals: int = 300, | |
| n_tasks: int = 8, | |
| n_alts: int = 3, | |
| seed: int = 42, | |
| attribute_names: list[str] | None = None, | |
| covariate_names: list[str] | None = None, | |
| bws: bool = False, | |
| ) -> SimulationOutput: | |
| """ | |
| Generate a panel choice dataset in long format. | |
| Parameters | |
| ---------- | |
| attribute_names : list[str] or None | |
| Custom attribute column names. When None, uses the hardcoded defaults | |
| (price, time, comfort, reliability) with their original distributions. | |
| covariate_names : list[str] or None | |
| Custom covariate column names. When None, uses the hardcoded defaults | |
| (income, age) with their original distributions. | |
| bws : bool | |
| If True, generate a ``worst`` column (binary 0/1) alongside the best | |
| choice column. Requires ``n_alts >= 3``. | |
| """ | |
| if n_individuals < 1 or n_tasks < 1 or n_alts < 2: | |
| raise ValueError("Need n_individuals >= 1, n_tasks >= 1, n_alts >= 2.") | |
| if bws and n_alts < 3: | |
| raise ValueError("BWS simulation requires n_alts >= 3.") | |
| rng = np.random.default_rng(seed) | |
| alt_constants = np.linspace(0.0, 0.35, num=n_alts) | |
| # ── custom vs hardcoded path ────────────────────────────────── | |
| use_custom_attrs = attribute_names is not None | |
| use_custom_covs = covariate_names is not None | |
| if use_custom_attrs: | |
| # Generate true parameters for each custom attribute | |
| true_params: dict[str, float] = {} | |
| attr_true_mu: list[float] = [] | |
| attr_true_sd: list[float] = [] | |
| for attr in attribute_names: | |
| mu = float(rng.uniform(-1.5, 1.5)) | |
| sd = abs(mu) * 0.2 | |
| true_params[f"mu_{attr}"] = round(mu, 4) | |
| true_params[f"sd_{attr}"] = round(sd, 4) | |
| attr_true_mu.append(mu) | |
| attr_true_sd.append(sd) | |
| else: | |
| attribute_names = ["price", "time", "comfort", "reliability"] | |
| true_params = { | |
| "mu_price": -1.20, | |
| "sd_price": 0.25, | |
| "mu_time": -0.70, | |
| "sd_time": 0.18, | |
| "mu_comfort": 0.85, | |
| "sd_comfort": 0.12, | |
| "beta_reliability": 0.55, | |
| } | |
| if not use_custom_covs: | |
| covariate_names = ["income", "age"] | |
| for alt_idx in range(1, n_alts): | |
| true_params[f"asc_alt_{alt_idx + 1}"] = float(alt_constants[alt_idx]) | |
| rows: list[dict[str, float | int]] = [] | |
| # ── covariate arrays (constant within respondent) ───────────── | |
| if use_custom_covs: | |
| cov_arrays = { | |
| name: rng.normal(50, 15, size=n_individuals).clip(10, 100) | |
| for name in covariate_names | |
| } | |
| else: | |
| cov_arrays = { | |
| "income": rng.normal(60_000, 15_000, size=n_individuals).clip(18_000, 180_000), | |
| "age": rng.normal(42, 12, size=n_individuals).clip(18, 80), | |
| } | |
| for individual in range(1, n_individuals + 1): | |
| # individual-level random coefficients | |
| if use_custom_attrs: | |
| betas = [rng.normal(attr_true_mu[k], max(attr_true_sd[k], 1e-6)) | |
| for k in range(len(attribute_names))] | |
| else: | |
| betas_named = { | |
| "price": rng.normal(true_params["mu_price"], true_params["sd_price"]), | |
| "time": rng.normal(true_params["mu_time"], true_params["sd_time"]), | |
| "comfort": rng.normal(true_params["mu_comfort"], true_params["sd_comfort"]), | |
| } | |
| cov_values = {name: float(arr[individual - 1]) for name, arr in cov_arrays.items()} | |
| for task in range(1, n_tasks + 1): | |
| # ── attribute values ────────────────────────────────── | |
| if use_custom_attrs: | |
| attr_values = { | |
| attr: rng.uniform(1, 10, size=n_alts) for attr in attribute_names | |
| } | |
| deterministic_utility = alt_constants.copy() | |
| for k, attr in enumerate(attribute_names): | |
| deterministic_utility = deterministic_utility + betas[k] * attr_values[attr] | |
| else: | |
| price = rng.uniform(3.0, 30.0, size=n_alts) | |
| time = rng.uniform(10.0, 90.0, size=n_alts) | |
| comfort = rng.integers(0, 2, size=n_alts) | |
| reliability = rng.uniform(0.70, 1.00, size=n_alts) | |
| deterministic_utility = ( | |
| alt_constants | |
| + betas_named["price"] * np.log1p(price) | |
| + betas_named["time"] * (time / 10.0) | |
| + betas_named["comfort"] * comfort | |
| + true_params["beta_reliability"] * reliability | |
| ) | |
| attr_values = { | |
| "price": price, | |
| "time": time, | |
| "comfort": comfort, | |
| "reliability": reliability, | |
| } | |
| random_shock = rng.gumbel(0.0, 1.0, size=n_alts) | |
| utilities = deterministic_utility + random_shock | |
| chosen_alt = int(np.argmax(utilities)) | |
| # ── worst choice (BWS) ─────────────────────────────── | |
| if bws: | |
| remaining_utils = utilities.copy() | |
| remaining_utils[chosen_alt] = np.inf # exclude best | |
| worst_alt = int(np.argmin(remaining_utils)) | |
| for alt in range(n_alts): | |
| row: dict[str, float | int] = { | |
| "respondent_id": individual, | |
| "task_id": task, | |
| "alternative": alt + 1, | |
| "choice": int(alt == chosen_alt), | |
| } | |
| for attr in attribute_names: | |
| val = attr_values[attr][alt] | |
| row[attr] = int(val) if isinstance(val, (np.integer,)) else float(val) | |
| for cov_name, cov_val in cov_values.items(): | |
| row[cov_name] = cov_val | |
| if bws: | |
| row["worst"] = int(alt == worst_alt) | |
| rows.append(row) | |
| return SimulationOutput(data=pd.DataFrame(rows), true_parameters=true_params) | |