RobustiPy

https://robustipy.readthedocs.io/en/latest/
Admin
RobustiPy is a Python library for robust statistical inference and specification curve analysis...
Tokens:86,668
Snippets:341
Trust Score:9.6
Update:1 month ago
Show doc for...
Tokens:
Raw
### robustipy.utils.prepare_asc

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Load and preprocess the ASC example dataset for illustration.

```APIDOC
## robustipy.utils.prepare_asc

### Description
Load and preprocess the ASC example dataset for illustration.

### Parameters
- **asc_path** (str) - Required - Path to the Stata (.dta) file containing ASC data.

### Returns
- tuple - (y (str): Dependent variable name, x (List[str]): Continuous predictor names, c (List[str]): Control variable names, group (str): Grouping variable name ('pidp'), ASC_df (pd.DataFrame): Cleaned DataFrame).
```

--------------------------------

### LRobust fit method: rescaling, validation, and control-set loop

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

This is the main body of the LRobust fit method, showing the sequence of setup steps: rescaling, parameter preparation, validation, and the core loop over control specifications. It includes the warning about unsupported demeaning, the validation call with allowed metrics, and the initialization of arrays for model statistics. The final block also shows the conditional import of SHAP and the data preparation for SHAP values, which is only executed when compute_shap is True.

```python
# Rescale independent variables (x)
        if rescale_x is True:
            for column in self.x:
                self.data[column] = rescale(self.data[column])

        # Rescale control variables (z)
        if rescale_z is True:
            for column in controls:
                self.data[column] = rescale(self.data[column])
```

```python
# Prepare modeling setup (includes seed, CPU count, metric, etc.)
        draws, kfold, oos_metric, n_cpu, seed = make_inquiry(
            self.model_name,
            self.y,
            self.data,
            draws,
            kfold,
            oos_metric,
            n_cpu,
            seed
        )
```

```python
self.seed = seed
        self.data = self.data.copy()
        if oos_metric == 'mcfadden-r2':
            oos_metric = 'mcfaddens-r2'
```

```python
# Logistic fixed-effects demeaning is not implemented. If `group` is
        # provided, we still use it for grouped CV/bootstrap resampling, but do
        # not demean y/x/z.
        if group is not None:
            warnings.warn(
                "[LRobust] Logistic fixed-effects demeaning is unsupported; "
                "`group` will be used for grouped CV/bootstrap resampling only (no demeaning).",
                UserWarning
            )
```

```python
# Ensure a single constant term in the design matrix.
        self.x, controls = _ensure_single_constant(self.data, self.y, self.x, controls)
```

```python
# Validate modeling parameters
        self._validate_fit_args(
            controls=controls,
            group=group,
            draws=draws,
            kfold=kfold,
            oos_metric=oos_metric,
            n_cpu=n_cpu,
            seed=seed,
            valid_oos_metrics=['pseudo-r2', 'mcfaddens-r2', 'rmse', 'cross-entropy', 'imv'],
            threshold=threshold
        )
```

```python
# Set seed for reproducibility
        np.random.seed(self.seed)
```

```python
# Store the selected out-of-sample evaluation metric
        self.oos_metric_name = oos_metric
```

```python
# Log model configuration
        print(
            f"LRobust is running with n_cpu={n_cpu}, draws={draws}, "
            f"folds={kfold}, seed={seed}.\n"
            f"We're evaluating our out-of-sample predictions with the "
            f"{oos_metric} metric.\n"
            f"The target variable of interest is {self.x[0]}. "
            f"Let's begin the calculations..."
        )
```

```python
# Default to using the full dataset size if sample_size is not provided
        if sample_size is None:
            sample_size = self.data.shape[0]
```

```python
# Currently only supports one dependent variable for logistic regression
        if len(self.y) > 1:
            raise NotImplementedError("Not implemented yet for logistic regression")  # TODO: Add multi-y logistic support
```

```python
else:
            z_specs = [tuple(spec) for spec in all_subsets(controls)]
            space_n = len(z_specs)
            control_positions = {name: idx for idx, name in enumerate(controls)}
            predictor_prefix = self.x + ([group] if group else [])
            base_predictors_df = self.data[predictor_prefix + controls]
            y_single_df = self.data[self.y]
            spec_predictor_cache: dict[int, pd.DataFrame] = {}

            def _get_lr_predictors_for_spec(spec_tuple: tuple[str, ...]) -> pd.DataFrame:
                mask = _spec_bitmask(spec_tuple, control_positions)
                cached = spec_predictor_cache.get(mask)
                if cached is None:
                    cols = predictor_prefix + list(spec_tuple)
                    cached = base_predictors_df.loc[:, cols]
                    spec_predictor_cache[mask] = cached
                return cached

            notebook_draws_bar = _make_notebook_draws_bar(
                total=space_n * draws,
                description="Bootstrap draws"
            )
            specs = []
            all_predictors = []
            b_all_list = []
            p_all_list = []

            # Initialize arrays to store model statistics across all control sets
            b_array = np.empty([space_n, draws])
            p_array = np.empty([space_n, draws])
            b_array_ystar = np.empty([space_n, draws])
            p_array_ystar = np.empty([space_n, draws])
            r2_array = np.empty([space_n, draws])
            r2i_array = np.empty([space_n])
            ll_array = np.empty([space_n])
            aic_array = np.empty([space_n])
            bic_array = np.empty([space_n])
            hqic_array = np.empty([space_n])
            av_k_metric_array = np.empty([space_n])

            shap_return = None
            if compute_shap:
                import shap

                # Preprocess data for SHAP values. Do not demean for logistic models.
                if group:
                    SHAP_comb = self.data[self.y + self.x + [group] + controls]
                else:
                    SHAP_comb = self.data[self.y + self.x + controls]
```

--------------------------------

### robustipy.utils.prepare_union

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Load and preprocess the classic union dataset for example analyses.

```APIDOC
## robustipy.utils.prepare_union

### Description
Load and preprocess the classic union dataset for example analyses.

### Parameters
- **path_to_union** (str) - Required - Path to the Stata (.dta) file containing union data.

### Returns
- tuple - (y (str): Dependent variable name ('log_wage'), c (List[str]): Control variable names, x (str): Treatment variable name ('union'), final_data (pd.DataFrame): Cleaned DataFrame ready for modeling).

### Raises
- **FileNotFoundError** - If the specified file does not exist.
```

--------------------------------

### robustipy.utils module imports and fallback setup

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/utils.html

Imports required libraries and sets up optional InquirerPy dependency with a fallback ValidationError class and _HAS_INQUIRERPY flag.

```python
# Module containing utility functions for the library
from typing import Optional, List, Tuple, Iterable, Sequence, Union
import copy
import numpy as np
import random
import warnings
import scipy
import pandas as pd
from itertools import chain, combinations
import statsmodels.api as sm
from scipy.optimize import minimize
from multiprocessing import cpu_count
import os
import sys

try:
    from InquirerPy import inquirer
    from InquirerPy.validator import ValidationError
    from InquirerPy.themes import GreenPassion
    _HAS_INQUIRERPY = True
except ImportError:
    inquirer = None
    GreenPassion = None



[docs]
    class ValidationError(Exception):
        """Fallback so IntegerRangeValidator can raise a typed error safely."""

        def __init__(self, *args, reason: str = ""):
            super().__init__(reason)
            self.reason = reason



    _HAS_INQUIRERPY = False
```

--------------------------------

### LRobust.get_results

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Gets the results of the logistic regression. Returns an object containing the regression results.

```APIDOC
## LRobust.get_results

### Description
Gets the results of the logistic regression. Returns an object containing the regression results.

### Method
get_results

### Endpoint
LRobust.get_results()

### Parameters
No parameters documented.

### Request Example
```json
{}
```

### Response
#### Success Response (200)
- **results** (OLSResult) - Object containing the regression results.

#### Response Example
```json
{
  "results": "OLSResult object"
}
```

### Error Handling
- No explicit error handling documented.
```

--------------------------------

### robustipy package initialization with compact warning formatting

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy.html

Initialization module for the robustipy package. It avoids global warning-hook side effects at import time and provides two public functions: enable_compact_warnings() and disable_compact_warnings(). The internal _robustipy_showwarning function formats warnings from robustipy modules without file paths, falling back to the original handler for non-robustipy warnings or on errors.

```python
"""robustipy package initialization.

This module intentionally avoids global warning-hook side effects at import
time. If compact robustipy-only warning formatting is desired, call
`enable_compact_warnings()` explicitly.
"""

from __future__ import annotations

import sys
import warnings

__all__ = ["enable_compact_warnings", "disable_compact_warnings"]

_ORIGINAL_SHOWWARNING = warnings.showwarning


def _robustipy_showwarning(message, category, filename, lineno, file=None, line=None):
    """Format robustipy warnings without file paths."""
    path = str(filename).replace("\\", "/")
    if "robustipy" in path:
        if file is None:
            file = sys.stderr
        try:
            file.write(f"{category.__name__}: {message}\n")
            return
        except Exception:
            # Fall back to the original handler if needed.
            pass
    _ORIGINAL_SHOWWARNING(message, category, filename, lineno, file=file, line=line)




[docs]
def enable_compact_warnings() -> None:
    """Enable compact formatting for warnings emitted from robustipy modules."""
    if warnings.showwarning is not _robustipy_showwarning:
        warnings.showwarning = _robustipy_showwarning






[docs]
def disable_compact_warnings() -> None:
    """Restore the original warning formatting handler."""
    if warnings.showwarning is _robustipy_showwarning:
        warnings.showwarning = _ORIGINAL_SHOWWARNING
```

--------------------------------

### Prepare output directory and SHAP data for univariate 8-panel grid

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/figures.html

Prepare the output directory and set up an 8-panel grid for univariate y_name. Validates SHAP values are 2D, drops a bias column if present, and filters SHAP values and features to control variables only.

```python
    # Prepare output directory for saving figures
    outdir = _prepare_output_dir(Path(figpath) if figpath else None, project_name)

    # If y_name is univariate, plot full 8-panel grid
    if max(len(t) for t in results_object.y_name) == 1:
        fig = plt.figure(figsize=figsize)
        gs = GridSpec(9, 24, wspace=0.5, hspace=1.5)

        # Set up axes for subplots
        ax1 = fig.add_subplot(gs[0:3, 0:12])
        ax2 = fig.add_subplot(gs[0:3, 13:24])
        ax4 = fig.add_subplot(gs[3:5, 6:12])
        ax3 = fig.add_subplot(gs[3:5, 0:6])
        ax4_cbar = fig.add_subplot(gs[3:5, 12:13])
        ax5 = fig.add_subplot(gs[3:5, 14:23])
        spec_gs = GridSpecFromSubplotSpec(
            2,
            2,
            subplot_spec=gs[5:9, 0:13],
            height_ratios=[2, 1],
            width_ratios=[12, 1],
            hspace=0.12,
            wspace=0.05
        )
        ax6 = fig.add_subplot(spec_gs[0, 0])
        ax6m = fig.add_subplot(spec_gs[1, 0], sharex=ax6)
        ax6m_cbar = fig.add_subplot(spec_gs[1, 1])
        ax6_cbar_spacer = fig.add_subplot(spec_gs[0, 1])
        ax6_cbar_spacer.axis('off')
        ax7 = fig.add_subplot(gs[5:7, 12:21])
        ax8 = fig.add_subplot(gs[7:9, 14:23])

        # Prepare SHAP values and input matrix (control/z variables only).
        # This keeps panel c/d/spec-matrix focused on specification-varying terms.
        shap_vals_full = np.asarray(results_object.shap_return[0])
        shap_x_full = results_object.shap_return[1]
        if not isinstance(shap_x_full, pd.DataFrame):
            shap_x_full = pd.DataFrame(shap_x_full)
        if shap_vals_full.ndim != 2:
            raise ValueError(
                f"Expected 2D SHAP values, received shape {shap_vals_full.shape!r}."
            )
        # Some SHAP pipelines prepend a bias column; drop it if present.
        if shap_vals_full.shape[1] == shap_x_full.shape[1] + 1:
            shap_vals_full = shap_vals_full[:, 1:]
        if shap_vals_full.shape[1] != shap_x_full.shape[1]:
            raise ValueError(
                f"SHAP values columns ({shap_vals_full.shape[1]}) do not match "
                f"feature matrix columns ({shap_x_full.shape[1]})."
            )

        shap_cols = [c for c in results_object.controls if c in shap_x_full.columns]
        shap_idx = [shap_x_full.columns.get_loc(c) for c in shap_cols]
        shap_vals = shap_vals_full[:, shap_idx]
        shap_x = shap_x_full[shap_cols].to_numpy()
```

--------------------------------

### Run parallel bootstrap and assemble OLSResult

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Runs parallel bootstrap seeds, stores per-spec results, and assembles the final OLSResult object. The dispatch mode is 'streaming' when group is set, otherwise 'chunked'. The progress bar is closed only if notebook_draws_bar is not None.

```python
bootstrap_out = _run_parallel_seed_batches(
    seeds=shared_seeds,
    n_cpu=n_cpu,
    run_one_seed=_run_one_seed,
    draws_bar=notebook_draws_bar,
    dispatch_mode="streaming" if group else "chunked",
)
b_list, p_list, r2_list, b_list_ystar, p_list_ystar = zip(*bootstrap_out)

# Store results for current spec
specs.append(frozenset(spec))
all_predictors.append(self.x + list(spec))
b_array[index, :] = b_list
p_array[index, :] = p_list
b_array_ystar[index, :] = b_list_ystar
p_array_ystar[index, :] = p_list_ystar
r2_array[index, :] = r2_list
r2i_array[index] = r2_i
ll_array[index] = ll_i
ll_raw_array[index] = ll_raw_i
ll_null_array[index] = ll_null_i
ll_gain_array[index] = ll_gain_i
ll_gain_per_obs_array[index] = ll_gain_per_obs_i
nobs_array[index] = nobs_i
aic_array[index] = aic_i
bic_array[index] = bic_i
hqic_array[index] = hqic_i
av_k_metric_array[index] = av_k_metric_i
b_all_list.append(b_all)
p_all_list.append(p_all)

# Assemble final results object
results = OLSResult(y=self.y[0],
                    x=self.x[0],
                    data=self.data,
                    specs=specs,
                    all_predictors=all_predictors,
                    controls=controls,
                    draws=draws,
                    kfold=kfold,
                    all_b=b_all_list,
                    all_p=p_all_list,
                    estimates=b_array,
                    p_values=p_array,
                    estimates_ystar=b_array_ystar,
                    p_values_ystar=p_array_ystar,
                    r2_values=r2_array,
                    r2i_array=r2i_array,
                    ll_array=ll_array,
                    ll_raw_array=ll_raw_array,
                    ll_null_array=ll_null_array,
                    ll_gain_array=ll_gain_array,
                    ll_gain_per_obs_array=ll_gain_per_obs_array,
                    nobs_array=nobs_array,
                    aic_array=aic_array,
                    bic_array=bic_array,
                    hqic_array=hqic_array,
                    av_k_metric_array=av_k_metric_array,
                    model_name=self.model_name,
                    name_av_k_metric=self.oos_metric_name,
                    shap_return=shap_return)
self.results = results
if notebook_draws_bar is not None:
    notebook_draws_bar.close()
```

--------------------------------

### Helper function _run_parallel_seed_batches for parallel bootstrap

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Executes bootstrap seeds while preserving seed order and allowing notebook progress to advance. Supports streaming, batched, or chunked dispatch modes. Uses joblib with loky backend and disables memmapping.

```python
def _run_parallel_seed_batches(
    *,
    seeds,
    n_cpu: int,
    run_one_seed,
    draws_bar=None,
    dispatch_mode: str = "streaming",
    task_batch_size: Optional[int] = None,
):
    """
    Execute bootstrap seeds while preserving seed order and allowing notebook
    progress to advance during a specification.
    """
    outputs = []
    if len(seeds) == 0:
        return outputs

    cpu = max(1, int(n_cpu))
    progress_batch_size = max(8, cpu)
    if dispatch_mode not in {"streaming", "batched", "chunked"}:
        raise ValueError("dispatch_mode must be 'streaming', 'batched', or 'chunked'.")
    parallel_kwargs = {
        "n_jobs": cpu,
        # Keep process parallelism for CPU-bound bootstrap fits, but disable
        # joblib's automatic memmapping so large profiler runs do not accumulate
        # thousands of temporary memmap folders.
        "backend": "loky",
        "max_nbytes": None,
        "mmap_mode": None,
    }

    def _update_draws_bar(n: int) -> None:
        if draws_bar is None or n <= 0:
            return
        draws_bar.update(n)

    if cpu == 1:
        pending_progress = 0
        for seed in seeds:
            outputs.append(run_one_seed(int(seed)))
            pending_progress += 1
            if pending_progress >= progress_batch_size:
                _update_draws_bar(pending_progress)
                pending_progress = 0
        _update_draws_bar(pending_progress)
        return outputs
```

--------------------------------

### Helper function _make_notebook_draws_bar for tqdm progress bar

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Builds a tqdm progress bar for notebooks. Returns None outside notebooks or when tqdm isn't available. Configures bar format with percentage, bar, and timing.

```python
def _make_notebook_draws_bar(*, total: int, description: str):
    """
    Build a tqdm progress bar for notebooks. Returns None outside notebooks or
    when tqdm isn't available.
    """
    if not _in_jupyter():
        return None
    try:
        from tqdm.auto import tqdm
    except ImportError:
        return None
    return tqdm(
        total=total,
        desc=description,
        leave=True,
        miniters=1,
        mininterval=1.0,
        smoothing=0,
        unit="draw",
        unit_scale=False,
        bar_format=(
            "{desc}: {percentage:5.1f}%|{bar}| "
            "{n_fmt}/{total_fmt} [{elapsed}<{remaining}, {rate_fmt}]"
        ),
    )
```

--------------------------------

### Train logistic regression for SHAP explainability

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Prepare SHAP data, split into training and testing sets, and train a logistic regression model with L2 penalty (C=0.1) for SHAP explainability. Uses a linear explainer to compute SHAP values on the test set.

```python
SHAP_comb = SHAP_comb.dropna().reset_index(drop=True).copy()

# Split data for training and testing
x_train, x_test, y_train, _ = train_test_split(
    SHAP_comb[self.x + controls].drop(columns=['const'], errors='ignore'),
    SHAP_comb[self.y],
    test_size=0.2,
    random_state=self.seed
)

# Train logistic regression for SHAP explainability
model = sklearn.linear_model.LogisticRegression(penalty="l2", C=0.1)
model.fit(x_train, y_train.squeeze())
explainer = shap.LinearExplainer(model, x_train)
shap_return = [explainer.shap_values(x_test), x_test]
```

--------------------------------

### Loop through control variable subsets and run bootstrap

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Iterates over all control variable subsets, builds the design matrix, checks colinearity, estimates full-sample metrics, and runs parallel bootstrap sampling. Saves coefficients, p-values, R-squared, and information criteria (AIC, BIC, HQIC) for each specification.

```python
# Loop through all subsets of control variables
for index, spec in _progress_iter(
    enumerate(z_specs),
    total=space_n,
    description="Control specs",
):
    spec_predictors = _get_lr_predictors_for_spec(spec)
    comb = pd.concat([y_single_df, spec_predictors], axis=1)

    comb = comb.dropna().reset_index(drop=True).copy()

    if group:
        X_design = comb.drop(columns=[comb.columns[0], group], errors='ignore')
    else:
        X_design = comb.drop(columns=comb.columns[0])  # Drop y column
    self._check_colinearity(X_design)

    # Estimate full-sample model metrics
    (b_all, p_all, r2_i, ll_i,
    aic_i, bic_i, hqic_i,
    av_k_metric_i) = self._full_sample(
        comb,
        kfold=kfold,
        group=group,
        oos_metric_name=self.oos_metric_name
    )

    # Bootstrap sampling (parallelized)
    seeds = np.random.randint(0, 2**32 - 1, size=draws, dtype=np.int64)
    group_bootstrap_lookup = (
        _make_group_bootstrap_lookup(comb, group)
        if group else None
    )
    if group:
        n_predictors = comb.drop(
            columns=[comb.columns[0], group],
            errors="ignore"
        ).shape[1]
        min_rows_after_filter = max(5, n_predictors + 2)
    else:
        min_rows_after_filter = None
    def _run_one_seed(seed_i: int):
        return self._strap_regression(
            comb,
            group,
            sample_size,
            seed_i,
            group_bootstrap_lookup=group_bootstrap_lookup,
            min_rows_after_filter=min_rows_after_filter,
        )

    bootstrap_out = _run_parallel_seed_batches(
        seeds=seeds,
        n_cpu=n_cpu,
        run_one_seed=_run_one_seed,
        draws_bar=notebook_draws_bar
    )
    b_list, p_list, r2_list = zip(*bootstrap_out)

    # Save results
    specs.append(frozenset(spec))
    all_predictors.append(self.x + list(spec))
    b_array[index, :] = b_list
    p_array[index, :] = p_list
    b_array_ystar[index, :] = np.nan * len(b_list)
    p_array_ystar[index, :] = np.nan * len(p_list)
    r2_array[index, :] = r2_list
    r2i_array[index] = r2_i
    ll_array[index] = ll_i
    aic_array[index] = aic_i
    bic_array[index] = bic_i
    hqic_array[index] = hqic_i
    av_k_metric_array[index] = av_k_metric_i
    b_all_list.append(b_all)
    p_all_list.append(p_all)
```

--------------------------------

### robustipy.prototypes module

Source: https://robustipy.readthedocs.io/en/latest/modules.html

Base classes and prototypes for robust models and results, including BaseRobust, Protomodel, and Protoresult.

```APIDOC
## robustipy.prototypes module

### BaseRobust
- `y`
- `x`
- `data`
- `model_name`
- `results`
- `parameters`
- `fit()`
- `get_results()`
- `multiple_y()`

### MissingValueWarning

### Protomodel
- `fit()`

### Protoresult
- `plot()`
- `summary()`
```

--------------------------------

### load(_filename: str_) → OLSResult

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Loads an OLSResult object from a pickle file.

```APIDOC
## load(_filename: str_) → OLSResult

### Description
Loads an OLSResult object from a pickle file.

### Method
Class method

### Endpoint
load(filename)

### Parameters
#### Path Parameters
- **filename** (str) - Required - Path to the pickle file.

### Returns
- **OLSResult** - The loaded OLSResult object.

### Return Type
OLSResult

```

--------------------------------

### Prompt for oos_metric, n_cpu, and seed with InquirerPy fallback

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/utils.html

Prompts for missing out-of-sample metric, CPU count, and seed using InquirerPy when available, otherwise falls back to input() in Jupyter/TTY or uses defaults ('pseudo-r2', default_cpus, 192735). The seed prompt validates integer input within [0, 2**31-1] and rejects non-integers or out-of-range values.

```python
    # ───────────────────────────────────────────────────────────────
    # 7) If oos_metric is missing, prompt the user to choose one:
    # ───────────────────────────────────────────────────────────────
    if oos_metric is None:
        if use_inquirer:
            question_oos = [
                inquirer.List(
                    'oos_inq',
                    message="Select the out-of-sample evaluation metric",
                    choices=oos_metric_choices,
                    carousel=True
                )
            ]
            oos_metric = inquirer.prompt(question_oos, theme=GreenPassion())['oos_inq']
        elif in_jupyter or in_real_tty:
            oos_metric = _ask_choice("Select the out-of-sample evaluation metric:", oos_metric_choices)
        else:
            oos_metric = 'pseudo-r2'

    # ───────────────────────────────────────────────────────────────
    # 8) If n_cpu is missing, ask about CPU count:
    # ───────────────────────────────────────────────────────────────
    if n_cpu is None:
        default_cpus = max(1, cpu_count() - 1)
        if use_inquirer:
            question_ncpu = [
                inquirer.List(
                    'ncpu_inq',
                    message=f"You haven’t specified the number of CPUs. Is {default_cpus} ok?",
                    choices=['Yes', 'No'],
                    carousel=True
                )
            ]
            n_cpu_answer = inquirer.prompt(question_ncpu, theme=GreenPassion())['ncpu_inq']
            if n_cpu_answer == 'Yes':
                n_cpu = default_cpus
            else:
                question_ncpu2 = [
                    inquirer.Text(
                        'ncpu_inq',
                        message="Enter number of CPUs to use",
                        validate=IntegerRangeValidator(1, cpu_count())
                    )
                ]
                n_cpu = int(inquirer.prompt(question_ncpu2, theme=GreenPassion())['ncpu_inq'])
        elif in_jupyter or in_real_tty:
            # Jupyter/TTY fallback without InquirerPy: ask via input().
            ans = input(f"You haven’t specified the number of CPUs. Is {default_cpus} okay? (yes/no): ").strip().lower()
            if ans in ('y', 'yes'):
                n_cpu = default_cpus
            else:
                n_cpu = _ask_integer("Enter number of CPUs to use", 1, cpu_count())
        else:
            n_cpu = default_cpus

        # ───────────────────────────────────────────────────────────────
        #  9) If seed is missing, prompt the user (TTY, Jupyter, or fallback)
        # ───────────────────────────────────────────────────────────────
    SEED_MIN = 0
    SEED_MAX = 2 ** 31 - 1
    in_jupyter = _running_in_jupyter()
    in_real_tty = _is_real_tty()
    def _ask_seed_input(prompt_text: str, min_val: int, max_val: int) -> int:
        """
        Repeatedly call input() until user enters a valid integer in [min_val, max_val].
        """
        while True:
            raw = input(f"{prompt_text} [{min_val}–{max_val}]: ").strip()
            try:
                val = int(raw)
            except ValueError:
                print(f"Invalid: '{raw}' is not an integer.")
                continue
            if val < min_val or val > max_val:
                print(f"Invalid: must be between {min_val} and {max_val}.")
                continue
            return val
    if seed is None:
        if use_inquirer:
            question_seed = [
                inquirer.Text(
                    'seed_inq',
                    message="Enter integer seed for reproducibility",
                    validate=IntegerRangeValidator(SEED_MIN, SEED_MAX)
                )
            ]
            seed = int(inquirer.prompt(question_seed, theme=GreenPassion())['seed_inq'])
        elif in_jupyter or in_real_tty:
            seed = _ask_seed_input("Enter integer seed for reproducibility", SEED_MIN, SEED_MAX)
        else:
            seed = 192735
    return draws, kfolds, oos_metric, n_cpu, seed
```

--------------------------------

### enable_compact_warnings()

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Enables compact formatting for warnings emitted from robustipy modules. This function must be called explicitly to activate the compact warning style.

```APIDOC
## enable_compact_warnings()

### Description
Enables compact formatting for warnings emitted from robustipy modules. This function must be called explicitly to activate the compact warning style.

### Method
Function call

### Endpoint
`robustipy.enable_compact_warnings()`

### Parameters
None

### Returns
None

### Example
```python
import robustipy
robustipy.enable_compact_warnings()
```

```

--------------------------------

### Store modeling results in OLSResult

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Constructs an OLSResult object with all modeling outputs and closes the notebook progress bar if present. Use after all bootstrap draws and metrics are computed.

```python
self.results = OLSResult(
    y=self.y[0],
    x=self.x[0],
    data=self.data,
    specs=specs,
    all_predictors=all_predictors,
    controls=controls,
    draws=draws,
    kfold=kfold,
    all_b=b_all_list,
    all_p=p_all_list,
    estimates=b_array,
    p_values=p_array,
    estimates_ystar=b_array_ystar,
    p_values_ystar=p_array_ystar,
    r2_values=r2_array,
    r2i_array=r2i_array,
    ll_array=ll_array,
    aic_array=aic_array,
    bic_array=bic_array,
    hqic_array=hqic_array,
    av_k_metric_array=av_k_metric_array,
    model_name=self.model_name,
    name_av_k_metric=self.oos_metric_name,
    shap_return=shap_return
)
if notebook_draws_bar is not None:
    notebook_draws_bar.close()
```

--------------------------------

### Initialize containers and compute SHAP values

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/models.html

Initializes containers and preallocates arrays for storing results across z-specs. If compute_shap is True, it imports shap, prepares the SHAP dataset (optionally group-demeaned), splits it into train/test, and fits a linear regression model for SHAP values. This block is used in the else branch when not assembling the final result.

```python
# Initialize containers for storing results across z-specs
specs = []
all_predictors = []
b_all_list = []
p_all_list = []

# Preallocate arrays for bootstrap and model stats
b_array = np.empty([space_n, draws])
p_array = np.empty([space_n, draws])
b_array_ystar = np.empty([space_n, draws])
p_array_ystar = np.empty([space_n, draws])
r2_array = np.empty([space_n, draws])
r2i_array = np.empty([space_n])
ll_array = np.empty([space_n])
ll_raw_array = np.empty([space_n])
ll_null_array = np.empty([space_n])
ll_gain_array = np.empty([space_n])
ll_gain_per_obs_array = np.empty([space_n])
nobs_array = np.empty([space_n], dtype=int)
aic_array = np.empty([space_n])
bic_array = np.empty([space_n])
hqic_array = np.empty([space_n])
av_k_metric_array = np.empty([space_n])

shap_return = None
if compute_shap:
    import shap

    # Prepare SHAP dataset (includes y, x, z, and optional group)
    if group:
        SHAP_comb = self.data[self.y + self.x + [group] + controls]
        SHAP_comb = group_demean(SHAP_comb, group=group)
    else:
        SHAP_comb = self.data[self.y + self.x + controls]
    SHAP_comb = SHAP_comb.dropna()
    SHAP_comb = SHAP_comb.reset_index(drop=True).copy()

    # Fit linear model for SHAP values
    x_train, x_test, y_train, _ = train_test_split(
        SHAP_comb[self.x + controls].drop(columns=['const'], errors='ignore'),
        SHAP_comb[self.y],
        test_size=0.2,
        random_state=self.seed
    )
    model = sklearn.linear_model.LinearRegression()
    model.fit(x_train, y_train)
```

--------------------------------

### get_colors(specs: List[List[str]], color_set_name: Optional[str] = 'Set1') -> List[Tuple[float, float, float, float]]

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/utils.html

Generates a palette of colors for a list of specifications using a categorical colormap.

```APIDOC
## get_colors(specs: List[List[str]], color_set_name: Optional[str] = 'Set1') -> List[Tuple[float, float, float, float]]

### Description
Generate a palette of colors for a list of specifications using a categorical colormap.

### Parameters
- **specs** (list of list of str) - Required - Each inner list represents one specification (set of variable names).
- **color_set_name** (str, optional) - Name of a Matplotlib qualitative colormap (default 'Set1').

### Returns
- **List[Tuple[float, float, float, float]]** - A list of RGBA tuples, one per specification.

### Raises
- **ValueError** - If `specs` is not a list of lists.

```

--------------------------------

### Generate and save robustipy figure panels (full grid, reduced layout, individual R2 hexbin)

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/figures.html

These blocks show the plotting logic inside robustipy.figures. The first block builds the full 8-panel grid (a–h) with SHAP, BMA, k-folds, curve, spec-matrix, IC, and beta-distribution panels, then saves the combined figure as '<project_stem>_all.<ext>' (PNG at dpi=800, others with tight bbox). The second block handles the reduced 3-panel layout used when y is multivariate, saving the same combined filename. The third block saves an individual R2 hexbin panel as '<project_stem>_R2hexbin.<ext>' and closes the figure.

```python
        # Generate plots in the grid
        plot_hexbin_r2(results_object, ax1, fig, oddsratio, colormap, title='a.')
        plot_hexbin_log(results_object, ax2, fig, oddsratio, colormap, title='b.')
        cbar_width = 0.046
        cbar_width_fig = cbar_width * ax1.get_position().width
        feature_order = shap_violin(
            ax4,
            shap_vals,
            shap_x,
            shap_cols,
            title='d.',
            clear_yticklabels=True,
            cmap=colormap,
            cbar_ax=ax4_cbar,
            cbar_width=cbar_width,
            cbar_width_fig=cbar_width_fig
        )
        plot_bma(results_object, colormap, ax3, feature_order, title='c.')
        plot_kfolds(results_object, colormap, ax5, title='e.', despine_left=True)
        plot_curve(results_object=results_object, loess=loess, ci=ci, specs=specs,
                   ax=ax6, highlights=highlights, title='f.', oddsratio=oddsratio, colormap=colormap)
        order_idx = _spec_order_idx(results_object, oddsratio)
        # Build the spec-matrix control list: all controls, ordered by
        # SHAP importance (feature_order) where available, with any
        # controls not ranked by SHAP appended at the end.
        controls_set = set(results_object.controls)
        spec_matrix_controls = [c for c in feature_order if c in controls_set]
        for c in results_object.controls:
            if c not in spec_matrix_controls:
                spec_matrix_controls.append(c)
        plot_spec_matrix(
            results_object=results_object,
            ax=ax6m,
            order_idx=order_idx,
            oddsratio=oddsratio,
            controls=spec_matrix_controls,
            bins=spec_matrix_bins,
            heatmap_threshold=spec_matrix_threshold,
            colormap=colormap,
            cbar_ax=ax6m_cbar,
            cbar_width=cbar_width,
            cbar_width_fig=cbar_width_fig
        )
        n_specs = len(results_object.specs_names)
        use_heatmap = (
            spec_matrix_bins is not None
            and spec_matrix_bins > 0
            and n_specs > spec_matrix_threshold
        )
        right_x1 = ax5.get_position().x1
        expanded_left_x0 = ax6_cbar_spacer.get_position().x0
        # Panel g can always expand into the blank spacer to the right of f.
        _set_axes_horizontal_span(expanded_left_x0, right_x1, ax7)
        # Panel h can only expand when the lower row is a dot plot; when a
        # heatmap is used, keep h aligned to the e-panel column.
        if use_heatmap:
            _right_align_axes_to(ax5, ax8)
        else:
            _set_axes_horizontal_span(expanded_left_x0, right_x1, ax8)
        locator = mticker.MaxNLocator(5)
        ax6.xaxis.set_major_locator(locator)
        ax6m.xaxis.set_major_locator(locator)
        ax6.minorticks_off()
        ax6m.minorticks_off()
        ax6.xaxis.set_minor_locator(mticker.NullLocator())
        ax6m.xaxis.set_minor_locator(mticker.NullLocator())
        ax6.set_xlabel('')
        ax6.tick_params(axis='x', which='major', bottom=True, labelbottom=False)
        plot_ic(results_object=results_object, ic=ic, specs=specs, ax=ax7,
                colormap=colormap, title='g.', despine_left=True)
        plot_bdist(results_object=results_object, specs=specs, ax=ax8,
                   oddsratio=oddsratio, highlights=highlights, colormap=colormap,
                   title='h.', despine_left=True)

        # Save the full panel figure
        if ext == 'png':
            plt.savefig(os.path.join(outdir, project_stem + '_all.' + ext), bbox_inches='tight', dpi=800)
        else:
            plt.savefig(os.path.join(outdir, project_stem + '_all.' + ext), bbox_inches='tight')
```

```python
    else:
        # Plot reduced layout when y is multivariate
        fig = plt.figure(figsize=figsize)
        gs = GridSpec(6, 24, wspace=-.25, hspace=5)
        ax1 = fig.add_subplot(gs[0:6, 0:16])
        ax2 = fig.add_subplot(gs[0:3, 17:24])
        ax3 = fig.add_subplot(gs[3:6, 17:24])

        # Generate and save each plot
        plot_curve(results_object=results_object, loess=loess, ci=ci, specs=specs,
                   ax=ax1, highlights=highlights, title='a.', oddsratio=oddsratio, colormap=colormap)
        plot_hexbin_r2(results_object, ax2, fig, oddsratio, colormap, title='b.', side='right')
        plot_bdist(results_object=results_object, specs=specs, ax=ax3,
                   oddsratio=oddsratio, highlights=highlights, colormap=colormap,
                   title='c.', despine_left=True)

        # Save the full panel figure
        plt.savefig(os.path.join(outdir, project_stem + '_all.' + ext), bbox_inches='tight')
```

```python
    # Plot and save individual panels
    fig, ax = plt.subplots(figsize=(8.5, 5))
    plot_hexbin_r2(results_object, ax, fig, oddsratio, colormap)
    plt.savefig(os.path.join(outdir, project_stem + '_R2hexbin.' + ext), bbox_inches='tight')
    plt.close(fig)
```

--------------------------------

### sample_z_specs

Source: https://robustipy.readthedocs.io/en/latest/robustipy.html

Generates a sample of z specifications by randomly selecting subsets of control variables. This method creates a set of binary masks to determine which subsets of the given controls should be included in each z specification. The number of subsets sampled is controlled by z_specs_sample_size.

```APIDOC
## sample_z_specs(controls, z_specs_sample_size)

### Description
Generates a sample of z specifications by randomly selecting subsets of control variables. This method creates a set of binary masks to determine which subsets of the given controls should be included in each z specification. The number of subsets sampled is controlled by z_specs_sample_size.

### Method
sample_z_specs

### Endpoint
OLSRobust.sample_z_specs

### Parameters
#### Path Parameters
- **controls** (list of str) - Required - A list of control variable names from which to build z specifications.
- **z_specs_sample_size** (int) - Required - The number of z specifications to sample. Must be a positive integer.

### Returns
- **space_n_sample** (int) - The number of sampled z specifications.
- **z_specs_sample** (list of tuple of str) - A list where each element is a tuple of selected control variable names representing one sampled z specification.

### Notes
- The method uses sample_z_masks to generate binary inclusion masks.
- If self.seed is defined, it is passed to sample_z_masks to ensure reproducibility.
- Each sampled mask determines a unique subset of controls.
```

--------------------------------

### Prompt for bootstrap draws (draws)

Source: https://robustipy.readthedocs.io/en/latest/_modules/robustipy/utils.html

If draws is None, prompts the user based on environment: uses InquirerPy with IntegerRangeValidator(2, 1000000) in a real TTY, falls back to _ask_integer in Jupyter or real TTY, otherwise defaults to 1000.

```python
    # ───────────────────────────────────────────────────────────────
    # 5) If draws is missing, decide which prompt to run:
    # ───────────────────────────────────────────────────────────────
    if draws is None:
        if use_inquirer:
            # (a) Real TTY: use inquirer.Text with IntegerRangeValidator
            question_draws = [
                inquirer.Text(
                    'draws_inq',
                    message="Enter number of bootstrap draws",
                    validate=IntegerRangeValidator(2, 1000000)
                )
            ]
            # inquirer.prompt(...) returns a dict; we extract and convert to int
            draws = int(inquirer.prompt(question_draws, theme=GreenPassion())['draws_inq'])
        elif in_jupyter or in_real_tty:
            # (b) Jupyter or real TTY fallback: pure input() + validator
            draws = _ask_integer("Enter number of bootstrap draws", 2, 1000000)
        else:
            # (c) Neither TTY nor Jupyter: use default
            draws = 1000
```
Tokens:
Raw