From 72b8aac4d5e07a237b65ab687bb016b89ff00e5b Mon Sep 17 00:00:00 2001 From: gitRasheed Date: Sun, 5 Jul 2026 10:59:38 +0100 Subject: [PATCH 1/2] Consolidate duplicated input-coercion helpers into _helpers Five near-identical validation helpers were re-defined per module: _as_matrix in causation/norm/distance (strict 2D), _as_matrix in boost/stack (1D-to-column promoting), _as_point_matrix and _as_vector in boost/stack, and _as_pair in dependence/part. Each pair or triple was byte-identical modulo the name interpolated into error messages. Move one canonical copy of each into nns._helpers: _as_matrix (strict), _as_vector_or_matrix (promoting), _as_point_matrix, _as_flat_vector, and _as_pair. The promoting and flattening variants are renamed so the shared module distinguishes them from the strict ones. All error messages and validation semantics are unchanged. Deliberate variants stay local: co_moments._as_pair (permits NaN/inf), pm_matrix._as_matrix (no finite check), copula._as_matrix (>=2-column rule), and the norm/distance _as_vector helpers (bespoke messages). --- src/nns/_helpers.py | 67 +++++++++++++++++++++++++++++++++++++++++++ src/nns/boost.py | 39 ++----------------------- src/nns/causation.py | 15 ++-------- src/nns/dependence.py | 18 +----------- src/nns/distance.py | 13 ++------- src/nns/norm.py | 16 ++--------- src/nns/part.py | 18 +----------- src/nns/stack.py | 45 ++++++----------------------- 8 files changed, 87 insertions(+), 144 deletions(-) diff --git a/src/nns/_helpers.py b/src/nns/_helpers.py index 6618aaa..3ddf467 100644 --- a/src/nns/_helpers.py +++ b/src/nns/_helpers.py @@ -24,6 +24,73 @@ def _warn_unsupported(**was_set: bool) -> None: ) +def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: + """Return x as a non-empty, all-finite 2D float64 matrix.""" + values = np.asarray(x, dtype=np.float64) + if values.ndim != 2: + raise ValueError(f"{name} must be 2D.") + if values.shape[0] == 0 or values.shape[1] == 0: + raise ValueError(f"{name} must be non-empty.") + if not np.all(np.isfinite(values)): + raise ValueError(f"{name} must contain only finite values.") + return values + + +def _as_vector_or_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: + """Return x as a non-empty, all-finite 2D float64 matrix, promoting 1D to a column.""" + values = np.asarray(x, dtype=np.float64) + if values.ndim == 1: + values = values.reshape(-1, 1) + if values.ndim != 2 or values.shape[0] == 0 or values.shape[1] == 0: + raise ValueError(f"{name} must be a non-empty numeric vector or matrix.") + if not np.all(np.isfinite(values)): + raise ValueError(f"{name} must contain only finite values.") + return values + + +def _as_point_matrix(x: NDArray[np.float64], n_cols: int) -> NDArray[np.float64]: + """Return evaluation points as an all-finite matrix with the training column count.""" + values = np.asarray(x, dtype=np.float64) + if values.ndim == 1: + if n_cols == 1: + values = values.reshape(-1, 1) + else: + values = values.reshape(1, -1) + if values.ndim != 2 or values.shape[1] != n_cols: + raise ValueError("ivs_test must have the same column count as ivs_train.") + if not np.all(np.isfinite(values)): + raise ValueError("ivs_test must contain only finite values.") + return values + + +def _as_flat_vector(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: + """Return x flattened to a non-empty, all-finite 1D float64 vector.""" + values = np.asarray(x, dtype=np.float64).reshape(-1) + if values.size == 0: + raise ValueError(f"{name} must be non-empty.") + if not np.all(np.isfinite(values)): + raise ValueError(f"{name} must contain only finite values.") + return values + + +def _as_pair( + x: NDArray[np.float64], + y: NDArray[np.float64], +) -> tuple[NDArray[np.float64], NDArray[np.float64]]: + """Return x and y as equal-length, non-empty, all-finite 1D float64 vectors.""" + x_values = np.asarray(x, dtype=np.float64) + y_values = np.asarray(y, dtype=np.float64) + if x_values.ndim != 1 or y_values.ndim != 1: + raise ValueError("x and y must be 1D.") + if x_values.size == 0: + raise ValueError("x and y must be non-empty.") + if x_values.size != y_values.size: + raise ValueError("x and y must have the same length.") + if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)): + raise ValueError("x and y must contain only finite values.") + return x_values, y_values + + def _fast_lm(x: NDArray[np.float64], y: NDArray[np.float64]) -> tuple[float, float]: """Return intercept and slope matching R's fast_lm helper.""" x_values = np.asarray(x, dtype=np.float64) diff --git a/src/nns/boost.py b/src/nns/boost.py index 99e4408..2120533 100644 --- a/src/nns/boost.py +++ b/src/nns/boost.py @@ -9,6 +9,7 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_flat_vector, _as_point_matrix, _as_vector_or_matrix from nns.categorical import _balance_class_training, _dense_factor_codes, encode_factor_codes from nns.dependence import _gravity from nns.regression import ( @@ -80,7 +81,7 @@ def nns_boost( ): raise ValueError("string/object predictor values require explicit factor_levels.") - x_train = _as_matrix(x_input, "ivs_train") + x_train = _as_vector_or_matrix(x_input, "ivs_train") x_test = ( x_train.copy() if x_test_input is None else _as_point_matrix(x_test_input, x_train.shape[1]) ) @@ -97,7 +98,7 @@ def nns_boost( ) class_codes = np.unique(y_train[np.isfinite(y_train)]) else: - y_train = _as_vector(dv_train, "dv_train") + y_train = _as_flat_vector(dv_train, "dv_train") class_codes = np.empty(0, dtype=np.float64) if x_train.shape[0] != y_train.size: raise ValueError("ivs_train and dv_train must have the same row count.") @@ -663,37 +664,3 @@ def _sse(predicted: NDArray[np.float64], actual: NDArray[np.float64]) -> float: def _accuracy(predicted: NDArray[np.float64], actual: NDArray[np.float64]) -> float: return float(np.mean(predicted == actual)) - - -def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim == 1: - values = values.reshape(-1, 1) - if values.ndim != 2 or values.shape[0] == 0 or values.shape[1] == 0: - raise ValueError(f"{name} must be a non-empty numeric vector or matrix.") - if not np.all(np.isfinite(values)): - raise ValueError(f"{name} must contain only finite values.") - return values - - -def _as_point_matrix(x: NDArray[np.float64], n_cols: int) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim == 1: - if n_cols == 1: - values = values.reshape(-1, 1) - else: - values = values.reshape(1, -1) - if values.ndim != 2 or values.shape[1] != n_cols: - raise ValueError("ivs_test must have the same column count as ivs_train.") - if not np.all(np.isfinite(values)): - raise ValueError("ivs_test must contain only finite values.") - return values - - -def _as_vector(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64).reshape(-1) - if values.size == 0: - raise ValueError(f"{name} must be non-empty.") - if not np.all(np.isfinite(values)): - raise ValueError(f"{name} must contain only finite values.") - return values diff --git a/src/nns/causation.py b/src/nns/causation.py index 40dc9ac..1827b92 100644 --- a/src/nns/causation.py +++ b/src/nns/causation.py @@ -5,9 +5,9 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_matrix, _as_pair from nns.core import lpm_ratio, upm_ratio from nns.dependence import ( - _as_pair, _copula_degree0_unsigned, _copula_signed, _directional_dep, @@ -68,7 +68,7 @@ def causal_matrix( tau: int | str = 0, ) -> NDArray[np.float64]: """Return R's NNS.caus.matrix antisymmetric net-causation matrix.""" - values = _as_matrix(x) + values = _as_matrix(x, "x") if tau != "ts": tau = _tau_value(tau) n_variables = values.shape[1] @@ -175,14 +175,3 @@ def _cap_inf100(value: float, cap: float = 100.0) -> float: if abs(value) > cap: return math.copysign(cap, value) return value - - -def _as_matrix(x: NDArray[np.float64]) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim != 2: - raise ValueError("x must be 2D.") - if values.shape[0] == 0 or values.shape[1] == 0: - raise ValueError("x must be non-empty.") - if not np.all(np.isfinite(values)): - raise ValueError("x must contain only finite values.") - return values diff --git a/src/nns/dependence.py b/src/nns/dependence.py index 9cc5c2b..479d392 100644 --- a/src/nns/dependence.py +++ b/src/nns/dependence.py @@ -6,6 +6,7 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_pair from nns._native import native_fn from nns.co_moments import co_lpm, co_upm, d_lpm, d_upm @@ -409,20 +410,3 @@ def _finite_or_zero(value: float) -> float: def _is_constant(values: NDArray[np.float64]) -> bool: return bool(np.all(values == values[0])) - - -def _as_pair( - x: NDArray[np.float64], - y: NDArray[np.float64], -) -> tuple[NDArray[np.float64], NDArray[np.float64]]: - x_values = np.asarray(x, dtype=np.float64) - y_values = np.asarray(y, dtype=np.float64) - if x_values.ndim != 1 or y_values.ndim != 1: - raise ValueError("x and y must be 1D.") - if x_values.size == 0: - raise ValueError("x and y must be non-empty.") - if x_values.size != y_values.size: - raise ValueError("x and y must have the same length.") - if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)): - raise ValueError("x and y must contain only finite values.") - return x_values, y_values diff --git a/src/nns/distance.py b/src/nns/distance.py index a96f6cd..c0849b4 100644 --- a/src/nns/distance.py +++ b/src/nns/distance.py @@ -6,6 +6,8 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_matrix + KValue = int | Literal["all"] @@ -231,14 +233,3 @@ def _as_vector(x: NDArray[np.float64]) -> NDArray[np.float64]: if not np.all(np.isfinite(values)): raise ValueError("dist_estimate must contain only finite values.") return values - - -def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim != 2: - raise ValueError(f"{name} must be 2D.") - if values.shape[0] == 0 or values.shape[1] == 0: - raise ValueError(f"{name} must be non-empty.") - if not np.all(np.isfinite(values)): - raise ValueError(f"{name} must contain only finite values.") - return values diff --git a/src/nns/norm.py b/src/nns/norm.py index dfe77b0..7c583fe 100644 --- a/src/nns/norm.py +++ b/src/nns/norm.py @@ -6,6 +6,7 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_matrix from nns.dependence import nns_dep @@ -39,14 +40,14 @@ def nns_norm( vector. """ if isinstance(x, np.ndarray): - values = _as_matrix(x) + values = _as_matrix(x, "x") else: series = [_as_vector(item, index) for index, item in enumerate(x)] if not series: raise ValueError("x must be non-empty.") if len({item.size for item in series}) > 1: return _norm_unequal_series(series) - values = _as_matrix(np.column_stack(series)) + values = _as_matrix(np.column_stack(series), "x") means = np.mean(values, axis=0) means = means.copy() means[means == 0.0] = 1e-10 @@ -97,14 +98,3 @@ def _as_vector(x: NDArray[np.float64], index: int) -> NDArray[np.float64]: if not np.all(np.isfinite(values)): raise ValueError(f"x[{index}] must contain only finite values.") return values - - -def _as_matrix(x: NDArray[np.float64]) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim != 2: - raise ValueError("x must be 2D.") - if values.shape[0] == 0 or values.shape[1] == 0: - raise ValueError("x must be non-empty.") - if not np.all(np.isfinite(values)): - raise ValueError("x must contain only finite values.") - return values diff --git a/src/nns/part.py b/src/nns/part.py index 276d6e7..da6f5a4 100644 --- a/src/nns/part.py +++ b/src/nns/part.py @@ -6,6 +6,7 @@ import numpy as np from numpy.typing import NDArray +from nns._helpers import _as_pair from nns._native import native_fn from nns.central_tendencies import _nearest_int_half_up_array, nns_mode from nns.dependence import _gravity @@ -259,20 +260,3 @@ def _validate_noise_reduction(value: str) -> NoiseReduction: "noise_reduction must be one of 'mean', 'median', 'mode', 'mode_class', 'off'." ) return cast(NoiseReduction, noise) - - -def _as_pair( - x: NDArray[np.float64], - y: NDArray[np.float64], -) -> tuple[NDArray[np.float64], NDArray[np.float64]]: - x_values = np.asarray(x, dtype=np.float64) - y_values = np.asarray(y, dtype=np.float64) - if x_values.ndim != 1 or y_values.ndim != 1: - raise ValueError("x and y must be 1D.") - if x_values.size == 0: - raise ValueError("x and y must be non-empty.") - if x_values.size != y_values.size: - raise ValueError("x and y must have the same length.") - if not np.all(np.isfinite(x_values)) or not np.all(np.isfinite(y_values)): - raise ValueError("x and y must contain only finite values.") - return x_values, y_values diff --git a/src/nns/stack.py b/src/nns/stack.py index 2f843eb..ba90fb1 100644 --- a/src/nns/stack.py +++ b/src/nns/stack.py @@ -7,7 +7,12 @@ import numpy as np from numpy.typing import NDArray -from nns._helpers import _warn_unsupported +from nns._helpers import ( + _as_flat_vector, + _as_point_matrix, + _as_vector_or_matrix, + _warn_unsupported, +) from nns.categorical import _balance_class_training, _dense_factor_codes from nns.central_tendencies import nns_mode from nns.dependence import _gravity @@ -107,14 +112,14 @@ def nns_stack( factor_levels=factor_levels, ) - x_train = _as_matrix(x_input, "ivs_train") + x_train = _as_vector_or_matrix(x_input, "ivs_train") if balance: y_train, class_codes = _dense_factor_codes(dv_train, levels=class_levels) elif type_value == "class": y_train, _ = _prepare_y_values(dv_train, type_value=type_value, class_levels=class_levels) class_codes = np.unique(y_train[np.isfinite(y_train)]) else: - y_train = _as_vector(dv_train, "dv_train") + y_train = _as_flat_vector(dv_train, "dv_train") class_codes = np.empty(0, dtype=np.float64) if x_train.shape[0] != y_train.size: raise ValueError("ivs_train and dv_train must have the same row count.") @@ -1005,37 +1010,3 @@ def _all_predictors_are_factor( if len(levels_by_column) < x.shape[1]: raise ValueError("factor_levels must provide levels for every predictor column.") return all(levels_by_column[col] is not None for col in range(x.shape[1])) - - -def _as_matrix(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim == 1: - values = values.reshape(-1, 1) - if values.ndim != 2 or values.shape[0] == 0 or values.shape[1] == 0: - raise ValueError(f"{name} must be a non-empty numeric vector or matrix.") - if not np.all(np.isfinite(values)): - raise ValueError(f"{name} must contain only finite values.") - return values - - -def _as_point_matrix(x: NDArray[np.float64], n_cols: int) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64) - if values.ndim == 1: - if n_cols == 1: - values = values.reshape(-1, 1) - else: - values = values.reshape(1, -1) - if values.ndim != 2 or values.shape[1] != n_cols: - raise ValueError("ivs_test must have the same column count as ivs_train.") - if not np.all(np.isfinite(values)): - raise ValueError("ivs_test must contain only finite values.") - return values - - -def _as_vector(x: NDArray[np.float64], name: str) -> NDArray[np.float64]: - values = np.asarray(x, dtype=np.float64).reshape(-1) - if values.size == 0: - raise ValueError(f"{name} must be non-empty.") - if not np.all(np.isfinite(values)): - raise ValueError(f"{name} must contain only finite values.") - return values From 0edc4f90eb1e501398d94aeee155bf4a94e16c2c Mon Sep 17 00:00:00 2001 From: gitRasheed Date: Sun, 5 Jul 2026 11:02:10 +0100 Subject: [PATCH 2/2] Apply ruff format to stack.py --- src/nns/stack.py | 92 +++++++++++++++++++++++++++--------------------- 1 file changed, 52 insertions(+), 40 deletions(-) diff --git a/src/nns/stack.py b/src/nns/stack.py index ba90fb1..7984d97 100644 --- a/src/nns/stack.py +++ b/src/nns/stack.py @@ -329,16 +329,19 @@ def _evaluate_method2( fold_scores.append(float(scores[best_index])) if stack and methods == (1, 2): - fit = cast("RegResult", nns_reg( - cv_x_train, - cv_y_train, - point_est=cv_x_test, - dim_red_method=dim_red_method, - threshold=best_threshold, - order=order, - dist=dist, - point_only=False, - )) + fit = cast( + "RegResult", + nns_reg( + cv_x_train, + cv_y_train, + point_est=cv_x_test, + dim_red_method=dim_red_method, + threshold=best_threshold, + order=order, + dist=dist, + point_only=False, + ), + ) train_star = cast(dict[str, NDArray[np.float64]], fit["x.star"])["x"] test_star = _xstar_for_points( fit, @@ -352,18 +355,21 @@ def _evaluate_method2( final_class_threshold = ( _threshold_mode(threshold_results) if type_value == "class" else math.nan ) - final_fit = cast("RegResult", nns_reg( - x_train, - y_train, - point_est=x_test, - dim_red_method=dim_red_method, - threshold=final_threshold, - order=order, - dist=dist, - point_only=False, - confidence_interval=pred_int, - type=type_value, - )) + final_fit = cast( + "RegResult", + nns_reg( + x_train, + y_train, + point_est=x_test, + dim_red_method=dim_red_method, + threshold=final_threshold, + order=order, + dist=dist, + point_only=False, + confidence_interval=pred_int, + type=type_value, + ), + ) fitted = cast(dict[str, NDArray[np.float64]], final_fit["Fitted.xy"]) fitted_yhat = fitted["y.hat"] prediction = _as_prediction(final_fit["Point.est"], x_test.shape[0]) @@ -609,16 +615,19 @@ def _fold_xstar( predicted = _class_threshold_round(predicted, threshold, cv_y_train) scores[idx] = objective_fn(predicted, cv_y_test) best_index = int(np.nanargmin(scores) if objective == "min" else np.nanargmax(scores)) - fit = cast("RegResult", nns_reg( - cv_x_train, - cv_y_train, - point_est=cv_x_test, - dim_red_method=dim_red_method, - threshold=float(cutoffs[best_index]), - order=order, - dist=dist, - point_only=False, - )) + fit = cast( + "RegResult", + nns_reg( + cv_x_train, + cv_y_train, + point_est=cv_x_test, + dim_red_method=dim_red_method, + threshold=float(cutoffs[best_index]), + order=order, + dist=dist, + point_only=False, + ), + ) return cast(dict[str, NDArray[np.float64]], fit["x.star"])["x"], _xstar_for_points( fit, cv_x_train, @@ -692,14 +701,17 @@ def _threshold_grid( elif isinstance(dim_red_method, str) and dim_red_method.lower() == "equal": return np.array([0.0], dtype=np.float64) else: - fit = cast("RegResult", nns_reg( - x, - y, - dim_red_method=dim_red_method, - order=order, - dist=dist, - point_only=True, - )) + fit = cast( + "RegResult", + nns_reg( + x, + y, + dim_red_method=dim_red_method, + order=order, + dist=dist, + point_only=True, + ), + ) equation = cast(dict[str, NDArray[np.float64]], fit["equation"]) scores = np.abs(np.round(equation["Coefficient"][:-1], 2)) scores = np.asarray(scores, dtype=np.float64)