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

Apply pylint's linting rules to bfxapi/rest/endpoints/rest_authenticated_endpoints.py.

Browse Source
This commit is contained in:
Davide Casale
2023-03-07 18:02:26 +01:00
parent 9e1b336a67
commit 06482ea3d3
1 changed files with 287 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

379
bfxapi/rest/endpoints/rest_authenticated_endpoints.py
View File

@@ -1,42 +1,80 @@
from typing import List, Tuple, Union, Literal, Optional #pylint: disable=redefined-builtin
from typing import Dict, List, Tuple, Union, Literal, Optional
from decimal import Decimal from decimal import Decimal
from datetime import datetime from datetime import datetime
from .. types import * from .. types import Notification, \
UserInfo, LoginHistory, BalanceAvailable, \
Order, Position, Trade, \
FundingTrade, OrderTrade, Ledger, \
FundingOffer, FundingCredit, FundingLoan, \
FundingAutoRenew, FundingInfo, Wallet, \
Transfer, Withdrawal, DepositAddress, \
LightningNetworkInvoice, Movement, SymbolMarginInfo, \
BaseMarginInfo, PositionClaim, PositionIncreaseInfo, \
PositionIncrease, PositionHistory, PositionSnapshot, \
PositionAudit, DerivativePositionCollateral, DerivativePositionCollateralLimits
from .. import serializers from .. import serializers
from .. serializers import _Notification
from .. enums import Sort, OrderType, FundingOfferType from .. enums import Sort, OrderType, FundingOfferType
from .. middleware import Middleware from .. middleware import Middleware
from ...utils.json_encoder import JSON
class RestAuthenticatedEndpoints(Middleware): class RestAuthenticatedEndpoints(Middleware):
def get_user_info(self) -> UserInfo: def get_user_info(self) -> UserInfo:
return serializers.UserInfo.parse(*self._post(f"auth/r/info/user")) return serializers.UserInfo.parse(*self._post("auth/r/info/user"))
def get_login_history(self) -> List[LoginHistory]: def get_login_history(self) -> List[LoginHistory]:
return [ serializers.LoginHistory.parse(*sub_data) for sub_data in self._post("auth/r/logins/hist") ] return [ serializers.LoginHistory.parse(*sub_data) \
for sub_data in self._post("auth/r/logins/hist") ]
def get_balance_available_for_orders_or_offers(self, symbol: str, type: str, dir: Optional[int] = None, rate: Optional[str] = None, lev: Optional[str] = None) -> BalanceAvailable: def get_balance_available_for_orders_or_offers(self,
symbol: str,
type: str,
*,
dir: Optional[int] = None,
rate: Optional[str] = None,
lev: Optional[str] = None) -> BalanceAvailable:
return serializers.BalanceAvailable.parse(*self._post("auth/calc/order/avail", body={ return serializers.BalanceAvailable.parse(*self._post("auth/calc/order/avail", body={
"symbol": symbol, "type": type, "dir": dir, "symbol": symbol, "type": type, "dir": dir,
"rate": rate, "lev": lev "rate": rate, "lev": lev
})) }))
def get_wallets(self) -> List[Wallet]: def get_wallets(self) -> List[Wallet]:
return [ serializers.Wallet.parse(*sub_data) for sub_data in self._post("auth/r/wallets") ] return [ serializers.Wallet.parse(*sub_data) \
for sub_data in self._post("auth/r/wallets") ]
def get_orders(self, symbol: Optional[str] = None, ids: Optional[List[str]] = None) -> List[Order]: def get_orders(self,
*,
symbol: Optional[str] = None,
ids: Optional[List[str]] = None) -> List[Order]:
endpoint = "auth/r/orders" endpoint = "auth/r/orders"
if symbol != None: if symbol is not None:
endpoint += f"/{symbol}" endpoint += f"/{symbol}"
return [ serializers.Order.parse(*sub_data) for sub_data in self._post(endpoint, body={ "id": ids }) ] return [ serializers.Order.parse(*sub_data) \
for sub_data in self._post(endpoint, body={ "id": ids }) ]
def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str], def submit_order(self,
price: Optional[Union[Decimal, float, str]] = None, lev: Optional[int] = None, type: OrderType,
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, symbol: str,
gid: Optional[int] = None, cid: Optional[int] = None, amount: Union[Decimal, float, str],
flags: Optional[int] = 0, tif: Optional[Union[datetime, str]] = None, meta: Optional[JSON] = None) -> Notification[Order]: *,
price: Optional[Union[Decimal, float, str]] = None,
lev: Optional[int] = 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,
flags: Optional[int] = 0,
tif: Optional[Union[datetime, str]] = None,
meta: Optional[JSON] = None) -> Notification[Order]:
body = { body = {
"type": type, "symbol": symbol, "amount": amount, "type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev, "price": price, "lev": lev,
@@ -45,12 +83,22 @@ class RestAuthenticatedEndpoints(Middleware):
"flags": flags, "tif": tif, "meta": meta "flags": flags, "tif": tif, "meta": meta
} }
return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/submit", body=body)) return _Notification[Order](serializers.Order).parse(*self._post("auth/w/order/submit", body=body))
def update_order(self, id: int, amount: Optional[Union[Decimal, float, str]] = None, price: Optional[Union[Decimal, float, str]] = None, def update_order(self,
cid: Optional[int] = None, cid_date: Optional[str] = None, gid: Optional[int] = None, id: int,
flags: Optional[int] = 0, lev: Optional[int] = None, 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) -> Notification[Order]: 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,
flags: Optional[int] = 0,
lev: Optional[int] = None,
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) -> Notification[Order]:
body = { body = {
"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,
@@ -58,18 +106,29 @@ class RestAuthenticatedEndpoints(Middleware):
"price_aux_limit": price_aux_limit, "price_trailing": price_trailing, "tif": tif "price_aux_limit": price_aux_limit, "price_trailing": price_trailing, "tif": tif
} }
return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/update", body=body)) return _Notification[Order](serializers.Order) \
.parse(*self._post("auth/w/order/update", body=body))
def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None) -> Notification[Order]: def cancel_order(self,
*,
id: Optional[int] = None,
cid: Optional[int] = None,
cid_date: Optional[str] = None) -> Notification[Order]:
body = { body = {
"id": id, "id": id,
"cid": cid, "cid": cid,
"cid_date": cid_date "cid_date": cid_date
} }
return serializers._Notification[Order](serializers.Order).parse(*self._post("auth/w/order/cancel", body=body)) return _Notification[Order](serializers.Order) \
.parse(*self._post("auth/w/order/cancel", body=body))
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]]: 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]]:
body = { body = {
"ids": ids, "ids": ids,
"cids": cids, "cids": cids,
@@ -78,10 +137,17 @@ class RestAuthenticatedEndpoints(Middleware):
"all": int(all) "all": int(all)
} }
return serializers._Notification[List[Order]](serializers.Order, is_iterable=True).parse(*self._post("auth/w/order/cancel/multi", body=body)) return _Notification[List[Order]](serializers.Order, is_iterable=True) \
.parse(*self._post("auth/w/order/cancel/multi", body=body))
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]: def get_orders_history(self,
if symbol == None: *,
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 is None:
endpoint = "auth/r/orders/hist" endpoint = "auth/r/orders/hist"
else: endpoint = f"auth/r/orders/{symbol}/hist" else: endpoint = f"auth/r/orders/{symbol}/hist"
@@ -93,11 +159,20 @@ class RestAuthenticatedEndpoints(Middleware):
return [ serializers.Order.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] return [ serializers.Order.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]
def get_order_trades(self, symbol: str, id: int) -> List[OrderTrade]: def get_order_trades(self,
return [ serializers.OrderTrade.parse(*sub_data) for sub_data in self._post(f"auth/r/order/{symbol}:{id}/trades") ] symbol: str,
id: int) -> List[OrderTrade]:
return [ serializers.OrderTrade.parse(*sub_data) \
for sub_data in self._post(f"auth/r/order/{symbol}:{id}/trades") ]
def get_trades_history(self, symbol: Optional[str] = None, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Trade]: def get_trades_history(self,
if symbol == None: *,
symbol: Optional[str] = None,
sort: Optional[Sort] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[Trade]:
if symbol is None:
endpoint = "auth/r/trades/hist" endpoint = "auth/r/trades/hist"
else: endpoint = f"auth/r/trades/{symbol}/hist" else: endpoint = f"auth/r/trades/{symbol}/hist"
@@ -109,17 +184,24 @@ class RestAuthenticatedEndpoints(Middleware):
return [ serializers.Trade.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] return [ serializers.Trade.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]
def get_ledgers(self, currency: str, category: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Ledger]: def get_ledgers(self,
currency: str,
*,
category: Optional[int] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[Ledger]:
body = { body = {
"category": category, "category": category,
"start": start, "end": end, "start": start, "end": end,
"limit": limit "limit": limit
} }
return [ serializers.Ledger.parse(*sub_data) for sub_data in self._post(f"auth/r/ledgers/{currency}/hist", body=body) ] return [ serializers.Ledger.parse(*sub_data) \
for sub_data in self._post(f"auth/r/ledgers/{currency}/hist", body=body) ]
def get_base_margin_info(self) -> BaseMarginInfo: def get_base_margin_info(self) -> BaseMarginInfo:
return serializers.BaseMarginInfo.parse(*(self._post(f"auth/r/info/margin/base")[1])) return serializers.BaseMarginInfo.parse(*(self._post("auth/r/info/margin/base")[1]))
def get_symbol_margin_info(self, symbol: str) -> SymbolMarginInfo: def get_symbol_margin_info(self, symbol: str) -> SymbolMarginInfo:
response = self._post(f"auth/r/info/margin/{symbol}") response = self._post(f"auth/r/info/margin/{symbol}")
@@ -127,51 +209,88 @@ class RestAuthenticatedEndpoints(Middleware):
return serializers.SymbolMarginInfo.parse(*data) return serializers.SymbolMarginInfo.parse(*data)
def get_all_symbols_margin_info(self) -> List[SymbolMarginInfo]: def get_all_symbols_margin_info(self) -> List[SymbolMarginInfo]:
return [ serializers.SymbolMarginInfo.parse(*([sub_data[1]] + sub_data[2])) for sub_data in self._post(f"auth/r/info/margin/sym_all") ] return [ serializers.SymbolMarginInfo.parse(*([sub_data[1]] + sub_data[2])) \
for sub_data in self._post("auth/r/info/margin/sym_all") ]
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 claim_position(self, id: int, amount: Optional[Union[Decimal, float, str]] = None) -> Notification[PositionClaim]: def claim_position(self,
return serializers._Notification[PositionClaim](serializers.PositionClaim).parse( id: int,
*self._post("auth/w/position/claim", body={ "id": id, "amount": amount }) *,
) amount: Optional[Union[Decimal, float, str]] = None) -> Notification[PositionClaim]:
return _Notification[PositionClaim](serializers.PositionClaim) \
.parse(*self._post("auth/w/position/claim", body={ "id": id, "amount": amount }))
def increase_position(self, symbol: str, amount: Union[Decimal, float, str]) -> Notification[PositionIncrease]: def increase_position(self,
return serializers._Notification[PositionIncrease](serializers.PositionIncrease).parse( symbol: str,
*self._post("auth/w/position/increase", body={ "symbol": symbol, "amount": amount }) amount: Union[Decimal, float, str]) -> Notification[PositionIncrease]:
) return _Notification[PositionIncrease](serializers.PositionIncrease) \
.parse(*self._post("auth/w/position/increase", body={ "symbol": symbol, "amount": amount }))
def get_increase_position_info(self, symbol: str, amount: Union[Decimal, float, str]) -> PositionIncreaseInfo: def get_increase_position_info(self,
response = self._post(f"auth/r/position/increase/info", body={ "symbol": symbol, "amount": amount }) symbol: str,
amount: Union[Decimal, float, str]) -> PositionIncreaseInfo:
response = self._post("auth/r/position/increase/info", body={ "symbol": symbol, "amount": amount })
data = response[0] + [response[1][0]] + response[1][1] + [response[1][2]] + response[4] + response[5] data = response[0] + [response[1][0]] + response[1][1] + [response[1][2]] + response[4] + response[5]
return serializers.PositionIncreaseInfo.parse(*data) return serializers.PositionIncreaseInfo.parse(*data)
def get_positions_history(self, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionHistory]: def get_positions_history(self,
return [ serializers.PositionHistory.parse(*sub_data) for sub_data in self._post("auth/r/positions/hist", body={ "start": start, "end": end, "limit": limit }) ] *,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[PositionHistory]:
return [ serializers.PositionHistory.parse(*sub_data) \
for sub_data in self._post("auth/r/positions/hist", \
body={ "start": start, "end": end, "limit": limit }) ]
def get_positions_snapshot(self, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionSnapshot]: def get_positions_snapshot(self,
return [ serializers.PositionSnapshot.parse(*sub_data) for sub_data in self._post("auth/r/positions/snap", body={ "start": start, "end": end, "limit": limit }) ] *,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[PositionSnapshot]:
return [ serializers.PositionSnapshot.parse(*sub_data) \
for sub_data in self._post("auth/r/positions/snap", \
body={ "start": start, "end": end, "limit": limit }) ]
def get_positions_audit(self, ids: Optional[List[int]] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[PositionAudit]: def get_positions_audit(self,
return [ serializers.PositionAudit.parse(*sub_data) for sub_data in self._post("auth/r/positions/audit", body={ "ids": ids, "start": start, "end": end, "limit": limit }) ] *,
ids: Optional[List[int]] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[PositionAudit]:
return [ serializers.PositionAudit.parse(*sub_data) \
for sub_data in self._post("auth/r/positions/audit", \
body={ "ids": ids, "start": start, "end": end, "limit": limit }) ]
def set_derivative_position_collateral(self, symbol: str, collateral: Union[Decimal, float, str]) -> DerivativePositionCollateral: def set_derivative_position_collateral(self,
return serializers.DerivativePositionCollateral.parse(*(self._post("auth/w/deriv/collateral/set", body={ "symbol": symbol, "collateral": collateral })[0])) symbol: str,
collateral: Union[Decimal, float, str]) -> DerivativePositionCollateral:
body = { "symbol": symbol, "collateral": collateral }
response = self._post("auth/w/deriv/collateral/set", body=body)
return serializers.DerivativePositionCollateral.parse(*(response)[0])
def get_derivative_position_collateral_limits(self, symbol: str) -> DerivativePositionCollateralLimits: def get_derivative_position_collateral_limits(self, symbol: str) -> DerivativePositionCollateralLimits:
return serializers.DerivativePositionCollateralLimits.parse(*self._post("auth/calc/deriv/collateral/limits", body={ "symbol": symbol })) body = { "symbol": symbol }
response = self._post("auth/calc/deriv/collateral/limit", body=body)
return serializers.DerivativePositionCollateralLimits.parse(*response)
def get_funding_offers(self, symbol: Optional[str] = None) -> List[FundingOffer]: def get_funding_offers(self, *, symbol: Optional[str] = None) -> List[FundingOffer]:
endpoint = "auth/r/funding/offers" endpoint = "auth/r/funding/offers"
if symbol != None: if symbol is not None:
endpoint += f"/{symbol}" endpoint += f"/{symbol}"
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, float, str], #pylint: disable-next=too-many-arguments
rate: Union[Decimal, float, str], period: int, def submit_funding_offer(self,
type: FundingOfferType,
symbol: str,
amount: Union[Decimal, float, str],
rate: Union[Decimal, float, str],
period: int,
*,
flags: Optional[int] = 0) -> Notification[FundingOffer]: flags: Optional[int] = 0) -> Notification[FundingOffer]:
body = { body = {
"type": type, "symbol": symbol, "amount": amount, "type": type, "symbol": symbol, "amount": amount,
@@ -179,37 +298,52 @@ class RestAuthenticatedEndpoints(Middleware):
"flags": flags "flags": flags
} }
return serializers._Notification[FundingOffer](serializers.FundingOffer).parse(*self._post("auth/w/funding/offer/submit", body=body)) return _Notification[FundingOffer](serializers.FundingOffer) \
.parse(*self._post("auth/w/funding/offer/submit", body=body))
def cancel_funding_offer(self, id: int) -> Notification[FundingOffer]: def cancel_funding_offer(self, id: int) -> Notification[FundingOffer]:
return serializers._Notification[FundingOffer](serializers.FundingOffer).parse(*self._post("auth/w/funding/offer/cancel", body={ "id": id })) return _Notification[FundingOffer](serializers.FundingOffer) \
.parse(*self._post("auth/w/funding/offer/cancel", body={ "id": id }))
def cancel_all_funding_offers(self, currency: str) -> Notification[Literal[None]]: def cancel_all_funding_offers(self, currency: str) -> Notification[Literal[None]]:
return serializers._Notification[Literal[None]](None).parse( return _Notification[Literal[None]](None) \
*self._post("auth/w/funding/offer/cancel/all", body={ "currency": currency }) .parse(*self._post("auth/w/funding/offer/cancel/all", body={ "currency": currency }))
)
def submit_funding_close(self, id: int) -> Notification[Literal[None]]: def submit_funding_close(self, id: int) -> Notification[Literal[None]]:
return serializers._Notification[Literal[None]](None).parse( return _Notification[Literal[None]](None) \
*self._post("auth/w/funding/close", body={ "id": id }) .parse(*self._post("auth/w/funding/close", body={ "id": id }))
)
def toggle_auto_renew(self, status: bool, currency: str, amount: Optional[str] = None, rate: Optional[int] = None, period: Optional[int] = None) -> Notification[FundingAutoRenew]: def toggle_auto_renew(self,
return serializers._Notification[FundingAutoRenew](serializers.FundingAutoRenew).parse(*self._post("auth/w/funding/auto", body={ status: bool,
"status": int(status), currency: str,
"currency": currency, "amount": amount, *,
amount: Optional[str] = None,
rate: Optional[int] = None,
period: Optional[int] = None) -> Notification[FundingAutoRenew]:
return _Notification[FundingAutoRenew](serializers.FundingAutoRenew) \
.parse(*self._post("auth/w/funding/auto", body={
"status": int(status), "currency": currency, "amount": amount,
"rate": rate, "period": period "rate": rate, "period": period
})) }))
def toggle_keep_funding(self, type: Literal["credit", "loan"], ids: Optional[List[int]] = None, changes: Optional[Dict[int, Literal[1, 2]]] = None) -> Notification[Literal[None]]: def toggle_keep_funding(self,
return serializers._Notification[Literal[None]](None).parse(*self._post("auth/w/funding/keep", body={ type: Literal["credit", "loan"],
*,
ids: Optional[List[int]] = None,
changes: Optional[Dict[int, Literal[1, 2]]] = None) -> Notification[Literal[None]]:
return _Notification[Literal[None]](None).parse(*self._post("auth/w/funding/keep", body={
"type": type, "type": type,
"id": ids, "id": ids,
"changes": changes "changes": changes
})) }))
def get_funding_offers_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingOffer]: def get_funding_offers_history(self,
if symbol == None: *,
symbol: Optional[str] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[FundingOffer]:
if symbol is None:
endpoint = "auth/r/funding/offers/hist" endpoint = "auth/r/funding/offers/hist"
else: endpoint = f"auth/r/funding/offers/{symbol}/hist" else: endpoint = f"auth/r/funding/offers/{symbol}/hist"
@@ -220,15 +354,20 @@ class RestAuthenticatedEndpoints(Middleware):
return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] return [ serializers.FundingOffer.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]
def get_funding_loans(self, symbol: Optional[str] = None) -> List[FundingLoan]: def get_funding_loans(self, *, symbol: Optional[str] = None) -> List[FundingLoan]:
if symbol == None: if symbol is None:
endpoint = "auth/r/funding/loans" endpoint = "auth/r/funding/loans"
else: endpoint = f"auth/r/funding/loans/{symbol}" else: endpoint = f"auth/r/funding/loans/{symbol}"
return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint) ] return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint) ]
def get_funding_loans_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingLoan]: def get_funding_loans_history(self,
if symbol == None: *,
symbol: Optional[str] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[FundingLoan]:
if symbol is None:
endpoint = "auth/r/funding/loans/hist" endpoint = "auth/r/funding/loans/hist"
else: endpoint = f"auth/r/funding/loans/{symbol}/hist" else: endpoint = f"auth/r/funding/loans/{symbol}/hist"
@@ -239,15 +378,20 @@ class RestAuthenticatedEndpoints(Middleware):
return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] return [ serializers.FundingLoan.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]
def get_funding_credits(self, symbol: Optional[str] = None) -> List[FundingCredit]: def get_funding_credits(self, *, symbol: Optional[str] = None) -> List[FundingCredit]:
if symbol == None: if symbol is None:
endpoint = "auth/r/funding/credits" endpoint = "auth/r/funding/credits"
else: endpoint = f"auth/r/funding/credits/{symbol}" else: endpoint = f"auth/r/funding/credits/{symbol}"
return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint) ] return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint) ]
def get_funding_credits_history(self, symbol: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingCredit]: def get_funding_credits_history(self,
if symbol == None: *,
symbol: Optional[str] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[FundingCredit]:
if symbol is None:
endpoint = "auth/r/funding/credits/hist" endpoint = "auth/r/funding/credits/hist"
else: endpoint = f"auth/r/funding/credits/{symbol}/hist" else: endpoint = f"auth/r/funding/credits/{symbol}/hist"
@@ -258,8 +402,14 @@ class RestAuthenticatedEndpoints(Middleware):
return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ] return [ serializers.FundingCredit.parse(*sub_data) for sub_data in self._post(endpoint, body=body) ]
def get_funding_trades_history(self, symbol: Optional[str] = None, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingTrade]: def get_funding_trades_history(self,
if symbol == None: *,
symbol: Optional[str] = None,
sort: Optional[Sort] = None,
start: Optional[str] = None,
end: Optional[str] = None,
limit: Optional[int] = None) -> List[FundingTrade]:
if symbol is None:
endpoint = "auth/r/funding/trades/hist" endpoint = "auth/r/funding/trades/hist"
else: endpoint = f"auth/r/funding/trades/{symbol}/hist" else: endpoint = f"auth/r/funding/trades/{symbol}/hist"
@@ -276,40 +426,57 @@ class RestAuthenticatedEndpoints(Middleware):
data = [response[1]] + response[2] data = [response[1]] + response[2]
return serializers.FundingInfo.parse(*data) return serializers.FundingInfo.parse(*data)
def transfer_between_wallets(self, from_wallet: str, to_wallet: str, currency: str, currency_to: str, amount: Union[Decimal, float, str]) -> Notification[Transfer]: #pylint: disable-next=too-many-arguments
def transfer_between_wallets(self,
from_wallet: str,
to_wallet: str,
currency: str,
currency_to: str,
amount: Union[Decimal, float, str]) -> Notification[Transfer]:
body = { body = {
"from": from_wallet, "to": to_wallet, "from": from_wallet, "to": to_wallet,
"currency": currency, "currency_to": currency_to, "currency": currency, "currency_to": currency_to,
"amount": amount "amount": amount
} }
return serializers._Notification[Transfer](serializers.Transfer).parse(*self._post("auth/w/transfer", body=body)) return _Notification[Transfer](serializers.Transfer) \
.parse(*self._post("auth/w/transfer", body=body))
def submit_wallet_withdrawal(self, wallet: str, method: str, address: str, amount: Union[Decimal, float, str]) -> Notification[Withdrawal]: def submit_wallet_withdrawal(self,
return serializers._Notification[Withdrawal](serializers.Withdrawal).parse(*self._post("auth/w/withdraw", body={ wallet: str,
"wallet": wallet, "method": method, method: str,
"address": address, "amount": amount, address: str,
amount: Union[Decimal, float, str]) -> Notification[Withdrawal]:
return _Notification[Withdrawal](serializers.Withdrawal) \
.parse(*self._post("auth/w/withdraw", body={
"wallet": wallet, "method": method, "address": address,
"amount": amount,
})) }))
def get_deposit_address(self, wallet: str, method: str, renew: bool = False) -> Notification[DepositAddress]: def get_deposit_address(self,
body = { wallet: str,
"wallet": wallet, method: str,
"method": method, renew: bool = False) -> Notification[DepositAddress]:
"renew": int(renew) return _Notification[DepositAddress](serializers.DepositAddress) \
} .parse(*self._post("auth/w/deposit/address", body={
"wallet": wallet, "method": method, "renew": int(renew)
}))
return serializers._Notification[DepositAddress](serializers.DepositAddress).parse(*self._post("auth/w/deposit/address", body=body)) def generate_deposit_invoice(self,
wallet: str,
currency: str,
amount: Union[Decimal, float, str]) -> LightningNetworkInvoice:
return serializers.LightningNetworkInvoice.parse(*self._post("auth/w/deposit/invoice", body={
"wallet": wallet, "currency": currency, "amount": amount
}))
def generate_deposit_invoice(self, wallet: str, currency: str, amount: Union[Decimal, float, str]) -> LightningNetworkInvoice: def get_movements(self,
body = { *,
"wallet": wallet, "currency": currency, currency: Optional[str] = None,
"amount": amount start: Optional[str] = None,
} end: Optional[str] = None,
limit: Optional[int] = None) -> List[Movement]:
return serializers.LightningNetworkInvoice.parse(*self._post("auth/w/deposit/invoice", body=body)) if currency is None:
def get_movements(self, currency: Optional[str] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Movement]:
if currency == None:
endpoint = "auth/r/movements/hist" endpoint = "auth/r/movements/hist"
else: endpoint = f"auth/r/movements/{currency}/hist" else: endpoint = f"auth/r/movements/{currency}/hist"