Skip to content

GRGWR

This page documents 2 public symbols. Each entry includes its purpose, import path, full API docstring, and the maintained example that exercises it.

Conceptual guide

GRGWR

Geo-Regime Geographically Weighted Regression.

Property Value
Type class
Import from pygwrx.models import GRGWR
Signature GRGWR(n_regimes: 'int' = 3, bandwidth: 'BandwidthLike' = 20, kernel: 'str' = 'bisquare', lambda_boundary: 'float' = 1.0, max_iter: 'int' = 10, tol: 'float' = 0.0001, spatial_constraint_weight: 'float' = 0.5, fit_intercept: 'bool' = True, verbose: 'bool' = False, *, n_neighbors: 'int' = 8, min_regime_size: 'Optional[int]' = None, enforce_connectivity: 'bool' = True, random_state: 'Optional[int]' = 42) -> 'None'
Maintained example examples/models/19_gr_gwr.py

GRGWR

GRGWR(
    n_regimes: int = 3,
    bandwidth: BandwidthLike = 20,
    kernel: str = "bisquare",
    lambda_boundary: float = 1.0,
    max_iter: int = 10,
    tol: float = 0.0001,
    spatial_constraint_weight: float = 0.5,
    fit_intercept: bool = True,
    verbose: bool = False,
    *,
    n_neighbors: int = 8,
    min_regime_size: Optional[int] = None,
    enforce_connectivity: bool = True,
    random_state: Optional[int] = 42
)

Geo-Regime Geographically Weighted Regression.

GR-GWR models a piecewise-smooth coefficient field. An initial full-domain GWR provides local slope features. Spatially constrained agglomerative clustering produces connected initial regimes, and a sequential ICM update refines labels under

.. math::

L(z)=\sum_i(y_i-x_i^T\beta_i^{(z_i)})^2+\lambda B(z),

where :math:B(z) counts undirected neighbouring pairs with different regime labels. Every accepted ICM move preserves the source regime's connectivity and attaches the point to an adjacent target regime. A full refit is accepted only when the reported objective does not increase.

Parameters:

Name Type Description Default
n_regimes int

Requested number of regimes.

3
bandwidth BandwidthLike

Positive fixed distance or one-based adaptive neighbour count.

20
kernel str

"bisquare", "gaussian" or "exponential".

'bisquare'
lambda_boundary float

Non-negative boundary-length penalty.

1.0
max_iter int

Maximum ICM sweeps.

10
tol float

Objective tolerance.

0.0001
spatial_constraint_weight float

:math:\gamma in [0, 1]. Clustering uses sqrt(1-gamma) times standardized slope coefficients and sqrt(gamma) times normalized coordinates, so the endpoints are exactly coefficient-only and coordinate-only.

0.5
fit_intercept bool

Add a local intercept. A legacy leading all-ones column is detected and removed.

True
n_neighbors int

k for the symmetric kNN adjacency graph. A minimum spanning tree is added so the graph is connected.

8
min_regime_size Optional[int]

Minimum members per regime. None uses the number of design parameters plus two.

None
enforce_connectivity bool

Preserve connected regimes during ICM.

True
random_state Optional[int]

Deterministic clustering and ICM order seed.

42
verbose bool

Print fitting progress.

False
Notes

The reported AICc and ENP are conditional on the discovered regime labels. They measure the final piecewise local smoother and do not include the full discrete search complexity of regime discovery.

Source code in src/pygwrx/models/grgwr.py
def __init__(
    self,
    n_regimes: int = 3,
    bandwidth: BandwidthLike = 20,
    kernel: str = "bisquare",
    lambda_boundary: float = 1.0,
    max_iter: int = 10,
    tol: float = 1e-4,
    spatial_constraint_weight: float = 0.5,
    fit_intercept: bool = True,
    verbose: bool = False,
    *,
    n_neighbors: int = 8,
    min_regime_size: Optional[int] = None,
    enforce_connectivity: bool = True,
    random_state: Optional[int] = 42,
) -> None:
    self.n_regimes = self._positive_int(n_regimes, "n_regimes")
    self.bandwidth = self._validate_bandwidth(bandwidth)
    self.kernel = self._choice(kernel, "kernel", self._KERNELS)
    self.lambda_boundary = self._nonnegative_float(
        lambda_boundary, "lambda_boundary"
    )
    self.max_iter = self._nonnegative_int(max_iter, "max_iter")
    self.tol = self._positive_float(tol, "tol")
    self.spatial_constraint_weight = self._unit_float(
        spatial_constraint_weight, "spatial_constraint_weight"
    )
    self.fit_intercept = self._boolean(fit_intercept, "fit_intercept")
    self.verbose = self._boolean(verbose, "verbose")
    self.n_neighbors = self._positive_int(n_neighbors, "n_neighbors")
    self.min_regime_size = (
        None
        if min_regime_size is None
        else self._positive_int(min_regime_size, "min_regime_size")
    )
    self.enforce_connectivity = self._boolean(
        enforce_connectivity, "enforce_connectivity"
    )
    self.random_state = random_state
    self._reset_fit_state()

fit

fit(
    X: ArrayLike, y: VectorLike, coords: ArrayLike
) -> "GRGWR"

Fit GR-GWR and return self.

Source code in src/pygwrx/models/grgwr.py
def fit(self, X: ArrayLike, y: VectorLike, coords: ArrayLike) -> "GRGWR":
    """Fit GR-GWR and return ``self``."""
    self._reset_fit_state()
    try:
        X_raw, names = self._coerce_X_fit(X)
        y_arr = self._numeric_y(y)
        coords_arr = np.asarray(validate_coords(coords), dtype=float)
        if X_raw.shape[0] != y_arr.size or coords_arr.shape[0] != y_arr.size:
            raise ValueError(
                "X, y and coords must contain the same number of rows."
            )
        self.X_ = X_raw.copy()
        self._Xd = add_intercept(X_raw) if self.fit_intercept else X_raw.copy()
        self.y_ = y_arr.copy()
        self.coords_ = coords_arr.copy()
        self.feature_names_ = names
        self.feature_names_in_ = np.asarray(names, dtype=object)
        self.n_features_in_ = X_raw.shape[1]
        p = self._Xd.shape[1]
        self._min_regime_size_ = (
            p + 2 if self.min_regime_size is None else self.min_regime_size
        )
        if isinstance(self.bandwidth, Integral) and int(self.bandwidth) < p + 1:
            raise ValueError(
                "Adaptive bandwidth must be at least the number of design "
                "parameters plus one."
            )
        if y_arr.size < self._min_regime_size_:
            raise ValueError("The sample is smaller than min_regime_size.")

        self._build_graph(coords_arr)
        self.global_coef_ = self._fit_global_gwr()
        self.clustering_features_ = self._clustering_features(
            self.global_coef_, coords_arr
        )
        labels = self._initial_regimes(self.clustering_features_)
        labels = self._relabel_contiguous(labels)

        current_fit = self._fit_for_labels(labels)
        current_objective = self._objective(labels, current_fit)
        self.objective_history_ = [float(current_objective)]
        converged = False
        stop_reason = "max_iter"
        accepted_iterations = 0

        for iteration in range(self.max_iter):
            proposed, changed = self._icm_sweep(labels, iteration)
            if changed == 0:
                converged = True
                stop_reason = "labels_stable"
                break
            proposed_fit = self._fit_for_labels(proposed)
            proposed_objective = self._objective(proposed, proposed_fit)
            if proposed_objective > current_objective + self.tol:
                converged = True
                stop_reason = "objective_guard"
                break
            improvement = current_objective - proposed_objective
            labels = proposed
            current_fit = proposed_fit
            current_objective = proposed_objective
            self.objective_history_.append(float(current_objective))
            accepted_iterations += 1
            if improvement < self.tol:
                converged = True
                stop_reason = "objective_tolerance"
                break

        self.regimes_ = self._relabel_contiguous(labels)
        # Relabelling cannot change the fit when labels are already contiguous,
        # but refit explicitly so every stored block follows final label order.
        final_fit = self._fit_for_labels(self.regimes_)
        self.n_regimes_actual_ = int(np.max(self.regimes_)) + 1
        self.regime_sizes_ = np.bincount(
            self.regimes_, minlength=self.n_regimes_actual_
        )
        self.regime_component_counts_ = np.asarray(
            [
                self._component_count(self.regimes_, regime)
                for regime in range(self.n_regimes_actual_)
            ],
            dtype=int,
        )
        self.local_parameters_ = final_fit.local_parameters
        self.coefficients_ = final_fit.coefficients_by_regime
        self.fitted_values_ = final_fit.fitted_values
        self.hat_matrix_ = final_fit.hat_matrix
        self.residuals_ = self.y_ - self.fitted_values_
        self.regime_boundaries_ = tuple(
            (i, j) for i, j in self.edges_ if self.regimes_[i] != self.regimes_[j]
        )
        self.n_iter_ = accepted_iterations
        self.converged_ = converged
        self.stop_reason_ = stop_reason
        self._finalise_parameters()
        self._compute_diagnostics()
        self._is_fitted = True
        return self
    except Exception:
        self._reset_fit_state()
        raise

predict_result

predict_result(
    X: ArrayLike, coords: ArrayLike
) -> GRGWRPredictionResult

Assign query regimes and recalibrate local WLS coefficients.

Source code in src/pygwrx/models/grgwr.py
def predict_result(self, X: ArrayLike, coords: ArrayLike) -> GRGWRPredictionResult:
    """Assign query regimes and recalibrate local WLS coefficients."""
    self._require_fitted()
    X_raw = self._coerce_X_predict(X)
    X_design = add_intercept(X_raw) if self.fit_intercept else X_raw.copy()
    coords_arr = np.asarray(validate_coords(coords), dtype=float)
    if X_raw.shape[0] != coords_arr.shape[0]:
        raise ValueError("X and coords must contain the same rows.")
    distances = cdist(coords_arr, self.coords_)
    n_query = X_raw.shape[0]
    parameters = np.zeros((n_query, self._Xd.shape[1]))
    regimes = np.zeros(n_query, dtype=int)
    global_beta = np.linalg.lstsq(self._Xd, self.y_, rcond=None)[0]
    for i in range(n_query):
        regime = self._assigned_regime(distances[i])
        regimes[i] = regime
        indices = np.flatnonzero(self.regimes_ == regime)
        if indices.size < self._Xd.shape[1]:
            parameters[i] = global_beta
            continue
        beta, _ = self._solve_local(
            self._Xd[indices],
            self.y_[indices],
            self._weights(distances[i, indices]),
        )
        parameters[i] = beta if np.all(np.isfinite(beta)) else global_beta
    predictions = np.einsum("ij,ij->i", X_design, parameters)
    if self.fit_intercept:
        intercepts = parameters[:, 0]
        coefficients = parameters[:, 1:]
    else:
        intercepts = np.zeros(n_query)
        coefficients = parameters
    return GRGWRPredictionResult(
        predictions=predictions,
        coefficients=coefficients,
        intercepts=intercepts,
        regimes=regimes,
        coords=coords_arr.copy(),
        feature_names=self.feature_names_,
    )

predict

predict(X: ArrayLike, coords: ArrayLike) -> np.ndarray

Return direct GR-GWR predictions at query locations.

Source code in src/pygwrx/models/grgwr.py
def predict(self, X: ArrayLike, coords: ArrayLike) -> np.ndarray:
    """Return direct GR-GWR predictions at query locations."""
    return self.predict_result(X, coords).predictions

results_frame

results_frame() -> pd.DataFrame

Return training regimes, local parameters and fitted values.

Source code in src/pygwrx/models/grgwr.py
def results_frame(self) -> pd.DataFrame:
    """Return training regimes, local parameters and fitted values."""
    self._require_fitted()
    data: Dict[str, np.ndarray] = {
        "coord_0": self.coords_[:, 0],
        "coord_1": self.coords_[:, 1],
        "regime": self.regimes_,
        "fitted": self.fitted_values_,
        "residual": self.residuals_,
        "intercept": self.intercept_,
    }
    for index, name in enumerate(self.feature_names_):
        data[f"coef_{name}"] = self.coef_[:, index]
    return pd.DataFrame(data)

to_frame

to_frame() -> pd.DataFrame

Alias for :meth:results_frame.

Source code in src/pygwrx/models/grgwr.py
def to_frame(self) -> pd.DataFrame:
    """Alias for :meth:`results_frame`."""
    return self.results_frame()

select_parameters classmethod

select_parameters(
    X: ArrayLike,
    y: VectorLike,
    coords: ArrayLike,
    *,
    n_regimes_grid: Tuple[int, ...] = (2, 3),
    bandwidth_grid: Tuple[BandwidthLike, ...] = (20, 30),
    lambda_boundary_grid: Tuple[float, ...] = (0.0, 1.0),
    spatial_constraint_grid: Tuple[float, ...] = (
        0.25,
        0.5,
        0.75,
    ),
    criterion: str = "conditional_aicc",
    cv_folds: int = 5,
    random_state: Optional[int] = 42,
    **model_kwargs: Any
) -> Tuple["GRGWR", pd.DataFrame]

Select a modest GR-GWR parameter grid and fit the best model.

criterion="conditional_aicc" compares final smoothers conditional on their discovered labels. criterion="spatial_cv" forms compact coordinate clusters and reports mean held-out squared error. The search is intentionally explicit and exhaustive; users should keep the grids small because every candidate contains a regime-discovery fit.

Returns:

Type Description
Tuple['GRGWR', DataFrame]

(best_model, search_table) sorted by ascending score.

Source code in src/pygwrx/models/grgwr.py
@classmethod
def select_parameters(
    cls,
    X: ArrayLike,
    y: VectorLike,
    coords: ArrayLike,
    *,
    n_regimes_grid: Tuple[int, ...] = (2, 3),
    bandwidth_grid: Tuple[BandwidthLike, ...] = (20, 30),
    lambda_boundary_grid: Tuple[float, ...] = (0.0, 1.0),
    spatial_constraint_grid: Tuple[float, ...] = (0.25, 0.5, 0.75),
    criterion: str = "conditional_aicc",
    cv_folds: int = 5,
    random_state: Optional[int] = 42,
    **model_kwargs: Any,
) -> Tuple["GRGWR", pd.DataFrame]:
    """Select a modest GR-GWR parameter grid and fit the best model.

    ``criterion="conditional_aicc"`` compares final smoothers conditional
    on their discovered labels. ``criterion="spatial_cv"`` forms compact
    coordinate clusters and reports mean held-out squared error. The search
    is intentionally explicit and exhaustive; users should keep the grids
    small because every candidate contains a regime-discovery fit.

    Returns:
        ``(best_model, search_table)`` sorted by ascending score.
    """
    criterion_key = cls._choice(
        criterion, "criterion", {"conditional_aicc", "spatial_cv"}
    )
    grids = (
        tuple(n_regimes_grid),
        tuple(bandwidth_grid),
        tuple(lambda_boundary_grid),
        tuple(spatial_constraint_grid),
    )
    if any(len(grid) == 0 for grid in grids):
        raise ValueError("All parameter grids must contain at least one value.")
    forbidden = {
        "n_regimes",
        "bandwidth",
        "lambda_boundary",
        "spatial_constraint_weight",
        "random_state",
    } & set(model_kwargs)
    if forbidden:
        raise ValueError(
            "Searched parameters must be supplied through their grid arguments: "
            f"{sorted(forbidden)}."
        )

    y_array = cls._numeric_y(y)
    coords_array = np.asarray(validate_coords(coords), dtype=float)
    X_rows = X.shape[0] if hasattr(X, "shape") else np.asarray(X).shape[0]
    if X_rows != y_array.size or coords_array.shape[0] != y_array.size:
        raise ValueError("X, y and coords must contain the same number of rows.")

    fold_labels: Optional[np.ndarray] = None
    if criterion_key == "spatial_cv":
        folds = cls._positive_int(cv_folds, "cv_folds")
        folds = min(folds, y_array.size)
        if folds < 2:
            raise ValueError("spatial_cv requires at least two folds.")
        cluster = import_optional_dependency(
            "sklearn.cluster", extra="ml", purpose="GRGWR spatial cross-validation"
        )
        fold_labels = cluster.KMeans(
            n_clusters=folds,
            n_init=10,
            random_state=random_state,
        ).fit_predict(coords_array)

    def subset_rows(value: Any, indices: np.ndarray) -> Any:
        if isinstance(value, (pd.DataFrame, pd.Series)):
            return value.iloc[indices]
        return np.asarray(value)[indices]

    records = []
    candidate_parameters = list(product(*grids))
    for candidate_id, (n_regimes, bandwidth, boundary, gamma) in enumerate(
        candidate_parameters
    ):
        parameters = {
            "n_regimes": n_regimes,
            "bandwidth": bandwidth,
            "lambda_boundary": boundary,
            "spatial_constraint_weight": gamma,
            "random_state": random_state,
            **model_kwargs,
        }
        if criterion_key == "conditional_aicc":
            candidate = cls(**parameters).fit(X, y, coords)
            score = float(candidate.diagnostics_["conditional_aicc"])
            fold_scores: Tuple[float, ...] = ()
        else:
            scores = []
            for fold in range(int(np.max(fold_labels)) + 1):
                test_index = np.flatnonzero(fold_labels == fold)
                train_index = np.flatnonzero(fold_labels != fold)
                candidate = cls(**parameters).fit(
                    subset_rows(X, train_index),
                    subset_rows(y, train_index),
                    coords_array[train_index],
                )
                prediction = candidate.predict(
                    subset_rows(X, test_index), coords_array[test_index]
                )
                scores.append(
                    float(np.mean((y_array[test_index] - prediction) ** 2))
                )
            fold_scores = tuple(scores)
            score = float(np.mean(scores))
        records.append(
            {
                "candidate_id": candidate_id,
                "n_regimes": int(n_regimes),
                "bandwidth": bandwidth,
                "lambda_boundary": float(boundary),
                "spatial_constraint_weight": float(gamma),
                "criterion": criterion_key,
                "score": score,
                "fold_scores": fold_scores,
            }
        )

    table = pd.DataFrame(records).sort_values(
        ["score", "n_regimes", "lambda_boundary"], ignore_index=True
    )
    best = table.iloc[0]
    best_record = records[int(best["candidate_id"])]
    best_model = cls(
        n_regimes=int(best_record["n_regimes"]),
        bandwidth=best_record["bandwidth"],
        lambda_boundary=float(best_record["lambda_boundary"]),
        spatial_constraint_weight=float(best_record["spatial_constraint_weight"]),
        random_state=random_state,
        **model_kwargs,
    ).fit(X, y, coords)
    best_model.search_results_ = table.copy()
    best_model.selection_criterion_ = criterion_key
    return best_model, table

summary

summary() -> str

Return a plain-text fitted-model summary.

Source code in src/pygwrx/models/grgwr.py
def summary(self) -> str:
    """Return a plain-text fitted-model summary."""
    self._require_fitted()
    return format_summary(
        "GR-GWR Summary",
        {
            "model": "GR-GWR",
            "n_samples": int(self.y_.size),
            "n_features": int(self.n_features_in_),
            "n_regimes_requested": int(self.n_regimes),
            "n_regimes_actual": int(self.n_regimes_actual_),
            "regime_sizes": tuple(int(value) for value in self.regime_sizes_),
            "component_counts": tuple(
                int(value) for value in self.regime_component_counts_
            ),
            "bandwidth": self.bandwidth,
            "kernel": self.kernel,
            "lambda_boundary": self.lambda_boundary,
            "n_neighbors": self.n_neighbors,
            "min_regime_size": self._min_regime_size_,
            "n_iterations": self.n_iter_,
            "converged": self.converged_,
            "stop_reason": self.stop_reason_,
            "objective_history": tuple(self.objective_history_),
            "r2": float(self.diagnostics_["r2"]),
            "adj_r2": float(self.diagnostics_["adj_r2"]),
            "rmse": float(self.diagnostics_["rmse"]),
            "conditional_aicc": float(self.diagnostics_["conditional_aicc"]),
            "conditional_enp": float(self.diagnostics_["conditional_enp"]),
            "n_boundaries": len(self.regime_boundaries_),
        },
    )

GRGWRPredictionResult

Detailed GR-GWR predictions at evaluation locations.

Property Value
Type class
Import from pygwrx.models import GRGWRPredictionResult
Signature GRGWRPredictionResult(predictions: 'np.ndarray', coefficients: 'np.ndarray', intercepts: 'np.ndarray', regimes: 'np.ndarray', coords: 'np.ndarray', feature_names: 'Tuple[str, ...]') -> None
Maintained example examples/models/19_gr_gwr.py

GRGWRPredictionResult dataclass

GRGWRPredictionResult(
    predictions: ndarray,
    coefficients: ndarray,
    intercepts: ndarray,
    regimes: ndarray,
    coords: ndarray,
    feature_names: Tuple[str, ...],
)

Detailed GR-GWR predictions at evaluation locations.

to_frame

to_frame() -> pd.DataFrame

Return predictions, regimes and local parameters as a DataFrame.

Source code in src/pygwrx/models/grgwr.py
def to_frame(self) -> pd.DataFrame:
    """Return predictions, regimes and local parameters as a DataFrame."""
    data: Dict[str, np.ndarray] = {
        "coord_0": self.coords[:, 0],
        "coord_1": self.coords[:, 1],
        "prediction": self.predictions,
        "regime": self.regimes,
        "intercept": self.intercepts,
    }
    for index, name in enumerate(self.feature_names):
        data[f"coef_{name}"] = self.coefficients[:, index]
    return pd.DataFrame(data)

Runnable examples used on this page

examples/models/19_gr_gwr.py
# SPDX-FileCopyrightText: 2026 Jinghao Hu
# SPDX-License-Identifier: MIT

"""Fit geo-regime GWR and inspect connected spatial regimes."""

# Allow this script to run directly from any working directory.
import sys
from pathlib import Path

_PROJECT_ROOT = Path(__file__).resolve().parents[2]
_EXAMPLES_ROOT = _PROJECT_ROOT / "examples"
_SRC_ROOT = _PROJECT_ROOT / "src"
for _path in (_SRC_ROOT, _EXAMPLES_ROOT):
    if str(_path) not in sys.path:
        sys.path.insert(0, str(_path))

from _common import print_model_result, regime_regression

from pygwrx import GRGWR, GRGWRPredictionResult

X, y, coords, truth = regime_regression(n=56)
model = GRGWR(n_regimes=2, bandwidth=18, max_iter=2, random_state=0).fit(X, y, coords)
print_model_result(model)
print("regime_sizes=", model.regime_sizes_)
print(
    "truth_agreement_or_label_swap=",
    max((model.regimes_ == truth).mean(), (model.regimes_ != truth).mean()),
)
result = model.predict_result(X.iloc[:3], coords.iloc[:3])
assert isinstance(result, GRGWRPredictionResult)
print(result.to_frame())