omnibias.core.refine

0.5.0rc1Release candidatePython >=3.10Apache-2.0

Coefficient recurrences, activation contracts, jet algebra, and verified arithmetic shared by the numerical backends.

import omnibias.core.refine

Module contract & mathematical conventions

Adaptive pack refinement algebra (theory 03-13).

Three structural moves — birth, growth, and death — act on a bank of tempered packs. Birth and growth are **exactly** zero-perturbation (new outer weight c = 0, or a higher-order sibling with c = 0). Death is bounded-perturbation: the API reports ||c_g p_g|| / ||u|| and never pretends the change is zero.

Scale alpha is the founding scaling law sigma_alpha^(n)(u) = alpha^n sigma^(n)(alpha u). That is a third axis, not bias collapse (delta -> 0) and not temperature collapse (beta -> inf via omnibias.core.spec.tempered).

Singularity and scale-flow indicators use the Domb-Sykes / ratio estimators that spec 03-10 and 03-07 need. They are not the full Padé tracker or the RG beta-function; those parent specs stay designed.

This module is backend-free. Tensor evaluation lives in the torch / jax twins. GrowableOperatorMultiBiasUnit.grow remains the literal-K growth move; pack-bank *p*-type growth is the collapsed sibling (add sigma^(n+1) at the same centre with c = 0).

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 →

CONSTANT

HpMove

omnibias.core.refine.HpMoveSource ↗

HpMove = Literal['h', 'p', 'none']

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Read the implementation

HpMove = Literal["h", "p", "none"]

CONSTANT

HpRule

omnibias.core.refine.HpRuleSource ↗

HpRule = Literal['coefficient_decay', 'always_h', 'always_p']

This exported constant is defined by the expression above. Its exact value and annotation come from the pinned source; use it with the module contract.

Read the implementation

HpRule = Literal["coefficient_decay", "always_h", "always_p"]

CLASS

Indicator

omnibias.core.IndicatorSource ↗

class Indicator(str, Enum)

Where-to-refine signal. Scale comes only from jet-aware members.

Regression example · hysteresis blocks weak birth

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_hysteresis_blocks_weak_birth() -> None:
    packs = (RefinedPack(order=0, center=0.5, weight=1.0, scale=2.0),)
    policy = RefinePolicy(
        indicator=Indicator.RESIDUAL,
        birth_threshold=0.1,
        death_threshold=0.05,
        hysteresis=4.0,
    )
    # score 0.15 exceeds birth_threshold but not hysteresis * death = 0.2
    proposal = propose_refinement(
        packs,
        policy=policy,
        peak_location=0.0,
        peak_value=0.15,
        derivatives=None,
        birth_score=0.15,
    )
    assert proposal is None

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_hysteresis_blocks_weak_birth -q

Regression example · g3 scale flow

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_g3_scale_flow() -> None:
    bank = _initial_bank()
    probes = jnp.linspace(0.0, 1.0, 65)

    def residual(x: Array) -> Array:
        return jnp.exp(-100.0 * x)

    _bank, report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.SCALE_FLOW,
            birth_threshold=0.1,
            death_threshold=0.0,
            min_age=10_000,
            hysteresis=1.0,
            min_scale_ratio=1.0,
        ),
        step=0,
        probe_jet=(_BL_CENTER, _BL_DERIVS),
    )
    assert report.proposed_scale == pytest.approx(_BL_SCALE)

python -m pytest packages/omnibias-jax/tests/test_refine.py -k test_g3_scale_flow -q

Read the implementation

class Indicator(str, Enum):
    """Where-to-refine signal. Scale comes only from jet-aware members."""

    RESIDUAL = "residual"
    CERTIFIED_ERROR = "certified_error"
    SINGULARITY = "singularity"
    SCALE_FLOW = "scale_flow"

Other public import paths

omnibias.core.Indicator
omnibias.core.refine.Indicator
omnibias.jax.refine.Indicator
omnibias.torch.refine.Indicator

CLASS

RefinePolicy

omnibias.core.RefinePolicySource ↗

class RefinePolicy

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

Discrete refinement policy. Decisions are not a smooth optimiser.

Parameters

Name	Type & default	Contract
`indicator`dataclass field	`Indicator`Default: Indicator.RESIDUAL	See the operation contract and implementation below.
`birth_threshold`dataclass field	`float`Default: 0.1	See the operation contract and implementation below.
`death_threshold`dataclass field	`float`Default: 0.01	See the operation contract and implementation below.
`min_age`dataclass field	`int`Default: 100	See the operation contract and implementation below.
`max_packs`dataclass field	`int \| None`Default: None	See the operation contract and implementation below.
`hp_rule`dataclass field	`HpRule`Default: 'coefficient_decay'	See the operation contract and implementation below.
`hysteresis`dataclass field	`float`Default: 2.0	See the operation contract and implementation below.
`jet_order`dataclass field	`int`Default: 6	See the operation contract and implementation below.
`default_order`dataclass field	`int`Default: 0	See the operation contract and implementation below.
`min_center_separation`dataclass field	`float`Default: 0.0	See the operation contract and implementation below.
`min_scale_ratio`dataclass field	`float`Default: 1.5	See the operation contract and implementation below.
`debug_zero_perturbation`dataclass field	`bool`Default: False	See the operation contract and implementation below.

Executed regression · min age protects newborn

Recorded coverage confirms this test executed the API body. 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_min_age_protects_newborn() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=1.0, age=200),
        RefinedPack(order=0, center=0.1, weight=0.0, scale=1.0, age=0),
    )
    contrib = [0.9, 0.0]
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.1))
    assert died == []

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_min_age_protects_newborn" -q

Regression example · hysteresis blocks weak birth

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_hysteresis_blocks_weak_birth() -> None:
    packs = (RefinedPack(order=0, center=0.5, weight=1.0, scale=2.0),)
    policy = RefinePolicy(
        indicator=Indicator.RESIDUAL,
        birth_threshold=0.1,
        death_threshold=0.05,
        hysteresis=4.0,
    )
    # score 0.15 exceeds birth_threshold but not hysteresis * death = 0.2
    proposal = propose_refinement(
        packs,
        policy=policy,
        peak_location=0.0,
        peak_value=0.15,
        derivatives=None,
        birth_score=0.15,
    )
    assert proposal is None

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_hysteresis_blocks_weak_birth -q

Fields & methods

indicatorfield

indicator: Indicator = Indicator.RESIDUAL

Declared Indicator field on RefinePolicy. Default expression: Indicator.RESIDUAL.

View source ↗

birth_thresholdfield

birth_threshold: float = 0.1

Declared float field on RefinePolicy. Default expression: 0.1.

View source ↗

death_thresholdfield

death_threshold: float = 0.01

Declared float field on RefinePolicy. Default expression: 0.01.

View source ↗

min_agefield

min_age: int = 100

Declared int field on RefinePolicy. Default expression: 100.

View source ↗

max_packsfield

max_packs: int | None = None

Declared int | None field on RefinePolicy. Default expression: None.

View source ↗

hp_rulefield

hp_rule: HpRule = 'coefficient_decay'

Declared HpRule field on RefinePolicy. Default expression: 'coefficient_decay'.

View source ↗

hysteresisfield

hysteresis: float = 2.0

Declared float field on RefinePolicy. Default expression: 2.0.

View source ↗

jet_orderfield

jet_order: int = 6

Declared int field on RefinePolicy. Default expression: 6.

View source ↗

default_orderfield

default_order: int = 0

Declared int field on RefinePolicy. Default expression: 0.

View source ↗

min_center_separationfield

min_center_separation: float = 0.0

Declared float field on RefinePolicy. Default expression: 0.0.

View source ↗

min_scale_ratiofield

min_scale_ratio: float = 1.5

Declared float field on RefinePolicy. Default expression: 1.5.

View source ↗

debug_zero_perturbationfield

debug_zero_perturbation: bool = False

Declared bool field on RefinePolicy. Default expression: False.

View source ↗

__post_init__method

__post_init__(self) -> None

See the class contract above and the exact method implementation below.

Returns: None

Validation & failure cases

When self.birth_threshold < 0.0, raises ValueError(f'birth_threshold must be >= 0, got {self.birth_threshold}').
When self.death_threshold < 0.0, raises ValueError(f'death_threshold must be >= 0, got {self.death_threshold}').
When self.min_age < 0, raises ValueError(f'min_age must be >= 0, got {self.min_age}').
When self.max_packs is not None and int(self.max_packs) < 1, raises ValueError(f'max_packs must be >= 1 or None, got {self.max_packs}').
When self.hysteresis < 1.0, raises ValueError(f'hysteresis must be >= 1, got {self.hysteresis}').
When self.jet_order < 1, raises ValueError(f'jet_order must be >= 1, got {self.jet_order}').
When self.default_order < 0, raises ValueError(f'default_order must be >= 0, got {self.default_order}').
When self.min_center_separation < 0.0, raises ValueError(f'min_center_separation must be >= 0, got {self.min_center_separation}').
When self.min_scale_ratio < 1.0, raises ValueError(f'min_scale_ratio must be >= 1, got {self.min_scale_ratio}').
When self.hp_rule not in ('coefficient_decay', 'always_h', 'always_p'), raises ValueError(f'unknown hp_rule {self.hp_rule!r}').

Executed regression · min age protects newborn

Full file, imports & fixtures ↗

def test_min_age_protects_newborn() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=1.0, age=200),
        RefinedPack(order=0, center=0.1, weight=0.0, scale=1.0, age=0),
    )
    contrib = [0.9, 0.0]
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.1))
    assert died == []

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_min_age_protects_newborn" -q

Method implementation

    def __post_init__(self) -> None:
        if self.birth_threshold < 0.0:
            raise ValueError(f"birth_threshold must be >= 0, got {self.birth_threshold}")
        if self.death_threshold < 0.0:
            raise ValueError(f"death_threshold must be >= 0, got {self.death_threshold}")
        if self.min_age < 0:
            raise ValueError(f"min_age must be >= 0, got {self.min_age}")
        if self.max_packs is not None and int(self.max_packs) < 1:
            raise ValueError(f"max_packs must be >= 1 or None, got {self.max_packs}")
        if self.hysteresis < 1.0:
            raise ValueError(f"hysteresis must be >= 1, got {self.hysteresis}")
        if self.jet_order < 1:
            raise ValueError(f"jet_order must be >= 1, got {self.jet_order}")
        if self.default_order < 0:
            raise ValueError(f"default_order must be >= 0, got {self.default_order}")
        if self.min_center_separation < 0.0:
            raise ValueError(
                f"min_center_separation must be >= 0, got {self.min_center_separation}"
            )
        if self.min_scale_ratio < 1.0:
            raise ValueError(f"min_scale_ratio must be >= 1, got {self.min_scale_ratio}")
        if self.hp_rule not in ("coefficient_decay", "always_h", "always_p"):
            raise ValueError(f"unknown hp_rule {self.hp_rule!r}")
        object.__setattr__(self, "indicator", Indicator(self.indicator))

View source ↗

Read the implementation

class RefinePolicy:
    """Discrete refinement policy. Decisions are not a smooth optimiser."""

    indicator: Indicator = Indicator.RESIDUAL
    birth_threshold: float = 0.1
    death_threshold: float = 0.01
    min_age: int = 100
    max_packs: int | None = None
    hp_rule: HpRule = "coefficient_decay"
    hysteresis: float = 2.0
    jet_order: int = 6
    default_order: int = 0
    min_center_separation: float = 0.0
    min_scale_ratio: float = 1.5
    debug_zero_perturbation: bool = False

    def __post_init__(self) -> None:
        if self.birth_threshold < 0.0:
            raise ValueError(f"birth_threshold must be >= 0, got {self.birth_threshold}")
        if self.death_threshold < 0.0:
            raise ValueError(f"death_threshold must be >= 0, got {self.death_threshold}")
        if self.min_age < 0:
            raise ValueError(f"min_age must be >= 0, got {self.min_age}")
        if self.max_packs is not None and int(self.max_packs) < 1:
            raise ValueError(f"max_packs must be >= 1 or None, got {self.max_packs}")
        if self.hysteresis < 1.0:
            raise ValueError(f"hysteresis must be >= 1, got {self.hysteresis}")
        if self.jet_order < 1:
            raise ValueError(f"jet_order must be >= 1, got {self.jet_order}")
        if self.default_order < 0:
            raise ValueError(f"default_order must be >= 0, got {self.default_order}")
        if self.min_center_separation < 0.0:
            raise ValueError(
                f"min_center_separation must be >= 0, got {self.min_center_separation}"
            )
        if self.min_scale_ratio < 1.0:
            raise ValueError(f"min_scale_ratio must be >= 1, got {self.min_scale_ratio}")
        if self.hp_rule not in ("coefficient_decay", "always_h", "always_p"):
            raise ValueError(f"unknown hp_rule {self.hp_rule!r}")
        object.__setattr__(self, "indicator", Indicator(self.indicator))

Other public import paths

omnibias.core.RefinePolicy
omnibias.core.refine.RefinePolicy
omnibias.jax.refine.RefinePolicy
omnibias.torch.refine.RefinePolicy

CLASS

RefineProposal

omnibias.core.refine.RefineProposalSource ↗

class RefineProposal

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

A single structural suggestion before it is applied.

Parameters

Name	Type & default	Contract
`move`dataclass field	`HpMove`Required	See the operation contract and implementation below.
`center`dataclass field	`float`Required	See the operation contract and implementation below.
`scale`dataclass field	`float`Required	See the operation contract and implementation below.
`order`dataclass field	`int`Required	See the operation contract and implementation below.
`parent_index`dataclass field	`int \| None`Default: None	See the operation contract and implementation below.
`indicator_peak`dataclass field	`float`Default: 0.0	See the operation contract and implementation below.
`inherit_scale`dataclass field	`bool`Default: False	See the operation contract and implementation below.

Regression example · birth is zero weight

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_birth_is_zero_weight() -> None:
    packs = (RefinedPack(order=0, center=0.25, weight=1.0, scale=2.0),)
    proposal = RefineProposal(move="h", center=_BL_CENTER, scale=_BL_SCALE, order=0)
    after, born = apply_birth(packs, proposal)
    assert born.weight == 0.0
    xs = [0.0, 0.01, 0.5, 1.0]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_birth_is_zero_weight -q

Regression example · growth sibling is zero perturbation

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_growth_sibling_is_zero_perturbation() -> None:
    packs = (RefinedPack(order=0, center=0.0, weight=0.3, scale=-100.0),)
    proposal = RefineProposal(
        move="p",
        center=0.0,
        scale=-100.0,
        order=1,
        parent_index=0,
    )
    after, grown, parent = apply_growth(packs, proposal)
    assert parent == 0
    assert grown.weight == 0.0
    assert grown.order == 1
    xs = [0.0, 0.01, 0.25]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_growth_sibling_is_zero_perturbation -q

Fields & methods

movefield

move: HpMove

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

View source ↗

centerfield

center: float

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

View source ↗

scalefield

scale: float

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

View source ↗

orderfield

order: int

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

View source ↗

parent_indexfield

parent_index: int | None = None

Declared int | None field on RefineProposal. Default expression: None.

View source ↗

indicator_peakfield

indicator_peak: float = 0.0

Declared float field on RefineProposal. Default expression: 0.0.

View source ↗

inherit_scalefield

inherit_scale: bool = False

Declared bool field on RefineProposal. Default expression: False.

View source ↗

Read the implementation

class RefineProposal:
    """A single structural suggestion before it is applied."""

    move: HpMove
    center: float
    scale: float
    order: int
    parent_index: int | None = None
    indicator_peak: float = 0.0
    inherit_scale: bool = False

CLASS

RefineReport

omnibias.core.RefineReportSource ↗

class RefineReport

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

What changed, plus the death bound that the caller must see.

Parameters

Name	Type & default	Contract
`born`dataclass field	`tuple[RefinedPack, ...]`Required	See the operation contract and implementation below.
`grown`dataclass field	`tuple[int, ...]`Required	See the operation contract and implementation below.
`died`dataclass field	`tuple[int, ...]`Required	See the operation contract and implementation below.
`death_perturbation`dataclass field	`float`Required	See the operation contract and implementation below.
`indicator_values`dataclass field	`tuple[float, ...]`Required	See the operation contract and implementation below.
`proposed_center`dataclass field	`float \| None`Default: None	See the operation contract and implementation below.
`proposed_scale`dataclass field	`float \| None`Default: None	See the operation contract and implementation below.
`hp_move`dataclass field	`HpMove`Default: 'none'	See the operation contract and implementation below.

Fields & methods

bornfield

born: tuple[RefinedPack, ...]

Declared tuple[RefinedPack, ...] field on RefineReport. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

grownfield

grown: tuple[int, ...]

Declared tuple[int, ...] field on RefineReport. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

diedfield

died: tuple[int, ...]

Declared tuple[int, ...] field on RefineReport. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

death_perturbationfield

death_perturbation: float

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

View source ↗

indicator_valuesfield

indicator_values: tuple[float, ...]

Declared tuple[float, ...] field on RefineReport. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.

View source ↗

proposed_centerfield

proposed_center: float | None = None

Declared float | None field on RefineReport. Default expression: None.

View source ↗

proposed_scalefield

proposed_scale: float | None = None

Declared float | None field on RefineReport. Default expression: None.

View source ↗

hp_movefield

hp_move: HpMove = 'none'

Declared HpMove field on RefineReport. Default expression: 'none'.

View source ↗

Read the implementation

class RefineReport:
    """What changed, plus the death bound that the caller must see."""

    born: tuple[RefinedPack, ...]
    grown: tuple[int, ...]
    died: tuple[int, ...]
    death_perturbation: float
    indicator_values: tuple[float, ...]
    proposed_center: float | None = None
    proposed_scale: float | None = None
    hp_move: HpMove = "none"

Other public import paths

omnibias.core.RefineReport
omnibias.core.refine.RefineReport
omnibias.jax.refine.RefineReport
omnibias.torch.refine.RefineReport

CLASS

RefinedPack

omnibias.core.RefinedPackSource ↗

class RefinedPack

Dataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.

One tempered pack: c * alpha^n sigma^(n)(alpha (z - center)).

center is the physical location of the pack (spec 03-13 mu). This is *not* ~omnibias.core.multipack.PackSpec.mean, which is added (z + mean). Convert with from_pack_spec / to_pack_spec.

Parameters

Name	Type & default	Contract
`order`dataclass field	`int`Required	See the operation contract and implementation below.
`center`dataclass field	`float`Required	See the operation contract and implementation below.
`weight`dataclass field	`float`Default: 0.0	See the operation contract and implementation below.
`scale`dataclass field	`float`Default: 1.0	See the operation contract and implementation below.
`age`dataclass field	`int`Default: 0	See the operation contract and implementation below.

Executed regression · increment ages

Full file, imports & fixtures ↗

def test_increment_ages() -> None:
    packs = increment_ages((RefinedPack(order=0, center=0.0, weight=1.0, age=3),), 2)
    assert packs[0].age == 5

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_increment_ages" -q

Regression example · pack spec roundtrip

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_pack_spec_roundtrip() -> None:
    spec = PackSpec(order=2, mean=-0.25, weight=0.5)
    pack = RefinedPack.from_pack_spec(spec, scale=3.0)
    assert pack.center == pytest.approx(0.25)
    assert pack.to_pack_spec().mean == pytest.approx(spec.mean)
    assert pack.to_pack_spec().order == spec.order

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_pack_spec_roundtrip -q

Fields & methods

orderfield

order: int

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

View source ↗

centerfield

center: float

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

View source ↗

weightfield

weight: float = 0.0

Declared float field on RefinedPack. Default expression: 0.0.

View source ↗

scalefield

scale: float = 1.0

Declared float field on RefinedPack. Default expression: 1.0.

View source ↗

agefield

age: int = 0

Declared int field on RefinedPack. Default expression: 0.

View source ↗

__post_init__method

__post_init__(self) -> None

See the class contract above and the exact method implementation below.

Returns: None

Validation & failure cases

When int(self.order) < 0, raises ValueError(f'order must be >= 0, got {self.order}').
When not math.isfinite(self.center), raises ValueError(f'center must be finite, got {self.center}').
When not math.isfinite(self.weight), raises ValueError(f'weight must be finite, got {self.weight}').
When not math.isfinite(self.scale) or self.scale == 0.0, raises ValueError(f'scale must be finite and nonzero, got {self.scale}').
When int(self.age) < 0, raises ValueError(f'age must be >= 0, got {self.age}').

Executed regression · increment ages

Full file, imports & fixtures ↗

def test_increment_ages() -> None:
    packs = increment_ages((RefinedPack(order=0, center=0.0, weight=1.0, age=3),), 2)
    assert packs[0].age == 5

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_increment_ages" -q

Method implementation

    def __post_init__(self) -> None:
        if int(self.order) < 0:
            raise ValueError(f"order must be >= 0, got {self.order}")
        if not math.isfinite(self.center):
            raise ValueError(f"center must be finite, got {self.center}")
        if not math.isfinite(self.weight):
            raise ValueError(f"weight must be finite, got {self.weight}")
        if not math.isfinite(self.scale) or self.scale == 0.0:
            raise ValueError(f"scale must be finite and nonzero, got {self.scale}")
        if int(self.age) < 0:
            raise ValueError(f"age must be >= 0, got {self.age}")
        object.__setattr__(self, "order", int(self.order))
        object.__setattr__(self, "center", float(self.center))
        object.__setattr__(self, "weight", float(self.weight))
        object.__setattr__(self, "scale", float(self.scale))
        object.__setattr__(self, "age", int(self.age))

View source ↗

from_pack_specmethod

from_pack_spec(cls, spec: PackSpec, *, scale: float=1.0, age: int=0) -> RefinedPack

Map a Birkhoff PackSpec (z + mean) onto a centred pack.

Parameters

Name	Type & default	Contract
`spec`positional-or-keyword	`PackSpec`Required	See the operation contract and implementation below.
`scale`keyword-only	`float`Default: 1.0	See the operation contract and implementation below.
`age`keyword-only	`int`Default: 0	See the operation contract and implementation below.

Returns: RefinedPack

Executed regression · pack spec roundtrip

Full file, imports & fixtures ↗

def test_pack_spec_roundtrip() -> None:
    spec = PackSpec(order=2, mean=-0.25, weight=0.5)
    pack = RefinedPack.from_pack_spec(spec, scale=3.0)
    assert pack.center == pytest.approx(0.25)
    assert pack.to_pack_spec().mean == pytest.approx(spec.mean)
    assert pack.to_pack_spec().order == spec.order

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_pack_spec_roundtrip" -q

Method implementation

    def from_pack_spec(cls, spec: PackSpec, *, scale: float = 1.0, age: int = 0) -> RefinedPack:
        """Map a Birkhoff :class:`PackSpec` (``z + mean``) onto a centred pack."""
        return cls(
            order=spec.order,
            center=-float(spec.mean),
            weight=float(spec.weight),
            scale=float(scale),
            age=int(age),
        )

View source ↗

to_pack_specmethod

to_pack_spec(self) -> PackSpec

Inverse of from_pack_spec (drops scale and age).

Returns: PackSpec

Executed regression · pack spec roundtrip

Full file, imports & fixtures ↗

def test_pack_spec_roundtrip() -> None:
    spec = PackSpec(order=2, mean=-0.25, weight=0.5)
    pack = RefinedPack.from_pack_spec(spec, scale=3.0)
    assert pack.center == pytest.approx(0.25)
    assert pack.to_pack_spec().mean == pytest.approx(spec.mean)
    assert pack.to_pack_spec().order == spec.order

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_pack_spec_roundtrip" -q

Method implementation

    def to_pack_spec(self) -> PackSpec:
        """Inverse of :meth:`from_pack_spec` (drops ``scale`` and ``age``)."""
        return PackSpec(order=self.order, mean=-self.center, weight=self.weight)

View source ↗

with_weightmethod

with_weight(self, weight: float) -> RefinedPack

See the class contract above and the exact method implementation below.

Parameters

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

Returns: RefinedPack

Method implementation

    def with_weight(self, weight: float) -> RefinedPack:
        return RefinedPack(
            order=self.order,
            center=self.center,
            weight=float(weight),
            scale=self.scale,
            age=self.age,
        )

View source ↗

agedmethod

aged(self, steps: int=1) -> RefinedPack

See the class contract above and the exact method implementation below.

Parameters

Name	Type & default	Contract
`steps`positional-or-keyword	`int`Default: 1	See the operation contract and implementation below.

Returns: RefinedPack

Executed regression · increment ages

Full file, imports & fixtures ↗

def test_increment_ages() -> None:
    packs = increment_ages((RefinedPack(order=0, center=0.0, weight=1.0, age=3),), 2)
    assert packs[0].age == 5

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_increment_ages" -q

Method implementation

    def aged(self, steps: int = 1) -> RefinedPack:
        return RefinedPack(
            order=self.order,
            center=self.center,
            weight=self.weight,
            scale=self.scale,
            age=self.age + int(steps),
        )

View source ↗

Read the implementation

class RefinedPack:
    """One tempered pack: ``c * alpha^n sigma^(n)(alpha (z - center))``.

    ``center`` is the physical location of the pack (spec 03-13 ``mu``).
    This is *not* :class:`~omnibias.core.multipack.PackSpec.mean`, which
    is added (``z + mean``). Convert with :meth:`from_pack_spec` /
    :meth:`to_pack_spec`.
    """

    order: int
    center: float
    weight: float = 0.0
    scale: float = 1.0
    age: int = 0

    def __post_init__(self) -> None:
        if int(self.order) < 0:
            raise ValueError(f"order must be >= 0, got {self.order}")
        if not math.isfinite(self.center):
            raise ValueError(f"center must be finite, got {self.center}")
        if not math.isfinite(self.weight):
            raise ValueError(f"weight must be finite, got {self.weight}")
        if not math.isfinite(self.scale) or self.scale == 0.0:
            raise ValueError(f"scale must be finite and nonzero, got {self.scale}")
        if int(self.age) < 0:
            raise ValueError(f"age must be >= 0, got {self.age}")
        object.__setattr__(self, "order", int(self.order))
        object.__setattr__(self, "center", float(self.center))
        object.__setattr__(self, "weight", float(self.weight))
        object.__setattr__(self, "scale", float(self.scale))
        object.__setattr__(self, "age", int(self.age))

    @classmethod
    def from_pack_spec(cls, spec: PackSpec, *, scale: float = 1.0, age: int = 0) -> RefinedPack:
        """Map a Birkhoff :class:`PackSpec` (``z + mean``) onto a centred pack."""
        return cls(
            order=spec.order,
            center=-float(spec.mean),
            weight=float(spec.weight),
            scale=float(scale),
            age=int(age),
        )

    def to_pack_spec(self) -> PackSpec:
        """Inverse of :meth:`from_pack_spec` (drops ``scale`` and ``age``)."""
        return PackSpec(order=self.order, mean=-self.center, weight=self.weight)

    def with_weight(self, weight: float) -> RefinedPack:
        return RefinedPack(
            order=self.order,
            center=self.center,
            weight=float(weight),
            scale=self.scale,
            age=self.age,
        )

    def aged(self, steps: int = 1) -> RefinedPack:
        return RefinedPack(
            order=self.order,
            center=self.center,
            weight=self.weight,
            scale=self.scale,
            age=self.age + int(steps),
        )

Other public import paths

omnibias.core.RefinedPack
omnibias.core.refine.RefinedPack
omnibias.jax.refine.RefinedPack
omnibias.torch.refine.RefinedPack

FUNCTION

apply_birth

omnibias.core.refine.apply_birthSource ↗

apply_birth(packs: Sequence[RefinedPack], proposal: RefineProposal) -> tuple[list[RefinedPack], RefinedPack]

Append a zero-weight pack. The represented function is unchanged.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`proposal`positional-or-keyword	`RefineProposal`Required	See the operation contract and implementation below.

Returnstuple[list[RefinedPack], RefinedPack]

Executed regression · birth is zero weight

Full file, imports & fixtures ↗

def test_birth_is_zero_weight() -> None:
    packs = (RefinedPack(order=0, center=0.25, weight=1.0, scale=2.0),)
    proposal = RefineProposal(move="h", center=_BL_CENTER, scale=_BL_SCALE, order=0)
    after, born = apply_birth(packs, proposal)
    assert born.weight == 0.0
    xs = [0.0, 0.01, 0.5, 1.0]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_birth_is_zero_weight" -q

Read the implementation

def apply_birth(packs: Sequence[RefinedPack], proposal: RefineProposal) -> tuple[list[RefinedPack], RefinedPack]:
    """Append a zero-weight pack. The represented function is unchanged."""
    born = RefinedPack(
        order=proposal.order,
        center=proposal.center,
        weight=0.0,
        scale=proposal.scale,
        age=0,
    )
    return [*packs, born], born

FUNCTION

apply_deaths

omnibias.core.refine.apply_deathsSource ↗

apply_deaths(packs: Sequence[RefinedPack], indices: Sequence[int], contributions: Sequence[float]) -> tuple[list[RefinedPack], float]

Remove packs. Reported perturbation is the sum of their contributions.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`indices`positional-or-keyword	`Sequence[int]`Required	See the operation contract and implementation below.
`contributions`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

Returnstuple[list[RefinedPack], float]

Executed regression · g3 scale flow

Full file, imports & fixtures ↗

def test_g3_scale_flow() -> None:
    bank = _initial_bank()
    probes = jnp.linspace(0.0, 1.0, 65)

    def residual(x: Array) -> Array:
        return jnp.exp(-100.0 * x)

    _bank, report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.SCALE_FLOW,
            birth_threshold=0.1,
            death_threshold=0.0,
            min_age=10_000,
            hysteresis=1.0,
            min_scale_ratio=1.0,
        ),
        step=0,
        probe_jet=(_BL_CENTER, _BL_DERIVS),
    )
    assert report.proposed_scale == pytest.approx(_BL_SCALE)

python -m pytest "packages/omnibias-jax/tests/test_refine.py::test_g3_scale_flow" -q

Regression example · death perturbation bounds measured change

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_death_perturbation_bounds_measured_change() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=-2.0, age=200),
        RefinedPack(order=0, center=0.25, weight=0.003, scale=2.0, age=200),
    )
    xs = [i / 32.0 for i in range(33)]
    terms = []
    field_sq = 0.0
    for pack in packs:
        values = [pack_term_scalar(x, pack, _exp_sigma) for x in xs]
        terms.append(math.sqrt(sum(v * v for v in values)))
        field_sq += sum(v * v for v in values)
    field_norm = math.sqrt(field_sq)
    contrib = pack_contributions(terms, field_norm)
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.05))
    assert 1 in died
    kept, reported = apply_deaths(packs, died, contrib)
    assert len(kept) == 1
    before = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after = [sum(pack_term_scalar(x, p, _exp_sigma) for p in kept) for x in xs]
    delta = math.sqrt(sum((a - b) ** 2 for a, b in zip(after, before, strict=True)))
    measured = delta / field_norm
    assert death_bound_holds(measured, reported, atol=1e-15)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_death_perturbation_bounds_measured_change -q

Read the implementation

def apply_deaths(
    packs: Sequence[RefinedPack],
    indices: Sequence[int],
    contributions: Sequence[float],
) -> tuple[list[RefinedPack], float]:
    """Remove packs. Reported perturbation is the sum of their contributions."""
    doomed = set(int(i) for i in indices)
    perturbation = sum(float(contributions[i]) for i in doomed)
    kept = [p for i, p in enumerate(packs) if i not in doomed]
    return kept, perturbation

FUNCTION

apply_growth

omnibias.core.refine.apply_growthSource ↗

apply_growth(packs: Sequence[RefinedPack], proposal: RefineProposal) -> tuple[list[RefinedPack], RefinedPack, int]

*p*-type: sibling at the same centre / scale, order n+1, c = 0.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`proposal`positional-or-keyword	`RefineProposal`Required	See the operation contract and implementation below.

Returnstuple[list[RefinedPack], RefinedPack, int]

Explicit validation & failure cases

When proposal.parent_index is None, raises ValueError('p-type growth requires a parent_index').

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

Executed regression · growth sibling is zero perturbation

Full file, imports & fixtures ↗

def test_growth_sibling_is_zero_perturbation() -> None:
    packs = (RefinedPack(order=0, center=0.0, weight=0.3, scale=-100.0),)
    proposal = RefineProposal(
        move="p",
        center=0.0,
        scale=-100.0,
        order=1,
        parent_index=0,
    )
    after, grown, parent = apply_growth(packs, proposal)
    assert parent == 0
    assert grown.weight == 0.0
    assert grown.order == 1
    xs = [0.0, 0.01, 0.25]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_growth_sibling_is_zero_perturbation" -q

Read the implementation

def apply_growth(
    packs: Sequence[RefinedPack],
    proposal: RefineProposal,
) -> tuple[list[RefinedPack], RefinedPack, int]:
    """*p*-type: sibling at the same centre / scale, order ``n+1``, ``c = 0``."""
    if proposal.parent_index is None:
        raise ValueError("p-type growth requires a parent_index")
    parent = packs[proposal.parent_index]
    grown = RefinedPack(
        order=proposal.order,
        center=parent.center,
        weight=0.0,
        scale=parent.scale,
        age=0,
    )
    return [*packs, grown], grown, proposal.parent_index

FUNCTION

assert_zero_perturbation

omnibias.core.assert_zero_perturbationSource ↗

assert_zero_perturbation(before: Sequence[float], after: Sequence[float]) -> None

Birth and growth must be bit-identical, not 'within tolerance'.

Parameters

Name	Type & default	Contract
`before`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.
`after`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

ReturnsNone

Explicit validation & failure cases

When len(before) != len(after), raises AssertionError(f'zero-perturbation length mismatch: {len(before)} vs {len(after)}').
When a != b, raises AssertionError(f'zero-perturbation violated at index {i}: {a!r} vs {b!r}').

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

Executed regression · birth is zero weight

Full file, imports & fixtures ↗

def test_birth_is_zero_weight() -> None:
    packs = (RefinedPack(order=0, center=0.25, weight=1.0, scale=2.0),)
    proposal = RefineProposal(move="h", center=_BL_CENTER, scale=_BL_SCALE, order=0)
    after, born = apply_birth(packs, proposal)
    assert born.weight == 0.0
    xs = [0.0, 0.01, 0.5, 1.0]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_birth_is_zero_weight" -q

Regression example · growth sibling is zero perturbation

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_growth_sibling_is_zero_perturbation() -> None:
    packs = (RefinedPack(order=0, center=0.0, weight=0.3, scale=-100.0),)
    proposal = RefineProposal(
        move="p",
        center=0.0,
        scale=-100.0,
        order=1,
        parent_index=0,
    )
    after, grown, parent = apply_growth(packs, proposal)
    assert parent == 0
    assert grown.weight == 0.0
    assert grown.order == 1
    xs = [0.0, 0.01, 0.25]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_growth_sibling_is_zero_perturbation -q

Read the implementation

def assert_zero_perturbation(
    before: Sequence[float],
    after: Sequence[float],
) -> None:
    """Birth and growth must be bit-identical, not 'within tolerance'."""
    if len(before) != len(after):
        raise AssertionError(
            f"zero-perturbation length mismatch: {len(before)} vs {len(after)}"
        )
    for i, (a, b) in enumerate(zip(before, after, strict=True)):
        if a != b:
            raise AssertionError(
                f"zero-perturbation violated at index {i}: {a!r} vs {b!r}"
            )

Other public import paths

omnibias.core.assert_zero_perturbation
omnibias.core.refine.assert_zero_perturbation

FUNCTION

certified_local_error

omnibias.core.refine.certified_local_errorSource ↗

certified_local_error(derivatives: Sequence[float], step: float) -> float

Lagrange remainder of the jet using |phi^(N)| as M for order N-1.

Sound when |phi^(N)| bounds the next derivative on the segment of length |step|. Uses the same identity as omnibias.core.line_search.lagrange_remainder_bound.

Parameters

Name	Type & default	Contract
`derivatives`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.
`step`positional-or-keyword	`float`Required	See the operation contract and implementation below.

Returnsfloat

Explicit validation & failure cases

When len(derivatives) < 2, raises ValueError('certified_local_error needs at least two derivatives').

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

Executed regression · certified local error exp

Full file, imports & fixtures ↗

def test_certified_local_error_exp() -> None:
    # exp on [0, 0.1]; order-3 model, M = |phi^(4)(0)| = 1 under-bounds
    # the true max e^{0.1}; still a well-defined remainder number.
    derivs = tuple(1.0 for _ in range(5))
    bound = certified_local_error(derivs, 0.1)
    remainder = abs(math.exp(0.1) - sum((0.1**k) / math.factorial(k) for k in range(4)))
    # Using M=1 < e^{0.1} is not sound; the API is only as sound as M.
    # With the true max M the bound must cover the remainder:
    sound = certified_local_error((1.0, 1.0, 1.0, 1.0, math.exp(0.1)), 0.1)
    assert sound >= remainder - 1e-15
    assert bound > 0.0

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_certified_local_error_exp" -q

Read the implementation

def certified_local_error(
    derivatives: Sequence[float],
    step: float,
) -> float:
    """Lagrange remainder of the jet using ``|phi^(N)|`` as ``M`` for order ``N-1``.

    Sound when ``|phi^(N)|`` bounds the next derivative on the segment of
    length ``|step|``. Uses the same identity as
    :func:`omnibias.core.line_search.lagrange_remainder_bound`.
    """
    if len(derivatives) < 2:
        raise ValueError("certified_local_error needs at least two derivatives")
    order = len(derivatives) - 2
    m = abs(float(derivatives[-1]))
    enclosure = lagrange_remainder_bound(m, float(step), order)
    return float(enclosure.hi)

FUNCTION

death_bound_holds

omnibias.core.refine.death_bound_holdsSource ↗

death_bound_holds(measured_relative: float, reported: float, *, atol: float=0.0) -> bool

measured ||Δu|| / ||u|| must not exceed the reported bound.

Parameters

Name	Type & default	Contract
`measured_relative`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`reported`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`atol`keyword-only	`float`Default: 0.0	See the operation contract and implementation below.

Returnsbool

Explicit validation & failure cases

When reported < 0.0, raises ValueError(f'reported perturbation must be >= 0, got {reported}').

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

Executed regression · death perturbation bounds change

Full file, imports & fixtures ↗

def test_death_perturbation_bounds_change() -> None:
    bank = init_pack_bank(
        (
            RefinedPack(order=0, center=0.0, weight=1.0, scale=-2.0, age=200),
            RefinedPack(order=0, center=0.25, weight=0.003, scale=2.0, age=200),
        ),
        max_packs=4,
        base="exp",
    )
    probes = jnp.linspace(0.0, 1.0, 33)
    before = bank_forward(bank, probes)

    def residual(x: Array) -> Array:
        return jnp.zeros_like(x)

    bank, report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.RESIDUAL,
            birth_threshold=10.0,
            death_threshold=0.05,
            min_age=0,
            hysteresis=1.0,
        ),
        step=0,
    )
    assert report.died
    after = bank_forward(bank, probes)
    delta = float(jnp.sqrt(jnp.sum((after - before) ** 2)))
    field_norm = float(jnp.sqrt(jnp.sum(before**2)))
    assert death_bound_holds(delta / field_norm, report.death_perturbation, atol=1e-12)

python -m pytest "packages/omnibias-jax/tests/test_refine.py::test_death_perturbation_bounds_change" -q

Regression example · death perturbation bounds change

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_death_perturbation_bounds_change() -> None:
    bank = AdaptivePackBank(
        (
            RefinedPack(order=0, center=0.0, weight=1.0, scale=-2.0, age=200),
            RefinedPack(order=0, center=0.25, weight=0.003, scale=2.0, age=200),
        ),
        max_packs=4,
        base="exp",
    )
    probes = torch.linspace(0.0, 1.0, 33)
    before = bank(probes).clone()

    def residual(x: torch.Tensor) -> torch.Tensor:
        return torch.zeros_like(x)

    report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.RESIDUAL,
            birth_threshold=10.0,
            death_threshold=0.05,
            min_age=0,
            hysteresis=1.0,
        ),
        step=0,
    )
    assert report.died
    after = bank(probes)
    delta = float((after - before).square().sum().sqrt().item())
    field_norm = float(before.square().sum().sqrt().item())
    assert death_bound_holds(delta / field_norm, report.death_perturbation, atol=1e-12)

python -m pytest packages/omnibias-torch/tests/test_refine.py -k test_death_perturbation_bounds_change -q

Read the implementation

def death_bound_holds(
    measured_relative: float,
    reported: float,
    *,
    atol: float = 0.0,
) -> bool:
    """``measured ||Δu|| / ||u||`` must not exceed the reported bound."""
    if reported < 0.0:
        raise ValueError(f"reported perturbation must be >= 0, got {reported}")
    return float(measured_relative) <= float(reported) + float(atol)

FUNCTION

domb_sykes_fit

omnibias.core.refine.domb_sykes_fitSource ↗

domb_sykes_fit(derivatives: Sequence[float]) -> tuple[float, float]

Linear fit of a_k / a_{k-1} against 1/k.

Returns (intercept, slope). For a singularity (1 - x/x_s)^{-p} the intercept is 1/x_s. An entire function (the worked exp(-100 x) example) has intercept 0.

Parameters

Name	Type & default	Contract
`derivatives`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

Returnstuple[float, float]

Explicit validation & failure cases

When n < 2, raises ValueError('Domb-Sykes needs at least two finite ratios').
When denom == 0.0, raises ValueError('Domb-Sykes abscissae are degenerate').

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

Executed regression · boundary layer hp is h then p

Full file, imports & fixtures ↗

def test_boundary_layer_hp_is_h_then_p() -> None:
    assert hp_decision(_BL_DERIVS, existing_scale=2.0) == "h"
    assert hp_decision(_BL_DERIVS, existing_scale=-100.0) == "p"

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_boundary_layer_hp_is_h_then_p" -q

Regression example · boundary layer domb sykes entire

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_boundary_layer_domb_sykes_entire() -> None:
    intercept, slope = domb_sykes_fit(_BL_DERIVS)
    assert intercept == pytest.approx(0.0, abs=1e-12)
    assert slope == pytest.approx(_BL_SCALE, rel=1e-12)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_boundary_layer_domb_sykes_entire -q

Read the implementation

def domb_sykes_fit(derivatives: Sequence[float]) -> tuple[float, float]:
    r"""Linear fit of ``a_k / a_{k-1}`` against ``1/k``.

    Returns ``(intercept, slope)``. For a singularity
    ``(1 - x/x_s)^{-p}`` the intercept is ``1/x_s``. An entire function
    (the worked ``exp(-100 x)`` example) has intercept ``0``.
    """
    ratios = successive_ratios(derivatives)
    xs: list[float] = []
    ys: list[float] = []
    for k, ratio in enumerate(ratios, start=1):
        if math.isfinite(ratio):
            xs.append(1.0 / float(k))
            ys.append(ratio)
    n = len(xs)
    if n < 2:
        raise ValueError("Domb-Sykes needs at least two finite ratios")
    sum_x = sum(xs)
    sum_y = sum(ys)
    sum_xx = sum(x * x for x in xs)
    sum_xy = sum(x * y for x, y in zip(xs, ys, strict=True))
    denom = float(n) * sum_xx - sum_x * sum_x
    if denom == 0.0:
        raise ValueError("Domb-Sykes abscissae are degenerate")
    slope = (float(n) * sum_xy - sum_x * sum_y) / denom
    intercept = (sum_y - slope * sum_x) / float(n)
    return intercept, slope

FUNCTION

factorial_scale_term

omnibias.core.refine.factorial_scale_termSource ↗

factorial_scale_term(scale: float, order: int) -> float

alpha^n with an integer order (used by the twins' host path).

Parameters

Name	Type & default	Contract
`scale`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`order`positional-or-keyword	`int`Required	See the operation contract and implementation below.

Returnsfloat

Explicit validation & failure cases

When order < 0, raises ValueError(f'order must be >= 0, got {order}').

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

Read the implementation

def factorial_scale_term(scale: float, order: int) -> float:
    """``alpha^n`` with an integer order (used by the twins' host path)."""
    if order < 0:
        raise ValueError(f"order must be >= 0, got {order}")
    return float(scale) ** int(order)

FUNCTION

hp_decision

omnibias.core.hp_decisionSource ↗

hp_decision(derivatives: Sequence[float], existing_scale: float, *, rule: HpRule='coefficient_decay', scale_mismatch: float=2.0, entire_intercept: float=0.05) -> HpMove

Choose *h* (new scale) or *p* (raise order) from jet decay.

Classical *hp*: *p* where the solution is locally analytic at the current scale, *h* where the scale is wrong or a singularity is nearby. The worked boundary-layer jet is entire (intercept ~ 0) but the early ratios demand |alpha| ~ 100, so the first move is *h*.

Parameters

Name	Type & default	Contract
`derivatives`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.
`existing_scale`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`rule`keyword-only	`HpRule`Default: 'coefficient_decay'	See the operation contract and implementation below.
`scale_mismatch`keyword-only	`float`Default: 2.0	See the operation contract and implementation below.
`entire_intercept`keyword-only	`float`Default: 0.05	See the operation contract and implementation below.

ReturnsHpMove

Explicit validation & failure cases

When existing_scale == 0.0 or not math.isfinite(existing_scale), raises ValueError(f'existing_scale must be finite and nonzero, got {existing_scale}').

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

Executed regression · boundary layer hp is h then p

Full file, imports & fixtures ↗

def test_boundary_layer_hp_is_h_then_p() -> None:
    assert hp_decision(_BL_DERIVS, existing_scale=2.0) == "h"
    assert hp_decision(_BL_DERIVS, existing_scale=-100.0) == "p"

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_boundary_layer_hp_is_h_then_p" -q

Read the implementation

def hp_decision(
    derivatives: Sequence[float],
    existing_scale: float,
    *,
    rule: HpRule = "coefficient_decay",
    scale_mismatch: float = 2.0,
    entire_intercept: float = 0.05,
) -> HpMove:
    """Choose *h* (new scale) or *p* (raise order) from jet decay.

    Classical *hp*: *p* where the solution is locally analytic at the
    current scale, *h* where the scale is wrong or a singularity is
    nearby. The worked boundary-layer jet is entire (intercept ~ 0) but
    the early ratios demand ``|alpha| ~ 100``, so the first move is *h*.
    """
    if rule == "always_h":
        return "h"
    if rule == "always_p":
        return "p"
    if existing_scale == 0.0 or not math.isfinite(existing_scale):
        raise ValueError(f"existing_scale must be finite and nonzero, got {existing_scale}")
    scale = local_scale_from_derivatives(derivatives)
    if abs(scale) / abs(existing_scale) >= float(scale_mismatch) or abs(existing_scale) / max(
        abs(scale), 1e-300
    ) >= float(scale_mismatch):
        return "h"
    intercept, _slope = domb_sykes_fit(derivatives)
    if math.isfinite(intercept) and abs(intercept) <= float(entire_intercept):
        return "p"
    return "h"

Other public import paths

omnibias.core.hp_decision
omnibias.core.refine.hp_decision

FUNCTION

increment_ages

omnibias.core.refine.increment_agesSource ↗

increment_ages(packs: Sequence[RefinedPack], steps: int=1) -> list[RefinedPack]

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

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`steps`positional-or-keyword	`int`Default: 1	See the operation contract and implementation below.

Returnslist[RefinedPack]

Executed regression · increment ages

Full file, imports & fixtures ↗

def test_increment_ages() -> None:
    packs = increment_ages((RefinedPack(order=0, center=0.0, weight=1.0, age=3),), 2)
    assert packs[0].age == 5

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_increment_ages" -q

Regression example · min age protects newborn

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_min_age_protects_newborn() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=1.0, age=200),
        RefinedPack(order=0, center=0.1, weight=0.0, scale=1.0, age=0),
    )
    contrib = [0.9, 0.0]
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.1))
    assert died == []

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_min_age_protects_newborn -q

Read the implementation

def increment_ages(packs: Sequence[RefinedPack], steps: int = 1) -> list[RefinedPack]:
    return [p.aged(steps) for p in packs]

FUNCTION

inherit_scale

omnibias.core.refine.inherit_scaleSource ↗

inherit_scale(packs: Sequence[RefinedPack], center: float, default: float=1.0) -> float

Scale of the nearest existing pack, or default if the bank is empty.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`center`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`default`positional-or-keyword	`float`Default: 1.0	See the operation contract and implementation below.

Returnsfloat

Executed regression · inherit scale empty

Full file, imports & fixtures ↗

def test_inherit_scale_empty() -> None:
    assert inherit_scale((), 0.0, default=4.0) == 4.0

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_inherit_scale_empty" -q

Read the implementation

def inherit_scale(packs: Sequence[RefinedPack], center: float, default: float = 1.0) -> float:
    """Scale of the nearest existing pack, or ``default`` if the bank is empty."""
    idx = nearest_pack_index(packs, center)
    if idx is None:
        return float(default)
    return packs[idx].scale

FUNCTION

jet_order_needed

omnibias.core.refine.jet_order_neededSource ↗

jet_order_needed(policy: RefinePolicy) -> int

Derivatives 0 .. N required for the configured indicator.

Parameters

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

Returnsint

Read the implementation

def jet_order_needed(policy: RefinePolicy) -> int:
    """Derivatives ``0 .. N`` required for the configured indicator."""
    return max(int(policy.jet_order), 2)

FUNCTION

local_scale_from_derivatives

omnibias.core.local_scale_from_derivativesSource ↗

local_scale_from_derivatives(derivatives: Sequence[float]) -> float

Signed generation scale from early jet ratios.

For exp(alpha (z - z0)), a_k / a_{k-1} = alpha / k, so k (a_k / a_{k-1}) is constantly alpha. The estimator is the median of those products over finite ratios. The sign is the sign of a_1 / a_0 (the first directional scale).

Parameters

Name	Type & default	Contract
`derivatives`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

Returnsfloat

Explicit validation & failure cases

When not products, raises ValueError('no finite successive ratio; cannot estimate scale').

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

Executed regression · boundary layer local scale

Full file, imports & fixtures ↗

def test_boundary_layer_local_scale() -> None:
    scale = local_scale_from_derivatives(_BL_DERIVS)
    assert scale == pytest.approx(_BL_SCALE, rel=1e-12)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_boundary_layer_local_scale" -q

Read the implementation

def local_scale_from_derivatives(derivatives: Sequence[float]) -> float:
    r"""Signed generation scale from early jet ratios.

    For ``exp(alpha (z - z0))``, ``a_k / a_{k-1} = alpha / k``, so
    ``k (a_k / a_{k-1})`` is constantly ``alpha``. The estimator is the
    median of those products over finite ratios. The sign is the sign of
    ``a_1 / a_0`` (the first directional scale).
    """
    ratios = successive_ratios(derivatives)
    products: list[float] = []
    for k, ratio in enumerate(ratios, start=1):
        if math.isfinite(ratio):
            products.append(float(k) * ratio)
    if not products:
        raise ValueError("no finite successive ratio; cannot estimate scale")
    products.sort()
    mid = len(products) // 2
    if len(products) % 2 == 1:
        return products[mid]
    return 0.5 * (products[mid - 1] + products[mid])

Other public import paths

omnibias.core.local_scale_from_derivatives
omnibias.core.refine.local_scale_from_derivatives

FUNCTION

nearest_pack_index

omnibias.core.refine.nearest_pack_indexSource ↗

nearest_pack_index(packs: Sequence[RefinedPack], center: float) -> int | None

Index of the pack whose centre is closest to center.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`center`positional-or-keyword	`float`Required	See the operation contract and implementation below.

Returnsint | None

Executed regression · inherit scale empty

Full file, imports & fixtures ↗

def test_inherit_scale_empty() -> None:
    assert inherit_scale((), 0.0, default=4.0) == 4.0

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_inherit_scale_empty" -q

Read the implementation

def nearest_pack_index(
    packs: Sequence[RefinedPack],
    center: float,
) -> int | None:
    """Index of the pack whose centre is closest to ``center``."""
    if not packs:
        return None
    best = 0
    best_d = abs(packs[0].center - float(center))
    for i, pack in enumerate(packs[1:], start=1):
        d = abs(pack.center - float(center))
        if d < best_d:
            best = i
            best_d = d
    return best

FUNCTION

pack_contributions

omnibias.core.refine.pack_contributionsSource ↗

pack_contributions(terms: Sequence[float], field_norm: float) -> list[float]

||c_g p_g|| / ||u|| from already-reduced per-pack norms.

Parameters

Name	Type & default	Contract
`terms`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.
`field_norm`positional-or-keyword	`float`Required	See the operation contract and implementation below.

Returnslist[float]

Explicit validation & failure cases

When field_norm < 0.0, raises ValueError(f'field_norm must be >= 0, got {field_norm}').

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

Executed regression · g3 scale flow

Full file, imports & fixtures ↗

def test_g3_scale_flow() -> None:
    bank = _initial_bank()
    probes = jnp.linspace(0.0, 1.0, 65)

    def residual(x: Array) -> Array:
        return jnp.exp(-100.0 * x)

    _bank, report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.SCALE_FLOW,
            birth_threshold=0.1,
            death_threshold=0.0,
            min_age=10_000,
            hysteresis=1.0,
            min_scale_ratio=1.0,
        ),
        step=0,
        probe_jet=(_BL_CENTER, _BL_DERIVS),
    )
    assert report.proposed_scale == pytest.approx(_BL_SCALE)

python -m pytest "packages/omnibias-jax/tests/test_refine.py::test_g3_scale_flow" -q

Regression example · min age protects newborn

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_min_age_protects_newborn() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=1.0, age=200),
        RefinedPack(order=0, center=0.1, weight=0.0, scale=1.0, age=0),
    )
    contrib = [0.9, 0.0]
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.1))
    assert died == []

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_min_age_protects_newborn -q

Read the implementation

def pack_contributions(
    terms: Sequence[float],
    field_norm: float,
) -> list[float]:
    """``||c_g p_g|| / ||u||`` from already-reduced per-pack norms."""
    if field_norm < 0.0:
        raise ValueError(f"field_norm must be >= 0, got {field_norm}")
    denom = float(field_norm)
    if denom == 0.0:
        return [0.0 for _ in terms]
    return [abs(float(t)) / denom for t in terms]

FUNCTION

pack_term_scalar

omnibias.core.refine.pack_term_scalarSource ↗

pack_term_scalar(z: float, pack: RefinedPack, sigma_n: Callable[[float, int], float]) -> float

Scalar c * alpha^n sigma^(n)(alpha (z - center)).

Parameters

Name	Type & default	Contract
`z`positional-or-keyword	`float`Required	See the operation contract and implementation below.
`pack`positional-or-keyword	`RefinedPack`Required	See the operation contract and implementation below.
`sigma_n`positional-or-keyword	`Callable[[float, int], float]`Required	See the operation contract and implementation below.

Returnsfloat

Executed regression · birth is zero weight

Full file, imports & fixtures ↗

def test_birth_is_zero_weight() -> None:
    packs = (RefinedPack(order=0, center=0.25, weight=1.0, scale=2.0),)
    proposal = RefineProposal(move="h", center=_BL_CENTER, scale=_BL_SCALE, order=0)
    after, born = apply_birth(packs, proposal)
    assert born.weight == 0.0
    xs = [0.0, 0.01, 0.5, 1.0]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_birth_is_zero_weight" -q

Regression example · growth sibling is zero perturbation

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_growth_sibling_is_zero_perturbation() -> None:
    packs = (RefinedPack(order=0, center=0.0, weight=0.3, scale=-100.0),)
    proposal = RefineProposal(
        move="p",
        center=0.0,
        scale=-100.0,
        order=1,
        parent_index=0,
    )
    after, grown, parent = apply_growth(packs, proposal)
    assert parent == 0
    assert grown.weight == 0.0
    assert grown.order == 1
    xs = [0.0, 0.01, 0.25]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_growth_sibling_is_zero_perturbation -q

Read the implementation

def pack_term_scalar(
    z: float,
    pack: RefinedPack,
    sigma_n: Callable[[float, int], float],
) -> float:
    """Scalar ``c * alpha^n sigma^(n)(alpha (z - center))``."""
    u = pack.scale * (float(z) - pack.center)
    return pack.weight * (pack.scale ** pack.order) * float(sigma_n(u, pack.order))

FUNCTION

omnibias.core.refine.propose_refinementSource ↗

propose_refinement(packs: Sequence[RefinedPack], *, policy: RefinePolicy, peak_location: float, peak_value: float, derivatives: Sequence[float] | None, birth_score: float) -> RefineProposal | None

Turn an indicator peak into an *h* or *p* proposal, or None.

Residual / certified-error proposals inherit the nearest pack's scale (they do not carry a scale). Singularity and scale-flow read derivatives and propose that scale.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`policy`keyword-only	`RefinePolicy`Required	See the operation contract and implementation below.
`peak_location`keyword-only	`float`Required	See the operation contract and implementation below.
`peak_value`keyword-only	`float`Required	See the operation contract and implementation below.
`derivatives`keyword-only	`Sequence[float] \| None`Required	See the operation contract and implementation below.
`birth_score`keyword-only	`float`Required	See the operation contract and implementation below.

ReturnsRefineProposal | None

Executed regression · hysteresis blocks weak birth

Full file, imports & fixtures ↗

def test_hysteresis_blocks_weak_birth() -> None:
    packs = (RefinedPack(order=0, center=0.5, weight=1.0, scale=2.0),)
    policy = RefinePolicy(
        indicator=Indicator.RESIDUAL,
        birth_threshold=0.1,
        death_threshold=0.05,
        hysteresis=4.0,
    )
    # score 0.15 exceeds birth_threshold but not hysteresis * death = 0.2
    proposal = propose_refinement(
        packs,
        policy=policy,
        peak_location=0.0,
        peak_value=0.15,
        derivatives=None,
        birth_score=0.15,
    )
    assert proposal is None

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_hysteresis_blocks_weak_birth" -q

Regression example · birth is zero weight

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_birth_is_zero_weight() -> None:
    packs = (RefinedPack(order=0, center=0.25, weight=1.0, scale=2.0),)
    proposal = RefineProposal(move="h", center=_BL_CENTER, scale=_BL_SCALE, order=0)
    after, born = apply_birth(packs, proposal)
    assert born.weight == 0.0
    xs = [0.0, 0.01, 0.5, 1.0]
    before_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after_vals = [sum(pack_term_scalar(x, p, _exp_sigma) for p in after) for x in xs]
    assert_zero_perturbation(before_vals, after_vals)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_birth_is_zero_weight -q

Read the implementation

def propose_refinement(
    packs: Sequence[RefinedPack],
    *,
    policy: RefinePolicy,
    peak_location: float,
    peak_value: float,
    derivatives: Sequence[float] | None,
    birth_score: float,
) -> RefineProposal | None:
    """Turn an indicator peak into an *h* or *p* proposal, or ``None``.

    Residual / certified-error proposals inherit the nearest pack's
    scale (they do not carry a scale). Singularity and scale-flow read
    ``derivatives`` and propose that scale.
    """
    threshold = max(policy.birth_threshold, policy.hysteresis * policy.death_threshold)
    if birth_score < threshold:
        return None
    if policy.max_packs is not None and len(packs) >= int(policy.max_packs):
        return None

    inherit = policy.indicator in (Indicator.RESIDUAL, Indicator.CERTIFIED_ERROR)
    if inherit or derivatives is None:
        scale = inherit_scale(packs, peak_location)
        move: HpMove = "h" if policy.hp_rule != "always_p" else "p"
        if policy.hp_rule == "always_p":
            move = "p"
        elif policy.hp_rule == "always_h":
            move = "h"
        else:
            move = "h"
    else:
        existing = inherit_scale(packs, peak_location)
        scale = local_scale_from_derivatives(derivatives)
        if policy.indicator is Indicator.SCALE_FLOW and policy.hp_rule == "coefficient_decay":
            # Scale flow names the cutoff to add; that is an h-move.
            move = "h"
        else:
            move = hp_decision(derivatives, existing, rule=policy.hp_rule)

    parent = nearest_pack_index(packs, peak_location)
    if move == "p":
        if parent is None:
            move = "h"
        else:
            order = packs[parent].order + 1
            return RefineProposal(
                move="p",
                center=packs[parent].center,
                scale=packs[parent].scale,
                order=order,
                parent_index=parent,
                indicator_peak=float(peak_value),
                inherit_scale=True,
            )

    if _too_close(packs, float(peak_location), float(scale), policy):
        return None
    return RefineProposal(
        move="h",
        center=float(peak_location),
        scale=float(scale),
        order=int(policy.default_order),
        parent_index=parent,
        indicator_peak=float(peak_value),
        inherit_scale=inherit,
    )

FUNCTION

refine_bank

omnibias.core.refine_bankSource ↗

refine_bank(packs: Sequence[RefinedPack], policy: RefinePolicy, *, probes: Sequence[float], residuals: Sequence[float], contributions: Sequence[float], derivatives: Sequence[float] | None, certified_values: Sequence[float] | None=None, peak_location: float | None=None) -> tuple[list[RefinedPack], RefineReport]

One discrete refine step on a host-side pack list.

Age every pack, then birth or grow from the indicator peak, then die. died indexes the bank *after* appends (newborns sit at the end, age 0, so min_age protects them). Torch / jax twins apply this report to pre-allocated slots so optimizer moments of live packs stay put.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`policy`positional-or-keyword	`RefinePolicy`Required	See the operation contract and implementation below.
`probes`keyword-only	`Sequence[float]`Required	See the operation contract and implementation below.
`residuals`keyword-only	`Sequence[float]`Required	See the operation contract and implementation below.
`contributions`keyword-only	`Sequence[float]`Required	See the operation contract and implementation below.
`derivatives`keyword-only	`Sequence[float] \| None`Required	See the operation contract and implementation below.
`certified_values`keyword-only	`Sequence[float] \| None`Default: None	See the operation contract and implementation below.
`peak_location`keyword-only	`float \| None`Default: None	See the operation contract and implementation below.

Returnstuple[list[RefinedPack], RefineReport]

Explicit validation & failure cases

When len(probes) != len(residuals), raises ValueError('probes and residuals must have the same length').
When len(contributions) != len(packs), raises ValueError('contributions length must match packs').
When not probes, raises ValueError('refine_bank needs at least one probe').
When certified_values is None, raises ValueError('certified_error indicator requires certified_values').
When len(certified_values) != len(probes), raises ValueError('certified_values length must match probes').

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

Executed regression · g3 scale flow

Full file, imports & fixtures ↗

def test_g3_scale_flow() -> None:
    bank = _initial_bank()
    probes = jnp.linspace(0.0, 1.0, 65)

    def residual(x: Array) -> Array:
        return jnp.exp(-100.0 * x)

    _bank, report = refine(
        bank,
        residual,
        probes,
        RefinePolicy(
            indicator=Indicator.SCALE_FLOW,
            birth_threshold=0.1,
            death_threshold=0.0,
            min_age=10_000,
            hysteresis=1.0,
            min_scale_ratio=1.0,
        ),
        step=0,
        probe_jet=(_BL_CENTER, _BL_DERIVS),
    )
    assert report.proposed_scale == pytest.approx(_BL_SCALE)

python -m pytest "packages/omnibias-jax/tests/test_refine.py::test_g3_scale_flow" -q

Read the implementation

def refine_bank(
    packs: Sequence[RefinedPack],
    policy: RefinePolicy,
    *,
    probes: Sequence[float],
    residuals: Sequence[float],
    contributions: Sequence[float],
    derivatives: Sequence[float] | None,
    certified_values: Sequence[float] | None = None,
    peak_location: float | None = None,
) -> tuple[list[RefinedPack], RefineReport]:
    """One discrete refine step on a host-side pack list.

    Age every pack, then birth or grow from the indicator peak, then die.
    ``died`` indexes the bank *after* appends (newborns sit at the end,
    age 0, so ``min_age`` protects them). Torch / jax twins apply this
    report to pre-allocated slots so optimizer moments of live packs
    stay put.
    """
    if len(probes) != len(residuals):
        raise ValueError("probes and residuals must have the same length")
    if len(contributions) != len(packs):
        raise ValueError("contributions length must match packs")
    if not probes:
        raise ValueError("refine_bank needs at least one probe")

    aged = increment_ages(packs, 1)
    if policy.indicator is Indicator.CERTIFIED_ERROR:
        if certified_values is None:
            raise ValueError("certified_error indicator requires certified_values")
        if len(certified_values) != len(probes):
            raise ValueError("certified_values length must match probes")
        values: Sequence[float] = certified_values
    else:
        values = residuals
    peak_i, peak_v = residual_indicator(values)
    loc = float(probes[peak_i]) if peak_location is None else float(peak_location)
    proposal = propose_refinement(
        aged,
        policy=policy,
        peak_location=loc,
        peak_value=peak_v,
        derivatives=derivatives,
        birth_score=peak_v,
    )
    working = list(aged)
    born: list[RefinedPack] = []
    grown: list[int] = []
    hp: HpMove = "none"
    proposed_center: float | None = None
    proposed_scale: float | None = None
    if proposal is not None:
        proposed_center = proposal.center
        proposed_scale = proposal.scale
        hp = proposal.move
        if proposal.move == "p":
            working, _sibling, parent = apply_growth(working, proposal)
            grown.append(parent)
        else:
            working, newborn = apply_birth(working, proposal)
            born.append(newborn)

    contrib_ext = [float(c) for c in contributions]
    while len(contrib_ext) < len(working):
        contrib_ext.append(0.0)
    died = select_deaths(working, contrib_ext, policy)
    working, perturbation = apply_deaths(working, died, contrib_ext)
    report = RefineReport(
        born=tuple(born),
        grown=tuple(grown),
        died=tuple(died),
        death_perturbation=perturbation,
        indicator_values=tuple(float(v) for v in values),
        proposed_center=proposed_center,
        proposed_scale=proposed_scale,
        hp_move=hp,
    )
    return working, report

Other public import paths

omnibias.core.refine_bank
omnibias.core.refine.refine_bank

FUNCTION

residual_indicator

omnibias.core.refine.residual_indicatorSource ↗

residual_indicator(values: Sequence[float]) -> tuple[int, float]

Argmax of |value| and that magnitude.

Parameters

Name	Type & default	Contract
`values`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

Returnstuple[int, float]

Explicit validation & failure cases

When not values, raises ValueError('residual_indicator needs at least one sample').

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

Executed regression · residual argmax

Full file, imports & fixtures ↗

def test_residual_argmax() -> None:
    idx, mag = residual_indicator((0.1, -0.4, 0.2))
    assert idx == 1
    assert mag == pytest.approx(0.4)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_residual_argmax" -q

Read the implementation

def residual_indicator(values: Sequence[float]) -> tuple[int, float]:
    """Argmax of ``|value|`` and that magnitude."""
    if not values:
        raise ValueError("residual_indicator needs at least one sample")
    best = 0
    best_v = abs(float(values[0]))
    for i, raw in enumerate(values[1:], start=1):
        v = abs(float(raw))
        if v > best_v:
            best = i
            best_v = v
    return best, best_v

FUNCTION

select_deaths

omnibias.core.refine.select_deathsSource ↗

select_deaths(packs: Sequence[RefinedPack], contributions: Sequence[float], policy: RefinePolicy, *, keep_at_least: int=1) -> list[int]

Indices that may die. Newly born packs are protected by min_age.

Parameters

Name	Type & default	Contract
`packs`positional-or-keyword	`Sequence[RefinedPack]`Required	See the operation contract and implementation below.
`contributions`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.
`policy`positional-or-keyword	`RefinePolicy`Required	See the operation contract and implementation below.
`keep_at_least`keyword-only	`int`Default: 1	See the operation contract and implementation below.

Returnslist[int]

Explicit validation & failure cases

When len(packs) != len(contributions), raises ValueError('contributions length must match packs').
When keep_at_least < 0, raises ValueError(f'keep_at_least must be >= 0, got {keep_at_least}').

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

Executed regression · min age protects newborn

Full file, imports & fixtures ↗

def test_min_age_protects_newborn() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=1.0, age=200),
        RefinedPack(order=0, center=0.1, weight=0.0, scale=1.0, age=0),
    )
    contrib = [0.9, 0.0]
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.1))
    assert died == []

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_min_age_protects_newborn" -q

Regression example · death perturbation bounds measured change

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_death_perturbation_bounds_measured_change() -> None:
    packs = (
        RefinedPack(order=0, center=0.0, weight=1.0, scale=-2.0, age=200),
        RefinedPack(order=0, center=0.25, weight=0.003, scale=2.0, age=200),
    )
    xs = [i / 32.0 for i in range(33)]
    terms = []
    field_sq = 0.0
    for pack in packs:
        values = [pack_term_scalar(x, pack, _exp_sigma) for x in xs]
        terms.append(math.sqrt(sum(v * v for v in values)))
        field_sq += sum(v * v for v in values)
    field_norm = math.sqrt(field_sq)
    contrib = pack_contributions(terms, field_norm)
    died = select_deaths(packs, contrib, RefinePolicy(min_age=100, death_threshold=0.05))
    assert 1 in died
    kept, reported = apply_deaths(packs, died, contrib)
    assert len(kept) == 1
    before = [sum(pack_term_scalar(x, p, _exp_sigma) for p in packs) for x in xs]
    after = [sum(pack_term_scalar(x, p, _exp_sigma) for p in kept) for x in xs]
    delta = math.sqrt(sum((a - b) ** 2 for a, b in zip(after, before, strict=True)))
    measured = delta / field_norm
    assert death_bound_holds(measured, reported, atol=1e-15)

python -m pytest packages/omnibias-core/tests/test_refine.py -k test_death_perturbation_bounds_measured_change -q

Read the implementation

def select_deaths(
    packs: Sequence[RefinedPack],
    contributions: Sequence[float],
    policy: RefinePolicy,
    *,
    keep_at_least: int = 1,
) -> list[int]:
    """Indices that may die. Newly born packs are protected by ``min_age``."""
    if len(packs) != len(contributions):
        raise ValueError("contributions length must match packs")
    if keep_at_least < 0:
        raise ValueError(f"keep_at_least must be >= 0, got {keep_at_least}")
    candidates = [
        i
        for i, (pack, contrib) in enumerate(zip(packs, contributions, strict=True))
        if pack.age >= policy.min_age and contrib <= policy.death_threshold
    ]
    budget = len(packs) - keep_at_least
    if budget <= 0:
        return []
    candidates.sort(key=lambda i: contributions[i])
    return candidates[:budget]

FUNCTION

successive_ratios

omnibias.core.refine.successive_ratiosSource ↗

successive_ratios(derivatives: Sequence[float]) -> list[float]

Taylor-coefficient ratios a_k / a_{k-1} for k >= 1.

Parameters

Name	Type & default	Contract
`derivatives`positional-or-keyword	`Sequence[float]`Required	See the operation contract and implementation below.

Returnslist[float]

Explicit validation & failure cases

When len(coeffs) < 2, raises ValueError('need at least two derivatives for a ratio').

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

Executed regression · boundary layer local scale

Full file, imports & fixtures ↗

def test_boundary_layer_local_scale() -> None:
    scale = local_scale_from_derivatives(_BL_DERIVS)
    assert scale == pytest.approx(_BL_SCALE, rel=1e-12)

python -m pytest "packages/omnibias-core/tests/test_refine.py::test_boundary_layer_local_scale" -q

Read the implementation

def successive_ratios(derivatives: Sequence[float]) -> list[float]:
    """Taylor-coefficient ratios ``a_k / a_{k-1}`` for ``k >= 1``."""
    coeffs = taylor_coeffs_from_derivatives(derivatives)
    if len(coeffs) < 2:
        raise ValueError("need at least two derivatives for a ratio")
    out: list[float] = []
    for k in range(1, len(coeffs)):
        prev = coeffs[k - 1]
        if prev == 0.0 or not math.isfinite(prev):
            out.append(float("nan"))
        else:
            out.append(coeffs[k] / prev)
    return out

← Package overview Runnable learning examples →

omnibias.core.refine

HpMove​

HpRule​

Indicator​

RefinePolicy​

Parameters

Fields & methods

RefineProposal​

Parameters

Fields & methods

RefineReport​

Parameters

Fields & methods

RefinedPack​

Parameters

Fields & methods

Parameters

Parameters

Parameters

apply_birth​

Parameters

apply_deaths​

Parameters

apply_growth​

Parameters

assert_zero_perturbation​

Parameters

certified_local_error​

Parameters

death_bound_holds​

Parameters

domb_sykes_fit​

Parameters

factorial_scale_term​

Parameters

hp_decision​

Parameters

increment_ages​

Parameters

inherit_scale​

Parameters

jet_order_needed​

Parameters

local_scale_from_derivatives​

Parameters

nearest_pack_index​

Parameters

pack_contributions​

Parameters

pack_term_scalar​

Parameters

propose_refinement​

Parameters

refine_bank​

Parameters

residual_indicator​

Parameters

select_deaths​

Parameters

successive_ratios​

Parameters

HpMove

HpRule

Indicator

RefinePolicy

RefineProposal

RefineReport

RefinedPack

apply_birth

apply_deaths

apply_growth

assert_zero_perturbation

certified_local_error

death_bound_holds

domb_sykes_fit

factorial_scale_term

hp_decision

increment_ages

inherit_scale

jet_order_needed

local_scale_from_derivatives

nearest_pack_index

pack_contributions

pack_term_scalar

propose_refinement

refine_bank

residual_indicator

select_deaths

successive_ratios