Treasury and wallets

Makefile

@@ -51,7 +51,7 @@ rebuild: down build up
 scratch: clean-volumes build up
 
 logs:
-	$(dashboards_command) logs -f -t eth-exporter historical-eth-exporter ftm-exporter historical-ftm-exporter treasury-exporter historical-treasury-exporter transactions-exporter wallet-exporter
+	$(dashboards_command) logs -f -t eth-exporter historical-eth-exporter ftm-exporter historical-ftm-exporter treasury-exporter historical-treasury-exporter transactions-exporter wallet-exporter sms-exporter historical-sms-exporter
 
 test:
 	$(test_command) up

requirements.txt

@@ -15,5 +15,5 @@ boto3==1.17.88
 rich>=10.11.0
 matplotlib>=3.4.3
 pandas>=1.3.0
-pyodbc>=4.0.32
+pony>=0.6.7
 sqlalchemy>=1.4.26

scripts/historical_sms_exporter.py

@@ -0,0 +1,44 @@
+import logging
+from datetime import datetime, timezone
+
+from yearn.historical_helper import export_historical, time_tracking
+from yearn.treasury.treasury import StrategistMultisig
+from yearn.utils import closest_block_after_timestamp
+
+logger = logging.getLogger('yearn.historical_sms_exporter')
+
+def main():
+    start = datetime.now(tz=timezone.utc)
+    # end: 2020-12-23 first sms tx
+    end = datetime(2020, 12, 23, tzinfo=timezone.utc)
+    export_historical(
+        start,
+        end,
+        export_chunk,
+        export_snapshot,
+        'sms_assets'
+    )
+
+
+def export_chunk(chunk, export_snapshot_func):
+    sms = StrategistMultisig()
+    for snapshot in chunk:
+        ts = snapshot.timestamp()
+        export_snapshot_func(
+            {
+                'treasury': sms,
+                'snapshot': snapshot,
+                'ts': ts,
+                'exporter_name': 'historical_sms'
+            }
+        )   
+
+
+@time_tracking
+def export_snapshot(sms, snapshot, ts, exporter_name):
+    block = closest_block_after_timestamp(ts)
+    assert block is not None, "no block after timestamp found"
+    sms.export(block, ts)
+    logger.info("exported SMS snapshot %s", snapshot)
+
+

scripts/historical_treasury_exporter.py

@@ -1,14 +1,15 @@
 import logging
 from datetime import datetime, timezone
+
 from yearn.historical_helper import export_historical, time_tracking
+from yearn.treasury.treasury import YearnTreasury
 from yearn.utils import closest_block_after_timestamp
-from yearn.treasury.treasury import Treasury
 
 logger = logging.getLogger('yearn.historical_treasury_exporter')
 
 def main():
     start = datetime.now(tz=timezone.utc)
-    # end: 2020-02-12 first treasury tx
+    # end: 2020-07-21 first treasury tx
     end = datetime(2020, 7, 21, tzinfo=timezone.utc)
     export_historical(
         start,
@@ -20,7 +21,7 @@ def main():
 
 
 def export_chunk(chunk, export_snapshot_func):
-    treasury = Treasury()
+    treasury = YearnTreasury()
     for snapshot in chunk:
         ts = snapshot.timestamp()
         export_snapshot_func(

scripts/sms_exporter.py

@@ -0,0 +1,20 @@
+import logging
+import os
+import time
+
+from brownie import chain
+from yearn.treasury.treasury import StrategistMultisig
+from yearn.outputs import victoria
+
+logger = logging.getLogger('yearn.sms_exporter')
+sleep_interval = int(os.environ.get('SLEEP_SECONDS', '0'))
+
+
+def main():
+    treasury = StrategistMultisig(watch_events_forever=True)
+    for block in chain.new_blocks(height_buffer=12):
+        start_time = time.time()
+        treasury.export(block.number, block.timestamp)
+        duration = time.time() - start_time
+        victoria.export_duration(duration, 1, "sms_forwards", block.timestamp)
+        time.sleep(sleep_interval)

scripts/transactions_exporter.py

@@ -1,63 +1,74 @@
+import logging
 import time
 import warnings
 from decimal import Decimal
-import logging
 
 import pandas as pd
 from brownie import ZERO_ADDRESS, Contract, chain, web3
 from brownie.exceptions import BrownieEnvironmentWarning
+from pony.orm import db_session
 from joblib import Parallel, delayed
-import sqlalchemy
-from tqdm import tqdm
 from web3._utils.abi import filter_by_name
 from web3._utils.events import construct_event_topic_set
-from yearn.events import create_filter, decode_logs, get_logs_asap
-from yearn.outputs.postgres.postgres import postgres
+from yearn.entities import UserTx  # , TreasuryTx
+from yearn.events import decode_logs, get_logs_asap
+from yearn.outputs.postgres.utils import (cache_address, cache_token,
+                                          last_recorded_block)
 from yearn.prices import magic
-from yearn.utils import contract
-from yearn.v1.registry import Registry as RegistryV1
-from yearn.v2.registry import Registry as RegistryV2
-from yearn.treasury.treasury import Treasury
-
-treasury = Treasury()
+from yearn.yearn import Yearn
 
 warnings.simplefilter("ignore", BrownieEnvironmentWarning)
 
-registryV1 = RegistryV1()
-registryV2 = RegistryV2()
+yearn = Yearn(load_strategies=False)
 
 logger = logging.getLogger(__name__)
 
+BATCH_SIZE = 5000
 
 def main():
     for block in chain.new_blocks(height_buffer=1):
-        process_and_cache_user_txs(postgres.last_recorded_block('user_txs'))
-
-
-def active_vaults_at(end_block) -> set:
-    v1 = {vault.vault for vault in registryV1.active_vaults_at(end_block)}
-    v2 = {vault.vault for vault in registryV2.active_vaults_at(end_block)}
-    return v1.union(v2)
+        process_and_cache_user_txs(last_recorded_block(UserTx))
 
 
+@db_session
 def process_and_cache_user_txs(last_saved_block=None):
-    max_block_to_cache = (
-        chain.height - 50
-    )  # We look 50 blocks back to avoid uncles and reorgs
+    # NOTE: We look 50 blocks back to avoid uncles and reorgs
+    max_block_to_cache = chain.height - 50
     start_block = last_saved_block + 1 if last_saved_block else None
     end_block = (
-        10650000
+        9480000 # NOTE block some arbitrary time after iearn's first deployment
         if start_block is None
-        else start_block + 500
-        if start_block + 500 < max_block_to_cache
+        else start_block + BATCH_SIZE
+        if start_block + BATCH_SIZE < max_block_to_cache
         else max_block_to_cache
     )
     df = pd.DataFrame()
-    for vault in tqdm(active_vaults_at(end_block)):
-        df = df.append(get_token_transfers(vault, start_block, end_block))
-    df = df.rename(columns={'token': 'vault'})
-    df.to_sql('user_txs', postgres.sqla_engine, if_exists='append', index=False)
-    print(f'user txs batch {start_block}-{end_block} exported to postrges')
+    for vault in yearn.active_vaults_at(end_block):
+        df = df.append(get_token_transfers(vault.vault, start_block, end_block))
+    if len(df):
+        # NOTE: We want to insert txs in the order they took place, so wallet exporter
+        #       won't have issues in the event that transactions exporter fails mid-run.
+        df = df.sort_values('block')  
+        for index, row in df.iterrows():
+            UserTx(
+                vault=cache_token(row.token),
+                timestamp=row.timestamp,
+                block=row.block,
+                hash=row.hash,
+                log_index=row.log_index,
+                type=row.type,
+                from_address=row['from'],
+                to_address=row['to'],
+                amount = row.amount,
+                price = row.price,
+                value_usd = row.value_usd,
+                gas_used = row.gas_used,
+                gas_price = row.gas_price
+                )
+        if start_block == end_block:
+            logger.warn(f'{len(df)} user txs exported to postrges [block {start_block}]')
+        else:
+            logger.warn(f'{len(df)} user txs exported to postrges [blocks {start_block}-{end_block}]')
 
 
 # Helper functions
@@ -66,76 +77,62 @@ def get_token_transfers(token, start_block, end_block) -> pd.DataFrame:
         filter_by_name('Transfer', token.abi)[0],
         web3.codec,
     )
-    postgres.cache_token(token.address)
-    decimals = contract(token.address).decimals()
+    token_entity = cache_token(token.address)
     events = decode_logs(
         get_logs_asap(token.address, topics, from_block=start_block, to_block=end_block)
     )
-    return pd.DataFrame(
-        Parallel(1, 'threading')(
-            delayed(_process_transfer_event)(event, token, decimals) for event in events
-        )
-    )
+    return pd.DataFrame([_process_transfer_event(event, token_entity) for event in events])
 
 
-def _process_transfer_event(event, vault, decimals) -> dict:
+def _process_transfer_event(event, token_entity) -> dict:
     sender, receiver, amount = event.values()
-    postgres.cache_address(sender)
-    postgres.cache_address(receiver)
-    price = _get_price(event, vault)
+    cache_address(sender)
+    cache_address(receiver)
+    price = _get_price(event, token_entity)
     if (
-        vault.address == '0x7F83935EcFe4729c4Ea592Ab2bC1A32588409797'
+        # NOTE magic.get_price() returns erroneous price due to erroneous ppfs
+        token_entity.token.address == '0x7F83935EcFe4729c4Ea592Ab2bC1A32588409797'
         and event.block_number == 12869164
     ):
-        # NOTE magic.get_price() returns erroneous price due to erroneous ppfs
         price = 99999
-    if price > 100000:
-        logger.warn(f'token: {vault.address}')
-        logger.warn(f'price: {price}')
-        logger.warn(f'block: {event.block_number}')
     txhash = event.transaction_hash.hex()
     return {
         'chainid': chain.id,
         'block': event.block_number,
         'timestamp': chain[event.block_number].timestamp,
         'hash': txhash,
         'log_index': event.log_index,
-        'token': vault.address,
-        'type': _event_type(sender, receiver, vault.address),
+        'token': token_entity.token.address,
+        'type': _event_type(sender, receiver, token_entity.token.address),
         'from': sender,
         'to': receiver,
-        'amount': Decimal(amount) / Decimal(10 ** decimals),
+        'amount': Decimal(amount) / Decimal(10 ** token_entity.decimals),
         'price': price,
-        'value_usd': Decimal(amount) / Decimal(10 ** decimals) * Decimal(price),
+        'value_usd': Decimal(amount) / Decimal(10 ** token_entity.decimals) * Decimal(price),
         'gas_used': web3.eth.getTransactionReceipt(txhash).gasUsed,
-        'gas_price': web3.eth.getTransaction(
-            txhash
-        ).gasPrice,  # * 1.0 # force pandas to insert this as decimal not bigint
+        'gas_price': web3.eth.getTransaction(txhash).gasPrice
     }
 
 
-def _get_price(event, vault):
+def _get_price(event, token_entity):
     while True:
         try:
             try:
-                return magic.get_price(vault.address, event.block_number)
+                return magic.get_price(token_entity.token.address, event.block_number)
             except TypeError:  # magic.get_price fails because all liquidity was removed for testing and `share_price` returns None
-                return magic.get_price(vault.token(), event.block_number)
+                return magic.get_price(Contract(token_entity.token.address).token(), event.block_number)
         except ConnectionError as e:
             # Try again
-            print(f'ConnectionError: {str(e)}')
+            logger.warn(f'ConnectionError: {str(e)}')
             time.sleep(1)
         except ValueError as e:
-            print(f'ValueError: {str(e)}')
-            if str(e) in [
-                "Failed to retrieve data from API: {'status': '0', 'message': 'NOTOK', 'result': 'Max rate limit reached'}",
-                "Failed to retrieve data from API: {'status': '0', 'message': 'NOTOK', 'result': 'Max rate limit reached, please use API Key for higher rate limit'}",
-            ]:
+            logger.warn(f'ValueError: {str(e)}')
+            if 'Max rate limit reached' in str(e):
                 # Try again
-                print('trying again...')
+                logger.warn('trying again...')
                 time.sleep(5)
             else:
-                print(f'vault: {vault.address}')
+                logger.warn(f'vault: {token_entity.token.address}')
                 raise Exception(str(e))
 
 

scripts/treasury_exporter.py

@@ -3,15 +3,15 @@
 import time
 
 from brownie import chain
-from yearn.treasury.treasury import Treasury
+from yearn.treasury.treasury import YearnTreasury
 from yearn.outputs import victoria
 
 logger = logging.getLogger('yearn.treasury_exporter')
 sleep_interval = int(os.environ.get('SLEEP_SECONDS', '0'))
 
 
 def main():
-    treasury = Treasury(watch_events_forever=True)
+    treasury = YearnTreasury(watch_events_forever=True)
     for block in chain.new_blocks(height_buffer=12):
         start_time = time.time()
         treasury.export(block.number, block.timestamp)

scripts/wallet_exporter.py

@@ -1,18 +1,22 @@
 import logging
+import os
 from datetime import datetime, timezone
 from itertools import count
 
+import requests
 from brownie import chain
-from yearn.outputs.postgres.postgres import postgres
+from yearn.entities import UserTx
+from yearn.historical_helper import export_historical, has_data, time_tracking
+from yearn.outputs.postgres.utils import last_recorded_block
 from yearn.utils import closest_block_after_timestamp
-from yearn.historical_helper import export_historical, time_tracking, has_data
 from yearn.yearn import Yearn
 
 logger = logging.getLogger('yearn.wallet_exporter')
 
-postgres_cached_thru_block = postgres.last_recorded_block('user_txs')
+postgres_cached_thru_block = last_recorded_block(UserTx)
 postgres_cached_thru_ts = chain[postgres_cached_thru_block].timestamp
 
+
 def main():
     start = datetime.now(tz=timezone.utc)
     # end: 2020-02-12 first iearn deployment