MGTWR

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

MGTWR

Gaussian multiscale geographically and temporally weighted regression.

Property	Value
Type	class
Import	from pygwrx.models import MGTWR
Signature	MGTWR(bandwidths: 'BandwidthInput' = None, taus: 'TauInput' = None, *, kernel: 'str' = 'bisquare', adaptive: 'bool' = True, fit_intercept: 'bool' = True, bandwidth_method: 'str' = 'aicc', bandwidth_range: 'BandwidthRange' = None, tau_range: 'Tuple[float, float]' = (0.0, 4.0), init_bandwidth: 'Optional[Bandwidth]' = None, init_tau: 'Optional[float]' = None, tol: 'float' = 1e-06, tol_multi: 'float' = 1e-05, max_iter: 'int' = 200, rss_score: 'bool' = False, calculate_inference: 'bool' = True, n_chunks: 'int' = 1, verbose: 'bool' = False) -> 'None'
Maintained example	examples/models/17_mgtwr.py

MGTWR

MGTWR(
    bandwidths: BandwidthInput = None,
    taus: TauInput = None,
    *,
    kernel: str = "bisquare",
    adaptive: bool = True,
    fit_intercept: bool = True,
    bandwidth_method: str = "aicc",
    bandwidth_range: BandwidthRange = None,
    tau_range: Tuple[float, float] = (0.0, 4.0),
    init_bandwidth: Optional[Bandwidth] = None,
    init_tau: Optional[float] = None,
    tol: float = 1e-06,
    tol_multi: float = 1e-05,
    max_iter: int = 200,
    rss_score: bool = False,
    calculate_inference: bool = True,
    n_chunks: int = 1,
    verbose: bool = False
)

Bases: MGWR

Gaussian multiscale geographically and temporally weighted regression.

MGTWR represents the response as a sum of coefficient-specific spatiotemporal terms. Each fitted parameter receives an independent spatial bandwidth and temporal scale parameter. Calibration starts from a common GTWR fit and then updates one additive term at a time from its partial residual until the score of change converges.

Parameters:

Name	Type	Description	Default
bandwidths	BandwidthInput	Optional scalar or one spatial bandwidth per fitted parameter. The fitted parameter count includes the intercept when fit_intercept=True. When omitted, scales are selected during backfitting.	None
taus	TauInput	Optional scalar or one non-negative temporal scale per fitted parameter. It must be supplied together with bandwidths. Distances are combined as sqrt(ds**2 + tau * dt**2). A value of zero removes temporal distance for that coefficient.	None
kernel	str	"gaussian", "bisquare", or "exponential".	'bisquare'
adaptive	bool	Interpret spatial bandwidths as integer nearest-neighbour counts in combined spatiotemporal distance.	True
fit_intercept	bool	Include a spatiotemporally varying intercept.	True
bandwidth_method	str	Scale-selection criterion: "aicc", "aic", "bic", or "cv".	'aicc'
bandwidth_range	BandwidthRange	Common lower and upper spatial bandwidth bounds for automatic selection.	None
tau_range	Tuple[float, float]	Common lower and upper temporal-scale bounds for automatic selection.	(0.0, 4.0)
init_bandwidth	Optional[Bandwidth]	Optional common spatial bandwidth for the initial GTWR fit.	None
init_tau	Optional[float]	Optional common temporal scale for the initial GTWR fit.	None
tol	float	Resolution target used by the deterministic two-dimensional scale search.	1e-06
tol_multi	float	Backfitting score-of-change convergence tolerance.	1e-05
max_iter	int	Maximum number of backfitting iterations.	200
rss_score	bool	Use relative RSS change instead of smooth-function change as the convergence score.	False
calculate_inference	bool	Compute exact smoother traces, effective parameter counts, local standard errors, and information criteria.	True
n_chunks	int	Number of column chunks used by exact smoother inference.	1
verbose	bool	Print scale-search and backfitting progress.	False

Attributes:

Name	Type	Description
bandwidths_	Optional[ndarray]	Final variable-specific spatial bandwidths.
taus_	Optional[ndarray]	Final variable-specific temporal scale parameters.
temporal_bandwidths_	Optional[ndarray]	Equivalent temporal bandwidths, computed as bandwidth / sqrt(tau) and set to infinity when tau == 0.
bandwidth_history_	Optional[ndarray]	Spatial bandwidth vector from every iteration.
tau_history_	Optional[ndarray]	Temporal-scale vector from every iteration.
convergence_history_	Optional[Any]	Score of change from every iteration.
params_	Optional[Any]	Local parameters including the intercept when fitted.
effective_params_by_variable_	Optional[Any]	Exact effective parameter count for each coefficient surface when inference is enabled.
parameter_standard_errors_	Optional[Any]	Local parameter standard errors when inference is enabled.
parameter_t_values_	Optional[Any]	Local parameter t statistics when inference is enabled.

Notes

Automatic scale selection uses a deterministic coarse-to-fine candidate search with explicit boundary evaluation; it is not an exhaustive proof of the global optimum. The numerical interpretation of tau depends on the coordinate and time units. Independent-target prediction is not exposed because a stable MGTWR prediction operator for independently supplied locations is not yet part of the package contract. Use fitted_values_ for calibration-location estimates.

References

Wu, C., Ren, F., Hu, W., and Du, Q. (2019). Multiscale geographically and temporally weighted regression: exploring the spatiotemporal determinants of housing prices. International Journal of Geographical Information Science, 33(3), 489-511.

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

def __init__(
    self,
    bandwidths: BandwidthInput = None,
    taus: TauInput = None,
    *,
    kernel: str = "bisquare",
    adaptive: bool = True,
    fit_intercept: bool = True,
    bandwidth_method: str = "aicc",
    bandwidth_range: BandwidthRange = None,
    tau_range: Tuple[float, float] = (0.0, 4.0),
    init_bandwidth: Optional[Bandwidth] = None,
    init_tau: Optional[float] = None,
    tol: float = 1e-6,
    tol_multi: float = 1e-5,
    max_iter: int = 200,
    rss_score: bool = False,
    calculate_inference: bool = True,
    n_chunks: int = 1,
    verbose: bool = False,
) -> None:
    kernel_name = str(kernel).strip().lower()
    if kernel_name not in self._VALID_KERNELS:
        raise ValueError(f"kernel must be one of {sorted(self._VALID_KERNELS)}.")
    method = str(bandwidth_method).strip().lower()
    if method not in self._VALID_CRITERIA:
        raise ValueError(
            "bandwidth_method must be one of 'aicc', 'aic', 'bic', or 'cv'."
        )
    if (bandwidths is None) != (taus is None):
        raise ValueError("bandwidths and taus must be supplied together.")
    if taus is not None:
        try:
            raw_taus = cast(
                NDArray[np.float64],
                np.asarray(taus, dtype=np.float64).reshape(-1),
            )
        except (TypeError, ValueError) as error:
            raise TypeError("taus must contain numeric values.") from error
        if raw_taus.size == 0:
            raise ValueError("taus cannot be empty.")
        if not np.all(np.isfinite(raw_taus)) or np.any(raw_taus < 0.0):
            raise ValueError("All taus must be finite and non-negative.")
    if not isinstance(calculate_inference, (bool, np.bool_)):
        raise TypeError("calculate_inference must be boolean.")
    if not isinstance(n_chunks, (int, np.integer)) or isinstance(
        n_chunks, (bool, np.bool_)
    ):
        raise TypeError("n_chunks must be an integer.")
    if int(n_chunks) < 1:
        raise ValueError("n_chunks must be at least 1.")
    self._validate_tau_range(tau_range)
    if init_tau is not None:
        self._validate_tau_value(init_tau, name="init_tau")

    super().__init__(
        kernel=kernel_name,
        bandwidths=bandwidths,
        bandwidth_method=method,
        adaptive=adaptive,
        bandwidth_range=bandwidth_range,
        init_bandwidth=init_bandwidth,
        optimization_method="golden_section",
        search_tol=tol,
        search_max_iter=max(50, min(int(max_iter), 500)),
        max_iter=max_iter,
        tol=tol_multi,
        rss_score=rss_score,
        bws_same_times=5,
        fit_intercept=fit_intercept,
        distance_metric="euclidean",
        sigma2_v1=True,
        verbose=verbose,
    )
    self.taus = taus
    self.tau_range = (float(tau_range[0]), float(tau_range[1]))
    self.init_tau = None if init_tau is None else float(init_tau)
    self.calculate_inference = bool(calculate_inference)
    self.n_chunks = int(n_chunks)
    self._reset_mgtwr_state()

fit

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

Fit MGTWR and replace all prior fitted state atomically.

Parameters:

Name	Type	Description	Default
X	ArrayLike	Predictor matrix with shape (n_samples, n_features).	required
y	VectorLike	Response vector with shape (n_samples,).	required
coords	ArrayLike	Spatial coordinates with shape (n_samples, 2).	required
times	VectorLike	Numeric time coordinate with one value per observation.	required

Returns:

Type	Description
'MGTWR'	The fitted model instance.

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

def fit(
    self,
    X: ArrayLike,
    y: VectorLike,
    coords: ArrayLike,
    times: VectorLike,
) -> "MGTWR":
    """Fit MGTWR and replace all prior fitted state atomically.

    Args:
        X: Predictor matrix with shape ``(n_samples, n_features)``.
        y: Response vector with shape ``(n_samples,)``.
        coords: Spatial coordinates with shape ``(n_samples, 2)``.
        times: Numeric time coordinate with one value per observation.

    Returns:
        The fitted model instance.
    """
    self._reset_fit_state()
    try:
        X_arr, y_arr, coords_arr = self._validate_inputs(X, y, coords)
        times_arr = np.asarray(times, dtype=float)
        if times_arr.ndim == 2 and 1 in times_arr.shape:
            times_arr = times_arr.reshape(-1)
        if times_arr.ndim != 1:
            raise ValueError("times must be one-dimensional.")
        if times_arr.shape[0] != X_arr.shape[0]:
            raise ValueError(
                "X, y, coords, and times must contain the same number of rows."
            )
        if not np.all(np.isfinite(times_arr)):
            raise ValueError("times must contain only finite values.")

        feature_names = (
            None
            if self.feature_names_in_ is None
            else self.feature_names_in_.copy()
        )
        self._store_training_data(X_arr, y_arr, coords_arr, copy=True)
        self.feature_names_in_ = feature_names
        self.times_train_ = times_arr.copy()
        X_design = (
            add_intercept(self.X_train_) if self.fit_intercept else self.X_train_
        )
        n_samples, n_parameters = X_design.shape
        if n_samples <= n_parameters + 2:
            raise ValueError(
                "MGTWR requires more observations than fitted parameters plus two."
            )

        self.kernel_func_ = get_kernel_function(self.kernel)
        spatial_distances = np.asarray(
            compute_distance_matrix(
                self.coords_train_,
                self.coords_train_,
                metric=self.distance_metric,
            ),
            dtype=float,
        )
        temporal_distances = self._temporal_distances(self.times_train_)
        manual_bandwidths = self._resolve_manual_bandwidths(
            n_parameters=n_parameters,
            n_samples=n_samples,
        )
        manual_taus = self._resolve_manual_taus(n_parameters=n_parameters)

        if self.init_bandwidth is not None:
            initial_bandwidth = self._validate_bandwidth_value(
                self.init_bandwidth,
                n_samples=n_samples,
                minimum_adaptive=n_parameters + 1,
                name="init_bandwidth",
            )
        elif manual_bandwidths is not None:
            median_bandwidth = float(np.median(manual_bandwidths))
            initial_bandwidth = self._validate_bandwidth_value(
                (
                    max(n_parameters + 1, int(round(median_bandwidth)))
                    if self.adaptive
                    else median_bandwidth
                ),
                n_samples=n_samples,
                minimum_adaptive=n_parameters + 1,
                name="initial bandwidth",
            )
        else:
            initial_bandwidth = None

        if self.init_tau is not None:
            initial_tau = self._validate_tau_value(self.init_tau, name="init_tau")
        elif manual_taus is not None:
            initial_tau = float(np.median(manual_taus))
        else:
            initial_tau = None

        if initial_bandwidth is None or initial_tau is None:
            selected_bandwidth, selected_tau, _ = self._select_scale(
                X_design,
                self.y_train_,
                spatial_distances,
                temporal_distances,
                self.bandwidth_range,
                initial_tau=initial_tau,
            )
            if initial_bandwidth is None:
                initial_bandwidth = self._validate_bandwidth_value(
                    selected_bandwidth,
                    n_samples=n_samples,
                    minimum_adaptive=n_parameters + 1,
                    name="initial selected bandwidth",
                )
            if initial_tau is None:
                initial_tau = selected_tau

        self.initial_bandwidth_ = initial_bandwidth
        self.initial_tau_ = float(initial_tau)
        self.bandwidth_ = initial_bandwidth

        backfit = self._backfit(
            X_design,
            self.y_train_,
            spatial_distances,
            temporal_distances,
            initial_bandwidth,
            float(initial_tau),
            manual_bandwidths,
            manual_taus,
        )
        self.bandwidths_ = backfit.bandwidths.copy()
        self.taus_ = backfit.taus.copy()
        self.temporal_bandwidths_ = np.divide(
            self.bandwidths_.astype(float),
            np.sqrt(self.taus_),
            out=np.full(self.taus_.shape, np.inf, dtype=float),
            where=self.taus_ > 0.0,
        )
        self.bandwidth_history_ = backfit.bandwidth_history.copy()
        self.tau_history_ = backfit.tau_history.copy()
        self.convergence_history_ = backfit.convergence_history.copy()
        self.n_iter_ = backfit.n_iter
        self.converged_ = backfit.converged
        self.parameter_contributions_ = backfit.contributions.copy()
        self.params_ = backfit.params.copy()

        if self.fit_intercept:
            self.intercept_ = self.params_[:, 0].copy()
            self.coef_ = self.params_[:, 1:].copy()
        else:
            self.intercept_ = np.zeros(n_samples, dtype=float)
            self.coef_ = self.params_.copy()
        self.fitted_values_ = np.sum(backfit.contributions, axis=1)
        self.residuals_ = self.y_train_ - self.fitted_values_
        self.local_r2_ = None

        rss = float(np.dot(self.residuals_, self.residuals_))
        tss = float(
            np.dot(
                self.y_train_ - np.mean(self.y_train_),
                self.y_train_ - np.mean(self.y_train_),
            )
        )
        self.rss_ = rss
        self.r2_ = float(1.0 - rss / tss) if tss > 0.0 else np.nan
        self.inference_enabled_ = self.calculate_inference

        if self.calculate_inference:
            inference = self._compute_exact_inference(
                X_design,
                spatial_distances,
                temporal_distances,
                n_chunks=self.n_chunks,
                compute_covariance=True,
            )
            self.effective_params_by_variable_ = (
                inference.effective_params_by_variable.copy()
            )
            self.ENP_j_ = self.effective_params_by_variable_
            self.effective_params_ = float(
                np.sum(self.effective_params_by_variable_)
            )
            self.hat_matrix_ = inference.hat_matrix
            self.partial_hat_matrices_ = inference.partial_hat_matrices
            self.diagnostics_ = compute_diagnostics(
                self.y_train_,
                self.fitted_values_,
                compute_gwr_stats=True,
                trace_S=inference.trace_S,
                trace_StS=inference.trace_StS,
            )
            self._set_inference_results(inference, self.params_)
            self.aic_ = float(self.diagnostics_["aic"])
            self.aicc_ = float(self.diagnostics_["aicc"])
            self.bic_ = float(self.diagnostics_["bic"])
            alpha = np.asarray([0.10, 0.05, 0.01], dtype=float)
            safe_enp = np.maximum(
                self.effective_params_by_variable_, np.finfo(float).eps
            )
            self.adjusted_alpha_by_variable_ = alpha / safe_enp[:, None]
            self.critical_t_values_ = student_t.ppf(
                1.0 - self.adjusted_alpha_by_variable_[:, 1] / 2.0,
                max(self.n_samples_ - 1, 1),
            )
        else:
            self.diagnostics_ = compute_diagnostics(
                self.y_train_,
                self.fitted_values_,
                n_features=n_parameters,
            )

        self._mark_fitted()
        return self
    except Exception:
        self._reset_fit_state()
        raise

predict

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

Reject unsupported independent-target prediction.

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

def predict(self, X: ArrayLike, coords: ArrayLike, times: VectorLike) -> np.ndarray:
    """Reject unsupported independent-target prediction."""
    self._check_is_fitted()
    raise NotImplementedError(
        "Out-of-sample MGTWR prediction is not implemented. Use "
        "fitted_values_ for calibration locations."
    )

to_frame

to_frame() -> pd.DataFrame

Return calibration-location parameters and diagnostics.

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

def to_frame(self) -> pd.DataFrame:
    """Return calibration-location parameters and diagnostics."""
    frame = super().to_frame()
    if self.times_train_ is not None:
        frame.insert(self.coords_train_.shape[1], "time", self.times_train_)
    return frame

summary

summary() -> str

Return fitted model diagnostics as a plain-text table.

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

def summary(self) -> str:
    """Return fitted model diagnostics as a plain-text table."""
    self._check_is_fitted()
    if (
        self.bandwidths_ is None
        or self.taus_ is None
        or self.temporal_bandwidths_ is None
    ):
        raise RuntimeError("MGTWR scale results are unavailable.")
    temporal_bandwidths = self.temporal_bandwidths_
    return format_summary(
        "MGTWR Summary",
        {
            "n_samples": int(self.n_samples_),
            "n_features": int(self.n_features_in_),
            "fit_intercept": bool(self.fit_intercept),
            "initial_bandwidth": self.initial_bandwidth_,
            "initial_tau": self.initial_tau_,
            "bandwidths": self.bandwidths_.tolist(),
            "taus": self.taus_.tolist(),
            "temporal_bandwidths": temporal_bandwidths.tolist(),
            "adaptive": bool(self.adaptive),
            "kernel": self.kernel,
            "iterations": int(self.n_iter_),
            "converged": bool(self.converged_),
            "rss": self.rss_,
            "r2": self.r2_,
            "sigma2": self.sigma2_,
            "effective_params": (
                None
                if self.effective_params_by_variable_ is None
                else float(np.sum(self.effective_params_by_variable_))
            ),
            "aic": self.aic_,
            "aicc": self.aicc_,
            "bic": self.bic_,
        },
    )

Runnable examples used on this page

examples/models/17_mgtwr.py

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

"""Fit the self-contained multiscale geographically and temporally weighted regression."""

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

from pygwrx import MGTWR

X, y, coords, times = temporal_regression(n=20, p=2)
model = MGTWR(
    bandwidths=[12, 12, 12],
    taus=[1.0, 1.0, 1.0],
    adaptive=True,
    calculate_inference=False,
).fit(X, y, coords, times)
print_model_result(model)
print("spatial_bandwidths=", model.bandwidths_)
print("temporal_scales=", model.taus_)
try:
    model.predict(X.iloc[:2], coords.iloc[:2], times[:2])
except NotImplementedError as exc:
    print("Expected MGTWR prediction limitation:", exc)