STWR

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

STWR

Spatiotemporal weighted regression.

Property	Value
Type	class
Import	from pygwrx.models import STWR
Signature	STWR(spatial_bandwidth: 'Bandwidth' = 'cv', *, adaptive: 'bool' = True, kernel: 'str' = 'bisquare', alpha: 'SelectionValue' = 0.3, theta: 'SelectionValue' = 0.0, tick_nums: 'Union[int, str, None]' = None, bandwidth_candidates: 'Optional[Sequence[Number]]' = None, alpha_candidates: 'Optional[Sequence[Number]]' = None, theta_candidates: 'Optional[Sequence[Number]]' = None, tick_candidates: 'Optional[Sequence[int]]' = None, fit_intercept: 'bool' = True, distance_metric: 'str' = 'euclidean', sigma2_v1: 'bool' = True, ridge: 'float' = 0.0, store_weights: 'bool' = True, verbose: 'bool' = False) -> 'None'
Maintained example	examples/models/04_stwr.py

STWR

STWR(
    spatial_bandwidth: Bandwidth = "cv",
    *,
    adaptive: bool = True,
    kernel: str = "bisquare",
    alpha: SelectionValue = 0.3,
    theta: SelectionValue = 0.0,
    tick_nums: Union[int, str, None] = None,
    bandwidth_candidates: Optional[Sequence[Number]] = None,
    alpha_candidates: Optional[Sequence[Number]] = None,
    theta_candidates: Optional[Sequence[Number]] = None,
    tick_candidates: Optional[Sequence[int]] = None,
    fit_intercept: bool = True,
    distance_metric: str = "euclidean",
    sigma2_v1: bool = True,
    ridge: float = 0.0,
    store_weights: bool = True,
    verbose: bool = False
)

Spatiotemporal weighted regression.

STWR uses the response-value variation rate as its time distance. For a current regression point :math:i and a past observation :math:j, the public STWR v1.0 code computes

.. math::

d^T_{ij} = \frac{\Delta t_{\mathrm{all}}}{\Delta t_q}
\left|\frac{y_{j,t-q}-y_{i,t}}{y_{j,t-q}}\right|,

and maps it through :math:\tanh(d^T_{ij}/2). The final weight is a convex combination of a spatial kernel and that temporal effect. The spatial bandwidth at earlier stages is changed linearly by tan(theta).

Parameters:

Name	Type	Description	Default
spatial_bandwidth	Bandwidth	Current-stage fixed distance or adaptive neighbour count. "cv" selects from bandwidth_candidates.	'cv'
adaptive	bool	Interpret spatial bandwidths as neighbour counts at the latest stage before conversion to local distance scales.	True
kernel	str	Spatial kernel. The published implementation primarily uses "bisquare" and "gaussian".	'bisquare'
alpha	SelectionValue	Temporal contribution in [0, 1]. "cv" selects from alpha_candidates.	0.3
theta	SelectionValue	Spatial-bandwidth time slope in radians. Earlier bandwidths equal the latest bandwidth minus tan(theta) * elapsed_time. "cv" selects from theta_candidates.	0.0
tick_nums	Union[int, str, None]	Number of most recent stages used. None uses all stages; "cv" selects from tick_candidates.	None
bandwidth_candidates	Optional[Sequence[Number]]	Optional candidates for automatic bandwidth search.	None
alpha_candidates	Optional[Sequence[Number]]	Optional candidates for automatic alpha search.	None
theta_candidates	Optional[Sequence[Number]]	Optional candidates for automatic theta search.	None
tick_candidates	Optional[Sequence[int]]	Optional candidates for automatic stage-count search.	None
fit_intercept	bool	Include a local intercept.	True
distance_metric	str	Spatial distance metric supported by pyGWRx.	'euclidean'
sigma2_v1	bool	Use n - trace(S) rather than the v2 residual denominator.	True
ridge	float	Non-negative numerical ridge added to local normal matrices. The intercept is not penalized.	0.0
store_weights	bool	Store the final latest-stage-to-history weight matrix.	True
verbose	bool	Print selection and fit information.	False

References

Que, X., Ma, X., Ma, C., & Chen, Q. (2020). A spatiotemporal weighted regression model (STWR v1.0) for analyzing local nonstationarity in space and time. Geoscientific Model Development, 13, 6149-6164.

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

def __init__(
    self,
    spatial_bandwidth: Bandwidth = "cv",
    *,
    adaptive: bool = True,
    kernel: str = "bisquare",
    alpha: SelectionValue = 0.3,
    theta: SelectionValue = 0.0,
    tick_nums: Union[int, str, None] = None,
    bandwidth_candidates: Optional[Sequence[Number]] = None,
    alpha_candidates: Optional[Sequence[Number]] = None,
    theta_candidates: Optional[Sequence[Number]] = None,
    tick_candidates: Optional[Sequence[int]] = None,
    fit_intercept: bool = True,
    distance_metric: str = "euclidean",
    sigma2_v1: bool = True,
    ridge: float = 0.0,
    store_weights: bool = True,
    verbose: bool = False,
) -> None:
    self.spatial_bandwidth = spatial_bandwidth
    self.adaptive = self._boolean(adaptive, "adaptive")
    self.kernel = self._kernel_name(kernel)
    self.alpha = alpha
    self.theta = theta
    self.tick_nums = tick_nums
    self.bandwidth_candidates = bandwidth_candidates
    self.alpha_candidates = alpha_candidates
    self.theta_candidates = theta_candidates
    self.tick_candidates = tick_candidates
    self.fit_intercept = self._boolean(fit_intercept, "fit_intercept")
    self.distance_metric = self._distance_metric(distance_metric)
    self.sigma2_v1 = self._boolean(sigma2_v1, "sigma2_v1")
    self.ridge = self._nonnegative(ridge, "ridge")
    self.store_weights = self._boolean(store_weights, "store_weights")
    self.verbose = self._boolean(verbose, "verbose")
    self.kernel_func_ = get_kernel_function(self.kernel)
    self._reset_fit_state()

fit

fit(
    X_list: Sequence[Union[ndarray, DataFrame]],
    y_list: Sequence[Union[ndarray, Series]],
    coords_list: Sequence[Union[ndarray, DataFrame]],
    time_intervals: Sequence[Number],
) -> "STWR"

Fit STWR for the latest time stage using recent historical stages.

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

def fit(
    self,
    X_list: Sequence[Union[np.ndarray, pd.DataFrame]],
    y_list: Sequence[Union[np.ndarray, pd.Series]],
    coords_list: Sequence[Union[np.ndarray, pd.DataFrame]],
    time_intervals: Sequence[Number],
) -> "STWR":
    """Fit STWR for the latest time stage using recent historical stages."""
    self._reset_fit_state()
    try:
        X_stages, y_stages, coords_stages, intervals = self._validate_stages(
            X_list, y_list, coords_list, time_intervals
        )
        bandwidth, alpha, theta, tick_nums = self._select_parameters(
            X_stages, y_stages, coords_stages, intervals
        )
        X_source, y_source, stage_slices = self._source_arrays(
            X_stages, y_stages, tick_nums
        )
        X_design = (
            add_intercept(X_source) if self.fit_intercept else X_source.copy()
        )
        query_X = X_stages[-1]
        query_design = (
            add_intercept(query_X) if self.fit_intercept else query_X.copy()
        )
        spatial, temporal, weights, _ = self._weight_components(
            coords_stages[-1],
            y_stages[-1],
            coords_stages,
            y_stages,
            intervals,
            bandwidth=bandwidth,
            alpha=alpha,
            theta=theta,
            tick_nums=tick_nums,
        )
        local_fit = self._fit_weight_matrix(
            X_design,
            y_source,
            weights,
            query_design,
            compute_covariance=True,
        )
        y_latest = y_stages[-1]
        residuals = y_latest - local_fit.fitted_values
        n_latest = y_latest.size
        trace_s = float(
            np.sum(
                local_fit.smoother_rows[np.arange(n_latest), np.arange(n_latest)]
            )
        )
        trace_sts = float(np.sum(local_fit.smoother_rows**2))
        diagnostics = compute_diagnostics(
            y_latest,
            local_fit.fitted_values,
            trace_S=max(trace_s, 0.0),
            trace_StS=max(trace_sts, 0.0),
            compute_gwr_stats=True,
        )
        denominator = (
            n_latest - trace_s
            if self.sigma2_v1
            else n_latest - 2.0 * trace_s + trace_sts
        )
        if denominator <= 0.0:
            raise ValueError(
                "STWR residual effective degrees of freedom are not positive; "
                "increase the bandwidth or use fewer historical stages."
            )
        sigma2 = float(np.dot(residuals, residuals) / denominator)
        if local_fit.covariance_diagonal is None:
            raise RuntimeError("STWR covariance factors were not computed.")
        parameter_se = np.sqrt(
            np.maximum(local_fit.covariance_diagonal * sigma2, 0.0)
        )
        parameter_t = np.divide(
            local_fit.parameters,
            parameter_se,
            out=np.full_like(local_fit.parameters, np.nan),
            where=parameter_se > 0.0,
        )

        self.X_stages_ = [stage.copy() for stage in X_stages]
        self.y_stages_ = [stage.copy() for stage in y_stages]
        self.coords_stages_ = [stage.copy() for stage in coords_stages]
        self.time_intervals_ = intervals.copy()
        self.spatial_bandwidth_ = bandwidth
        self.alpha_ = alpha
        self.theta_ = theta
        self.tick_nums_ = tick_nums
        self.stage_slices_ = stage_slices
        self.parameters_ = local_fit.parameters
        if self.fit_intercept:
            self.intercept_ = local_fit.parameters[:, 0]
            self.coef_ = local_fit.parameters[:, 1:]
            self.intercept_se_ = parameter_se[:, 0]
            self.coef_se_ = parameter_se[:, 1:]
            self.intercept_t_ = parameter_t[:, 0]
            self.coef_t_ = parameter_t[:, 1:]
        else:
            self.intercept_ = np.zeros(n_latest, dtype=float)
            self.coef_ = local_fit.parameters
            self.intercept_se_ = np.zeros(n_latest, dtype=float)
            self.coef_se_ = parameter_se
            self.intercept_t_ = np.full(n_latest, np.nan)
            self.coef_t_ = parameter_t
        self.coefficients_ = self.coef_
        self.fitted_values_ = local_fit.fitted_values
        self.residuals_ = residuals
        self.smoother_rows_ = local_fit.smoother_rows
        self.influence_ = local_fit.smoother_rows[
            np.arange(n_latest), np.arange(n_latest)
        ]
        self.parameter_covariance_diagonal_ = local_fit.covariance_diagonal
        self.parameter_standard_errors_ = parameter_se
        self.parameter_t_values_ = parameter_t
        self.sigma2_ = sigma2
        self.diagnostics_ = diagnostics
        if self.store_weights:
            self.weights_ = weights
            self.spatial_weights_ = spatial
            self.temporal_weights_ = temporal
        self._is_fitted = True
    except Exception:
        self._reset_fit_state()
        raise
    if self.verbose:
        print(
            "STWR fitted: "
            f"bandwidth={self.spatial_bandwidth_}, alpha={self.alpha_:.4f}, "
            f"theta={self.theta_:.4f}, tick_nums={self.tick_nums_}, "
            f"AICc={self.diagnostics_['aicc']:.6f}"
        )
    return self

predict_result

predict_result(
    X: Union[ndarray, DataFrame],
    coords: Union[ndarray, DataFrame],
    *,
    reference_y: Optional[Union[ndarray, Series]] = None
) -> STWRPredictionResult

Predict at new locations in the latest modeled time stage.

reference_y supplies the current-stage response baseline required by the STWR variation-rate time distance. When omitted, it is estimated by inverse-distance weighting from the latest observed responses, following the prediction strategy in the public STWR code.

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

def predict_result(
    self,
    X: Union[np.ndarray, pd.DataFrame],
    coords: Union[np.ndarray, pd.DataFrame],
    *,
    reference_y: Optional[Union[np.ndarray, pd.Series]] = None,
) -> STWRPredictionResult:
    """Predict at new locations in the latest modeled time stage.

    ``reference_y`` supplies the current-stage response baseline required by
    the STWR variation-rate time distance. When omitted, it is estimated by
    inverse-distance weighting from the latest observed responses, following
    the prediction strategy in the public STWR code.
    """
    self._require_fitted()
    if (
        self.X_stages_ is None
        or self.y_stages_ is None
        or self.coords_stages_ is None
        or self.time_intervals_ is None
        or self.spatial_bandwidth_ is None
        or self.alpha_ is None
        or self.theta_ is None
        or self.tick_nums_ is None
    ):
        raise RuntimeError("STWR training state is incomplete.")
    X_arr, _ = self._coerce_X_stage(X, expected_names=self.feature_names_)
    coords_arr = validate_coords(coords)
    if X_arr.shape[0] != coords_arr.shape[0]:
        raise ValueError("X and coords must contain the same number of rows.")
    reference = (
        self._estimate_reference_y(coords_arr)
        if reference_y is None
        else self._numeric_vector(reference_y, "reference_y")
    )
    if reference.size != X_arr.shape[0]:
        raise ValueError("reference_y must contain one value per prediction row.")
    X_source, y_source, _ = self._source_arrays(
        self.X_stages_, self.y_stages_, self.tick_nums_
    )
    X_design = add_intercept(X_source) if self.fit_intercept else X_source.copy()
    query_design = add_intercept(X_arr) if self.fit_intercept else X_arr.copy()
    _, _, weights, _ = self._weight_components(
        coords_arr,
        reference,
        self.coords_stages_,
        self.y_stages_,
        self.time_intervals_,
        bandwidth=self.spatial_bandwidth_,
        alpha=self.alpha_,
        theta=self.theta_,
        tick_nums=self.tick_nums_,
    )
    local_fit = self._fit_weight_matrix(
        X_design,
        y_source,
        weights,
        query_design,
        compute_covariance=False,
    )
    if self.fit_intercept:
        intercept = local_fit.parameters[:, 0]
        coef = local_fit.parameters[:, 1:]
    else:
        intercept = np.zeros(X_arr.shape[0], dtype=float)
        coef = local_fit.parameters
    return STWRPredictionResult(
        predictions=local_fit.fitted_values,
        coef=coef,
        intercept=intercept,
        coords=coords_arr.copy(),
        feature_names=self.feature_names_,
        reference_y=reference.copy(),
    )

predict

predict(
    X: Union[ndarray, DataFrame],
    coords: Union[ndarray, DataFrame],
    *,
    reference_y: Optional[Union[ndarray, Series]] = None
) -> np.ndarray

Return STWR predictions at latest-stage locations.

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

def predict(
    self,
    X: Union[np.ndarray, pd.DataFrame],
    coords: Union[np.ndarray, pd.DataFrame],
    *,
    reference_y: Optional[Union[np.ndarray, pd.Series]] = None,
) -> np.ndarray:
    """Return STWR predictions at latest-stage locations."""
    return self.predict_result(X, coords, reference_y=reference_y).predictions

get_results

get_results() -> pd.DataFrame

Return latest-stage fitted values and local coefficients.

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

def get_results(self) -> pd.DataFrame:
    """Return latest-stage fitted values and local coefficients."""
    self._require_fitted()
    if self.coords_stages_ is None or self.fitted_values_ is None:
        raise RuntimeError("STWR training results are incomplete.")
    result = STWRPredictionResult(
        predictions=self.fitted_values_,
        coef=self.coef_,
        intercept=self.intercept_,
        coords=self.coords_stages_[-1],
        feature_names=self.feature_names_,
        reference_y=self.y_stages_[-1].copy(),
    ).to_frame()
    result["residual"] = self.residuals_
    return result

STWRPredictionResult

Detailed predictions produced at the latest modeled time stage.

Property	Value
Type	class
Import	from pygwrx.models import STWRPredictionResult
Signature	STWRPredictionResult(predictions: 'np.ndarray', coef: 'np.ndarray', intercept: 'np.ndarray', coords: 'np.ndarray', feature_names: 'Tuple[str, ...]', reference_y: 'np.ndarray') -> None
Maintained example	examples/models/04_stwr.py

STWRPredictionResult dataclass

STWRPredictionResult(
    predictions: ndarray,
    coef: ndarray,
    intercept: ndarray,
    coords: ndarray,
    feature_names: Tuple[str, ...],
    reference_y: ndarray,
)

Detailed predictions produced at the latest modeled time stage.

to_frame

to_frame() -> pd.DataFrame

Return predictions and local parameters as a DataFrame.

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

def to_frame(self) -> pd.DataFrame:
    """Return predictions and local parameters as a DataFrame."""
    data: Dict[str, np.ndarray] = {
        "coord_0": self.coords[:, 0],
        "coord_1": self.coords[:, 1],
        "prediction": self.predictions,
        "reference_y": self.reference_y,
        "intercept": self.intercept,
    }
    for index, name in enumerate(self.feature_names):
        data[f"coef_{name}"] = self.coef[:, index]
    return pd.DataFrame(data)

Runnable examples used on this page

examples/models/04_stwr.py

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

"""Fit STWR from multiple observation snapshots."""

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

from pygwrx import STWR, STWRPredictionResult

X_list, y_list, coords_list, intervals = stwr_stages()
model = STWR(
    spatial_bandwidth=10,
    adaptive=True,
    alpha=0.3,
    theta=0.0,
    tick_nums=2,
    store_weights=True,
).fit(X_list, y_list, coords_list, intervals)
print_model_result(model)
result = model.predict_result(
    X_list[-1].iloc[:3],
    coords_list[-1].iloc[:3],
    reference_y=y_list[-1][:3],
)
assert isinstance(result, STWRPredictionResult)
print(result.to_frame())