aljaz/bitfinex-api-py
1
0
Fork 0
mirror of https://github.com/aljazceru/bitfinex-api-py.git synced 2025-12-20 15:24:21 +01:00
Code Issues Packages Projects Releases Wiki Activity

Add _Requests and _RestPublicEndpoints classes in bfxapi/rest/BfxRestInterface.py.

Browse Source
This commit is contained in:
Davide Casale
2022-12-14 18:17:29 +01:00
parent ec821a0752
commit 07241b1ba8
1 changed files with 28 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
bfxapi/rest/BfxRestInterface.py
View File

@@ -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]
return self._GET(f"conf/{config}")[0]