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

Add funding related rest endpoints, refactor pre-existent rest endpoints to use get_ prefix. Add function to calculate flags easily. Add example test to create a funding offer.

Browse Source
This commit is contained in:
itsdeka
2023-01-06 15:18:57 +01:00
committed by Davide Casale
parent 72a3252e32
commit ef836bbe1a
7 changed files with 170 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
bfxapi/client.py
View File

@@ -1,3 +1,4 @@
from .rest import BfxRestInterface
from .websocket import BfxWebsocketClient
from typing import Optional
@@ -12,7 +13,20 @@ class Constants(str, Enum):
PUB_WSS_HOST = "wss://api-pub.bitfinex.com/ws/2"
class Client(object):
def __init__(self, WSS_HOST: str = Constants.WSS_HOST, API_KEY: Optional[str] = None, API_SECRET: Optional[str] = None, log_level: str = "WARNING"):
def __init__(
self,
REST_HOST: str = Constants.REST_HOST,
WSS_HOST: str = Constants.WSS_HOST,
API_KEY: Optional[str] = None,
API_SECRET: Optional[str] = None,
log_level: str = "WARNING"
):
self.rest = BfxRestInterface(
host=REST_HOST,
API_KEY=API_KEY,
API_SECRET=API_SECRET
)
self.wss = BfxWebsocketClient(
host=WSS_HOST,
API_KEY=API_KEY,

5
bfxapi/enums.py
View File

@@ -16,6 +16,11 @@ class OrderType(str, Enum):
IOC = "IOC"
EXCHANGE_IOC = "EXCHANGE IOC"
class FundingOfferType(str, Enum):
LIMIT = "LIMIT"
FRR_DELTA_FIX = "FRRDELTAFIX"
FRR_DELTA_VAR = "FRRDELTAVAR"
class Flag(int, Enum):
HIDDEN = 64
CLOSE = 512

91
bfxapi/rest/BfxRestInterface.py
View File

@@ -9,7 +9,7 @@ from typing import List, Union, Literal, Optional, Any, cast
from . import serializers
from .typings import *
from .enums import Config, Precision, Sort, OrderType, Error
from .enums import Config, Precision, Sort, OrderType, FundingOfferType, Error
from .exceptions import ResourceNotFound, RequestParametersError, InvalidAuthenticationCredentials, UnknownGenericError
from .. utils.integers import Int16, int32, int45, int64
@@ -84,39 +84,39 @@ class _Requests(object):
return data
class _RestPublicEndpoints(_Requests):
def platform_status(self) -> PlatformStatus:
def get_platform_status(self) -> PlatformStatus:
return serializers.PlatformStatus.parse(*self._GET("platform/status"))
def tickers(self, symbols: List[str]) -> List[Union[TradingPairTicker, FundingCurrencyTicker]]:
def get_tickers(self, symbols: List[str]) -> List[Union[TradingPairTicker, FundingCurrencyTicker]]:
data = self._GET("tickers", params={ "symbols": ",".join(symbols) })
parsers = { "t": serializers.TradingPairTicker.parse, "f": serializers.FundingCurrencyTicker.parse }
return [ parsers[subdata[0][0]](*subdata) for subdata in data ]
def t_tickers(self, pairs: Union[List[str], Literal["ALL"]]) -> List[TradingPairTicker]:
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.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.tickers([ "t" + pair for pair in pairs ])
data = self.get_tickers([ "t" + pair for pair in pairs ])
return cast(List[TradingPairTicker], data)
def f_tickers(self, currencies: Union[List[str], Literal["ALL"]]) -> List[FundingCurrencyTicker]:
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.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.tickers([ "f" + currency for currency in currencies ])
data = self.get_tickers([ "f" + currency for currency in currencies ])
return cast(List[FundingCurrencyTicker], data)
def t_ticker(self, pair: str) -> TradingPairTicker:
def get_t_ticker(self, pair: str) -> TradingPairTicker:
return serializers.TradingPairTicker.parse(*self._GET(f"ticker/t{pair}"), skip=["SYMBOL"])
def f_ticker(self, currency: str) -> FundingCurrencyTicker:
def get_f_ticker(self, currency: str) -> FundingCurrencyTicker:
return serializers.FundingCurrencyTicker.parse(*self._GET(f"ticker/f{currency}"), skip=["SYMBOL"])
def tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[TickersHistory]:
def get_tickers_history(self, symbols: List[str], start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[TickersHistory]:
params = {
"symbols": ",".join(symbols),
"start": start, "end": end,
@@ -127,29 +127,29 @@ class _RestPublicEndpoints(_Requests):
return [ serializers.TickersHistory.parse(*subdata) for subdata in data ]
def t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[TradingPairTrade]:
def get_t_trades(self, pair: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[TradingPairTrade]:
params = { "limit": limit, "start": start, "end": end, "sort": sort }
data = self._GET(f"trades/{'t' + pair}/hist", params=params)
return [ serializers.TradingPairTrade.parse(*subdata) for subdata in data ]
def f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[FundingCurrencyTrade]:
def get_f_trades(self, currency: str, limit: Optional[int] = None, start: Optional[str] = None, end: Optional[str] = None, sort: Optional[Sort] = None) -> List[FundingCurrencyTrade]:
params = { "limit": limit, "start": start, "end": end, "sort": sort }
data = self._GET(f"trades/{'f' + currency}/hist", params=params)
return [ serializers.FundingCurrencyTrade.parse(*subdata) for subdata in data ]
def t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairBook]:
def get_t_book(self, pair: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairBook]:
return [ serializers.TradingPairBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/{precision}", params={ "len": len }) ]
def f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyBook]:
def get_f_book(self, currency: str, precision: Literal["P0", "P1", "P2", "P3", "P4"], len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyBook]:
return [ serializers.FundingCurrencyBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/{precision}", params={ "len": len }) ]
def t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairRawBook]:
def get_t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairRawBook]:
return [ serializers.TradingPairRawBook.parse(*subdata) for subdata in self._GET(f"book/{'t' + pair}/R0", params={ "len": len }) ]
def f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyRawBook]:
def get_f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyRawBook]:
return [ serializers.FundingCurrencyRawBook.parse(*subdata) for subdata in self._GET(f"book/{'f' + currency}/R0", params={ "len": len }) ]
def stats_hist(
def get_stats_hist(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -158,7 +158,7 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"stats1/{resource}/hist", params=params)
return [ serializers.Statistic.parse(*subdata) for subdata in data ]
def stats_last(
def get_stats_last(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -167,7 +167,7 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"stats1/{resource}/last", params=params)
return serializers.Statistic.parse(*data)
def candles_hist(
def get_candles_hist(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -176,7 +176,7 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"candles/{resource}/hist", params=params)
return [ serializers.Candle.parse(*subdata) for subdata in data ]
def candles_last(
def get_candles_last(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -185,14 +185,14 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"candles/{resource}/last", params=params)
return serializers.Candle.parse(*data)
def derivatives_status(self, type: str, keys: List[str]) -> List[DerivativesStatus]:
def get_derivatives_status(self, type: str, keys: List[str]) -> List[DerivativesStatus]:
params = { "keys": ",".join(keys) }
data = self._GET(f"status/{type}", params=params)
return [ serializers.DerivativesStatus.parse(*subdata) for subdata in data ]
def derivatives_status_history(
def get_derivatives_status_history(
self,
type: str, symbol: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -203,14 +203,14 @@ class _RestPublicEndpoints(_Requests):
return [ serializers.DerivativesStatus.parse(*subdata, skip=[ "KEY" ]) for subdata in data ]
def liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Liquidation]:
def get_liquidations(self, sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[Liquidation]:
params = { "sort": sort, "start": start, "end": end, "limit": limit }
data = self._GET("liquidations/hist", params=params)
return [ serializers.Liquidation.parse(*subdata[0]) for subdata in data ]
def leaderboards_hist(
def get_leaderboards_hist(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -219,7 +219,7 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"rankings/{resource}/hist", params=params)
return [ serializers.Leaderboard.parse(*subdata) for subdata in data ]
def leaderboards_last(
def get_leaderboards_last(
self,
resource: str,
sort: Optional[Sort] = None, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None
@@ -228,7 +228,7 @@ class _RestPublicEndpoints(_Requests):
data = self._GET(f"rankings/{resource}/last", params=params)
return serializers.Leaderboard.parse(*data)
def funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingStatistic]:
def get_funding_stats(self, symbol: str, start: Optional[str] = None, end: Optional[str] = None, limit: Optional[int] = None) -> List[FundingStatistic]:
params = { "start": start, "end": end, "limit": limit }
data = self._GET(f"funding/stats/{symbol}/hist", params=params)
@@ -239,17 +239,17 @@ class _RestPublicEndpoints(_Requests):
return self._GET(f"conf/{config}")[0]
class _RestAuthenticatedEndpoints(_Requests):
def wallets(self) -> List[Wallet]:
def get_wallets(self) -> List[Wallet]:
return [ serializers.Wallet.parse(*subdata) for subdata in self._POST("auth/r/wallets") ]
def retrieve_orders(self, ids: Optional[List[str]] = None) -> List[Order]:
def get_orders(self, ids: Optional[List[str]] = None) -> List[Order]:
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[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] = None, 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:
data = {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev,
@@ -262,7 +262,7 @@ class _RestAuthenticatedEndpoints(_Requests):
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,
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:
data = {
"id": id, "amount": amount, "price": price,
@@ -293,7 +293,7 @@ class _RestAuthenticatedEndpoints(_Requests):
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]:
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:
endpoint = "auth/r/orders/hist"
else: endpoint = f"auth/r/orders/{symbol}/hist"
@@ -306,14 +306,33 @@ class _RestAuthenticatedEndpoints(_Requests):
return [ serializers.Order.parse(*subdata) for subdata in self._POST(endpoint, data=data) ]
def trades(self, symbol: str) -> List[Trade]:
def get_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]:
def get_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) ]
return [ serializers.Ledger.parse(*subdata) for subdata in self._POST(f"auth/r/ledgers/{currency}/hist", data=data) ]
def get_active_funding_offers(self, symbol: Optional[str] = None) -> List[FundingOffer]:
endpoint = "auth/r/funding/offers"
if symbol != None:
endpoint += f"/{symbol}"
return [ serializers.FundingOffer.parse(*subdata) for subdata in self._POST(endpoint) ]
def submit_funding_offer(self, type: FundingOfferType, symbol: str, amount: Union[Decimal, str],
rate: Union[Decimal, str], period: int,
flags: Optional[int] = 0) -> Notification:
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))

24
bfxapi/rest/serializers.py
View File

@@ -234,6 +234,30 @@ Order = _Serializer[typings.Order]("Order", labels=[
"META"
])
FundingOffer = _Serializer[typings.FundingOffer]("FundingOffer", labels=[
"ID",
"SYMBOL",
"MTS_CREATED",
"MTS_UPDATED",
"AMOUNT",
"AMOUNT_ORIG",
"OFFER_TYPE",
"_PLACEHOLDER",
"_PLACEHOLDER",
"FLAGS",
"OFFER_STATUS",
"_PLACEHOLDER",
"_PLACEHOLDER",
"_PLACEHOLDER",
"RATE",
"PERIOD",
"NOTIFY",
"HIDDEN",
"_PLACEHOLDER",
"RENEW",
"_PLACEHOLDER"
])
Trade = _Serializer[typings.Trade]("Trade", labels=[
"ID",
"PAIR",

16
bfxapi/rest/typings.py
View File

@@ -169,6 +169,22 @@ class Order(TypedDict):
ROUTING: str
META: JSON
class FundingOffer(TypedDict):
ID: int
SYMBOL: str
MTS_CREATE: int
MTS_UPDATE: int
AMOUNT: float
AMOUNT_ORIG: float
OFFER_TYPE: str
FLAGS: int
OFFER_STATUS: str
RATE: float
PERIOD: int
NOTIFY: bool
HIDDEN: int
RENEW: bool
class Trade(TypedDict):
ID: int
SYMBOL: str

29
bfxapi/utils/flags.py Normal file
View File

@@ -0,0 +1,29 @@
from .. enums import Flag
def calculate_order_flags(
hidden : bool = False,
close : bool = False,
reduce_only : bool = False,
post_only : bool = False,
oco : bool = False,
no_var_rates: bool = False
) -> int:
flags = 0
if hidden: flags += Flag.HIDDEN
if close: flags += Flag.CLOSE
if reduce_only: flags += Flag.REDUCE_ONLY
if post_only: flags += Flag.POST_ONLY
if oco: flags += Flag.OCO
if no_var_rates: flags += Flag.NO_VAR_RATES
return flags
def calculate_offer_flags(
hidden : bool = False
) -> int:
flags = 0
if hidden: flags += Flag.HIDDEN
return flags

26
examples/rest/create_funding_offer.py Normal file
View File

@@ -0,0 +1,26 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.utils.flags import calculate_offer_flags
from bfxapi.rest.typings import List, FundingOffer, Notification
bfx = Client(
REST_HOST=Constants.REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)
notification: Notification = bfx.rest.auth.submit_funding_offer(
type="LIMIT",
symbol="fUSD",
amount="123.45",
rate="0.001",
period=2,
flags=calculate_offer_flags(hidden=True)
)
print("Offer notification:", notification)
offers: List[FundingOffer] = bfx.rest.auth.get_active_funding_offers()
print("Offers:", offers)