alchemy_getTokenMetadata

plex/plex.py

@@ -7,30 +7,22 @@
 import streamlit as st
 import yaml
 from pandas import DataFrame
-from utils.coingecko import myCoinGeckoAPI
+from utils.coingecko import ScannerAPI
+
 
 class PnlExplainer:
-    def __init__(self, categories: dict[str, str]):
+    def __init__(self, categories: dict[str, str], alchemy_key: str = None):
         self.categories = categories
-        with st.spinner('fetching meta data'):
-            address_df = []
-            line_list = myCoinGeckoAPI().get_coins_list(include_platform='true')
-            for line in line_list:
-                if line['platforms']:
-                    address_df += [{'symbol': line['symbol'],
-                                    'chain': myCoinGeckoAPI.debank_mapping[chain] if chain in myCoinGeckoAPI.debank_mapping else chain,
-                                    'address': address}
-                                   for chain, address in line['platforms'].items()]
-            self.address_map = pd.DataFrame(address_df).set_index('address')
-
-    def validate_categories(self, data) -> bool:
+        self.etherscan_api = ScannerAPI(alchemy_key)
+
+    def underlying(self, asset: str) -> str:
+        categories = {key.lower(): value for key, value in self.categories.items()}
+        return categories.get(asset.lower(), asset)
+
+    def validate_categories(self, data) -> None:
         if missing_category := set(data['asset']) - set(self.categories.keys()):
-            st.warning(f"Categories need to be updated. Please categorize the following assets: {missing_category}")
-            return False
-        # if missing_underlying := set(self.categories.values()) - set(data['asset']):
-        #     st.warning(f"I need underlying {missing_underlying} to have a position, maybe get some dust? Sorry...")
-        #     return False
-        return True
+            st.error(f"Categories need to be updated. Please categorize the following assets: {missing_category}")
+            st.stop()
 
     def explain(self, start_snapshot: pd.DataFrame, end_snapshot: pd.DataFrame) -> DataFrame:
         snapshot_start = start_snapshot.set_index([col for col in start_snapshot.columns if col not in ['price', 'amount', 'value', 'timestamp']])
@@ -40,12 +32,11 @@ def explain(self, start_snapshot: pd.DataFrame, end_snapshot: pd.DataFrame) -> D
         before_pos = data[data.index.isin(set(snapshot_start.index) - set(snapshot_end.index))].reset_index()
         after_pos = data[data.index.isin(set(snapshot_end.index) - set(snapshot_start.index))].reset_index()
 
-        if not self.validate_categories(common_pos):
-            return pd.DataFrame()
+        self.validate_categories(common_pos)
 
-        common_pos['underlying'] = common_pos['asset'].apply(lambda x: self.categories[x])
-        before_pos['underlying'] = before_pos['asset'].apply(lambda x: self.categories[x])
-        after_pos['underlying'] = after_pos['asset'].apply(lambda x: self.categories[x])
+        common_pos['underlying'] = common_pos['asset'].apply(self.underlying)
+        before_pos['underlying'] = before_pos['asset'].apply(self.underlying)
+        after_pos['underlying'] = after_pos['asset'].apply(self.underlying)
         # TODO: messy since we need position on same chain, USD and EUR don't work...need coingecko snap.
         common_pos[['P_underlying_start', 'P_underlying_end']] = common_pos.apply(lambda x: common_pos.loc[common_pos['asset'] == x['underlying'], ['price_start', 'price_end']].mean(), axis=1)
         # data['dP_basis'] = data['dP'] / data['dP_underlying']
@@ -87,10 +78,7 @@ def format_transactions(self, start_snapshot_timestamp: int, end_snapshot_timest
         tx_pnl['timestamp_end'] = end_snapshot_timestamp
         tx_pnl['hold_mode'] = tx_pnl['type']
         tx_pnl['asset'] = tx_pnl.apply(
-            lambda x: self.address_map[self.address_map['chain'] == x['chain']].loc[x['asset'], 'symbol']
-            if x['asset'] in self.address_map.index else x['asset'],
-            axis=1)
-        categories = {key.lower(): value for key, value in self.categories.items()}
-        tx_pnl['underlying'] = tx_pnl['asset'].apply(lambda x: categories[x.lower()] if x.lower() in categories.keys() else x)
+                lambda x: self.etherscan_api.get_token_symbol(x['asset'], x['chain']), axis=1)
+        tx_pnl['underlying'] = tx_pnl['asset'].apply(self.underlying)
 
         return tx_pnl

pnl_explain.py

@@ -37,7 +37,7 @@
     st.session_state.api = DebankAPI(json_db=raw_data_db,
                                      plex_db=st.session_state.plex_db,
                                      parameters=st.session_state.parameters)
-    st.session_state.pnl_explainer = PnlExplainer(st.session_state.plex_db.query_categories())
+    st.session_state.pnl_explainer = PnlExplainer(st.session_state.plex_db.query_categories(),st.secrets['alchemy_key'])
 
 addresses = st.session_state.parameters['profile']['addresses']
 risk_tab, pnl_tab = st.tabs(
@@ -123,11 +123,11 @@
 
         transactions = st.session_state.plex_db.query_table_between(addresses, start_timestamp, end_timestamp, "transactions")
         st.session_state.transactions = st.session_state.pnl_explainer.format_transactions(start_timestamp, end_timestamp, transactions)
-
+        st.session_state.transactions.rename(columns={'pnl': 'value'}, inplace=True)
         display_pivot(st.session_state.transactions,
                       rows=['underlying', 'asset'],
                       columns=['type'],
-                      values=['gas', 'pnl'],
+                      values=['gas', 'value'],
                       hidden=['id', 'protocol', 'chain'])
 
         if 'plex' in st.session_state:

utils/coingecko.py

@@ -7,9 +7,41 @@
 
 import pandas as pd
 import pycoingecko
+import requests
 import streamlit
 
 
+class ScannerAPI:
+    def __init__(self, api_key):
+        self.api_key = api_key
+        self.network_map = {
+            "eth": "eth-mainnet",
+            "op": "op-mainnet",
+            "arb": "arb-mainnet",
+            "matic": "polygon-mainnet",
+            "base": "base-mainnet",
+        }
+
+    @streamlit.cache_data
+    def get_token_symbol(_self, address, network):
+        try:
+            url = f"https://{_self.network_map[network]}.g.alchemy.com/v2/{_self.api_key}"
+            payload = {
+                # "id": 1,
+                "jsonrpc": "2.0",
+                "method": "alchemy_getTokenMetadata",
+                "params": [address]
+            }
+            headers = {
+                "accept": "application/json",
+                "content-type": "application/json"
+            }
+            response = requests.post(url, json=payload, headers=headers)
+            data = response.json()
+            return data["result"]["symbol"].lower()
+        except Exception as e:
+            return address
+
 class myCoinGeckoAPI(pycoingecko.CoinGeckoAPI):
     defillama_mapping = ({'id': 'id',
                          'symbol': 'symbol',