aljaz/bitfinex-api-py
1
0
Fork 0
mirror of https://github.com/aljazceru/bitfinex-api-py.git synced 2025-12-19 06:44:22 +01:00
Code Issues Packages Projects Releases Wiki Activity

Improve some endpoints in RestPublicEndpoints (bfxapi.rest.endpoints.rest_public_endpoints).

Browse Source
This commit is contained in:
Davide Casale
2023-04-20 02:45:33 +02:00
parent 6f207010e8
commit 34a1b0099e
3 changed files with 38 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
bfxapi/rest/endpoints/rest_public_endpoints.py
View File

@@ -1,4 +1,4 @@
from typing import List, Union, Literal, Optional, Any, cast
from typing import List, Dict, Union, Literal, Optional, Any, cast
from decimal import Decimal
@@ -22,37 +22,46 @@ class RestPublicEndpoints(Middleware):
def get_platform_status(self) -> PlatformStatus:
return serializers.PlatformStatus.parse(*self._get("platform/status"))
def get_tickers(self, symbols: List[str]) -> List[Union[TradingPairTicker, FundingCurrencyTicker]]:
def get_tickers(self, symbols: List[str]) -> Dict[str, Union[TradingPairTicker, FundingCurrencyTicker]]:
data = self._get("tickers", params={ "symbols": ",".join(symbols) })
parsers = { "t": serializers.TradingPairTicker.parse, "f": serializers.FundingCurrencyTicker.parse }
return [ cast(Union[TradingPairTicker, FundingCurrencyTicker], \
parsers[sub_data[0][0]](*sub_data)) for sub_data in data ]
return {
symbol: cast(Union[TradingPairTicker, FundingCurrencyTicker],
parsers[symbol[0]](*sub_data)) for sub_data in data
if (symbol := sub_data.pop(0))
}
def get_t_tickers(self, pairs: Union[List[str], Literal["ALL"]]) -> List[TradingPairTicker]:
if isinstance(pairs, str) and pairs == "ALL":
return [ cast(TradingPairTicker, sub_data) for sub_data in self.get_tickers([ "ALL" ]) \
if cast(str, sub_data.symbol).startswith("t") ]
def get_t_tickers(self, symbols: Union[List[str], Literal["ALL"]]) -> Dict[str, TradingPairTicker]:
if isinstance(symbols, str) and symbols == "ALL":
return {
symbol: cast(TradingPairTicker, sub_data)
for symbol, sub_data in self.get_tickers([ "ALL" ]).items()
if symbol.startswith("t")
}
data = self.get_tickers(list(pairs))
data = self.get_tickers(list(symbols))
return cast(List[TradingPairTicker], data)
return cast(Dict[str, TradingPairTicker], data)
def get_f_tickers(self, currencies: Union[List[str], Literal["ALL"]]) -> List[FundingCurrencyTicker]:
if isinstance(currencies, str) and currencies == "ALL":
return [ cast(FundingCurrencyTicker, sub_data) for sub_data in self.get_tickers([ "ALL" ]) \
if cast(str, sub_data.symbol).startswith("f") ]
def get_f_tickers(self, symbols: Union[List[str], Literal["ALL"]]) -> Dict[str, FundingCurrencyTicker]:
if isinstance(symbols, str) and symbols == "ALL":
return {
symbol: cast(FundingCurrencyTicker, sub_data)
for symbol, sub_data in self.get_tickers([ "ALL" ]).items()
if symbol.startswith("f")
}
data = self.get_tickers(list(currencies))
data = self.get_tickers(list(symbols))
return cast(List[FundingCurrencyTicker], data)
return cast(Dict[str, FundingCurrencyTicker], data)
def get_t_ticker(self, pair: str) -> TradingPairTicker:
return serializers.TradingPairTicker.parse(*([pair] + self._get(f"ticker/{pair}")))
def get_t_ticker(self, symbol: str) -> TradingPairTicker:
return serializers.TradingPairTicker.parse(*self._get(f"ticker/{symbol}"))
def get_f_ticker(self, currency: str) -> FundingCurrencyTicker:
return serializers.FundingCurrencyTicker.parse(*([currency] + self._get(f"ticker/{currency}")))
def get_f_ticker(self, symbol: str) -> FundingCurrencyTicker:
return serializers.FundingCurrencyTicker.parse(*self._get(f"ticker/{symbol}"))
def get_tickers_history(self,
symbols: List[str],
@@ -164,26 +173,29 @@ class RestPublicEndpoints(Middleware):
data = self._get(f"candles/trade:{tf}:{symbol}/last", params=params)
return serializers.Candle.parse(*data)
def get_derivatives_status(self, keys: Union[List[str], Literal["ALL"]]) -> List[DerivativesStatus]:
def get_derivatives_status(self, keys: Union[List[str], Literal["ALL"]]) -> Dict[str, DerivativesStatus]:
if keys == "ALL":
params = { "keys": "ALL" }
else: params = { "keys": ",".join(keys) }
data = self._get("status/deriv", params=params)
return [ serializers.DerivativesStatus.parse(*sub_data) for sub_data in data ]
return {
key: serializers.DerivativesStatus.parse(*sub_data)
for sub_data in data
if (key := sub_data.pop(0))
}
def get_derivatives_status_history(self,
type: str,
symbol: str,
key: str,
*,
sort: Optional[Sort] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> 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(*([symbol] + sub_data)) for sub_data in data ]
data = self._get(f"status/deriv/{key}/hist", params=params)
return [ serializers.DerivativesStatus.parse(*sub_data) for sub_data in data ]
def get_liquidations(self,
*,