openavmkit.inference

This module contains various inference models for predicting missing values in spatial datasets (e.g. you are missing some amount of building square footage). It uses proxy variables supplied by the user (e.g. building footprint size) and works out geospatial correlations to predict the missing values.

perform_spatial_inference() is the main function that orchestrates the inference process based on user settings.

CategoricalEncoder

CategoricalEncoder()

Universal categorical encoder that handles unseen categories.

Source code in openavmkit/inference.py

def __init__(self):
    self.label_encoder = LabelEncoder()
    self.unknown_value = None  # Will be set during fit

fit

fit(series)

Fit encoder adding a special unknown value.

Parameters:

Name	Type	Description	Default
series	Series	data series to fit	required

Source code in openavmkit/inference.py

def fit(self, series: pd.Series) -> None:
    """Fit encoder adding a special unknown value.

    Parameters
    ----------
    series : pd.Series
        data series to fit
    """

    # Get unique non-null values
    unique_values = series.dropna().unique()

    # Add an extra category for unknown
    self.label_encoder.fit(np.append(unique_values, ["__UNKNOWN__"]))
    self.unknown_value = self.label_encoder.transform(["__UNKNOWN__"])[0]

fit_transform

fit_transform(series)

Fit and transform in one step.

Parameters:

Name	Type	Description	Default
series	Series	data series to fit & transform	required

Returns:

Type	Description
ndarray	the transformed data series

Source code in openavmkit/inference.py

def fit_transform(self, series: pd.Series) -> np.ndarray:
    """Fit and transform in one step.

    Parameters
    ----------
    series : pd.Series
        data series to fit & transform

    Returns
    -------
    np.ndarray
        the transformed data series

    """
    self.fit(series)
    return self.transform(series)

transform

transform(series)

Transform values, mapping unseen categories to unknown.

Parameters:

Name	Type	Description	Default
series	Series	data series to transform	required

Returns:

Type	Description
ndarray	the transformed data sereies

Source code in openavmkit/inference.py

def transform(self, series: pd.Series) -> np.ndarray:
    """Transform values, mapping unseen categories to unknown.

    Parameters
    ----------
    series : pd.Series
        data series to transform

    Returns
    -------
    np.ndarray
        the transformed data sereies
    """
    # Handle nulls first
    series = series.fillna("__UNKNOWN__")

    # Create output array
    result = np.full(len(series), self.unknown_value)

    # Get mask of known categories
    known_mask = series.isin(self.label_encoder.classes_)

    # Transform known categories
    if known_mask.any():
        result[known_mask] = self.label_encoder.transform(series[known_mask])

    return result

EnsembleModel

EnsembleModel()

Bases: InferenceModel

Ensemble model combining LightGBM, XGBoost, and Random Forest.

Source code in openavmkit/inference.py

def __init__(self):
    self.lgb_model = LightGBMModel()
    self.xgb_model = XGBoostModel()
    self.rf_model = RandomForestModel()
    self.weights = None
    self.encoders = {}
    self.proxy_fields = None
    self.location_fields = None
    self.interaction_fields = None
    self.imputer = SimpleImputer(strategy="median")
    self.feature_order = None

evaluate

evaluate(df, target)

Evaluate model performance.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing features and true target values.	required
target	str	Name of the target variable column in df.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of evaluation metrics (e.g., R², RMSE).

Source code in openavmkit/inference.py

def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """
    Evaluate model performance.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing features and true target values.
    target : str
        Name of the target variable column in `df`.

    Returns
    -------
    Dict[str, float]
        Dictionary of evaluation metrics (e.g., R², RMSE).
    """
    predictions = self.predict(df)
    actuals = df[target]

    metrics = {
        "mae": np.abs(predictions - actuals).mean(),
        "mape": np.abs((predictions - actuals) / actuals).mean() * 100,
        "rmse": np.sqrt(((predictions - actuals) ** 2).mean()),
        "r2": 1
        - ((predictions - actuals) ** 2).sum()
        / ((actuals - actuals.mean()) ** 2).sum(),
    }

    return metrics

fit

fit(df, target, settings)

Fit ensemble model and determine optimal weights.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Source code in openavmkit/inference.py

def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """Fit ensemble model and determine optimal weights.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.
    """
    proxies = settings.get("proxies", [])
    locations = [
        loc for loc in settings.get("locations", []) if loc != "___everything___"
    ]
    interactions = settings.get("interactions", [])

    # Format interaction fields
    self.interaction_fields = []
    for interaction in interactions:
        if isinstance(interaction, list):
            self.interaction_fields.append("_x_".join(interaction))
        else:
            self.interaction_fields.append(interaction)

    self.proxy_fields = proxies
    self.location_fields = locations

    # Split data for weight optimization
    df_train, df_val = train_test_split(df, test_size=0.2, random_state=42)

    # Fit individual models
    print("\nFitting LightGBM model...")
    self.lgb_model.fit(df_train, target, settings)
    print("\nFitting XGBoost model...")
    self.xgb_model.fit(df_train, target, settings)
    print("\nFitting Random Forest model...")
    self.rf_model.fit(df_train, target, settings)

    # Get predictions on validation set
    lgb_preds = self.lgb_model.predict(df_val)
    xgb_preds = self.xgb_model.predict(df_val)
    rf_preds = self.rf_model.predict(df_val)
    actuals = df_val[target].values.astype(
        float
    )  # Convert to numpy array and ensure float type

    # Optimize weights to minimize RMSE
    def objective(weights):
        ensemble_preds = (
            weights[0] * lgb_preds + weights[1] * xgb_preds + weights[2] * rf_preds
        )
        return np.sqrt(((ensemble_preds - actuals) ** 2).mean())

    initial_weights = np.array([1 / 3, 1 / 3, 1 / 3])
    bounds = [(0, 1), (0, 1), (0, 1)]
    constraints = {"type": "eq", "fun": lambda w: np.sum(w) - 1}

    result = minimize(
        objective, initial_weights, bounds=bounds, constraints=constraints
    )
    self.weights = result.x

    print("\nEnsemble weights:")
    print(f"--> LightGBM: {self.weights[0]:.4f}")
    print(f"--> XGBoost: {self.weights[1]:.4f}")
    print(f"--> Random Forest: {self.weights[2]:.4f}")

    # Fit final models on full data
    self.lgb_model.fit(df, target, settings)
    self.xgb_model.fit(df, target, settings)
    self.rf_model.fit(df, target, settings)

predict

predict(df)

Make predictions on new data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Data to perform predictions on.	required

Returns:

Type	Description
Series	Predicted values of the target variable chosen during fit()

Source code in openavmkit/inference.py

def predict(self, df: pd.DataFrame) -> pd.Series:
    """Make predictions on new data.

    Parameters
    ----------
    df : pd.DataFrame
        Data to perform predictions on.

    Returns
    -------
    pd.Series
        Predicted values of the target variable chosen during fit()
    """
    lgb_preds = self.lgb_model.predict(df)
    xgb_preds = self.xgb_model.predict(df)
    rf_preds = self.rf_model.predict(df)

    return pd.Series(
        self.weights[0] * lgb_preds
        + self.weights[1] * xgb_preds
        + self.weights[2] * rf_preds,
        index=df.index,
    )

InferenceModel

Bases: ABC

Base class for inference models.

evaluate abstractmethod

evaluate(df, target)

Evaluate model performance on training data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required

Returns:

Type	Description
Dict[str, float]	Dictionary containing the following metrics: "mae", "mape", "rmse", "r2"

Source code in openavmkit/inference.py

@abstractmethod
def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """Evaluate model performance on training data.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.

    Returns
    -------
    Dict[str, float]
        Dictionary containing the following metrics: "mae", "mape", "rmse", "r2"
    """
    pass

fit abstractmethod

fit(df, target, settings)

Fit the model using training data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Returns:

Type	Description
None

Source code in openavmkit/inference.py

@abstractmethod
def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """
    Fit the model using training data.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.

    Returns
    -------
    None
    """
    pass

predict abstractmethod

predict(df)

Make predictions on new data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Data to perform predictions on.	required

Returns:

Type	Description
Series	Predicted values of the target variable chosen during fit()

Source code in openavmkit/inference.py

@abstractmethod
def predict(self, df: pd.DataFrame) -> pd.Series:
    """Make predictions on new data.

    Parameters
    ----------
    df : pd.DataFrame
        Data to perform predictions on.

    Returns
    -------
    pd.Series
        Predicted values of the target variable chosen during fit()
    """
    pass

LightGBMModel

LightGBMModel()

Bases: InferenceModel

LightGBM model with improved validation and parameters.

Source code in openavmkit/inference.py

def __init__(self):
    self.model = lgb.LGBMRegressor(
        n_estimators=200,
        max_depth=-1,
        num_leaves=31,
        learning_rate=0.05,
        subsample=0.8,
        colsample_bytree=0.8,
        reg_alpha=0.1,
        reg_lambda=0.1,
        random_state=42,
        n_jobs=-1,
    )
    self.encoders = {}
    self.proxy_fields = None
    self.location_fields = None
    self.interaction_fields = None
    self.imputer = SimpleImputer(strategy="median")
    self.feature_order = None  # Store the order of features from training

evaluate

evaluate(df, target)

Evaluate model performance.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing features and true target values.	required
target	str	Name of the target variable column in df.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of evaluation metrics (e.g., R², RMSE).

Source code in openavmkit/inference.py

def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """
    Evaluate model performance.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing features and true target values.
    target : str
        Name of the target variable column in `df`.

    Returns
    -------
    Dict[str, float]
        Dictionary of evaluation metrics (e.g., R², RMSE).
    """
    predictions = self.predict(df)
    actuals = df[target]

    metrics = {
        "mae": np.abs(predictions - actuals).mean(),
        "mape": np.abs((predictions - actuals) / actuals).mean() * 100,
        "rmse": np.sqrt(((predictions - actuals) ** 2).mean()),
        "r2": 1
        - ((predictions - actuals) ** 2).sum()
        / ((actuals - actuals.mean()) ** 2).sum(),
    }

    return metrics

fit

fit(df, target, settings)

Fit the model using training data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Source code in openavmkit/inference.py

def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """
    Fit the model using training data.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.
    """

    proxies = settings.get("proxies", [])
    locations = [
        loc for loc in settings.get("locations", []) if loc != "___everything___"
    ]
    interactions = settings.get("interactions", [])

    # Format interaction fields
    self.interaction_fields = []
    for interaction in interactions:
        if isinstance(interaction, list):
            self.interaction_fields.append("_x_".join(interaction))
        else:
            self.interaction_fields.append(interaction)

    self.proxy_fields = proxies
    self.location_fields = locations

    # Create feature matrix
    X = self._create_feature_matrix(df, fit=True)
    y = df[target].values.astype(
        float
    )  # Convert to numpy array and ensure float type

    print("\nFitting LightGBM model:")
    print(f"Features being used: {list(X.columns)}")
    if self.interaction_fields:
        print(f"Interaction features: {self.interaction_fields}")

    # Fit model
    self.model.fit(X, y)

    # Print feature importances
    importances = pd.Series(
        self.model.feature_importances_, index=X.columns
    ).sort_values(ascending=False)

    print("\nFeature importances:")
    for feat, imp in importances.items():
        print(f"--> {feat}: {imp:.4f}")

    if self.interaction_fields:
        print("\nInteraction feature importances:")
        interaction_importances = importances[
            importances.index.isin(self.interaction_fields)
        ]
        for feat, imp in interaction_importances.items():
            print(f"--> {feat}: {imp:.4f}")

predict

predict(df)

Make predictions on new data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Data to perform predictions on.	required

Returns:

Type	Description
Series	Predicted values of the target variable chosen during fit()

Source code in openavmkit/inference.py

def predict(self, df: pd.DataFrame) -> pd.Series:
    """Make predictions on new data.

    Parameters
    ----------
    df : pd.DataFrame
        Data to perform predictions on.

    Returns
    -------
    pd.Series
        Predicted values of the target variable chosen during fit()
    """
    # Create feature matrix using same process as fit
    X = self._create_feature_matrix(df, fit=False)

    # Verify we have all features
    missing_features = set(self.feature_order) - set(X.columns)
    if missing_features:
        raise ValueError(f"Missing features during prediction: {missing_features}")

    return pd.Series(self.model.predict(X), index=df.index)

RandomForestModel

RandomForestModel()

Bases: InferenceModel

Random Forest with improved validation and parameters.

Source code in openavmkit/inference.py

def __init__(self):
    self.model = RandomForestRegressor(
        n_estimators=200,
        max_depth=None,
        min_samples_split=5,
        min_samples_leaf=2,
        max_features="sqrt",
        random_state=42,
        n_jobs=-1,
    )
    self.encoders = {}
    self.proxy_fields = None
    self.location_fields = None
    self.interaction_fields = None
    self.imputer = SimpleImputer(strategy="median")
    self.feature_order = None  # Store the order of features from training

evaluate

evaluate(df, target)

Evaluate model performance.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing features and true target values.	required
target	str	Name of the target variable.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of evaluation metrics (e.g., R², RMSE).

Source code in openavmkit/inference.py

def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """
    Evaluate model performance.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing features and true target values.
    target : str
        Name of the target variable.

    Returns
    -------
    Dict[str, float]
        Dictionary of evaluation metrics (e.g., R², RMSE).
    """
    predictions = self.predict(df)
    actuals = df[target]

    metrics = {
        "mae": np.abs(predictions - actuals).mean(),
        "mape": np.abs((predictions - actuals) / actuals).mean() * 100,
        "rmse": np.sqrt(((predictions - actuals) ** 2).mean()),
        "r2": 1
        - ((predictions - actuals) ** 2).sum()
        / ((actuals - actuals.mean()) ** 2).sum(),
    }

    return metrics

fit

fit(df, target, settings)

Fit the model using training data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Source code in openavmkit/inference.py

def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """
    Fit the model using training data.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.
    """

    proxies = settings.get("proxies", [])
    locations = [
        loc for loc in settings.get("locations", []) if loc != "___everything___"
    ]
    interactions = settings.get("interactions", [])

    # Format interaction fields
    self.interaction_fields = []
    for interaction in interactions:
        if isinstance(interaction, list):
            self.interaction_fields.append("_x_".join(interaction))
        else:
            self.interaction_fields.append(interaction)

    self.proxy_fields = proxies
    self.location_fields = locations

    # Create feature matrix
    X = self._create_feature_matrix(df, fit=True)
    y = df[target].values.astype(
        float
    )  # Convert to numpy array and ensure float type

    print("\nFitting Random Forest model:")
    print(f"Features being used: {list(X.columns)}")
    if self.interaction_fields:
        print(f"Interaction features: {self.interaction_fields}")

    # Fit model
    self.model.fit(X, y)

    # Print feature importances
    importances = pd.Series(
        self.model.feature_importances_, index=X.columns
    ).sort_values(ascending=False)

    print("\nFeature importances:")
    for feat, imp in importances.items():
        print(f"--> {feat}: {imp:.4f}")

    if self.interaction_fields:
        print("\nInteraction feature importances:")
        interaction_importances = importances[
            importances.index.isin(self.interaction_fields)
        ]
        for feat, imp in interaction_importances.items():
            print(f"--> {feat}: {imp:.4f}")

predict

predict(df)

Make predictions using best model.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame for making predictions.	required

Returns:

Type	Description
Series	Predicted values from the best model.

Source code in openavmkit/inference.py

def predict(self, df: pd.DataFrame) -> pd.Series:
    """
    Make predictions using best model.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame for making predictions.

    Returns
    -------
    pandas.Series
        Predicted values from the best model.
    """
    # Create feature matrix using same process as fit
    X = self._create_feature_matrix(df, fit=False)

    # Verify we have all features
    missing_features = set(self.feature_order) - set(X.columns)
    if missing_features:
        raise ValueError(f"Missing features during prediction: {missing_features}")

    return pd.Series(self.model.predict(X), index=df.index)

RatioProxyModel

RatioProxyModel()

Bases: InferenceModel

Ratio-based proxy model with proper validation handling.

Source code in openavmkit/inference.py

def __init__(self):
    self.proxy_ratios = {}
    self.proxy_stats = {}

evaluate

evaluate(df, target)

Evaluate model performance.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing features and true target values.	required
target	str	Name of the target variable column in df.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of evaluation metrics (e.g., R², RMSE).

Source code in openavmkit/inference.py

def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """
    Evaluate model performance.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing features and true target values.
    target : str
        Name of the target variable column in `df`.

    Returns
    -------
    Dict[str, float]
        Dictionary of evaluation metrics (e.g., R², RMSE).
    """
    valid_mask = df[target].notna()
    predictions = self.predict(df[valid_mask])

    actuals = df.loc[valid_mask, target]

    # Calculate metrics
    metrics = {
        "mae": np.abs(predictions - actuals).mean(),
        "mape": np.abs((predictions - actuals) / actuals).mean() * 100,
        "rmse": np.sqrt(((predictions - actuals) ** 2).mean()),
        "r2": 1
        - ((predictions - actuals) ** 2).sum()
        / ((actuals - actuals.mean()) ** 2).sum(),
    }

    return metrics

fit

fit(df, target, settings)

Fit the model using training data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Source code in openavmkit/inference.py

def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """
    Fit the model using training data.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.
    """
    proxies = settings.get("proxies", [])
    locations = settings.get("locations", [])
    group_by = settings.get("group_by", [])

    # Add global grouping
    locations.append("___everything___")
    df["___everything___"] = "1"

    self.proxy_ratios = {}
    self.proxy_stats = {}

    # Calculate ratios for each proxy
    for proxy in proxies:
        # Handle missing values
        valid_mask = (
            df[target].notna()
            & df[proxy].notna()
            & df[proxy].gt(0)
            & df[target].gt(0)
        )

        if valid_mask.sum() == 0:
            warnings.warn(f"No valid data for proxy {proxy}")
            continue

        # Calculate ratios
        df_valid = df[valid_mask].copy()
        df_valid[f"ratio_{proxy}"] = df_valid[target] / df_valid[proxy]

        # Remove outliers
        q1, q99 = df_valid[f"ratio_{proxy}"].quantile([0.01, 0.99])
        valid_range = (df_valid[f"ratio_{proxy}"] >= q1) & (
            df_valid[f"ratio_{proxy}"] <= q99
        )
        df_valid = df_valid[valid_range]

        # Store global ratio
        global_ratio = df_valid[f"ratio_{proxy}"].median()
        self.proxy_ratios[(proxy, ())] = global_ratio

        # Calculate ratios for each location/group combination
        for location in locations:
            if location == "___everything___":
                continue

            group_list = group_by.copy() if group_by else []
            group_list.append(location)

            try:
                grouped = df_valid.groupby(group_list)
                median_ratios = grouped[f"ratio_{proxy}"].median()
                if not median_ratios.empty:
                    self.proxy_ratios[(proxy, tuple(group_list))] = median_ratios
            except Exception as e:
                warnings.warn(
                    f"Failed to calculate grouped ratios for {proxy} with groups {group_list}: {str(e)}"
                )

predict

predict(df)

Make predictions using fitted ratios.

Parameters:

Name	Type	Description	Default
df	DataFrame	Input DataFrame for making predictions.	required

Returns:

Type	Description
Series	Predicted values.

Source code in openavmkit/inference.py

def predict(self, df: pd.DataFrame) -> pd.Series:
    """
    Make predictions using fitted ratios.

    Parameters
    ----------
    df : pandas.DataFrame
        Input DataFrame for making predictions.

    Returns
    -------
    pandas.Series
        Predicted values.
    """
    predictions = pd.Series(index=df.index, dtype="float64")

    for proxy, group_list in sorted(
        self.proxy_ratios.keys(), key=lambda x: len(x[1]), reverse=True
    ):
        if len(group_list) > 0:
            try:
                # Get group-specific ratios
                group_key = df[list(group_list)].astype(str).agg("_".join, axis=1)
                ratios = self.proxy_ratios[(proxy, group_list)]

                # Only apply ratios for existing group combinations
                common_keys = group_key[group_key.isin(ratios.index)]
                if not common_keys.empty:
                    # Create initial mask
                    mask = (
                        predictions.isna()
                        & df[proxy].notna()
                        & df[proxy].gt(0)
                        & group_key.isin(ratios.index)
                    )

                    # Additional validation
                    proxy_values = df.loc[mask, proxy]
                    ratio_values = ratios[group_key[mask]]
                    predicted_values = ratio_values * proxy_values

                    # Create validation mask aligned with original mask
                    valid_predictions = pd.Series(False, index=df.index)
                    valid_predictions.loc[mask] = (predicted_values > 100) & (
                        predicted_values < 100000
                    )

                    # Combine masks
                    final_mask = mask & valid_predictions

                    # Apply predictions
                    predictions.loc[final_mask] = predicted_values[
                        valid_predictions[mask]
                    ]
            except Exception as e:
                warnings.warn(
                    f"Failed to apply grouped ratios for {proxy} with groups {group_list}: {str(e)}"
                )
        else:
            # Apply global ratio to remaining missing values
            ratio = self.proxy_ratios[(proxy, ())]
            mask = predictions.isna() & df[proxy].notna() & df[proxy].gt(0)

            # Additional validation for global ratio
            proxy_values = df.loc[mask, proxy]
            predicted_values = ratio * proxy_values

            # Create validation mask aligned with original mask
            valid_predictions = pd.Series(False, index=df.index)
            valid_predictions.loc[mask] = (predicted_values > 100) & (
                predicted_values < 100000
            )

            # Combine masks
            final_mask = mask & valid_predictions

            # Apply predictions
            predictions.loc[final_mask] = ratio * df.loc[final_mask, proxy]

    return predictions

XGBoostModel

XGBoostModel()

Bases: InferenceModel

XGBoost model with improved validation and parameters.

Source code in openavmkit/inference.py

def __init__(self):
    self.model = xgb.XGBRegressor(
        n_estimators=200,
        max_depth=6,
        learning_rate=0.05,
        subsample=0.8,
        colsample_bytree=0.8,
        reg_alpha=0.1,
        reg_lambda=0.1,
        random_state=42,
        n_jobs=-1,
    )
    self.encoders = {}
    self.proxy_fields = None
    self.location_fields = None
    self.interaction_fields = None
    self.imputer = SimpleImputer(strategy="median")
    self.feature_order = None  # Store the order of features from training

evaluate

evaluate(df, target)

Evaluate model performance.

Parameters:

Name	Type	Description	Default
df	DataFrame	DataFrame containing features and true target values.	required
target	str	Name of the target variable column in df.	required

Returns:

Type	Description
Dict[str, float]	Dictionary of evaluation metrics (e.g., R², RMSE).

Source code in openavmkit/inference.py

def evaluate(self, df: pd.DataFrame, target: str) -> Dict[str, float]:
    """
    Evaluate model performance.

    Parameters
    ----------
    df : pandas.DataFrame
        DataFrame containing features and true target values.
    target : str
        Name of the target variable column in `df`.

    Returns
    -------
    Dict[str, float]
        Dictionary of evaluation metrics (e.g., R², RMSE).
    """
    predictions = self.predict(df)
    actuals = df[target]

    metrics = {
        "mae": np.abs(predictions - actuals).mean(),
        "mape": np.abs((predictions - actuals) / actuals).mean() * 100,
        "rmse": np.sqrt(((predictions - actuals) ** 2).mean()),
        "r2": 1
        - ((predictions - actuals) ** 2).sum()
        / ((actuals - actuals.mean()) ** 2).sum(),
    }

    return metrics

fit

fit(df, target, settings)

Fit model with proper categorical handling and interactions.

Parameters:

Name	Type	Description	Default
df	DataFrame	Training data.	required
target	str	Field name of the target variable.	required
settings	Dict[str, Any]	Settings dictionary.	required

Source code in openavmkit/inference.py

def fit(self, df: pd.DataFrame, target: str, settings: Dict[str, Any]) -> None:
    """
    Fit model with proper categorical handling and interactions.

    Parameters
    ----------
    df : pd.DataFrame
        Training data.
    target : str
        Field name of the target variable.
    settings : Dict[str, Any]
        Settings dictionary.
    """
    proxies = settings.get("proxies", [])
    locations = [
        loc for loc in settings.get("locations", []) if loc != "___everything___"
    ]
    interactions = settings.get("interactions", [])

    # Format interaction fields
    self.interaction_fields = []
    for interaction in interactions:
        if isinstance(interaction, list):
            self.interaction_fields.append("_x_".join(interaction))
        else:
            self.interaction_fields.append(interaction)

    self.proxy_fields = proxies
    self.location_fields = locations

    # Create feature matrix
    X = self._create_feature_matrix(df, fit=True)
    y = df[target].values.astype(
        float
    )  # Convert to numpy array and ensure float type

    print("\nFitting XGBoost model:")
    print(f"Features being used: {list(X.columns)}")
    if self.interaction_fields:
        print(f"Interaction features: {self.interaction_fields}")

    # Fit model
    self.model.fit(X, y)

    # Print feature importances
    importances = pd.Series(
        self.model.feature_importances_, index=X.columns
    ).sort_values(ascending=False)

    print("\nFeature importances:")
    for feat, imp in importances.items():
        print(f"--> {feat}: {imp:.4f}")

    if self.interaction_fields:
        print("\nInteraction feature importances:")
        interaction_importances = importances[
            importances.index.isin(self.interaction_fields)
        ]
        for feat, imp in interaction_importances.items():
            print(f"--> {feat}: {imp:.4f}")

predict

predict(df)

Make predictions on new data.

Parameters:

Name	Type	Description	Default
df	DataFrame	Data to perform predictions on.	required

Returns:

Type	Description
Series	Predicted values of the target variable chosen during fit()

Source code in openavmkit/inference.py

def predict(self, df: pd.DataFrame) -> pd.Series:
    """Make predictions on new data.

    Parameters
    ----------
    df : pd.DataFrame
        Data to perform predictions on.

    Returns
    -------
    pd.Series
        Predicted values of the target variable chosen during fit()
    """
    # Create feature matrix using same process as fit
    X = self._create_feature_matrix(df, fit=False)

    # Verify we have all features
    missing_features = set(self.feature_order) - set(X.columns)
    if missing_features:
        raise ValueError(f"Missing features during prediction: {missing_features}")

    return pd.Series(self.model.predict(X), index=df.index)

perform_spatial_inference

perform_spatial_inference(df, s_infer, key, verbose=False)

Perform spatial inference using specified model(s)

Parameters:

Name	Type	Description	Default
df	DataFrame	Input GeoDataFrame	required
s_infer	dict	Inference settings from config	required
key	str	Key field name	required
verbose	bool	Whether to print progress	False

Returns:

Type	Description
GeoDataFrame	GeoDataFrame with inferred values

Source code in openavmkit/inference.py

def perform_spatial_inference(
    df: gpd.GeoDataFrame, s_infer: dict, key: str, verbose: bool = False
) -> gpd.GeoDataFrame:
    """Perform spatial inference using specified model(s)

    Parameters
    -----------
    df : pd.DataFrame
        Input GeoDataFrame
    s_infer : dict
        Inference settings from config
    key : str
        Key field name
    verbose : bool
        Whether to print progress

    Returns
    -------
    gpd.GeoDataFrame
        GeoDataFrame with inferred values
    """
    # Suppress all numpy warnings for the entire inference process
    with np.errstate(all="ignore"):
        df_out = df.copy()
        for field in s_infer:
            entry = s_infer[field]
            df_out = _do_perform_spatial_inference(
                df_out, entry, field, key, verbose=verbose
            )
        return df_out