omnibias.torch.confluence

0.5.0rc1Release candidatePython >=3.10Apache-2.0

PyTorch activation derivatives, trainable operator layers, network jets, and direct differential-operator contractions.

import omnibias.torch.confluence

Module contract & mathematical conventions

Stable centered collision atoms, live through the attained rho=0 boundary.

Finite formulas are exact in real arithmetic. For abs(h*eta)<=series_radius an explicit even Taylor polynomial is evaluated (series_terms coefficients). Its remainder is bounded by the shared core confluence derivative bounds. The inactive finite branch uses positive denominators so backward/JIT are safe.

Shared formulas do not imply universal floating-point identity. Verify dtype, shape, derivative-order support, and the stated numerical assumptions for the path you use.

Module source ↗Source revision 9a2d73df0573Running reference examples →

CLASS

CenteredPair

omnibias.torch.confluence.CenteredPairSource ↗

class CenteredPair(NamedTuple)

The module contract describes the context for this operation. Its signature, explicit validation, return expressions, and implementation are recorded below.

Fields & methods

averagefield

average: Tensor

Declared Tensor field on CenteredPair. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

divided_differencefield

divided_difference: Tensor

Declared Tensor field on CenteredPair. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

Read the implementation

class CenteredPair(NamedTuple):
    average: Tensor
    divided_difference: Tensor

FUNCTION

centered_pair

omnibias.torch.confluence.centered_pairSource ↗

centered_pair(z: Tensor, rho: Tensor, eta: Tensor, *, activation: str='sigmoid', series_terms: int=6, series_radius: float=0.01, tower: Tensor | Sequence[Tensor] | None=None) -> CenteredPair

A=(sigma(z+h*eta)+sigma(z-h*eta))/2, B=odd difference/(2h).

rho=h**2 is nonnegative (projected in this function). At rho=0, A=sigma(z) and B=eta*sigma'(z). Includes live derivatives in rho, z, and eta. Sigmoid and tanh have stable finite-spread formulas; other activations are rejected. The integer series_terms and radius are static compilation parameters.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`rho`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`eta`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`activation`keyword-only	`str`Default: 'sigmoid'	See the operation contract and implementation below.
`series_terms`keyword-only	`int`Default: 6	See the operation contract and implementation below.
`series_radius`keyword-only	`float`Default: 0.01	See the operation contract and implementation below.
`tower`keyword-only	`Tensor \| Sequence[Tensor] \| None`Default: None	See the operation contract and implementation below.

ReturnsCenteredPair

Explicit validation & failure cases

When activation not in ('sigmoid', 'tanh'), raises ValueError('centered_pair supports sigmoid and tanh').
When series_terms < 1 or not 0 < series_radius <= 0.1, raises ValueError('positive terms and radius in (0,0.1] required').
When spec.fastpath is None, raises ValueError('activation requires a derivative provider').
When len(tower) <= maximum, raises ValueError('supplied tower is too short for the centered series').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Regression example · tails and safe inactive branches

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_tails_and_safe_inactive_branches():
    z = torch.tensor([-80.0, -20.0, 20.0, 80.0], dtype=torch.float64, requires_grad=True)
    rho = torch.tensor([0.0, 1e-30, 1.0, 1e20], dtype=torch.float64, requires_grad=True)
    a, b = tpair(z, rho, torch.ones_like(z))
    assert torch.isfinite(a).all() and torch.isfinite(b).all()
    for derivative in torch.autograd.grad((a + b).sum(), (z, rho)):
        assert torch.isfinite(derivative).all()

python -m pytest tests/test_confluence_parity.py -k test_tails_and_safe_inactive_branches -q

Regression example · fourth spread derivative at boundary

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_fourth_spread_derivative_at_boundary():
    z = torch.tensor(0.2, dtype=torch.float64)
    p = torch.tensor(0.0, dtype=torch.float64, requires_grad=True)
    eta = torch.tensor(0.7, dtype=torch.float64)
    a, b = tpair(z, p, eta)
    value = a + b
    for _ in range(4):
        value = torch.autograd.grad(value, p, create_graph=True)[0]
    zz = z.clone().requires_grad_()
    ds = [torch.sigmoid(zz)]
    for _ in range(9):
        ds.append(torch.autograd.grad(ds[-1], zz, create_graph=True)[0])
    expected = math.factorial(4) * (
        ds[8] * eta**8 / math.factorial(8) + ds[9] * eta**9 / math.factorial(9)
    )
    torch.testing.assert_close(value, expected, rtol=1e-10, atol=1e-12)

python -m pytest tests/test_confluence_parity.py -k test_fourth_spread_derivative_at_boundary -q

Read the implementation

def centered_pair(
    z: Tensor, rho: Tensor, eta: Tensor, *, activation: str = "sigmoid",
    series_terms: int = 6, series_radius: float = 0.01,
    tower: Tensor | Sequence[Tensor] | None = None,
) -> CenteredPair:
    """A=(sigma(z+h*eta)+sigma(z-h*eta))/2, B=odd difference/(2h).

    rho=h**2 is nonnegative (projected in this function). At rho=0, A=sigma(z)
    and B=eta*sigma'(z). Includes live derivatives in rho, z, and eta. Sigmoid
    and tanh have stable finite-spread formulas; other activations are rejected.
    The integer series_terms and radius are static compilation parameters.
    """
    if activation not in ("sigmoid", "tanh"):
        raise ValueError("centered_pair supports sigmoid and tanh")
    if series_terms < 1 or not 0 < series_radius <= 0.1:
        raise ValueError("positive terms and radius in (0,0.1] required")
    spec = get_activation(activation)
    if spec.fastpath is None:
        raise ValueError("activation requires a derivative provider")
    p = retract_spread(rho)
    small = p * eta**2 <= series_radius**2
    safe_rho = torch.where(small, torch.ones_like(p), p)
    h = torch.sqrt(safe_rho)
    u = torch.abs(h * eta)
    u = torch.where(small, torch.ones_like(u), u)
    # Three sigmoid evaluations in the finite path; no subtraction of nearby
    # activation values. tanh(z)=2*sigmoid(2*z)-1 supplies the same formula.
    scale = 1.0 if activation == "sigmoid" else 2.0
    zz = scale * z
    uu = scale * u
    splus = torch.sigmoid(zz + uu)
    sminus = torch.sigmoid(zz - uu)
    complement = torch.sigmoid(-zz + uu)
    a_finite = (splus + sminus) * 0.5
    b_finite = eta * scale * splus * complement * (-torch.expm1(-2 * uu) / (2 * uu))
    if activation == "tanh":
        a_finite = 2 * a_finite - 1
        b_finite = 2 * b_finite
    # Evaluate the near branch at safe spread even when it is inactive: huge
    # finite spreads must not overflow powers in a discarded Taylor polynomial.
    r = torch.where(small, p, torch.zeros_like(p))
    e = torch.where(small, eta, torch.zeros_like(eta))
    maximum = 2 * series_terms - 1
    if tower is None:
        tower = (spec.tower(z, maximum) if spec.tower is not None
                 else tuple(spec.fastpath(z, k) for k in range(maximum + 1)))
    if len(tower) <= maximum:
        raise ValueError("supplied tower is too short for the centered series")
    a_series = torch.zeros_like(z)
    b_series = torch.zeros_like(z)
    for k in range(series_terms):
        a_series = a_series + tower[2*k] * (r**k * e**(2*k)) / factorial(2*k)
        b_series = b_series + tower[2*k+1] * (r**k * e**(2*k+1)) / factorial(2*k+1)
    return CenteredPair(torch.where(small, a_series, a_finite),
                        torch.where(small, b_series, b_finite))

FUNCTION

collision_atom

omnibias.torch.confluence.collision_atomSource ↗

collision_atom(z: Tensor, rho: Tensor, eta: Tensor, m0: Tensor, m1: Tensor, *, activation: str='sigmoid', series_terms: int=6, tower: Tensor | Sequence[Tensor] | None=None) -> Tensor

m0*A + m1*B; moments remain bounded as a pair becomes a derivative atom.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`rho`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`eta`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`m0`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`m1`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`activation`keyword-only	`str`Default: 'sigmoid'	See the operation contract and implementation below.
`series_terms`keyword-only	`int`Default: 6	See the operation contract and implementation below.
`tower`keyword-only	`Tensor \| Sequence[Tensor] \| None`Default: None	See the operation contract and implementation below.

ReturnsTensor

Read the implementation

def collision_atom(z: Tensor, rho: Tensor, eta: Tensor, m0: Tensor, m1: Tensor,
                   *, activation: str = "sigmoid", series_terms: int = 6,
                   tower: Tensor | Sequence[Tensor] | None = None) -> Tensor:
    """m0*A + m1*B; moments remain bounded as a pair becomes a derivative atom."""
    pair = centered_pair(z, rho, eta, activation=activation, series_terms=series_terms, tower=tower)
    return m0 * pair.average + m1 * pair.divided_difference

FUNCTION

moment_atom

omnibias.torch.confluence.moment_atomSource ↗

moment_atom(z: Tensor, moments: Tensor, *, activation: str='sigmoid', tower: Tensor | Sequence[Tensor] | None=None) -> Tensor

A truncated common-weight bias cluster, sum(m[k]*sigma**(k)(z)).

The host initialize_moments/cluster_remainder API supplies its error bound. This is exact only when the selected atom architecture is itself the model.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`moments`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`activation`keyword-only	`str`Default: 'sigmoid'	See the operation contract and implementation below.
`tower`keyword-only	`Tensor \| Sequence[Tensor] \| None`Default: None	See the operation contract and implementation below.

ReturnsTensor

Explicit validation & failure cases

When moments.ndim != 1 or moments.shape[0] < 1, raises ValueError('moments must be a nonempty vector').
When spec.fastpath is None, raises ValueError('activation requires a derivative provider').
When len(tower) <= n, raises ValueError('supplied tower is too short for the moment expansion').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Read the implementation

def moment_atom(z: Tensor, moments: Tensor, *, activation: str = "sigmoid",
                tower: Tensor | Sequence[Tensor] | None = None) -> Tensor:
    """A truncated common-weight bias cluster, sum(m[k]*sigma**(k)(z)).

    The host initialize_moments/cluster_remainder API supplies its error bound.
    This is exact only when the selected atom architecture is itself the model.
    """
    if moments.ndim != 1 or moments.shape[0] < 1:
        raise ValueError("moments must be a nonempty vector")
    spec = get_activation(activation)
    if spec.fastpath is None:
        raise ValueError("activation requires a derivative provider")
    n = moments.shape[0] - 1
    if tower is None:
        tower = (spec.tower(z, n) if spec.tower is not None
                 else tuple(spec.fastpath(z, k) for k in range(n+1)))
    if len(tower) <= n:
        raise ValueError("supplied tower is too short for the moment expansion")
    out = torch.zeros_like(z)
    for k in range(n+1):
        out = out + moments[k] * tower[k]
    return out

FUNCTION

polynomial_pair

omnibias.torch.confluence.polynomial_pairSource ↗

polynomial_pair(z: Tensor, rho: Tensor, eta: Tensor, coefficients: Tensor) -> CenteredPair

Exact finite polynomial control in the same centered coordinates.

Coefficients are live and in ascending powers. No finite-spread truncation occurs: the even/odd binomial expansion terminates at the supplied degree.

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`rho`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`eta`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.
`coefficients`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.

ReturnsCenteredPair

Explicit validation & failure cases

When coefficients.ndim != 1 or coefficients.shape[0] < 1, raises ValueError('nonempty polynomial coefficient vector required').

These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.

Regression example · polynomial pair rejects invalid coefficient shape

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_polynomial_pair_rejects_invalid_coefficient_shape():
    for coefficients in (np.empty(0), np.ones((2, 2))):
        with pytest.raises(ValueError, match="coefficient"):
            torch_pair(
                torch.tensor(0.0), torch.tensor(0.0), torch.tensor(1.0), torch.tensor(coefficients)
            )
        with pytest.raises(ValueError, match="coefficient"):
            jax_pair(
                jnp.asarray(0.0), jnp.asarray(0.0), jnp.asarray(1.0), jnp.asarray(coefficients)
            )

python -m pytest tests/test_polynomial_pair.py -k test_polynomial_pair_rejects_invalid_coefficient_shape -q

Regression example · finite polynomial pair matches defining pair without truncation

This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.

Full file, imports & fixtures ↗

def test_finite_polynomial_pair_matches_defining_pair_without_truncation():
    coefficients = np.array([0.125, -0.5, 0.25, 0.75, -0.125, 0.5, 0.25, -0.125])
    polynomial = Polynomial(coefficients)
    z = np.array([-0.5, 0.0, 0.25])
    eta = np.array([-0.75, 1.0, 0.5])
    h = 0.25
    plus, minus = polynomial(z + h * eta), polynomial(z - h * eta)
    a, b = torch_pair(
        torch.tensor(z), torch.tensor(h * h), torch.tensor(eta), torch.tensor(coefficients)
    )
    np.testing.assert_allclose(a, (plus + minus) / 2, atol=1e-15)
    np.testing.assert_allclose(b, (plus - minus) / (2 * h), atol=1e-15)

python -m pytest tests/test_polynomial_pair.py -k test_finite_polynomial_pair_matches_defining_pair_without_truncation -q

Read the implementation

def polynomial_pair(z: Tensor, rho: Tensor, eta: Tensor, coefficients: Tensor) -> CenteredPair:
    """Exact finite polynomial control in the same centered coordinates.

    Coefficients are live and in ascending powers. No finite-spread truncation
    occurs: the even/odd binomial expansion terminates at the supplied degree.
    """
    from math import comb
    if coefficients.ndim != 1 or coefficients.shape[0] < 1:
        raise ValueError('nonempty polynomial coefficient vector required')
    p = retract_spread(rho)
    a, b = torch.zeros_like(z), torch.zeros_like(z)
    for degree in range(coefficients.shape[0]):
        for power in range(degree+1):
            term = coefficients[degree]*comb(degree,power)*z**(degree-power)
            if power % 2:
                b = b+term*p**((power-1)//2)*eta**power
            else:
                a = a+term*p**(power//2)*eta**power
    return CenteredPair(a,b)

FUNCTION

retract_spread

omnibias.torch.confluence.retract_spreadSource ↗

retract_spread(rho: Tensor) -> Tensor

Project onto rho>=0; call between steps to attain zero exactly.

Parameters

Name	Type & default	Contract
`rho`positional-or-keyword	`Tensor`Required	See the operation contract and implementation below.

ReturnsTensor

Read the implementation

def retract_spread(rho: Tensor) -> Tensor:
    """Project onto rho>=0; call between steps to attain zero exactly."""
    return torch.where(rho >= 0, rho, torch.zeros_like(rho))

← Package overview Runnable learning examples →

omnibias.torch.confluence

CenteredPair​

Fields & methods

centered_pair​

Parameters

collision_atom​

Parameters

moment_atom​

Parameters

polynomial_pair​

Parameters

retract_spread​

Parameters

CenteredPair

centered_pair

collision_atom

moment_atom

polynomial_pair

retract_spread