文件 1 的 1：Vyper_contract.vy
# @version 0.3.1
"""
@title StableSwap
@author Curve.Fi
@license Copyright (c) Curve.Fi, 2020-2021 - all rights reserved
"""
from vyper.interfaces import ERC20
interface ERC1271:
def isValidSignature(_hash: bytes32, _signature: Bytes[65]) -> bytes32: view
event Transfer:
sender: indexed(address)
receiver: indexed(address)
value: uint256
event Approval:
owner: indexed(address)
spender: indexed(address)
value: uint256
event TokenExchange:
buyer: indexed(address)
sold_id: int128
tokens_sold: uint256
bought_id: int128
tokens_bought: uint256
event AddLiquidity:
provider: indexed(address)
token_amounts: uint256[N_COINS]
fees: uint256[N_COINS]
invariant: uint256
token_supply: uint256
event RemoveLiquidity:
provider: indexed(address)
token_amounts: uint256[N_COINS]
fees: uint256[N_COINS]
token_supply: uint256
event RemoveLiquidityOne:
provider: indexed(address)
token_amount: uint256
coin_amount: uint256
token_supply: uint256
event RemoveLiquidityImbalance:
provider: indexed(address)
token_amounts: uint256[N_COINS]
fees: uint256[N_COINS]
invariant: uint256
token_supply: uint256
event RampA:
old_A: uint256
new_A: uint256
initial_time: uint256
future_time: uint256
event StopRampA:
A: uint256
t: uint256
event TransferOwnership:
_old_owner: address
_new_owner: address
N_COINS: constant(int128) = 4
PRECISION: constant(uint256) = 10 ** 18
FEE_DENOMINATOR: constant(uint256) = 10 ** 10
ADMIN_FEE: constant(uint256) = 5000000000
A_PRECISION: constant(uint256) = 100
MAX_A: constant(uint256) = 10 ** 6
MAX_A_CHANGE: constant(uint256) = 10
MIN_RAMP_TIME: constant(uint256) = 86400
EIP712_TYPEHASH: constant(bytes32) = keccak256("EIP712Domain(string name,string version,uint256 chainId,address verifyingContract)")
PERMIT_TYPEHASH: constant(bytes32) = keccak256("Permit(address owner,address spender,uint256 value,uint256 nonce,uint256 deadline)")
# keccak256("isValidSignature(bytes32,bytes)")[:4] << 224
ERC1271_MAGIC_VAL: constant(bytes32) = 0x1626ba7e00000000000000000000000000000000000000000000000000000000
VERSION: constant(String[8]) = "v5.0.0"
COINS: immutable(address[N_COINS])
RATE_MULTIPLIERS: immutable(uint256[N_COINS])
NAME: immutable(String[64])
SYMBOL: immutable(String[32])
DOMAIN_SEPARATOR: immutable(bytes32)
admin_balances: public(uint256[N_COINS])
fee: public(uint256) # fee * 1e10
initial_A: public(uint256)
future_A: public(uint256)
initial_A_time: public(uint256)
future_A_time: public(uint256)
balanceOf: public(HashMap[address, uint256])
allowance: public(HashMap[address, HashMap[address, uint256]])
totalSupply: public(uint256)
nonces: public(HashMap[address, uint256])
owner: public(address)
future_owner: public(address)
@external
def __init__(
_name: String[64],
_symbol: String[32],
_coins: address[4],
_rate_multipliers: uint256[4],
_A: uint256,
_fee: uint256,
):
"""
@notice Contract constructor
@param _name Name of the new pool
@param _symbol Token symbol
@param _coins List of all ERC20 conract addresses of coins
@param _rate_multipliers List of number of decimals in coins
@param _A Amplification coefficient multiplied by n ** (n - 1)
@param _fee Fee to charge for exchanges
"""
COINS = _coins
RATE_MULTIPLIERS = _rate_multipliers
A: uint256 = _A * A_PRECISION
self.initial_A = A
self.future_A = A
self.fee = _fee
self.owner = msg.sender
NAME = _name
SYMBOL = _symbol
DOMAIN_SEPARATOR = keccak256(
_abi_encode(EIP712_TYPEHASH, keccak256(_name), keccak256(VERSION), chain.id, self)
)
# fire a transfer event so block explorers identify the contract as an ERC20
log Transfer(ZERO_ADDRESS, self, 0)
### ERC20 Functionality ###
@view
@external
def decimals() -> uint256:
"""
@notice Get the number of decimals for this token
@dev Implemented as a view method to reduce gas costs
@return uint256 decimal places
"""
return 18
@internal
def _transfer(_from: address, _to: address, _value: uint256):
# # NOTE: vyper does not allow underflows
# # so the following subtraction would revert on insufficient balance
self.balanceOf[_from] -= _value
self.balanceOf[_to] += _value
log Transfer(_from, _to, _value)
@external
def transfer(_to : address, _value : uint256) -> bool:
"""
@dev Transfer token for a specified address
@param _to The address to transfer to.
@param _value The amount to be transferred.
"""
self._transfer(msg.sender, _to, _value)
return True
@external
def transferFrom(_from : address, _to : address, _value : uint256) -> bool:
"""
@dev Transfer tokens from one address to another.
@param _from address The address which you want to send tokens from
@param _to address The address which you want to transfer to
@param _value uint256 the amount of tokens to be transferred
"""
self._transfer(_from, _to, _value)
_allowance: uint256 = self.allowance[_from][msg.sender]
if _allowance != MAX_UINT256:
self.allowance[_from][msg.sender] = _allowance - _value
return True
@external
def approve(_spender : address, _value : uint256) -> bool:
"""
@notice Approve the passed address to transfer the specified amount of
tokens on behalf of msg.sender
@dev Beware that changing an allowance via this method brings the risk that
someone may use both the old and new allowance by unfortunate transaction
ordering: https://github.com/ethereum/EIPs/issues/20#issuecomment-263524729
@param _spender The address which will transfer the funds
@param _value The amount of tokens that may be transferred
@return bool success
"""
self.allowance[msg.sender][_spender] = _value
log Approval(msg.sender, _spender, _value)
return True
@external
def permit(
_owner: address,
_spender: address,
_value: uint256,
_deadline: uint256,
_v: uint8,
_r: bytes32,
_s: bytes32
) -> bool:
"""
@notice Approves spender by owner's signature to expend owner's tokens.
See https://eips.ethereum.org/EIPS/eip-2612.
@dev Inspired by https://github.com/yearn/yearn-vaults/blob/main/contracts/Vault.vy#L753-L793
@dev Supports smart contract wallets which implement ERC1271
https://eips.ethereum.org/EIPS/eip-1271
@param _owner The address which is a source of funds and has signed the Permit.
@param _spender The address which is allowed to spend the funds.
@param _value The amount of tokens to be spent.
@param _deadline The timestamp after which the Permit is no longer valid.
@param _v The bytes[64] of the valid secp256k1 signature of permit by owner
@param _r The bytes[0:32] of the valid secp256k1 signature of permit by owner
@param _s The bytes[32:64] of the valid secp256k1 signature of permit by owner
@return True, if transaction completes successfully
"""
assert _owner != ZERO_ADDRESS
assert block.timestamp <= _deadline
nonce: uint256 = self.nonces[_owner]
digest: bytes32 = keccak256(
concat(
b"\x19\x01",
DOMAIN_SEPARATOR,
keccak256(_abi_encode(PERMIT_TYPEHASH, _owner, _spender, _value, nonce, _deadline))
)
)
if _owner.is_contract:
sig: Bytes[65] = concat(_abi_encode(_r, _s), slice(convert(_v, bytes32), 31, 1))
# reentrancy not a concern since this is a staticcall
assert ERC1271(_owner).isValidSignature(digest, sig) == ERC1271_MAGIC_VAL
else:
assert ecrecover(digest, convert(_v, uint256), convert(_r, uint256), convert(_s, uint256)) == _owner
self.allowance[_owner][_spender] = _value
self.nonces[_owner] = nonce + 1
log Approval(_owner, _spender, _value)
return True
### StableSwap Functionality ###
@view
@internal
def _balances() -> uint256[N_COINS]:
result: uint256[N_COINS] = empty(uint256[N_COINS])
for i in range(N_COINS):
result[i] = ERC20(COINS[i]).balanceOf(self) - self.admin_balances[i]
return result
@view
@external
def balances(i: uint256) -> uint256:
"""
@notice Get the current balance of a coin within the
pool, less the accrued admin fees
@param i Index value for the coin to query balance of
@return Token balance
"""
return self._balances()[i]
@view
@external
def get_balances() -> uint256[N_COINS]:
return self._balances()
@view
@internal
def _A() -> uint256:
"""
Handle ramping A up or down
"""
t1: uint256 = self.future_A_time
A1: uint256 = self.future_A
if block.timestamp < t1:
A0: uint256 = self.initial_A
t0: uint256 = self.initial_A_time
# Expressions in uint256 cannot have negative numbers, thus "if"
if A1 > A0:
return A0 + (A1 - A0) * (block.timestamp - t0) / (t1 - t0)
else:
return A0 - (A0 - A1) * (block.timestamp - t0) / (t1 - t0)
else: # when t1 == 0 or block.timestamp >= t1
return A1
@view
@external
def admin_fee() -> uint256:
return ADMIN_FEE
@view
@external
def A() -> uint256:
return self._A() / A_PRECISION
@view
@external
def A_precise() -> uint256:
return self._A()
@pure
@internal
def _xp_mem(_rates: uint256[N_COINS], _balances: uint256[N_COINS]) -> uint256[N_COINS]:
result: uint256[N_COINS] = empty(uint256[N_COINS])
for i in range(N_COINS):
result[i] = _rates[i] * _balances[i] / PRECISION
return result
@pure
@internal
def get_D(_xp: uint256[N_COINS], _amp: uint256) -> uint256:
"""
D invariant calculation in non-overflowing integer operations
iteratively
A * sum(x_i) * n**n + D = A * D * n**n + D**(n+1) / (n**n * prod(x_i))
Converging solution:
D[j+1] = (A * n**n * sum(x_i) - D[j]**(n+1) / (n**n prod(x_i))) / (A * n**n - 1)
"""
S: uint256 = 0
Dprev: uint256 = 0
for x in _xp:
S += x
if S == 0:
return 0
D: uint256 = S
Ann: uint256 = _amp * N_COINS
for i in range(255):
D_P: uint256 = D
for x in _xp:
D_P = D_P * D / (x * N_COINS) # If division by 0, this will be borked: only withdrawal will work. And that is good
Dprev = D
D = (Ann * S / A_PRECISION + D_P * N_COINS) * D / ((Ann - A_PRECISION) * D / A_PRECISION + (N_COINS + 1) * D_P)
# Equality with the precision of 1
if D > Dprev:
if D - Dprev <= 1:
return D
else:
if Dprev - D <= 1:
return D
# convergence typically occurs in 4 rounds or less, this should be unreachable!
# if it does happen the pool is borked and LPs can withdraw via `remove_liquidity`
raise
@view
@internal
def get_D_mem(_rates: uint256[N_COINS], _balances: uint256[N_COINS], _amp: uint256) -> uint256:
xp: uint256[N_COINS] = self._xp_mem(_rates, _balances)
return self.get_D(xp, _amp)
@view
@external
def get_virtual_price() -> uint256:
"""
@notice The current virtual price of the pool LP token
@dev Useful for calculating profits
@return LP token virtual price normalized to 1e18
"""
amp: uint256 = self._A()
balances: uint256[N_COINS] = self._balances()
xp: uint256[N_COINS] = self._xp_mem(RATE_MULTIPLIERS, balances)
D: uint256 = self.get_D(xp, amp)
# D is in the units similar to DAI (e.g. converted to precision 1e18)
# When balanced, D = n * x_u - total virtual value of the portfolio
return D * PRECISION / self.totalSupply
@view
@external
def calc_token_amount(_amounts: uint256[N_COINS], _is_deposit: bool) -> uint256:
"""
@notice Calculate addition or reduction in token supply from a deposit or withdrawal
@dev This calculation accounts for slippage, but not fees.
Needed to prevent front-running, not for precise calculations!
@param _amounts Amount of each coin being deposited
@param _is_deposit set True for deposits, False for withdrawals
@return Expected amount of LP tokens received
"""
amp: uint256 = self._A()
balances: uint256[N_COINS] = self._balances()
D0: uint256 = self.get_D_mem(RATE_MULTIPLIERS, balances, amp)
for i in range(N_COINS):
amount: uint256 = _amounts[i]
if _is_deposit:
balances[i] += amount
else:
balances[i] -= amount
D1: uint256 = self.get_D_mem(RATE_MULTIPLIERS, balances, amp)
diff: uint256 = 0
if _is_deposit:
diff = D1 - D0
else:
diff = D0 - D1
return diff * self.totalSupply / D0
@external
@nonreentrant('lock')
def add_liquidity(
_amounts: uint256[N_COINS],
_min_mint_amount: uint256,
_receiver: address = msg.sender
) -> uint256:
"""
@notice Deposit coins into the pool
@param _amounts List of amounts of coins to deposit
@param _min_mint_amount Minimum amount of LP tokens to mint from the deposit
@param _receiver Address that owns the minted LP tokens
@return Amount of LP tokens received by depositing
"""
amp: uint256 = self._A()
old_balances: uint256[N_COINS] = self._balances()
rates: uint256[N_COINS] = RATE_MULTIPLIERS
# Initial invariant
D0: uint256 = self.get_D_mem(rates, old_balances, amp)
total_supply: uint256 = self.totalSupply
new_balances: uint256[N_COINS] = old_balances
for i in range(N_COINS):
amount: uint256 = _amounts[i]
if amount > 0:
coin: address = COINS[i]
initial: uint256 = ERC20(coin).balanceOf(self)
response: Bytes[32] = raw_call(
coin,
concat(
method_id("transferFrom(address,address,uint256)"),
convert(msg.sender, bytes32),
convert(self, bytes32),
convert(amount, bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool) # dev: failed transfer
new_balances[i] += ERC20(coin).balanceOf(self) - initial
else:
assert total_supply != 0 # dev: initial deposit requires all coins
# Invariant after change
D1: uint256 = self.get_D_mem(rates, new_balances, amp)
assert D1 > D0
# We need to recalculate the invariant accounting for fees
# to calculate fair user's share
fees: uint256[N_COINS] = empty(uint256[N_COINS])
mint_amount: uint256 = 0
if total_supply > 0:
# Only account for fees if we are not the first to deposit
base_fee: uint256 = self.fee * N_COINS / (4 * (N_COINS - 1))
for i in range(N_COINS):
ideal_balance: uint256 = D1 * old_balances[i] / D0
difference: uint256 = 0
new_balance: uint256 = new_balances[i]
if ideal_balance > new_balance:
difference = ideal_balance - new_balance
else:
difference = new_balance - ideal_balance
fees[i] = base_fee * difference / FEE_DENOMINATOR
self.admin_balances[i] += fees[i] * ADMIN_FEE / FEE_DENOMINATOR
new_balances[i] -= fees[i]
D2: uint256 = self.get_D_mem(rates, new_balances, amp)
mint_amount = total_supply * (D2 - D0) / D0
else:
mint_amount = D1 # Take the dust if there was any
assert mint_amount >= _min_mint_amount, "Slippage screwed you"
# Mint pool tokens
total_supply += mint_amount
self.balanceOf[_receiver] += mint_amount
self.totalSupply = total_supply
log Transfer(ZERO_ADDRESS, _receiver, mint_amount)
log AddLiquidity(msg.sender, _amounts, fees, D1, total_supply)
return mint_amount
@view
@internal
def get_y(i: int128, j: int128, x: uint256, xp: uint256[N_COINS]) -> uint256:
"""
Calculate x[j] if one makes x[i] = x
Done by solving quadratic equation iteratively.
x_1**2 + x_1 * (sum' - (A*n**n - 1) * D / (A * n**n)) = D ** (n + 1) / (n ** (2 * n) * prod' * A)
x_1**2 + b*x_1 = c
x_1 = (x_1**2 + c) / (2*x_1 + b)
"""
# x in the input is converted to the same price/precision
assert i != j # dev: same coin
assert j >= 0 # dev: j below zero
assert j < N_COINS # dev: j above N_COINS
# should be unreachable, but good for safety
assert i >= 0
assert i < N_COINS
amp: uint256 = self._A()
D: uint256 = self.get_D(xp, amp)
S_: uint256 = 0
_x: uint256 = 0
y_prev: uint256 = 0
c: uint256 = D
Ann: uint256 = amp * N_COINS
for _i in range(N_COINS):
if _i == i:
_x = x
elif _i != j:
_x = xp[_i]
else:
continue
S_ += _x
c = c * D / (_x * N_COINS)
c = c * D * A_PRECISION / (Ann * N_COINS)
b: uint256 = S_ + D * A_PRECISION / Ann # - D
y: uint256 = D
for _i in range(255):
y_prev = y
y = (y*y + c) / (2 * y + b - D)
# Equality with the precision of 1
if y > y_prev:
if y - y_prev <= 1:
return y
else:
if y_prev - y <= 1:
return y
raise
@view
@external
def get_dy(i: int128, j: int128, dx: uint256) -> uint256:
"""
@notice Calculate the current output dy given input dx
@dev Index values can be found via the `coins` public getter method
@param i Index value for the coin to send
@param j Index valie of the coin to recieve
@param dx Amount of `i` being exchanged
@return Amount of `j` predicted
"""
rates: uint256[N_COINS] = RATE_MULTIPLIERS
xp: uint256[N_COINS] = self._xp_mem(rates, self._balances())
x: uint256 = xp[i] + (dx * rates[i] / PRECISION)
y: uint256 = self.get_y(i, j, x, xp)
dy: uint256 = xp[j] - y - 1
fee: uint256 = self.fee * dy / FEE_DENOMINATOR
return (dy - fee) * PRECISION / rates[j]
@external
@nonreentrant('lock')
def exchange(
i: int128,
j: int128,
_dx: uint256,
_min_dy: uint256,
_receiver: address = msg.sender,
) -> uint256:
"""
@notice Perform an exchange between two coins
@dev Index values can be found via the `coins` public getter method
@param i Index value for the coin to send
@param j Index valie of the coin to recieve
@param _dx Amount of `i` being exchanged
@param _min_dy Minimum amount of `j` to receive
@return Actual amount of `j` received
"""
rates: uint256[N_COINS] = RATE_MULTIPLIERS
old_balances: uint256[N_COINS] = self._balances()
xp: uint256[N_COINS] = self._xp_mem(rates, old_balances)
coin: address = COINS[i]
dx: uint256 = ERC20(coin).balanceOf(self)
response: Bytes[32] = raw_call(
coin,
concat(
method_id("transferFrom(address,address,uint256)"),
convert(msg.sender, bytes32),
convert(self, bytes32),
convert(_dx, bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool)
dx = ERC20(coin).balanceOf(self) - dx
x: uint256 = xp[i] + dx * rates[i] / PRECISION
y: uint256 = self.get_y(i, j, x, xp)
dy: uint256 = xp[j] - y - 1 # -1 just in case there were some rounding errors
dy_fee: uint256 = dy * self.fee / FEE_DENOMINATOR
# Convert all to real units
dy = (dy - dy_fee) * PRECISION / rates[j]
assert dy >= _min_dy, "Exchange resulted in fewer coins than expected"
self.admin_balances[j] += (dy_fee * ADMIN_FEE / FEE_DENOMINATOR) * PRECISION / rates[j]
response = raw_call(
COINS[j],
concat(
method_id("transfer(address,uint256)"),
convert(_receiver, bytes32),
convert(dy, bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool)
log TokenExchange(msg.sender, i, _dx, j, dy)
return dy
@external
@nonreentrant('lock')
def remove_liquidity(
_burn_amount: uint256,
_min_amounts: uint256[N_COINS],
_receiver: address = msg.sender
) -> uint256[N_COINS]:
"""
@notice Withdraw coins from the pool
@dev Withdrawal amounts are based on current deposit ratios
@param _burn_amount Quantity of LP tokens to burn in the withdrawal
@param _min_amounts Minimum amounts of underlying coins to receive
@param _receiver Address that receives the withdrawn coins
@return List of amounts of coins that were withdrawn
"""
total_supply: uint256 = self.totalSupply
amounts: uint256[N_COINS] = empty(uint256[N_COINS])
balances: uint256[N_COINS] = self._balances()
for i in range(N_COINS):
value: uint256 = balances[i] * _burn_amount / total_supply
assert value >= _min_amounts[i], "Withdrawal resulted in fewer coins than expected"
amounts[i] = value
response: Bytes[32] = raw_call(
COINS[i],
concat(
method_id("transfer(address,uint256)"),
convert(_receiver, bytes32),
convert(value, bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool)
total_supply -= _burn_amount
self.balanceOf[msg.sender] -= _burn_amount
self.totalSupply = total_supply
log Transfer(msg.sender, ZERO_ADDRESS, _burn_amount)
log RemoveLiquidity(msg.sender, amounts, empty(uint256[N_COINS]), total_supply)
return amounts
@external
@nonreentrant('lock')
def remove_liquidity_imbalance(
_amounts: uint256[N_COINS],
_max_burn_amount: uint256,
_receiver: address = msg.sender
) -> uint256:
"""
@notice Withdraw coins from the pool in an imbalanced amount
@param _amounts List of amounts of underlying coins to withdraw
@param _max_burn_amount Maximum amount of LP token to burn in the withdrawal
@param _receiver Address that receives the withdrawn coins
@return Actual amount of the LP token burned in the withdrawal
"""
amp: uint256 = self._A()
old_balances: uint256[N_COINS] = self._balances()
rates: uint256[N_COINS] = RATE_MULTIPLIERS
D0: uint256 = self.get_D_mem(rates, old_balances, amp)
new_balances: uint256[N_COINS] = old_balances
for i in range(N_COINS):
amount: uint256 = _amounts[i]
if amount != 0:
new_balances[i] -= amount
response: Bytes[32] = raw_call(
COINS[i],
concat(
method_id("transfer(address,uint256)"),
convert(_receiver, bytes32),
convert(amount, bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool)
D1: uint256 = self.get_D_mem(rates, new_balances, amp)
fees: uint256[N_COINS] = empty(uint256[N_COINS])
base_fee: uint256 = self.fee * N_COINS / (4 * (N_COINS - 1))
for i in range(N_COINS):
ideal_balance: uint256 = D1 * old_balances[i] / D0
difference: uint256 = 0
new_balance: uint256 = new_balances[i]
if ideal_balance > new_balance:
difference = ideal_balance - new_balance
else:
difference = new_balance - ideal_balance
fees[i] = base_fee * difference / FEE_DENOMINATOR
self.admin_balances[i] += fees[i] * ADMIN_FEE / FEE_DENOMINATOR
new_balances[i] -= fees[i]
D2: uint256 = self.get_D_mem(rates, new_balances, amp)
total_supply: uint256 = self.totalSupply
burn_amount: uint256 = ((D0 - D2) * total_supply / D0) + 1
assert burn_amount > 1 # dev: zero tokens burned
assert burn_amount <= _max_burn_amount, "Slippage screwed you"
total_supply -= burn_amount
self.totalSupply = total_supply
self.balanceOf[msg.sender] -= burn_amount
log Transfer(msg.sender, ZERO_ADDRESS, burn_amount)
log RemoveLiquidityImbalance(msg.sender, _amounts, fees, D1, total_supply)
return burn_amount
@pure
@internal
def get_y_D(A: uint256, i: int128, xp: uint256[N_COINS], D: uint256) -> uint256:
"""
Calculate x[i] if one reduces D from being calculated for xp to D
Done by solving quadratic equation iteratively.
x_1**2 + x_1 * (sum' - (A*n**n - 1) * D / (A * n**n)) = D ** (n + 1) / (n ** (2 * n) * prod' * A)
x_1**2 + b*x_1 = c
x_1 = (x_1**2 + c) / (2*x_1 + b)
"""
# x in the input is converted to the same price/precision
assert i >= 0 # dev: i below zero
assert i < N_COINS # dev: i above N_COINS
S_: uint256 = 0
_x: uint256 = 0
y_prev: uint256 = 0
c: uint256 = D
Ann: uint256 = A * N_COINS
for _i in range(N_COINS):
if _i != i:
_x = xp[_i]
else:
continue
S_ += _x
c = c * D / (_x * N_COINS)
c = c * D * A_PRECISION / (Ann * N_COINS)
b: uint256 = S_ + D * A_PRECISION / Ann
y: uint256 = D
for _i in range(255):
y_prev = y
y = (y*y + c) / (2 * y + b - D)
# Equality with the precision of 1
if y > y_prev:
if y - y_prev <= 1:
return y
else:
if y_prev - y <= 1:
return y
raise
@view
@internal
def _calc_withdraw_one_coin(_burn_amount: uint256, i: int128) -> uint256[2]:
# First, need to calculate
# * Get current D
# * Solve Eqn against y_i for D - _token_amount
amp: uint256 = self._A()
rates: uint256[N_COINS] = RATE_MULTIPLIERS
xp: uint256[N_COINS] = self._xp_mem(rates, self._balances())
D0: uint256 = self.get_D(xp, amp)
total_supply: uint256 = self.totalSupply
D1: uint256 = D0 - _burn_amount * D0 / total_supply
new_y: uint256 = self.get_y_D(amp, i, xp, D1)
base_fee: uint256 = self.fee * N_COINS / (4 * (N_COINS - 1))
xp_reduced: uint256[N_COINS] = empty(uint256[N_COINS])
for j in range(N_COINS):
dx_expected: uint256 = 0
xp_j: uint256 = xp[j]
if j == i:
dx_expected = xp_j * D1 / D0 - new_y
else:
dx_expected = xp_j - xp_j * D1 / D0
xp_reduced[j] = xp_j - base_fee * dx_expected / FEE_DENOMINATOR
dy: uint256 = xp_reduced[i] - self.get_y_D(amp, i, xp_reduced, D1)
dy_0: uint256 = (xp[i] - new_y) * PRECISION / rates[i] # w/o fees
dy = (dy - 1) * PRECISION / rates[i] # Withdraw less to account for rounding errors
return [dy, dy_0 - dy]
@view
@external
def calc_withdraw_one_coin(_burn_amount: uint256, i: int128) -> uint256:
"""
@notice Calculate the amount received when withdrawing a single coin
@param _burn_amount Amount of LP tokens to burn in the withdrawal
@param i Index value of the coin to withdraw
@return Amount of coin received
"""
return self._calc_withdraw_one_coin(_burn_amount, i)[0]
@external
@nonreentrant('lock')
def remove_liquidity_one_coin(
_burn_amount: uint256,
i: int128,
_min_received: uint256,
_receiver: address = msg.sender,
) -> uint256:
"""
@notice Withdraw a single coin from the pool
@param _burn_amount Amount of LP tokens to burn in the withdrawal
@param i Index value of the coin to withdraw
@param _min_received Minimum amount of coin to receive
@param _receiver Address that receives the withdrawn coins
@return Amount of coin received
"""
dy: uint256[2] = self._calc_withdraw_one_coin(_burn_amount, i)
assert dy[0] >= _min_received, "Not enough coins removed"
self.admin_balances[i] += dy[1] * ADMIN_FEE / FEE_DENOMINATOR
total_supply: uint256 = self.totalSupply - _burn_amount
self.totalSupply = total_supply
self.balanceOf[msg.sender] -= _burn_amount
log Transfer(msg.sender, ZERO_ADDRESS, _burn_amount)
response: Bytes[32] = raw_call(
COINS[i],
concat(
method_id("transfer(address,uint256)"),
convert(_receiver, bytes32),
convert(dy[0], bytes32),
),
max_outsize=32,
)
if len(response) > 0:
assert convert(response, bool)
log RemoveLiquidityOne(msg.sender, _burn_amount, dy[0], total_supply)
return dy[0]
@external
def ramp_A(_future_A: uint256, _future_time: uint256):
assert msg.sender == self.owner # dev: only owner
assert block.timestamp >= self.initial_A_time + MIN_RAMP_TIME
assert _future_time >= block.timestamp + MIN_RAMP_TIME # dev: insufficient time
_initial_A: uint256 = self._A()
_future_A_p: uint256 = _future_A * A_PRECISION
assert _future_A > 0 and _future_A < MAX_A
if _future_A_p < _initial_A:
assert _future_A_p * MAX_A_CHANGE >= _initial_A
else:
assert _future_A_p <= _initial_A * MAX_A_CHANGE
self.initial_A = _initial_A
self.future_A = _future_A_p
self.initial_A_time = block.timestamp
self.future_A_time = _future_time
log RampA(_initial_A, _future_A_p, block.timestamp, _future_time)
@external
def stop_ramp_A():
assert msg.sender == self.owner # dev: only owner
current_A: uint256 = self._A()
self.initial_A = current_A
self.future_A = current_A
self.initial_A_time = block.timestamp
self.future_A_time = block.timestamp
# now (block.timestamp < t1) is always False, so we return saved A
log StopRampA(current_A, block.timestamp)
@external
def withdraw_admin_fees():
assert msg.sender == self.owner
for i in range(N_COINS):
amount: uint256 = self.admin_balances[i]
if amount != 0:
coin: address = COINS[i]
raw_call(
coin,
concat(
method_id("transfer(address,uint256)"),
convert(msg.sender, bytes32),
convert(amount, bytes32)
)
)
self.admin_balances[i] = 0
@external
def commit_transfer_ownership(_future_owner: address):
assert msg.sender == self.owner
self.future_owner = _future_owner
@external
def accept_transfer_ownership():
assert msg.sender == self.future_owner
log TransferOwnership(self.owner, msg.sender)
self.owner = msg.sender
@view
@external
def version() -> String[8]:
"""
@notice Get the version of this token contract
"""
return VERSION
@pure
@external
def coins(_i: uint256) -> address:
return COINS[_i]
@pure
@external
def name() -> String[64]:
return NAME
@pure
@external
def symbol() -> String[32]:
return SYMBOL
@pure
@external
def DOMAIN_SEPARATOR() -> bytes32:
return DOMAIN_SEPARATOR
