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

Rewrite and extend custom JSONEncoder in bfxapi/utils/encoder.py to automatically convert floats to strs. Change every Union[Decimal, str] type to Union[Decimal, float, str]. Fix type hinting bug in labeler.py.

Browse Source
This commit is contained in:
Davide Casale
2023-01-25 18:18:15 +01:00
parent 3a6a863796
commit ed12bf473f
6 changed files with 49 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
bfxapi/labeler.py
View File

@@ -29,7 +29,7 @@ class _Serializer(Generic[T]):
return cast(T, self.klass(**dict(self._serialize(*values, skip=skip)))) return cast(T, self.klass(**dict(self._serialize(*values, skip=skip))))
class _RecursiveSerializer(_Serializer, Generic[T]): class _RecursiveSerializer(_Serializer, Generic[T]):
def __init__(self, name: str, klass: Type[_Type], labels: List[str], serializers: Dict[str, Type[_Serializer]], IGNORE: List[str] = ["_PLACEHOLDER"]): def __init__(self, name: str, klass: Type[_Type], labels: List[str], serializers: Dict[str, _Serializer[Any]], IGNORE: List[str] = ["_PLACEHOLDER"]):
super().__init__(name, klass, labels, IGNORE) super().__init__(name, klass, labels, IGNORE)
self.serializers = serializers self.serializers = serializers
@@ -46,5 +46,5 @@ class _RecursiveSerializer(_Serializer, Generic[T]):
def generate_labeler_serializer(name: str, klass: Type[T], labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _Serializer[T]: def generate_labeler_serializer(name: str, klass: Type[T], labels: List[str], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _Serializer[T]:
return _Serializer[T](name, klass, labels, IGNORE) return _Serializer[T](name, klass, labels, IGNORE)
def generate_recursive_serializer(name: str, klass: Type[T], labels: List[str], serializers: Dict[str, Type[_Serializer]], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _RecursiveSerializer[T]: def generate_recursive_serializer(name: str, klass: Type[T], labels: List[str], serializers: Dict[str, _Serializer[Any]], IGNORE: List[str] = [ "_PLACEHOLDER" ]) -> _RecursiveSerializer[T]:
return _RecursiveSerializer[T](name, klass, labels, serializers, IGNORE) return _RecursiveSerializer[T](name, klass, labels, serializers, IGNORE)

26
bfxapi/rest/BfxRestInterface.py
View File

@@ -268,14 +268,14 @@ class _RestPublicEndpoints(_Requests):
return messages return messages
def get_trading_market_average_price(self, symbol: str, amount: Union[Decimal, str], price_limit: Optional[Union[Decimal, str]] = None) -> TradingMarketAveragePrice: def get_trading_market_average_price(self, symbol: str, amount: Union[Decimal, float, str], price_limit: Optional[Union[Decimal, float, str]] = None) -> TradingMarketAveragePrice:
data = { data = {
"symbol": symbol, "amount": amount, "price_limit": price_limit "symbol": symbol, "amount": amount, "price_limit": price_limit
} }
return serializers.TradingMarketAveragePrice.parse(*self._POST("calc/trade/avg", data=data)) return serializers.TradingMarketAveragePrice.parse(*self._POST("calc/trade/avg", data=data))
def get_funding_market_average_price(self, symbol: str, amount: Union[Decimal, str], period: int, rate_limit: Optional[Union[Decimal, str]] = None) -> FundingMarketAveragePrice: def get_funding_market_average_price(self, symbol: str, amount: Union[Decimal, float, str], period: int, rate_limit: Optional[Union[Decimal, float, str]] = None) -> FundingMarketAveragePrice:
data = { data = {
"symbol": symbol, "amount": amount, "period": period, "symbol": symbol, "amount": amount, "period": period,
"rate_limit": rate_limit "rate_limit": rate_limit
@@ -301,9 +301,9 @@ class _RestAuthenticatedEndpoints(_Requests):
def get_positions(self) -> List[Position]: def get_positions(self) -> List[Position]:
return [ serializers.Position.parse(*sub_data) for sub_data in self._POST("auth/r/positions") ] return [ serializers.Position.parse(*sub_data) for sub_data in self._POST("auth/r/positions") ]
def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, str], def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str],
price: Optional[Union[Decimal, str]] = None, lev: Optional[int] = None, price: Optional[Union[Decimal, float, 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, price_trailing: Optional[Union[Decimal, float, str]] = None, price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_oco_stop: Optional[Union[Decimal, float, str]] = None,
gid: Optional[int] = None, cid: Optional[int] = None, gid: Optional[int] = None, cid: Optional[int] = None,
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification[Order]: flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification[Order]:
data = { data = {
@@ -316,10 +316,10 @@ class _RestAuthenticatedEndpoints(_Requests):
return serializers._Notification[Order](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, def update_order(self, id: int, amount: Optional[Union[Decimal, float, str]] = None, price: Optional[Union[Decimal, float, str]] = None,
cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = 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, flags: Optional[int] = 0, lev: Optional[int] = None, delta: Optional[Union[Decimal, float, str]] = None,
price_aux_limit: Optional[Union[Decimal, str]] = None, price_trailing: Optional[Union[Decimal, str]] = None, tif: Optional[Union[datetime, str]] = None) -> Notification[Order]: price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_trailing: Optional[Union[Decimal, float, str]] = None, tif: Optional[Union[datetime, str]] = None) -> Notification[Order]:
data = { data = {
"id": id, "amount": amount, "price": price, "id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid, "cid": cid, "cid_date": cid_date, "gid": gid,
@@ -395,8 +395,8 @@ class _RestAuthenticatedEndpoints(_Requests):
return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._POST(endpoint) ] return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._POST(endpoint) ]
def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, str], def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, float, str],
rate: Union[Decimal, str], period: int, rate: Union[Decimal, float, str], period: int,
flags: Optional[int] = 0) -> Notification[FundingOffer]: flags: Optional[int] = 0) -> Notification[FundingOffer]:
data = { data = {
"type": type, "symbol": symbol, "amount": amount, "type": type, "symbol": symbol, "amount": amount,
@@ -440,7 +440,7 @@ class _RestAuthenticatedEndpoints(_Requests):
return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._POST(endpoint, data=data) ] return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._POST(endpoint, data=data) ]
def submit_wallet_transfer(self, from_wallet: str, to_wallet: str, currency: str, currency_to: str, amount: Union[Decimal, str]) -> Notification[Transfer]: def submit_wallet_transfer(self, from_wallet: str, to_wallet: str, currency: str, currency_to: str, amount: Union[Decimal, float, str]) -> Notification[Transfer]:
data = { data = {
"from": from_wallet, "to": to_wallet, "from": from_wallet, "to": to_wallet,
"currency": currency, "currency_to": currency_to, "currency": currency, "currency_to": currency_to,
@@ -449,7 +449,7 @@ class _RestAuthenticatedEndpoints(_Requests):
return serializers._Notification[Transfer](serializer=serializers.Transfer).parse(*self._POST("auth/w/transfer", data=data)) return serializers._Notification[Transfer](serializer=serializers.Transfer).parse(*self._POST("auth/w/transfer", data=data))
def submit_wallet_withdraw(self, wallet: str, method: str, address: str, amount: Union[Decimal, str]) -> Notification[Withdrawal]: def submit_wallet_withdraw(self, wallet: str, method: str, address: str, amount: Union[Decimal, float, str]) -> Notification[Withdrawal]:
data = { data = {
"wallet": wallet, "method": method, "wallet": wallet, "method": method,
"address": address, "amount": amount, "address": address, "amount": amount,
@@ -466,7 +466,7 @@ class _RestAuthenticatedEndpoints(_Requests):
return serializers._Notification[DepositAddress](serializer=serializers.DepositAddress).parse(*self._POST("auth/w/deposit/address", data=data)) return serializers._Notification[DepositAddress](serializer=serializers.DepositAddress).parse(*self._POST("auth/w/deposit/address", data=data))
def get_deposit_invoice(self, wallet: str, currency: str, amount: Union[Decimal, str]) -> Invoice: def get_deposit_invoice(self, wallet: str, currency: str, amount: Union[Decimal, float, str]) -> Invoice:
data = { data = {
"wallet": wallet, "currency": currency, "wallet": wallet, "currency": currency,
"amount": amount "amount": amount

4
bfxapi/rest/types.py
View File

@@ -3,10 +3,8 @@ from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Any
from dataclasses import dataclass from dataclasses import dataclass
from .. labeler import _Type from .. labeler import _Type
from .. notification import Notification from .. notification import Notification
from .. utils.encoder import JSON
JSON = Union[Dict[str, "JSON"], List["JSON"], bool, int, float, str, Type[None]]
#region Type hinting for Rest Public Endpoints #region Type hinting for Rest Public Endpoints

26
bfxapi/utils/encoder.py
View File

@@ -2,8 +2,30 @@ import json
from decimal import Decimal from decimal import Decimal
from datetime import datetime from datetime import datetime
from typing import Type, List, Dict, Union, Any
JSON = Union[Dict[str, "JSON"], List["JSON"], bool, int, float, str, Type[None]]
class JSONEncoder(json.JSONEncoder): class JSONEncoder(json.JSONEncoder):
def default(self, obj): def encode(self, obj: JSON) -> str:
if isinstance(obj, Decimal) or isinstance(obj, datetime): def _convert_float_to_str(data: JSON) -> JSON:
if isinstance(data, float):
return format(Decimal(repr(data)), "f")
elif isinstance(data, list):
return [ _convert_float_to_str(sub_data) for sub_data in data ]
elif isinstance(data, dict):
return { key: _convert_float_to_str(value) for key, value in data.items() }
else: return data
data = _convert_float_to_str(obj)
return json.JSONEncoder.encode(self, data)
def default(self, obj: Any) -> Any:
if isinstance(obj, Decimal):
return format(obj, "f")
if isinstance(obj, datetime):
return str(obj) return str(obj)
return json.JSONEncoder.default(self, obj) return json.JSONEncoder.default(self, obj)

16
bfxapi/websocket/_BfxWebsocketInputs.py
View File

@@ -12,9 +12,9 @@ class _BfxWebsocketInputs(object):
def __init__(self, __handle_websocket_input): def __init__(self, __handle_websocket_input):
self.__handle_websocket_input = __handle_websocket_input self.__handle_websocket_input = __handle_websocket_input
async def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, str], async def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str],
price: Optional[Union[Decimal, str]] = None, lev: Optional[int] = None, price: Optional[Union[Decimal, float, 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, price_trailing: Optional[Union[Decimal, float, str]] = None, price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_oco_stop: Optional[Union[Decimal, float, str]] = None,
gid: Optional[int] = None, cid: Optional[int] = None, gid: Optional[int] = None, cid: Optional[int] = None,
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None): flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None):
data = _strip({ data = _strip({
@@ -27,10 +27,10 @@ class _BfxWebsocketInputs(object):
await self.__handle_websocket_input("on", data) await self.__handle_websocket_input("on", data)
async def update_order(self, id: int, amount: Optional[Union[Decimal, str]] = None, price: Optional[Union[Decimal, str]] = None, async def update_order(self, id: int, amount: Optional[Union[Decimal, float, str]] = None, price: Optional[Union[Decimal, float, str]] = None,
cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = 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, flags: Optional[int] = 0, lev: Optional[int] = None, delta: Optional[Union[Decimal, float, str]] = None,
price_aux_limit: Optional[Union[Decimal, str]] = None, price_trailing: Optional[Union[Decimal, str]] = None, tif: Optional[Union[datetime, str]] = None): price_aux_limit: Optional[Union[Decimal, float, str]] = None, price_trailing: Optional[Union[Decimal, float, str]] = None, tif: Optional[Union[datetime, str]] = None):
data = _strip({ data = _strip({
"id": id, "amount": amount, "price": price, "id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid, "cid": cid, "cid_date": cid_date, "gid": gid,
@@ -60,8 +60,8 @@ class _BfxWebsocketInputs(object):
await self.__handle_websocket_input("oc_multi", data) await self.__handle_websocket_input("oc_multi", data)
async def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, str], async def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, float, str],
rate: Union[Decimal, str], period: int, rate: Union[Decimal, float, str], period: int,
flags: Optional[int] = 0): flags: Optional[int] = 0):
data = { data = {
"type": type, "symbol": symbol, "amount": amount, "type": type, "symbol": symbol, "amount": amount,

4
bfxapi/websocket/types.py
View File

@@ -3,10 +3,8 @@ from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Any
from dataclasses import dataclass from dataclasses import dataclass
from ..labeler import _Type from ..labeler import _Type
from ..notification import Notification from ..notification import Notification
from .. utils.encoder import JSON
JSON = Union[Dict[str, "JSON"], List["JSON"], bool, int, float, str, Type[None]]
#region Type hinting for Websocket Public Channels #region Type hinting for Websocket Public Channels