StableSwap-NG Oracles

Price and D Oracles¶

StableSwap-NG pools have the following oracles:

The AMM implementation utilizes two private variables, last_prices_packed and last_D_packed, to store the latest spot and EMA values. These values serve as the foundation for calculating the oracles.

@internal
@view
def _calc_moving_average(
    packed_value: uint256,
    averaging_window: uint256,
    ma_last_time: uint256
) -> uint256:

    last_spot_value: uint256 = packed_value & (2**128 - 1)
    last_ema_value: uint256 = (packed_value >> 128)

    if ma_last_time < block.timestamp:  # calculate new_ema_value and return that.
        alpha: uint256 = self.exp(
            -convert(
                unsafe_div(unsafe_mul(unsafe_sub(block.timestamp, ma_last_time), 10**18), averaging_window), int256
            )
        )
        return unsafe_div(last_spot_value * (10**18 - alpha) + last_ema_value * alpha, 10**18)

    return last_ema_value

The formula to calculate the exponential moving-average essentially comes down to:

with:

price_oracle calculation is based on the two values stored in last_prices_packed, last_price and ema_price. These values are conditionally updated: Generally speaking, both values are simultaneously updated whenever upkeep_oracles is called. This happens at certain actions, see here.

While last_price (spot price) is always updated at every relevant action, the ema_price is maximally updated once per block. There might be the case that there is more than one relevant action within the same block. Let's say there are two relevant actions within the block which would update both values: If this is the case, last_price is updated at every action, so there will be two updated. ema_price on the other hand will only be updated once (at the first action) and will not change a second time. Reasoning behind this is to prevent single-block manipulation. The ema_price will just be updated at the next action outside of this block.

D_oracle calculation is based on the two values stored in last_D_packed, last_D and ma_D.

price_oracle¶

last_prices_packed: DynArray[uint256, MAX_COINS]  #  packing: last_price, ma_price
ma_exp_time: public(uint256)
ma_last_time: public(uint256)                     # packing: ma_last_time_p, ma_last_time_D

@external
@view
@nonreentrant('lock')
def price_oracle(i: uint256) -> uint256:
    return self._calc_moving_average(
        self.last_prices_packed[i],
        self.ma_exp_time,
        self.ma_last_time & (2**128 - 1)
    )

@internal
@view
def _calc_moving_average(
    packed_value: uint256,
    averaging_window: uint256,
    ma_last_time: uint256
) -> uint256:

    last_spot_value: uint256 = packed_value & (2**128 - 1)
    last_ema_value: uint256 = (packed_value >> 128)

    if ma_last_time < block.timestamp:  # calculate new_ema_value and return that.
        alpha: uint256 = self.exp(
            -convert(
                (block.timestamp - ma_last_time) * 10**18 / averaging_window, int256
            )
        )
        return (last_spot_value * (10**18 - alpha) + last_ema_value * alpha) / 10**18

    return last_ema_value

>>> StableSwap.price_oracle(0)
1000187813326452556

D_oracle¶

last_D_packed: uint256                            #  packing: last_D, ma_D
D_ma_time: public(uint256)
ma_last_time: public(uint256)                     # packing: ma_last_time_p, ma_last_time_D

@external
@view
@nonreentrant('lock')
def D_oracle() -> uint256:
    return self._calc_moving_average(
        self.last_D_packed,
        self.D_ma_time,
        self.ma_last_time >> 128
    )

@internal
@view
def _calc_moving_average(
    packed_value: uint256,
    averaging_window: uint256,
    ma_last_time: uint256
) -> uint256:

    last_spot_value: uint256 = packed_value & (2**128 - 1)
    last_ema_value: uint256 = (packed_value >> 128)

    if ma_last_time < block.timestamp:  # calculate new_ema_value and return that.
        alpha: uint256 = self.exp(
            -convert(
                (block.timestamp - ma_last_time) * 10**18 / averaging_window, int256
            )
        )
        return (last_spot_value * (10**18 - alpha) + last_ema_value * alpha) / 10**18

    return last_ema_value

>>> StableSwap.D_oracle()
2183776033162328612308290

Other Methods¶

last_price¶

last_prices_packed: DynArray[uint256, MAX_COINS]  #  packing: last_price, ma_price

@view
@external
def last_price(i: uint256) -> uint256:
    return self.last_prices_packed[i] & (2**128 - 1)

>>> StableSwap.last_price(0)
1000187811171795736

ema_price¶

last_prices_packed: DynArray[uint256, MAX_COINS]  #  packing: last_price, ma_price

@view
@external
def ema_price(i: uint256) -> uint256:
    return (self.last_prices_packed[i] >> 128)

>>> StableSwap.ema_price(0)
1000187824576102231

get_p¶

@external
@view
def get_p(i: uint256) -> uint256:
    """
    @notice Returns the AMM State price of token
    @dev if i = 0, it will return the state price of coin[1].
    @param i index of state price (0 for coin[1], 1 for coin[2], ...)
    @return uint256 The state price quoted by the AMM for coin[i+1]
    """
    amp: uint256 = self._A()
    xp: DynArray[uint256, MAX_COINS] = self._xp_mem(
        self._stored_rates(), self._balances()
    )
    D: uint256 = self.get_D(xp, amp)
    return self._get_p(xp, amp, D)[i]

@internal
@pure
def _get_p(
    xp: DynArray[uint256, MAX_COINS],
    amp: uint256,
    D: uint256,
) -> DynArray[uint256, MAX_COINS]:

    # dx_0 / dx_1 only, however can have any number of coins in pool
    ANN: uint256 = unsafe_mul(amp, N_COINS)
    Dr: uint256 = unsafe_div(D, pow_mod256(N_COINS, N_COINS))

    for i in range(MAX_COINS_128):

        if i == N_COINS_128:
            break

        Dr = Dr * D / xp[i]

    p: DynArray[uint256, MAX_COINS] = empty(DynArray[uint256, MAX_COINS])
    xp0_A: uint256 = ANN * xp[0] / A_PRECISION

    for i in range(1, MAX_COINS):

        if i == N_COINS:
            break

        p.append(10**18 * (xp0_A + Dr * xp[0] / xp[i]) / (xp0_A + Dr))

    return p

>>> StableSwap.get_p(0)
1000187811171795736

get_virtual_price¶

@view
@external
@nonreentrant('lock')
def get_virtual_price() -> uint256:
    """
    @notice The current virtual price of the pool LP token
    @dev Useful for calculating profits.
        The method may be vulnerable to donation-style attacks if implementation
        contains rebasing tokens. For integrators, caution is advised.
    @return LP token virtual price normalized to 1e18
    """
    amp: uint256 = self._A()
    xp: DynArray[uint256, MAX_COINS] = self._xp_mem(
        self._stored_rates(), self._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.total_supply

>>> StableSwap.get_virtual_price()
1000063971106330426

ma_exp_time¶

ma_exp_time: public(uint256)

>>> StableSwap.ma_exp_time()
866

D_ma_time¶

D_ma_time: public(uint256)

>>> StableSwap.D_ma_time()
62324

ma_last_time¶

ma_last_time: public(uint256)                     # packing: ma_last_time_p, ma_last_time_D

>>> StableSwap.ma_last_time()
579359617954437487117250992339883299967854142015

Upkeeping Oracles¶

The internal upkeep_oracles method is responsible for updating the price and D oracle.

@internal
def upkeep_oracles(xp: DynArray[uint256, MAX_COINS], amp: uint256, D: uint256):
    """
    @notice Upkeeps price and D oracles.
    """
    ma_last_time_unpacked: uint256[2] = self.unpack_2(self.ma_last_time)
    last_prices_packed_current: DynArray[uint256, MAX_COINS] = self.last_prices_packed
    last_prices_packed_new: DynArray[uint256, MAX_COINS] = last_prices_packed_current

    spot_price: DynArray[uint256, MAX_COINS] = self._get_p(xp, amp, D)

    # -------------------------- Upkeep price oracle -------------------------

    for i in range(MAX_COINS):

        if i == N_COINS - 1:
            break

        if spot_price[i] != 0:

            # Update packed prices -----------------
            last_prices_packed_new[i] = self.pack_2(
                min(spot_price[i], 2 * 10**18),  # <----- Cap spot value by 2.
                self._calc_moving_average(
                    last_prices_packed_current[i],
                    self.ma_exp_time,
                    ma_last_time_unpacked[0],  # index 0 is ma_last_time for prices
                )
            )

    self.last_prices_packed = last_prices_packed_new

    # ---------------------------- Upkeep D oracle ---------------------------

    last_D_packed_current: uint256 = self.last_D_packed
    self.last_D_packed = self.pack_2(
        D,
        self._calc_moving_average(
            last_D_packed_current,
            self.D_ma_time,
            ma_last_time_unpacked[1],  # index 1 is ma_last_time for D
        )
    )

    # Housekeeping: Update ma_last_time for p and D oracles ------------------
    for i in range(2):
        if ma_last_time_unpacked[i] < block.timestamp:
            ma_last_time_unpacked[i] = block.timestamp

    self.ma_last_time = self.pack_2(ma_last_time_unpacked[0], ma_last_time_unpacked[1])

Price Oracles¶

The price oracle is updated when the upkeep_oracles method is called. This occurs in response to one of the following actions:

• Token exchange (__exchange)
• Liquidity addition (add_liquidity)
• Single-sided liquidity (remove_liquidity_one_coin)
• Liquidity removal in an imbalanced proportion (remove_liquidity_imbalance)

When price oracles are upkept, the code calculates both the spot price and the moving-average price. These values are then packed and stored together in last_prices_packed.

# -------------------------- Upkeep price oracle -------------------------

for i in range(MAX_COINS):

    if i == N_COINS - 1:
        break

    if spot_price[i] != 0:

        # Update packed prices -----------------
        last_prices_packed_new[i] = self.pack_2(
            min(spot_price[i], 2 * 10**18),  # <----- Cap spot value by 2.
            self._calc_moving_average(
                last_prices_packed_current[i],
                self.ma_exp_time,
                ma_last_time_unpacked[0],  # index 0 is ma_last_time for prices
            )
        )

self.last_prices_packed = last_prices_packed_new

1. last_price which represents the last stored spot price within the AMM is calculated using _get_p. Additionally, the value is capped at 2 * 10**18 to prevent price oracle manipulation. Note: It's not actually the spot price which is capped, but rather the spot price that is used in the calculation for the ema price oracle.

   _get_p

   @internal
   @pure
   def _get_p(
       xp: DynArray[uint256, MAX_COINS],
       amp: uint256,
       D: uint256,
   ) -> DynArray[uint256, MAX_COINS]:
   
       # dx_0 / dx_1 only, however can have any number of coins in pool
       ANN: uint256 = unsafe_mul(amp, N_COINS)
       Dr: uint256 = unsafe_div(D, pow_mod256(N_COINS, N_COINS))
   
       for i in range(N_COINS_128, bound=MAX_COINS_128):
           Dr = Dr * D / xp[i]
   
       p: DynArray[uint256, MAX_COINS] = empty(DynArray[uint256, MAX_COINS])
       xp0_A: uint256 = unsafe_div(ANN * xp[0], A_PRECISION)
   
       for i in range(1, MAX_COINS):
   
           if i == N_COINS:
               break
   
           p.append(10**18 * (xp0_A + unsafe_div(Dr * xp[0], xp[i])) / (xp0_A + Dr))
   
       return p
   

2. The moving-average price (ema_price) is calculated using _calc_moving_average. This value can only be updated once per block. If there are two actions which would update the value, only the first action will update it. For the second action, only the last_price is updated, while ema_price will not be updated and have the same value as in the first action.

   _calc_moving_average

   @internal
   @view
   def _calc_moving_average(
       packed_value: uint256,
       averaging_window: uint256,
       ma_last_time: uint256
   ) -> uint256:
   
       last_spot_value: uint256 = packed_value & (2**128 - 1)
       last_ema_value: uint256 = (packed_value >> 128)
   
       if ma_last_time < block.timestamp:  # calculate new_ema_value and return that.
           alpha: uint256 = self.exp(
               -convert(
                   unsafe_div(unsafe_mul(unsafe_sub(block.timestamp, ma_last_time), 10**18), averaging_window), int256
               )
           )
           return unsafe_div(last_spot_value * (10**18 - alpha) + last_ema_value * alpha, 10**18)
   
       return last_ema_value
   

D Oracle¶

The D oracle is updated when the upkeep_oracles method is called. This occurs in response to one of the following actions:

• Token exchange (__exchange)
• Liquidity addition (add_liquidity)
• Single-sided liquidity (remove_liquidity_one_coin)
• Liquidity removal in an imbalanced proportion (remove_liquidity_imbalance)
• Balanced proportion liquidity removal. For this action, the remove_liquidity function, which executes it, does not directly call the upkeep_oracles method. Instead, the D oracle update is performed "manually" within the function. The rationale behind this approach is that updating the price oracle is not necessary in this scenario, because removing in a balanced proportion does not change the prices within the AMM.

When the D oracle is updated, the code calculates both the "spot" D invariant and the moving-average D invariant value. These values are then packed and stored together in last_D_packed.

# ---------------------------- Upkeep D oracle ---------------------------

last_D_packed_current: uint256 = self.last_D_packed
self.last_D_packed = self.pack_2(
    D,
    self._calc_moving_average(
        last_D_packed_current,
        self.D_ma_time,
        ma_last_time_unpacked[1],  # index 1 is ma_last_time for D
    )
)