diff --git a/credmark/cmf/types/pool.py b/credmark/cmf/types/pool.py new file mode 100644 index 00000000..cb3796e7 --- /dev/null +++ b/credmark/cmf/types/pool.py @@ -0,0 +1,91 @@ +from typing import List, Union +from credmark.cmf.model.errors import ModelDataError +from credmark.cmf.types.token import Tokens, Token +from credmark.cmf.types.contract import Contract +from credmark.dto import DTOField, IterableListGenericDTO, PrivateAttr +from credmark.cmf.types.portfolio import Portfolio + + +class Pool(Contract): + """ + A pool is a contract that consolidates assets on behalf of many accounts to perform + a financial function. + """ + + tokens: Tokens = DTOField(description="The tokens that are supported by the pool.") + portfolio: Portfolio = DTOField(description="The positions of the pool") + + def __init__(self, **data): + if len(data.get("tokens")) != len(data.get("balances")): + raise ModelDataError( + message="Pool balances and Pool tokens must be the same size array.") + super().__init__(**data) + + +class SwapPool(Pool): + """ + Swap Pools are pools that represent an liquidity pool enabling Swaps between tokens. + These are commonly the building blocks of DEXes, such as Uniswap and Curve Finance. + + Note: The token reserves may not be custodied by the pool contract itself, representing + virtual custody. + """ + + +class ConstantProductSwapPool(SwapPool): + """ + SymmetricSwapPools are Swap Pools that follow an x*y=k based swap algoritm. + """ + + def __init__(self, **data): + if len(data.get("tokens")) != 2: + raise ModelDataError( + message="There must be 2 tokens and 2 balances in a FlatAMMSwapPool") + super().__init__(**data) + + @property + def swap_spot_price(self, base: Union[Token, None] = None): + quote = self.tokens[1] + if base is None: + base = self.tokens[0] + if base == self.tokens[1]: + quote = self.tokens[0] + base_liquidity = None + quote_liquidity = None + for position in self.portfolio: + if position.asset == base: + base_liquidity = position.amount + if position.asset == quote: + quote_liquidity = position.amount + if base_liquidity is None: + raise ModelDataError("Token " + base.symbol + " not found in Pool.") + if quote_liquidity is None: + raise ModelDataError("Token " + quote.symbol + " not found in Pool.") + if quote_liquidity == 0 or base_liquidity == 0: + raise ModelDataError("No Liquidity in Pool " + self.address) + return base_liquidity / quote_liquidity + + +class TokenizedConstantProductSwapPool(SwapPool): + + lp_token: Union[Token, None] = DTOField(default=None, + description="The Liquidity Provider " + "Token for this pool.") + + +class Pools(IterableListGenericDTO[Pool]): + pools: List[Pool] = DTOField( + default=[], description="An iterable list of Pool Objects") + _iterator: str = PrivateAttr('pools') + + +class SwapPools(IterableListGenericDTO[SwapPool]): + pools: List[Pool] = DTOField( + default=[], description="An iterable list of SwapPool Objects") + _iterator: str = PrivateAttr('pools') + + +class ConstantProductSwapPools(IterableListGenericDTO[ConstantProductSwapPool]): + pools: List[Pool] = DTOField( + default=[], description="An iterable list of FlatAmmPairedSwapPool Objects") + _iterator: str = PrivateAttr('pools') diff --git a/credmark/dto/__init__.py b/credmark/dto/__init__.py index 93bc595d..699112d8 100644 --- a/credmark/dto/__init__.py +++ b/credmark/dto/__init__.py @@ -224,3 +224,20 @@ def dict(self): json_dump, json_dumps ) + + +class MetadataDto(): + _meta: Union[dict, DTO] = PrivateAttr( + default_factory=lambda: {}) # pylint: disable=unnecessary-lambda + + def __init__(self, **data): + super().__init__(**data) + meta = data.get('meta', None) + if meta is not None: + if isinstance(meta, dict): + self._meta = type(self._meta)(**meta) + if isinstance(meta, type(self._meta)): + self._meta = meta + + for key, value in data.items(): + self._meta.__setattr__(key, value)