From db500e603ef325b489ff4d3cb92cf99c8bae06bb Mon Sep 17 00:00:00 2001 From: daviidarr Date: Tue, 23 Apr 2024 15:44:15 +0200 Subject: [PATCH] alchemy_getTokenMetadata --- plex/plex.py | 48 +++++++++++++++++----------------------------- pnl_explain.py | 6 +++--- utils/coingecko.py | 32 +++++++++++++++++++++++++++++++ 3 files changed, 53 insertions(+), 33 deletions(-) diff --git a/plex/plex.py b/plex/plex.py index 5e371f5..6fbeab8 100644 --- a/plex/plex.py +++ b/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 diff --git a/pnl_explain.py b/pnl_explain.py index 10a6ae1..a2067f5 100644 --- a/pnl_explain.py +++ b/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: diff --git a/utils/coingecko.py b/utils/coingecko.py index fc0bdbf..9286305 100644 --- a/utils/coingecko.py +++ b/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',