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

Fix small bug in BfxRestInterface.py file.

Browse Source
This commit is contained in:
Davide Casale
2022-12-12 15:23:43 +01:00
parent 6a368d139d
commit 376ac37273
1 changed files with 34 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
bfxapi/rest/BfxRestInterface.py
View File

@@ -30,20 +30,24 @@ class BfxRestInterface(object):
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) })
return [
{
"t": serializers.TradingPairTicker.parse,
"f": serializers.FundingCurrencyTicker.parse
}[subdata[0][0]](*subdata)
for subdata in self.__GET("tickers", params={ "symbols": ",".join(symbols) })
for subdata in data
]
def ticker(self, symbol: str) -> Union[TradingPairTicker, FundingCurrencyTicker]:
data = self.__GET(f"ticker/{symbol}")
return {
"t": serializers.TradingPairTicker.parse,
"f": serializers.FundingCurrencyTicker.parse
}[symbol[0]](*self.__GET(f"ticker/{symbol}"), skip=["SYMBOL"])
}[symbol[0]](*data, skip=["SYMBOL"])
def tickers_history(self, symbols: List[str], start: Optional[int] = None, end: Optional[int] = None, limit: Optional[int] = None) -> TickerHistories:
params = {
@@ -52,28 +56,34 @@ class BfxRestInterface(object):
"limit": limit
}
return [ serializers.TickerHistory.parse(*subdata) for subdata in self.__GET("tickers/hist", params=params) ]
data = self.__GET("tickers/hist", params=params)
return [ serializers.TickerHistory.parse(*subdata) for subdata in data ]
def trades(self, symbol: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[int] = None) -> Union[TradingPairTrades, FundingCurrencyTrades]:
params = { "symbol": symbol, "limit": limit, "start": start, "end": end, "sort": sort }
data = self.__GET(f"trades/{symbol}/hist", params=params)
return [
{
"t": serializers.TradingPairTrade.parse,
"f": serializers.FundingCurrencyTrade.parse
}[symbol[0]](*subdata)
for subdata in self.__GET(f"trades/{symbol}/hist", params=params)
for subdata in data
]
def book(self, symbol: str, precision: str, len: Optional[int] = None) -> Union[TradingPairBooks, FundingCurrencyBooks, TradingPairRawBooks, FundingCurrencyRawBooks]:
data = self.__GET(f"book/{symbol}/{precision}", params={ "len": len })
return [
{
"t": precision == "R0" and serializers.TradingPairRawBook.parse or serializers.TradingPairBook.parse,
"f": precision == "R0" and serializers.FundingCurrencyRawBook.parse or serializers.FundingCurrencyBook.parse,
}[symbol[0]](*subdata)
for subdata in self.__GET(f"book/{symbol}/{precision}", params={ "len": len })
for subdata in data
]
def stats(
@@ -81,47 +91,48 @@ class BfxRestInterface(object):
resource: str, section: Literal["hist", "last"],
sort: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
) -> Union[Stat, Stats]:
endpoint = f"stats1/{resource}/{section}"
params = { "sort": sort, "start": start, "end": end, "limit": limit }
data = self.__GET(f"stats1/{resource}/{section}", params=params)
if section == "last":
return serializers.Stat.parse(*self.__GET(endpoint, params=params))
return [ serializers.Stat.parse(*subdata) for subdata in self.__GET(endpoint, params=params) ]
return serializers.Stat.parse(*data)
return [ serializers.Stat.parse(*subdata) for subdata in data ]
def candles(
self,
resource: str, section: Literal["hist", "last"],
sort: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
) -> Union[Candle, Candles]:
endpoint = f"candles/{resource}/{section}"
params = { "sort": sort, "start": start, "end": end, "limit": limit }
data = self.__GET(f"candles/{resource}/{section}", params=params)
if section == "last":
return serializers.Candle.parse(*self.__GET(endpoint, params=params))
return [ serializers.Candle.parse(*subdata) for subdata in self.__GET(endpoint, params=params) ]
return serializers.Candle.parse(*data)
return [ serializers.Candle.parse(*subdata) for subdata in data ]
def derivatives_status(self, type: str, keys: Optional[List[str]] = None) -> DerivativeStatuses:
params = None
if keys != None:
def derivatives_status(self, type: str, keys: List[str] = None) -> DerivativeStatuses:
params = { "keys": ",".join(keys) }
return [ serializers.DerivativesStatus.parse(*subdata) for subdata in self.__GET(f"status/{type}", params=params) ]
data = self.__GET(f"status/{type}", params=params)
return [ serializers.DerivativesStatus.parse(*subdata) for subdata in data ]
def derivatives_status_history(
self,
type: str, symbol: str,
sort: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
) -> DerivativeStatuses:
endpoint = f"status/{type}/{symbol}/hist"
params = { "sort": sort, "start": start, "end": end, "limit": limit }
return [ serializers.DerivativesStatus.parse(*subdata, skip=[ "KEY" ]) for subdata in self.__GET(endpoint, 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[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> Liquidations:
params = { "sort": sort, "start": start, "end": end, "limit": limit }
return [ serializers.Liquidation.parse(*subdata[0]) for subdata in self.__GET("liquidations/hist", params=params) ]
data = self.__GET("liquidations/hist", params=params)
return [ serializers.Liquidation.parse(*subdata[0]) for subdata in data ]