From 07241b1ba850bec476842e710fd2363dd8730859 Mon Sep 17 00:00:00 2001 From: Davide Casale Date: Wed, 14 Dec 2022 18:17:29 +0100 Subject: [PATCH] Add _Requests and _RestPublicEndpoints classes in bfxapi/rest/BfxRestInterface.py. --- bfxapi/rest/BfxRestInterface.py | 51 ++++++++++++++++++--------------- 1 file changed, 28 insertions(+), 23 deletions(-) diff --git a/bfxapi/rest/BfxRestInterface.py b/bfxapi/rest/BfxRestInterface.py index 018ce7b..30109c0 100644 --- a/bfxapi/rest/BfxRestInterface.py +++ b/bfxapi/rest/BfxRestInterface.py @@ -12,9 +12,13 @@ from .exceptions import RequestParametersError, ResourceNotFound class BfxRestInterface(object): def __init__(self, host): + self.public = _RestPublicEndpoints(host=host) + +class _Requests(object): + def __init__(self, host: str): self.host = host - def __GET(self, endpoint, params = None): + def _GET(self, endpoint, params = None): response = requests.get(f"{self.host}/{endpoint}", params=params) if response.status_code == HTTPStatus.NOT_FOUND: @@ -28,11 +32,12 @@ class BfxRestInterface(object): return data +class _RestPublicEndpoints(_Requests): def platform_status(self) -> PlatformStatus: - return serializers.PlatformStatus.parse(*self.__GET("platform/status")) + return serializers.PlatformStatus.parse(*self._GET("platform/status")) def tickers(self, symbols: List[str]) -> List[Union[TradingPairTicker, FundingCurrencyTicker]]: - data = self.__GET("tickers", params={ "symbols": ",".join(symbols) }) + data = self._GET("tickers", params={ "symbols": ",".join(symbols) }) parsers = { "t": serializers.TradingPairTicker.parse, "f": serializers.FundingCurrencyTicker.parse } @@ -55,10 +60,10 @@ class BfxRestInterface(object): return cast(List[FundingCurrencyTicker], data) def t_ticker(self, pair: str) -> TradingPairTicker: - return serializers.TradingPairTicker.parse(*self.__GET(f"ticker/t{pair}"), skip=["SYMBOL"]) + return serializers.TradingPairTicker.parse(*self._GET(f"ticker/t{pair}"), skip=["SYMBOL"]) def f_ticker(self, currency: str) -> FundingCurrencyTicker: - return serializers.FundingCurrencyTicker.parse(*self.__GET(f"ticker/f{currency}"), skip=["SYMBOL"]) + 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: params = { @@ -67,31 +72,31 @@ class BfxRestInterface(object): "limit": limit } - data = self.__GET("tickers/hist", params=params) + data = self._GET("tickers/hist", params=params) 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: params = { "limit": limit, "start": start, "end": end, "sort": sort } - data = self.__GET(f"trades/{'t' + pair}/hist", params=params) + 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: params = { "limit": limit, "start": start, "end": end, "sort": sort } - data = self.__GET(f"trades/{'f' + currency}/hist", params=params) + 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: - return [ serializers.TradingPairBook.parse(*subdata) for subdata in self.__GET(f"book/{'t' + pair}/{precision}", params={ "len": len }) ] + 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: - return [ serializers.FundingCurrencyBook.parse(*subdata) for subdata in self.__GET(f"book/{'f' + currency}/{precision}", params={ "len": len }) ] + 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: - return [ serializers.TradingPairRawBook.parse(*subdata) for subdata in self.__GET(f"book/{'t' + pair}/R0", params={ "len": len }) ] + 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: - return [ serializers.FundingCurrencyRawBook.parse(*subdata) for subdata in self.__GET(f"book/{'f' + currency}/R0", params={ "len": len }) ] + return [ serializers.FundingCurrencyRawBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/R0", params={ "len": len }) ] def stats_hist( self, @@ -99,7 +104,7 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Stats: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"stats1/{resource}/hist", params=params) + data = self._GET(f"stats1/{resource}/hist", params=params) return [ serializers.Stat.parse(*subdata) for subdata in data ] def stats_last( @@ -108,7 +113,7 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Stat: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"stats1/{resource}/last", params=params) + data = self._GET(f"stats1/{resource}/last", params=params) return serializers.Stat.parse(*data) def candles_hist( @@ -117,7 +122,7 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Candles: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"candles/{resource}/hist", params=params) + data = self._GET(f"candles/{resource}/hist", params=params) return [ serializers.Candle.parse(*subdata) for subdata in data ] def candles_last( @@ -126,13 +131,13 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Candle: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"candles/{resource}/last", params=params) + data = self._GET(f"candles/{resource}/last", params=params) return serializers.Candle.parse(*data) def derivatives_status(self, type: str, keys: List[str]) -> DerivativeStatuses: params = { "keys": ",".join(keys) } - data = self.__GET(f"status/{type}", params=params) + data = self._GET(f"status/{type}", params=params) return [ serializers.DerivativesStatus.parse(*subdata) for subdata in data ] @@ -143,14 +148,14 @@ class BfxRestInterface(object): ) -> DerivativeStatuses: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"status/{type}/{symbol}/hist", params=params) + 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: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET("liquidations/hist", params=params) + data = self._GET("liquidations/hist", params=params) return [ serializers.Liquidation.parse(*subdata[0]) for subdata in data ] @@ -160,7 +165,7 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Leaderboards: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"rankings/{resource}/hist", params=params) + data = self._GET(f"rankings/{resource}/hist", params=params) return [ serializers.Leaderboard.parse(*subdata) for subdata in data ] def leaderboards_last( @@ -169,15 +174,15 @@ class BfxRestInterface(object): sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None ) -> Leaderboard: params = { "sort": sort, "start": start, "end": end, "limit": limit } - data = self.__GET(f"rankings/{resource}/last", params=params) + 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: params = { "start": start, "end": end, "limit": limit } - data = self.__GET(f"funding/stats/{symbol}/hist", params=params) + data = self._GET(f"funding/stats/{symbol}/hist", params=params) return [ serializers.FundingStat.parse(*subdata) for subdata in data ] def conf(self, config: Config) -> Any: - return self.__GET(f"conf/{config}")[0] \ No newline at end of file + return self._GET(f"conf/{config}")[0] \ No newline at end of file