Async Support

examples/async_direct_session.py

@@ -0,0 +1,59 @@
+from pybit.unified_trading import AsyncHTTP
+import asyncio
+
+
+BYBIT_API_KEY = "api_key"
+BYBIT_API_SECRET = "api_secret"
+TESTNET = True  # True means your API keys were generated on testnet.bybit.com
+
+
+async def main():
+    session = AsyncHTTP(api_key=BYBIT_API_KEY,
+                        api_secret=BYBIT_API_SECRET,
+                        testnet=True)
+
+    # Place order
+
+    response = await session.place_order(
+        category="spot",
+        symbol="ETHUSDT",
+        side="Sell",
+        orderType="Market",
+        qty="0.1",
+        timeInForce="GTC",
+    )
+
+    # Example to cancel orders
+
+    response = await session.get_open_orders(
+        category="linear",
+        symbol="BTCUSDT",
+    )
+
+    orders = response["result"]["list"]
+
+    for order in orders:
+        if order["orderStatus"] == "Untriggered":
+            await session.cancel_order(
+                category="linear",
+                symbol=order["symbol"],
+                orderId=order["orderId"],
+            )
+
+    # Batch cancel orders
+
+    orders_to_cancel = [
+        {"category": "option", "symbol": o["symbol"], "orderId": o["orderId"]}
+        for o in response["result"]["list"]
+        if o["orderStatus"] == "New"
+    ]
+
+    response = await session.cancel_batch_order(
+        category="option",
+        request=orders_to_cancel,
+    )
+
+
+loop = asyncio.new_event_loop()
+loop.run_until_complete(main())
+

examples/async_websocket_example_quickstart.py

@@ -0,0 +1,33 @@
+import asyncio
+from pybit.asyncio.websocket import (
+    AsyncWebsocket,
+    WSState
+)
+
+
+async def main():
+    ws = AsyncWebsocket(
+        testnet=True,
+        api_key="api_key",
+        api_secret="api_secret"
+    )
+    private_session = ws.order_stream(category="linear")
+    async with private_session as active_session:
+        while True:
+            if active_session.ws_state == WSState.EXITING:
+                break
+            response = await active_session.recv()
+            print(response)
+            break
+
+    public_session = ws.kline_stream(symbols=["kline.60.BTCUSDT"], channel_type="linear")
+    async with public_session as active_session:
+        while True:
+            if active_session.ws_state == WSState.EXITING:
+                break
+            response = await active_session.recv()
+            print(response)
+
+
+loop = asyncio.new_event_loop()
+loop.run_until_complete(main())

pybit/_http_manager/__init__.py

@@ -0,0 +1 @@
+from ._http_manager import _V5HTTPManager

pybit/_http_manager/_auth.py

@@ -0,0 +1,79 @@
+from dataclasses import dataclass, field
+import hmac
+import hashlib
+from Crypto.Hash import SHA256
+from Crypto.PublicKey import RSA
+from Crypto.Signature import PKCS1_v1_5
+import base64
+
+from pybit import _helpers
+
+
+def generate_signature(use_rsa_authentication, secret, param_str):
+    def generate_hmac():
+        hash = hmac.new(
+            bytes(secret, "utf-8"),
+            param_str.encode("utf-8"),
+            hashlib.sha256,
+        )
+        return hash.hexdigest()
+
+    def generate_rsa():
+        hash = SHA256.new(param_str.encode("utf-8"))
+        encoded_signature = base64.b64encode(
+            PKCS1_v1_5.new(RSA.importKey(secret)).sign(
+                hash
+            )
+        )
+        return encoded_signature.decode()
+
+    if not use_rsa_authentication:
+        return generate_hmac()
+    else:
+        return generate_rsa()
+
+
+@dataclass
+class AuthService:
+    api_key: str = field(default=None)
+    api_secret: str = field(default=None)
+    rsa_authentication: str = field(default=False)
+
+    def _auth(self, payload, recv_window, timestamp):
+        """
+        Prepares authentication signature per Bybit API specifications.
+        """
+
+        if self.api_key is None or self.api_secret is None:
+            raise PermissionError("Authenticated endpoints require keys.")
+
+        param_str = str(timestamp) + self.api_key + str(recv_window) + payload
+
+        return generate_signature(
+            self.rsa_authentication, self.api_secret, param_str
+        )
+
+    def _prepare_auth_headers(self, recv_window, req_params) -> dict:
+        # Prepare signature.
+        timestamp = _helpers.generate_timestamp()
+        signature = self._auth(
+            payload=req_params,
+            recv_window=recv_window,
+            timestamp=timestamp,
+        )
+        return {
+            "Content-Type": "application/json",
+            "X-BAPI-API-KEY": self.api_key,
+            "X-BAPI-SIGN": signature,
+            "X-BAPI-SIGN-TYPE": "2",
+            "X-BAPI-TIMESTAMP": str(timestamp),
+            "X-BAPI-RECV-WINDOW": str(recv_window),
+        }
+
+    def _change_floating_numbers_for_auth_signature(self, query):
+        # Bug fix: change floating whole numbers to integers to prevent
+        # auth signature errors.
+        if query is not None:
+            for i in query.keys():
+                if isinstance(query[i], float) and query[i] == int(query[i]):
+                    query[i] = int(query[i])

pybit/_http_manager/_http_helpers.py

@@ -0,0 +1,26 @@
+import logging
+
+from datetime import datetime as dt
+
+from pybit import _helpers
+
+
+def set_logger_handler(logger, logging_level):
+    if len(logging.root.handlers) == 0:
+        # no handler on root logger set -> we add handler just for this logger to not mess with custom logic from outside
+        handler = logging.StreamHandler()
+        handler.setFormatter(
+            logging.Formatter(
+                fmt="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+                datefmt="%Y-%m-%d %H:%M:%S",
+            )
+        )
+        handler.setLevel(logging_level)
+        logger.addHandler(handler)
+
+
+def calculate_rate_limit_delay_time(x_bapi_limit_reset_timestamp: int):
+    limit_reset_str = dt.fromtimestamp(x_bapi_limit_reset_timestamp / 10 ** 3).strftime(
+        "%H:%M:%S.%f")[:-3]
+    delay_time = (int(x_bapi_limit_reset_timestamp) - _helpers.generate_timestamp()) / 10 ** 3
+    return delay_time, limit_reset_str