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

Add type hints and type checks in bfxapi.websocket.client.bfx_websocket_inputs.

Browse Source
This commit is contained in:
Davide Casale
2023-05-25 20:40:31 +02:00
parent c8290f144b
commit dbc61ab275
1 changed files with 21 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
bfxapi/websocket/client/bfx_websocket_inputs.py
View File

@@ -1,16 +1,24 @@
from typing import TYPE_CHECKING, Callable, Awaitable, \
Tuple, List, Union, Optional, Any
from decimal import Decimal
from datetime import datetime
from typing import Union, Optional, List, Tuple
from ..enums import OrderType, FundingOfferType
from ...types import JSON
if TYPE_CHECKING:
from bfxapi.enums import \
OrderType, FundingOfferType
from bfxapi.types import JSON
_Handler = Callable[[str, Any], Awaitable[None]]
class BfxWebSocketInputs:
def __init__(self, handle_websocket_input):
def __init__(self, handle_websocket_input: "_Handler") -> None:
self.__handle_websocket_input = handle_websocket_input
async def submit_order(self,
type: OrderType,
type: "OrderType",
symbol: str,
amount: Union[Decimal, float, str],
*,
@@ -23,7 +31,7 @@ class BfxWebSocketInputs:
cid: Optional[int] = None,
flags: Optional[int] = 0,
tif: Optional[Union[datetime, str]] = None,
meta: Optional[JSON] = None):
meta: Optional["JSON"] = None) -> None:
await self.__handle_websocket_input("on", {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev, "price_trailing": price_trailing,
@@ -45,7 +53,7 @@ class BfxWebSocketInputs:
delta: Optional[Union[Decimal, float, 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):
tif: Optional[Union[datetime, str]] = None) -> None:
await self.__handle_websocket_input("ou", {
"id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid,
@@ -57,7 +65,7 @@ class BfxWebSocketInputs:
*,
id: Optional[int] = None,
cid: Optional[int] = None,
cid_date: Optional[str] = None):
cid_date: Optional[str] = None) -> None:
await self.__handle_websocket_input("oc", {
"id": id, "cid": cid, "cid_date": cid_date
})
@@ -67,7 +75,7 @@ class BfxWebSocketInputs:
ids: Optional[List[int]] = None,
cids: Optional[List[Tuple[int, str]]] = None,
gids: Optional[List[int]] = None,
all: bool = False):
all: bool = False) -> None:
await self.__handle_websocket_input("oc_multi", {
"ids": ids, "cids": cids, "gids": gids,
"all": int(all)
@@ -75,20 +83,20 @@ class BfxWebSocketInputs:
#pylint: disable-next=too-many-arguments
async def submit_funding_offer(self,
type: FundingOfferType,
type: "FundingOfferType",
symbol: str,
amount: Union[Decimal, float, str],
rate: Union[Decimal, float, str],
period: int,
*,
flags: Optional[int] = 0):
flags: Optional[int] = 0) -> None:
await self.__handle_websocket_input("fon", {
"type": type, "symbol": symbol, "amount": amount,
"rate": rate, "period": period, "flags": flags
})
async def cancel_funding_offer(self, id: int):
async def cancel_funding_offer(self, id: int) -> None:
await self.__handle_websocket_input("foc", { "id": id })
async def calc(self, *args: str):
async def calc(self, *args: str) -> None:
await self.__handle_websocket_input("calc", list(map(lambda arg: [arg], args)))