openavmkit.model_runner

Model-run orchestration (the high-level coordinator that drives modeling).

Runs the full configured model menu (MRA, GWR, XGBoost, LightGBM, CatBoost, NGBoost, layered comps, kernel regression, ensembles, and several "naive" baselines) across each model group, with optional variable-importance experiments, then compares model outputs (the "benchmark" comparison) and produces ensemble predictions.

This module is the high-level coordinator that drives :mod:openavmkit.modeling. The main entry points (run_models, try_variables, try_models, finalize_models, ensemble runners) are exposed as wrappers in :mod:openavmkit.pipeline.

History

Formerly named openavmkit.benchmark; renamed to openavmkit.model_runner because the module orchestrates the whole model run, not only the benchmark comparison (and to avoid confusion with the research benchmark/ harness). A deprecating compatibility shim remains at openavmkit.benchmark and is slated for removal in the 0.8.0 release.

Notes

The list of models to run for each main/vacant stage is configured in settings.json under modeling.instructions.<stage>.run. Per-model-group skip lists live under modeling.instructions.<stage>.skip.<model_group>.

BenchmarkResults

BenchmarkResults(df_time, df_stats_test, df_stats_test_post_val, df_stats_full, assessor_in_test=True)

Container for benchmark results.

Attributes:

Name	Type	Description
df_time	DataFrame	DataFrame containing timing information.
df_stats_test	DataFrame	DataFrame with statistics for the test set.
df_stats_test_post_val	DataFrame	DataFrame with statistics for the test set (post-valuation-date only).
df_stats_full	DataFrame	DataFrame with statistics for the full universe.
test_empty	bool	Whether df_stats_test contains no records
full_empty	bool	Whether df_stats_full contains no records
test_post_val_empty	bool	Whether df_stats_test_post_val contains no records

Initialize a BenchmarkResults instance.

Parameters:

Name	Type	Description	Default
df_time	DataFrame	DataFrame containing timing data.	required
df_stats_test	DataFrame	DataFrame with test set statistics.	required
df_stats_test_post_val	DataFrame	DataFrame with test set (post-valuation-date only) statistics.	required
df_stats_full	DataFrame	DataFrame with full universe statistics.	required

Source code in openavmkit/model_runner.py

def __init__(
    self,
    df_time: pd.DataFrame,
    df_stats_test: pd.DataFrame,
    df_stats_test_post_val: pd.DataFrame,
    df_stats_full: pd.DataFrame,
    assessor_in_test: bool = True,
):
    """
    Initialize a BenchmarkResults instance.

    Parameters
    ----------
    df_time : pandas.DataFrame
        DataFrame containing timing data.
    df_stats_test : pandas.DataFrame
        DataFrame with test set statistics.
    df_stats_test_post_val : pandas.DataFrame
        DataFrame with test set (post-valuation-date only) statistics.
    df_stats_full : pandas.DataFrame
        DataFrame with full universe statistics.
    """
    self.df_time = df_time
    self.df_stats_test = df_stats_test
    self.df_stats_test_post_val = df_stats_test_post_val
    self.df_stats_full = df_stats_full

    test_empty = False == (df_stats_test["count_sales"].sum() > 0)
    full_empty = False == (df_stats_full["count_sales"].sum() > 0)

    if df_stats_test_post_val is not None:
        test_post_val_empty = False == (df_stats_test_post_val["count_sales"].sum() > 0)
    else:
        test_post_val_empty = True

    self.test_empty = test_empty
    self.full_empty = full_empty
    self.test_post_val_empty = test_post_val_empty
    self.assessor_in_test = assessor_in_test

print

print()

Return a formatted string summarizing the benchmark results.

Returns:

Type	Description
str	A string that includes timings, test set stats, and universe set stats.

Source code in openavmkit/model_runner.py

def print(self) -> str:
    """
    Return a formatted string summarizing the benchmark results.

    Returns
    -------
    str
        A string that includes timings, test set stats, and universe set stats.
    """
    result = "Timings:\n"
    result += _format_benchmark_df(self.df_time)
    result += "\n\n"
    if (
        self.df_stats_test_post_val is not None
        and not self.test_post_val_empty
    ):
        result += "Holdout set (post-valuation-date only):\n"
        result += (
            "  (Like-for-like vs. the assessor: these sales postdate the valuation date,\n"
            "   so they are out-of-sample for both -- as long as valuation_date is aligned\n"
            "   with the assessor's roll-close date.)\n"
        )
        result += _format_benchmark_df(self.df_stats_test_post_val)
        result += "\n\n"
    result += "Holdout set:\n"
    if self.assessor_in_test:
        result += (
            "  (Assessor shown here because you've declared its values honor this same\n"
            "   holdout (analysis.ratio_study.assessor_holdout: shared). Otherwise it is\n"
            "   left off, since the holdout status of values we didn't generate is unknown.)\n"
        )
    else:
        result += (
            "  (Assessor not shown here: this is a random pre-valuation holdout we draw\n"
            "   ourselves. Our figures are out-of-sample, but we can't know whether values\n"
            "   we didn't generate were held out the same way, so the comparison wouldn't be\n"
            "   like-for-like. If you are the assessor and know the holdout status, see\n"
            "   analysis.ratio_study.assessor_holdout.)\n"
        )
    result += _format_benchmark_df(self.df_stats_test)
    result += "\n\n"
    result += "Study set:\n"
    result += (
        "  (Assessor shown as an audit of the finished roll over all sales -- the standard\n"
        "   IAAO frame, not a predictive holdout. See the sales-chasing check in the ratio\n"
        "   study report for context on interpreting a very tight assessor result.)\n"
    )
    result += _format_benchmark_df(self.df_stats_full)
    result += "\n\n"
    return result

MultiModelResults

MultiModelResults(model_results, benchmark, df_univ, df_sales, drop_assessor_from_test=False)

Container for results from multiple models along with a benchmark.

Attributes: model_results (dict[str, SingleModelResults]): Dictionary mapping model names to their results. benchmark (BenchmarkResults): Benchmark results computed from the model results.

Initialize a MultiModelResults instance.

Parameters:

Name	Type	Description	Default
model_results	dict[str, SingleModelResults]	Dictionary of individual model results.	required
benchmark	BenchmarkResults	Benchmark results.	required
drop_assessor_from_test	bool	Whether the assessor should be left off the pre-valuation "Test set". Stored so that add_model (which recomputes the benchmark, e.g. when the ensemble is added) preserves the same choice as the initial _calc_benchmark call.	False

Source code in openavmkit/model_runner.py

def __init__(
    self, model_results: dict[str, SingleModelResults], benchmark: BenchmarkResults, df_univ: pd.DataFrame, df_sales: pd.DataFrame, drop_assessor_from_test: bool = False
):
    """Initialize a MultiModelResults instance.

    Parameters
    ----------
    model_results: dict[str, SingleModelResults]
        Dictionary of individual model results.
    benchmark: BenchmarkResults
        Benchmark results.
    drop_assessor_from_test: bool
        Whether the assessor should be left off the pre-valuation "Test set". Stored so
        that ``add_model`` (which recomputes the benchmark, e.g. when the ensemble is
        added) preserves the same choice as the initial ``_calc_benchmark`` call.
    """
    self.model_results = model_results
    self.benchmark = benchmark
    self.df_univ_orig = df_univ
    self.df_sales_orig = df_sales
    self.drop_assessor_from_test = drop_assessor_from_test

add_model

add_model(model, results)

Add a new model's results and update the benchmark.

Parameters:

Name	Type	Description	Default
model	str	The model name.	required
results	SingleModelResults	The results for the given model.	required

Source code in openavmkit/model_runner.py

def add_model(self, model: str, results: SingleModelResults):
    """Add a new model's results and update the benchmark.

    Parameters
    ----------
    model: str
        The model name.
    results: SingleModelResults
        The results for the given model.
    """
    self.model_results[model] = results
    # Recalculate the benchmark based on updated model results. Preserve the assessor
    # drop choice -- otherwise adding the ensemble model would silently re-introduce the
    # assessor into the Test-set comparison.
    self.benchmark = _calc_benchmark(
        self.model_results, drop_assessor_from_test=self.drop_assessor_from_test
    )

calc_df_mape

calc_df_mape(df, field_prediction, settings, dep_var, is_land_predictions=False)

Calculate MAPE

Parameters:

Name	Type	Description	Default
df	DataFrame	Dataframe you want to calculate MAPE for	required
field_prediction	str	The field name for predictions.	required
settings	dict	Settings dictionary	required
dep_var	str	The field you're trying to predict	required
is_land_predictions	bool	Are you predicting land values or not. If true, uses the valid_for_land_ratio_study validity flag.	False

Source code in openavmkit/model_runner.py

def calc_df_mape(
    df: pd.DataFrame,
    field_prediction: str,
    settings: dict,
    dep_var: str,
    is_land_predictions: bool = False
):
    """
    Calculate MAPE 

    Parameters
    ----------
    df : pd.DataFrame
        Dataframe you want to calculate MAPE for
    field_prediction : str
        The field name for predictions.
    settings : dict
        Settings dictionary
    dep_var: str
        The field you're trying to predict
    is_land_predictions: bool
        Are you predicting land values or not. If true, uses the `valid_for_land_ratio_study` validity flag.
    """

    # Clean arrays
    y = df[dep_var].to_numpy()
    df[dep_var] = pd.to_numeric(df[dep_var], errors="coerce")
    df[field_prediction] = pd.to_numeric(df[field_prediction], errors="coerce")

    # Get validity field
    valid_field = "valid_for_ratio_study"
    if is_land_predictions:
        valid_field = "valid_for_land_ratio_study"

    # select only values that are not NaN in either and are valid for ratio study:
    df_clean = df[
        df[valid_field] & 
        ~pd.isna(df[dep_var]) & 
        ~pd.isna(df[field_prediction])
    ]

    # Get y & y_pred
    y = df_clean[dep_var].to_numpy()
    y_pred = df_clean[field_prediction].to_numpy()

    # Calculate MAPE
    if len(y) > 0 and len(y_pred) > 0:
        return mean_absolute_percentage_error(y, y_pred)

    return float("nan")

generate_variable_report

generate_variable_report(report, settings, model_group, best_variables)

Generate a variable selection report.

This function updates the MarkdownReport with various threshold values, weights, and summary tables based on the best variables.

Parameters:

Name	Type	Description	Default
report	MarkdownReport	The markdown report object.	required
settings	dict	The settings dictionary.	required
model_group	str	The model group identifier.	required
best_variables	list[str]	List of selected best variables.	required

Returns:

Type	Description
MarkdownReport	The updated markdown report.

Source code in openavmkit/model_runner.py

def generate_variable_report(
    report: MarkdownReport, settings: dict, model_group: str, best_variables: list[str]
):
    """
    Generate a variable selection report.

    This function updates the MarkdownReport with various threshold values, weights, and
    summary tables based on the best variables.

    Parameters
    ----------
    report : MarkdownReport
        The markdown report object.
    settings : dict
        The settings dictionary.
    model_group : str
        The model group identifier.
    best_variables : list[str]
        List of selected best variables.

    Returns
    -------
    MarkdownReport
        The updated markdown report.
    """
    locality = settings.get("locality", {})
    report.set_var("locality", locality.get("name", "...LOCALITY..."))

    mg = get_model_group(settings, model_group)
    report.set_var("val_date", get_valuation_date(settings).strftime("%Y-%m-%d"))
    report.set_var("model_group", mg.get("name", mg))

    instructions = settings.get("modeling", {}).get("instructions", {})
    feature_selection = instructions.get("feature_selection", {})
    thresh = feature_selection.get("thresholds", {})

    report.set_var("thresh_correlation", thresh.get("correlation", ".2f"))
    report.set_var("thresh_enr_coef", thresh.get("enr_coef", ".2f"))
    report.set_var("thresh_vif", thresh.get("vif", ".2f"))
    report.set_var("thresh_p_value", thresh.get("p_value", ".2f"))
    report.set_var("thresh_t_value", thresh.get("t_value", ".2f"))
    report.set_var("thresh_adj_r2", thresh.get("adj_r2", ".2f"))

    weights = feature_selection.get("weights", {})
    df_weights = pd.DataFrame(weights.items(), columns=["Statistic", "Weight"])
    df_weights["Statistic"] = df_weights["Statistic"].map(
        {
            "vif": "VIF",
            "p_value": "P-value",
            "t_value": "T-value",
            "corr_score": "Correlation",
            "enr_coef": "ENR",
            "coef_sign": "Coef. sign",
            "adj_r2": "R-squared",
        }
    )
    df_weights.set_index("Statistic", inplace=True)
    report.set_var("pre_model_weights", df_weights.to_markdown())

    # TODO: Construct summary and post-model tables as needed.
    post_model_table = "...POST MODEL TABLE..."
    report.set_var("post_model_table", post_model_table)

    return report

get_data_split_for

get_data_split_for(model_name, model_engine, model_entry, model_group, location_fields, ind_vars, df_sales, df_universe, settings, dep_var, dep_var_test, fields_cat, interactions, test_keys, train_keys, vacant_only)

Prepare a DataSplit object for a given model.

Parameters:

Name	Type	Description	Default
model_name	str	Model unique identifier	required
model_engine	str	Model engine ("xgboost", "mra", etc.)	required
model_entry	dict	Model parameters	required
model_group	str	The model group identifier.	required
location_fields	list[str] or None	List of location fields.	required
ind_vars	list[str]	List of independent variables.	required
df_sales	DataFrame	Sales DataFrame.	required
df_universe	DataFrame	Universe DataFrame.	required
settings	dict	The settings dictionary.	required
dep_var	str	Dependent variable for training.	required
dep_var_test	str	Dependent variable for testing.	required
fields_cat	list[str]	List of categorical fields.	required
interactions	dict	Dictionary of variable interactions.	required
test_keys	list[str]	Keys for test split.	required
train_keys	list[str]	Keys for training split.	required
vacant_only	bool	Whether to consider only vacant sales.	required

Returns:

Type	Description
DataSplit	A DataSplit object.

Source code in openavmkit/model_runner.py

def get_data_split_for(
    model_name: str,
    model_engine: str,
    model_entry: dict,
    model_group: str,
    location_fields: list[str] | None,
    ind_vars: list[str],
    df_sales: pd.DataFrame,
    df_universe: pd.DataFrame,
    settings: dict,
    dep_var: str,
    dep_var_test: str,
    fields_cat: list[str],
    interactions: dict,
    test_keys: list[str],
    train_keys: list[str],
    vacant_only: bool,
):
    """
    Prepare a DataSplit object for a given model.

    Parameters
    ----------
    model_name: str,
        Model unique identifier
    model_engine : str
        Model engine ("xgboost", "mra", etc.)
    model_entry : dict
        Model parameters
    model_group : str
        The model group identifier.
    location_fields : list[str] or None
        List of location fields.
    ind_vars : list[str]
        List of independent variables.
    df_sales : pandas.DataFrame
        Sales DataFrame.
    df_universe : pandas.DataFrame
        Universe DataFrame.
    settings : dict
        The settings dictionary.
    dep_var : str
        Dependent variable for training.
    dep_var_test : str
        Dependent variable for testing.
    fields_cat : list[str]
        List of categorical fields.
    interactions : dict
        Dictionary of variable interactions.
    test_keys : list[str]
        Keys for test split.
    train_keys : list[str]
        Keys for training split.
    vacant_only : bool
        Whether to consider only vacant sales.

    Returns
    -------
    DataSplit
        A DataSplit object.
    """

    unit = area_unit(settings)
    lenunit = length_unit(settings)

    if model_engine == "local_area":
        _ind_vars = location_fields + [f"bldg_area_finished_{unit}", f"land_area_{unit}"]
    elif model_engine == "multi_mra":
        _ind_vars = [v for v in ind_vars if v not in location_fields]
    elif model_engine == "assessor":
        _ind_vars = ["assr_market_value"]
    elif model_engine == "pass_through":
        field = model_entry.get("field")
        if field is None:
            raise ValueError("pass_through model \"{model_name}\" has no .field parameter!")
        _ind_vars = [field]
    elif model_engine == "ground_truth":
        _ind_vars = ["true_market_value"]
    elif model_engine == "spatial_lag":
        sale_field = get_sale_field(settings)
        field = f"spatial_lag_{sale_field}"
        if vacant_only:
            field = f"{field}_vacant"
        _ind_vars = [field]
    elif model_engine == "spatial_lag_area":
        sale_field = get_sale_field(settings)
        _ind_vars = [
            f"spatial_lag_{sale_field}_impr_{unit}",
            f"spatial_lag_{sale_field}_land_{unit}",
            f"bldg_area_finished_{unit}",
            f"land_area_{unit}",
        ]
    elif model_engine == "catboost":
        df_sales = _clean_categoricals(df_sales, fields_cat, settings)
        df_universe = _clean_categoricals(df_universe, fields_cat, settings)
        _ind_vars = ind_vars
    elif model_engine == "lcomp":
        _ind_vars = ind_vars
    else:
        _ind_vars = ind_vars
        if model_engine == "gwr" or model_engine == "kernel":
            exclude_vars = ["latitude", "longitude", "latitude_norm", "longitude_norm"]
            _ind_vars = [var for var in _ind_vars if var not in exclude_vars]

    _validate_ind_vars_across_frames(
        _ind_vars, df_sales, df_universe, fields_cat, model_name, model_group
    )

    return DataSplit(
        model_name,
        df_sales,
        df_universe,
        model_group,
        settings,
        dep_var,
        dep_var_test,
        _ind_vars,
        fields_cat,
        interactions,
        test_keys,
        train_keys,
        vacant_only=vacant_only,
    )

get_variable_recommendations

get_variable_recommendations(df_sales, df_universe, vacant_only, settings, model_group, variables_to_use=None, tests_to_run=None, do_cross=True, do_report=False, do_plots=False, verbose=False, t=None)

Determine which variables are most likely to be meaningful in a model.

This function examines sales and universe data, applies feature selection via correlations, elastic net regularization, R², p-values, t-values, and VIF, and produces a set of recommended variables along with a written report.

Parameters:

Name	Type	Description	Default
df_sales	DataFrame	The sales data.	required
df_universe	DataFrame	The parcel universe data.	required
vacant_only	bool	Whether to consider only vacant sales.	required
settings	dict	The settings dictionary.	required
model_group	str	The model group to consider.	required
variables_to_use	list[str] or None	A list of variables to use for feature selection. If None, variables are pulled from modeling section	None
tests_to_run	list[str] or None	A list of tests to run. If None, all tests are run. Legal values are "corr", "r2", "p_value", "t_value", "enr", and "vif"	None
do_report	bool	If True, generates a report of the variable selection process.	False
do_plots	bool	If True, prints correlation plots	False
verbose	bool	If True, prints additional debugging information.	False
t	TimingData or None	TimingData object	None

Returns:

Type	Description
dict	A dictionary with keys "variables" (the best variables list) and "report" (the generated report).

Source code in openavmkit/model_runner.py

def get_variable_recommendations(
    df_sales: pd.DataFrame,
    df_universe: pd.DataFrame,
    vacant_only: bool,
    settings: dict,
    model_group: str,
    variables_to_use: list[str] | None = None,
    tests_to_run: list[str] | None = None,
    do_cross: bool = True,
    do_report: bool = False,
    do_plots: bool = False,
    verbose: bool = False,
    t: TimingData = None
) -> dict:
    """Determine which variables are most likely to be meaningful in a model.

    This function examines sales and universe data, applies feature selection via
    correlations, elastic net regularization, R², p-values, t-values, and VIF, and
    produces a set of recommended variables along with a written report.

    Parameters
    ----------
    df_sales : pandas.DataFrame
        The sales data.
    df_universe : pandas.DataFrame
        The parcel universe data.
    vacant_only : bool
        Whether to consider only vacant sales.
    settings : dict
        The settings dictionary.
    model_group : str
        The model group to consider.
    variables_to_use : list[str] or None
        A list of variables to use for feature selection. If None, variables are pulled
        from modeling section
    tests_to_run : list[str] or None
        A list of tests to run. If None, all tests are run. Legal values are "corr",
        "r2", "p_value", "t_value", "enr", and "vif"
    do_report : bool
        If True, generates a report of the variable selection process.
    do_plots: bool, optional
        If True, prints correlation plots
    verbose : bool, optional
        If True, prints additional debugging information.
    t : TimingData or None
        TimingData object

    Returns
    -------
    dict
        A dictionary with keys "variables" (the best variables list) and "report"
        (the generated report).
    """    
    if t is None:
        t = TimingData()

    t.start("variables.markdown")
    report: MarkdownReport = MarkdownReport("variables")
    t.stop("variables.markdown")
    if tests_to_run is None:
        tests_to_run: list[str] = ["corr", "r2", "p_value", "t_value", "enr", "vif"]

    if "sale_price_time_adj" not in df_sales.columns:
        warnings.warn("Time adjustment was not found in sales data. Calculating now...")
        t.start("variables.time_adjustment")
        df_sales = enrich_time_adjustment(df_sales, settings, write=False, verbose=verbose)
        t.stop("variables.time_adjustment")

    t.start("variables.stuff")
    settings_model = settings.get("modeling", {})
    vacant_status = "vacant" if vacant_only else "main"
    model_entries = settings_model.get("models", {}).get(vacant_status, {})
    model_entries = model_entries.get(model_group, model_entries)
    entry: dict | None = model_entries.get("model", model_entries.get("default", {}))
    if variables_to_use is None:
        variables_to_use: list | None = entry.get("ind_vars", None)

    if variables_to_use is None or len(variables_to_use) == 0:
        raise ValueError("No independent variables provided! Please define some!")

    categoricals = get_fields_categorical(settings, df_sales, include_boolean=False)

    flagged = []
    categoricals_to_use = [x for x in variables_to_use if x in categoricals]
    for variable in categoricals_to_use:
        if df_sales[variable].nunique() > 50:
            warnings.warn(
                f"Variable '{variable}' has more than 50 unique values. No variable analysis will be done on it and it will not be auto-dropped. Hope you know what you're doing!"
            )
            flagged.append(variable)

    if len(flagged) > 0:
        variables_to_use = [
            variable for variable in variables_to_use if variable not in flagged
        ]

    # Check for duplicate variables in variables_to_use
    if variables_to_use is not None:
        seen_vars = set()
        duplicates = []
        deduped_vars = []

        for var in variables_to_use:
            if var in seen_vars:
                duplicates.append(var)
            else:
                seen_vars.add(var)
                deduped_vars.append(var)

        if duplicates:
            print(
                f"\n⚠️ WARNING: Found duplicate variables in variables_to_use: {duplicates}"
            )
            print(f"Using only the first occurrence of each variable for analysis.")
            variables_to_use = deduped_vars

    # Check for duplicate columns in DataFrame (could happen from merges)
    duplicate_cols = df_sales.columns[df_sales.columns.duplicated()].tolist()
    if duplicate_cols:
        print(
            f"\n⚠️ WARNING: Found duplicate columns in sales DataFrame: {duplicate_cols}"
        )
        print(
            f"This could cause errors in analysis. Keeping only first occurrence of each column."
        )
        df_sales = df_sales.loc[:, ~df_sales.columns.duplicated()]

    duplicate_cols_univ = df_universe.columns[df_universe.columns.duplicated()].tolist()
    if duplicate_cols_univ:
        print(
            f"\n⚠️ WARNING: Found duplicate columns in universe DataFrame: {duplicate_cols_univ}"
        )
        print(
            f"This could cause errors in analysis. Keeping only first occurrence of each column."
        )
        df_universe = df_universe.loc[:, ~df_universe.columns.duplicated()]

    t.stop("variables.stuff")

    t.start("variables.prepare_ds")
    ds = _prepare_ds(
        "var_recs", df_sales, df_universe, model_group, vacant_only, settings, variables_to_use
    )
    t.stop("variables.prepare_ds")
    t.start("variables.one_hot")
    ds = ds.encode_categoricals_with_one_hot()
    t.stop("variables.one_hot")
    t.start("variables.split")
    ds.split()
    t.stop("variables.split")
    feature_selection = (
        settings.get("modeling", {})
        .get("instructions", {})
        .get("feature_selection", {})
    )
    thresh = feature_selection.get("thresholds", {})

    X_sales = ds.X_sales[ds.ind_vars]
    y_sales = ds.y_sales

    X_univ = ds.X_univ[ds.ind_vars]

    t.start("variables.rep")
    rep_results = calc_representation(X_sales, X_univ, do_plots=do_plots)
    bad_vars = rep_results["bad_vars"]
    t.stop("variables.rep")

    # Remove bad variables
    ind_vars = [var for var in ds.ind_vars if var not in bad_vars]

    if "corr" in tests_to_run:
        # Correlation
        X_corr = ds.df_sales[[ds.dep_var] + ind_vars]
        t.start("variables.corr")
        corr_results = calc_correlations(X_corr, thresh.get("correlation", 0.1), do_plots=do_plots)

        # Remove bad variables
        ind_vars = [var for var in ds.ind_vars if var not in corr_results["bad_vars"]]
        t.stop("variables.corr")
    else:
        corr_results = None

    if "enr" in tests_to_run:
        # Elastic net regularization
        try:
            t.start("variables.enr")
            enr_coefs = calc_elastic_net_regularization(
                X_sales, y_sales, thresh.get("enr", 0.01)
            )
            t.stop("variables.enr")
        except ValueError as e:
            nulls_in_X = X_sales[X_sales.isna().any(axis=1)]
            print(f"Found {len(nulls_in_X)} rows with nulls in X:")
            # identify columns with nulls in them:
            cols_with_null = nulls_in_X.columns[nulls_in_X.isna().any()].tolist()
            print(f"Columns with nulls: {cols_with_null}")
            raise e
    else:
        enr_coefs = None

    if "r2" in tests_to_run:
        # R² values
        t.start("variables.r2")
        r2_values = calc_r2(ds.df_sales, ind_vars, y_sales)
        t.stop("variables.r2")
    else:
        r2_values = None

    if "p_value" in tests_to_run:
        # P Values
        t.start("variables.p")
        p_values = calc_p_values_recursive_drop(
            X_sales, y_sales, thresh.get("p_value", 0.05)
        )
        t.stop("variables.p")
    else:
        p_values = None

    if "t_value" in tests_to_run:
        # T Values
        t.start("variables.t")
        t_values = calc_t_values_recursive_drop(
            X_sales, y_sales, thresh.get("t_value", 2)
        )
        t.stop("variables.t")
    else:
        t_values = None

    if "vif" in tests_to_run:
        t.start("variables.vif")
        # VIF
        # Filter out boolean columns before VIF calculation
        bool_cols = []
        vif_X = X_sales.copy()

        for col in X_sales.columns:
            # Check if column is boolean or contains only 0/1 values
            if X_sales[col].dtype == bool or (
                X_sales[col].isin([0, 1, True, False]).all()
                and len(X_sales[col].unique()) <= 2
            ):
                bool_cols.append(col)

        if bool_cols:
            vif_X = vif_X.drop(columns=bool_cols)

        # Don't run VIF if we have no columns left or too few rows
        if 0 < vif_X.shape[1] < len(vif_X):
            vif = calc_vif_recursive_drop(vif_X, thresh.get("vif", 10), settings)

            # Add boolean columns back to the final VIF results with NaN VIF values
            if bool_cols and vif is not None and "final" in vif:
                for bool_col in bool_cols:
                    vif["final"] = pd.concat(
                        [
                            vif["final"],
                            pd.DataFrame(
                                {"variable": [bool_col], "vif": [float("nan")]}
                            ),
                        ],
                        ignore_index=True,
                    )
        else:
            if verbose:
                print(
                    "Skipping VIF calculation - not enough non-boolean variables or samples"
                )
            vif = {
                "initial": pd.DataFrame(columns=["variable", "vif"]),
                "final": pd.DataFrame(columns=["variable", "vif"]),
            }
        t.stop("variables.vif")
    else:
        vif = None

    t.start("variables.calc_recs")
    # Generate final results & recommendations
    df_results = _calc_variable_recommendations(
        ds=ds,
        settings=settings,
        rep_results=rep_results,
        correlation_results=corr_results,
        enr_results=enr_coefs,
        r2_values_results=r2_values,
        p_values_results=p_values,
        t_values_results=t_values,
        vif_results=vif,
        report=report
    )
    t.stop("variables.calc_recs")

    t.start("variables.final_stuff")
    curr_variables = df_results["variable"].tolist()
    best_variables = curr_variables.copy()
    best_score = float("inf")

    df_cross = df_results.copy()
    y = ds.y_sales

    t.start("variables.final_stuff.while")
    if do_cross:
        while len(curr_variables) > 0:
            X = ds.df_sales[curr_variables]
            t.start("variables.final_stuff.while.cross")
            cv_score = calc_cross_validation_score(X, y)
            t.stop("variables.final_stuff.while.cross")
            if cv_score < best_score:
                best_score = cv_score
                best_variables = curr_variables.copy()
            worst_idx = df_cross["weighted_score"].idxmin()
            worst_variable = df_cross.loc[worst_idx, "variable"]
            curr_variables.remove(worst_variable)
            # Remove the variable from the results dataframe.
            df_cross = df_cross[df_cross["variable"].ne(worst_variable)]
    t.stop("variables.final_stuff.while")

    # Create a table from the list of best variables.
    df_best = pd.DataFrame(best_variables, columns=["Variable"])
    df_best["Rank"] = range(1, len(df_best) + 1)
    df_best["Description"] = df_best["Variable"]

    t.start("variables.final_stuff.apply_dd")
    df_best = _apply_dd_to_df_rows(
        df_best, "Variable", settings, ds.one_hot_descendants, "name"
    )
    df_best = _apply_dd_to_df_rows(
        df_best, "Description", settings, ds.one_hot_descendants, "description"
    )
    t.stop("variables.final_stuff.apply_dd")
    df_best = df_best[["Rank", "Variable", "Description"]]
    df_best.loc[df_best["Variable"].eq(df_best["Description"]), "Description"] = ""
    df_best.set_index("Rank", inplace=True)

    if do_report:
        report.set_var("summary_table", df_best.to_markdown())
        report = generate_variable_report(report, settings, model_group, best_variables)
    else:
        report = None
    t.stop("variables.final_stuff")

    print(t.print())

    return {"variables": best_variables, "report": report, "df_results": df_results}

run_ensemble

run_ensemble(df_sales, df_universe, model_group, vacant_only, dep_var, dep_var_test, outpath, all_results, settings, verbose=False)

Run an ensemble model based on the provided parameters.

This function optimizes the ensemble model and runs it, returning the results and the list of models used in the ensemble.

Parameters:

Name	Type	Description	Default
df_sales	DataFrame or None	Sales DataFrame. If None, it will be read from the MultiModelResults.	required
df_universe	DataFrame or None	Universe DataFrame. If None, it will be read from the MultiModelResults.	required
model_group	str	Model group identifier.	required
vacant_only	bool	Whether to use only vacant sales.	required
dep_var	str	Dependent variable for training.	required
dep_var_test	str	Dependent variable for testing.	required
outpath	str	Output path for saving results.	required
all_results	MultiModelResults	MultiModelResults containing all model results.	required
settings	dict	Settings dictionary.	required
verbose	bool	If True, prints additional information. Defaults to False.	False

Returns:

Type	Description
tuple[SingleModelResults, list[str]]	A tuple containing the SingleModelResults of the ensemble model and a list of models used in the ensemble.

Source code in openavmkit/model_runner.py

def run_ensemble(
    df_sales: pd.DataFrame | None,
    df_universe: pd.DataFrame | None,
    model_group: str,
    vacant_only: bool,
    dep_var: str,
    dep_var_test: str,
    outpath: str,
    all_results: MultiModelResults,
    settings: dict,
    verbose: bool = False,
) -> tuple[SingleModelResults, list[str]]:
    """Run an ensemble model based on the provided parameters.

    This function optimizes the ensemble model and runs it, returning the results and the list of models used in the ensemble.

    Parameters
    ----------
    df_sales : pandas.DataFrame or None
        Sales DataFrame. If None, it will be read from the MultiModelResults.
    df_universe : pandas.DataFrame or None
        Universe DataFrame. If None, it will be read from the MultiModelResults.
    model_group : str
        Model group identifier.
    vacant_only : bool
        Whether to use only vacant sales.
    dep_var : str
        Dependent variable for training.
    dep_var_test : str
        Dependent variable for testing.
    outpath : str
        Output path for saving results.
    all_results : MultiModelResults
        MultiModelResults containing all model results.
    settings : dict
        Settings dictionary.
    verbose : bool, optional
        If True, prints additional information. Defaults to False.

    Returns
    -------
    tuple[SingleModelResults, list[str]]
        A tuple containing the SingleModelResults of the ensemble model and a list of models used in the ensemble.
    """
    if verbose:
        print("Optimizing ensemble...")

    ensemble_list = _optimize_ensemble(
        df_sales,
        df_universe,
        model_group,
        vacant_only,
        dep_var,
        dep_var_test,
        all_results,
        settings,
        verbose=verbose,
        ensemble_list=None,
    )
    if verbose:
        print("Running ensemble...")
    ensemble = _run_ensemble(
        df_sales,
        df_universe,
        model_group,
        vacant_only=vacant_only,
        dep_var=dep_var,
        dep_var_test=dep_var_test,
        outpath=outpath,
        ensemble_list=ensemble_list,
        all_results=all_results,
        settings=settings,
        verbose=verbose,
    )
    if verbose:
        print("Finished ensemble!")
    return ensemble, ensemble_list

run_models

run_models(sup, settings, save_params=False, use_saved_params=True, save_results=False, verbose=False, run_main=True, run_vacant=True, run_ensemble=True, do_shaps=False, do_plots=False)

Runs predictive models on the given SalesUniversePair.

This function takes detailed instructions from the provided settings dictionary and handles all the internal details like splitting the data, training the models, and saving the results. It performs basic statistic analysis on each model, and optionally combines results into an ensemble model.

If "run_main" is true, it will run normal (full market value) models. If "run_vacant" is true, it will run vacant models as well -- models that only use vacant sales as evidence to generate land values.

This function iterates over model groups and runs models for both main and vacant cases.

Parameters:

Name	Type	Description	Default
sup	SalesUniversePair	Sales and universe data.	required
settings	dict	The settings dictionary.	required
save_params	bool	Whether to save model parameters.	False
use_saved_params	bool	Whether to use saved model parameters.	True
save_results	bool	Whether to save model results.	False
verbose	bool	If True, prints additional information.	False
run_main	bool	Whether to run main (non-vacant) models.	True
run_vacant	bool	Whether to run vacant models.	True
run_ensemble	bool	Whether to run ensemble models.	True
do_shaps	bool	Whether to compute SHAP values.	False
do_plots	bool	Whether to plot scatterplots	False

Returns:

Type	Description
MultiModelResults	The MultiModelResults containing all model results and benchmarks.

Source code in openavmkit/model_runner.py

def run_models(
    sup: SalesUniversePair,
    settings: dict,
    save_params: bool = False,
    use_saved_params: bool = True,
    save_results: bool = False,
    verbose: bool = False,
    run_main: bool = True,
    run_vacant: bool = True,
    run_ensemble: bool = True,
    do_shaps: bool = False,
    do_plots: bool = False
):
    """
    Runs predictive models on the given SalesUniversePair.

    This function takes detailed instructions from the provided settings dictionary and handles all the internal
    details like splitting the data, training the models, and saving the results. It performs basic statistic analysis
    on each model, and optionally combines results into an ensemble model.

    If "run_main" is true, it will run normal (full market value) models.
    If "run_vacant" is true, it will run vacant models as well -- models that only use vacant sales as evidence
    to generate land values.

    This function iterates over model groups and runs models for both main and vacant cases.

    Parameters
    ----------
    sup : SalesUniversePair
        Sales and universe data.
    settings : dict
        The settings dictionary.
    save_params : bool, optional
        Whether to save model parameters.
    use_saved_params : bool, optional
        Whether to use saved model parameters.
    save_results : bool, optional
        Whether to save model results.
    verbose : bool, optional
        If True, prints additional information.
    run_main : bool, optional
        Whether to run main (non-vacant) models.
    run_vacant : bool, optional
        Whether to run vacant models.
    run_ensemble : bool, optional
        Whether to run ensemble models.
    do_shaps : bool, optional
        Whether to compute SHAP values.
    do_plots : bool, optional
        Whether to plot scatterplots

    Returns
    -------
    MultiModelResults
        The MultiModelResults containing all model results and benchmarks.
    """

    t = TimingData()

    t.start("setup")
    s = settings
    s_model = s.get("modeling", {})
    s_inst = s_model.get("instructions", {})
    model_groups = s_inst.get("model_groups", [])

    df_univ = sup["universe"]

    if len(model_groups) == 0:
        model_groups = get_model_group_ids(settings, df_univ)

    dict_all_results = {}
    t.stop("setup")

    t.start("run model groups")
    for model_group in model_groups:
        t.start(f"model group: {model_group}")
        for main_vacant in ["main", "vacant"]:
            if main_vacant == "main" and not run_main:
                continue
            if main_vacant == "vacant" and not run_vacant:
                continue

            models_to_skip = s_inst.get(main_vacant, {}).get("skip", {}).get(model_group, [])

            if "all" in models_to_skip:
                if verbose:
                    print(
                        f"Skipping all models for model_group: {model_group}/{main_vacant}"
                    )
                continue

            if verbose:
                print("")
                print("")
                print("******************************************************")
                print(f"Running models for model_group: {model_group}")
                print("******************************************************")
                print("")
                print("")

            mg_results = _run_models(
                sup,
                model_group,
                settings,
                main_vacant,
                save_params,
                use_saved_params,
                save_results,
                verbose,
                run_ensemble,
                do_shaps=do_shaps,
                do_plots=do_plots
            )
            if mg_results is not None and save_results:
                dict_all_results[model_group] = mg_results
        t.stop(f"model group: {model_group}")
    t.stop("run model groups")

    if save_results:
        t.start("write")
        write_out_all_results(sup, dict_all_results, settings)
        t.stop("write")

    print("**********TIMING FOR RUN ALL MODELS***********")
    print(t.print())
    print("***********************************************")

    return dict_all_results

run_one_model

run_one_model(df_sales, df_universe, vacant_only, model_group, model_name, model_entries, settings, dep_var, dep_var_test, best_variables, fields_cat, outpath, save_params, use_saved_params, save_results, verbose=False, test_keys=None, train_keys=None)

Run a single model based on provided parameters and return its results.

Parameters:

Name	Type	Description	Default
df_sales	DataFrame	Sales DataFrame.	required
df_universe	DataFrame	Universe DataFrame.	required
vacant_only	bool	Whether to use only vacant sales.	required
model_group	str	Model group identifier.	required
model_name	str	Model's unique identifier.	required
model_entries	dict	Dictionary of model configuration entries.	required
settings	dict	Settings dictionary.	required
dep_var	str	Dependent variable for training.	required
dep_var_test	str	Dependent variable for testing.	required
best_variables	list[str]	List of best variables selected.	required
fields_cat	list[str]	List of categorical fields.	required
outpath	str	Output path for saving results.	required
save_params	bool	Whether to save parameters.	required
use_saved_params	bool	Whether to use saved parameters.	required
save_results	bool	Whether to save results.	required
verbose	bool	If True, prints additional information.	False
test_keys	list[str] or None	Optional list of test keys (will be read from disk if not provided).	None
train_keys	list[str] or None	Optional list of training keys (will be read from disk if not provided).	None

Returns:

Type	Description
SingleModelResults or None	SingleModelResults if successful, else None.

Source code in openavmkit/model_runner.py

def run_one_model(
    df_sales: pd.DataFrame,
    df_universe: pd.DataFrame,
    vacant_only: bool,
    model_group: str,
    model_name: str,
    model_entries: dict,
    settings: dict,
    dep_var: str,
    dep_var_test: str,
    best_variables: list[str],
    fields_cat: list[str],
    outpath: str,
    save_params: bool,
    use_saved_params: bool,
    save_results: bool,
    verbose: bool = False,
    test_keys: list[str] | None = None,
    train_keys: list[str] | None = None,
) -> SingleModelResults | None:
    """
    Run a single model based on provided parameters and return its results.

    Parameters
    ----------
    df_sales : pandas.DataFrame
        Sales DataFrame.
    df_universe : pandas.DataFrame
        Universe DataFrame.
    vacant_only : bool
        Whether to use only vacant sales.
    model_group : str
        Model group identifier.
    model_name : str
        Model's unique identifier.
    model_entries : dict
        Dictionary of model configuration entries.
    settings : dict
        Settings dictionary.
    dep_var : str
        Dependent variable for training.
    dep_var_test : str
        Dependent variable for testing.
    best_variables : list[str]
        List of best variables selected.
    fields_cat : list[str]
        List of categorical fields.
    outpath : str
        Output path for saving results.
    save_params : bool
        Whether to save parameters.
    use_saved_params : bool
        Whether to use saved parameters.
    save_results : bool
        Whether to save results.
    verbose : bool, optional
        If True, prints additional information.
    test_keys : list[str] or None, optional
        Optional list of test keys (will be read from disk if not provided).
    train_keys : list[str] or None, optional
        Optional list of training keys (will be read from disk if not provided).

    Returns
    -------
    SingleModelResults or None
        SingleModelResults if successful, else None.
    """

    t = TimingData()

    t.start("setup")

    entry: dict | None = model_entries.get(model_name, None)
    default_entry: dict | None = model_entries.get("default", {})
    entry_is_default = entry is None
    if entry is None:
        entry = default_entry
        if entry is None:
            raise ValueError(
                f"Model entry for {model_name} not found, and there is no default entry!"
            )
    # No "model" key means the name IS the engine (e.g. "mra", "lightgbm"). That fallback is
    # load-bearing for plain entries, but it silently turns an *alias* with no entry in this
    # model group's block (e.g. "lgbm_x") into a bogus engine name -- see the dispatch else.
    model_engine = entry.get("model", model_name)
    if model_engine == "default":
        # this isn't a real model, just a settings object to fill in for others
        return None

    if "*" in model_engine:
        sales_chase = 0.01
        model_engine = model_engine.replace("*", "")
    else:
        sales_chase = False

    if verbose:
        print(f"------------------------------------------------")
        print(f"Running model {model_name} on {len(df_sales)} rows...")

    are_ind_vars_default = entry.get("ind_vars", None) is None
    ind_vars: list | None = entry.get("ind_vars", default_entry.get("ind_vars", None))

    if ind_vars is None:
        raise ValueError(f"ind_vars not found for model {model_name}")
    # De-duplicate with a DETERMINISTIC order: sorted() is independent of PYTHONHASHSEED, so the
    # feature column order (and thus XGBoost/LightGBM column-subsampling selections) is reproducible
    # across processes and separate invocations. Plain list(set(...)) is hash-ordered and was a
    # latent source of run-to-run nondeterminism (see run_one_model_cv worker pinning).
    ind_vars = sorted(set(ind_vars))

    if are_ind_vars_default:
        if (best_variables is not None) and (set(ind_vars) != set(best_variables)):
            if verbose:
                print(
                    f"--> using default variables, auto-optimized variable list: {best_variables}"
                )
            ind_vars = best_variables

    interactions = get_variable_interactions(entry, settings, df_sales)
    location_fields = entry.get("locations", get_locations(settings, df_sales))

    if test_keys is None or train_keys is None:
        test_keys, train_keys = _read_split_keys(model_group)
    t.stop("setup")

    t.start("data split")
    ds = get_data_split_for(
        model_name=model_name,
        model_engine=model_engine,
            model_entry=entry,
            model_group=model_group,
            location_fields=location_fields,
            ind_vars=ind_vars,
            df_sales=df_sales,
            df_universe=df_universe,
            settings=settings,
            dep_var=dep_var,
            dep_var_test=dep_var_test,
            fields_cat=fields_cat,
            interactions=interactions,
            test_keys=test_keys,
            train_keys=train_keys,
            vacant_only=vacant_only,
        )

    # safeguards against invalid splits
    n_sales = len(ds.df_sales) if ds.df_sales is not None else 0
    n_train = len(ds.df_train) if ds.df_train is not None else 0
    n_test  = len(ds.df_test)  if ds.df_test  is not None else 0
    p = ds.X_train.shape[1] if getattr(ds, "X_train", None) is not None else 0

    if n_train == 0 or p == 0:
        # compute some helpful diagnostics
        missing_vars = []
        if p == 0:
            # requested vars that are not present in train
            train_cols = set(ds.df_train.columns) if ds.df_train is not None else set()
            missing_vars = [v for v in ds.ind_vars if v not in train_cols]

        why = []
        if n_train == 0:
            if n_test == n_sales and n_sales > 0:
                why.append("all sales ended up in the test split (train set empty) — check split keys")
            else:
                why.append("filters/slicing removed all training rows")
        if p == 0:
            why.append(f"no usable features in X_train (missing ind_vars: {missing_vars[:20]}{'...' if len(missing_vars)>20 else ''})")

        warnings.warn(
            f"Skipping model {model_group}/{model_name} ({model_engine}): "
            f"sales={n_sales}, train={n_train}, test={n_test}, X_train_cols={p}. "
            + " ".join(why),
            RuntimeWarning
        )
        return None

    t.stop("data split")

    t.start("setup")
    if len(ds.y_sales) < 15:
        if verbose:
            print(f"--> model {model_name} has less than 15 sales. Skipping...")
        return None

    optimize_vars = entry.get("optimize_vars", False)
    intercept = entry.get("intercept", True)
    # Per-model opt-in to log-target training (mra / multi_mra only). Fitting on log(price) keeps
    # the linear models from extrapolating negative values; the model exponentiates its own
    # predictions back to price space, so this stays contained to the model (no dep_var changes).
    log = entry.get("log", False)
    n_trials = entry.get("n_trials", 50)
    use_gpu = entry.get("use_gpu", True)
    seed = get_model_seed(settings)
    t.stop("setup")

    t.start("run")
    if model_engine == "garbage":
        results = run_garbage(
            ds, normal=False, sales_chase=sales_chase, verbose=verbose
        )
    elif model_engine == "garbage_normal":
        results = run_garbage(ds, normal=True, sales_chase=sales_chase, verbose=verbose)
    elif model_engine == "mean":
        results = run_average(
            ds, average_type="mean", sales_chase=sales_chase, verbose=verbose
        )
    elif model_engine == "median":
        results = run_average(
            ds, average_type="median", sales_chase=sales_chase, verbose=verbose
        )
    elif model_engine == "naive_area":
        results = run_naive_area(ds, sales_chase=sales_chase, verbose=verbose)
    elif model_engine == "local_area":
        results = run_local_area(
            ds,
            location_fields=location_fields,
            sales_chase=sales_chase,
            verbose=verbose,
        )
    elif model_engine == "assessor" or model_engine == "pass_through":
        results = run_pass_through(ds, model_engine, verbose=verbose)
    elif model_engine == "ground_truth":
        results = run_ground_truth(ds, verbose=verbose)
    elif model_engine == "spatial_lag":
        results = run_spatial_lag(ds, per_area=False, verbose=verbose)
    elif model_engine == "spatial_lag_area":
        results = run_spatial_lag(ds, per_area=True, verbose=verbose)
    elif model_engine == "mra":
        results = run_mra(ds, intercept=intercept, verbose=verbose, log=log)
    elif model_engine == "multi_mra":
        results = run_multi_mra(ds, outpath, location_fields, optimize_vars=optimize_vars, intercept=intercept, verbose=verbose, log=log)
    elif model_engine == "kernel":
        results = run_kernel(
            ds, outpath, save_params, use_saved_params, verbose=verbose
        )
    elif model_engine == "gwr":
        results = run_gwr(ds, outpath, save_params, use_saved_params, verbose=verbose)
    elif model_engine == "xgboost":
        results = run_xgboost(
            ds, outpath, save_params, use_saved_params, n_trials=n_trials, verbose=verbose, seed=seed
        )
    elif model_engine == "lightgbm":
        results = run_lightgbm(
            ds, outpath, save_params, use_saved_params, n_trials=n_trials, verbose=verbose, seed=seed
        )
    elif model_engine == "catboost":
        results = run_catboost(
            ds, outpath, save_params, use_saved_params, n_trials=n_trials, verbose=verbose, use_gpu=use_gpu, seed=seed
        )
    elif model_engine == "ngboost":
        results = run_ngboost(
            ds, outpath, save_params, use_saved_params, n_trials=n_trials, verbose=verbose, seed=seed
        )
    elif model_engine == "lcomp":
        results = run_layeredcomp(
            ds, outpath, save_params, use_saved_params, n_trials=n_trials, verbose=verbose, seed=seed
        )
    else:
        if entry_is_default:
            raise ValueError(
                f'Model "{model_name}" (model_group "{model_group}") has no entry in '
                f'modeling.models.{"vacant" if vacant_only else "main"}, so it fell back to the '
                f'"default" entry, which does not name an engine -- leaving "{model_engine}" '
                f"(the model's own name) as the engine, and that is not a known engine. If "
                f'"{model_name}" is an alias, it needs its own entry with a "model" key. Note '
                f"that a per-model-group override block REPLACES the top-level block wholesale, "
                f"so every name in instructions.run must be defined in whichever block applies "
                f"to this group."
            )
        raise ValueError(
            f'Model engine "{model_engine}" (from model "{model_name}", model_group '
            f'"{model_group}") not found!'
        )
    t.stop("run")

    if results is None:
        return None

    if ds.vacant_only:
        # If this is a vacant model, we attempt to load a corresponding "full value" model
        max_trim = _get_max_ratio_study_trim(settings, results.ds.model_group)

    if save_results:
        t.start("write")
        main_vacant = "vacant" if vacant_only else "main"
        location = get_model_location(settings, main_vacant, model_name, model_group)
        _write_model_results(results, outpath, settings, location, verbose=verbose)
        t.stop("write")

    return results

run_one_model_cv

run_one_model_cv(df_sales, df_universe, vacant_only, model_group, model_name, model_entries, settings, dep_var, dep_var_test, best_variables, fields_cat, outpath, save_params, use_saved_params, save_results, verbose=False)

Nested cross-validation wrapper around :func:run_one_model.

Two phases, both driven through the ordinary single-model path so every engine, ratio study, and back-transform is reused unchanged:

• Phase 1 (holdout): run the model once per fold — each fold trains on the other folds (parcels intact, post-val excluded) and predicts its own held-out slice. Because each fold re-tunes from scratch on its own training data (distinct outpath → distinct params.json), the held-out slice is leakage-free w.r.t. training and HP selection. The slices stitch into an out-of-fold (OOF) prediction for every trainable sale.
• Phase 2 (study/ship): refit once on ALL trainable sales; its predictions on all sales (study) and the universe (shipped values) are kept as-is, and its held-out post-val predictions cover the post-valuation sales in the report (leakage-free — post-val never trains).

The returned result is the Phase-2 result with its test side replaced by the 100%-coverage OOF frame (see :meth:SingleModelResults.override_test_predictions), so pred_test is the honest full-coverage holdout, while pred_sales / pred_univ are the study / shipped values.

Falls back to a single split only when no folds.csv exists (legacy mode / skipped group). Log-target (log_-prefixed dep_var_test) is handled uniformly: :meth:SingleModelResults.override_test_predictions re-applies the same log back-transform the ordinary path does.

Source code in openavmkit/model_runner.py

def run_one_model_cv(
    df_sales: pd.DataFrame,
    df_universe: pd.DataFrame,
    vacant_only: bool,
    model_group: str,
    model_name: str,
    model_entries: dict,
    settings: dict,
    dep_var: str,
    dep_var_test: str,
    best_variables: list[str],
    fields_cat: list[str],
    outpath: str,
    save_params: bool,
    use_saved_params: bool,
    save_results: bool,
    verbose: bool = False,
) -> SingleModelResults | None:
    """Nested cross-validation wrapper around :func:`run_one_model`.

    Two phases, both driven through the ordinary single-model path so every engine, ratio
    study, and back-transform is reused unchanged:

    - **Phase 1 (holdout):** run the model once per fold — each fold trains on the other folds
      (parcels intact, post-val excluded) and predicts its own held-out slice. Because each
      fold re-tunes from scratch on its own training data (distinct ``outpath`` → distinct
      ``params.json``), the held-out slice is leakage-free w.r.t. training *and* HP selection.
      The slices stitch into an out-of-fold (OOF) prediction for every trainable sale.
    - **Phase 2 (study/ship):** refit once on ALL trainable sales; its predictions on all sales
      (study) and the universe (shipped values) are kept as-is, and its held-out post-val
      predictions cover the post-valuation sales in the report (leakage-free — post-val never
      trains).

    The returned result is the Phase-2 result with its test side replaced by the 100%-coverage
    OOF frame (see :meth:`SingleModelResults.override_test_predictions`), so ``pred_test`` is the
    honest full-coverage holdout, while ``pred_sales`` / ``pred_univ`` are the study / shipped
    values.

    Falls back to a single split only when no ``folds.csv`` exists (legacy mode / skipped group).
    Log-target (``log_``-prefixed ``dep_var_test``) is handled uniformly:
    :meth:`SingleModelResults.override_test_predictions` re-applies the same log back-transform
    the ordinary path does.
    """
    fold_data = _read_fold_keys(model_group)
    if fold_data is None or fold_data["n_folds"] < 2:
        return run_one_model(
            df_sales, df_universe, vacant_only, model_group, model_name, model_entries,
            settings, dep_var, dep_var_test, best_variables, fields_cat, outpath,
            save_params, use_saved_params, save_results, verbose=verbose,
        )

    train_all = [str(k) for k in fold_data["train_all"]]
    post_val = [str(k) for k in fold_data["post_val"]]
    cv_prod_mode = settings.get("modeling", {}).get("instructions", {}).get(
        "cv_production_params", "aggregate"
    )
    # Out-of-fold contributions. Off => the legacy behavior, where the Phase-2 model explains
    # the OOF holdout frame (attributions and predictions from different models; see
    # `_stitch_oof_contributions`).
    emit_oof_contribs = bool(
        settings.get("modeling", {}).get("instructions", {}).get("cv_oof_contributions", True)
    )

    # ---- Phase 1: per-fold out-of-fold predictions ----
    # Folds are independent and each is internally deterministic, so running them in parallel is
    # bit-identical to sequential. To avoid CPU oversubscription (each fold's tuner already runs a
    # trial-level thread pool), each worker caps its tuner threads to a fair share of the cores.
    tasks = [
        (k, [str(x) for x in hk], [str(x) for x in tk])
        for k, (hk, tk) in enumerate(fold_data["folds"])
        if len(hk) > 0 and len(tk) > 0
    ]
    cpu = os.cpu_count() or 2
    cv_max_workers = int(settings.get("modeling", {}).get("instructions", {}).get(
        "cv_max_workers", min(len(tasks), max(1, cpu - 2))
    ))
    cv_max_workers = max(1, min(cv_max_workers, len(tasks)))
    fold_args = (
        df_sales, df_universe, vacant_only, model_group, model_name, model_entries,
        settings, dep_var, dep_var_test, best_variables, fields_cat, outpath,
        use_saved_params, verbose, emit_oof_contribs and save_results,
    )

    # Parallelism comes from the fold axis, and each fold tunes its trials SERIALLY (1 thread).
    # This is deliberate: XGBoost/LightGBM threaded trial-concurrency is not numerically identical
    # across concurrency levels (tiny MAPE differences flip which trial wins), so varying the
    # trial-thread count would make results depend on cv_max_workers. Fixing trials to serial makes
    # every CV run bit-identical regardless of worker count, and fold_workers x 1 thread never
    # oversubscribes. (The single-split path is untouched — it keeps full trial parallelism.)
    if cv_max_workers > 1 and len(tasks) > 1:
        if verbose:
            print(f"Running {len(tasks)} CV folds for {model_group}/{model_name} across "
                  f"{cv_max_workers} workers (serial tuning per fold)...")
        # Pin the math libraries to a single thread in the worker processes (inherited at spawn).
        # XGBoost/LightGBM `hist` otherwise use OpenMP threads whose count varies with system load,
        # and parallel float reductions are non-associative -> non-reproducible results run-to-run.
        # Single-threaded per worker makes each fold deterministic; parallelism comes from folds.
        # PYTHONHASHSEED must be fixed too: loky workers otherwise get a randomized hash seed,
        # so hash-ordered constructs in the fit path (e.g. list(set(ind_vars))) order features
        # differently per worker/run, and XGBoost column subsampling then picks different features
        # -> non-reproducible models. Pinning it makes every worker (and re-run) deterministic.
        _thread_env = {
            "OMP_NUM_THREADS": "1", "OPENBLAS_NUM_THREADS": "1", "MKL_NUM_THREADS": "1",
            "NUMEXPR_NUM_THREADS": "1", "VECLIB_MAXIMUM_THREADS": "1", "PYTHONHASHSEED": "0",
        }
        _saved_env = {k: os.environ.get(k) for k in _thread_env}
        for k, v in _thread_env.items():
            os.environ[k] = v
        try:
            from joblib import Parallel, delayed
            fold_results = Parallel(n_jobs=cv_max_workers, backend="loky")(
                delayed(_run_cv_fold)(k, hk, tk, 1, *fold_args) for (k, hk, tk) in tasks
            )
        finally:
            for k, v in _saved_env.items():
                if v is None:
                    os.environ.pop(k, None)
                else:
                    os.environ[k] = v
    else:
        # cv_max_workers=1 is the explicit "no fold parallelism" escape hatch (e.g. low memory);
        # let each fold keep full trial-level parallelism (tune_workers=None) so it isn't crippled.
        fold_results = [_run_cv_fold(k, hk, tk, None, *fold_args) for (k, hk, tk) in tasks]

    oof_map: dict[str, float] = {}
    n_fold_ok = 0
    fold_param_files: list[tuple[float, str]] = []  # (holdout utility, params.json path)
    for res in fold_results:
        if res is None:
            warnings.warn(
                f"A CV fold for {model_group}/{model_name} produced no result; its holdout "
                f"sales fall back to the study prediction."
            )
            continue
        oof_pairs, util, fpath = res
        for ks, pred in oof_pairs:
            oof_map[ks] = pred
        n_fold_ok += 1
        if fpath:
            fold_param_files.append((util, fpath))

    # ---- Production hyperparameters (skip the Phase-2 tune when possible) ----
    # For tunable engines, reuse the fold HPs instead of a fresh production study:
    #   aggregate (default) = median/mode across folds; best_fold = the best-holdout fold's HPs.
    # We write them to the prod params.json with the CV-aggregate sentinel so _get_params trusts
    # them regardless of fingerprint. refit (or non-tunable / no fold params) tunes as normal.
    prod_outpath = f"{outpath}/cv_prod"
    prod_use_saved = use_saved_params
    if fold_param_files and cv_prod_mode in ("aggregate", "best_fold"):
        dicts = []
        for _util, fpath in fold_param_files:
            d = json.load(open(fpath))
            d.pop("__fingerprint", None)
            dicts.append(d)
        if cv_prod_mode == "best_fold":
            best_i = min(
                range(len(fold_param_files)),
                key=lambda i: fold_param_files[i][0] if fold_param_files[i][0] == fold_param_files[i][0] else float("inf"),
            )
            chosen = dicts[best_i]
        else:
            chosen = _aggregate_fold_params(dicts)
        os.makedirs(prod_outpath, exist_ok=True)
        with open(f"{prod_outpath}/{model_name}_params.json", "w") as fh:
            json.dump({**chosen, "__fingerprint": _CV_AGGREGATE_FINGERPRINT}, fh)
        prod_use_saved = True
        if verbose:
            print(f"--> CV production HPs for {model_group}/{model_name}: {cv_prod_mode}")

    # ---- Phase 2: refit on all trainable; study + universe + post-val holdout ----
    # test_keys = post-val (leakage-free full-model holdout). If there are no post-val sales,
    # any small non-empty slice satisfies DataSplit.split(); its stats are overwritten below.
    prod_test_keys = post_val if len(post_val) > 0 else train_all[: max(1, len(train_all) // 5)]
    prod_train_keys = [k for k in train_all if k not in set(prod_test_keys)] if len(post_val) == 0 else train_all
    phase2 = run_one_model(
        df_sales, df_universe, vacant_only, model_group, model_name, model_entries,
        settings, dep_var, dep_var_test, best_variables, fields_cat, prod_outpath,
        save_params, prod_use_saved, save_results=False, verbose=verbose,
        test_keys=prod_test_keys, train_keys=prod_train_keys,
    )
    if phase2 is None:
        return None

    main_vacant = "vacant" if vacant_only else "main"
    location = get_model_location(settings, main_vacant, model_name, model_group)

    # ---- Post-valuation slice, explained while the frames still line up ----
    # Right now `phase2.df_test` IS the post-valuation holdout (it was built from
    # prod_test_keys) and ds.X_test matches it row for row. Phase 2 never trained on post-val,
    # so its explanation of those rows is exactly as leakage-free as each fold's is of its own
    # holdout -- they belong in the same OOF file. This is the only moment we can isolate them:
    # `override_test_predictions` below swaps df_test for the full-coverage frame.
    wrote_post_val_contribs = False
    if save_results and emit_oof_contribs and n_fold_ok > 0 and len(post_val) > 0:
        try:
            pv_dir = _model_artifact_dir(prod_outpath, model_name)
            os.makedirs(pv_dir, exist_ok=True)
            write_model_parameters(
                phase2.model, phase2, location, pv_dir, verbose=verbose, subsets={"test"},
            )
            wrote_post_val_contribs = True
        except Exception as e:
            warnings.warn(
                f"{model_group}/{model_name}: could not write post-valuation contributions "
                f"({type(e).__name__}: {e}); those rows will be missing from "
                f"contributions_test.csv."
            )

    # ---- Stitch: replace the test side with the full-coverage OOF frame ----
    if n_fold_ok == 0:
        warnings.warn(
            f"No CV folds succeeded for {model_group}/{model_name}; holdout stats fall back to "
            f"the single Phase-2 holdout."
        )
    else:
        field = phase2.field_prediction
        raw_sales = phase2.ds.df_sales.copy()
        ks = raw_sales["key_sale"].astype(str)
        post_val_map = dict(
            zip(phase2.df_test["key_sale"].astype(str).values, phase2.df_test[field].values)
        )
        pred = ks.map(oof_map)
        pred = pred.where(pred.notna(), ks.map(post_val_map))
        raw_sales[field] = pred.values
        df_test_full = raw_sales[raw_sales[field].notna()].reset_index(drop=True)
        phase2.override_test_predictions(df_test_full)

    if save_results:
        # The Phase-2 model must NOT explain the OOF holdout frame: its attributions would
        # describe a different model than the predictions sitting in that frame. Hold back the
        # "test" subset and let the stitcher assemble it from the per-fold writes instead.
        write_subsets = _cv_phase2_write_subsets(emit_oof_contribs, n_fold_ok)
        _write_model_results(
            phase2, outpath, settings, location, verbose=verbose, subsets=write_subsets
        )
        if write_subsets is not None:
            _stitch_oof_contributions(
                outpath,
                model_name,
                [k for (k, _hk, _tk) in tasks],
                prod_outpath,
                include_post_val=wrote_post_val_contribs,
                verbose=verbose,
            )

    return phase2

try_variables

try_variables(sup, settings, verbose=False, plot=False, do_report=False)

Experiment with variables to determine which are most useful for modeling.

Parameters:

Name	Type	Description	Default
sup	SalesUniversePair	The SalesUniversePair containing sales and universe data.	required
settings	dict	Settings dictionary	required
verbose	bool	Whether to print verbose output. Default is False.	False
plot	bool	Whether to generate plots. Default is False.	False
do_report	bool	Whether to generate a pdf report. Default is False.	False

Source code in openavmkit/model_runner.py

def try_variables(
    sup: SalesUniversePair,
    settings: dict,
    verbose: bool = False,
    plot: bool = False,
    do_report: bool = False,
):
    """Experiment with variables to determine which are most useful for modeling.

    Parameters
    ----------
    sup: SalesUniversePair
        The SalesUniversePair containing sales and universe data.
    settings: dict
        Settings dictionary
    verbose: bool
        Whether to print verbose output. Default is False.
    plot: bool
        Whether to generate plots. Default is False.
    do_report: bool
        Whether to generate a pdf report. Default is False.

    """

    df_hydrated = get_hydrated_sales_from_sup(sup)

    idx_vacant = df_hydrated["vacant_sale"].eq(True)

    df_vacant = df_hydrated[idx_vacant].copy()

    df_vacant = _simulate_removed_buildings(df_vacant, settings, idx_vacant)

    # update df_hydrated with *all* the characteristics of df_vacant where their keys match:
    df_hydrated.loc[idx_vacant, df_vacant.columns] = df_vacant.values

    all_best_variables = {}
    all_reports = {}

    try_vars = settings.get("modeling", {}).get("try_variables", {})
    model_groups_to_skip = try_vars.get("skip", [])

    def _try_variables(
        df_in: pd.DataFrame,
        model_group: str,
        df_univ: pd.DataFrame,
        do_report: bool,
        settings: dict,
        verbose: bool,
        results: dict,
        reports: dict
    ):
        bests = {}
        local_reports = {}

        for vacant_only in [False, True]:

            if vacant_only:
                if df_in["vacant_sale"].sum() == 0:
                    if verbose:
                        print("No vacant sales found, skipping...")
                    continue
            else:
                if df_in["valid_sale"].sum() == 0:
                    if verbose:
                        print("No valid sales found, skipping...")
                    continue

            try_vars = settings.get("modeling", {}).get("try_variables", {})
            variables_to_use = (
                try_vars.get("variables", [])
            )

            if len(variables_to_use) == 0:
                raise ValueError(
                    "No variables defined. Please check settings `modeling.try_variables.variables`"
                )

            df_univ = df_univ[df_univ["model_group"].eq(model_group)].copy()

            try:
                var_recs = get_variable_recommendations(
                    df_in,
                    df_univ,
                    vacant_only,
                    settings,
                    model_group,
                    variables_to_use=variables_to_use,
                    tests_to_run=["corr", "r2"],
                    do_report=True,
                    do_cross=True,
                    do_plots=plot,
                    verbose=verbose,
                )
            except Exception as e:
                # A model group can be too small or too degenerate (e.g. all-constant /
                # all-NaN features, too few sales) for the correlation/R2 recommendation
                # step to produce a result -- it would otherwise crash the whole
                # try_variables run. Warn and skip this group instead.
                warnings.warn(
                    f"try_variables: skipping model group '{model_group}' "
                    f"({'vacant' if vacant_only else 'main'}) -- could not compute variable "
                    f"recommendations ({type(e).__name__}: {e}). This usually means the group "
                    f"has too few or too-degenerate sales for the requested variables."
                )
                continue

            best_variables = var_recs["variables"]
            df_results = var_recs["df_results"]
            report = var_recs["report"]

            if vacant_only:
                bests["vacant_only"] = df_results
                local_reports["vacant_only"] = report
            else:
                bests["main"] = df_results
                local_reports["main"] = report

        results[model_group] = bests
        reports[model_group] = local_reports

    do_per_model_group(
        df_hydrated,
        settings,
        _try_variables,
        params={
            "settings": settings,
            "df_univ": sup.universe,
            "do_report": do_report,
            "verbose": verbose,
            "results": all_best_variables,
            "reports": all_reports
        },
        key="key_sale",
        skip=model_groups_to_skip
    )

    sale_field = get_sale_field(settings)

    print("")
    print("********** BEST VARIABLES ***********")
    for model_group in all_best_variables:
        entry = all_best_variables[model_group]
        report_entry = all_reports[model_group]
        for vacant_status in entry:
            print("")
            print(f"model group: {model_group} / {vacant_status}")
            results = entry[vacant_status]
            report = report_entry[vacant_status]
            results = results[~results["corr_strength"].isna()]

            styled = results.style.format(
                {
                    "corr_strength": "{:,.2f}",
                    "corr_clarity": "{:,.2f}",
                    "corr_score": "{:,.2f}",
                    "r2": "{:,.2f}",
                    "adj_r2": "{:,.2f}",
                    "coef_sign": "{:,.0f}"
                }
            )

            pd.set_option("display.max_rows", None)
            display(styled)
            pd.set_option("display.max_rows", 15)

            file_out = f"out/try/{model_group}/{vacant_status}.csv"
            report_out = f"out/try/{model_group}/{vacant_status}_report"
            if not os.path.exists(os.path.dirname(file_out)):
                os.makedirs(os.path.dirname(file_out))
            results.to_csv(file_out, index=False)

            if do_report:
                finish_report(report, report_out, "variable", settings)

write_out_all_results

write_out_all_results(sup, all_results, settings)

Write out all model results to CSV and Parquet files.

This function collects predictions from all model groups and writes them to a single DataFrame, which is then saved to both CSV and Parquet formats. It also merges the predictions with the universe DataFrame to include all keys.

It additionally writes the openratiostudy.com export (open_ratio_study_sales.csv and open_ratio_study_test.csv) — the slim per-sale frames that website consumes — by concatenating the ensemble (production) results across all model groups.

Parameters:

Name	Type	Description	Default
sup	SalesUniversePair	The SalesUniversePair containing sales and universe data.	required
all_results	dict	A dictionary where keys are model group identifiers and values are MultiModelResults containing the results for each model group.	required
settings	dict	The settings dictionary, used to resolve the report-location fields for the open-ratio-study export.	required

Source code in openavmkit/model_runner.py

def write_out_all_results(sup: SalesUniversePair, all_results: dict, settings: dict):
    """Write out all model results to CSV and Parquet files.

    This function collects predictions from all model groups and writes them to a single
    DataFrame, which is then saved to both CSV and Parquet formats. It also merges the
    predictions with the universe DataFrame to include all keys.

    It additionally writes the openratiostudy.com export (``open_ratio_study_sales.csv``
    and ``open_ratio_study_test.csv``) — the slim per-sale frames that website consumes —
    by concatenating the ensemble (production) results across all model groups.

    Parameters
    ----------
    sup : SalesUniversePair
        The SalesUniversePair containing sales and universe data.
    all_results : dict
        A dictionary where keys are model group identifiers and values are MultiModelResults
        containing the results for each model group.
    settings : dict
        The settings dictionary, used to resolve the report-location fields for the
        open-ratio-study export.
    """
    t = TimingData()
    df_all = None
    ors_frames = {"sales": [], "test": []}

    for model_group in all_results:
        t.start(f"model group: {model_group}")
        t.start("read")
        mm_results: MultiModelResults = all_results[model_group]

        # Skip if no results for this model group
        if mm_results is None:
            t.stop("read")
            t.stop(f"model group: {model_group}")
            continue

        # Collect all ensemble types to output
        output_models = []
        if "ensemble" in mm_results.model_results:
            output_models.append("ensemble")
        if not output_models:
            t.stop("read")
            t.stop(f"model group: {model_group}")
            continue

        # Accumulate the openratiostudy.com export from the ensemble (production) model.
        ors = _assemble_open_ratio_study(mm_results.model_results["ensemble"], settings)
        for subset, df_ors in ors.items():
            df_ors["model_group"] = model_group
            ors_frames[subset].append(df_ors)

        # For each output model, extract predictions and add to df_univ_local
        df_univ_local = None
        for model_type in output_models:
            smr = mm_results.model_results[model_type]
            col_name = (
                f"market_value_{model_type}"
                if "ensemble" not in model_type
                else "market_value"
            )
            df_pred = smr.df_universe[["key", smr.field_prediction]].rename(
                columns={smr.field_prediction: col_name}
            )
            if df_univ_local is None:
                df_univ_local = df_pred
            else:
                df_univ_local = df_univ_local.merge(df_pred, on="key", how="outer")
        df_univ_local["model_group"] = model_group

        if df_all is None:
            df_all = df_univ_local
        else:
            t.start("concat")
            df_all = pd.concat([df_all, df_univ_local])
            t.stop("concat")

        t.stop(f"model group: {model_group}")

    # Only proceed with writing if we have results
    if df_all is not None:
        t.start("copy")
        df_univ = sup.universe.copy()
        t.stop("copy")
        t.start("merge")
        df_univ = df_univ.merge(df_all, on="key", how="left")
        t.stop("merge")

        outpath = "out/models/all_model_groups"
        if not os.path.exists(outpath):
            os.makedirs(outpath)

        t.start("csv")
        df_univ.to_csv(f"{outpath}/universe.csv", index=False)
        t.stop("csv")
        t.start("parquet")
        df_univ.to_parquet(f"{outpath}/universe.parquet", engine="pyarrow")
        t.stop("parquet")

    # Write the openratiostudy.com export (study + test subsets, all model groups).
    if ors_frames["sales"] or ors_frames["test"]:
        outpath = "out/models/all_model_groups"
        if not os.path.exists(outpath):
            os.makedirs(outpath)
        t.start("open_ratio_study")
        for subset, fname in (("sales", "open_ratio_study_sales.csv"),
                              ("test", "open_ratio_study_test.csv")):
            frames = ors_frames[subset]
            if not frames:
                continue
            pd.concat(frames, ignore_index=True).to_csv(f"{outpath}/{fname}", index=False)
        t.stop("open_ratio_study")