diff --git a/bfxapi/labeler.py b/bfxapi/labeler.py new file mode 100644 index 0000000..eb7076c --- /dev/null +++ b/bfxapi/labeler.py @@ -0,0 +1,20 @@ +from typing import Generic, TypeVar, Iterable, Optional, List, Any + +T = TypeVar("T") + +class _Serializer(Generic[T]): + def __init__(self, name: str, labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]): + self.name, self.__labels, self.__IGNORE = name, labels, IGNORE + + def __serialize(self, *args: Any, skip: Optional[List[str]]) -> Iterable[T]: + labels = list(filter(lambda label: label not in (skip or list()), self.__labels)) + + if len(labels) > len(args): + raise Exception(" and <*args> arguments should contain the same amount of elements.") + + for index, label in enumerate(labels): + if label not in self.__IGNORE: + yield label, args[index] + + def parse(self, *values: Any, skip: Optional[List[str]] = None) -> T: + return dict(self.__serialize(*values, skip=skip)) \ No newline at end of file diff --git a/bfxapi/rest/BfxRestInterface.py b/bfxapi/rest/BfxRestInterface.py index b7a8893..6644247 100644 --- a/bfxapi/rest/BfxRestInterface.py +++ b/bfxapi/rest/BfxRestInterface.py @@ -104,7 +104,7 @@ class _RestPublicEndpoints(_Requests): def f_ticker(self, currency: str) -> FundingCurrencyTicker: return serializers.FundingCurrencyTicker.parse(*self._GET(f"ticker/f{currency}"), skip=["SYMBOL"]) - def tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> TickersHistories: + def tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[TickersHistory]: params = { "symbols": ",".join(symbols), "start": start, "end": end, @@ -115,51 +115,51 @@ class _RestPublicEndpoints(_Requests): return [ serializers.TickersHistory.parse(*subdata) for subdata in data ] - def t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> TradingPairTrades: + def t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[TradingPairTrade]: params = { "limit": limit, "start": start, "end": end, "sort": sort } data = self._GET(f"trades/{'t' + pair}/hist", params=params) return [ serializers.TradingPairTrade.parse(*subdata) for subdata in data ] - def f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> FundingCurrencyTrades: + def f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[FundingCurrencyTrade]: params = { "limit": limit, "start": start, "end": end, "sort": sort } data = self._GET(f"trades/{'f' + currency}/hist", params=params) return [ serializers.FundingCurrencyTrade.parse(*subdata) for subdata in data ] - def t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> TradingPairBooks: + def t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairBook]: return [ serializers.TradingPairBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/{precision}", params={ "len": len }) ] - def f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> FundingCurrencyBooks: + def f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyBook]: return [ serializers.FundingCurrencyBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/{precision}", params={ "len": len }) ] - def t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> TradingPairRawBooks: + def t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairRawBook]: return [ serializers.TradingPairRawBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/R0", params={ "len": len }) ] - def f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> FundingCurrencyRawBooks: + def f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyRawBook]: return [ serializers.FundingCurrencyRawBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/R0", params={ "len": len }) ] def stats_hist( self, resource: str, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None - ) -> Stats: + ) -> List[Statistic]: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET(f"stats1/{resource}/hist", params=params) - return [ serializers.Stat.parse(*subdata) for subdata in data ] + return [ serializers.Statistic.parse(*subdata) for subdata in data ] def stats_last( self, resource: str, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None - ) -> Stat: + ) -> Statistic: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET(f"stats1/{resource}/last", params=params) - return serializers.Stat.parse(*data) + return serializers.Statistic.parse(*data) def candles_hist( self, resource: str, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None - ) -> Candles: + ) -> List[Candle]: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET(f"candles/{resource}/hist", params=params) return [ serializers.Candle.parse(*subdata) for subdata in data ] @@ -173,7 +173,7 @@ class _RestPublicEndpoints(_Requests): data = self._GET(f"candles/{resource}/last", params=params) return serializers.Candle.parse(*data) - def derivatives_status(self, type: str, keys: List[str]) -> DerivativeStatuses: + def derivatives_status(self, type: str, keys: List[str]) -> List[DerivativesStatus]: params = { "keys": ",".join(keys) } data = self._GET(f"status/{type}", params=params) @@ -184,14 +184,14 @@ class _RestPublicEndpoints(_Requests): self, type: str, symbol: str, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None - ) -> DerivativeStatuses: + ) -> List[DerivativesStatus]: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET(f"status/{type}/{symbol}/hist", params=params) return [ serializers.DerivativesStatus.parse(*subdata, skip=[ "KEY" ]) for subdata in data ] - def liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> Liquidations: + def liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Liquidation]: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET("liquidations/hist", params=params) @@ -202,7 +202,7 @@ class _RestPublicEndpoints(_Requests): self, resource: str, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None - ) -> Leaderboards: + ) -> List[Leaderboard]: params = { "sort": sort, "start": start, "end": end, "limit": limit } data = self._GET(f"rankings/{resource}/hist", params=params) return [ serializers.Leaderboard.parse(*subdata) for subdata in data ] @@ -216,12 +216,12 @@ class _RestPublicEndpoints(_Requests): data = self._GET(f"rankings/{resource}/last", params=params) return serializers.Leaderboard.parse(*data) - def funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> FundingStats: + def funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingStatistic]: params = { "start": start, "end": end, "limit": limit } data = self._GET(f"funding/stats/{symbol}/hist", params=params) - return [ serializers.FundingStat.parse(*subdata) for subdata in data ] + return [ serializers.FundingStatistic.parse(*subdata) for subdata in data ] def conf(self, config: Config) -> Any: return self._GET(f"conf/{config}")[0] diff --git a/bfxapi/rest/serializers.py b/bfxapi/rest/serializers.py index 8bcdf34..95f4c26 100644 --- a/bfxapi/rest/serializers.py +++ b/bfxapi/rest/serializers.py @@ -1,27 +1,6 @@ -from typing import Generic, TypeVar, Iterable, Optional, List, Any - from . import typings -from .exceptions import BfxRestException - -T = TypeVar("T") - -class _Serializer(Generic[T]): - def __init__(self, name: str, labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]): - self.name, self.__labels, self.__IGNORE = name, labels, IGNORE - - def __serialize(self, *args: Any, skip: Optional[List[str]]) -> Iterable[T]: - labels = list(filter(lambda label: label not in (skip or list()), self.__labels)) - - if len(labels) > len(args): - raise BfxRestException(" and <*args> arguments should contain the same amount of elements.") - - for index, label in enumerate(labels): - if label not in self.__IGNORE: - yield label, args[index] - - def parse(self, *values: Any, skip: Optional[List[str]] = None) -> T: - return dict(self.__serialize(*values, skip=skip)) +from .. labeler import _Serializer #region Serializers definition for Rest Public Endpoints @@ -120,7 +99,7 @@ FundingCurrencyRawBook = _Serializer[typings.FundingCurrencyRawBook]("FundingCur "AMOUNT" ]) -Stat = _Serializer[typings.Stat]("Stat", labels=[ +Statistic = _Serializer[typings.Statistic]("Statistic", labels=[ "MTS", "VALUE" ]) @@ -189,7 +168,7 @@ Leaderboard = _Serializer[typings.Leaderboard]("Leaderboard", labels=[ "TWITTER_HANDLE" ]) -FundingStat = _Serializer[typings.FundingStat]("FundingStat", labels=[ +FundingStatistic = _Serializer[typings.FundingStatistic]("FundingStatistic", labels=[ "TIMESTAMP", "_PLACEHOLDER", "_PLACEHOLDER", diff --git a/bfxapi/rest/typings.py b/bfxapi/rest/typings.py index 4b225f4..7af75fb 100644 --- a/bfxapi/rest/typings.py +++ b/bfxapi/rest/typings.py @@ -2,140 +2,130 @@ from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Any #region Type hinting for Rest Public Endpoints -PlatformStatus = TypedDict("PlatformStatus", { - "OPERATIVE": int -}) +class PlatformStatus(TypedDict): + OPERATIVE: int -TradingPairTicker = TypedDict("TradingPairTicker", { - "SYMBOL": Optional[str], - "BID": float, - "BID_SIZE": float, - "ASK": float, - "ASK_SIZE": float, - "DAILY_CHANGE": float, - "DAILY_CHANGE_RELATIVE": float, - "LAST_PRICE": float, - "VOLUME": float, - "HIGH": float, - "LOW": float -}) +class TradingPairTicker(TypedDict): + SYMBOL: Optional[str] + BID: float + BID_SIZE: float + ASK: float + ASK_SIZE: float + DAILY_CHANGE: float + DAILY_CHANGE_RELATIVE: float + LAST_PRICE: float + VOLUME: float + HIGH: float + LOW: float -FundingCurrencyTicker = TypedDict("FundingCurrencyTicker", { - "SYMBOL": Optional[str], - "FRR": float, - "BID": float, - "BID_PERIOD": int, - "BID_SIZE": float, - "ASK": float, - "ASK_PERIOD": int, - "ASK_SIZE": float, - "DAILY_CHANGE": float, - "DAILY_CHANGE_RELATIVE": float, - "LAST_PRICE": float, - "VOLUME": float, - "HIGH": float, - "LOW": float, - "FRR_AMOUNT_AVAILABLE": float -}) +class FundingCurrencyTicker(TypedDict): + SYMBOL: Optional[str] + FRR: float + BID: float + BID_PERIOD: int + BID_SIZE: float + ASK: float + ASK_PERIOD: int + ASK_SIZE: float + DAILY_CHANGE: float + DAILY_CHANGE_RELATIVE: float + LAST_PRICE: float + VOLUME: float + HIGH: float + LOW: float + FRR_AMOUNT_AVAILABLE: float -TickersHistory = TypedDict("TickersHistory", { - "SYMBOL": str, - "BID": float, - "ASK": float, - "MTS": int -}) +class TickersHistory(TypedDict): + SYMBOL: str + BID: float + ASK: float + MTS: int -TickersHistories = List[TickersHistory] +class TradingPairTrade(TypedDict): + ID: int + MTS: int + AMOUNT: float + PRICE: float -(TradingPairTrade, FundingCurrencyTrade) = ( - TypedDict("TradingPairTrade", { "ID": int, "MTS": int, "AMOUNT": float, "PRICE": float }), - TypedDict("FundingCurrencyTrade", { "ID": int, "MTS": int, "AMOUNT": float, "RATE": float, "PERIOD": int }) -) +class FundingCurrencyTrade(TypedDict): + ID: int + MTS: int + AMOUNT: float + RATE: float + PERIOD: int -(TradingPairTrades, FundingCurrencyTrades) = (List[TradingPairTrade], List[FundingCurrencyTrade]) +class TradingPairBook(TypedDict): + PRICE: float + COUNT: int + AMOUNT: float + +class FundingCurrencyBook(TypedDict): + RATE: float + PERIOD: int + COUNT: int + AMOUNT: float + +class TradingPairRawBook(TypedDict): + ORDER_ID: int + PRICE: float + AMOUNT: float + +class FundingCurrencyRawBook(TypedDict): + OFFER_ID: int + PERIOD: int + RATE: float + AMOUNT: float -(TradingPairBook, FundingCurrencyBook) = ( - TypedDict("TradingPairBook", { "PRICE": float, "COUNT": int, "AMOUNT": float }), - TypedDict("FundingCurrencyBook", { "RATE": float, "PERIOD": int, "COUNT": int, "AMOUNT": float }) -) +class Statistic(TypedDict): + MTS: int + VALUE: float -(TradingPairBooks, FundingCurrencyBooks) = (List[TradingPairBook], List[FundingCurrencyBook]) +class Candle(TypedDict): + MTS: int + OPEN: float + CLOSE: float + HIGH: float + LOW: float + VOLUME: float -(TradingPairRawBook, FundingCurrencyRawBook) = ( - TypedDict("TradingPairRawBook", { "ORDER_ID": int, "PRICE": float, "AMOUNT": float }), - TypedDict("FundingCurrencyRawBook", { "OFFER_ID": int, "PERIOD": int, "RATE": float, "AMOUNT": float }), -) +class DerivativesStatus(TypedDict): + KEY: Optional[str] + MTS: int + DERIV_PRICE: float + SPOT_PRICE: float + INSURANCE_FUND_BALANCE: float + NEXT_FUNDING_EVT_TIMESTAMP_MS: int + NEXT_FUNDING_ACCRUED: float + NEXT_FUNDING_STEP: int + CURRENT_FUNDING: float + MARK_PRICE: float + OPEN_INTEREST: float + CLAMP_MIN: float + CLAMP_MAX: float -(TradingPairRawBooks, FundingCurrencyRawBooks) = (List[TradingPairRawBook], List[FundingCurrencyRawBook]) +class Liquidation(TypedDict): + POS_ID: int + MTS: int + SYMBOL: str + AMOUNT: float + BASE_PRICE: float + IS_MATCH: int + IS_MARKET_SOLD: int + PRICE_ACQUIRED: float -Stat = TypedDict("Stat", { - "MTS": int, - "VALUE": float -}) +class Leaderboard(TypedDict): + MTS: int + USERNAME: str + RANKING: int + VALUE: float + TWITTER_HANDLE: Optional[str] -Stats = List[Stat] - -Candle = TypedDict("Candle", { - "MTS": int, - "OPEN": float, - "CLOSE": float, - "HIGH": float, - "LOW": float, - "VOLUME": float -}) - -Candles = List[Candle] - -DerivativesStatus = TypedDict("DerivativesStatus", { - "KEY": Optional[str], - "MTS": int, - "DERIV_PRICE": float, - "SPOT_PRICE": float, - "INSURANCE_FUND_BALANCE": float, - "NEXT_FUNDING_EVT_TIMESTAMP_MS": int, - "NEXT_FUNDING_ACCRUED": float, - "NEXT_FUNDING_STEP": int, - "CURRENT_FUNDING": float, - "MARK_PRICE": float, - "OPEN_INTEREST": float, - "CLAMP_MIN": float, - "CLAMP_MAX": float -}) - -DerivativeStatuses = List[DerivativesStatus] - -Liquidation = TypedDict("Liquidation", { - "POS_ID": int, - "MTS": int, - "SYMBOL": str, - "AMOUNT": float, - "BASE_PRICE": float, - "IS_MATCH": int, - "IS_MARKET_SOLD": int, - "PRICE_ACQUIRED": float -}) - -Liquidations = List[Liquidation] - -Leaderboard = TypedDict("Leaderboard", { - "MTS": int, - "USERNAME": str, - "RANKING": int, - "VALUE": float, - "TWITTER_HANDLE": Optional[str] -}) - -Leaderboards = List[Leaderboard] - -FundingStat = TypedDict("FundingStat", { - "TIMESTAMP": int, - "FRR": float, - "AVG_PERIOD": float, - "FUNDING_AMOUNT": float, - "FUNDING_AMOUNT_USED": float, - "FUNDING_BELOW_THRESHOLD": float -}) - -FundingStats = List[FundingStat] +class FundingStatistic(TypedDict): + TIMESTAMP: int + FRR: float + AVG_PERIOD: float + FUNDING_AMOUNT: float + FUNDING_AMOUNT_USED: float + FUNDING_BELOW_THRESHOLD: float #endregion \ No newline at end of file