degenbot.balancer.pools

Balancer V2 weighted pool implementation.

Module Contents

degenbot.balancer.pools.detect_pow_version(bytecode: str) → degenbot.balancer.libraries.constants.PowVersion

Detect which FixedPoint library version a pool contract uses from its bytecode.

V2 (WeightedPool) contracts include fast paths for y == ONE, TWO, FOUR in powDown/powUp. These reference the TWO and FOUR constants, which are absent from V1 (WeightedPool2Tokens) bytecode.

The detection works by checking for the TWO constant (0x1bc16d674ec80000) in the deployed bytecode. This constant is used in the y == TWO fast-path comparison and does not appear in V1 contracts.

Returns:

The computed value.

class degenbot.balancer.pools.BalancerV2Pool(*args: Any, **kwargs: Any)

Bases: degenbot.types.abstract.AbstractLiquidityPool

BalancerV2Pool class.

variant: ClassVar[str | None] = 'balancer_weighted'

type PoolState = BalancerV2PoolState

FEE_DENOMINATOR = 1000000000000000000

address: degenbot.types.chain.ChecksummedAddress

pool_id: bytes

pool_specialization: int

vault: degenbot.types.chain.ChecksummedAddress

scaling_factors: tuple[int, ...]

fee: fractions.Fraction

weights: tuple[int, ...]

pow_version: degenbot.balancer.libraries.constants.PowVersion

classmethod from_handle(py_pool: degenbot.types.Pool) → Self

Wrap a Rust-owned Pool handle as a Python companion.

Internal seam (ADR-005, Polars-style _from_pydf pattern). Every identity field (vault, pool_id, tokens, weights, scaling_factors, swap_fee, pow_version, address) is read off the handle.

Returns:

A cls instance wrapping py_pool.

Raises:

DegenbotValueError – If the handle is not a Balancer weighted pool or any token is not registered.

property balances: tuple[int, ...]

Balances.

Read from the Rust core via the Pool handle (ADR-005 slice 12b). Rust BotState is the single source of truth for the mutable balances slot; this getter returns the live tuple.

property state: PoolState

State.

Built from one atomic Rust snapshot (snapshot_balancer_weighted() — (balances, block)) so callers see a coherent tuple (no torn read mid-external_update). Mirrors V3/V4’s snapshot_v3() and Curve’s snapshot_curve() contract.

Raises:

DegenbotValueError – If the Rust snapshot is absent (the pool is not registered in Rust as a Balancer weighted pool — unreachable for a companion built over a registered handle).

property tokens: tuple[degenbot.erc20.Erc20Token, ...]

Tokens.

calculate_tokens_out_from_tokens_in(token_in: degenbot.erc20.Erc20Token, token_out: degenbot.erc20.Erc20Token, token_in_quantity: int, override_state: PoolState | None = None) → int

Calculate tokens out from tokens in.

Thin driver shell over the Rust core: token resolution stays Python-side, the swap math is fully Rust-owned.

Returns:

The computed integer value.

calculate_tokens_in_from_tokens_out(token_in: degenbot.erc20.Erc20Token, token_out: degenbot.erc20.Erc20Token, token_out_quantity: int, override_state: PoolState | None = None) → int

Compute how many tokens must be sent to take token_out_quantity out.

Thin driver shell over the Rust core.

Returns:

The computed integer value.

external_update(update: degenbot.balancer.types.BalancerV2WeightedPoolExternalUpdate) → None

Apply an external state update with new balances.

Delegates to the Rust core (Pool.apply_balancer_weighted_balance_update) which journals the prior balances (genesis-anchor V2-style discipline) and lands the new balances + update_block atomically (ADR-005 slice 12b). The _state_lock + double-check-after-acquire pattern it used to apply is gone — Rust’s internal write lock handles atomicity; the registration-state precondition is enforced by the Rust core’s silent-no-op-on-older-block contract.

Raises:

DegenbotValueError – If the Rust core rejects the update (the pool is not registered as a Balancer weighted pool — unreachable for a companion built over a registered handle).