Merge pull request #132 from OVVO-Financial/claude/dy-d-v057-restore

OVVO-Financial · web-flow · commit acce9682564f · 2026-07-17T15:17:30.000-04:00
diff: add dy_d_best — reconciled dy.d_ (dy.dx step + NNS.stack on X*)
diff --git a/src/nns/__init__.py b/src/nns/__init__.py
@@ -26,6 +26,7 @@
     "d_lpm": ("nns.co_moments", "d_lpm"),
     "dpm_nd": ("nns.dependence", "dpm_nd"),
     "dy_d": ("nns.diff", "dy_d"),
+    "dy_d_best": ("nns.diff", "dy_d_best"),
     "dy_dx": ("nns.diff", "dy_dx"),
     "d_upm": ("nns.co_moments", "d_upm"),
     "ecdf_pm": ("nns.classical", "ecdf_pm"),
diff --git a/src/nns/diff.py b/src/nns/diff.py
@@ -262,6 +262,299 @@ def _combine_dy_d_outputs(
     }
 
 
+def dy_d_best(
+    x: NDArray[Any],
+    y: NDArray[Any],
+    wrt: int | NDArray[np.int64],
+    eval_points: str | float | NDArray[np.float64] = "obs",
+    *,
+    mixed: bool = False,
+    messages: bool = False,
+    factor_levels: Sequence[Sequence[Any] | None] | None = None,
+) -> dict[str, NDArray[np.float64]]:
+    """Reconciled ``dy.d_`` partial-derivative estimator.
+
+    This is the reconciliation of the original NNS 0.5.7 ``dy.d_`` (Vinod & Viole
+    2020, SSRN 3681436) with the current regression engine. It keeps the v0.5.7
+    finite-difference scaffolding but makes two changes that make the estimates
+    uniform across identically-distributed regressors and independent of the old
+    data.table machinery:
+
+    * **Step (``h_step``) shares the ``dy.dx`` logic** - a locally-adaptive step
+      centred on the evaluation point's percentile,
+      ``h_step = VaR(p + H, 1, x) - VaR(p - H, 1, x)`` with
+      ``p = LPM.ratio(1, eval, x)`` - instead of a global quantile spacing with a
+      cumulative window. This is what removes the cross-regressor scatter.
+    * **Estimates come from** ``nns_stack`` on the equal-weight synthetic regressor
+      ``X*`` via the increased-dimension trick ``cbind(X*, X*)``, with
+      ``method=(1, 2)``, ``dim_red_method="equal"``, ``order="max"`` and
+      ``folds=5``. The stack's cross-validated ``n.best`` regularises the (sharper)
+      current engine back toward the paper's regime.
+
+    Bandwidths follow v0.5.7: ``h_s = 1/log(size(x), [2, 10])`` extended by
+    ``10 * h_s`` and doubled when ``nns_dep(x[:, wrt], y) < 0.5``. First and second
+    derivatives use the central-difference forms
+    ``First = (upper - lower) / (2 h_step)`` and
+    ``Second = (upper - 2 f(x) + lower) / h_step ** 2`` (matching ``dy.dx``),
+    blended across bandwidths with a plain ``nanmean``.
+
+    ``wrt`` uses R-style 1-based indexing into the factor-expanded predictor
+    matrix. A scalar/1-D ``eval_points`` evaluates only the selected regressor
+    (averaged over the distribution of the other regressors); a 2-D array
+    evaluates complete predictor tuples.
+    """
+    raw_x = np.asarray(x)
+    if raw_x.ndim != 2:
+        raise ValueError("Please ensure (x) is a matrix or data.frame type object.")
+    if raw_x.shape[1] < 2:
+        raise ValueError("Please use NNS::dy.dx(...) for univariate partial derivatives.")
+
+    raw_y = np.asarray(y).reshape(-1)
+    if raw_y.size != raw_x.shape[0]:
+        raise ValueError("x and y must have compatible row counts.")
+    if _dy_d_has_missing(raw_x) or _dy_d_has_missing(raw_y):
+        raise ValueError("You have some missing values, please address.")
+
+    x_values = _dy_d_expand_predictors(x, factor_levels=factor_levels)
+    y_values = np.asarray(raw_y, dtype=np.float64)
+    if _dy_d_has_missing(x_values) or _dy_d_has_missing(y_values):
+        raise ValueError("You have some missing values, please address.")
+
+    wrt_values = _dy_d_validate_wrt(wrt, x_values.shape[1])
+    outputs = [
+        _dy_d_best_scalar(
+            x_values,
+            y_values,
+            int(wrt_value) - 1,
+            eval_points,
+            mixed=bool(mixed),
+            messages=bool(messages),
+            wrt_label=int(wrt_value),
+        )
+        for wrt_value in wrt_values
+    ]
+    if len(outputs) == 1:
+        return outputs[0]
+    return _combine_dy_d_outputs(outputs)
+
+
+def _dy_d_best_reg_estimates(
+    x: NDArray[np.float64],
+    y: NDArray[np.float64],
+    test_points: NDArray[np.float64],
+) -> NDArray[np.float64]:
+    """f(x +/- h) via NNS.stack on the equal-weight synthetic regressor X*.
+
+    Reduces both the training design and the test points to the equal-weight
+    synthetic regressor X* = rowMeans(.), then estimates with the NNS
+    increased-dimension trick cbind(X*, X*) under
+    ``method=(1, 2), dim_red_method="equal", order="max", folds=5``. The
+    cross-validated ``n.best`` is what regularises the current engine.
+    """
+    from nns.stack import nns_stack
+
+    x_star = np.asarray(x, dtype=np.float64).mean(axis=1)
+    test_star = np.asarray(test_points, dtype=np.float64).mean(axis=1)
+    result = nns_stack(
+        ivs_train=np.column_stack([x_star, x_star]),
+        dv_train=y,
+        ivs_test=np.column_stack([test_star, test_star]),
+        method=(1, 2),
+        dim_red_method="equal",
+        status=False,
+        order="max",
+        folds=5,
+        ncores=1,
+        dist=None,
+    )
+    return np.asarray(result["stack"], dtype=np.float64).reshape(-1)
+
+
+def _r_seq(start: float, stop: float, by: float) -> NDArray[np.float64]:
+    """Reproduce R's seq(start, stop, by)."""
+    if not np.isfinite(by) or by == 0.0:
+        return np.asarray([start], dtype=np.float64)
+    count = int(np.floor((stop - start) / by + 1e-9))
+    return start + by * np.arange(0, count + 1, dtype=np.float64)
+
+
+def _v057_bandwidths(x_size: int, wrt_dependence: float) -> NDArray[np.float64]:
+    """h_s = 1/log(length(x), c(2, 10)); c(h_s, 10*h_s); doubled if dependence < .5."""
+    base = np.log(np.asarray([2.0, 10.0])) / np.log(float(x_size))
+    h_s = np.concatenate([base, 10.0 * base])
+    if wrt_dependence < 0.5:
+        h_s = 2.0 * h_s
+    return h_s
+
+
+def _dydx_step(column: NDArray[np.float64], eval_value: float, band: float) -> float:
+    """Locally-adaptive dy.dx step: VaR(p + H, 1, x) - VaR(p - H, 1, x), p = CDF(eval)."""
+    from nns.core import lpm_ratio
+    from nns.var import lpm_var
+
+    p = float(lpm_ratio(1.0, float(eval_value), column))
+    upper = lpm_var(min(1.0, p + band), 1.0, column)
+    lower = lpm_var(max(0.0, p - band), 1.0, column)
+    return float(upper - lower)
+
+
+def _dy_d_best_scalar(
+    x_values: NDArray[np.float64],
+    y_values: NDArray[np.float64],
+    wrt_index: int,
+    eval_points: str | float | NDArray[np.float64],
+    *,
+    mixed: bool,
+    messages: bool,
+    wrt_label: int,
+) -> dict[str, NDArray[np.float64]]:
+    from nns.dependence import nns_dep
+    from nns.var import lpm_var
+
+    _n_rows, n_predictors = x_values.shape
+    if wrt_index < 0 or wrt_index >= n_predictors:
+        raise ValueError("`wrt` must select exactly one column of the expanded predictor matrix.")
+    if n_predictors != 2:
+        mixed = False
+
+    if messages:
+        print(
+            "Currently generating NNS.reg finite difference estimates...Regressor "
+            f"{wrt_label}\r"
+        )
+
+    eval_values, vector_branch = _dy_d_eval_points(x_values, wrt_index, eval_points)
+
+    column = x_values[:, wrt_index]
+    dependence = float(nns_dep(column, y_values)["Dependence"])
+    h_s = _v057_bandwidths(x_values.size, dependence)
+
+    firsts: list[NDArray[np.float64]] = []
+    seconds: list[NDArray[np.float64]] = []
+    mixeds: list[NDArray[np.float64]] = []
+
+    if vector_branch:
+        eval_vec = np.asarray(eval_values, dtype=np.float64).reshape(-1)
+        grid = np.column_stack(
+            [
+                np.asarray(
+                    [lpm_var(float(p), 0.0, x_values[:, col]) for p in _r_seq(0.0, 1.0, 0.05)],
+                    dtype=np.float64,
+                )
+                for col in range(n_predictors)
+            ]
+        )
+        sample_size = grid.shape[0]
+        k = eval_vec.size
+        for band in h_s:
+            steps = np.array([_dydx_step(column, ev, float(band)) for ev in eval_vec])
+            blocks: list[NDArray[np.float64]] = []
+            position: list[str] = []
+            ids: list[int] = []
+            for g in range(k):
+                lower = grid.copy()
+                middle = grid.copy()
+                upper = grid.copy()
+                lower[:, wrt_index] = eval_vec[g] - steps[g]
+                middle[:, wrt_index] = eval_vec[g]
+                upper[:, wrt_index] = eval_vec[g] + steps[g]
+                blocks.extend((lower, middle, upper))
+                position.extend(["l"] * sample_size + ["m"] * sample_size + ["u"] * sample_size)
+                ids.extend([g] * (3 * sample_size))
+            estimates = _dy_d_best_reg_estimates(x_values, y_values, np.vstack(blocks))
+            pos = np.asarray(position, dtype=object)
+            idx = np.asarray(ids)
+            band_first = np.empty(k)
+            band_second = np.empty(k)
+            for g in range(k):
+                lo = np.mean(estimates[(pos == "l") & (idx == g)])
+                mid = np.mean(estimates[(pos == "m") & (idx == g)])
+                up = np.mean(estimates[(pos == "u") & (idx == g)])
+                h = steps[g]
+                if np.isfinite(h) and h != 0.0:
+                    band_first[g] = (up - lo) / (2.0 * h)
+                    band_second[g] = (up - 2.0 * mid + lo) / (h**2)
+                else:
+                    band_first[g] = np.nan
+                    band_second[g] = np.nan
+            firsts.append(band_first)
+            seconds.append(band_second)
+        mixed_eval: NDArray[np.float64] | None = None
+    else:
+        eval_mat = _as_eval_matrix(eval_values, n_predictors)
+        n_eval = eval_mat.shape[0]
+        for band in h_s:
+            steps = np.array(
+                [_dydx_step(column, eval_mat[i, wrt_index], float(band)) for i in range(n_eval)]
+            )
+            finite = np.isfinite(steps) & (steps != 0.0)
+            lower = eval_mat.copy()
+            upper = eval_mat.copy()
+            lower[:, wrt_index] = eval_mat[:, wrt_index] - steps
+            upper[:, wrt_index] = eval_mat[:, wrt_index] + steps
+            if messages:
+                print(
+                    "Currently generating NNS.reg finite difference estimates...bandwidth\r"
+                )
+            estimates = _dy_d_best_reg_estimates(
+                x_values, y_values, np.vstack((lower, eval_mat, upper))
+            )
+            lo = estimates[:n_eval]
+            mid = estimates[n_eval : 2 * n_eval]
+            up = estimates[2 * n_eval :]
+            with np.errstate(invalid="ignore", divide="ignore"):
+                first = (up - lo) / (2.0 * steps)
+                second = (up - 2.0 * mid + lo) / (steps**2)
+            first[~finite] = np.nan
+            second[~finite] = np.nan
+            firsts.append(first)
+            seconds.append(second)
+        mixed_eval = eval_mat
+
+    def _row_nanmean(bands: list[NDArray[np.float64]]) -> NDArray[np.float64]:
+        matrix = np.column_stack([np.asarray(b, dtype=np.float64).reshape(-1) for b in bands])
+        with np.errstate(invalid="ignore"):
+            return np.asarray(np.nanmean(matrix, axis=1), dtype=np.float64)
+
+    output = {"First": _row_nanmean(firsts), "Second": _row_nanmean(seconds)}
+
+    if mixed:
+        if vector_branch:
+            tuple_eval = np.asarray(eval_values, dtype=np.float64).reshape(-1)
+            if tuple_eval.size != 2:
+                raise ValueError("Mixed Derivatives are only for 2 IV")
+            mixed_points = tuple_eval.reshape(1, 2)
+        else:
+            assert mixed_eval is not None
+            if mixed_eval.shape[1] != 2:
+                raise ValueError("Mixed Derivatives are only for 2 IV")
+            mixed_points = mixed_eval
+        for band in h_s:
+            band_vals: list[float] = []
+            for point in mixed_points:
+                s1 = _dydx_step(x_values[:, 0], point[0], float(band))
+                s2 = _dydx_step(x_values[:, 1], point[1], float(band))
+                if not (np.isfinite(s1) and np.isfinite(s2) and s1 != 0.0 and s2 != 0.0):
+                    band_vals.append(np.nan)
+                    continue
+                corners = np.array(
+                    [
+                        [point[0] + s1, point[1] + s2],
+                        [point[0] - s1, point[1] + s2],
+                        [point[0] + s1, point[1] - s2],
+                        [point[0] - s1, point[1] - s2],
+                    ]
+                )
+                z = _dy_d_best_reg_estimates(x_values, y_values, corners)
+                band_vals.append((z[0] + z[3] - z[1] - z[2]) / (4.0 * s1 * s2))
+            mixeds.append(np.asarray(band_vals, dtype=np.float64))
+        output["Mixed"] = _row_nanmean(mixeds)
+
+    if messages:
+        print("\r")
+    return output
+
+
 def _dy_d_scalar(
     x_values: NDArray[np.float64],
     y_values: NDArray[np.float64],
diff --git a/tests/invariants/test_diff.py b/tests/invariants/test_diff.py
@@ -115,3 +115,80 @@ def test_dy_d_point_modes_preserve_linear_slope_direction() -> None:
     for eval_points in ("mean", "median", "last"):
         assert dy_d(x, positive, wrt=1, eval_points=eval_points)["First"][0] > 0.0
         assert dy_d(x, negative, wrt=1, eval_points=eval_points)["First"][0] < 0.0
+
+
+def test_dy_d_best_is_exported() -> None:
+    import nns
+
+    assert "dy_d_best" in nns.__all__
+    assert callable(nns.dy_d_best)
+
+
+def test_dy_d_best_vectorized_wrt_tuple_returns_first_second() -> None:
+    from nns import dy_d_best
+
+    x = np.random.RandomState(0).randn(60, 3)
+    y = x[:, 0] + 2.0 * x[:, 1] - x[:, 2]
+
+    result = dy_d_best(x, y, wrt=np.array([1, 2, 3]), eval_points=np.zeros((1, 3)))
+
+    assert result.keys() == {"First", "Second"}
+    assert result["First"].shape == (1, 3)
+    assert result["Second"].shape == (1, 3)
+    assert np.all(np.isfinite(result["First"]))
+
+
+def test_dy_d_best_single_wrt_returns_one_dimensional_first() -> None:
+    from nns import dy_d_best
+
+    x = np.random.RandomState(0).randn(60, 3)
+    y = x[:, 0] + 2.0 * x[:, 1] - x[:, 2]
+
+    # A single `wrt` returns 1-D arrays (one value per evaluation point).
+    result = dy_d_best(x, y, wrt=1, eval_points=np.array([[0.0, 0.0, 0.0]]))
+
+    assert result["First"].shape == (1,)
+    assert np.all(np.isfinite(result["First"]))
+
+
+def test_dy_d_best_two_column_mixed_returns_mixed() -> None:
+    from nns import dy_d_best
+
+    x = np.random.RandomState(1).randn(60, 2)
+    y = x[:, 0] * x[:, 1]
+
+    result = dy_d_best(x, y, wrt=np.array([1, 2]), eval_points=np.array([[0.0, 0.0]]), mixed=True)
+
+    assert result.keys() == {"First", "Second", "Mixed"}
+    assert result["First"].shape == (1, 2)
+    assert result["Mixed"].shape == (1, 2)
+
+
+def test_dy_d_best_is_deterministic() -> None:
+    from nns import dy_d_best
+
+    x = np.random.RandomState(0).randn(60, 3)
+    y = x[:, 0] + 2.0 * x[:, 1] - x[:, 2]
+
+    a = dy_d_best(x, y, wrt=np.array([1, 2, 3]), eval_points=np.zeros((1, 3)))
+    b = dy_d_best(x, y, wrt=np.array([1, 2, 3]), eval_points=np.zeros((1, 3)))
+
+    assert np.allclose(a["First"], b["First"])
+    assert np.allclose(a["Second"], b["Second"])
+
+
+def test_dy_d_best_matches_pinned_values() -> None:
+    # Golden values guard the v0.5.7 finite-difference design against regressions.
+    from nns import dy_d_best
+
+    x = np.random.RandomState(0).randn(60, 3)
+    y = x[:, 0] + 2.0 * x[:, 1] - x[:, 2]
+
+    result = dy_d_best(x, y, wrt=np.array([1, 2, 3]), eval_points=np.zeros((1, 3)))
+
+    np.testing.assert_allclose(
+        np.ravel(result["First"]),
+        np.array([0.754963, 0.792164, 0.699243]),
+        rtol=0.0,
+        atol=1e-4,
+    )
