Repository: hummingbot/hummingbot
Pull request: feat/ Implement an `APIThrottler` that accommodates the various Rate Limit types
**Before submitting this PR, please make sure**:

- [x] Your code builds clean without any errors or warnings
- [x] You are using approved title ("feat/", "fix/", "docs/", "refactor/")

**A description of the changes proposed in the pull request**:

Implement a new `APIThrottler` class to handle all rate limit implementations listed in #2601 

Includes an example of its usage with the Crypto.com connector.

**Test to be performed by developers:**
Added 3 APIThrottler classes for the 3 commonly identified rate limit types exchanges implement.
Added Unit Tests cases for each of the throttler classes demonstrating the behaviour of the throttler in various scenarios.

**Hints for QA Team**
Ensure that the `crypto_com` connector is able to limit and throttle the no. of API requests for a specific endpoints.
Refer to Crypto.com's rate limits defined [here](https://exchange-docs.crypto.com/spot/index.html#rate-limits)

Rate Limits are declared in `crypto_com_constants.py`. You can modify certain values to forcibly lower the rate limits.

Problem statement:
Implement `APIThrottler` class
**Initial Investigation**
- There are several ways Rate Limits are being implemented by the various exchanges. 

  1. Weighted Requests, where each user is assigned a fix rate limit and each method is assigned a weight. This is used by Binance. Where certain API endpoints has a higher weight depending on the resource consumed.
  2. Rate Limit by Methods, exchanges assigns rate limits for every API endpoint per API key pair. This is observed in Crypto.com i.e. `private/create-order` is limited to `15 requests per 100ms`
  3. Fixed Rate Limit, this is implemented by most other exchanges. i.e. AscendEx has a fixed rate limit of 6,000 requests/min

**Suggested Solution**
- Populate the various exchange specific rate limits for each API endpoint.
- Refactor of a `Throttler` class to deal with these 3 types of rate limits.

Hints from issue discussion:

Commit message: (feat) refactor APIThrottler class into a base class

Patch to review:
diff --git a/hummingbot/connector/exchange/crypto_com/crypto_com_constants.py b/hummingbot/connector/exchange/crypto_com/crypto_com_constants.py
index 7e871e40200..63efd68a6ed 100644
--- a/hummingbot/connector/exchange/crypto_com/crypto_com_constants.py
+++ b/hummingbot/connector/exchange/crypto_com/crypto_com_constants.py
@@ -1,5 +1,6 @@
 # A single source of truth for constant variables related to the exchange
 
+from hummingbot.core.utils.api_throttler import RateLimit
 
 EXCHANGE_NAME = "crypto_com"
 REST_URL = "https://api.crypto.com/v2"
@@ -8,6 +9,27 @@
 # WSS_PUBLIC_URL = "wss://stream.crypto.com/v2/market"
 WSS_PUBLIC_URL = "wss://d10tq1f9ygdz7y.cloudfront.net/v2/market"
 
+# REST API ENDPOINTS
+CHECK_NETWORK_PATH_URL = "public/get-ticker?instrument_name=BTC_USDT"
+GET_TRADING_RULES_PATH_URL = "public/get-instruments"
+CREATE_ORDER_PATH_URL = "private/create-order"
+CANCEL_ORDER_PATH_URL = "private/cancel-order"
+GET_ACCOUNT_SUMMARY_PATH_URL = "private/get-account-summary"
+GET_ORDER_DETAIL_PATH_URL = "private/get-order-detail"
+GET_OPEN_ORDERS_PATH_URL = "private/get-open-orders"
+
+# Crypto.com has a per method API limit
+RATE_LIMITS = [
+    # RateLimit(Limit, TimeInterval, RequestPath)
+    RateLimit(100, 1, path_url=CHECK_NETWORK_PATH_URL),
+    RateLimit(100, 1, path_url=GET_TRADING_RULES_PATH_URL),
+    RateLimit(15, 0.1, path_url=CREATE_ORDER_PATH_URL),
+    RateLimit(15, 0.1, path_url=CANCEL_ORDER_PATH_URL),
+    RateLimit(3, 0.1, path_url=GET_ACCOUNT_SUMMARY_PATH_URL),
+    RateLimit(30, 0.1, path_url=GET_ORDER_DETAIL_PATH_URL),
+    RateLimit(3, 0.1, path_url=GET_OPEN_ORDERS_PATH_URL),
+]
+
 API_REASONS = {
     0: "Success",
     10001: "Malformed request, (E.g. not using application/json for REST)",
diff --git a/hummingbot/connector/exchange/crypto_com/crypto_com_exchange.py b/hummingbot/connector/exchange/crypto_com/crypto_com_exchange.py
index 0bc4898df51..226bae71a99 100644
--- a/hummingbot/connector/exchange/crypto_com/crypto_com_exchange.py
+++ b/hummingbot/connector/exchange/crypto_com/crypto_com_exchange.py
@@ -40,8 +40,10 @@
 from hummingbot.connector.exchange.crypto_com.crypto_com_auth import CryptoComAuth
 from hummingbot.connector.exchange.crypto_com.crypto_com_in_flight_order import CryptoComInFlightOrder
 from hummingbot.connector.exchange.crypto_com import crypto_com_utils
-from hummingbot.connector.exchange.crypto_com import crypto_com_constants as Constants
+from hummingbot.connector.exchange.crypto_com import crypto_com_constants as CONSTANTS
 from hummingbot.core.data_type.common import OpenOrder
+from hummingbot.core.api_throttler.varied_rate_api_throttler import VariedRateThrottler
+
 ctce_logger = None
 s_decimal_NaN = Decimal("nan")
 
@@ -92,6 +94,9 @@ def __init__(self,
         self._user_stream_event_listener_task = None
         self._trading_rules_polling_task = None
         self._last_poll_timestamp = 0
+        self._throttler = VariedRateThrottler(
+            rate_limit_list=CONSTANTS.RATE_LIMITS,
+        )
 
     @property
     def name(self) -> str:
@@ -196,6 +201,7 @@ async def stop_network(self):
         This function is required by NetworkIterator base class and is called automatically.
         """
         self._order_book_tracker.stop()
+        self._throttler.stop()
         if self._status_polling_task is not None:
             self._status_polling_task.cancel()
             self._status_polling_task = None
@@ -219,7 +225,7 @@ async def check_network(self) -> NetworkStatus:
         """
         try:
             # since there is no ping endpoint, the lowest rate call is to get BTC-USDT ticker
-            await self._api_request("get", "public/get-ticker?instrument_name=BTC_USDT")
+            await self._api_request("get", CONSTANTS.CHECK_NETWORK_PATH_URL)
         except asyncio.CancelledError:
             raise
         except Exception:
@@ -252,7 +258,7 @@ async def _trading_rules_polling_loop(self):
                 await asyncio.sleep(0.5)
 
     async def _update_trading_rules(self):
-        instruments_info = await self._api_request("get", path_url="public/get-instruments")
+        instruments_info = await self._api_request("get", path_url=CONSTANTS.GET_TRADING_RULES_PATH_URL)
         self._trading_rules.clear()
         self._trading_rules = self._format_trading_rules(instruments_info)
 
@@ -315,37 +321,38 @@ async def _api_request(self,
         signature to the request.
         :returns A response in json format.
         """
-        url = f"{Constants.REST_URL}/{path_url}"
-        client = await self._http_client()
-        if is_auth_required:
-            request_id = crypto_com_utils.RequestId.generate_request_id()
-            data = {"params": params}
-            params = self._crypto_com_auth.generate_auth_dict(path_url, request_id,
-                                                              crypto_com_utils.get_ms_timestamp(), data)
-            headers = self._crypto_com_auth.get_headers()
-        else:
-            headers = {"Content-Type": "application/json"}
-
-        if method == "get":
-            response = await client.get(url, headers=headers)
-        elif method == "post":
-            post_json = json.dumps(params)
-            response = await client.post(url, data=post_json, headers=headers)
-        else:
-            raise NotImplementedError
+        async with self._throttler.execute_task(path_url):
+            url = f"{CONSTANTS.REST_URL}/{path_url}"
+            client = await self._http_client()
+            if is_auth_required:
+                request_id = crypto_com_utils.RequestId.generate_request_id()
+                data = {"params": params}
+                params = self._crypto_com_auth.generate_auth_dict(path_url, request_id,
+                                                                  crypto_com_utils.get_ms_timestamp(), data)
+                headers = self._crypto_com_auth.get_headers()
+            else:
+                headers = {"Content-Type": "application/json"}
+
+            if method == "get":
+                response = await client.get(url, headers=headers)
+            elif method == "post":
+                post_json = json.dumps(params)
+                response = await client.post(url, data=post_json, headers=headers)
+            else:
+                raise NotImplementedError
 
-        try:
-            parsed_response = json.loads(await response.text())
-        except Exception as e:
-            raise IOError(f"Error parsing data from {url}. Error: {str(e)}")
-        if response.status != 200:
-            raise IOError(f"Error fetching data from {url}. HTTP status is {response.status}. "
-                          f"Message: {parsed_response}")
-        if parsed_response["code"] != 0:
-            raise IOError(f"{url} API call failed, response: {parsed_response}")
-        # print(f"REQUEST: {method} {path_url} {params}")
-        # print(f"RESPONSE: {parsed_response}")
-        return parsed_response
+            try:
+                parsed_response = json.loads(await response.text())
+            except Exception as e:
+                raise IOError(f"Error parsing data from {url}. Error: {str(e)}")
+            if response.status != 200:
+                raise IOError(f"Error fetching data from {url}. HTTP status is {response.status}. "
+                              f"Message: {parsed_response}")
+            if parsed_response["code"] != 0:
+                raise IOError(f"{url} API call failed, response: {parsed_response}")
+            # print(f"REQUEST: {method} {path_url} {params}")
+            # print(f"RESPONSE: {parsed_response}")
+            return parsed_response
 
     def get_order_price_quantum(self, trading_pair: str, price: Decimal):
         """
@@ -449,7 +456,7 @@ async def _create_order(self,
                                   order_type
                                   )
         try:
-            order_result = await self._api_request("post", "private/create-order", api_params, True)
+            order_result = await self._api_request("post", CONSTANTS.CREATE_ORDER_PATH_URL, api_params, True)
             exchange_order_id = str(order_result["result"]["order_id"])
             tracked_order = self._in_flight_orders.get(order_id)
             if tracked_order is not None:
@@ -527,7 +534,7 @@ async def _execute_cancel(self, trading_pair: str, order_id: str) -> str:
             ex_order_id = tracked_order.exchange_order_id
             await self._api_request(
                 "post",
-                "private/cancel-order",
+                CONSTANTS.CANCEL_ORDER_PATH_URL,
                 {"instrument_name": crypto_com_utils.convert_to_exchange_trading_pair(trading_pair),
                  "order_id": ex_order_id},
                 True
@@ -573,7 +580,7 @@ async def _update_balances(self):
         """
         local_asset_names = set(self._account_balances.keys())
         remote_asset_names = set()
-        account_info = await self._api_request("post", "private/get-account-summary", {}, True)
+        account_info = await self._api_request("post", CONSTANTS.GET_ACCOUNT_SUMMARY_PATH_URL, {}, True)
         for account in account_info["result"]["accounts"]:
             asset_name = account["currency"]
             self._account_available_balances[asset_name] = Decimal(str(account["available"]))
@@ -598,7 +605,7 @@ async def _update_order_status(self):
             for tracked_order in tracked_orders:
                 order_id = await tracked_order.get_exchange_order_id()
                 tasks.append(self._api_request("post",
-                                               "private/get-order-detail",
+                                               CONSTANTS.GET_ORDER_DETAIL_PATH_URL,
                                                {"order_id": order_id},
                                                True))
             self.logger().debug(f"Polling for order status updates of {len(tasks)} orders.")
@@ -715,7 +722,7 @@ async def cancel_all(self, timeout_seconds: float):
                     "order_id": order.exchange_order_id,
                 }
                 tasks.append(self._api_request(method="post",
-                                               path_url="private/cancel-order",
+                                               path_url=CONSTANTS.CANCEL_ORDER_PATH_URL,
                                                params=api_params,
                                                is_auth_required=True))
 
@@ -815,7 +822,7 @@ async def _user_stream_event_listener(self):
     async def get_open_orders(self) -> List[OpenOrder]:
         result = await self._api_request(
             "post",
-            "private/get-open-orders",
+            CONSTANTS.GET_OPEN_ORDERS_PATH_URL,
             {},
             True
         )
diff --git a/hummingbot/core/api_throttler/__init__.py b/hummingbot/core/api_throttler/__init__.py
new file mode 100644
index 00000000000..e69de29bb2d
diff --git a/hummingbot/core/api_throttler/api_request_context_base.py b/hummingbot/core/api_throttler/api_request_context_base.py
new file mode 100644
index 00000000000..df965dbd748
--- /dev/null
+++ b/hummingbot/core/api_throttler/api_request_context_base.py
@@ -0,0 +1,82 @@
+import asyncio
+import time
+
+from typing import (
+    Deque,
+)
+
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    TaskLog,
+    Seconds
+)
+
+
+class APIRequestContextBase:
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 pending_tasks: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 stop_event: asyncio.Event,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        """
+        Asynchronous context associated with each API request.
+        :param task_logs: Shared task logs
+        :param rate_limit: Rate limit for the associated API request
+        :param rate_limit_type: Rate limit type
+        :param period_safety_margin: estimate for the network latency
+        :param retry_interval: Time between each limit check
+        """
+        self._lock = asyncio.Lock()
+        self._stop_event = stop_event
+        self._task_logs: Deque[TaskLog] = task_logs
+        self._pending_tasks: Deque[TaskLog] = pending_tasks
+
+        self._rate_limit: RateLimit = rate_limit
+
+        self._time_interval: float = rate_limit.time_interval
+        self._retry_interval: float = retry_interval
+        self._period_safety_margin = period_safety_margin
+
+    def flush(self):
+        """
+        Remove task logs that have passed rate limit periods
+        :return:
+        """
+        now: float = time.time()
+        while self._task_logs:
+            task_log: TaskLog = self._task_logs[0]
+            elapsed: float = now - task_log.timestamp
+            if elapsed > self._time_interval - self._period_safety_margin:
+                self._task_logs.popleft()
+            else:
+                break
+
+    def within_capacity(self) -> bool:
+        raise NotImplementedError
+
+    async def acquire(self):
+        self._pending_tasks.append(id(self))
+        while True:
+            self.flush()
+
+            if self.within_capacity():
+                break
+            await asyncio.sleep(self._retry_interval)
+
+        self._pending_tasks.remove(id(self))
+        task = TaskLog(timestamp=time.time(),
+                       path_url=self._rate_limit.path_url,
+                       weight=self._rate_limit.weight)
+
+        self._task_logs.append(task)
+
+    async def __aenter__(self):
+        async with self._lock:
+            await self.acquire()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        if self._stop_event.is_set():
+            raise asyncio.CancelledError
diff --git a/hummingbot/core/api_throttler/api_throttler_base.py b/hummingbot/core/api_throttler/api_throttler_base.py
new file mode 100644
index 00000000000..7897b5d000c
--- /dev/null
+++ b/hummingbot/core/api_throttler/api_throttler_base.py
@@ -0,0 +1,50 @@
+import asyncio
+
+from collections import deque
+from typing import (
+    Deque,
+    Dict,
+    List,
+)
+
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    RequestPath,
+    Seconds,
+    TaskLog
+)
+
+
+class APIThrottlerBase:
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        """
+        The APIThrottlerBase is an abstract class meant to describe the functions necessary to handle the
+        throttling of API requests through the usage of asynchronous context managers
+        """
+
+        # Rate Limit Definitions
+        self._rate_limit_list: List[RateLimit] = rate_limit_list
+
+        self._path_rate_limit_map: Dict[RequestPath, RateLimit] = {
+            limit.path_url: limit
+            for limit in self._rate_limit_list
+        }
+
+        # Throttler Parameters
+        self._retry_interval: float = retry_interval
+        self._period_safety_margin = period_safety_margin
+
+        self._task_logs: Deque[TaskLog] = deque()
+        self._pending_task: Deque[int]
+
+        # Event
+        self._throttler_stop_event = asyncio.Event()
+
+    def execute_task(self):
+        raise NotImplementedError
+
+    def stop(self):
+        self._throttler_stop_event.set()
diff --git a/hummingbot/core/api_throttler/data_types.py b/hummingbot/core/api_throttler/data_types.py
new file mode 100644
index 00000000000..6e3fe2e1e00
--- /dev/null
+++ b/hummingbot/core/api_throttler/data_types.py
@@ -0,0 +1,23 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class RateLimit():
+    limit: int
+    time_interval: float
+    path_url: str = ""
+    weight: int = 1
+
+
+@dataclass
+class TaskLog():
+    timestamp: float
+    path_url: str = ""
+    weight: int = 1
+
+
+Limit = int             # Integer representing the no. of requests be time interval
+RequestPath = str       # String representing the request path url
+RequestWeight = int     # Integer representing the request weight of the path url
+
+Seconds = float
diff --git a/hummingbot/core/api_throttler/fixed_rate_api_throttler.py b/hummingbot/core/api_throttler/fixed_rate_api_throttler.py
new file mode 100644
index 00000000000..25d1dba7d87
--- /dev/null
+++ b/hummingbot/core/api_throttler/fixed_rate_api_throttler.py
@@ -0,0 +1,42 @@
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class FixedRateThrottler(APIThrottlerBase):
+
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(rate_limit_list, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def execute_task(self):
+        rate_limit: RateLimit = next(iter(self._path_rate_limit_map.values()))
+
+        return FixedRateRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            retry_interval=self._retry_interval,
+        )
+
+
+class FixedRateRequestContext(APIRequestContextBase):
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 period_safety_margin: Seconds,
+                 retry_interval: Seconds):
+        super().__init__(task_logs, rate_limit, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def within_capacity(self) -> bool:
+        current_capacity: int = self._rate_limit.limit - len(self._task_logs)
+
+        return current_capacity - self._rate_limit.weight >= 0
diff --git a/hummingbot/core/api_throttler/varied_rate_api_throttler.py b/hummingbot/core/api_throttler/varied_rate_api_throttler.py
new file mode 100644
index 00000000000..d96e8e68433
--- /dev/null
+++ b/hummingbot/core/api_throttler/varied_rate_api_throttler.py
@@ -0,0 +1,54 @@
+import asyncio
+
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class VariedRateThrottler(APIThrottlerBase):
+
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(rate_limit_list, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def execute_task(self, path_url: str):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return VariedRateRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            stop_event=self._throttler_stop_event,
+            period_safety_margin=self._period_safety_margin,
+            retry_interval=self._retry_interval,
+        )
+
+
+class VariedRateRequestContext(APIRequestContextBase):
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 stop_event: asyncio.Event,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(
+            task_logs,
+            rate_limit,
+            stop_event,
+            period_safety_margin=period_safety_margin,
+            retry_interval=retry_interval)
+
+    def within_capacity(self) -> bool:
+        current_capacity: int = self._rate_limit.limit - len([task
+                                                              for task in self._task_logs
+                                                              if task.path_url == self._rate_limit.path_url])
+
+        return current_capacity - self._rate_limit.weight >= 0
diff --git a/hummingbot/core/api_throttler/weighted_api_throttler.py b/hummingbot/core/api_throttler/weighted_api_throttler.py
new file mode 100644
index 00000000000..ac9506d4ea2
--- /dev/null
+++ b/hummingbot/core/api_throttler/weighted_api_throttler.py
@@ -0,0 +1,44 @@
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class WeightedAPIThrottler(APIThrottlerBase):
+
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(rate_limit_list, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def execute_task(self, path_url: str):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return WeightedRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            retry_interval=self._retry_interval,
+        )
+
+
+class WeightedRequestContext(APIRequestContextBase):
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(task_logs, rate_limit, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def within_capacity(self) -> bool:
+        current_capacity: int = self._rate_limit.limit - sum([task.weight
+                                                              for task in self._task_logs
+                                                              if task.path_url == self._rate_limit.path_url])
+
+        return current_capacity - self._rate_limit.weight >= 0
diff --git a/hummingbot/core/mock_api/mock_web_server.py b/hummingbot/core/mock_api/mock_web_server.py
index a148d8e20db..1eb77a95bda 100644
--- a/hummingbot/core/mock_api/mock_web_server.py
+++ b/hummingbot/core/mock_api/mock_web_server.py
@@ -53,29 +53,25 @@ class MockWebServer:
     reroute_request(self, method, url, **kwargs)
     """
     TEST_RESPONSE = f"hello {str(random.randint(0, 10000000))}"
-    __instance = None
     _hosts_to_mock = {}
     host = "127.0.0.1"
     _port: Optional[int] = None
 
-    @staticmethod
-    def get_instance():
+    @classmethod
+    def get_instance(cls):
         """
         Initiate a Humming Web App instance
         :return: An instance of Humming Web App
         """
-        if MockWebServer.__instance is None:
-            MockWebServer()
-        return MockWebServer.__instance
+        instance = cls.__dict__.get("__instance__")
+        if instance is None:
+            cls.__instance__ = instance = cls()
+        return instance
 
     def __init__(self):
         """
         Constructs all the necessary attributes for the Humming Web object
         """
-        if MockWebServer.__instance is not None:
-            raise Exception("This class is a singleton!")
-        else:
-            MockWebServer.__instance = self
         self._ev_loop: None
         self._impl: Optional[web.Application] = None
         self._runner: Optional[web.AppRunner] = None
diff --git a/hummingbot/core/utils/api_throttler.py b/hummingbot/core/utils/api_throttler.py
new file mode 100644
index 00000000000..c51a5a733d8
--- /dev/null
+++ b/hummingbot/core/utils/api_throttler.py
@@ -0,0 +1,170 @@
+import asyncio
+import time
+
+from collections import deque
+from dataclasses import dataclass
+from enum import Enum
+from typing import (
+    Deque,
+    Dict,
+    List,
+)
+
+
+class RateLimitType(Enum):
+    FIXED = 1
+    WEIGHTED = 2
+    PER_METHOD = 3
+
+
+@dataclass
+class RateLimit():
+    limit: int
+    time_interval: float
+    path_url: str = ""
+    weight: int = 1
+
+
+@dataclass
+class TaskLog():
+    timestamp: float
+    path_url: str = ""
+    weight: int = 1
+
+
+Limit = int             # Integer representing the no. of requests be time interval
+RequestPath = str       # String representing the request path url
+RequestWeight = int     # Integer representing the request weight of the path url
+
+Seconds = float
+
+
+class APIThrottler:
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 rate_limit_type: RateLimitType = RateLimitType.FIXED,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        """
+        The APIThrottler class handles the throttling of API requests through the usage of asynchronous context
+        managers
+        """
+
+        # Rate Limit Definitions
+        self._rate_limit_type: RateLimitType = rate_limit_type
+        self._rate_limit_list: List[RateLimit] = rate_limit_list
+
+        self._path_rate_limit_map: Dict[RequestPath, RateLimit] = {
+            limit.path_url: limit
+            for limit in self._rate_limit_list
+        }
+
+        # Throttler Parameters
+        self._retry_interval: float = retry_interval
+        self._period_safety_margin = period_safety_margin
+
+        self._task_logs: Deque[TaskLog] = deque()
+
+    def weighted_task(self, path_url: str):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return APIRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            rate_limit_type=self._rate_limit_type,
+            retry_interval=self._retry_interval,
+        )
+
+    def fixed_rate_task(self):
+        rate_limit: RateLimit = next(iter(self._path_rate_limit_map.values()))
+
+        return APIRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            rate_limit_type=self._rate_limit_type,
+            retry_interval=self._retry_interval,
+        )
+
+    def per_method_task(self, path_url):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return APIRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            rate_limit_type=self._rate_limit_type,
+            retry_interval=self._retry_interval,
+        )
+
+
+class APIRequestContext:
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 rate_limit_type: RateLimitType,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        """
+        Asynchronous context associated with each API request.
+        :param task_logs: Shared task logs
+        :param rate_limit: Rate limit for the associated API request
+        :param rate_limit_type: Rate limit type
+        :param period_safety_margin: estimate for the network latency
+        :param retry_interval: Time between each limit check
+        """
+        self._lock = asyncio.Lock()
+        self._task_logs: Deque[TaskLog] = task_logs
+
+        self._rate_limit: RateLimit = rate_limit
+        self._rate_limit_type: RateLimitType = rate_limit_type
+
+        self._time_interval: float = rate_limit.time_interval
+        self._retry_interval: float = retry_interval
+        self._period_safety_margin = period_safety_margin
+
+    def flush(self):
+        """
+        Remove task logs that have passed rate limit periods
+        :return:
+        """
+        now: float = time.time()
+        while self._task_logs:
+            task_log: TaskLog = self._task_logs[0]
+            elapsed: float = now - task_log.timestamp
+            if elapsed > self._time_interval - self._period_safety_margin:
+                self._task_logs.popleft()
+            else:
+                break
+
+    async def acquire(self):
+        while True:
+            self.flush()
+
+            if self._rate_limit_type == RateLimitType.PER_METHOD:
+                current_capacity: int = self._rate_limit.limit - len([task
+                                                                      for task in self._task_logs
+                                                                      if task.path_url == self._rate_limit.path_url])
+            elif self._rate_limit_type == RateLimitType.FIXED:
+                current_capacity: int = self._rate_limit.limit - len(self._task_logs)
+            elif self._rate_limit_type == RateLimitType.WEIGHTED:
+                current_capacity: int = self._rate_limit.limit - sum([task.weight
+                                                                      for task in self._task_logs
+                                                                      if task.path_url == self._rate_limit.path_url])
+
+            # Request Weight for non-weighted requests defaults to 1
+            if current_capacity - self._rate_limit.weight >= 0:
+                break
+            await asyncio.sleep(self._retry_interval)
+
+        task = TaskLog(timestamp=time.time(),
+                       path_url=self._rate_limit.path_url,
+                       weight=self._rate_limit.weight)
+
+        self._task_logs.append(task)
+
+    async def __aenter__(self):
+        async with self._lock:
+            await self.acquire()
+
+    async def __aexit__(self, exc_type, exc, tb):
+        pass
diff --git a/test/hummingbot/core/utils/test_api_throttler.py b/test/hummingbot/core/utils/test_api_throttler.py
new file mode 100644
index 00000000000..69d6e584204
--- /dev/null
+++ b/test/hummingbot/core/utils/test_api_throttler.py
@@ -0,0 +1,222 @@
+import unittest
+import aiohttp
+import asyncio
+import random
+import requests
+
+from aiohttp import web
+from unittest.mock import patch
+
+from hummingbot.core.mock_api.mock_web_server import MockWebServer
+from hummingbot.core.utils.api_throttler import APIThrottler, RateLimit, RateLimitType
+
+BASE_URL = "www.hbottesst.com"
+
+
+TEST_PATH_URL = "/test"
+
+FIXED_RATE_LIMIT = [
+    RateLimit(5, 5)
+]
+
+WEIGHTED_RATE_LIMIT = [
+    RateLimit(5, 5, TEST_PATH_URL, 2)
+]
+
+PER_METHOD_RATE_LIMIT = [
+    RateLimit(5, 5, TEST_PATH_URL)
+]
+
+
+class ThrottledMockServer(MockWebServer):
+
+    def __init__(self):
+        super().__init__()
+        self._request_count = 0
+
+    @property
+    def request_count(self) -> int:
+        return self._request_count
+
+    def reset_request_count(self):
+        self._request_count = 0
+
+    async def _handler(self, request: web.Request):
+        self._request_count += 1
+        return await super()._handler(request)
+
+
+class APIThrottlerUnitTests(unittest.TestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        super().setUpClass()
+        cls.ev_loop: asyncio.AbstractEventLoop = asyncio.get_event_loop()
+
+        cls.mock_server = ThrottledMockServer.get_instance()
+        cls.mock_server.add_host_to_mock(BASE_URL)
+
+        cls.mock_server.start()
+        cls.ev_loop.run_until_complete(cls.mock_server.wait_til_started())
+
+        cls._patcher = patch("aiohttp.client.URL")
+        cls._url_mock = cls._patcher.start()
+        cls._url_mock.side_effect = MockWebServer.reroute_local
+
+        cls._req_patcher = unittest.mock.patch.object(requests.Session, "request", autospec=True)
+        cls._req_url_mock = cls._req_patcher.start()
+        cls._req_url_mock.side_effect = MockWebServer.reroute_request
+
+        cls.mock_server.clear_responses()
+        cls.mock_server.update_response(method="GET",
+                                        host=BASE_URL,
+                                        path=TEST_PATH_URL,
+                                        data=cls.mock_server.TEST_RESPONSE,
+                                        is_json=False
+                                        )
+
+    def setUp(self) -> None:
+        super().setUp()
+
+        self.mock_server.reset_request_count()
+
+        self.per_method_rate_throttler = APIThrottler(rate_limit_list=PER_METHOD_RATE_LIMIT,
+                                                      rate_limit_type=RateLimitType.PER_METHOD,
+                                                      retry_interval=5.0)
+
+        self.fixed_rate_throttler = APIThrottler(rate_limit_list=FIXED_RATE_LIMIT,
+                                                 rate_limit_type=RateLimitType.FIXED,
+                                                 retry_interval=5.0)
+
+        self.weighted_rate_throttler = APIThrottler(rate_limit_list=WEIGHTED_RATE_LIMIT,
+                                                    rate_limit_type=RateLimitType.WEIGHTED,
+                                                    retry_interval=5.0)
+
+    @classmethod
+    def tearDownClass(cls):
+        cls.mock_server.stop()
+        cls._patcher.stop()
+        cls._req_patcher.stop()
+        super().tearDownClass()
+
+    async def execute_n_per_method_requests(self, n: int, throttler: APIThrottler):
+        for _ in range(n):
+            async with throttler.per_method_task(path_url=TEST_PATH_URL):
+                async with aiohttp.ClientSession() as client:
+                    await client.get(f"https://{BASE_URL + TEST_PATH_URL}")
+
+    async def execute_n_fixed_requests(self, n: int, throttler: APIThrottler):
+        for _ in range(n):
+            async with throttler.fixed_rate_task():
+                async with aiohttp.ClientSession() as client:
+                    await client.get(f"https://{BASE_URL + TEST_PATH_URL}")
+
+    async def execute_n_weighted_requests(self, n: int, throttler: APIThrottler):
+        for _ in range(n):
+            async with throttler.weighted_task(path_url=TEST_PATH_URL):
+                async with aiohttp.ClientSession() as client:
+                    await client.get(f"https://{BASE_URL + TEST_PATH_URL}")
+
+    def test_per_method_rate_throttler_above_limit(self):
+        # Test Scenario: API requests sent > Rate Limit
+        n: int = 10
+        limit: int = PER_METHOD_RATE_LIMIT[0].limit
+
+        # Note: We assert a timeout ensuring that the throttler does not wait for the limit interval
+        with self.assertRaises(asyncio.exceptions.TimeoutError):
+            self.ev_loop.run_until_complete(
+                asyncio.wait_for(self.execute_n_per_method_requests(n, throttler=self.per_method_rate_throttler), timeout=1.0)
+            )
+
+        self.assertEqual(limit, self.mock_server.request_count)
+
+    def test_per_method_rate_throttler_below_limit(self):
+        # Test Scenario: API requests sent < Rate Limit
+        n: int = random.randint(1, PER_METHOD_RATE_LIMIT[0].limit - 1)
+        limit: int = PER_METHOD_RATE_LIMIT[0].limit
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_per_method_requests(n, throttler=self.per_method_rate_throttler))
+
+        self.assertEqual(self.mock_server.request_count, n)
+        self.assertLess(self.mock_server.request_count, limit)
+
+    def test_per_method_rate_throttler_equal_limit(self):
+        # Test Scenario: API requests sent = Rate Limit
+        n = limit = PER_METHOD_RATE_LIMIT[0].limit
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_per_method_requests(n, throttler=self.per_method_rate_throttler))
+
+        self.assertEqual(self.mock_server.request_count, limit)
+
+    def test_fixed_rate_throttler_above_limit(self):
+        # Test Scenario: API requests sent > Rate Limit
+        n: int = 10
+        limit: int = FIXED_RATE_LIMIT[0].limit
+
+        # Note: We assert a timeout ensuring that the throttler does not wait for the limit interval
+        with self.assertRaises(asyncio.exceptions.TimeoutError):
+            self.ev_loop.run_until_complete(
+                asyncio.wait_for(self.execute_n_fixed_requests(n, throttler=self.fixed_rate_throttler), timeout=1.0)
+            )
+
+        self.assertEqual(limit, self.mock_server.request_count)
+
+    def test_fixed_rate_throttler_below_limit(self):
+        # Test Scenario: API requests sent < Rate Limit
+        n: int = random.randint(1, FIXED_RATE_LIMIT[0].limit - 1)
+        limit: int = FIXED_RATE_LIMIT[0].limit
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_fixed_requests(n, throttler=self.fixed_rate_throttler))
+
+        self.assertEqual(self.mock_server.request_count, n)
+        self.assertLess(self.mock_server.request_count, limit)
+
+    def test_fixed_rate_throttler_equal_limit(self):
+        # Test Scenario: API requests sent = Rate Limit
+        n = limit = FIXED_RATE_LIMIT[0].limit
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_fixed_requests(n, throttler=self.fixed_rate_throttler))
+
+        self.assertEqual(self.mock_server.request_count, limit)
+
+    def test_weighted_rate_throttler_above_limit(self):
+        # Test Scenario: API requests sent > Rate Limit
+        n: int = 10
+        limit: int = WEIGHTED_RATE_LIMIT[0].limit
+        weight: int = WEIGHTED_RATE_LIMIT[0].weight
+
+        expected_request_count: int = limit // weight
+
+        # Note: We assert a timeout ensuring that the throttler does not wait for the limit interval
+        with self.assertRaises(asyncio.exceptions.TimeoutError):
+            self.ev_loop.run_until_complete(
+                asyncio.wait_for(self.execute_n_weighted_requests(n, throttler=self.weighted_rate_throttler), timeout=1.0)
+            )
+
+        self.assertEqual(expected_request_count, self.mock_server.request_count)
+
+    def test_weighted_rate_throttler_below_limit(self):
+        n: int = WEIGHTED_RATE_LIMIT[0].limit // WEIGHTED_RATE_LIMIT[0].weight
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_weighted_requests(n, throttler=self.weighted_rate_throttler))
+
+        self.assertEqual(n, self.mock_server.request_count)
+
+    def test_weighted_rate_throttler_equal_limit(self):
+        rate_limit: RateLimit = RateLimit(5, 5, TEST_PATH_URL, 1)
+
+        n: int = rate_limit.limit // rate_limit.weight
+
+        self.weighted_rate_throttler = APIThrottler(rate_limit_list=[rate_limit],
+                                                    rate_limit_type=RateLimitType.WEIGHTED,
+                                                    retry_interval=5.0)
+
+        self.ev_loop.run_until_complete(
+            self.execute_n_weighted_requests(n, throttler=self.weighted_rate_throttler))
+
+        self.assertEqual(n, self.mock_server.request_count)


Code review comments:
diff_hunk: @@ -0,0 +1,44 @@
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class WeightedAPIThrottler(APIThrottlerBase):
+
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(rate_limit_list, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def execute_task(self, path_url: str):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return WeightedRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            retry_interval=self._retry_interval,
+        )
+
+
+class WeightedRequestContext(APIRequestContextBase):
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(task_logs, rate_limit, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
path: hummingbot/core/api_throttler/weighted_api_throttler.py
text: @user1:
`__init__` is not required because it just calls super()

diff_hunk: @@ -0,0 +1,82 @@
+import asyncio
+import time
+
+from typing import (
+    Deque,
+)
+
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    TaskLog,
+    Seconds
+)
+
+
+class APIRequestContextBase:
+
+    def __init__(self,
+                 task_logs: Deque[TaskLog],
+                 pending_tasks: Deque[TaskLog],
+                 rate_limit: RateLimit,
+                 stop_event: asyncio.Event,
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        """
+        Asynchronous context associated with each API request.
+        :param task_logs: Shared task logs
+        :param rate_limit: Rate limit for the associated API request
+        :param rate_limit_type: Rate limit type
+        :param period_safety_margin: estimate for the network latency
+        :param retry_interval: Time between each limit check
+        """
+        self._lock = asyncio.Lock()
+        self._stop_event = stop_event
path: hummingbot/core/api_throttler/api_request_context_base.py
text: @user1:
I don't understand the use of _stop_event. Is it only used to signal the cancelation of a task?
What happens when it is None because it was not provided as a parameter? (For example the `__init__` function in `FixedRateRequestContext` is not sending the parameter as part of the super() call)

diff_hunk: @@ -0,0 +1,44 @@
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class WeightedAPIThrottler(APIThrottlerBase):
+
+    def __init__(self,
path: hummingbot/core/api_throttler/weighted_api_throttler.py
text: @user1:
`__init__` here just calls super(). It is not required.

diff_hunk: @@ -0,0 +1,42 @@
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class FixedRateThrottler(APIThrottlerBase):
+
+    def __init__(self,
path: hummingbot/core/api_throttler/fixed_rate_api_throttler.py
text: @user1:
`__init__` is just calling super(). It is not required.

diff_hunk: @@ -0,0 +1,54 @@
+import asyncio
+
+from typing import Deque, List
+
+from hummingbot.core.api_throttler.api_request_context_base import APIRequestContextBase
+from hummingbot.core.api_throttler.api_throttler_base import APIThrottlerBase
+from hummingbot.core.api_throttler.data_types import (
+    RateLimit,
+    Seconds,
+    TaskLog
+)
+
+
+class VariedRateThrottler(APIThrottlerBase):
+
+    def __init__(self,
+                 rate_limit_list: List[RateLimit],
+                 period_safety_margin: Seconds = 0.1,
+                 retry_interval: Seconds = 0.1):
+        super().__init__(rate_limit_list, period_safety_margin=period_safety_margin, retry_interval=retry_interval)
+
+    def execute_task(self, path_url: str):
+        rate_limit: RateLimit = self._path_rate_limit_map[path_url]
+
+        return VariedRateRequestContext(
+            task_logs=self._task_logs,
+            rate_limit=rate_limit,
+            stop_event=self._throttler_stop_event,
+            period_safety_margin=self._period_safety_margin,
+            retry_interval=self._retry_interval,
+        )
+
+
+class VariedRateRequestContext(APIRequestContextBase):
+
+    def __init__(self,
path: hummingbot/core/api_throttler/varied_rate_api_throttler.py
text: @user1:
`__init__` here just calls super(). It is not required.