From b0f4d5ae10bc64564f1278500c95fd477ce104c1 Mon Sep 17 00:00:00 2001 From: v1nvn Date: Wed, 28 Aug 2024 15:21:00 +0530 Subject: [PATCH] fix: unscale by token decimals --- .vscode/settings.json | 5 +- HISTORY.rst | 3 + credmark/cmf/types/token_erc20.py | 291 +++++++++++++++++------------- 3 files changed, 171 insertions(+), 128 deletions(-) diff --git a/.vscode/settings.json b/.vscode/settings.json index 162115f..2e0069c 100644 --- a/.vscode/settings.json +++ b/.vscode/settings.json @@ -172,5 +172,8 @@ "yfarm", "ZEPPELINOS", "Zora" - ] + ], + "[python]": { + "editor.defaultFormatter": "charliermarsh.ruff" + }, } \ No newline at end of file diff --git a/HISTORY.rst b/HISTORY.rst index 7d70c07..941c81a 100644 --- a/HISTORY.rst +++ b/HISTORY.rst @@ -3,6 +3,9 @@ History ------- +0.8.148 [2024-08-28] +* fix: unscale by token decimals + 0.8.147 [2024-08-28] * fix: add raw_amount column to token balance diff --git a/credmark/cmf/types/token_erc20.py b/credmark/cmf/types/token_erc20.py index 21839f2..4d33c51 100644 --- a/credmark/cmf/types/token_erc20.py +++ b/credmark/cmf/types/token_erc20.py @@ -15,19 +15,24 @@ from .contract import Contract from .data.erc_standard_data import ERC20_BASE_ABI from .data.erc_standard_data_alt import ERC20_BASE_ABI_ALT -from .data.fungible_token_data import FUNGIBLE_TOKEN_DATA_BY_ADDRESS, FUNGIBLE_TOKEN_DATA_BY_SYMBOL, NATIVE_TOKEN +from .data.fungible_token_data import ( + FUNGIBLE_TOKEN_DATA_BY_ADDRESS, + FUNGIBLE_TOKEN_DATA_BY_SYMBOL, + NATIVE_TOKEN, +) def get_token_from_configuration( - chain_id: int, - symbol: Union[str, None] = None, - address: Union[Address, None] = None, - is_native_token: bool = False) -> Union[dict, None]: - + chain_id: int, + symbol: Union[str, None] = None, + address: Union[Address, None] = None, + is_native_token: bool = False, +) -> Union[dict, None]: if is_native_token: native_token_meta = NATIVE_TOKEN[chain_id] - if ((symbol is None or native_token_meta['symbol'] == symbol) and - (address is None or native_token_meta['address'] == Address(address))): + if (symbol is None or native_token_meta["symbol"] == symbol) and ( + address is None or native_token_meta["address"] == Address(address) + ): return native_token_meta else: return None @@ -37,8 +42,9 @@ def get_token_from_configuration( if symbol is not None: token_meta = chain_tokens_by_symbol.get(symbol, None) - if ((token_meta is not None) and - (address is None or token_meta['address'] == Address(address))): + if (token_meta is not None) and ( + address is None or token_meta["address"] == Address(address) + ): return token_meta else: return None @@ -80,14 +86,15 @@ def update_cache(self, field, chain_id, block_number, value): self._cache[field][chain_id] = {} self._cache[field][chain_id][block_number] = value - _meta: TokenMetadata = PrivateAttr( - default_factory=lambda: Token.TokenMetadata()) # pylint: disable=unnecessary-lambda + _meta: TokenMetadata = PrivateAttr(default_factory=lambda: Token.TokenMetadata()) # pylint: disable=unnecessary-lambda class Config: schema_extra = { - 'examples': [{'address': '0x7Fc66500c84A76Ad7e9c93437bFc5Ac33E2DDaE9'}, - {'symbol': 'AAVE'} - ] + Contract.Config.schema_extra['examples'] + "examples": [ + {"address": "0x7Fc66500c84A76Ad7e9c93437bFc5Ac33E2DDaE9"}, + {"symbol": "AAVE"}, + ] + + Contract.Config.schema_extra["examples"] } @classmethod @@ -100,8 +107,7 @@ def validate(cls, value): return value if isinstance(value, Token): return value - raise TypeError( - f'{cls.__name__} must be deserialized with an str or dict') + raise TypeError(f"{cls.__name__} must be deserialized with an str or dict") def __new__(cls, *args, **data): if cls == NativeToken: @@ -110,26 +116,24 @@ def __new__(cls, *args, **data): if len(args) > 0: if isinstance(args[0], str): if evm_address_regex.match(args[0]) is not None: - if 'address' not in data: - data['address'] = args[0] + if "address" not in data: + data["address"] = args[0] else: - if 'symbol' not in data: - data['symbol'] = args[0] + if "symbol" not in data: + data["symbol"] = args[0] elif isinstance(args[0], dict): data = args[0] | data context = credmark.cmf.model.ModelContext.current_context() - symbol = data.get('symbol', None) - address = data.get('address', None) + symbol = data.get("symbol", None) + address = data.get("address", None) if symbol is None and address is None: return super().__new__(cls) token_data = get_token_from_configuration( - chain_id=context.chain_id, - symbol=symbol, - address=address, - is_native_token=True) + chain_id=context.chain_id, symbol=symbol, address=address, is_native_token=True + ) if token_data is not None: return super().__new__(NativeToken) @@ -140,47 +144,48 @@ def __init__(self, *args, **data): if len(args) > 0: if isinstance(args[0], str): if evm_address_regex.match(args[0]) is not None: - if 'address' not in data: - data['address'] = args[0] + if "address" not in data: + data["address"] = args[0] else: - if 'symbol' not in data: - data['symbol'] = args[0] + if "symbol" not in data: + data["symbol"] = args[0] elif isinstance(args[0], dict): data = args[0] | data - if 'address' not in data and 'symbol' not in data: - raise ModelDataError('One of address or symbol is required') + if "address" not in data and "symbol" not in data: + raise ModelDataError("One of address or symbol is required") - if 'address' not in data and 'symbol' in data: + if "address" not in data and "symbol" in data: context = credmark.cmf.model.ModelContext.current_context() token_data = get_token_from_configuration( - chain_id=context.chain_id, symbol=data['symbol']) + chain_id=context.chain_id, symbol=data["symbol"] + ) if token_data is None: raise ModelDataError(f'Unsupported symbol: {data["symbol"]}') - data['address'] = token_data['address'] - if 'meta' not in data: - data['meta'] = {} - data['meta']['symbol'] = token_data['symbol'] - data['meta']['name'] = token_data['name'] - data['meta']['decimals'] = token_data['decimals'] - data['meta']['wrapped'] = (Address(token_data['wrapped']) - if 'wrapped' in token_data else None) - - if token_data.get('set_loaded', False): # Special case for BTC + data["address"] = token_data["address"] + if "meta" not in data: + data["meta"] = {} + data["meta"]["symbol"] = token_data["symbol"] + data["meta"]["name"] = token_data["name"] + data["meta"]["decimals"] = token_data["decimals"] + data["meta"]["wrapped"] = ( + Address(token_data["wrapped"]) if "wrapped" in token_data else None + ) + + if token_data.get("set_loaded", False): # Special case for BTC self._loaded = True - data['meta']['set_loaded'] = True + data["meta"]["set_loaded"] = True else: - data['meta']['set_loaded'] = False + data["meta"]["set_loaded"] = False - if data['address'] == Address.null(): - raise ModelDataError( - f'NULL address ({Address.null()}) is not a valid Token Address') + if data["address"] == Address.null(): + raise ModelDataError(f"NULL address ({Address.null()}) is not a valid Token Address") - if 'abi' not in data: - data['abi'] = ERC20_BASE_ABI + if "abi" not in data: + data["abi"] = ERC20_BASE_ABI super().__init__(**data) @@ -206,7 +211,8 @@ def as_erc20(self, set_loaded=False, use_alt=False): _ = self.proxy_for.abi except BlockNumberOutOfRangeError as err: raise BlockNumberOutOfRangeError( - err.data.message + f' for contract {self.address}') from err + err.data.message + f" for contract {self.address}" + ) from err except ModelDataError: self.proxy_for.set_abi(erc20_abi, set_loaded=True) @@ -226,8 +232,10 @@ def try_erc20_property(self, prop_name): prop_value = self.functions[prop_name]().call() # type: ignore except (BadFunctionCallOutput, ABIFunctionNotFound) as err: raise ModelDataError( - f'No {prop_name} function on token {self.address}, non ERC20 Compliant' - f' proxied by {self.proxy_for.address}' if self.proxy_for is not None else '' + f"No {prop_name} function on token {self.address}, non ERC20 Compliant" + f" proxied by {self.proxy_for.address}" + if self.proxy_for is not None + else "" ) from err if prop_value is None: raise ModelDataError(f"Token.{prop_name} is None") @@ -236,96 +244,115 @@ def try_erc20_property(self, prop_name): @property def symbol(self) -> str: self._load() - current_block = int(credmark.cmf.model.ModelContext.current_context().web3.eth.default_block) + current_block = int( + credmark.cmf.model.ModelContext.current_context().web3.eth.default_block + ) current_chain_id = credmark.cmf.model.ModelContext.current_context().chain_id if not self._meta.set_loaded: - self._meta.symbol = self._meta.get_cache(field='symbol', - chain_id=current_chain_id, block_number=current_block) + self._meta.symbol = self._meta.get_cache( + field="symbol", chain_id=current_chain_id, block_number=current_block + ) if self._meta.symbol is None: try: - symbol_tmp = self.try_erc20_property('symbol') + symbol_tmp = self.try_erc20_property("symbol") except ModelDataError: - symbol_tmp = self.try_erc20_property('SYMBOL') + symbol_tmp = self.try_erc20_property("SYMBOL") if isinstance(symbol_tmp, bytes): - symbol_tmp = symbol_tmp.decode( - 'utf-8', errors='strict').replace('\x00', '') + symbol_tmp = symbol_tmp.decode("utf-8", errors="strict").replace("\x00", "") elif isinstance(symbol_tmp, str): - symbol_tmp = symbol_tmp.replace('\x00', '') + symbol_tmp = symbol_tmp.replace("\x00", "") elif not isinstance(symbol_tmp, str): - raise ModelDataError(f'Unknown value for symbol {symbol_tmp}') + raise ModelDataError(f"Unknown value for symbol {symbol_tmp}") self._meta.symbol = symbol_tmp - self._meta.update_cache(field='symbol', - chain_id=current_chain_id, block_number=current_block, - value=self._meta.symbol) + self._meta.update_cache( + field="symbol", + chain_id=current_chain_id, + block_number=current_block, + value=self._meta.symbol, + ) return self._meta.symbol @property def decimals(self) -> int: self._load() - current_block = int(credmark.cmf.model.ModelContext.current_context().web3.eth.default_block) + current_block = int( + credmark.cmf.model.ModelContext.current_context().web3.eth.default_block + ) current_chain_id = credmark.cmf.model.ModelContext.current_context().chain_id if not self._meta.set_loaded: - self._meta.decimals = self._meta.get_cache(field='decimals', - chain_id=current_chain_id, - block_number=current_block) + self._meta.decimals = self._meta.get_cache( + field="decimals", chain_id=current_chain_id, block_number=current_block + ) if self._meta.decimals is None: try: - self._meta.decimals = self.try_erc20_property('decimals') + self._meta.decimals = self.try_erc20_property("decimals") except ModelDataError: - self._meta.decimals = self.try_erc20_property('DECIMALS') - self._meta.update_cache(field='decimals', - chain_id=current_chain_id, block_number=current_block, - value=self._meta.decimals) + self._meta.decimals = self.try_erc20_property("DECIMALS") + self._meta.update_cache( + field="decimals", + chain_id=current_chain_id, + block_number=current_block, + value=self._meta.decimals, + ) return self._meta.decimals @property def name(self) -> str: self._load() - current_block = int(credmark.cmf.model.ModelContext.current_context().web3.eth.default_block) + current_block = int( + credmark.cmf.model.ModelContext.current_context().web3.eth.default_block + ) current_chain_id = credmark.cmf.model.ModelContext.current_context().chain_id if not self._meta.set_loaded: - self._meta.name = self._meta.get_cache(field='name', - chain_id=current_chain_id, - block_number=current_block) + self._meta.name = self._meta.get_cache( + field="name", chain_id=current_chain_id, block_number=current_block + ) if self._meta.name is None: try: - name_tmp = self.try_erc20_property('name') + name_tmp = self.try_erc20_property("name") except ModelDataError: - name_tmp = self.try_erc20_property('NAME') + name_tmp = self.try_erc20_property("NAME") if isinstance(name_tmp, bytes): - name_tmp = name_tmp.decode( - 'utf-8', errors='strict').replace('\x00', '') + name_tmp = name_tmp.decode("utf-8", errors="strict").replace("\x00", "") elif isinstance(name_tmp, str): - name_tmp = name_tmp.replace('\x00', '') + name_tmp = name_tmp.replace("\x00", "") elif not isinstance(name_tmp, str): - raise ModelDataError(f'Unknown value for name {name_tmp}') + raise ModelDataError(f"Unknown value for name {name_tmp}") self._meta.name = name_tmp - self._meta.update_cache(field='name', - chain_id=current_chain_id, block_number=current_block, - value=self._meta.name) + self._meta.update_cache( + field="name", + chain_id=current_chain_id, + block_number=current_block, + value=self._meta.name, + ) return self._meta.name @property def total_supply(self) -> int: self._load() - current_block = int(credmark.cmf.model.ModelContext.current_context().web3.eth.default_block) + current_block = int( + credmark.cmf.model.ModelContext.current_context().web3.eth.default_block + ) current_chain_id = credmark.cmf.model.ModelContext.current_context().chain_id if not self._meta.set_loaded: - self._meta.total_supply = self._meta.get_cache(field='total_supply', - chain_id=current_chain_id, - block_number=current_block) + self._meta.total_supply = self._meta.get_cache( + field="total_supply", chain_id=current_chain_id, block_number=current_block + ) if self._meta.total_supply is None: - self._meta.total_supply = self.try_erc20_property('totalSupply') - self._meta.update_cache(field='total_supply', - chain_id=current_chain_id, block_number=current_block, - value=self._meta.total_supply) + self._meta.total_supply = self.try_erc20_property("totalSupply") + self._meta.update_cache( + field="total_supply", + chain_id=current_chain_id, + block_number=current_block, + value=self._meta.total_supply, + ) return self._meta.total_supply @property @@ -333,7 +360,10 @@ def total_supply_scaled(self) -> float: return self.scaled(self.total_supply) def scaled(self, value) -> float: - return value / (10 ** self.decimals) + return value / (10**self.decimals) + + def unscaled(self, value) -> float: + return value * (10**self.decimals) def balance_of(self, address: ChecksumAddress | str | bytes | Address) -> int: if isinstance(address, Address): @@ -359,6 +389,7 @@ class TokenInfo(Token): """ Subclass of Token containing its related metadata. """ + meta: Token.TokenMetadata @property @@ -375,37 +406,38 @@ def __init__(self, *args, **kwargs) -> None: context = credmark.cmf.model.ModelContext.current_context() token_data = NATIVE_TOKEN[context.chain_id] if token_data is None: - raise ModelRunError( - f'No native token specified for chain id: {context.chain_id}') + raise ModelRunError(f"No native token specified for chain id: {context.chain_id}") if len(args) > 0: if isinstance(args[0], str): if evm_address_regex.match(args[0]) is not None: - if 'address' not in kwargs: - kwargs['address'] = args[0] + if "address" not in kwargs: + kwargs["address"] = args[0] else: - if 'symbol' not in kwargs: - kwargs['symbol'] = args[0] + if "symbol" not in kwargs: + kwargs["symbol"] = args[0] elif isinstance(args[0], dict): kwargs = args[0] | kwargs - symbol = kwargs.get('symbol', None) - address = kwargs.get('address', None) - if symbol is not None and token_data['symbol'] != symbol: + symbol = kwargs.get("symbol", None) + address = kwargs.get("address", None) + if symbol is not None and token_data["symbol"] != symbol: raise ModelRunError( f'Wrong symbol {symbol} specified for {token_data["symbol"]} ' - f'for chain id: {context.chain_id}') - if address is not None and token_data['address'] != Address(address): + f'for chain id: {context.chain_id}' + ) + if address is not None and token_data["address"] != Address(address): raise ModelRunError( f'Wrong address {address} specified for {token_data["address"]} ' - f'for chain id: {context.chain_id}') - super().__init__(**({'address': token_data['address']})) + f'for chain id: {context.chain_id}' + ) + super().__init__(**({"address": token_data["address"]})) self._meta.abi = ABI([]) - self._meta.symbol = token_data['symbol'] - self._meta.name = token_data['name'] - self._meta.decimals = token_data['decimals'] - self._meta.wrapped = Address(token_data['wrapped']) + self._meta.symbol = token_data["symbol"] + self._meta.name = token_data["name"] + self._meta.decimals = token_data["decimals"] + self._meta.wrapped = Address(token_data["wrapped"]) self._meta.total_supply = 0 self._loaded = True @@ -417,25 +449,25 @@ def balance_of(self, address: ChecksumAddress) -> int: def balance_of_scaled(self, address: ChecksumAddress) -> float: context = credmark.cmf.model.ModelContext.current_context() balance = self.balance_of(address) - return float(context.web3.from_wei(balance, 'ether')) + return float(context.web3.from_wei(balance, "ether")) @property def symbol(self): if self._meta.symbol is not None: return self._meta.symbol - raise ModelRunError(f'No symbol found for {self}') + raise ModelRunError(f"No symbol found for {self}") @property def name(self): if self._meta.name is not None: return self._meta.name - raise ModelRunError(f'No name found for {self}') + raise ModelRunError(f"No name found for {self}") @property def decimals(self): if self._meta.decimals is not None: return self._meta.decimals - raise ModelRunError(f'No decimals found for {self}') + raise ModelRunError(f"No decimals found for {self}") @property def ledger(self) -> None: @@ -444,7 +476,7 @@ def ledger(self) -> None: def wrapped(self) -> Token: if self._meta.wrapped is not None: return Token(address=self._meta.wrapped) - raise ValueError('No wrapper Token found') + raise ValueError("No wrapper Token found") class Tokens(IterableListGenericDTO[Token]): @@ -452,25 +484,30 @@ class Tokens(IterableListGenericDTO[Token]): Iterable list of Token instances. """ - tokens: List[Token] = DTOField( - default=[], description="An iterable list of Token Objects") - _iterator: str = PrivateAttr('tokens') + tokens: List[Token] = DTOField(default=[], description="An iterable list of Token Objects") + _iterator: str = PrivateAttr("tokens") class Config: schema_extra = { - "examples": [{"tokens": ['0x6B175474E89094C44Da98b954EedeAC495271d0F', # DAI - '0x514910771AF9Ca656af840dff83E8264EcF986CA']}] # LINK + "examples": [ + { + "tokens": [ + "0x6B175474E89094C44Da98b954EedeAC495271d0F", # DAI + "0x514910771AF9Ca656af840dff83E8264EcF986CA", + ] + } + ] # LINK } @classmethod - def from_addresses(cls, addresses: List[Address]) -> 'Tokens': + def from_addresses(cls, addresses: List[Address]) -> "Tokens": """ Returns a Tokens instance from a list of addresses. """ return cls(tokens=[Token(address=address) for address in addresses]) @classmethod - def empty(cls) -> 'Tokens': + def empty(cls) -> "Tokens": """ Returns an empty Tokens instance. """