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

Merge pull request #35 from Davi0kProgramsThings/fix/refactoring

Browse Source 
Merge branch `fix/refactoring` in branch `feature/rest`.
This commit is contained in:
Davide Casale
2023-02-17 20:26:43 +01:00
committed by GitHub
parent f0d14a230f f4c6a21ef4
commit 21117f62d0
59 changed files with 741 additions and 541 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
LICENSE
View File

@@ -1,3 +1,4 @@
Apache License
Version 2.0, January 2004
http://www.apache.org/licenses/

7
bfxapi/__init__.py
View File

@@ -1 +1,6 @@
from .client import Client, Constants
from .client import Client
from .urls import REST_HOST, PUB_REST_HOST, STAGING_REST_HOST, \
WSS_HOST, PUB_WSS_HOST, STAGING_WSS_HOST
NAME = "bfxapi"

32
bfxapi/client.py
View File

@@ -1,37 +1,31 @@
from .rest import BfxRestInterface
from .websocket import BfxWebsocketClient
from .urls import REST_HOST, WSS_HOST
from typing import Optional
from enum import Enum
class Constants(str, Enum):
REST_HOST = "https://api.bitfinex.com/v2"
PUB_REST_HOST = "https://api-pub.bitfinex.com/v2"
STAGING_REST_HOST = "https://api.staging.bitfinex.com/v2"
WSS_HOST = "wss://api.bitfinex.com/ws/2"
PUB_WSS_HOST = "wss://api-pub.bitfinex.com/ws/2"
STAGING_WSS_HOST = "wss://api.staging.bitfinex.com/ws/2"
from typing import List, Optional
class Client(object):
def __init__(
self,
REST_HOST: str = Constants.REST_HOST,
WSS_HOST: str = Constants.WSS_HOST,
REST_HOST: str = REST_HOST,
WSS_HOST: str = WSS_HOST,
API_KEY: Optional[str] = None,
API_SECRET: Optional[str] = None,
log_level: str = "WARNING"
filter: Optional[List[str]] = None,
log_level: str = "INFO"
):
credentials = None
if API_KEY and API_SECRET:
credentials = { "API_KEY": API_KEY, "API_SECRET": API_SECRET, "filter": filter }
self.rest = BfxRestInterface(
host=REST_HOST,
API_KEY=API_KEY,
API_SECRET=API_SECRET
credentials=credentials
)
self.wss = BfxWebsocketClient(
host=WSS_HOST,
API_KEY=API_KEY,
API_SECRET=API_SECRET,
credentials=credentials,
log_level=log_level
)

3
bfxapi/rest/__init__.py
View File

@@ -1,3 +1,4 @@
from .endpoints import BfxRestInterface, RestPublicEndpoints, RestAuthenticatedEndpoints
from .endpoints import BfxRestInterface, RestPublicEndpoints, RestAuthenticatedEndpoints, \
RestMerchantEndpoints
NAME = "rest"

2
bfxapi/rest/endpoints/__init__.py
View File

@@ -1,5 +1,7 @@
from .bfx_rest_interface import BfxRestInterface
from .rest_public_endpoints import RestPublicEndpoints
from .rest_authenticated_endpoints import RestAuthenticatedEndpoints
from .rest_merchant_endpoints import RestMerchantEndpoints
NAME = "endpoints"

5
bfxapi/rest/endpoints/bfx_rest_interface.py
View File

@@ -7,7 +7,10 @@ from .rest_merchant_endpoints import RestMerchantEndpoints
class BfxRestInterface(object):
VERSION = 2
def __init__(self, host: str, API_KEY: Optional[str] = None, API_SECRET: Optional[str] = None):
def __init__(self, host, credentials = None):
API_KEY, API_SECRET = credentials and \
(credentials["API_KEY"], credentials["API_SECRET"]) or (None, None)
self.public = RestPublicEndpoints(host=host)
self.auth = RestAuthenticatedEndpoints(host=host, API_KEY=API_KEY, API_SECRET=API_SECRET)
self.merchant = RestMerchantEndpoints(host=host, API_KEY=API_KEY, API_SECRET=API_SECRET)

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

@@ -1,4 +1,4 @@
from typing import List, Union, Literal, Optional
from typing import List, Tuple, Union, Literal, Optional
from decimal import Decimal
from datetime import datetime

5
bfxapi/rest/endpoints/rest_merchant_endpoints.py
View File

@@ -1,9 +1,10 @@
from typing import List, Union, Literal, Optional
from typing import TypedDict, List, Union, Literal, Optional
from decimal import Decimal
from .. types import *
from .. middleware import Middleware
from ... utils.camel_and_snake_case_adapters import to_snake_case_keys, to_camel_case_keys
from ...utils.camel_and_snake_case_helpers import to_snake_case_keys, to_camel_case_keys
_CustomerInfo = TypedDict("_CustomerInfo", {
"nationality": str, "resid_country": str, "resid_city": str,

16
bfxapi/rest/endpoints/rest_public_endpoints.py
View File

@@ -25,7 +25,7 @@ class RestPublicEndpoints(Middleware):
if isinstance(pairs, str) and pairs == "ALL":
return [ cast(TradingPairTicker, sub_data) for sub_data in self.get_tickers([ "ALL" ]) if cast(str, sub_data.symbol).startswith("t") ]
data = self.get_tickers([ "t" + pair for pair in pairs ])
data = self.get_tickers([ pair for pair in pairs ])
return cast(List[TradingPairTicker], data)
@@ -33,7 +33,7 @@ class RestPublicEndpoints(Middleware):
if isinstance(currencies, str) and currencies == "ALL":
return [ cast(FundingCurrencyTicker, sub_data) for sub_data in self.get_tickers([ "ALL" ]) if cast(str, sub_data.symbol).startswith("f") ]
data = self.get_tickers([ "f" + currency for currency in currencies ])
data = self.get_tickers([ currency for currency in currencies ])
return cast(List[FundingCurrencyTicker], data)
@@ -52,25 +52,25 @@ class RestPublicEndpoints(Middleware):
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)
data = self._GET(f"trades/{pair}/hist", params=params)
return [ serializers.TradingPairTrade.parse(*sub_data) for sub_data in data ]
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)
data = self._GET(f"trades/{currency}/hist", params=params)
return [ serializers.FundingCurrencyTrade.parse(*sub_data) for sub_data in data ]
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(*sub_data) for sub_data in self._GET(f"book/{'t' + pair}/{precision}", params={ "len": len }) ]
return [ serializers.TradingPairBook.parse(*sub_data) for sub_data in self._GET(f"book/{pair}/{precision}", params={ "len": len }) ]
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(*sub_data) for sub_data in self._GET(f"book/{'f' + currency}/{precision}", params={ "len": len }) ]
return [ serializers.FundingCurrencyBook.parse(*sub_data) for sub_data in self._GET(f"book/{currency}/{precision}", params={ "len": len }) ]
def get_t_raw_book(self, pair: str, len: Optional[Literal[1, 25, 100]] = None) -> List[TradingPairRawBook]:
return [ serializers.TradingPairRawBook.parse(*sub_data) for sub_data in self._GET(f"book/{'t' + pair}/R0", params={ "len": len }) ]
return [ serializers.TradingPairRawBook.parse(*sub_data) for sub_data in self._GET(f"book/{pair}/R0", params={ "len": len }) ]
def get_f_raw_book(self, currency: str, len: Optional[Literal[1, 25, 100]] = None) -> List[FundingCurrencyRawBook]:
return [ serializers.FundingCurrencyRawBook.parse(*sub_data) for sub_data in self._GET(f"book/{'f' + currency}/R0", params={ "len": len }) ]
return [ serializers.FundingCurrencyRawBook.parse(*sub_data) for sub_data in self._GET(f"book/{currency}/R0", params={ "len": len }) ]
def get_stats_hist(
self,

41
bfxapi/rest/types.py
View File

@@ -1,4 +1,4 @@
from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Literal, Any
from typing import List, Dict, Optional, Literal, Any
from dataclasses import dataclass
@@ -575,48 +575,47 @@ class InvoiceSubmission(_Type):
currency: str
order_id: str
pay_currencies: List[str]
webhook: Optional[str]
redirect_url: Optional[str]
webhook: str
redirect_url: str
status: Literal["CREATED", "PENDING", "COMPLETED", "EXPIRED"]
customer_info: Optional["_CustomerInfo"]
invoices: List["_Invoice"]
payment: Optional["_Payment"]
additional_payments: Optional[List["_Payment"]]
customer_info: "CustomerInfo"
invoices: List["Invoice"]
payment: "Payment"
additional_payments: List["Payment"]
merchant_name: str
@classmethod
def parse(cls, data: Dict[str, Any]) -> "InvoiceSubmission":
if "customer_info" in data and data["customer_info"] != None:
data["customer_info"] = _CustomerInfo(**data["customer_info"])
data["customer_info"] = InvoiceSubmission.CustomerInfo(**data["customer_info"])
for index, invoice in enumerate(data["invoices"]):
data["invoices"][index] = _Invoice(**invoice)
data["invoices"][index] = InvoiceSubmission.Invoice(**invoice)
if "payment" in data and data["payment"] != None:
data["payment"] = _Payment(**data["payment"])
data["payment"] = InvoiceSubmission.Payment(**data["payment"])
if "additional_payments" in data and data["additional_payments"] != None:
for index, additional_payment in enumerate(data["additional_payments"]):
data["additional_payments"][index] = _Payment(**additional_payment)
data["additional_payments"][index] = InvoiceSubmission.Payment(**additional_payment)
return InvoiceSubmission(**data)
@compose(dataclass, partial)
class _CustomerInfo:
class CustomerInfo:
nationality: str
resid_country: str
resid_state: Optional[str]
resid_state: str
resid_city: str
resid_zip_code: str
resid_street: str
resid_building_no: Optional[str]
resid_building_no: str
full_name: str
email: str
tos_accepted: bool
@compose(dataclass, partial)
class _Invoice:
class Invoice:
amount: float
currency: str
pay_currency: str
@@ -625,7 +624,7 @@ class _Invoice:
ext: JSON
@compose(dataclass, partial)
class _Payment:
class Payment:
txid: str
amount: float
currency: str
@@ -634,10 +633,10 @@ class _Payment:
confirmations: int
created_at: str
updated_at: str
deposit_id: Optional[int]
ledger_id: Optional[int]
force_completed: Optional[bool]
amount_diff: Optional[str]
deposit_id: int
ledger_id: int
force_completed: bool
amount_diff: str
@dataclass
class InvoiceStats(_Type):

8
bfxapi/tests/__init__.py
View File

@@ -1,6 +1,6 @@
import unittest
from .test_rest_serializers_and_types import TestRestSerializersAndTypes
from .test_websocket_serializers_and_types import TestWebsocketSerializersAndTypes
from .test_rest_serializers import TestRestSerializers
from .test_websocket_serializers import TestWebsocketSerializers
from .test_labeler import TestLabeler
from .test_notification import TestNotification
@@ -8,8 +8,8 @@ NAME = "tests"
def suite():
return unittest.TestSuite([
unittest.makeSuite(TestRestSerializersAndTypes),
unittest.makeSuite(TestWebsocketSerializersAndTypes),
unittest.makeSuite(TestRestSerializers),
unittest.makeSuite(TestWebsocketSerializers),
unittest.makeSuite(TestLabeler),
unittest.makeSuite(TestNotification),
])

17
bfxapi/tests/test_rest_serializers.py Normal file
View File

@@ -0,0 +1,17 @@
import unittest
from ..labeler import _Type
from ..rest import serializers
class TestRestSerializers(unittest.TestCase):
def test_rest_serializers(self):
for serializer in map(serializers.__dict__.get, serializers.__serializers__):
self.assertTrue(issubclass(serializer.klass, _Type),
f"_Serializer <{serializer.name}>: .klass field must be a subclass of _Type (got {serializer.klass}).")
self.assertListEqual(serializer.get_labels(), list(serializer.klass.__annotations__),
f"_Serializer <{serializer.name}> and _Type <{serializer.klass.__name__}> must have matching labels and fields.")
if __name__ == "__main__":
unittest.main()

17
bfxapi/tests/test_rest_serializers_and_types.py
View File

@@ -1,17 +0,0 @@
import unittest
from ..rest import serializers, types
class TestRestSerializersAndTypes(unittest.TestCase):
def test_consistency(self):
for serializer in map(serializers.__dict__.get, serializers.__serializers__):
type = types.__dict__.get(serializer.name)
self.assertIsNotNone(type, f"_Serializer <{serializer.name}>: no respective _Type found in bfxapi.rest.types.")
self.assertEqual(serializer.klass, type, f"_Serializer <{serializer.name}>.klass: field does not match with respective _Type in bfxapi.rest.types.")
self.assertListEqual(serializer.get_labels(), list(type.__annotations__),
f"_Serializer <{serializer.name}> and _Type <{type.__name__}> must have matching labels and fields.")
if __name__ == "__main__":
unittest.main()

17
bfxapi/tests/test_websocket_serializers.py Normal file
View File

@@ -0,0 +1,17 @@
import unittest
from ..labeler import _Type
from ..websocket import serializers
class TestWebsocketSerializers(unittest.TestCase):
def test_websocket_serializers(self):
for serializer in map(serializers.__dict__.get, serializers.__serializers__):
self.assertTrue(issubclass(serializer.klass, _Type),
f"_Serializer <{serializer.name}>: .klass field must be a subclass of _Type (got {serializer.klass}).")
self.assertListEqual(serializer.get_labels(), list(serializer.klass.__annotations__),
f"_Serializer <{serializer.name}> and _Type <{serializer.klass.__name__}> must have matching labels and fields.")
if __name__ == "__main__":
unittest.main()

17
bfxapi/tests/test_websocket_serializers_and_types.py
View File

@@ -1,17 +0,0 @@
import unittest
from ..websocket import serializers, types
class TestWebsocketSerializersAndTypes(unittest.TestCase):
def test_consistency(self):
for serializer in map(serializers.__dict__.get, serializers.__serializers__):
type = types.__dict__.get(serializer.name)
self.assertIsNotNone(type, f"_Serializer <{serializer.name}>: no respective _Type found in bfxapi.websocket.types.")
self.assertEqual(serializer.klass, type, f"_Serializer <{serializer.name}>.klass: field does not match with respective _Type in bfxapi.websocket.types.")
self.assertListEqual(serializer.get_labels(), list(type.__annotations__),
f"_Serializer <{serializer.name}> and _Type <{type.__name__}> must have matching labels and fields.")
if __name__ == "__main__":
unittest.main()

7
bfxapi/urls.py Normal file
View File

@@ -0,0 +1,7 @@
REST_HOST = "https://api.bitfinex.com/v2"
PUB_REST_HOST = "https://api-pub.bitfinex.com/v2"
STAGING_REST_HOST = "https://api.staging.bitfinex.com/v2"
WSS_HOST = "wss://api.bitfinex.com/ws/2"
PUB_WSS_HOST = "wss://api-pub.bitfinex.com/ws/2"
STAGING_WSS_HOST = "wss://api.staging.bitfinex.com/ws/2"

2
bfxapi/utils/JSONEncoder.py
View File

@@ -2,8 +2,6 @@ import json
from decimal import Decimal
from datetime import datetime
from types import SimpleNamespace
from typing import Type, List, Dict, Union, Any
JSON = Union[Dict[str, "JSON"], List["JSON"], bool, int, float, str, Type[None]]

0
bfxapi/utils/camel_and_snake_case_adapters.py → bfxapi/utils/camel_and_snake_case_helpers.py
View File

99
bfxapi/utils/logger.py
View File

@@ -1,99 +1,52 @@
"""
Module used to describe all of the different data types
"""
import logging
BLACK, RED, GREEN, YELLOW, BLUE, MAGENTA, CYAN, WHITE = range(8)
RESET_SEQ = "\033[0m"
COLOR_SEQ = "\033[1;%dm"
ITALIC_COLOR_SEQ = "\033[3;%dm"
UNDERLINE_COLOR_SEQ = "\033[4;%dm"
BOLD_SEQ = "\033[1m"
UNDERLINE_SEQ = "\033[04m"
YELLOW = '\033[93m'
WHITE = '\33[37m'
BLUE = '\033[34m'
LIGHT_BLUE = '\033[94m'
RED = '\033[91m'
GREY = '\33[90m'
KEYWORD_COLORS = {
'WARNING': YELLOW,
'INFO': LIGHT_BLUE,
'DEBUG': WHITE,
'CRITICAL': YELLOW,
'ERROR': RED,
'TRADE': '\33[102m\33[30m'
}
def formatter_message(message, use_color = True):
"""
Syntax highlight certain keywords
"""
if use_color:
message = message.replace("$RESET", RESET_SEQ).replace("$BOLD", BOLD_SEQ)
else:
message = message.replace("$RESET", "").replace("$BOLD", "")
return message
def format_word(message, word, color_seq, bold=False, underline=False):
"""
Surround the given word with a sequence
"""
replacer = color_seq + word + RESET_SEQ
if underline:
replacer = UNDERLINE_SEQ + replacer
if bold:
replacer = BOLD_SEQ + replacer
return message.replace(word, replacer)
COLORS = {
"DEBUG": CYAN,
"INFO": BLUE,
"WARNING": YELLOW,
"ERROR": RED
}
class Formatter(logging.Formatter):
"""
This Formatted simply colors in the levelname i.e 'INFO', 'DEBUG'
"""
class _ColoredFormatter(logging.Formatter):
def __init__(self, msg, use_color = True):
logging.Formatter.__init__(self, msg)
logging.Formatter.__init__(self, msg, "%d-%m-%Y %H:%M:%S")
self.use_color = use_color
def format(self, record):
"""
Format and highlight certain keywords
"""
levelname = record.levelname
if self.use_color and levelname in KEYWORD_COLORS:
levelname_color = KEYWORD_COLORS[levelname] + levelname + RESET_SEQ
if self.use_color and levelname in COLORS:
levelname_color = COLOR_SEQ % (30 + COLORS[levelname]) + levelname + RESET_SEQ
record.levelname = levelname_color
record.name = GREY + record.name + RESET_SEQ
record.name = ITALIC_COLOR_SEQ % (30 + BLACK) + record.name + RESET_SEQ
return logging.Formatter.format(self, record)
class CustomLogger(logging.Logger):
"""
This adds extra logging functions such as logger.trade and also
sets the logger to use the custom formatter
"""
FORMAT = "[$BOLD%(name)s$RESET] [%(levelname)s] %(message)s"
class ColoredLogger(logging.Logger):
FORMAT = "[$BOLD%(name)s$RESET] [%(asctime)s] [%(levelname)s] %(message)s"
COLOR_FORMAT = formatter_message(FORMAT, True)
TRADE = 50
def __init__(self, name, logLevel='DEBUG'):
logging.Logger.__init__(self, name, logLevel)
color_formatter = Formatter(self.COLOR_FORMAT)
def __init__(self, name, level):
logging.Logger.__init__(self, name, level)
colored_formatter = _ColoredFormatter(self.COLOR_FORMAT)
console = logging.StreamHandler()
console.setFormatter(color_formatter)
console.setFormatter(colored_formatter)
self.addHandler(console)
logging.addLevelName(self.TRADE, "TRADE")
return
def set_level(self, level):
logging.Logger.setLevel(self, level)
def trade(self, message, *args, **kws):
"""
Print a syntax highlighted trade signal
"""
if self.isEnabledFor(self.TRADE):
message = format_word(message, 'CLOSED ', YELLOW, bold=True)
message = format_word(message, 'OPENED ', LIGHT_BLUE, bold=True)
message = format_word(message, 'UPDATED ', BLUE, bold=True)
message = format_word(message, 'CLOSED_ALL ', RED, bold=True)
# Yes, logger takes its '*args' as 'args'.
self._log(self.TRADE, message, args, **kws)

157
bfxapi/websocket/BfxWebsocketClient.py
View File

@@ -1,157 +0,0 @@
import traceback, json, asyncio, hmac, hashlib, time, uuid, websockets
from typing import Literal, TypeVar, Callable, cast
from pyee.asyncio import AsyncIOEventEmitter
from ._BfxWebsocketBucket import _HEARTBEAT, F, _require_websocket_connection, _BfxWebsocketBucket
from ._BfxWebsocketInputs import _BfxWebsocketInputs
from .handlers import Channels, PublicChannelsHandler, AuthenticatedChannelsHandler
from .exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported
from ..utils.JSONEncoder import JSONEncoder
from ..utils.logger import Formatter, CustomLogger
def _require_websocket_authentication(function: F) -> F:
async def wrapper(self, *args, **kwargs):
if self.authentication == False:
raise WebsocketAuthenticationRequired("To perform this action you need to authenticate using your API_KEY and API_SECRET.")
await _require_websocket_connection(function)(self, *args, **kwargs)
return cast(F, wrapper)
class BfxWebsocketClient(object):
VERSION = _BfxWebsocketBucket.VERSION
MAXIMUM_BUCKETS_AMOUNT = 20
EVENTS = [
"open", "subscribed", "authenticated", "wss-error",
*PublicChannelsHandler.EVENTS,
*AuthenticatedChannelsHandler.EVENTS
]
def __init__(self, host, buckets=5, log_level = "WARNING", API_KEY=None, API_SECRET=None, filter=None):
self.host, self.websocket, self.event_emitter = host, None, AsyncIOEventEmitter()
self.event_emitter.add_listener("error",
lambda exception: self.logger.error(str(exception) + "\n" +
str().join(traceback.format_exception(type(exception), exception, exception.__traceback__))[:-1])
)
self.API_KEY, self.API_SECRET, self.filter, self.authentication = API_KEY, API_SECRET, filter, False
self.handler = AuthenticatedChannelsHandler(event_emitter=self.event_emitter)
self.buckets = [ _BfxWebsocketBucket(self.host, self.event_emitter, self.__bucket_open_signal) for _ in range(buckets) ]
self.inputs = _BfxWebsocketInputs(self.__handle_websocket_input)
self.logger = CustomLogger("BfxWebsocketClient", logLevel=log_level)
if buckets > BfxWebsocketClient.MAXIMUM_BUCKETS_AMOUNT:
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_BUCKETS_AMOUNT} buckets from the same \
connection ({buckets} in use), the server could momentarily block the client with <429 Too Many Requests>.")
def run(self):
return asyncio.run(self.start())
async def start(self):
tasks = [ bucket._connect(index) for index, bucket in enumerate(self.buckets) ]
if self.API_KEY != None and self.API_SECRET != None:
tasks.append(self.__connect(self.API_KEY, self.API_SECRET, self.filter))
await asyncio.gather(*tasks)
async def __connect(self, API_KEY, API_SECRET, filter=None):
async for websocket in websockets.connect(self.host):
self.websocket = websocket
await self.__authenticate(API_KEY, API_SECRET, filter)
try:
async for message in websocket:
message = json.loads(message)
if isinstance(message, dict) and message["event"] == "auth":
if message["status"] == "OK":
self.event_emitter.emit("authenticated", message); self.authentication = True
else: raise InvalidAuthenticationCredentials("Cannot authenticate with given API-KEY and API-SECRET.")
elif isinstance(message, dict) and message["event"] == "error":
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) == 0 and message[1] != _HEARTBEAT:
self.handler.handle(message[1], message[2])
except websockets.ConnectionClosedError: continue
finally: await self.websocket.wait_closed(); break
async def __authenticate(self, API_KEY, API_SECRET, filter=None):
data = { "event": "auth", "filter": filter, "apiKey": API_KEY }
data["authNonce"] = int(time.time()) * 1000
data["authPayload"] = "AUTH" + str(data["authNonce"])
data["authSig"] = hmac.new(
API_SECRET.encode("utf8"),
data["authPayload"].encode("utf8"),
hashlib.sha384
).hexdigest()
await self.websocket.send(json.dumps(data))
async def subscribe(self, channel, **kwargs):
counters = [ len(bucket.pendings) + len(bucket.subscriptions) for bucket in self.buckets ]
index = counters.index(min(counters))
await self.buckets[index]._subscribe(channel, **kwargs)
async def unsubscribe(self, chanId):
for bucket in self.buckets:
if chanId in bucket.subscriptions.keys():
await bucket._unsubscribe(chanId=chanId)
async def close(self, code=1000, reason=str()):
if self.websocket != None and self.websocket.open == True:
await self.websocket.close(code=code, reason=reason)
for bucket in self.buckets:
await bucket._close(code=code, reason=reason)
@_require_websocket_authentication
async def notify(self, info, MESSAGE_ID=None, **kwargs):
await self.websocket.send(json.dumps([ 0, "n", MESSAGE_ID, { "type": "ucm-test", "info": info, **kwargs } ]))
@_require_websocket_authentication
async def __handle_websocket_input(self, input, data):
await self.websocket.send(json.dumps([ 0, input, None, data], cls=JSONEncoder))
def __bucket_open_signal(self, index):
if all(bucket.websocket != None and bucket.websocket.open == True for bucket in self.buckets):
self.event_emitter.emit("open")
def on(self, event, callback = None):
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
if callback != None:
return self.event_emitter.on(event, callback)
def handler(function):
self.event_emitter.on(event, function)
return handler
def once(self, event, callback = None):
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
if callback != None:
return self.event_emitter.once(event, callback)
def handler(function):
self.event_emitter.once(event, function)
return handler

4
bfxapi/websocket/__init__.py
View File

@@ -1 +1,3 @@
from .BfxWebsocketClient import BfxWebsocketClient
from .client import BfxWebsocketClient, BfxWebsocketBucket, BfxWebsocketInputs
NAME = "websocket"

5
bfxapi/websocket/client/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
from .bfx_websocket_client import BfxWebsocketClient
from .bfx_websocket_bucket import BfxWebsocketBucket
from .bfx_websocket_inputs import BfxWebsocketInputs
NAME = "client"

50
bfxapi/websocket/_BfxWebsocketBucket.py → bfxapi/websocket/client/bfx_websocket_bucket.py
View File

@@ -2,9 +2,9 @@ import json, uuid, websockets
from typing import Literal, TypeVar, Callable, cast
from .handlers import PublicChannelsHandler
from ..handlers import PublicChannelsHandler
from .exceptions import ConnectionNotOpen, TooManySubscriptions, OutdatedClientVersion
from ..exceptions import ConnectionNotOpen, TooManySubscriptions, OutdatedClientVersion
_HEARTBEAT = "hb"
@@ -19,32 +19,40 @@ def _require_websocket_connection(function: F) -> F:
return cast(F, wrapper)
class _BfxWebsocketBucket(object):
class BfxWebsocketBucket(object):
VERSION = 2
MAXIMUM_SUBSCRIPTIONS_AMOUNT = 25
def __init__(self, host, event_emitter, __bucket_open_signal):
self.host, self.event_emitter, self.__bucket_open_signal = host, event_emitter, __bucket_open_signal
def __init__(self, host, event_emitter, on_open_event):
self.host, self.event_emitter, self.on_open_event = host, event_emitter, on_open_event
self.websocket, self.subscriptions, self.pendings = None, dict(), list()
self.handler = PublicChannelsHandler(event_emitter=self.event_emitter)
async def _connect(self, index):
reconnection = False
async for websocket in websockets.connect(self.host):
self.websocket = websocket
self.__bucket_open_signal(index)
self.on_open_event.set()
if reconnection == True or (reconnection := False):
for pending in self.pendings:
await self.websocket.send(json.dumps(pending))
for _, subscription in self.subscriptions.items():
await self._subscribe(**subscription)
self.subscriptions.clear()
try:
async for message in websocket:
message = json.loads(message)
if isinstance(message, dict) and message["event"] == "info" and "version" in message:
if _BfxWebsocketBucket.VERSION != message["version"]:
raise OutdatedClientVersion(f"Mismatch between the client version and the server version. Update the library to the latest version to continue (client version: {_BfxWebsocketBucket.VERSION}, server version: {message['version']}).")
elif isinstance(message, dict) and message["event"] == "subscribed" and (chanId := message["chanId"]):
if isinstance(message, dict) and message["event"] == "subscribed" and (chanId := message["chanId"]):
self.pendings = [ pending for pending in self.pendings if pending["subId"] != message["subId"] ]
self.subscriptions[chanId] = message
self.event_emitter.emit("subscribed", message)
@@ -55,20 +63,27 @@ class _BfxWebsocketBucket(object):
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) and message[1] != _HEARTBEAT:
self.handler.handle(self.subscriptions[chanId], *message[1:])
except websockets.ConnectionClosedError: continue
finally: await self.websocket.wait_closed(); break
except websockets.ConnectionClosedError as error:
if error.code == 1006:
self.on_open_event.clear()
reconnection = True
continue
raise error
break
@_require_websocket_connection
async def _subscribe(self, channel, subId=None, **kwargs):
if len(self.subscriptions) + len(self.pendings) == _BfxWebsocketBucket.MAXIMUM_SUBSCRIPTIONS_AMOUNT:
if len(self.subscriptions) + len(self.pendings) == BfxWebsocketBucket.MAXIMUM_SUBSCRIPTIONS_AMOUNT:
raise TooManySubscriptions("The client has reached the maximum number of subscriptions.")
subscription = {
**kwargs,
"event": "subscribe",
"channel": channel,
"subId": subId or str(uuid.uuid4()),
**kwargs
}
self.pendings.append(subscription)
@@ -85,3 +100,8 @@ class _BfxWebsocketBucket(object):
@_require_websocket_connection
async def _close(self, code=1000, reason=str()):
await self.websocket.close(code=code, reason=reason)
def _get_chan_id(self, subId):
for subscription in self.subscriptions.values():
if subscription["subId"] == subId:
return subscription["chanId"]

246
bfxapi/websocket/client/bfx_websocket_client.py Normal file
View File

@@ -0,0 +1,246 @@
import traceback, json, asyncio, hmac, hashlib, time, websockets, socket, random
from typing import cast
from collections import namedtuple
from datetime import datetime
from pyee.asyncio import AsyncIOEventEmitter
from .bfx_websocket_bucket import _HEARTBEAT, F, _require_websocket_connection, BfxWebsocketBucket
from .bfx_websocket_inputs import BfxWebsocketInputs
from ..handlers import PublicChannelsHandler, AuthenticatedChannelsHandler
from ..exceptions import WebsocketAuthenticationRequired, InvalidAuthenticationCredentials, EventNotSupported, OutdatedClientVersion
from ...utils.JSONEncoder import JSONEncoder
from ...utils.logger import ColoredLogger
def _require_websocket_authentication(function: F) -> F:
async def wrapper(self, *args, **kwargs):
if hasattr(self, "authentication") and self.authentication == False:
raise WebsocketAuthenticationRequired("To perform this action you need to authenticate using your API_KEY and API_SECRET.")
await _require_websocket_connection(function)(self, *args, **kwargs)
return cast(F, wrapper)
class BfxWebsocketClient(object):
VERSION = BfxWebsocketBucket.VERSION
MAXIMUM_CONNECTIONS_AMOUNT = 20
EVENTS = [
"open", "subscribed", "authenticated", "wss-error",
*PublicChannelsHandler.EVENTS,
*AuthenticatedChannelsHandler.EVENTS
]
def __init__(self, host, credentials = None, log_level = "INFO"):
self.websocket = None
self.host, self.credentials, self.event_emitter = host, credentials, AsyncIOEventEmitter()
self.inputs = BfxWebsocketInputs(handle_websocket_input=self.__handle_websocket_input)
self.handler = AuthenticatedChannelsHandler(event_emitter=self.event_emitter)
self.logger = ColoredLogger("BfxWebsocketClient", level=log_level)
self.event_emitter.add_listener("error",
lambda exception: self.logger.error(f"{type(exception).__name__}: {str(exception)}" + "\n" +
str().join(traceback.format_exception(type(exception), exception, exception.__traceback__))[:-1])
)
def run(self, connections = 5):
return asyncio.run(self.start(connections))
async def start(self, connections = 5):
if connections > BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT:
self.logger.warning(f"It is not safe to use more than {BfxWebsocketClient.MAXIMUM_CONNECTIONS_AMOUNT} buckets from the same " +
f"connection ({connections} in use), the server could momentarily block the client with <429 Too Many Requests>.")
self.on_open_events = [ asyncio.Event() for _ in range(connections) ]
self.buckets = [
BfxWebsocketBucket(self.host, self.event_emitter, self.on_open_events[index])
for index in range(connections)
]
tasks = [ bucket._connect(index) for index, bucket in enumerate(self.buckets) ]
tasks.append(self.__connect(self.credentials))
await asyncio.gather(*tasks)
async def __connect(self, credentials = None):
Reconnection = namedtuple("Reconnection", ["status", "attempts", "timestamp"])
reconnection, delay = Reconnection(status=False, attempts=0, timestamp=None), None
async def _connection():
nonlocal reconnection
async with websockets.connect(self.host) as websocket:
if reconnection.status == True:
self.logger.info(f"Reconnect attempt successful (attempt no.{reconnection.attempts}): The " +
f"client has been offline for a total of {datetime.now() - reconnection.timestamp} " +
f"(connection lost at: {reconnection.timestamp:%d-%m-%Y at %H:%M:%S}).")
reconnection = Reconnection(status=False, attempts=0, timestamp=None)
self.websocket, self.authentication = websocket, False
if await asyncio.gather(*[on_open_event.wait() for on_open_event in self.on_open_events]):
self.event_emitter.emit("open")
if self.credentials:
await self.__authenticate(**self.credentials)
async for message in websocket:
message = json.loads(message)
if isinstance(message, dict) and message["event"] == "info" and "version" in message:
if BfxWebsocketClient.VERSION != message["version"]:
raise OutdatedClientVersion(f"Mismatch between the client version and the server version. " +
f"Update the library to the latest version to continue (client version: {BfxWebsocketClient.VERSION}, " +
f"server version: {message['version']}).")
elif isinstance(message, dict) and message["event"] == "info" and message["code"] == 20051:
rcvd = websockets.frames.Close(code=1012, reason="Stop/Restart Websocket Server (please reconnect).")
raise websockets.ConnectionClosedError(rcvd=rcvd, sent=None)
elif isinstance(message, dict) and message["event"] == "auth":
if message["status"] == "OK":
self.event_emitter.emit("authenticated", message); self.authentication = True
else: raise InvalidAuthenticationCredentials("Cannot authenticate with given API-KEY and API-SECRET.")
elif isinstance(message, dict) and message["event"] == "error":
self.event_emitter.emit("wss-error", message["code"], message["msg"])
elif isinstance(message, list) and (chanId := message[0]) == 0 and message[1] != _HEARTBEAT:
self.handler.handle(message[1], message[2])
class _Delay:
BACKOFF_MIN, BACKOFF_MAX = 1.92, 60.0
BACKOFF_INITIAL = 5.0
def __init__(self, backoff_factor):
self.__backoff_factor = backoff_factor
self.__backoff_delay = _Delay.BACKOFF_MIN
self.__initial_delay = random.random() * _Delay.BACKOFF_INITIAL
def next(self):
backoff_delay = self.peek()
__backoff_delay = self.__backoff_delay * self.__backoff_factor
self.__backoff_delay = min(__backoff_delay, _Delay.BACKOFF_MAX)
return backoff_delay
def peek(self):
return (self.__backoff_delay == _Delay.BACKOFF_MIN) \
and self.__initial_delay or self.__backoff_delay
while True:
if reconnection.status == True:
await asyncio.sleep(delay.next())
try:
await _connection()
except (websockets.ConnectionClosedError, socket.gaierror) as error:
if isinstance(error, websockets.ConnectionClosedError) and (error.code == 1006 or error.code == 1012):
if error.code == 1006:
self.logger.error("Connection lost: no close frame received "
+ "or sent (1006). Attempting to reconnect...")
if error.code == 1012:
self.logger.info("WSS server is about to restart, reconnection "
+ "required (client received 20051). Attempt in progress...")
reconnection = Reconnection(status=True, attempts=1, timestamp=datetime.now());
delay = _Delay(backoff_factor=1.618)
elif isinstance(error, socket.gaierror) and reconnection.status == True:
self.logger.warning(f"Reconnection attempt no.{reconnection.attempts} has failed. "
+ f"Next reconnection attempt in ~{round(delay.peek()):.1f} seconds."
+ f"(at the moment the client has been offline for {datetime.now() - reconnection.timestamp})")
reconnection = reconnection._replace(attempts=reconnection.attempts + 1)
else: raise error
if reconnection.status == False:
break
async def __authenticate(self, API_KEY, API_SECRET, filter=None):
data = { "event": "auth", "filter": filter, "apiKey": API_KEY }
data["authNonce"] = int(time.time()) * 1000
data["authPayload"] = "AUTH" + str(data["authNonce"])
data["authSig"] = hmac.new(
API_SECRET.encode("utf8"),
data["authPayload"].encode("utf8"),
hashlib.sha384
).hexdigest()
await self.websocket.send(json.dumps(data))
async def subscribe(self, channel, **kwargs):
counters = [ len(bucket.pendings) + len(bucket.subscriptions) for bucket in self.buckets ]
index = counters.index(min(counters))
await self.buckets[index]._subscribe(channel, **kwargs)
async def unsubscribe(self, subId):
for bucket in self.buckets:
if (chanId := bucket._get_chan_id(subId)):
await bucket._unsubscribe(chanId=chanId)
async def close(self, code=1000, reason=str()):
if self.websocket != None and self.websocket.open == True:
await self.websocket.close(code=code, reason=reason)
for bucket in self.buckets:
await bucket._close(code=code, reason=reason)
@_require_websocket_authentication
async def notify(self, info, MESSAGE_ID=None, **kwargs):
await self.websocket.send(json.dumps([ 0, "n", MESSAGE_ID, { "type": "ucm-test", "info": info, **kwargs } ]))
@_require_websocket_authentication
async def __handle_websocket_input(self, input, data):
await self.websocket.send(json.dumps([ 0, input, None, data], cls=JSONEncoder))
def on(self, *events, callback = None):
for event in events:
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
if callback != None:
for event in events:
self.event_emitter.on(event, callback)
if callback == None:
def handler(function):
for event in events:
self.event_emitter.on(event, function)
return handler
def once(self, *events, callback = None):
for event in events:
if event not in BfxWebsocketClient.EVENTS:
raise EventNotSupported(f"Event <{event}> is not supported. To get a list of available events print BfxWebsocketClient.EVENTS")
if callback != None:
for event in events:
self.event_emitter.once(event, callback)
if callback == None:
def handler(function):
for event in events:
self.event_emitter.once(event, function)
return handler

24
bfxapi/websocket/_BfxWebsocketInputs.py → bfxapi/websocket/client/bfx_websocket_inputs.py
View File

@@ -2,19 +2,19 @@ from decimal import Decimal
from datetime import datetime
from typing import Union, Optional, List, Tuple
from .types import JSON
from .enums import OrderType, FundingOfferType
from .. enums import OrderType, FundingOfferType
from ... utils.JSONEncoder import JSON
class _BfxWebsocketInputs(object):
def __init__(self, __handle_websocket_input):
self.__handle_websocket_input = __handle_websocket_input
class BfxWebsocketInputs(object):
def __init__(self, handle_websocket_input):
self.handle_websocket_input = handle_websocket_input
async def submit_order(self, type: OrderType, symbol: str, amount: Union[Decimal, float, str],
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):
await self.__handle_websocket_input("on", {
await self.handle_websocket_input("on", {
"type": type, "symbol": symbol, "amount": amount,
"price": price, "lev": lev,
"price_trailing": price_trailing, "price_aux_limit": price_aux_limit, "price_oco_stop": price_oco_stop,
@@ -26,7 +26,7 @@ class _BfxWebsocketInputs(object):
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):
await self.__handle_websocket_input("ou", {
await self.handle_websocket_input("ou", {
"id": id, "amount": amount, "price": price,
"cid": cid, "cid_date": cid_date, "gid": gid,
"flags": flags, "lev": lev, "delta": delta,
@@ -34,12 +34,12 @@ class _BfxWebsocketInputs(object):
})
async def cancel_order(self, id: Optional[int] = None, cid: Optional[int] = None, cid_date: Optional[str] = None):
await self.__handle_websocket_input("oc", {
await self.handle_websocket_input("oc", {
"id": id, "cid": cid, "cid_date": cid_date
})
async 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):
await self.__handle_websocket_input("oc_multi", {
await self.handle_websocket_input("oc_multi", {
"ids": ids, "cids": cids, "gids": gids,
"all": int(all)
})
@@ -47,14 +47,14 @@ class _BfxWebsocketInputs(object):
async 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):
await self.__handle_websocket_input("fon", {
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):
await self.__handle_websocket_input("foc", { "id": id })
await self.handle_websocket_input("foc", { "id": id })
async def calc(self, *args: str):
await self.__handle_websocket_input("calc", list(map(lambda arg: [arg], args)))
await self.handle_websocket_input("calc", list(map(lambda arg: [arg], args)))

2
bfxapi/websocket/enums.py
View File

@@ -1,6 +1,6 @@
from .. enums import *
class Channels(str, Enum):
class Channel(str, Enum):
TICKER = "ticker"
TRADES = "trades"
BOOK = "book"

7
bfxapi/websocket/exceptions.py
View File

@@ -59,3 +59,10 @@ class InvalidAuthenticationCredentials(BfxWebsocketException):
"""
pass
class HandlerNotFound(BfxWebsocketException):
"""
This error indicates that a handler was not found for an incoming message.
"""
pass

4
bfxapi/websocket/handlers/__init__.py Normal file
View File

@@ -0,0 +1,4 @@
from .public_channels_handler import PublicChannelsHandler
from .authenticated_channels_handler import AuthenticatedChannelsHandler
NAME = "handlers"

69
bfxapi/websocket/handlers/authenticated_channels_handler.py Normal file
View File

@@ -0,0 +1,69 @@
from .. import serializers
from .. types import *
from .. exceptions import HandlerNotFound
class AuthenticatedChannelsHandler(object):
__abbreviations = {
"os": "order_snapshot", "on": "order_new", "ou": "order_update", "oc": "order_cancel",
"ps": "position_snapshot", "pn": "position_new", "pu": "position_update", "pc": "position_close",
"te": "trade_executed", "tu": "trade_execution_update",
"fos": "funding_offer_snapshot", "fon": "funding_offer_new", "fou": "funding_offer_update", "foc": "funding_offer_cancel",
"fcs": "funding_credit_snapshot", "fcn": "funding_credit_new", "fcu": "funding_credit_update", "fcc": "funding_credit_close",
"fls": "funding_loan_snapshot", "fln": "funding_loan_new", "flu": "funding_loan_update", "flc": "funding_loan_close",
"ws": "wallet_snapshot", "wu": "wallet_update",
"bu": "balance_update",
}
__serializers = {
("os", "on", "ou", "oc",): serializers.Order,
("ps", "pn", "pu", "pc",): serializers.Position,
("te", "tu"): serializers.Trade,
("fos", "fon", "fou", "foc",): serializers.FundingOffer,
("fcs", "fcn", "fcu", "fcc",): serializers.FundingCredit,
("fls", "fln", "flu", "flc",): serializers.FundingLoan,
("ws", "wu",): serializers.Wallet,
("bu",): serializers.Balance
}
EVENTS = [
"notification",
"on-req-notification", "ou-req-notification", "oc-req-notification",
"oc_multi-notification",
"fon-req-notification", "foc-req-notification",
*list(__abbreviations.values())
]
def __init__(self, event_emitter, strict = True):
self.event_emitter, self.strict = event_emitter, strict
def handle(self, type, stream):
if type == "n":
return self.__notification(stream)
for types, serializer in AuthenticatedChannelsHandler.__serializers.items():
if type in types:
event = AuthenticatedChannelsHandler.__abbreviations[type]
if all(isinstance(substream, list) for substream in stream):
return self.event_emitter.emit(event, [ serializer.parse(*substream) for substream in stream ])
return self.event_emitter.emit(event, serializer.parse(*stream))
if self.strict:
raise HandlerNotFound(f"No handler found for event of type <{type}>.")
def __notification(self, stream):
type, serializer = "notification", serializers._Notification(serializer=None)
if stream[1] == "on-req" or stream[1] == "ou-req" or stream[1] == "oc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.Order)
if stream[1] == "oc_multi-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.Order, iterate=True)
if stream[1] == "fon-req" or stream[1] == "foc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification(serializer=serializers.FundingOffer)
return self.event_emitter.emit(type, serializer.parse(*stream))

95
bfxapi/websocket/handlers.py → bfxapi/websocket/handlers/public_channels_handler.py
View File

@@ -1,10 +1,8 @@
from typing import List
from .. import serializers
from .types import *
from .. types import *
from . import serializers
from .enums import Channels
from .exceptions import BfxWebsocketException
from .. exceptions import HandlerNotFound
class PublicChannelsHandler(object):
EVENTS = [
@@ -15,22 +13,25 @@ class PublicChannelsHandler(object):
"derivatives_status_update",
]
def __init__(self, event_emitter):
self.event_emitter = event_emitter
def __init__(self, event_emitter, strict = True):
self.event_emitter, self.strict = event_emitter, strict
self.__handlers = {
Channels.TICKER: self.__ticker_channel_handler,
Channels.TRADES: self.__trades_channel_handler,
Channels.BOOK: self.__book_channel_handler,
Channels.CANDLES: self.__candles_channel_handler,
Channels.STATUS: self.__status_channel_handler
"ticker": self.__ticker_channel_handler,
"trades": self.__trades_channel_handler,
"book": self.__book_channel_handler,
"candles": self.__candles_channel_handler,
"status": self.__status_channel_handler
}
def handle(self, subscription, *stream):
_clear = lambda dictionary, *args: { key: value for key, value in dictionary.items() if key not in args }
if channel := subscription["channel"] or channel in self.__handlers.keys():
return self.__handlers[channel](_clear(subscription, "event", "channel", "subId"), *stream)
if (channel := subscription["channel"]) and channel in self.__handlers.keys():
return self.__handlers[channel](_clear(subscription, "event", "channel", "chanId"), *stream)
if self.strict:
raise HandlerNotFound(f"No handler found for channel <{subscription['channel']}>.")
def __ticker_channel_handler(self, subscription, *stream):
if subscription["symbol"].startswith("t"):
@@ -48,7 +49,7 @@ class PublicChannelsHandler(object):
)
def __trades_channel_handler(self, subscription, *stream):
if type := stream[0] or type in [ "te", "tu", "fte", "ftu" ]:
if (type := stream[0]) and type in [ "te", "tu", "fte", "ftu" ]:
if subscription["symbol"].startswith("t"):
return self.event_emitter.emit(
{ "te": "t_trade_executed", "tu": "t_trade_execution_update" }[type],
@@ -118,67 +119,3 @@ class PublicChannelsHandler(object):
subscription,
serializers.DerivativesStatus.parse(*stream[0])
)
class AuthenticatedChannelsHandler(object):
__abbreviations = {
"os": "order_snapshot", "on": "order_new", "ou": "order_update", "oc": "order_cancel",
"ps": "position_snapshot", "pn": "position_new", "pu": "position_update", "pc": "position_close",
"te": "trade_executed", "tu": "trade_execution_update",
"fos": "funding_offer_snapshot", "fon": "funding_offer_new", "fou": "funding_offer_update", "foc": "funding_offer_cancel",
"fcs": "funding_credit_snapshot", "fcn": "funding_credit_new", "fcu": "funding_credit_update", "fcc": "funding_credit_close",
"fls": "funding_loan_snapshot", "fln": "funding_loan_new", "flu": "funding_loan_update", "flc": "funding_loan_close",
"ws": "wallet_snapshot", "wu": "wallet_update",
"bu": "balance_update",
}
__serializers = {
("os", "on", "ou", "oc",): serializers.Order,
("ps", "pn", "pu", "pc",): serializers.Position,
("te", "tu"): serializers.Trade,
("fos", "fon", "fou", "foc",): serializers.FundingOffer,
("fcs", "fcn", "fcu", "fcc",): serializers.FundingCredit,
("fls", "fln", "flu", "flc",): serializers.FundingLoan,
("ws", "wu",): serializers.Wallet,
("bu",): serializers.Balance
}
EVENTS = [
"notification",
"on-req-notification", "ou-req-notification", "oc-req-notification",
"oc_multi-notification",
"fon-req-notification", "foc-req-notification",
*list(__abbreviations.values())
]
def __init__(self, event_emitter, strict = False):
self.event_emitter, self.strict = event_emitter, strict
def handle(self, type, stream):
if type == "n":
return self.__notification(stream)
for types, serializer in AuthenticatedChannelsHandler.__serializers.items():
if type in types:
event = AuthenticatedChannelsHandler.__abbreviations[type]
if all(isinstance(substream, list) for substream in stream):
return self.event_emitter.emit(event, [ serializer.parse(*substream) for substream in stream ])
return self.event_emitter.emit(event, serializer.parse(*stream))
if self.strict == True:
raise BfxWebsocketException(f"Event of type <{type}> not found in self.__handlers.")
def __notification(self, stream):
type, serializer = "notification", serializers._Notification(serializer=None)
if stream[1] == "on-req" or stream[1] == "ou-req" or stream[1] == "oc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification[Order](serializer=serializers.Order)
if stream[1] == "oc_multi-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification[List[Order]](serializer=serializers.Order, iterate=True)
if stream[1] == "fon-req" or stream[1] == "foc-req":
type, serializer = f"{stream[1]}-notification", serializers._Notification[FundingOffer](serializer=serializers.FundingOffer)
return self.event_emitter.emit(type, serializer.parse(*stream))

18
bfxapi/websocket/subscriptions.py
View File

@@ -1,6 +1,8 @@
from typing import TypedDict, Optional
from typing import TypedDict, Union, Literal, Optional
__all__ = [
"Subscription",
"Ticker",
"Trades",
"Book",
@@ -8,18 +10,22 @@ __all__ = [
"Status"
]
_Header = TypedDict("_Header", { "event": Literal["subscribed"], "channel": str, "chanId": int })
Subscription = Union["Ticker", "Trades", "Book", "Candles", "Status"]
class Ticker(TypedDict):
chanId: int; symbol: str
subId: str; symbol: str
pair: Optional[str]
currency: Optional[str]
class Trades(TypedDict):
chanId: int; symbol: str
subId: str; symbol: str
pair: Optional[str]
currency: Optional[str]
class Book(TypedDict):
chanId: int
subId: str
symbol: str
prec: str
freq: str
@@ -27,9 +33,9 @@ class Book(TypedDict):
pair: str
class Candles(TypedDict):
chanId: int
subId: str
key: str
class Status(TypedDict):
chanId: int
subId: str
key: str

2
bfxapi/websocket/types.py
View File

@@ -1,4 +1,4 @@
from typing import Type, Tuple, List, Dict, TypedDict, Union, Optional, Any
from typing import Optional
from dataclasses import dataclass

4
examples/rest/claim_position.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/create_funding_offer.py
View File

@@ -2,11 +2,11 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
from bfxapi.enums import FundingOfferType, Flag
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/create_order.py
View File

@@ -1,11 +1,11 @@
# python -c "import examples.rest.create_order"
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
from bfxapi.enums import OrderType, Flag
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/derivatives.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/extra_calcs.py
View File

@@ -1,9 +1,9 @@
# python -c "import examples.rest.extra_calcs"
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST
REST_HOST=REST_HOST
)
t_symbol_response = bfx.rest.public.get_trading_market_average_price(

4
examples/rest/funding_auto_renew.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/get_authenticated_data.py
View File

@@ -3,10 +3,10 @@
import os
import time
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/get_candles_hist.py
View File

@@ -1,9 +1,9 @@
# python -c "import examples.rest.get_candles_hist"
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST
REST_HOST=REST_HOST
)
print(f"Candles: {bfx.rest.public.get_candles_hist(symbol='tBTCUSD')}")

4
examples/rest/get_funding_info.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/get_funding_trades_history.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/get_liquidations.py
View File

@@ -2,10 +2,10 @@
import time
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST
REST_HOST=REST_HOST
)
now = int(round(time.time() * 1000))

4
examples/rest/get_positions.py
View File

@@ -3,10 +3,10 @@
import os
import time
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/get_public_data.py
View File

@@ -2,10 +2,10 @@
import time
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST
REST_HOST=REST_HOST
)
now = int(round(time.time() * 1000))

4
examples/rest/get_pulse_data.py
View File

@@ -2,10 +2,10 @@
import time
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST
REST_HOST=REST_HOST
)
now = int(round(time.time() * 1000))

4
examples/rest/increase_position.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/keep_taken_funding.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/merchant.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/return_taken_funding.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/rest/transfer_wallet.py
View File

@@ -2,10 +2,10 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, REST_HOST
bfx = Client(
REST_HOST=Constants.REST_HOST,
REST_HOST=REST_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

4
examples/websocket/create_order.py
View File

@@ -2,12 +2,12 @@
import os
from bfxapi.client import Client, Constants
from bfxapi.client import Client, WSS_HOST
from bfxapi.websocket.enums import Error, OrderType
from bfxapi.websocket.types import Notification, Order
bfx = Client(
WSS_HOST=Constants.WSS_HOST,
WSS_HOST=WSS_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)

23
examples/websocket/derivatives_status.py Normal file
View File

@@ -0,0 +1,23 @@
# python -c "import examples.websocket.derivatives_status"
from bfxapi import Client, PUB_WSS_HOST
from bfxapi.websocket.enums import Error, Channel
from bfxapi.websocket.types import DerivativesStatus
from bfxapi.websocket import subscriptions
bfx = Client(WSS_HOST=PUB_WSS_HOST)
@bfx.wss.on("derivatives_status_update")
def on_derivatives_status_update(subscription: subscriptions.Status, data: DerivativesStatus):
print(f"{subscription}:", data)
@bfx.wss.on("wss-error")
def on_wss_error(code: Error, msg: str):
print(code, msg)
@bfx.wss.once("open")
async def open():
await bfx.wss.subscribe(Channel.STATUS, key="deriv:tBTCF0:USTF0")
bfx.wss.run()

8
examples/websocket/order_book.py
View File

@@ -4,10 +4,10 @@ from collections import OrderedDict
from typing import List
from bfxapi import Client, Constants
from bfxapi import Client, PUB_WSS_HOST
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels, Error
from bfxapi.websocket.enums import Channel, Error
from bfxapi.websocket.types import TradingPairBook
class OrderBook(object):
@@ -38,7 +38,7 @@ SYMBOLS = [ "tBTCUSD", "tLTCUSD", "tLTCBTC", "tETHUSD", "tETHBTC" ]
order_book = OrderBook(symbols=SYMBOLS)
bfx = Client(WSS_HOST=Constants.PUB_WSS_HOST)
bfx = Client(WSS_HOST=PUB_WSS_HOST)
@bfx.wss.on("wss-error")
def on_wss_error(code: Error, msg: str):
@@ -47,7 +47,7 @@ def on_wss_error(code: Error, msg: str):
@bfx.wss.on("open")
async def on_open():
for symbol in SYMBOLS:
await bfx.wss.subscribe(Channels.BOOK, symbol=symbol)
await bfx.wss.subscribe(Channel.BOOK, symbol=symbol)
@bfx.wss.on("subscribed")
def on_subscribed(subscription):

8
examples/websocket/raw_order_book.py
View File

@@ -4,10 +4,10 @@ from collections import OrderedDict
from typing import List
from bfxapi import Client, Constants
from bfxapi import Client, PUB_WSS_HOST
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels, Error
from bfxapi.websocket.enums import Channel, Error
from bfxapi.websocket.types import TradingPairRawBook
class RawOrderBook(object):
@@ -38,7 +38,7 @@ SYMBOLS = [ "tBTCUSD", "tLTCUSD", "tLTCBTC", "tETHUSD", "tETHBTC" ]
raw_order_book = RawOrderBook(symbols=SYMBOLS)
bfx = Client(WSS_HOST=Constants.PUB_WSS_HOST)
bfx = Client(WSS_HOST=PUB_WSS_HOST)
@bfx.wss.on("wss-error")
def on_wss_error(code: Error, msg: str):
@@ -47,7 +47,7 @@ def on_wss_error(code: Error, msg: str):
@bfx.wss.on("open")
async def on_open():
for symbol in SYMBOLS:
await bfx.wss.subscribe(Channels.BOOK, symbol=symbol, prec="R0")
await bfx.wss.subscribe(Channel.BOOK, symbol=symbol, prec="R0")
@bfx.wss.on("subscribed")
def on_subscribed(subscription):

10
examples/websocket/ticker.py
View File

@@ -1,21 +1,21 @@
# python -c "import examples.websocket.ticker"
from bfxapi import Client, Constants
from bfxapi import Client, PUB_WSS_HOST
from bfxapi.websocket import subscriptions
from bfxapi.websocket.enums import Channels
from bfxapi.websocket.enums import Channel
from bfxapi.websocket.types import TradingPairTicker
bfx = Client(WSS_HOST=Constants.PUB_WSS_HOST)
bfx = Client(WSS_HOST=PUB_WSS_HOST)
@bfx.wss.on("t_ticker_update")
def on_t_ticker_update(subscription: subscriptions.Ticker, data: TradingPairTicker):
print(f"Subscription with channel ID: {subscription['chanId']}")
print(f"Subscription with subId: {subscription['subId']}")
print(f"Data: {data}")
@bfx.wss.once("open")
async def open():
await bfx.wss.subscribe(Channels.TICKER, symbol="tBTCUSD")
await bfx.wss.subscribe(Channel.TICKER, symbol="tBTCUSD")
bfx.wss.run()

29
examples/websocket/trades.py Normal file
View File

@@ -0,0 +1,29 @@
# python -c "import examples.websocket.trades"
from bfxapi import Client, PUB_WSS_HOST
from bfxapi.websocket.enums import Error, Channel
from bfxapi.websocket.types import Candle, TradingPairTrade
from bfxapi.websocket import subscriptions
bfx = Client(WSS_HOST=PUB_WSS_HOST)
@bfx.wss.on("candles_update")
def on_candles_update(subscription: subscriptions.Candles, candle: Candle):
print(f"New candle: {candle}")
@bfx.wss.on("t_trade_executed")
def on_t_trade_executed(subscription: subscriptions.Trades, trade: TradingPairTrade):
print(f"New trade: {trade}")
@bfx.wss.on("wss-error")
def on_wss_error(code: Error, msg: str):
print(code, msg)
@bfx.wss.once("open")
async def open():
await bfx.wss.subscribe(Channel.CANDLES, key="trade:1m:tBTCUSD")
await bfx.wss.subscribe(Channel.TRADES, symbol="tBTCUSD")
bfx.wss.run()

30
examples/websocket/wallet_balance.py Normal file
View File

@@ -0,0 +1,30 @@
# python -c "import examples.websocket.wallet_balance"
import os
from typing import List
from bfxapi import Client, WSS_HOST
from bfxapi.websocket.enums import Error
from bfxapi.websocket.types import Wallet
bfx = Client(
WSS_HOST=WSS_HOST,
API_KEY=os.getenv("BFX_API_KEY"),
API_SECRET=os.getenv("BFX_API_SECRET")
)
@bfx.wss.on("wallet_snapshot")
def log_snapshot(wallets: List[Wallet]):
for wallet in wallets:
print(f"Balance: {wallet}")
@bfx.wss.on("wallet_update")
def log_update(wallet: Wallet):
print(f"Balance update: {wallet}")
@bfx.wss.on("wss-error")
def on_wss_error(code: Error, msg: str):
print(code, msg)
bfx.wss.run()

39
setup.py
View File

@@ -2,18 +2,36 @@ from distutils.core import setup
setup(
name="bitfinex-api-py",
version="3.0.0",
packages=[
"bfxapi", "bfxapi.utils",
"bfxapi.websocket",
"bfxapi.rest", "bfxapi.rest.endpoints", "bfxapi.rest.middleware",
],
version="3.0.0b1",
description="Official Bitfinex Python API",
long_description="A Python reference implementation of the Bitfinex API for both REST and websocket interaction",
long_description_content_type="text/markdown",
url="https://github.com/bitfinexcom/bitfinex-api-py",
license="OSI Approved :: Apache Software License",
author="Bitfinex",
author_email="support@bitfinex.com",
description="Official Bitfinex Python API",
license="Apache-2.0",
classifiers=[
"Development Status :: 4 - Beta",
"Intended Audience :: Developers",
"Topic :: Software Development :: Build Tools",
"License :: OSI Approved :: Apache-2.0",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
],
keywords="bitfinex,api,trading",
project_urls={
"Bug Reports": "https://github.com/bitfinexcom/bitfinex-api-py/issues",
"Source": "https://github.com/bitfinexcom/bitfinex-api-py",
},
packages=[
"bfxapi", "bfxapi.utils",
"bfxapi.websocket", "bfxapi.websocket.client", "bfxapi.websocket.handlers",
"bfxapi.rest", "bfxapi.rest.endpoints", "bfxapi.rest.middleware",
],
install_requires=[
"certifi~=2022.12.7",
"charset-normalizer~=2.1.1",
@@ -29,8 +47,5 @@ setup(
"urllib3~=1.26.13",
"websockets~=10.4",
],
project_urls={
"Bug Reports": "https://github.com/bitfinexcom/bitfinex-api-py/issues",
"Source": "https://github.com/bitfinexcom/bitfinex-api-py",
}
python_requires=">=3.8"
)