# balancer\_pool.py ## NOTE: this file was used as the initial porting of BPool code and the basis for system\_policies. Not used in the model. Path: [balancer\_pool.py](https://github.com/TokenEngineeringCommunity/BalancerAMM_Model/blob/main/model/balancer_pool.py) file discussed in this article.\ \ The file starts with importing balancer constants, and BalancerMath ``` from model.balancer_constants import MAX_TOTAL_WEIGHT, MAX_WEIGHT, MIN_BALANCE, MIN_FEE, MAX_BOUND_TOKENS, INIT_POOL_SUPPLY, EXIT_FEE, MAX_IN_RATIO, MAX_OUT_RATIO, MIN_WEIGHT from model.balancer_math import BalancerMath from dataclasses import dataclass ``` Data classes for storing information about the pool itself and swaps ``` @dataclass class TokenRecord: bound: bool name: str denorm: Decimal balance: Decimal @dataclass class SwapInResult: token_amount_out: Decimal spot_price_after: Decimal @dataclass class SwapOutResult: token_amount_in: Decimal spot_price_after: Decimal ``` The BalancerMath class and related functions ``` class BalancerPool(BalancerMath): ``` ### Balancer Pool functions Pool initialization with initial balances and Pool Shares ``` def __init__(self, initial_pool_supply: Decimal = INIT_POOL_SUPPLY): self._swap_fee = MIN_FEE self._records = {} # NOTE: we are assuming python 3.6+ ordered dictionaries self.total_weight = Decimal('0') self._pool_token_supply = initial_pool_supply self.factory_fees = Decimal('0') ``` Functions for requesting tokens weights ``` def get_total_denorm_weight(self): return self.total_weight def get_denorm_weight(self, token: str): return self._records[token].denorm def get_normal_weight(self, token: str): denorm = self._records[token].denorm return denorm / self.total_weight ``` Functions related to balances of tokens and Pool Shares ``` def get_balance(self, token: str): if self._records.get(token) is not None: return self._records[token].balance else: return 0 def get_num_tokens(self): return len(self._records) def get_pool_token_supply(self): return self._pool_token_supply ``` Functions for minting and burning Pool Shares (related to join and exit the pool) ``` def _mint_pool_share(self, amount: Decimal): self._pool_token_supply += amount def _burn_pool_share(self, amount: Decimal): self._pool_token_supply -= amount ``` Set swap fee function ``` def set_swap_fee(self, amount: Decimal): self._swap_fee = amount ``` Functions for receiving spot price (with and without fee) ``` def get_spot_price(self, token_in: str, token_out: str) -> Decimal: min_pool_amount_out = self._records[token_in] out_record = self._records[token_out] return self.calc_spot_price(min_pool_amount_out.balance, min_pool_amount_out.denorm, out_record.balance, out_record.denorm, self._swap_fee) def get_spot_price_sans_fee(self, token_in: str, token_out: str) -> Decimal: min_pool_amount_out = self._records[token_in] out_record = self._records[token_out] return self.calc_spot_price(min_pool_amount_out.balance, min_pool_amount_out.denorm, out_record.balance, out_record.denorm, Decimal('0')) ``` Join and exit pool functions (liquidity provision and liquidity withdrawal) ``` def join_pool(self, pool_amount_out: Decimal, max_amounts_in: dict) -> dict: ratio = pool_amount_out / self._pool_token_supply if ratio == 0: return Exception("ERR_MATH_APPROX") results = {} for token in self._records: record = self._records[token] token_amount_in = ratio * record.balance if token_amount_in == 0: return Exception("ERR_MATH_APPROX") if token_amount_in > max_amounts_in[token]: raise Exception('ERR_LIMIT_IN') record.balance += token_amount_in results[token] = token_amount_in self._mint_pool_share(pool_amount_out) return results def exit_pool(self, pool_amount_in: Decimal, min_amounts_out: dict) -> dict: pool_total = self._pool_token_supply exit_fee = pool_amount_in * EXIT_FEE pool_amount_in_afer_exit_fee = pool_amount_in - exit_fee ratio = pool_amount_in_afer_exit_fee / pool_total return_dict = { "exit_fee_pool_token": exit_fee } self._burn_pool_share(pool_amount_in_afer_exit_fee) for token in self._records: record = self._records[token] token_amount_out = ratio * record.balance if token_amount_out == 0: raise Exception("ERR_MATH_APPROX") if token_amount_out < min_amounts_out[token]: raise Exception("ERR_LIMIT_OUT") record.balance -= token_amount_out return_dict[token] = token_amount_out return return_dict ``` Token swaps functions (two of them - for exact amount in and for exact amount out) ``` def swap_exact_amount_in(self, token_in: str, token_amount_in: Decimal, token_out: str, min_amount_out: Decimal, max_price: Decimal) -> SwapInResult: min_pool_amount_out = self._records[token_in] if not min_pool_amount_out.bound: raise Exception('ERR_NOT_BOUND') out_record = self._records[token_out] if not out_record.bound: raise Exception('ERR_NOT_BOUND') if token_amount_in > min_pool_amount_out.balance * MAX_IN_RATIO: raise Exception("ERR_MAX_IN_RATIO") spot_price_before = self.calc_spot_price( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, swap_fee=self._swap_fee ) if spot_price_before > max_price: raise Exception("ERR_BAD_LIMIT_PRICE") token_amount_out = self.calc_out_given_in( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, token_amount_in=token_amount_in, swap_fee=self._swap_fee ) # TODO require(token_amount_out >= min_amount_out, "ERR_LIMIT_OUT") min_pool_amount_out.balance += token_amount_in out_record.balance -= token_amount_out spot_price_after = self.calc_spot_price( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, swap_fee=self._swap_fee ) if spot_price_after < spot_price_before: raise Exception("ERR_MATH_APPROX") if spot_price_after > max_price: raise Exception("ERR_LIMIT_PRICE") if spot_price_before > (token_amount_in / token_amount_out): raise Exception("ERR_MATH_APPROX") # NOTE: we are not doing user balances yet # _pullUnderlying(token_in, msg.sender, token_amount_in) # _pushUnderlying(token_out, msg.sender, token_amount_out) return SwapInResult(token_amount_out, spot_price_after) def swap_exact_amount_out(self, token_in: str, max_amount_in: Decimal, token_out: str, token_amount_out: Decimal, max_price: Decimal) -> SwapOutResult: min_pool_amount_out = self._records[token_in] if not min_pool_amount_out.bound: raise Exception('ERR_NOT_BOUND') out_record = self._records[token_out] if not out_record.bound: raise Exception('ERR_NOT_BOUND') if token_amount_out > (out_record.balance * MAX_OUT_RATIO): raise Exception("ERR_MAX_OUT_RATIO") spot_price_before = self.calc_spot_price( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, swap_fee=self._swap_fee ) if spot_price_before > max_price: raise Exception('ERR_BAD_LIMIT_PRICE') token_amount_in = self.calc_in_given_out( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, token_amount_out=token_amount_out, swap_fee=self._swap_fee ) if token_amount_in > max_amount_in: raise Exception('ERR_LIMIT_IN') min_pool_amount_out.balance += token_amount_in out_record.balance -= token_amount_out spot_price_after = self.calc_spot_price( token_balance_in=min_pool_amount_out.balance, token_weight_in=min_pool_amount_out.denorm, token_balance_out=out_record.balance, token_weight_out=out_record.denorm, swap_fee=self._swap_fee ) if spot_price_after < spot_price_before: raise Exception('ERR_MATH_APPROX') if spot_price_after > max_price: raise Exception('LIMIT PRICE') if spot_price_before > (token_amount_in / token_amount_out): raise Exception('ERR_MATH_APPROX') # NOTE not modeling balance change for sender # _pullUnderlying(token_in, msg.sender, token_amount_in) # _pushUnderlying(token_out, msg.sender, token_amount_out) return SwapOutResult(token_amount_in=token_amount_in, spot_price_after=spot_price_after) ``` Join pool functions (liquidity provision) ``` def join_swap_extern_amount_in(self, token_in: str, token_amount_in: Decimal, min_pool_amount_out: Decimal) -> Decimal: # require(_finalized, "ERR_NOT_FINALIZED") if not self._records[token_in].bound: raise Exception("ERR_NOT_BOUND") if token_amount_in > self._records[token_in].balance * MAX_IN_RATIO: raise Exception("ERR_MAX_IN_RATIO") in_record = self._records[token_in] pool_amount_out = self.calc_pool_out_given_single_in( token_balance_in=in_record.balance, token_weight_in=in_record.denorm, pool_supply=self._pool_token_supply, total_weight=self.total_weight, token_amount_in=token_amount_in, swap_fee=self._swap_fee ) if pool_amount_out < min_pool_amount_out: raise Exception("ERR_LIMIT_OUT") in_record.balance += token_amount_in self._mint_pool_share(pool_amount_out) # NOTE user balance can be inferred from params (substract tai), pool out is already returning # _pushPoolShare(msg.sender, pool_amount_out) # _pullUnderlying(token_in, msg.sender, token_amount_in) return pool_amount_out def join_swap_pool_amount_out(self, token_in: str, pool_amount_out: Decimal, max_amount_in: Decimal) -> Decimal: if not self._records[token_in].bound: raise Exception("ERR_NOT_BOUND") in_record = self._records[token_in] token_amount_in = self.calc_single_in_given_pool_out( token_balance_in = in_record.balance, token_weight_in = in_record.denorm, pool_supply = self._pool_token_supply, total_weight = self.total_weight, pool_amount_out = pool_amount_out, swap_fee = self._swap_fee) if token_amount_in == 0: raise Exception("ERR_MATH_APPROX") if token_amount_in > max_amount_in: raise Exception("ERR_LIMIT_IN") if token_amount_in > in_record.balance * MAX_IN_RATIO: raise Exception("ERR_MAX_IN_RATIO") in_record.balance = in_record.balance + token_amount_in self._mint_pool_share(pool_amount_out) # NOTE not modeling balance change for sender # _pushPoolShare(msg.sender, poolAmountOut) # _pullUnderlying(tokenIn, msg.sender, tokenAmountIn) return token_amount_in ``` Exit pool functions (liquidity withdrawal) ``` def exit_swap_pool_amount_in(self, token_out: str, pool_amount_in: Decimal, min_amount_out: Decimal) -> Decimal: if not self._records[token_out].bound: raise Exception("ERR_NOT_BOUND") out_record = self._records[token_out] token_amount_out = self.calc_single_out_given_pool_in( token_balance_out=out_record.balance, token_weight_out=out_record.denorm, pool_supply=self._pool_token_supply, total_weight=self.total_weight, pool_amount_in=pool_amount_in, swap_fee=self._swap_fee) if token_amount_out < min_amount_out: raise Exception("ERR_LIMIT_OUT") if token_amount_out > out_record.balance * MAX_OUT_RATIO: raise Exception("ERR_MAX_OUT_RATIO") out_record.balance = out_record.balance - token_amount_out exit_fee = pool_amount_in * EXIT_FEE self._burn_pool_share(pool_amount_in - exit_fee) # _pushPoolShare(_factory, exit_fee) # _pushUnderlying(token_out, msg.sender, token_amount_out) return token_amount_out def exit_swap_extern_amount_out(self, token_out: str, token_amount_out: Decimal, max_pool_amount_in: Decimal) -> Decimal: if not self._records[token_out].bound: raise Exception("ERR_NOT_BOUND") out_record = self._records[token_out] if token_amount_out > out_record.balance * MAX_OUT_RATIO: raise Exception("ERR_MAX_OUT_RATIO") pool_amount_in = self.calc_pool_in_given_single_out( token_balance_out=out_record.balance, token_weight_out=out_record.denorm, pool_supply=self._pool_token_supply, total_weight=self.total_weight, token_amount_out=token_amount_out, swap_fee=self._swap_fee ) if pool_amount_in == 0: raise Exception("ERR_MATH_APPROX") if pool_amount_in > max_pool_amount_in: raise Exception("ERR_LIMIT_IN") out_record.balance -= token_amount_out exitFee = pool_amount_in * EXIT_FEE self._burn_pool_share(pool_amount_in - exitFee) #_pushPoolShare(_factory, exitFee); return pool_amount_in; ``` --- # Agent Instructions: Querying This Documentation If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question. Perform an HTTP GET request on the current page URL with the `ask` query parameter: ``` GET https://token-engineering-balancer.gitbook.io/balancer-simulations/additional-code-and-instructions/balancer-the-python-edition/balancer_pool.py.md?ask= ``` The question should be specific, self-contained, and written in natural language. The response will contain a direct answer to the question and relevant excerpts and sources from the documentation. Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.