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 support for new rest authenticated endpoints.

Browse Source
This commit is contained in:
Davide Casale
2022-12-23 16:36:51 +01:00
parent db4438144d
commit 72a3252e32
5 changed files with 118 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
bfxapi/notification.py
View File

@@ -1,4 +1,4 @@
from typing import List, Dict, Optional, Any, TypedDict, cast
from typing import List, Dict, Union, Optional, Any, TypedDict, cast
from .labeler import _Serializer
@@ -6,7 +6,7 @@ class Notification(TypedDict):
MTS: int
TYPE: str
MESSAGE_ID: Optional[int]
NOTIFY_INFO: Dict[str, Any]
NOTIFY_INFO: Union[Dict[str, Any], List[Dict[str, Any]]]
CODE: Optional[int]
STATUS: str
TEXT: str
@@ -14,15 +14,17 @@ class Notification(TypedDict):
class _Notification(_Serializer):
__LABELS = [ "MTS", "TYPE", "MESSAGE_ID", "_PLACEHOLDER", "NOTIFY_INFO", "CODE", "STATUS", "TEXT" ]
def __init__(self, serializer: Optional[_Serializer] = None):
def __init__(self, serializer: Optional[_Serializer] = None, iterate: bool = False):
super().__init__("Notification", _Notification.__LABELS, IGNORE = [ "_PLACEHOLDER" ])
self.serializer = serializer
self.serializer, self.iterate = serializer, iterate
def parse(self, *values: Any, skip: Optional[List[str]] = None) -> Notification:
notification = dict(self._serialize(*values))
if isinstance(self.serializer, _Serializer):
notification["NOTIFY_INFO"] = dict(self.serializer._serialize(*notification["NOTIFY_INFO"], skip=skip))
if self.iterate == False:
notification["NOTIFY_INFO"] = dict(self.serializer._serialize(*notification["NOTIFY_INFO"], skip=skip))
else: notification["NOTIFY_INFO"] = [ dict(self.serializer._serialize(*data, skip=skip)) for data in notification["NOTIFY_INFO"] ]
return cast(Notification, notification)

60
bfxapi/rest/BfxRestInterface.py
View File

@@ -12,7 +12,7 @@ from .typings import *
from .enums import Config, Precision, Sort, OrderType, Error
from .exceptions import ResourceNotFound, RequestParametersError, InvalidAuthenticationCredentials, UnknownGenericError
from .. utils.integers import Int16, Int32, Int45, Int64
from .. utils.integers import Int16, int32, int45, int64
from .. utils.encoder import JSONEncoder
@@ -246,10 +246,10 @@ class _RestAuthenticatedEndpoints(_Requests):
return [ serializers.Order.parse(*subdata) for subdata in self._POST("auth/r/orders", data={ "id": ids }) ]
def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, str],
price: Optional[Union[Decimal, str]] = None, lev: Optional[Union[Int32, int]] = None,
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[Union[Int32, int]] = None, cid: Optional[Union[Int45, int]] = None,
flags: Optional[Union[Int16, int]] = None, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification:
gid: Optional[int] = None, cid: Optional[int] = None,
flags: Optional[int] = None, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification:
data = {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev,
@@ -260,9 +260,9 @@ class _RestAuthenticatedEndpoints(_Requests):
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/submit", data=data))
def update_order(self, id: Union[Int64, int], amount: Optional[Union[Decimal, str]] = None, price: Optional[Union[Decimal, str]] = None,
cid: Optional[Union[Int45, int]] = None, cid_date: Optional[str] = None, gid: Optional[Union[Int32, int]] = None,
flags: Optional[Union[Int16, int]] = None, lev: Optional[Union[Int32, int]] = None, delta: Optional[Union[Decimal, str]] = None,
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] = None, 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:
data = {
"id": id, "amount": amount, "price": price,
@@ -273,5 +273,47 @@ class _RestAuthenticatedEndpoints(_Requests):
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/update", data=data))
def cancel_order(self, id: Union[Int64, int]) -> Notification:
return serializers._Notification(serializer=serializers.Order).parse(*self._POST("auth/w/order/cancel", data={ "id": id }))
def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None) -> Notification:
data = {
"id": id,
"cid": cid,
"cid_date": cid_date
}
return serializers._Notification(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:
data = {
"ids": ids,
"cids": cids,
"gids": gids,
"all": int(all)
}
return serializers._Notification(serializer=serializers.Order, iterate=True).parse(*self._POST("auth/w/order/cancel/multi", data=data))
def 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:
endpoint = "auth/r/orders/hist"
else: endpoint = f"auth/r/orders/{symbol}/hist"
data = {
"id": ids,
"start": start, "end": end,
"limit": limit
}
return [ serializers.Order.parse(*subdata) for subdata in self._POST(endpoint, data=data) ]
def trades(self, symbol: str) -> List[Trade]:
return [ serializers.Trade.parse(*subdata) for subdata in self._POST(f"auth/r/trades/{symbol}/hist") ]
def ledgers(self, currency: str, category: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Ledger]:
data = {
"category": category,
"start": start, "end": end,
"limit": limit
}
return [ serializers.Ledger.parse(*subdata) for subdata in self._POST(f"auth/r/ledgers/{currency}/hist", data=data) ]

27
bfxapi/rest/serializers.py
View File

@@ -234,4 +234,31 @@ Order = _Serializer[typings.Order]("Order", labels=[
"META"
])
Trade = _Serializer[typings.Trade]("Trade", labels=[
"ID",
"PAIR",
"MTS_CREATE",
"ORDER_ID",
"EXEC_AMOUNT",
"EXEC_PRICE",
"ORDER_TYPE",
"ORDER_PRICE",
"MAKER",
"FEE",
"FEE_CURRENCY",
"CID"
])
Ledger = _Serializer[typings.Ledger]("Ledger", labels=[
"ID",
"CURRENCY",
"_PLACEHOLDER",
"MTS",
"_PLACEHOLDER",
"AMOUNT",
"BALANCE",
"_PLACEHOLDER",
"DESCRIPTION"
])
#endregion

22
bfxapi/rest/typings.py
View File

@@ -169,4 +169,26 @@ class Order(TypedDict):
ROUTING: str
META: JSON
class Trade(TypedDict):
ID: int
SYMBOL: str
MTS_CREATE: int
ORDER_ID: int
EXEC_AMOUNT: float
EXEC_PRICE: float
ORDER_TYPE: str
ORDER_PRICE: float
MAKER:int
FEE: float
FEE_CURRENCY: str
CID: int
class Ledger(TypedDict):
ID: int
CURRENCY: str
MTS: int
AMOUNT: float
BALANCE: float
description: str
#endregion

12
bfxapi/utils/integers.py
View File

@@ -1,4 +1,4 @@
from typing import cast, TypeVar
from typing import cast, TypeVar, Union
from .. exceptions import IntegerUnderflowError, IntegerOverflowflowError
@@ -25,11 +25,19 @@ class _Int(int):
class Int16(_Int):
_BITS = 16
int16 = Union[Int16, int]
class Int32(_Int):
_BITS = 32
int32 = Union[Int32, int]
class Int45(_Int):
_BITS = 45
int45 = Union[Int45, int]
class Int64(_Int):
_BITS = 64
_BITS = 64
int64 = Union[Int64, int]