Merge pull request hummingbot#6584 from cardosofede/fix/funding_fee_e…

…ndpoint_binance_perps

Fix: Order cancellation, status and revamp in Binance Perps

hummingbot/connector/derivative/binance_perpetual/binance_perpetual_auth.py

@@ -1,5 +1,6 @@
 import hashlib
 import hmac
+import json
 from collections import OrderedDict
 from typing import Any, Dict
 from urllib.parse import urlencode
@@ -26,11 +27,11 @@ def generate_signature_from_payload(self, payload: str) -> str:
 
     async def rest_authenticate(self, request: RESTRequest) -> RESTRequest:
         if request.method == RESTMethod.POST:
-            request.data = self.add_auth_to_params(request.data)
+            request.data = self.add_auth_to_params(params=json.loads(request.data))
         else:
             request.params = self.add_auth_to_params(request.params)
 
-        request.headers = {"X-MBX-APIKEY": self._api_key}
+        request.headers = self.header_for_authentication()
 
         return request
 
@@ -48,3 +49,6 @@ def add_auth_to_params(self,
         request_params["signature"] = self.generate_signature_from_payload(payload=payload)
 
         return request_params
+
+    def header_for_authentication(self) -> Dict[str, str]:
+        return {"X-MBX-APIKEY": self._api_key}

hummingbot/connector/derivative/binance_perpetual/binance_perpetual_constants.py

@@ -17,41 +17,38 @@
 PUBLIC_WS_ENDPOINT = "stream"
 PRIVATE_WS_ENDPOINT = "ws"
 
-API_VERSION = "v1"
-API_VERSION_V2 = "v2"
-
 TIME_IN_FORCE_GTC = "GTC"  # Good till cancelled
 TIME_IN_FORCE_GTX = "GTX"  # Good Till Crossing
 TIME_IN_FORCE_IOC = "IOC"  # Immediate or cancel
 TIME_IN_FORCE_FOK = "FOK"  # Fill or kill
 
 # Public API v1 Endpoints
-SNAPSHOT_REST_URL = "/depth"
-TICKER_PRICE_URL = "/ticker/bookTicker"
-TICKER_PRICE_CHANGE_URL = "/ticker/24hr"
-EXCHANGE_INFO_URL = "/exchangeInfo"
-RECENT_TRADES_URL = "/trades"
-PING_URL = "/ping"
-MARK_PRICE_URL = "/premiumIndex"
-SERVER_TIME_PATH_URL = "/time"
+SNAPSHOT_REST_URL = "v1/depth"
+TICKER_PRICE_URL = "v1/ticker/bookTicker"
+TICKER_PRICE_CHANGE_URL = "v1/ticker/24hr"
+EXCHANGE_INFO_URL = "v1/exchangeInfo"
+RECENT_TRADES_URL = "v1/trades"
+PING_URL = "v1/ping"
+MARK_PRICE_URL = "v1/premiumIndex"
+SERVER_TIME_PATH_URL = "v1/time"
 
 # Private API v1 Endpoints
-ORDER_URL = "/order"
-CANCEL_ALL_OPEN_ORDERS_URL = "/allOpenOrders"
-ACCOUNT_TRADE_LIST_URL = "/userTrades"
-SET_LEVERAGE_URL = "/leverage"
-GET_INCOME_HISTORY_URL = "/income"
-CHANGE_POSITION_MODE_URL = "/positionSide/dual"
+ORDER_URL = "v1/order"
+CANCEL_ALL_OPEN_ORDERS_URL = "v1/allOpenOrders"
+ACCOUNT_TRADE_LIST_URL = "v1/userTrades"
+SET_LEVERAGE_URL = "v1/leverage"
+GET_INCOME_HISTORY_URL = "v1/income"
+CHANGE_POSITION_MODE_URL = "v1/positionSide/dual"
 
 POST_POSITION_MODE_LIMIT_ID = f"POST{CHANGE_POSITION_MODE_URL}"
 GET_POSITION_MODE_LIMIT_ID = f"GET{CHANGE_POSITION_MODE_URL}"
 
 # Private API v2 Endpoints
-ACCOUNT_INFO_URL = "/account"
-POSITION_INFORMATION_URL = "/positionRisk"
+ACCOUNT_INFO_URL = "v2/account"
+POSITION_INFORMATION_URL = "v2/positionRisk"
 
 # Private API Endpoints
-BINANCE_USER_STREAM_ENDPOINT = "/listenKey"
+BINANCE_USER_STREAM_ENDPOINT = "v1/listenKey"
 
 # Funding Settlement Time Span
 FUNDING_SETTLEMENT_DURATION = (0, 30)  # seconds before snapshot, seconds after snapshot
@@ -129,3 +126,8 @@
     RateLimit(limit_id=MARK_PRICE_URL, limit=MAX_REQUEST, time_interval=ONE_MINUTE, weight=1,
               linked_limits=[LinkedLimitWeightPair(REQUEST_WEIGHT, weight=1)]),
 ]
+
+ORDER_NOT_EXIST_ERROR_CODE = -2013
+ORDER_NOT_EXIST_MESSAGE = "Order does not exist"
+UNKNOWN_ORDER_ERROR_CODE = -2011
+UNKNOWN_ORDER_MESSAGE = "Unknown order sent"

hummingbot/connector/derivative/binance_perpetual/binance_perpetual_derivative.py

@@ -30,7 +30,6 @@
 from hummingbot.core.data_type.user_stream_tracker_data_source import UserStreamTrackerDataSource
 from hummingbot.core.utils.async_utils import safe_gather
 from hummingbot.core.utils.estimate_fee import build_trade_fee
-from hummingbot.core.web_assistant.connections.data_types import RESTMethod, RESTRequest
 from hummingbot.core.web_assistant.web_assistants_factory import WebAssistantsFactory
 
 if TYPE_CHECKING:
@@ -114,7 +113,7 @@ def is_trading_required(self) -> bool:
 
     @property
     def funding_fee_poll_interval(self) -> int:
-        return 120
+        return 600
 
     def supported_order_types(self) -> List[OrderType]:
         """
@@ -143,18 +142,14 @@ def _is_request_exception_related_to_time_synchronizer(self, request_exception:
         return is_time_synchronizer_related
 
     def _is_order_not_found_during_status_update_error(self, status_update_exception: Exception) -> bool:
-        # TODO: implement this method correctly for the connector
-        # The default implementation was added when the functionality to detect not found orders was introduced in the
-        # ExchangePyBase class. Also fix the unit test test_lost_order_removed_if_not_found_during_order_status_update
-        # when replacing the dummy implementation
-        return False
+        return str(CONSTANTS.ORDER_NOT_EXIST_ERROR_CODE) in str(
+            status_update_exception
+        ) and CONSTANTS.ORDER_NOT_EXIST_MESSAGE in str(status_update_exception)
 
     def _is_order_not_found_during_cancelation_error(self, cancelation_exception: Exception) -> bool:
-        # TODO: implement this method correctly for the connector
-        # The default implementation was added when the functionality to detect not found orders was introduced in the
-        # ExchangePyBase class. Also fix the unit test test_cancel_order_not_found_in_the_exchange when replacing the
-        # dummy implementation
-        return False
+        return str(CONSTANTS.UNKNOWN_ORDER_ERROR_CODE) in str(
+            cancelation_exception
+        ) and CONSTANTS.UNKNOWN_ORDER_MESSAGE in str(cancelation_exception)
 
     def _create_web_assistants_factory(self) -> WebAssistantsFactory:
         return web_utils.build_api_factory(
@@ -581,10 +576,8 @@ async def _update_balances(self):
         local_asset_names = set(self._account_balances.keys())
         remote_asset_names = set()
 
-        account_info = await self._api_request(path_url=CONSTANTS.ACCOUNT_INFO_URL,
-                                               is_auth_required=True,
-                                               api_version=CONSTANTS.API_VERSION_V2,
-                                               )
+        account_info = await self._api_get(path_url=CONSTANTS.ACCOUNT_INFO_URL,
+                                           is_auth_required=True)
         assets = account_info.get("assets")
         for asset in assets:
             asset_name = asset.get("asset")
@@ -600,10 +593,8 @@ async def _update_balances(self):
             del self._account_balances[asset_name]
 
     async def _update_positions(self):
-        positions = await self._api_request(path_url=CONSTANTS.POSITION_INFORMATION_URL,
-                                            is_auth_required=True,
-                                            api_version=CONSTANTS.API_VERSION_V2,
-                                            )
+        positions = await self._api_get(path_url=CONSTANTS.POSITION_INFORMATION_URL,
+                                        is_auth_required=True)
         for position in positions:
             trading_pair = position.get("symbol")
             try:
@@ -639,7 +630,7 @@ async def _update_order_fills_from_trades(self):
                 trading_pairs_to_order_map[order.trading_pair][order.exchange_order_id] = order
             trading_pairs = list(trading_pairs_to_order_map.keys())
             tasks = [
-                self._api_request(
+                self._api_get(
                     path_url=CONSTANTS.ACCOUNT_TRADE_LIST_URL,
                     params={"symbol": await self.exchange_symbol_associated_to_pair(trading_pair=trading_pair)},
                     is_auth_required=True,
@@ -693,13 +684,12 @@ async def _update_order_status(self):
         if current_tick > last_tick and len(self._order_tracker.active_orders) > 0:
             tracked_orders = list(self._order_tracker.active_orders.values())
             tasks = [
-                self._api_request(
+                self._api_get(
                     path_url=CONSTANTS.ORDER_URL,
                     params={
                         "symbol": await self.exchange_symbol_associated_to_pair(trading_pair=order.trading_pair),
                         "origClientOrderId": order.client_order_id
                     },
-                    method=RESTMethod.GET,
                     is_auth_required=True,
                     return_err=True,
                 )
@@ -735,14 +725,13 @@ async def _update_order_status(self):
     async def _get_position_mode(self) -> Optional[PositionMode]:
         # To-do: ensure there's no active order or contract before changing position mode
         if self._position_mode is None:
-            response = await self._api_request(
-                method=RESTMethod.GET,
+            response = await self._api_get(
                 path_url=CONSTANTS.CHANGE_POSITION_MODE_URL,
                 is_auth_required=True,
                 limit_id=CONSTANTS.GET_POSITION_MODE_LIMIT_ID,
                 return_err=True
             )
-            self._position_mode = PositionMode.HEDGE if response["dualSidePosition"] else PositionMode.ONEWAY
+            self._position_mode = PositionMode.HEDGE if response.get("dualSidePosition") else PositionMode.ONEWAY
 
         return self._position_mode
 
@@ -754,8 +743,7 @@ async def _trading_pair_position_mode_set(self, mode: PositionMode, trading_pair
             params = {
                 "dualSidePosition": mode.value
             }
-            response = await self._api_request(
-                method=RESTMethod.POST,
+            response = await self._api_post(
                 path_url=CONSTANTS.CHANGE_POSITION_MODE_URL,
                 data=params,
                 is_auth_required=True,
@@ -771,10 +759,9 @@ async def _trading_pair_position_mode_set(self, mode: PositionMode, trading_pair
     async def _set_trading_pair_leverage(self, trading_pair: str, leverage: int) -> Tuple[bool, str]:
         symbol = await self.exchange_symbol_associated_to_pair(trading_pair)
         params = {'symbol': symbol, 'leverage': leverage}
-        set_leverage = await self._api_request(
+        set_leverage = await self._api_post(
             path_url=CONSTANTS.SET_LEVERAGE_URL,
             data=params,
-            method=RESTMethod.POST,
             is_auth_required=True,
         )
         success = False
@@ -787,22 +774,19 @@ async def _set_trading_pair_leverage(self, trading_pair: str, leverage: int) ->
 
     async def _fetch_last_fee_payment(self, trading_pair: str) -> Tuple[int, Decimal, Decimal]:
         exchange_symbol = await self.exchange_symbol_associated_to_pair(trading_pair)
-        payment_response = await self._api_request(
+        payment_response = await self._api_get(
             path_url=CONSTANTS.GET_INCOME_HISTORY_URL,
             params={
                 "symbol": exchange_symbol,
                 "incomeType": "FUNDING_FEE",
-                "limit": 10,
             },
-            method=RESTMethod.GET,
             is_auth_required=True,
         )
-        funding_info_response = await self._api_request(
+        funding_info_response = await self._api_get(
             path_url=CONSTANTS.MARK_PRICE_URL,
             params={
                 "symbol": exchange_symbol,
             },
-            method=RESTMethod.GET,
         )
         sorted_payment_response = sorted(payment_response, key=lambda a: a.get('time', 0), reverse=True)
         if len(sorted_payment_response) < 1:
@@ -817,54 +801,3 @@ async def _fetch_last_fee_payment(self, trading_pair: str) -> Tuple[int, Decimal
         else:
             timestamp, funding_rate, payment = 0, Decimal("-1"), Decimal("-1")
         return timestamp, funding_rate, payment
-
-    async def _api_request(
-            self,
-            path_url,
-            overwrite_url: Optional[str] = None,
-            method: RESTMethod = RESTMethod.GET,
-            params: Optional[Dict[str, Any]] = None,
-            data: Optional[Dict[str, Any]] = None,
-            is_auth_required: bool = False,
-            return_err: bool = False,
-            api_version: str = CONSTANTS.API_VERSION,
-            limit_id: Optional[str] = None,
-            **kwargs,
-    ) -> Dict[str, Any]:
-        last_exception = None
-        rest_assistant = await self._web_assistants_factory.get_rest_assistant()
-        url = web_utils.rest_url(path_url, self.domain, api_version)
-        for _ in range(2):
-            try:
-
-                async with self._throttler.execute_task(limit_id=limit_id if limit_id else path_url):
-                    request = RESTRequest(
-                        method=method,
-                        url=url,
-                        params=params,
-                        data=data,
-                        is_auth_required=is_auth_required,
-                        throttler_limit_id=limit_id if limit_id else path_url
-                    )
-                    response = await rest_assistant.call(request=request)
-
-                    if response.status != 200:
-                        if return_err:
-                            error_response = await response.json()
-                            return error_response
-                        else:
-                            error_response = await response.text()
-                            raise IOError(f"Error executing request {method.name} {path_url}. "
-                                          f"HTTP status is {response.status}. "
-                                          f"Error: {error_response}")
-                    return await response.json()
-            except IOError as request_exception:
-                last_exception = request_exception
-                if self._is_request_exception_related_to_time_synchronizer(request_exception=request_exception):
-                    self._time_synchronizer.clear_time_offset_ms_samples()
-                    await self._update_time_synchronizer()
-                else:
-                    raise
-
-        # Failed even after the last retry
-        raise last_exception

hummingbot/connector/derivative/binance_perpetual/binance_perpetual_user_stream_data_source.py

@@ -7,6 +7,7 @@
 from hummingbot.connector.derivative.binance_perpetual.binance_perpetual_auth import BinancePerpetualAuth
 from hummingbot.core.data_type.user_stream_tracker_data_source import UserStreamTrackerDataSource
 from hummingbot.core.utils.async_utils import safe_ensure_future
+from hummingbot.core.web_assistant.connections.data_types import RESTMethod
 from hummingbot.core.web_assistant.web_assistants_factory import WebAssistantsFactory
 from hummingbot.core.web_assistant.ws_assistant import WSAssistant
 from hummingbot.logger import HummingbotLogger
@@ -54,24 +55,23 @@ async def _get_ws_assistant(self) -> WSAssistant:
             self._ws_assistant = await self._api_factory.get_ws_assistant()
         return self._ws_assistant
 
-    async def get_listen_key(self):
-        data = None
-
+    async def _get_listen_key(self):
+        rest_assistant = await self._api_factory.get_rest_assistant()
         try:
-            data = await self._connector._api_post(
-                path_url=CONSTANTS.BINANCE_USER_STREAM_ENDPOINT,
-                is_auth_required=True)
+            data = await rest_assistant.execute_request(
+                url=web_utils.private_rest_url(path_url=CONSTANTS.BINANCE_USER_STREAM_ENDPOINT, domain=self._domain),
+                method=RESTMethod.POST,
+                throttler_limit_id=CONSTANTS.BINANCE_USER_STREAM_ENDPOINT,
+                headers=self._auth.header_for_authentication()
+            )
         except asyncio.CancelledError:
             raise
         except Exception as exception:
-            raise IOError(
-                f"Error fetching Binance Perpetual user stream listen key. "
-                f"The response was {data}. Error: {exception}"
-            )
+            raise IOError(f"Error fetching user stream listen key. Error: {exception}")
 
         return data["listenKey"]
 
-    async def ping_listen_key(self) -> bool:
+    async def _ping_listen_key(self) -> bool:
         try:
             data = await self._connector._api_put(
                 path_url=CONSTANTS.BINANCE_USER_STREAM_ENDPOINT,
@@ -93,24 +93,23 @@ async def ping_listen_key(self) -> bool:
     async def _manage_listen_key_task_loop(self):
         try:
             while True:
+                now = int(time.time())
                 if self._current_listen_key is None:
-                    self._current_listen_key = await self.get_listen_key()
+                    self._current_listen_key = await self._get_listen_key()
                     self.logger().info(f"Successfully obtained listen key {self._current_listen_key}")
                     self._listen_key_initialized_event.set()
-                else:
-                    success: bool = await self.ping_listen_key()
+                    self._last_listen_key_ping_ts = int(time.time())
+
+                if now - self._last_listen_key_ping_ts >= self.LISTEN_KEY_KEEP_ALIVE_INTERVAL:
+                    success: bool = await self._ping_listen_key()
                     if not success:
-                        self.logger().error("Error occurred renewing listen key... ")
+                        self.logger().error("Error occurred renewing listen key ...")
                         break
                     else:
                         self.logger().info(f"Refreshed listen key {self._current_listen_key}.")
                         self._last_listen_key_ping_ts = int(time.time())
-                await self._sleep(self.LISTEN_KEY_KEEP_ALIVE_INTERVAL)
-
-        except Exception as e:
-            self.logger().error(f"Unexpected error occurred with maintaining listen key. "
-                                f"Error {e}")
-            raise
+                else:
+                    await self._sleep(self.LISTEN_KEY_KEEP_ALIVE_INTERVAL)
         finally:
             self._current_listen_key = None
             self._listen_key_initialized_event.clear()