feat: asyncify yeth

yearn/yeth.py

@@ -1,26 +1,32 @@
 import asyncio
+import logging
 import os
 import re
 import logging
 from datetime import datetime, timezone, timedelta
+from pprint import pformat
+from typing import Optional, AsyncIterator
 
 import eth_retry
-
 from brownie import chain
-from pprint import pformat
+from brownie.network.event import _EventItem
 
-from y import Contract, magic, Network
-from y.time import get_block_timestamp, closest_block_after_timestamp
+from y import Contract, Network, magic
 from y.contracts import contract_creation_block_async
+from y.datatypes import Block
 from y.exceptions import PriceError, yPriceMagicError
+from y.time import get_block_timestamp_async, closest_block_after_timestamp
+from y.utils.events import Events
+from y.utils.dank_mids import dank_w3
 
-from yearn.apy.common import (Apy, ApyFees,
-                              ApySamples, SECONDS_PER_YEAR, SECONDS_PER_WEEK, SharePricePoint, calculate_roi, get_samples)
+from yearn.apy.common import (SECONDS_PER_WEEK, SECONDS_PER_YEAR, Apy, ApyFees,
+                              ApySamples, SharePricePoint, calculate_roi,
+                              get_samples)
 from yearn.common import Tvl
+from yearn.debug import Debug
 from yearn.events import decode_logs, get_logs_asap
-from yearn.utils import Singleton
 from yearn.prices.constants import weth
-from yearn.debug import Debug
+from yearn.utils import Singleton
 
 logger = logging.getLogger("yearn.yeth")
 
@@ -49,24 +55,20 @@ def decimals(self):
     def symbol(self):
         return 'st-yETH'
 
+    async def get_supply(self, block: Optional[Block] = None) -> float:
+        return (await YETH_POOL.vb_prod_sum.coroutine(block_identifier=block))[1] / 10 ** 18
 
-    async def _get_supply_price(self, block=None):
-        data = YETH_POOL.vb_prod_sum(block_identifier=block)
-        supply = data[1] / 1e18
+    async def get_price(self, block: Optional[Block] = None) -> Optional[float]:
         try:
-            price = await magic.get_price(YETH_TOKEN, block=block, sync=False)
+            return float(await magic.get_price(YETH_TOKEN, block=block, sync=False))
         except yPriceMagicError as e:
             if not isinstance(e.exception, PriceError):
                 raise e
-            price = None
-
-        return supply, price
-
 
     @eth_retry.auto_retry
     async def apy(self, samples: ApySamples) -> Apy:
         block = samples.now
-        now = get_block_timestamp(block)
+        now = await get_block_timestamp_async(block)
         seconds_til_eow = SECONDS_PER_WEEK - now % SECONDS_PER_WEEK
 
         data = STAKING_CONTRACT.get_amounts(block_identifier=block)
@@ -83,14 +85,13 @@ async def apy(self, samples: ApySamples) -> Apy:
 
     @eth_retry.auto_retry
     async def tvl(self, block=None) -> Tvl:
-        supply, price = await self._get_supply_price(block=block)
+        supply, price = await asyncio.gather(self.get_supply(block), self.get_price(block))
         tvl = supply * price if price else None
-
         return Tvl(supply, price, tvl)
 
 
     async def describe(self, block=None):
-        supply, price = await self._get_supply_price(block=block)
+        supply, price = await asyncio.gather(self.get_supply(block), self.get_price(block))
         try:
             pool_supply = YETH_POOL.supply(block_identifier=block)
             total_assets = STAKING_CONTRACT.totalAssets(block_identifier=block)
@@ -100,8 +101,8 @@ async def describe(self, block=None):
             boost = 0
 
         if block:
-            block_timestamp = get_block_timestamp(block)
-            samples = get_samples(datetime.fromtimestamp(block_timestamp))
+            block_timestamp = await get_block_timestamp_async(block)
+            samples = get_samples(datetime.fromtimestamp(block_timestamp, tz=timezone.utc))
         else:
             samples = get_samples()
 
@@ -118,7 +119,7 @@ async def describe(self, block=None):
 
 
     async def total_value_at(self, block=None):
-        supply, price = await self._get_supply_price(block=block)
+        supply, price = await asyncio.gather(self.get_supply(block), self.get_price(block))
         return supply * price
 
 
@@ -140,12 +141,16 @@ def decimals(self):
     def _sanitize(self, name):
         return re.sub(r"([\d]+)\.[\d]*", r"\1", name)
 
-    def _get_lst_data(self, block=None):
-        virtual_balance = YETH_POOL.virtual_balance(self.asset_id, block_identifier=block) / 1e18
-        weights = YETH_POOL.weight(self.asset_id, block_identifier=block)
+    async def _get_lst_data(self, block=None):
+        virtual_balance, weights, rate = await asyncio.gather(
+            YETH_POOL.virtual_balance.coroutine(self.asset_id, block_identifier=block),
+            YETH_POOL.weight.coroutine(self.asset_id, block_identifier=block),
+            RATE_PROVIDER.rate.coroutine(str(self.lst), block_identifier=block)
+        )
+        virtual_balance /= 1e18
         weight = weights[0] / 1e18
         target = weights[1] / 1e18
-        rate = RATE_PROVIDER.rate(str(self.lst), block_identifier=block) / 1e18
+        rate /= 1e18
 
         return {
           "virtual_balance": virtual_balance,
@@ -156,8 +161,12 @@ def _get_lst_data(self, block=None):
 
     @eth_retry.auto_retry
     async def apy(self, samples: ApySamples) -> Apy:
-        now_rate = RATE_PROVIDER.rate(str(self.lst), block_identifier=samples.now) / 1e18
-        week_ago_rate = RATE_PROVIDER.rate(str(self.lst), block_identifier=samples.week_ago) / 1e18
+        now_rate, week_ago_rate = await asyncio.gather(
+            RATE_PROVIDER.rate.coroutine(str(self.lst), block_identifier=samples.now),
+            RATE_PROVIDER.rate(str(self.lst), block_identifier=samples.week_ago),
+        )
+        now_rate /= 1e18
+        week_ago_rate /= 1e18
         now_point = SharePricePoint(samples.now, now_rate)
         week_ago_point = SharePricePoint(samples.week_ago, week_ago_rate)
         apy = calculate_roi(now_point, week_ago_point)
@@ -166,16 +175,18 @@ async def apy(self, samples: ApySamples) -> Apy:
 
     @eth_retry.auto_retry
     async def tvl(self, block=None) -> Tvl:
-        data = self._get_lst_data(block=block)
+        data = await self._get_lst_data(block=block)
         tvl = data["virtual_balance"] * data["rate"]
         return Tvl(data["virtual_balance"], data["rate"], tvl)
 
     async def describe(self, block=None):
-        weth_price = await magic.get_price(weth, block=block, sync=False)
-        data = self._get_lst_data(block=block)
+        weth_price, data = await asyncio.gather(
+            magic.get_price(weth, block=block, sync=False),
+            self._get_lst_data(block=block),
+        )
 
         if block:
-            block_timestamp = get_block_timestamp(block)
+            block_timestamp = await get_block_timestamp_async(block)
             samples = get_samples(datetime.fromtimestamp(block_timestamp))
         else:
             samples = get_samples()
@@ -201,7 +212,7 @@ async def describe(self, block=None):
         }
 
     async def total_value_at(self, block=None):
-        data = self._get_lst_data(block=block)
+        data = await self._get_lst_data(block=block)
         tvl = data["virtual_balance"] * data["rate"]
         return tvl
 
@@ -211,33 +222,34 @@ def __init__(self) -> None:
         self.st_yeth = StYETH()
         self.swap_volumes = {}
         self.resolution = os.environ.get('RESOLUTION', '1h') # Default: 1 hour
-
+        self.swap_events = Events(addresses=[str(YETH_POOL)])
+
+    async def _get_swaps(self, from_block, to_block) -> AsyncIterator[_EventItem]:
+        async for event in YETH_POOL.events.Swap.events(to_block):
+            if event.block_number < from_block:
+                continue
+            elif event.block_number > to_block:
+                return
+            yield event
+
     async def _get_swap_volumes(self, from_block, to_block):
-        logs = get_logs_asap([str(YETH_POOL)], None, from_block=from_block, to_block=to_block)
-        events = decode_logs(logs)
-
-        num_assets = YETH_POOL.num_assets(block_identifier=from_block)
+        num_assets = await YETH_POOL.num_assets.coroutine(block_identifier=from_block)
         volume_in_eth = [0] * num_assets
         volume_out_eth = [0] * num_assets
         volume_in_usd = [0] * num_assets
         volume_out_usd = [0] * num_assets
 
-        rates = []
-        for i in range(num_assets):
-            lst = self.st_yeth.lsts[i]
-            address = str(lst.lst)
-            rates.append(RATE_PROVIDER.rate(address, block_identifier=from_block) / 1e18)
-
-        for e in events:
-            if e.name == "Swap":
-                asset_in = e["asset_in"]
-                asset_out = e["asset_out"]
-                amount_in = e["amount_in"] / 1e18
-                amount_out = e["amount_out"] / 1e18
-                volume_in_eth[asset_in] += amount_in * rates[asset_in]
-                volume_out_eth[asset_out] += amount_out * rates[asset_out]
-
-        weth_price = await magic.get_price(weth, block=from_block, sync=False)
+        rates = [r / 1e18 for r in await asyncio.gather(*[RATE_PROVIDER.rate.coroutine(str(self.st_yeth.lsts[i].lst), block_identifier=from_block) for i in range(num_assets)])]
+
+        async for swap in self._get_swaps(from_block, to_block):
+            asset_in = swap["asset_in"]
+            asset_out = swap["asset_out"]
+            amount_in = swap["amount_in"] / 1e18
+            amount_out = swap["amount_out"] / 1e18
+            volume_in_eth[asset_in] += amount_in * rates[asset_in]
+            volume_out_eth[asset_out] += amount_out * rates[asset_out]
+
+        weth_price = float(await magic.get_price(weth, block=from_block, sync=False))
         for i, value in enumerate(volume_in_eth):
             volume_in_usd[i] = value * weth_price
 
@@ -252,26 +264,27 @@ async def _get_swap_volumes(self, from_block, to_block):
         }
 
     async def describe(self, block=None):
-        to_block = chain.height
-        now_time = datetime.today()
         if block:
             to_block = block
-            block_timestamp = get_block_timestamp(block)
-            now_time = datetime.fromtimestamp(block_timestamp)
+            block_timestamp = await get_block_timestamp_async(block)
+            now_time = datetime.fromtimestamp(block_timestamp, tz=timezone.utc)
+        else:
+            to_block = await dank_w3.eth.block_number
+            now_time = datetime.today()
 
         from_block = self._get_from_block(now_time)
 
         self.swap_volumes = await self._get_swap_volumes(from_block, to_block)
-        products = await self.active_products_at(block)
+        products = await self.active_vaults_at(block)
         data = await asyncio.gather(*[product.describe(block=block) for product in products])
         return {product.name: desc for product, desc in zip(products, data)}
 
     async def total_value_at(self, block=None):
-        products = await self.active_products_at(block)
+        products = await self.active_vaults_at(block)
         tvls = await asyncio.gather(*[product.total_value_at(block=block) for product in products])
         return {product.name: tvl for product, tvl in zip(products, tvls)}
 
-    async def active_products_at(self, block=None):
+    async def active_vaults_at(self, block=None):
         products = [self.st_yeth] + self.st_yeth.lsts
         if block:
             blocks = await asyncio.gather(*[contract_creation_block_async(str(product.address)) for product in products])