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

Merge branch 'fix/refactoring' of https://github.com/Davi0kProgramsThings/bitfinex-api-py into add-rest-examples

Browse Source
This commit is contained in:
itsdeka
2023-01-17 12:08:09 +01:00
parent b48a882f5e 6f8e37cad0
commit ace074edda
9 changed files with 205 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
bfxapi/rest/BfxRestInterface.py
View File

@@ -8,7 +8,7 @@ from typing import List, Union, Literal, Optional, Any, cast
from . import serializers
from .typings import *
from .types import *
from .enums import Config, Sort, OrderType, FundingOfferType, Error
from .exceptions import ResourceNotFound, RequestParametersError, InvalidAuthenticationCredentials, UnknownGenericError
@@ -97,11 +97,11 @@ class _RestPublicEndpoints(_Requests):
parsers = { "t": serializers.TradingPairTicker.parse, "f": serializers.FundingCurrencyTicker.parse }
return [ parsers[subdata[0][0]](*subdata) for subdata in data ]
return [ cast(Union[TradingPairTicker, FundingCurrencyTicker], parsers[subdata[0][0]](*subdata)) for subdata in data ]
def get_t_tickers(self, pairs: Union[List[str], Literal["ALL"]]) -> List[TradingPairTicker]:
if isinstance(pairs, str) and pairs == "ALL":
return [ cast(TradingPairTicker, subdata) for subdata in self.get_tickers([ "ALL" ]) if cast(str, subdata["SYMBOL"]).startswith("t") ]
return [ cast(TradingPairTicker, subdata) for subdata in self.get_tickers([ "ALL" ]) if cast(str, subdata.SYMBOL).startswith("t") ]
data = self.get_tickers([ "t" + pair for pair in pairs ])
@@ -109,7 +109,7 @@ class _RestPublicEndpoints(_Requests):
def get_f_tickers(self, currencies: Union[List[str], Literal["ALL"]]) -> List[FundingCurrencyTicker]:
if isinstance(currencies, str) and currencies == "ALL":
return [ cast(FundingCurrencyTicker, subdata) for subdata in self.get_tickers([ "ALL" ]) if cast(str, subdata["SYMBOL"]).startswith("f") ]
return [ cast(FundingCurrencyTicker, subdata) for subdata in self.get_tickers([ "ALL" ]) if cast(str, subdata.SYMBOL).startswith("f") ]
data = self.get_tickers([ "f" + currency for currency in currencies ])
@@ -262,7 +262,7 @@ class _RestAuthenticatedEndpoints(_Requests):
price: Optional[Union[Decimal, str]] = None, lev: Optional[int] = None,
price_trailing: Optional[Union[Decimal, str]] = None, price_aux_limit: Optional[Union[Decimal, str]] = None, price_oco_stop: Optional[Union[Decimal, str]] = None,
gid: Optional[int] = None, cid: Optional[int] = None,
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification:
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification[Order]:
data = {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev,
@@ -271,12 +271,12 @@ class _RestAuthenticatedEndpoints(_Requests):
"flags": flags, "tif": tif, "meta": meta
}
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/submit", data=data))
return serializers._Notification[Order](serializer=serializers.Order).parse(*self._POST("auth/w/order/submit", data=data))
def update_order(self, id: int, amount: Optional[Union[Decimal, str]] = None, price: Optional[Union[Decimal, str]] = None,
cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = None,
flags: Optional[int] = 0, lev: Optional[int] = None, delta: Optional[Union[Decimal, str]] = None,
price_aux_limit: Optional[Union[Decimal, str]] = None, price_trailing: Optional[Union[Decimal, str]] = None, tif: Optional[Union[datetime, str]] = None) -> Notification:
price_aux_limit: Optional[Union[Decimal, str]] = None, price_trailing: Optional[Union[Decimal, str]] = None, tif: Optional[Union[datetime, str]] = None) -> Notification[Order]:
data = {
"id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid,
@@ -284,18 +284,18 @@ class _RestAuthenticatedEndpoints(_Requests):
"price_aux_limit": price_aux_limit, "price_trailing": price_trailing, "tif": tif
}
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/update", data=data))
return serializers._Notification[Order](serializer=serializers.Order).parse(*self._POST("auth/w/order/update", data=data))
def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None) -> Notification:
def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None) -> Notification[Order]:
data = {
"id": id,
"cid": cid,
"cid_date": cid_date
}
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/cancel", data=data))
return serializers._Notification[Order](serializer=serializers.Order).parse(*self._POST("auth/w/order/cancel", data=data))
def cancel_order_multi(self, ids: Optional[List[int]] = None, cids: Optional[List[Tuple[int, str]]] = None, gids: Optional[List[int]] = None, all: bool = False) -> Notification:
def cancel_order_multi(self, ids: Optional[List[int]] = None, cids: Optional[List[Tuple[int, str]]] = None, gids: Optional[List[int]] = None, all: bool = False) -> Notification[List[Order]]:
data = {
"ids": ids,
"cids": cids,
@@ -304,7 +304,7 @@ class _RestAuthenticatedEndpoints(_Requests):
"all": int(all)
}
return serializers._Notification(serializer=serializers.Order, iterate=True).parse(*self._POST("auth/w/order/cancel/multi", data=data))
return serializers._Notification[List[Order]](serializer=serializers.Order, iterate=True).parse(*self._POST("auth/w/order/cancel/multi", data=data))
def get_orders_history(self, symbol: Optional[str] = None, ids: Optional[List[int]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Order]:
if symbol == None:
@@ -354,17 +354,17 @@ class _RestAuthenticatedEndpoints(_Requests):
def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, str],
rate: Union[Decimal, str], period: int,
flags: Optional[int] = 0) -> Notification:
flags: Optional[int] = 0) -> Notification[FundingOffer]:
data = {
"type": type, "symbol": symbol, "amount": amount,
"rate": rate, "period": period,
"flags": flags
}
return serializers._Notification(serializer=serializers.FundingOffer).parse(*self._POST("auth/w/funding/offer/submit", data=data))
return serializers._Notification[FundingOffer](serializer=serializers.FundingOffer).parse(*self._POST("auth/w/funding/offer/submit", data=data))
def cancel_funding_offer(self, id: int) -> Notification:
return serializers._Notification(serializer=serializers.FundingOffer).parse(*self._POST("auth/w/funding/offer/cancel", data={ "id": id }))
def cancel_funding_offer(self, id: int) -> Notification[FundingOffer]:
return serializers._Notification[FundingOffer](serializer=serializers.FundingOffer).parse(*self._POST("auth/w/funding/offer/cancel", data={ "id": id }))
def get_funding_offers_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingOffer]:
if symbol == None: