GWPCA

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

Conceptual guide

GWPCA

Fit a basic geographically weighted principal component analysis.

Property	Value
Type	class
Import	from pygwrx.models import GWPCA
Signature	GWPCA(n_components: 'int' = 2, kernel: 'str \| Any' = 'bisquare', bandwidth: 'float \| int \| str \| None' = 'cv', adaptive: 'bool' = True, scaling: 'bool' = True, compute_scores: 'bool' = False, verbose: 'bool' = False) -> 'None'
Maintained example	examples/models/09_gwpca.py

GWPCA

GWPCA(
    n_components: int = 2,
    kernel: str | Any = "bisquare",
    bandwidth: float | int | str | None = "cv",
    adaptive: bool = True,
    scaling: bool = True,
    compute_scores: bool = False,
    verbose: bool = False,
)

Fit a basic geographically weighted principal component analysis.

Parameters:

Name	Type	Description	Default
n_components	int	Number of local principal components to retain.	2
kernel	str | Any	Spatial kernel name or callable accepted by :func:pygwrx.core.kernels.get_kernel_function.	'bisquare'
bandwidth	float | int | str | None	Positive fixed distance or, when adaptive=True, a positive integer neighbour count. None or "cv" selects a bandwidth using GWmodel-compatible leave-one-out cross-validation and its golden-section search.	'cv'
adaptive	bool	Whether the bandwidth represents a nearest-neighbour count.	True
scaling	bool	Whether to globally standardize variables before local PCA. When false, variables are globally centered only. Both paths then apply local weighted centering, matching GWmodel.	True
compute_scores	bool	Whether to retain locally centered scores for all observations receiving positive weight at each evaluation location.	False
verbose	bool	Whether to print a compact completion message.	False

Notes

At evaluation location :math:u, basic GWPCA decomposes

.. math::

\sqrt{W(u)}\{X^* - \bar{X}_w(u)\} = U D V^T,

where :math:X^* is the globally centered or standardized matrix. Local loadings are columns of :math:V, and local component variances are :math:D^2 / \sum_i w_i(u).

Principal-component signs are mathematically indeterminate. pyGWRx applies a deterministic convention: the largest absolute loading in each component is made positive.

Source code in src/pygwrx/models/gwpca.py

def __init__(
    self,
    n_components: int = 2,
    kernel: str | Any = "bisquare",
    bandwidth: float | int | str | None = "cv",
    adaptive: bool = True,
    scaling: bool = True,
    compute_scores: bool = False,
    verbose: bool = False,
) -> None:
    if isinstance(n_components, (bool, np.bool_)) or not isinstance(
        n_components, Integral
    ):
        raise TypeError("n_components must be a positive integer.")
    if int(n_components) < 1:
        raise ValueError("n_components must be at least 1.")
    if not isinstance(adaptive, (bool, np.bool_)):
        raise TypeError("adaptive must be boolean.")
    if not isinstance(scaling, (bool, np.bool_)):
        raise TypeError("scaling must be boolean.")
    if not isinstance(compute_scores, (bool, np.bool_)):
        raise TypeError("compute_scores must be boolean.")
    if not isinstance(verbose, (bool, np.bool_)):
        raise TypeError("verbose must be boolean.")
    if isinstance(bandwidth, str):
        name = bandwidth.strip().lower()
        if name not in {"cv"}:
            if name == "adaptive":
                raise ValueError(
                    "Use adaptive=True with bandwidth='cv' or an integer "
                    "neighbour count; bandwidth='adaptive' is not supported."
                )
            raise ValueError("bandwidth string must be 'cv'.")
    get_kernel_function(kernel)

    self.n_components = int(n_components)
    self.kernel = kernel
    self.bandwidth = bandwidth
    self.adaptive = bool(adaptive)
    self.scaling = bool(scaling)
    self.compute_scores = bool(compute_scores)
    self.verbose = bool(verbose)
    if bandwidth is not None and not isinstance(bandwidth, str):
        self._validate_bandwidth(bandwidth, n_samples=None)
    self._clear_fit_state()

select_bandwidth

select_bandwidth(
    X: ndarray | DataFrame, coords: ndarray | DataFrame
) -> float | int

Select a fixed or adaptive bandwidth by leave-one-out CV.

Source code in src/pygwrx/models/gwpca.py

def select_bandwidth(
    self,
    X: np.ndarray | pd.DataFrame,
    coords: np.ndarray | pd.DataFrame,
) -> float | int:
    """Select a fixed or adaptive bandwidth by leave-one-out CV."""
    X_arr, _ = self._as_data_matrix(X)
    coords_arr = validate_coords(coords)
    if coords_arr.shape[0] != X_arr.shape[0]:
        raise ValueError("X and coords must contain the same number of rows.")
    if self.n_components > X_arr.shape[1]:
        raise ValueError("n_components cannot exceed the number of variables.")

    mean = np.mean(X_arr, axis=0)
    if self.scaling:
        scale = np.std(X_arr, axis=0, ddof=1)
        if np.any(scale <= 0) or np.any(~np.isfinite(scale)):
            raise ValueError(
                "scaling=True requires every variable to have positive "
                "finite sample standard deviation."
            )
    else:
        scale = np.ones(X_arr.shape[1], dtype=float)
    processed = (X_arr - mean) / scale
    distances = compute_distance_matrix(coords_arr, coords_arr)

    score_cache: dict[float | int, float] = {}

    def score(candidate: float | int) -> float:
        bandwidth: float | int
        if self.adaptive:
            bandwidth = int(candidate)
        else:
            bandwidth = float(candidate)
        if bandwidth not in score_cache:
            values = self._cv_contributions(processed, coords_arr, bandwidth)
            score_cache[bandwidth] = (
                float(np.sum(values)) if np.all(np.isfinite(values)) else np.inf
            )
        return score_cache[bandwidth]

    n_samples = X_arr.shape[0]
    if self.adaptive:
        selected = self._gwmodel_golden_search(
            score, 2.0, float(n_samples), adaptive=True
        )
        if not np.isfinite(score(selected)):
            raise RuntimeError("adaptive GWPCA bandwidth selection failed.")
        return int(selected)

    upper = float(np.max(distances))
    if upper <= 0:
        raise ValueError(
            "fixed-bandwidth selection requires non-identical coordinates."
        )
    selected = self._gwmodel_golden_search(
        score, upper / 5000.0, upper, adaptive=False
    )
    if not np.isfinite(score(selected)):
        raise RuntimeError("fixed GWPCA bandwidth selection failed.")
    return float(selected)

fit

fit(
    X: ndarray | DataFrame,
    coords: ndarray | DataFrame,
    eval_coords: ndarray | DataFrame | None = None,
    compute_cv: bool = False,
) -> "GWPCA"

Fit local principal components at observation or evaluation locations.

Parameters:

Name	Type	Description	Default
X	ndarray | DataFrame	Numeric matrix with observations in rows and variables in columns.	required
coords	ndarray | DataFrame	Observation coordinates with the same row count as X.	required
eval_coords	ndarray | DataFrame | None	Optional coordinates at which local loadings are evaluated. The observations in X remain the weighted data.	None
compute_cv	bool	Whether to retain leave-one-out reconstruction-error contributions for the selected or supplied bandwidth.	False

Returns:

Name	Type	Description
GWPCA	'GWPCA'	The fitted estimator.

Raises:

Type	Description
ValueError	If inputs, bandwidth, or local windows are invalid.

Source code in src/pygwrx/models/gwpca.py

def fit(
    self,
    X: np.ndarray | pd.DataFrame,
    coords: np.ndarray | pd.DataFrame,
    eval_coords: np.ndarray | pd.DataFrame | None = None,
    compute_cv: bool = False,
) -> "GWPCA":
    """Fit local principal components at observation or evaluation locations.

    Args:
        X: Numeric matrix with observations in rows and variables in columns.
        coords: Observation coordinates with the same row count as ``X``.
        eval_coords: Optional coordinates at which local loadings are
            evaluated. The observations in ``X`` remain the weighted data.
        compute_cv: Whether to retain leave-one-out reconstruction-error
            contributions for the selected or supplied bandwidth.

    Returns:
        GWPCA: The fitted estimator.

    Raises:
        ValueError: If inputs, bandwidth, or local windows are invalid.
    """
    self._clear_fit_state()
    X_arr, names = self._as_data_matrix(X)
    coords_arr = validate_coords(coords)
    if coords_arr.shape[0] != X_arr.shape[0]:
        raise ValueError("X and coords must contain the same number of rows.")
    if self.n_components > X_arr.shape[1]:
        raise ValueError("n_components cannot exceed the number of variables.")
    if not isinstance(compute_cv, (bool, np.bool_)):
        raise TypeError("compute_cv must be boolean.")

    if eval_coords is None:
        eval_arr = coords_arr.copy()
    else:
        eval_arr = validate_coords(eval_coords)
        if eval_arr.shape[1] != coords_arr.shape[1]:
            raise ValueError(
                "eval_coords and coords must have the same coordinate dimension."
            )

    decomposition = import_optional_dependency(
        "sklearn.decomposition", extra="ml", purpose="GWPCA"
    )
    processed = self._preprocess_fit(X_arr)
    self.pca_global_ = decomposition.PCA(n_components=min(X_arr.shape)).fit(
        processed
    )

    if self.bandwidth is None or isinstance(self.bandwidth, str):
        bandwidth = self.select_bandwidth(X_arr, coords_arr)
    else:
        bandwidth = self._validate_bandwidth(
            self.bandwidth, n_samples=X_arr.shape[0]
        )

    distances = compute_distance_matrix(eval_arr, coords_arr)
    n_eval = eval_arr.shape[0]
    n_features = X_arr.shape[1]
    loadings = np.empty((n_eval, n_features, self.n_components), dtype=float)
    variances = np.empty((n_eval, n_features), dtype=float)
    local_means = np.empty((n_eval, n_features), dtype=float)
    weights_all = np.empty((n_eval, X_arr.shape[0]), dtype=float)
    local_scores: list[np.ndarray] | None = [] if self.compute_scores else None

    for index in range(n_eval):
        weights = self._weights(distances[index], bandwidth)
        local_loading, local_var, local_mean, score_matrix = self._local_pca(
            processed, weights
        )
        loadings[index] = local_loading
        variances[index] = local_var
        local_means[index] = local_mean
        weights_all[index] = weights
        if local_scores is not None:
            local_scores.append(score_matrix)

    total_variance = np.sum(variances, axis=1)
    if np.any(~np.isfinite(total_variance)) or np.any(total_variance <= 0):
        raise ValueError("At least one local PCA has zero or invalid variance.")
    local_pv = variances[:, : self.n_components] / total_variance[:, None] * 100.0

    self.X_train_ = X_arr.copy()
    self.X_processed_ = processed
    self.coords_train_ = coords_arr.copy()
    self.eval_coords_ = eval_arr.copy()
    self.bandwidth_ = bandwidth
    self.feature_names_ = names
    self.loadings_ = loadings
    self.var_ = variances
    self.local_means_ = local_means
    self.local_pv_ = local_pv
    self.cumulative_pv_ = np.sum(local_pv, axis=1)
    self.scores_ = local_scores
    self.weights_ = weights_all
    self.focal_scores_ = None
    if eval_coords is None:
        self.focal_scores_ = np.einsum(
            "ij,ijk->ik", processed - local_means, loadings
        )
    self.cv_scores_ = (
        self._cv_contributions(processed, coords_arr, bandwidth)
        if compute_cv
        else None
    )
    self._is_fitted = True

    if self.verbose:
        print(
            "GWPCA fit complete: "
            f"n={X_arr.shape[0]}, p={n_features}, "
            f"components={self.n_components}, bandwidth={bandwidth}."
        )
    return self

transform

transform(
    X: ndarray | DataFrame,
    coords: ndarray | DataFrame | None = None,
) -> np.ndarray

Project rows using loadings already calibrated at matching locations.

This method does not interpolate or borrow the nearest loading surface. To score new locations, first fit with those locations in eval_coords.

Parameters:

Name	Type	Description	Default
X	ndarray | DataFrame	Rows to project, one row per fitted evaluation location.	required
coords	ndarray | DataFrame | None	Optional coordinates identifying the fitted evaluation locations. Every row must exactly match one fitted location.	None

Returns:

Name	Type	Description
ndarray	ndarray	Locally centered component scores.

Source code in src/pygwrx/models/gwpca.py

def transform(
    self,
    X: np.ndarray | pd.DataFrame,
    coords: np.ndarray | pd.DataFrame | None = None,
) -> np.ndarray:
    """Project rows using loadings already calibrated at matching locations.

    This method does not interpolate or borrow the nearest loading surface.
    To score new locations, first fit with those locations in ``eval_coords``.

    Args:
        X: Rows to project, one row per fitted evaluation location.
        coords: Optional coordinates identifying the fitted evaluation
            locations. Every row must exactly match one fitted location.

    Returns:
        ndarray: Locally centered component scores.
    """
    if not self._is_fitted:
        raise ValueError("Model is not fitted. Call fit() first.")
    X_arr, _ = self._as_data_matrix(X, min_rows=1)
    processed = self._preprocess_transform(X_arr)

    if coords is None:
        if X_arr.shape[0] != self.eval_coords_.shape[0]:
            raise ValueError(
                "Without coords, X must contain one row per fitted "
                "evaluation location."
            )
        indices = np.arange(X_arr.shape[0])
    else:
        coords_arr = validate_coords(coords)
        if coords_arr.shape[0] != X_arr.shape[0]:
            raise ValueError("X and coords must contain the same number of rows.")
        indices = []
        for coordinate in coords_arr:
            matches = np.flatnonzero(
                np.all(np.isclose(self.eval_coords_, coordinate), axis=1)
            )
            if matches.size != 1:
                raise ValueError(
                    "Every transform coordinate must match exactly one fitted "
                    "evaluation location; refit with the desired eval_coords."
                )
            indices.append(int(matches[0]))
        indices = np.asarray(indices, dtype=int)

    centered = processed - self.local_means_[indices]
    return np.einsum("ij,ijk->ik", centered, self.loadings_[indices])

get_winning_variable

get_winning_variable(component: int = 0) -> np.ndarray

Return the largest-absolute-loading variable index per location.

Source code in src/pygwrx/models/gwpca.py

def get_winning_variable(self, component: int = 0) -> np.ndarray:
    """Return the largest-absolute-loading variable index per location."""
    if not self._is_fitted:
        raise ValueError("Model is not fitted. Call fit() first.")
    if isinstance(component, (bool, np.bool_)) or not isinstance(
        component, Integral
    ):
        raise TypeError("component must be an integer index.")
    component_index = int(component)
    if component_index < 0 or component_index >= self.n_components:
        raise ValueError(f"component must lie in [0, {self.n_components - 1}].")
    return np.argmax(np.abs(self.loadings_[:, :, component_index]), axis=1)

to_frame

to_frame() -> pd.DataFrame

Return local variance proportions and winning PC1 variable.

Source code in src/pygwrx/models/gwpca.py

def to_frame(self) -> pd.DataFrame:
    """Return local variance proportions and winning PC1 variable."""
    if not self._is_fitted:
        raise ValueError("Model is not fitted. Call fit() first.")
    data: dict[str, Any] = {
        f"Comp.{index + 1}_PV": self.local_pv_[:, index]
        for index in range(self.n_components)
    }
    data["local_CP"] = self.cumulative_pv_
    winners = self.get_winning_variable(0)
    data["win_var_PC1"] = np.asarray(self.feature_names_, dtype=object)[winners]
    return pd.DataFrame(data)

summary

summary() -> str

Return global and local variance diagnostics as a plain-text table.

Source code in src/pygwrx/models/gwpca.py

def summary(self) -> str:
    """Return global and local variance diagnostics as a plain-text table."""
    if not self._is_fitted:
        raise ValueError("Model is not fitted. Call fit() first.")
    winning_pc1 = self.get_winning_variable(0)
    return format_summary(
        "GWPCA Summary",
        {
            "n_components": self.n_components,
            "bandwidth": self.bandwidth_,
            "global_variance": float(
                np.sum(
                    self.pca_global_.explained_variance_ratio_[: self.n_components]
                )
                * 100.0
            ),
            "local_variance_mean": float(np.mean(self.cumulative_pv_)),
            "local_variance_std": float(np.std(self.cumulative_pv_)),
            "local_variance_range": (
                float(np.min(self.cumulative_pv_)),
                float(np.max(self.cumulative_pv_)),
            ),
            "winning_var_pc1_mode": int(
                np.argmax(
                    np.bincount(winning_pc1, minlength=len(self.feature_names_))
                )
            ),
        },
    )

Runnable examples used on this page

examples/models/09_gwpca.py

# SPDX-FileCopyrightText: 2026 Jinghao Hu
# SPDX-License-Identifier: MIT

"""Fit GWPCA, inspect local loadings, and transform observations."""

# 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, spatial_regression

from pygwrx import GWPCA

X, _, coords = spatial_regression(n=48, p=3)
model = GWPCA(n_components=2, bandwidth=24, adaptive=True).fit(
    X, coords, compute_cv=True
)
print_model_result(model)
print("scores_shape=", model.transform(X, coords).shape)
print("explained_variance_first_location=", model.local_pv_[0])