Skip to content

GWSS

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

GWSS

Compute geographically weighted summary statistics.

Property Value
Type class
Import from pygwrx.models import GWSS
Signature GWSS(kernel: 'str \| Any' = 'bisquare', bandwidth: 'float \| int \| None' = None, adaptive: 'bool' = False, quantile: 'bool' = False, verbose: 'bool' = False) -> 'None'
Maintained example examples/models/11_gwss.py

GWSS

GWSS(
    kernel: str | Any = "bisquare",
    bandwidth: float | int | None = None,
    adaptive: bool = False,
    quantile: bool = False,
    verbose: bool = False,
)

Compute geographically weighted summary statistics.

Parameters:

Name Type Description Default
kernel str | Any

Kernel name or callable accepted by :func:pygwrx.core.kernels.get_kernel_function.

'bisquare'
bandwidth float | int | None

Positive fixed distance or, when adaptive=True, a positive integer number of nearest neighbours. If None, a leave-one-out cross-validation bandwidth for local means is selected during fitting.

None
adaptive bool

Whether bandwidth represents a nearest-neighbour count.

False
quantile bool

Whether to calculate local median, interquartile range, and quantile imbalance.

False
verbose bool

Whether to print a compact completion message.

False
Notes

Moment statistics follow GWmodel::gwss. In particular, local variance is the normalized weighted second central moment, whereas bivariate covariance uses the unbiased stats::cov.wt denominator 1 - sum(w**2). Weighted quantiles reproduce GWmodel's findq rule.

Source code in src/pygwrx/models/gwss.py
def __init__(
    self,
    kernel: str | Any = "bisquare",
    bandwidth: float | int | None = None,
    adaptive: bool = False,
    quantile: bool = False,
    verbose: bool = False,
) -> None:
    self.kernel = kernel
    self.bandwidth = bandwidth
    self.adaptive = bool(adaptive)
    self.quantile = bool(quantile)
    self.verbose = bool(verbose)
    get_kernel_function(kernel)
    if bandwidth is not None:
        self._validate_bandwidth(bandwidth, n_samples=None)
    self._clear_fit_state()

select_bandwidth

select_bandwidth(
    X: ndarray | DataFrame,
    coords: ndarray | DataFrame,
    *,
    statistic: str = "mean"
) -> float | int

Select a shared bandwidth by leave-one-out CV.

The score sums the GWmodel mean- or median-CV scores over all variables. This method returns one shared bandwidth suitable for gwss; GWmodel's bw.gwss.average additionally reports variable-specific bandwidths.

Source code in src/pygwrx/models/gwss.py
def select_bandwidth(
    self,
    X: np.ndarray | pd.DataFrame,
    coords: np.ndarray | pd.DataFrame,
    *,
    statistic: str = "mean",
) -> float | int:
    """Select a shared bandwidth by leave-one-out CV.

    The score sums the GWmodel mean- or median-CV scores over all variables.
    This method returns one shared bandwidth suitable for ``gwss``; GWmodel's
    ``bw.gwss.average`` additionally reports variable-specific bandwidths.
    """
    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.")
    statistic_name = str(statistic).strip().lower()
    if statistic_name not in {"mean", "median"}:
        raise ValueError("statistic must be 'mean' or 'median'.")
    distances = compute_distance_matrix(coords_arr, coords_arr)

    def score(candidate: float | int) -> float:
        bw = int(round(candidate)) if self.adaptive else float(candidate)
        total = 0.0
        for i in range(X_arr.shape[0]):
            raw = self._weights(distances[i], bw)
            if not np.isfinite(raw).all() or raw.sum() <= 0:
                return np.inf
            full = raw / raw.sum()
            leave = raw.copy()
            leave[i] = 0.0
            if leave.sum() <= 0:
                return np.inf
            leave /= leave.sum()
            if statistic_name == "mean":
                difference = full @ X_arr - leave @ X_arr
            else:
                full_m = np.array(
                    [
                        self._weighted_quantile(X_arr[:, j], full, (0.5,))[0]
                        for j in range(X_arr.shape[1])
                    ]
                )
                keep = np.arange(X_arr.shape[0]) != i
                leave_m = np.array(
                    [
                        self._weighted_quantile(
                            X_arr[keep, j], leave[keep], (0.5,)
                        )[0]
                        for j in range(X_arr.shape[1])
                    ]
                )
                difference = full_m - leave_m
            total += float(difference @ difference)
        return total

    n = X_arr.shape[0]
    if self.adaptive:
        lower = 2
        candidates = range(lower, n + 1)
        scores = [(candidate, score(candidate)) for candidate in candidates]
        return min(scores, key=lambda item: (item[1], item[0]))[0]
    upper = float(np.max(distances))
    if upper <= 0:
        raise ValueError(
            "fixed-bandwidth selection requires non-identical coordinates."
        )
    result = minimize_scalar(
        score, bounds=(upper / 5000.0, upper), method="bounded"
    )
    if not result.success or not np.isfinite(result.fun):
        raise RuntimeError("bandwidth selection failed.")
    return float(result.x)

fit

fit(
    X: ndarray | DataFrame,
    coords: ndarray | DataFrame,
    summary_coords: ndarray | DataFrame | None = None,
) -> "GWSS"

Calculate local summary statistics and store the fitted result.

Source code in src/pygwrx/models/gwss.py
def fit(
    self,
    X: np.ndarray | pd.DataFrame,
    coords: np.ndarray | pd.DataFrame,
    summary_coords: np.ndarray | pd.DataFrame | None = None,
) -> "GWSS":
    """Calculate local summary statistics and store the fitted result."""
    self._clear_fit_state()
    try:
        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.")
        summary_arr = (
            coords_arr
            if summary_coords is None
            else validate_coords(summary_coords)
        )
        if summary_arr.shape[1] != coords_arr.shape[1]:
            raise ValueError(
                "summary_coords and coords must have the same dimension."
            )
        bandwidth = (
            self.select_bandwidth(X_arr, coords_arr)
            if self.bandwidth is None
            else self._validate_bandwidth(self.bandwidth, X_arr.shape[0])
        )
        distances = compute_distance_matrix(summary_arr, coords_arr)
        weight_rows = []
        for i in range(summary_arr.shape[0]):
            raw = self._weights(distances[i], bandwidth)
            if not np.all(np.isfinite(raw)) or raw.sum() <= 0:
                raise ValueError(
                    f"kernel weights are undefined at summary location {i}."
                )
            weight_rows.append(raw / raw.sum())
        W = np.vstack(weight_rows)

        means = W @ X_arr
        centered = X_arr[None, :, :] - means[:, None, :]
        variances = np.einsum("sn,snv->sv", W, centered**2)
        std = np.sqrt(np.maximum(variances, 0.0))
        third = np.einsum("sn,snv->sv", W, centered**3)
        skew = np.divide(
            third, std**3, out=np.full_like(third, np.nan), where=std > 0
        )
        cv = np.divide(std, means, out=np.full_like(std, np.nan), where=means != 0)

        medians = iqrs = qis = None
        if self.quantile:
            quantiles = np.empty((summary_arr.shape[0], X_arr.shape[1], 3))
            for i, w in enumerate(W):
                for j in range(X_arr.shape[1]):
                    quantiles[i, j] = self._weighted_quantile(
                        X_arr[:, j], w, (0.25, 0.5, 0.75)
                    )
            medians = quantiles[:, :, 1]
            iqrs = quantiles[:, :, 2] - quantiles[:, :, 0]
            numerator = 2 * medians - quantiles[:, :, 2] - quantiles[:, :, 0]
            qis = np.divide(
                numerator, iqrs, out=np.full_like(iqrs, np.nan), where=iqrs != 0
            )

        covariances: dict[tuple[int, int], np.ndarray] = {}
        correlations: dict[tuple[int, int], np.ndarray] = {}
        rank_correlations: dict[tuple[int, int], np.ndarray] = {}
        ranks = np.column_stack(
            [rankdata(X_arr[:, j], method="average") for j in range(X_arr.shape[1])]
        )
        for j in range(X_arr.shape[1] - 1):
            for k in range(j + 1, X_arr.shape[1]):
                cov = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            X_arr[:, j], X_arr[:, k], w
                        )
                        for w in W
                    ]
                )
                var_j = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            X_arr[:, j], X_arr[:, j], w
                        )
                        for w in W
                    ]
                )
                var_k = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            X_arr[:, k], X_arr[:, k], w
                        )
                        for w in W
                    ]
                )
                denom = np.sqrt(var_j * var_k)
                corr = np.divide(
                    cov, denom, out=np.full_like(cov, np.nan), where=denom > 0
                )
                rcov = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            ranks[:, j], ranks[:, k], w
                        )
                        for w in W
                    ]
                )
                rvar_j = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            ranks[:, j], ranks[:, j], w
                        )
                        for w in W
                    ]
                )
                rvar_k = np.array(
                    [
                        self._unbiased_weighted_covariance(
                            ranks[:, k], ranks[:, k], w
                        )
                        for w in W
                    ]
                )
                rdenom = np.sqrt(rvar_j * rvar_k)
                rho = np.divide(
                    rcov, rdenom, out=np.full_like(rcov, np.nan), where=rdenom > 0
                )
                covariances[(j, k)] = cov
                correlations[(j, k)] = corr
                rank_correlations[(j, k)] = rho

        self.X_data_ = X_arr
        self.coords_data_ = coords_arr
        self.coords_summary_ = summary_arr
        self.bandwidth_ = bandwidth
        self.var_names_ = names
        self.weights_ = W
        self.local_mean_ = means
        self.local_var_ = variances
        self.local_std_ = std
        self.local_skewness_ = skew
        self.local_cv_ = cv
        self.local_median_ = medians
        self.local_iqr_ = iqrs
        self.local_qi_ = qis
        self.local_cov_ = covariances
        self.local_corr_ = correlations
        self.local_corr_spearman_ = rank_correlations
        self._is_fitted = True
    except Exception:
        self._clear_fit_state()
        raise
    if self.verbose:
        print(f"GWSS fitted at {self.coords_summary_.shape[0]} locations.")
    return self

summary

summary() -> str

Return a plain-text summary of the fitted local statistics.

Source code in src/pygwrx/models/gwss.py
def summary(self) -> str:
    """Return a plain-text summary of the fitted local statistics."""
    self._require_fitted()
    result: dict[str, Any] = {
        "n_vars": len(self.var_names_),
        "var_names": list(self.var_names_),
        "n_summary_locations": self.local_mean_.shape[0],
        "bandwidth": self.bandwidth_,
        "adaptive": self.adaptive,
    }
    for i, name in enumerate(self.var_names_):
        result[f"{name}_mean_range"] = (
            float(np.nanmin(self.local_mean_[:, i])),
            float(np.nanmax(self.local_mean_[:, i])),
        )
        result[f"{name}_std_range"] = (
            float(np.nanmin(self.local_std_[:, i])),
            float(np.nanmax(self.local_std_[:, i])),
        )
    return format_summary("GWSS Summary", result)

to_dataframe

to_dataframe() -> pd.DataFrame

Return local statistics in GWmodel-compatible column naming.

Source code in src/pygwrx/models/gwss.py
def to_dataframe(self) -> pd.DataFrame:
    """Return local statistics in GWmodel-compatible column naming."""
    self._require_fitted()
    data: dict[str, np.ndarray] = {
        "x": self.coords_summary_[:, 0],
        "y": self.coords_summary_[:, 1],
    }
    for i, name in enumerate(self.var_names_):
        data[f"{name}_LM"] = self.local_mean_[:, i]
        data[f"{name}_LSD"] = self.local_std_[:, i]
        data[f"{name}_LVar"] = self.local_var_[:, i]
        data[f"{name}_LSKe"] = self.local_skewness_[:, i]
        data[f"{name}_LCV"] = self.local_cv_[:, i]
        if self.quantile:
            data[f"{name}_Median"] = self.local_median_[:, i]
            data[f"{name}_IQR"] = self.local_iqr_[:, i]
            data[f"{name}_QI"] = self.local_qi_[:, i]
    for (i, j), cov in self.local_cov_.items():
        left, right = self.var_names_[i], self.var_names_[j]
        data[f"Cov_{left}.{right}"] = cov
        data[f"Corr_{left}.{right}"] = self.local_corr_[(i, j)]
        data[f"Spearman_rho_{left}.{right}"] = self.local_corr_spearman_[(i, j)]
    return pd.DataFrame(data)

Runnable examples used on this page

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

"""Compute geographically weighted summary statistics."""

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

from pygwrx import GWSS

X, _, coords = spatial_regression(n=48, p=3)
model = GWSS(bandwidth=24, adaptive=True, quantile=True).fit(X, coords)
print(model.summary())
print("local_means_shape=", model.local_mean_.shape)
print("local_correlation_pairs=", sorted(model.local_corr_))
print("first_correlation_shape=", next(iter(model.local_corr_.values())).shape)