ENH(mutation): add global independence proposals

CHANGELOG.md

@@ -6,6 +6,11 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Added
+
+- Global proposals with `IndependentProposalKernel`
+  ([#139](https://github.com/sequince-dev/sequince/pull/139)).
+
 ## [0.1.0b1] - 2026-10-23
 
 First public beta release.

docs/user_guide/sampling_options/mutation_configuration.rst

@@ -45,7 +45,8 @@ Each step takes a target callable or :class:`~sequince.target.Target` object.
 
 MH steps take a proposal through the ``kernel`` argument. See
 :mod:`sequince.steps.mutation.kernels` for Gaussian random walks, masked
-updates, mixtures, and differential-evolution kernels. Red-blue MH requires a
+updates, global proposals, mixtures, and differential-evolution kernels.
+Red-blue MH requires a
 :class:`~sequince.steps.mutation.kernels.ComplementaryEnsembleKernel`.
 
 Common configuration

src/sequince/array_api.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from collections.abc import Callable
 from typing import Any
 
 import numpy as _np
@@ -23,7 +24,7 @@
 from ._scipy_compat import enable_scipy_array_api as _enable_scipy_array_api
 from .typing import Array
 
-IS_NAMESPACE_FUNCTIONS = {
+IS_NAMESPACE_FUNCTIONS: dict[str, Callable[[Any], bool]] = {
     "numpy": is_numpy_namespace,
     "torch": is_torch_namespace,
     "jax": is_jax_namespace,

src/sequince/steps/__init__.py

@@ -38,6 +38,7 @@
     ESJDMetric,
     ESJDScaleWeightPolicy,
     FixedLengthMutation,
+    IndependentProposalKernel,
     KernelSample,
     MaskedGaussianRandomWalk,
     MetropolisBlock,
@@ -124,6 +125,7 @@
     "ErynMutation",
     "FixedLengthMutation",
     "MaskedGaussianRandomWalk",
+    "IndependentProposalKernel",
     "KernelSample",
     "MetropolisHastingsMutation",
     "RedBlueMetropolisHastingsMutation",

src/sequince/steps/mutation/__init__.py

@@ -37,6 +37,7 @@
     ComplementaryEnsembleKernel,
     DifferentialEvolutionKernel,
     GaussianRandomWalk,
+    IndependentProposalKernel,
     KernelSample,
     MaskedGaussianRandomWalk,
     MixtureMutationKernel,
@@ -98,6 +99,7 @@
     "RedBlueMetropolisHastingsMutation",
     "MiniPCNMutation",
     "MaskedGaussianRandomWalk",
+    "IndependentProposalKernel",
     "KernelSample",
     "MetropolisBlock",
     "MixtureMutationKernel",

src/sequince/steps/mutation/kernels.py

@@ -17,6 +17,8 @@
 
 if TYPE_CHECKING:
     from orng import RandomGenerator
+
+    from ...proposals import ProposalDistribution
 else:  # pragma: no cover - runtime fallback
     from orng import RandomGenerator  # type: ignore
 
@@ -321,6 +323,81 @@ def evaluate_log_transition_ratio(
     )
 
 
+@dataclass(frozen=True)
+class IndependentProposalKernel:
+    """Independence MH proposals in the coordinates supplied to the kernel.
+
+    Parameters
+    ----------
+    proposal : ProposalDistribution
+        Distribution defining samples and densities in mutation coordinates.
+        A physical-space prior is suitable only without a coordinate transform.
+        Sample batches must have exactly the input particle shape; densities
+        must have shape (n_particles,). Sampled densities are retained locally.
+    """
+
+    proposal: ProposalDistribution
+
+    def sample(self, particles: Array, rng: RandomGenerator) -> KernelSample:
+        """Draw a batch and retain its log densities for the MH correction."""
+        xp = namespace(particles)
+        values, density = self.proposal.sample_and_log_prob(
+            particles.shape[0], rng=rng, xp=xp
+        )
+        values = asarray(values, xp=xp, like=particles, dtype=particles.dtype)
+        density = asarray(
+            density, xp=xp, like=particles, dtype=particles.dtype
+        )
+        if values.shape != particles.shape:
+            raise ValueError("Proposal samples must match the particle shape.")
+        if density.shape != (particles.shape[0],):
+            raise ValueError(
+                "Proposal densities must have shape (n_particles,)."
+            )
+        return KernelSample(values, {"log_proposal": density})
+
+    def log_transition_ratio(
+        self, from_particles: Array, to_particles: Array
+    ) -> Array:
+        """Return ``log q(current) - log q(proposed)`` per particle."""
+        if from_particles.shape != to_particles.shape:
+            raise ValueError("Proposal samples must match the particle shape.")
+        return self.log_transition_ratio_from_sample(
+            from_particles,
+            KernelSample(
+                to_particles,
+                {"log_proposal": self.proposal.log_prob(to_particles)},
+            ),
+        )
+
+    def log_transition_ratio_from_sample(
+        self, from_particles: Array, sample: KernelSample
+    ) -> Array:
+        """Evaluate the correction using the cached proposed log densities."""
+        if from_particles.shape != sample.particles.shape:
+            raise ValueError("Proposal samples must match the particle shape.")
+        xp = namespace(from_particles)
+        current = asarray(
+            self.proposal.log_prob(from_particles),
+            xp=xp,
+            like=from_particles,
+            dtype=from_particles.dtype,
+        )
+        proposed = asarray(
+            sample.auxiliary["log_proposal"],
+            xp=xp,
+            like=from_particles,
+            dtype=from_particles.dtype,
+        )
+        if current.shape != (from_particles.shape[0],) or (
+            proposed.shape != current.shape
+        ):
+            raise ValueError(
+                "Proposal densities must have shape (n_particles,)."
+            )
+        return current - proposed
+
+
 @dataclass(frozen=True)
 class GaussianRandomWalk(SymmetricMutationKernel):
     """Isotropic Gaussian random-walk mutation kernel."""
@@ -882,6 +959,7 @@ def log_transition_ratio_from_sample(
 
 MUTATION_KERNELS: dict[str, type] = {
     "gaussian-random-walk": GaussianRandomWalk,
+    "independent-proposal": IndependentProposalKernel,
     "mixture": MixtureMutationKernel,
 }
 

tests/test_mutation/test_independent_proposal.py

@@ -0,0 +1,161 @@
+from __future__ import annotations
+
+from typing import Any
+
+import numpy as np
+import pytest
+from array_api_compat import device as get_device
+from orng import RandomGenerator
+
+from sequince import ParticleState
+from sequince.array_api import namespace, to_numpy
+from sequince.steps import (
+    IndependentProposalKernel,
+    MetropolisHastingsMutation,
+    StepContext,
+)
+from sequince.steps.mutation.kernels import (
+    evaluate_log_transition_ratio,
+    get_mutation_kernel,
+)
+from sequince.typing import Array
+from sequince.utils.random import resolve_rng
+
+
+class NormalProposal:
+    def sample(
+        self,
+        n_samples: int,
+        context: StepContext | None = None,
+        *,
+        rng: RandomGenerator | int | None = None,
+        xp: Any | None = None,
+    ) -> Array:
+        return resolve_rng(rng).normal(size=(n_samples, 2)) * 2
+
+    def log_prob(
+        self, particles: Array, context: StepContext | None = None
+    ) -> Array:
+        xp = namespace(particles)
+        return -xp.sum(particles**2, axis=-1) / 8
+
+    def sample_and_log_prob(
+        self,
+        n_samples: int,
+        context: StepContext | None = None,
+        *,
+        rng: RandomGenerator | int | None = None,
+        xp: Any | None = None,
+    ) -> tuple[Array, Array]:
+        values = self.sample(n_samples, rng=rng, xp=xp)
+        return values, self.log_prob(values)
+
+
+def test_independence_backend(backend, backend_factory) -> None:
+    case = backend_factory(backend)
+    particles = case.particles
+    kernel = IndependentProposalKernel(NormalProposal())
+    sample = kernel.sample(particles, case.rng(5))
+    ratio = evaluate_log_transition_ratio(kernel, particles, sample)
+    assert sample.particles.shape == particles.shape
+    assert sample.particles.dtype == particles.dtype
+    assert get_device(sample.particles) == get_device(particles)
+    assert case.matches_namespace(sample.particles)
+    np.testing.assert_allclose(
+        to_numpy(ratio),
+        to_numpy(
+            kernel.proposal.log_prob(particles)
+            - kernel.proposal.log_prob(sample.particles)
+        ),
+        rtol=1e-6,
+        atol=1e-6,
+    )
+    with pytest.raises(ValueError, match="shape"):
+        kernel.log_transition_ratio(particles, sample.particles[:1])
+
+
+def test_support_and_validation() -> None:
+    class Bounded(NormalProposal):
+        def log_prob(self, particles, context=None):
+            return np.where(particles[:, 0] >= 0, -particles[:, 0], -np.inf)
+
+    kernel = IndependentProposalKernel(Bounded())
+    ratio = kernel.log_transition_ratio(
+        np.array([[-1.0, 0], [1, 0]]), np.array([[1.0, 0], [-1, 0]])
+    )
+    np.testing.assert_array_equal(ratio, [-np.inf, np.inf])
+    with pytest.raises(ValueError, match="shape"):
+        kernel.sample(np.zeros((3, 1)), RandomGenerator(seed=5))
+
+    class BadDensity(NormalProposal):
+        def log_prob(self, particles, context=None):
+            return np.zeros((len(particles), 1))
+
+    with pytest.raises(ValueError, match="densities"):
+        IndependentProposalKernel(BadDensity()).sample(
+            np.zeros((3, 2)), RandomGenerator(seed=5)
+        )
+
+
+@pytest.mark.parametrize("shape", [(5,), (1, 2), (5, 2, 3)])
+@pytest.mark.parametrize("dtype", [np.float32, np.float64])
+def test_independence_particle_shapes(shape, dtype) -> None:
+    class ShapedProposal(NormalProposal):
+        def sample(self, n_samples, context=None, *, rng=None, xp=None):
+            return rng.normal(size=(n_samples, *shape[1:]))
+
+        def log_prob(self, particles, context=None):
+            return (
+                -np.sum(
+                    np.reshape(particles, (len(particles), -1)) ** 2, axis=1
+                )
+                / 2
+            )
+
+    particles = np.ones(shape, dtype=dtype)
+    kernel = IndependentProposalKernel(ShapedProposal())
+    sample = kernel.sample(particles, RandomGenerator(seed=3))
+    assert sample.particles.shape == shape
+    assert sample.particles.dtype == dtype
+    ratio = evaluate_log_transition_ratio(kernel, particles, sample)
+    assert ratio.shape == (shape[0],)
+    assert ratio.dtype == dtype
+
+
+def test_cached_density() -> None:
+    class CountedProposal(NormalProposal):
+        def __init__(self) -> None:
+            self.calls = []
+
+        def log_prob(self, particles, context=None):
+            self.calls.append(particles)
+            return super().log_prob(particles, context)
+
+    proposal = CountedProposal()
+    independent = IndependentProposalKernel(proposal)
+    particles = np.ones((10, 2))
+    rng = RandomGenerator(seed=22)
+    sample = independent.sample(particles, rng)
+    evaluate_log_transition_ratio(independent, particles, sample)
+    assert len(proposal.calls) == 2  # Sampled density plus current density.
+    assert proposal.calls[-1] is particles
+
+
+def test_known_target(context_factory) -> None:
+    rng = RandomGenerator(seed=44)
+    state = ParticleState(rng.normal(size=(1500, 2)) * 4)
+    step = MetropolisHastingsMutation(
+        lambda x: -np.sum(x * x, axis=-1) / 2,
+        IndependentProposalKernel(NormalProposal()),
+        n_steps=100,
+    )
+    state, _, _ = step(state, context_factory(), rng)
+    np.testing.assert_allclose(np.mean(state.particles, axis=0), 0, atol=0.10)
+    np.testing.assert_allclose(np.var(state.particles, axis=0), 1, atol=0.12)
+
+
+def test_factory() -> None:
+    proposal = NormalProposal()
+    kernel = get_mutation_kernel("independent-proposal", proposal=proposal)
+    assert isinstance(kernel, IndependentProposalKernel)
+    assert kernel.proposal is proposal