Repository: hummingbot/hummingbot
Pull request: fix/reduced infura consumption with websocket connection
**Before submitting this PR, please make sure**:
- [ ] Your code builds clean without any errors or warnings
- [ ] You are using approved title ("feat/", "fix/", "docs/", "refactor/", etc)


**A description of the changes proposed in the pull request**:

Reduced infura consumption below 100,000 per day limit with WebSocket connection to replace watcher for new blocks
- Clubhouse Ticket: https://app.clubhouse.io/coinalpha/story/9071/reduce-infura-api-calls-on-hb-client
- Github: #1848 

Need someone to also to test if the consumption is indeed below the rate (4166/hr) for them


Problem statement:
[Research] Assess Infura usage when using Hummingbot on DEXs
Due to recent changes in rate limits by Infura, user report that Hummingbot clients running PMM with basic settings now hit the daily 100k limit in approx 21 hours.

Assess how severe this issue is by running various bot configurations for 1 hour each and assessing how many Infura calls are made over that time. The main params that affect number of Infura calls should be:
- number of order levels
- order refresh time
- other params that increase number of markets tracked or API calls made

Hints from issue discussion:
For Infura's free "Core" account,  they restrict the number of Requests/Day to 100,000. Otherwise, you have to pay at least $50/month. 

We are therefore restricted to about 4,166 requests per hour. After some testing with our PMM strategy on Radar Relay, our client calls Infura ranging from 4000 - 6000 times per hour, depending on the configuration. `order_refresh_time` of less than 60 seconds and an `order_level` greater than 1 easily results in a rate higher than 4166 on average. 

**Top Infura Method Calls:**
- `eth_getLogs`: 39.38%
- `eth_getBlockByNumber`: 19.47%
- `eth_call`: 14.91%
- `eth_getBlockByHash`: 12.71%
- `eth_getBalance`: 5.58%
- `eth_gasPrice`: 5.50%
- `eth_accounts`, `eth_getTransactionCount`, `eth_blockNumber`: < 2.00%

**Infura Usage Data from Past 24-hours**
(Only ran bot for 7 Trials [19 hours total Maximum])
<img width="1091" alt="Screen Shot 2020-06-12 at 4 55 37 PM" src="https://user-images.githubusercontent.com/20906874/84545607-8e116800-accd-11ea-895f-cdba7c091028.png">

a hint for reducing smart contract calls is using helper contracts like the following any time multiple requests are sent for the same operation:
[https://github.com/RigoBlock/rigoblock-monorepo/blob/master/packages/contracts/src/examples/helpers/HGetMultipleBalances.sol](url)

Ofc logs, which are the biggest number of calls here, can be reduced only by substituting log queries with transaction queries and filters, or, as already discussed with @mifeng , by properly optimizing the code.

Commit message: Merge branch 'development' into bug/fix_infura_consumption

Patch to review:
diff --git a/hummingbot/client/command/connect_command.py b/hummingbot/client/command/connect_command.py
index b60e48d3e24..3d62e4da357 100644
--- a/hummingbot/client/command/connect_command.py
+++ b/hummingbot/client/command/connect_command.py
@@ -142,6 +142,7 @@ async def connect_ethereum(self,  # type: HummingbotApplication
             public_address = Security.add_private_key(private_key)
             global_config_map["ethereum_wallet"].value = public_address
             await self.prompt_a_config(global_config_map["ethereum_rpc_url"])
+            await self.prompt_a_config(global_config_map["ethereum_rpc_ws_url"])
             if self.app.to_stop_config:
                 self.app.to_stop_config = False
                 return
diff --git a/hummingbot/client/config/global_config_map.py b/hummingbot/client/config/global_config_map.py
index a9aaa2a7184..74188c9aa94 100644
--- a/hummingbot/client/config/global_config_map.py
+++ b/hummingbot/client/config/global_config_map.py
@@ -274,6 +274,11 @@ def using_bamboo_coordinator_mode() -> bool:
                   prompt="Which Ethereum node would you like your client to connect to? >>> ",
                   required_if=lambda: global_config_map["ethereum_wallet"].value is not None,
                   is_connect_key=True),
+    "ethereum_rpc_ws_url":
+        ConfigVar(key="ethereum_rpc_ws_url",
+                  prompt="Enter the Websocket Address of your Ethereum Node >>> ",
+                  required_if=lambda: global_config_map["ethereum_rpc_url"].value is not None,
+                  is_connect_key=True),
     "ethereum_chain_name":
         ConfigVar(key="ethereum_chain_name",
                   prompt="What is your preferred ethereum chain name? >>> ",
diff --git a/hummingbot/core/utils/ethereum.py b/hummingbot/core/utils/ethereum.py
index 0bd52f9e3e6..ff667d3a0ab 100644
--- a/hummingbot/core/utils/ethereum.py
+++ b/hummingbot/core/utils/ethereum.py
@@ -1,5 +1,9 @@
+import binascii
 import logging
+from hexbytes import HexBytes
 from web3 import Web3
+from web3.datastructures import AttributeDict
+from typing import Dict
 
 
 def check_web3(ethereum_rpc_url: str) -> bool:
@@ -18,3 +22,13 @@ def check_web3(ethereum_rpc_url: str) -> bool:
             logging.getLogger().warning("Please add \"https://\" to your Infura node url.")
     return ret
 
+
+def block_values_to_hex(block: AttributeDict) -> AttributeDict:
+    formatted_block: Dict = {}
+    for key in block.keys():
+        value = block[key]
+        try:
+            formatted_block[key] = HexBytes(value)
+        except binascii.Error:
+            formatted_block[key] = value
+    return AttributeDict(formatted_block)
diff --git a/hummingbot/templates/conf_global_TEMPLATE.yml b/hummingbot/templates/conf_global_TEMPLATE.yml
index 2b003c6b9cf..db4b4c2ec4b 100644
--- a/hummingbot/templates/conf_global_TEMPLATE.yml
+++ b/hummingbot/templates/conf_global_TEMPLATE.yml
@@ -45,6 +45,7 @@ celo_password: null
 # Ethereum wallet address: required for trading on a DEX
 ethereum_wallet: null
 ethereum_rpc_url: null
+ethereum_rpc_ws_url: null
 ethereum_chain_name: MAIN_NET
 ethereum_token_overrides: {}
 
diff --git a/hummingbot/user/user_balances.py b/hummingbot/user/user_balances.py
index 571368f337a..e7487c53359 100644
--- a/hummingbot/user/user_balances.py
+++ b/hummingbot/user/user_balances.py
@@ -132,6 +132,8 @@ def validate_ethereum_wallet() -> Optional[str]:
             return "Ethereum wallet is required."
         if global_config_map.get("ethereum_rpc_url").value is None:
             return "ethereum_rpc_url is required."
+        if global_config_map.get("ethereum_rpc_ws_url").value is None:
+            return "ethereum_rpc_ws_url is required."
         if global_config_map.get("ethereum_wallet").value not in Security.private_keys():
             return "Ethereum private key file does not exist or corrupts."
         try:
diff --git a/hummingbot/wallet/ethereum/watcher/account_balance_watcher.py b/hummingbot/wallet/ethereum/watcher/account_balance_watcher.py
index bb2c48c5e0c..e7f9f5b4814 100644
--- a/hummingbot/wallet/ethereum/watcher/account_balance_watcher.py
+++ b/hummingbot/wallet/ethereum/watcher/account_balance_watcher.py
@@ -23,7 +23,7 @@
     safe_gather,
 )
 from .base_watcher import BaseWatcher
-from .new_blocks_watcher import NewBlocksWatcher
+from .websocket_watcher import WSNewBlocksWatcher
 
 s_decimal_0 = Decimal(0)
 
@@ -39,12 +39,12 @@ def logger(cls) -> HummingbotLogger:
 
     def __init__(self,
                  w3: Web3,
-                 blocks_watcher: NewBlocksWatcher,
+                 blocks_watcher: WSNewBlocksWatcher,
                  account_address: str,
                  erc20_addresses: List[str],
                  erc20_abis: List[any]):
         super().__init__(w3)
-        self._blocks_watcher: NewBlocksWatcher = blocks_watcher
+        self._blocks_watcher: WSNewBlocksWatcher = blocks_watcher
         self._account_address: str = account_address
         self._addresses_to_contracts: Dict[str, Contract] = {
             address: w3.eth.contract(address=address, abi=abi)
@@ -148,7 +148,7 @@ async def update_balances(self):
         except asyncio.CancelledError:
             raise
         except Exception:
-            self.logger().network(f"Error fetching account balance updates.",
+            self.logger().network("Error fetching account balance updates.",
                                   exc_info=True,
-                                  app_warning_msg=f"Error account balance updates. "
-                                                  f"Check Ethereum node connection.")
+                                  app_warning_msg="Error account balance updates. "
+                                                  "Check Ethereum node connection.")
diff --git a/hummingbot/wallet/ethereum/watcher/erc20_events_watcher.py b/hummingbot/wallet/ethereum/watcher/erc20_events_watcher.py
index ae67b43a3d4..5ac7fa4f227 100644
--- a/hummingbot/wallet/ethereum/watcher/erc20_events_watcher.py
+++ b/hummingbot/wallet/ethereum/watcher/erc20_events_watcher.py
@@ -2,7 +2,6 @@
 
 import asyncio
 import cytoolz
-from hexbytes import HexBytes
 import logging
 import math
 from typing import (
@@ -30,7 +29,7 @@
     safe_gather,
 )
 from .base_watcher import BaseWatcher
-from .new_blocks_watcher import NewBlocksWatcher
+from .websocket_watcher import WSNewBlocksWatcher
 from .contract_event_logs import ContractEventLogger
 
 weth_sai_symbols: Set[str] = {"WETH", "SAI"}
@@ -54,7 +53,7 @@ def is_weth_sai(symbol: str) -> bool:
 
     def __init__(self,
                  w3: Web3,
-                 blocks_watcher: NewBlocksWatcher,
+                 blocks_watcher: WSNewBlocksWatcher,
                  contract_addresses: List[str],
                  contract_abi: List[any],
                  watch_addresses: Iterable[str]):
@@ -62,7 +61,7 @@ def __init__(self,
             raise ValueError("Each entry in contract_addresses must have a corresponding entry in contract_abi.")
 
         super().__init__(w3)
-        self._blocks_watcher: NewBlocksWatcher = blocks_watcher
+        self._blocks_watcher: WSNewBlocksWatcher = blocks_watcher
         self._addresses_to_contracts: Dict[str, Contract] = {
             address: w3.eth.contract(address=address, abi=abi)
             for address, abi in zip(contract_addresses, contract_abi)
@@ -141,9 +140,9 @@ async def poll_erc20_logs_loop(self):
             except asyncio.TimeoutError:
                 continue
             except Exception:
-                self.logger().network(f"Error fetching new events from ERC20 contracts.", exc_info=True,
-                                      app_warning_msg=f"Error fetching new events from ERC20 contracts. "
-                                                      f"Check wallet network connection")
+                self.logger().network("Error fetching new events from ERC20 contracts.", exc_info=True,
+                                      app_warning_msg="Error fetching new events from ERC20 contracts. "
+                                                      "Check wallet network connection")
 
     async def _handle_event_data(self, event_data: AttributeDict):
         event_type: str = event_data["event"]
diff --git a/hummingbot/wallet/ethereum/watcher/incoming_eth_watcher.py b/hummingbot/wallet/ethereum/watcher/incoming_eth_watcher.py
index b51ebadcde3..e6039e972ab 100644
--- a/hummingbot/wallet/ethereum/watcher/incoming_eth_watcher.py
+++ b/hummingbot/wallet/ethereum/watcher/incoming_eth_watcher.py
@@ -2,6 +2,7 @@
 
 import asyncio
 import cytoolz
+import logging
 from typing import (
     List,
     Set,
@@ -26,19 +27,26 @@
 )
 from hummingbot.logger import HummingbotLogger
 from .base_watcher import BaseWatcher
-from .new_blocks_watcher import NewBlocksWatcher
+# from .new_blocks_watcher import NewBlocksWatcher
+from .websocket_watcher import WSNewBlocksWatcher
 
 
 class IncomingEthWatcher(BaseWatcher):
-    logger: Optional[HummingbotLogger] = None
+    _iew_logger: Optional[HummingbotLogger] = None
+
+    @classmethod
+    def logger(cls) -> HummingbotLogger:
+        if cls._iew_logger is None:
+            cls._iew_logger = logging.getLogger(__name__)
+        return cls._iew_logger
 
     def __init__(self,
                  w3: Web3,
-                 blocks_watcher: NewBlocksWatcher,
+                 blocks_watcher: WSNewBlocksWatcher,
                  watch_addresses: Iterable[str]):
         super().__init__(w3)
         self._watch_addresses: Set[str] = set(watch_addresses)
-        self._blocks_watcher: NewBlocksWatcher = blocks_watcher
+        self._blocks_watcher: WSNewBlocksWatcher = blocks_watcher
         self._event_forwarder: EventForwarder = EventForwarder(self.did_receive_new_blocks)
 
     async def start_network(self):
diff --git a/hummingbot/wallet/ethereum/watcher/websocket_watcher.py b/hummingbot/wallet/ethereum/watcher/websocket_watcher.py
new file mode 100644
index 00000000000..0a7483b11ae
--- /dev/null
+++ b/hummingbot/wallet/ethereum/watcher/websocket_watcher.py
@@ -0,0 +1,169 @@
+import websockets
+from web3 import Web3
+from web3.exceptions import BlockNotFound
+from websockets.exceptions import ConnectionClosedOK
+import logging
+import ujson
+import asyncio
+
+from hexbytes import HexBytes
+from web3.datastructures import AttributeDict
+from cachetools import TTLCache
+
+from typing import Optional, Dict
+
+from hummingbot.core.utils.async_utils import safe_ensure_future
+from hummingbot.wallet.ethereum.watcher.base_watcher import BaseWatcher
+from hummingbot.logger import HummingbotLogger
+from hummingbot.core.event.events import NewBlocksWatcherEvent
+
+
+class WSNewBlocksWatcher(BaseWatcher):
+    def __init__(self, w3: Web3, websocket_url):
+        super().__init__(w3)
+        self._network_on = False
+        self._nonce: int = 0
+        self._current_block_number: int = -1
+        self._websocket_url = websocket_url
+        self._node_address = None
+        self._client: Optional[websockets.WebSocketClientProtocol] = None
+        self._fetch_new_blocks_task: Optional[asyncio.Task] = None
+        self._block_cache = TTLCache(maxsize=10, ttl=120)
+
+    _nbw_logger: Optional[HummingbotLogger] = None
+
+    @classmethod
+    def logger(cls) -> HummingbotLogger:
+        if cls._nbw_logger is None:
+            cls._nbw_logger = logging.getLogger(__name__)
+        return cls._nbw_logger
+
+    @property
+    def block_number(self) -> int:
+        return self._current_block_number
+
+    @property
+    def block_cache(self) -> Dict[HexBytes, AttributeDict]:
+        cache_dict: Dict[HexBytes, AttributeDict] = dict([(key, self._block_cache[key])
+                                                          for key in self._block_cache.keys()])
+        return cache_dict
+
+    async def start_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.stop_network()
+        else:
+            try:
+                self._current_block_number = await self.call_async(getattr, self._w3.eth, "blockNumber")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                self.logger().network("Error fetching newest Ethereum block number.",
+                                      app_warning_msg="Error fetching newest Ethereum block number. "
+                                                      "Check Ethereum node connection",
+                                      exc_info=True)
+            await self.connect()
+            await self.subscribe(["newHeads"])
+            self._fetch_new_blocks_task: asyncio.Task = safe_ensure_future(self.fetch_new_blocks_loop())
+            self._network_on = True
+
+    async def stop_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.disconnect()
+            self._fetch_new_blocks_task.cancel()
+            self._fetch_new_blocks_task = None
+            self._network_on = False
+
+    async def connect(self):
+        try:
+            self._client = await websockets.connect(uri=self._websocket_url)
+            return self._client
+        except Exception as e:
+            self.logger().network(f"ERROR in connection: {e}")
+
+    async def disconnect(self):
+        try:
+            await self._client.close()
+        except Exception as e:
+            self.logger().network(f"ERROR in disconnection: {e}")
+
+    async def _send(self, emit_data) -> int:
+        self._nonce += 1
+        emit_data["id"] = self._nonce
+        await self._client.send(ujson.dumps(emit_data))
+        return self._nonce
+
+    async def subscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_subscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        if raw_message is not None:
+            resp = ujson.loads(raw_message)
+            if resp.get("id", None) == nonce:
+                self._node_address = resp.get("result")
+                return True
+        return False
+
+    async def unsubscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_unsubscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        resp = ujson.loads(raw_message)
+        if resp.get("id", None) == nonce:
+            result = resp.get("result", None)
+            return (result is not None) and result
+
+    async def fetch_new_blocks_loop(self):
+        try:
+            while True:
+                try:
+                    raw_message = await self._client.recv()
+                    message_json = ujson.loads(raw_message) if raw_message is not None else None
+                    if message_json.get("method", None) == "eth_subscription":
+                        subscription_result_params = message_json.get("params", None)
+                        incoming_block = subscription_result_params.get("result", None) \
+                            if subscription_result_params is not None else None
+                        if incoming_block is not None:
+                            new_block: AttributeDict = await self.call_async(self._w3.eth.getBlock,
+                                                                             incoming_block.get("hash"), True)
+                            self._current_block_number = new_block.get("number")
+                            self._block_cache[new_block.get("hash")] = new_block
+                            self.trigger_event(NewBlocksWatcherEvent.NewBlocks, [new_block])
+                except asyncio.CancelledError:
+                    if self._network_on:
+                        self.logger().network("Network closed abnormally, reconnecting ...")
+                        self.connect()
+                    raise
+                except asyncio.TimeoutError:
+                    self.logger().network("Timed out fetching new block.", exc_info=True,
+                                          app_warning_msg="Timed out fetching new block. "
+                                                          "Check wallet network connection")
+                except BlockNotFound:
+                    pass
+                except ConnectionClosedOK:
+                    raise
+                except Exception as e:
+                    self.logger().network(f"Error fetching new block: {e}", exc_info=True,
+                                          app_warning_msg="Error fetching new block. "
+                                                          "Check wallet network connection")
+        except asyncio.CancelledError:
+            raise
+
+    async def get_timestamp_for_block(self, block_hash: HexBytes, max_tries: Optional[int] = 10) -> int:
+        counter = 0
+        block: AttributeDict = None
+        if block_hash in self._block_cache.keys():
+            block = self._block_cache.get(block_hash)
+        else:
+            while block is None:
+                if counter == max_tries:
+                    raise ValueError(f"Block hash {block_hash.hex()} does not exist.")
+                counter += 1
+                block = self._block_cache.get(block_hash)
+                await asyncio.sleep(0.5)
+        return block.get("timestamp")
diff --git a/hummingbot/wallet/ethereum/watcher/weth_watcher.py b/hummingbot/wallet/ethereum/watcher/weth_watcher.py
index f459f3b2a9d..09999f9066d 100644
--- a/hummingbot/wallet/ethereum/watcher/weth_watcher.py
+++ b/hummingbot/wallet/ethereum/watcher/weth_watcher.py
@@ -1,7 +1,6 @@
 #!/usr/bin/env python
 
 import asyncio
-from hexbytes import HexBytes
 from decimal import Decimal
 import logging
 from typing import (
@@ -25,7 +24,7 @@
 from hummingbot.core.event.event_forwarder import EventForwarder
 from hummingbot.core.utils.async_utils import safe_ensure_future
 from .base_watcher import BaseWatcher
-from .new_blocks_watcher import NewBlocksWatcher
+from .websocket_watcher import WSNewBlocksWatcher
 from .contract_event_logs import ContractEventLogger
 
 DEPOSIT_EVENT_NAME = "Deposit"
@@ -44,10 +43,10 @@ def logger(cls) -> HummingbotLogger:
     def __init__(self,
                  w3: Web3,
                  weth_token: ERC20Token,
-                 blocks_watcher: NewBlocksWatcher,
+                 blocks_watcher: WSNewBlocksWatcher,
                  watch_addresses: Iterable[str]):
         super().__init__(w3)
-        self._blocks_watcher: NewBlocksWatcher = blocks_watcher
+        self._blocks_watcher: WSNewBlocksWatcher = blocks_watcher
         self._watch_addresses: Set[str] = set(watch_addresses)
         self._asset_decimals: Dict[str, int] = {}
         self._weth_token = weth_token
@@ -96,9 +95,9 @@ async def poll_weth_logs_loop(self):
             except asyncio.TimeoutError:
                 continue
             except Exception:
-                self.logger().network(f"Unknown error trying to fetch new events from WETH contract.", exc_info=True,
-                                      app_warning_msg=f"Unknown error trying to fetch new events from WETH contract. "
-                                                      f"Check wallet network connection")
+                self.logger().network("Unknown error trying to fetch new events from WETH contract.", exc_info=True,
+                                      app_warning_msg="Unknown error trying to fetch new events from WETH contract. "
+                                                      "Check wallet network connection")
 
     async def _handle_event_data(self, event_data: AttributeDict):
         event_type: str = event_data["event"]
diff --git a/hummingbot/wallet/ethereum/watcher/zeroex_fill_watcher.py b/hummingbot/wallet/ethereum/watcher/zeroex_fill_watcher.py
index d3c85d6457d..985e8d783c8 100644
--- a/hummingbot/wallet/ethereum/watcher/zeroex_fill_watcher.py
+++ b/hummingbot/wallet/ethereum/watcher/zeroex_fill_watcher.py
@@ -34,7 +34,8 @@
 from hummingbot.core.utils.async_call_scheduler import AsyncCallScheduler
 from hummingbot.core.utils.async_utils import safe_ensure_future
 from .base_watcher import BaseWatcher
-from .new_blocks_watcher import NewBlocksWatcher
+# from .new_blocks_watcher import NewBlocksWatcher
+from .websocket_watcher import WSNewBlocksWatcher
 
 with open(realpath(join(__file__, "../../zero_ex/zero_ex_exchange_abi_v3.json"))) as exchange_abi_json:
     exchange_abi: List[any] = ujson.load(exchange_abi_json)
@@ -55,9 +56,9 @@ def logger(cls) -> HummingbotLogger:
 
     def __init__(self,
                  w3: Web3,
-                 blocks_watcher: NewBlocksWatcher):
+                 blocks_watcher: WSNewBlocksWatcher):
         super().__init__(w3)
-        self._blocks_watcher: NewBlocksWatcher = blocks_watcher
+        self._blocks_watcher: WSNewBlocksWatcher = blocks_watcher
         self._poll_fill_logs_task: asyncio.Task = None
         self._event_forwarder: EventForwarder = EventForwarder(self.did_receive_new_blocks)
         self._new_blocks_queue: asyncio.Queue = asyncio.Queue()
@@ -139,9 +140,9 @@ async def poll_zeroex_logs_loop(self):
             except asyncio.TimeoutError:
                 continue
             except Exception:
-                self.logger().network(f"Unknown error trying to fetch new events for ZeroEx fills.", exc_info=True,
-                                      app_warning_msg=f"Unknown error trying to fetch new events for ZeroEx fills. "
-                                                      f"Check wallet network connection")
+                self.logger().network("Unknown error trying to fetch new events for ZeroEx fills.", exc_info=True,
+                                      app_warning_msg="Unknown error trying to fetch new events for ZeroEx fills. "
+                                                      "Check wallet network connection")
 
     async def _get_logs(self,
                         event_filter_params: Dict[str, any],
diff --git a/hummingbot/wallet/ethereum/web3_wallet_backend.py b/hummingbot/wallet/ethereum/web3_wallet_backend.py
index 61d335b544f..7724b1350a3 100644
--- a/hummingbot/wallet/ethereum/web3_wallet_backend.py
+++ b/hummingbot/wallet/ethereum/web3_wallet_backend.py
@@ -50,15 +50,16 @@
     safe_gather,
 )
 from hummingbot.wallet.ethereum.watcher import (
-    NewBlocksWatcher,
     AccountBalanceWatcher,
     ERC20EventsWatcher,
     IncomingEthWatcher,
     WethWatcher,
     ZeroExFillWatcher,
 )
+from hummingbot.wallet.ethereum.watcher.websocket_watcher import WSNewBlocksWatcher
 from hummingbot.wallet.ethereum.erc20_token import ERC20Token
 from hummingbot.logger import HummingbotLogger
+from hummingbot.client.config.global_config_map import global_config_map
 
 s_decimal_0 = Decimal(0)
 
@@ -117,7 +118,7 @@ def __init__(self,
         self._local_nonce: int = -1
 
         # Watchers
-        self._new_blocks_watcher: Optional[NewBlocksWatcher] = None
+        self._new_blocks_watcher: Optional[WSNewBlocksWatcher] = None
         self._account_balance_watcher: Optional[AccountBalanceWatcher] = None
         self._erc20_events_watcher: Optional[ERC20EventsWatcher] = None
         self._incoming_eth_watcher: Optional[IncomingEthWatcher] = None
@@ -231,8 +232,9 @@ async def start_network(self):
             lambda: self.get_remote_nonce()
         )
 
-            # Create event watchers.
-        self._new_blocks_watcher = NewBlocksWatcher(self._w3)
+        # Create event watchers.
+        websocket_url: str = global_config_map["ethereum_rpc_ws_url"].value
+        self._new_blocks_watcher = WSNewBlocksWatcher(self._w3, websocket_url)
         self._account_balance_watcher = AccountBalanceWatcher(
             self._w3,
             self._new_blocks_watcher,
@@ -355,9 +357,9 @@ async def _check_network_loop(self):
             except asyncio.TimeoutError:
                 new_status = NetworkStatus.NOT_CONNECTED
             except Exception:
-                self.logger().network(f"Unexpected error while checking for network status.", exc_info=True,
-                                      app_warning_msg=f"Unexpected error while checking for network status. "
-                                                      f"Check wallet network connection")
+                self.logger().network("Unexpected error while checking for network status.", exc_info=True,
+                                      app_warning_msg="Unexpected error while checking for network status. "
+                                                      "Check wallet network connection")
                 new_status = NetworkStatus.NOT_CONNECTED
 
             self._network_status = new_status
@@ -376,9 +378,9 @@ async def check_transaction_receipts_loop(self):
                 raise
             except Exception:
                 self.logger().network(
-                    f"Unknown error occurred while checking for transaction receipts.", exc_info=True,
-                    app_warning_msg=f"Unknown error occurred while checking for transaction receipts. "
-                                    f"Check wallet network connection")
+                    "Unknown error occurred while checking for transaction receipts.", exc_info=True,
+                    app_warning_msg="Unknown error occurred while checking for transaction receipts. "
+                                    "Check wallet network connection")
                 await asyncio.sleep(5.0)
 
     async def _check_transaction_receipt(self, tx_hash: str, timestamp: int):
@@ -528,7 +530,7 @@ def execute_transaction(self, contract_function: ContractFunction, **kwargs) ->
             try:
                 estimate_gas = self._w3.eth.estimateGas(transaction)
             except ValueError:
-                self.logger().error(f"Failed to estimate gas. Using default of 1000000.")
+                self.logger().error("Failed to estimate gas. Using default of 1000000.")
             transaction["gas"] = estimate_gas
         signed_transaction: AttributeDict = self._account.signTransaction(transaction)
         tx_hash: str = signed_transaction.hash.hex()
@@ -597,7 +599,7 @@ def to_raw(self, asset_name: str, nominal_amount: Decimal) -> int:
 
     @staticmethod
     def to_raw_static(nominal_amount: Decimal) -> int:
-        return int(nominal_amount * Decimal(f"1e18"))
+        return int(nominal_amount * Decimal("1e18"))
 
     def _received_asset_event_listener(self, received_asset_event: WalletReceivedAssetEvent):
         self.logger().info(f"Received {received_asset_event.amount_received} {received_asset_event.asset_name} at "


Code review comments:
diff_hunk: @@ -0,0 +1,169 @@
+import websockets
+from web3 import Web3
+from web3.exceptions import BlockNotFound
+from websockets.exceptions import ConnectionClosedOK
+import logging
+import ujson
+import asyncio
+
+from hexbytes import HexBytes
+from web3.datastructures import AttributeDict
+from cachetools import TTLCache
+
+from typing import Optional, Dict
+
+from hummingbot.core.utils.async_utils import safe_ensure_future
+from hummingbot.wallet.ethereum.watcher.base_watcher import BaseWatcher
+from hummingbot.logger import HummingbotLogger
+from hummingbot.core.event.events import NewBlocksWatcherEvent
+
+
+class WSNewBlocksWatcher(BaseWatcher):
+    def __init__(self, w3: Web3, websocket_url):
+        super().__init__(w3)
+        self._network_on = False
+        self._nonce: int = 0
+        self._current_block_number: int = -1
+        self._websocket_url = websocket_url
+        self._node_address = None
+        self._client: Optional[websockets.WebSocketClientProtocol] = None
+        self._fetch_new_blocks_task: Optional[asyncio.Task] = None
+        self._block_cache = TTLCache(maxsize=10, ttl=120)
+
+    _nbw_logger: Optional[HummingbotLogger] = None
+
+    @classmethod
+    def logger(cls) -> HummingbotLogger:
+        if cls._nbw_logger is None:
+            cls._nbw_logger = logging.getLogger(__name__)
+        return cls._nbw_logger
+
+    @property
+    def block_number(self) -> int:
+        return self._current_block_number
+
+    @property
+    def block_cache(self) -> Dict[HexBytes, AttributeDict]:
+        cache_dict: Dict[HexBytes, AttributeDict] = dict([(key, self._block_cache[key])
+                                                          for key in self._block_cache.keys()])
+        return cache_dict
+
+    async def start_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.stop_network()
+        else:
+            try:
+                self._current_block_number = await self.call_async(getattr, self._w3.eth, "blockNumber")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                self.logger().network("Error fetching newest Ethereum block number.",
+                                      app_warning_msg="Error fetching newest Ethereum block number. "
+                                                      "Check Ethereum node connection",
+                                      exc_info=True)
+            await self.connect()
+            await self.subscribe(["newHeads"])
+            self._fetch_new_blocks_task: asyncio.Task = safe_ensure_future(self.fetch_new_blocks_loop())
+            self._network_on = True
+
+    async def stop_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.disconnect()
+            self._fetch_new_blocks_task.cancel()
+            self._fetch_new_blocks_task = None
+            self._network_on = False
+
+    async def connect(self):
+        try:
+            self._client = await websockets.connect(uri=self._websocket_url)
+            return self._client
+        except Exception as e:
+            self.logger().network(f"ERROR in connection: {e}")
+
+    async def disconnect(self):
+        try:
+            await self._client.close()
+        except Exception as e:
+            self.logger().network(f"ERROR in disconnection: {e}")
+
+    async def _send(self, emit_data) -> int:
+        self._nonce += 1
+        emit_data["id"] = self._nonce
+        await self._client.send(ujson.dumps(emit_data))
+        return self._nonce
+
+    async def subscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_subscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        if raw_message is not None:
+            resp = ujson.loads(raw_message)
+            if resp.get("id", None) == nonce:
+                self._node_address = resp.get("result")
+                return True
+        return False
+
+    async def unsubscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_unsubscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        resp = ujson.loads(raw_message)
+        if resp.get("id", None) == nonce:
+            result = resp.get("result", None)
+            return (result is not None) and result
+
+    async def fetch_new_blocks_loop(self):
+        try:
+            while True:
+                try:
+                    raw_message = await self._client.recv()
+                    message_json = ujson.loads(raw_message) if raw_message is not None else None
+                    if message_json.get("method", None) == "eth_subscription":
+                        subscription_result_params = message_json.get("params", None)
+                        incoming_block = subscription_result_params.get("result", None) \
+                            if subscription_result_params is not None else None
+                        if incoming_block is not None:
+                            new_block: AttributeDict = await self.call_async(self._w3.eth.getBlock,
+                                                                             incoming_block.get("hash"), True)
+                            self._current_block_number = new_block.get("number")
+                            self._block_cache[new_block.get("hash")] = new_block
+                            self.trigger_event(NewBlocksWatcherEvent.NewBlocks, [new_block])
+                except asyncio.CancelledError:
+                    if self._network_on:
+                        self.logger().network("Network closed abnormally, reconnecting ...")
+                        self.connect()
+                    raise
+                except asyncio.TimeoutError:
+                    self.logger().network("Timed out fetching new block.", exc_info=True,
+                                          app_warning_msg="Timed out fetching new block. "
+                                                          "Check wallet network connection")
+                except BlockNotFound:
+                    pass
+                except ConnectionClosedOK:
+                    raise
+                except Exception as e:
+                    self.logger().network(f"Error fetching new block: {e}", exc_info=True,
+                                          app_warning_msg="Error fetching new block. "
+                                                          "Check wallet network connection")
+        except asyncio.CancelledError:
path: hummingbot/wallet/ethereum/watcher/websocket_watcher.py
text: @user1:
This except and raise here is unnecessary, you need to do this only when you also catch other kinds of Exceptions.

@user1:
You don't need to except asyncio.CancelledError here, but you do need one in your inner try-except block.

diff_hunk: @@ -0,0 +1,169 @@
+import websockets
+from web3 import Web3
+from web3.exceptions import BlockNotFound
+from websockets.exceptions import ConnectionClosedOK
+import logging
+import ujson
+import asyncio
+
+from hexbytes import HexBytes
+from web3.datastructures import AttributeDict
+from cachetools import TTLCache
+
+from typing import Optional, Dict
+
+from hummingbot.core.utils.async_utils import safe_ensure_future
+from hummingbot.wallet.ethereum.watcher.base_watcher import BaseWatcher
+from hummingbot.logger import HummingbotLogger
+from hummingbot.core.event.events import NewBlocksWatcherEvent
+
+
+class WSNewBlocksWatcher(BaseWatcher):
+    def __init__(self, w3: Web3, websocket_url):
+        super().__init__(w3)
+        self._network_on = False
+        self._nonce: int = 0
+        self._current_block_number: int = -1
+        self._websocket_url = websocket_url
+        self._node_address = None
+        self._client: Optional[websockets.WebSocketClientProtocol] = None
+        self._fetch_new_blocks_task: Optional[asyncio.Task] = None
+        self._block_cache = TTLCache(maxsize=10, ttl=120)
+
+    _nbw_logger: Optional[HummingbotLogger] = None
+
+    @classmethod
+    def logger(cls) -> HummingbotLogger:
+        if cls._nbw_logger is None:
+            cls._nbw_logger = logging.getLogger(__name__)
+        return cls._nbw_logger
+
+    @property
+    def block_number(self) -> int:
+        return self._current_block_number
+
+    @property
+    def block_cache(self) -> Dict[HexBytes, AttributeDict]:
+        cache_dict: Dict[HexBytes, AttributeDict] = dict([(key, self._block_cache[key])
+                                                          for key in self._block_cache.keys()])
+        return cache_dict
+
+    async def start_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.stop_network()
+        else:
+            try:
+                self._current_block_number = await self.call_async(getattr, self._w3.eth, "blockNumber")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                self.logger().network("Error fetching newest Ethereum block number.",
+                                      app_warning_msg="Error fetching newest Ethereum block number. "
+                                                      "Check Ethereum node connection",
+                                      exc_info=True)
+            await self.connect()
+            await self.subscribe(["newHeads"])
+            self._fetch_new_blocks_task: asyncio.Task = safe_ensure_future(self.fetch_new_blocks_loop())
+            self._network_on = True
+
+    async def stop_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.disconnect()
+            self._fetch_new_blocks_task.cancel()
+            self._fetch_new_blocks_task = None
+            self._network_on = False
+
+    async def connect(self):
+        try:
+            self._client = await websockets.connect(uri=self._websocket_url)
+            return self._client
+        except Exception as e:
+            self.logger().network(f"ERROR in connection: {e}")
+
+    async def disconnect(self):
+        try:
+            await self._client.close()
+        except Exception as e:
+            self.logger().network(f"ERROR in disconnection: {e}")
+
+    async def _send(self, emit_data) -> int:
+        self._nonce += 1
+        emit_data["id"] = self._nonce
+        await self._client.send(ujson.dumps(emit_data))
+        return self._nonce
+
+    async def subscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_subscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        if raw_message is not None:
+            resp = ujson.loads(raw_message)
+            if resp.get("id", None) == nonce:
+                self._node_address = resp.get("result")
+                return True
+        return False
+
+    async def unsubscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_unsubscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        resp = ujson.loads(raw_message)
+        if resp.get("id", None) == nonce:
+            result = resp.get("result", None)
+            return (result is not None) and result
+
+    async def fetch_new_blocks_loop(self):
+        try:
+            while True:
+                try:
+                    raw_message = await self._client.recv()
+                    message_json = ujson.loads(raw_message) if raw_message is not None else None
+                    if message_json.get("method", None) == "eth_subscription":
+                        subscription_result_params = message_json.get("params", None)
+                        incoming_block = subscription_result_params.get("result", None) \
+                            if subscription_result_params is not None else None
+                        if incoming_block is not None:
+                            new_block: AttributeDict = await self.call_async(self._w3.eth.getBlock,
+                                                                             incoming_block.get("hash"), True)
+                            self._current_block_number = new_block.get("number")
+                            self._block_cache[new_block.get("hash")] = new_block
+                            self.trigger_event(NewBlocksWatcherEvent.NewBlocks, [new_block])
+                except asyncio.CancelledError:
+                    if self._network_on:
+                        self.logger().network("Network closed abnormally, reconnecting ...")
+                        self.connect()
path: hummingbot/wallet/ethereum/watcher/websocket_watcher.py
text: @user1:
Not sure if we should `connect()` here on this exception, seems to me the task is cancelled and exit from here. To connect here might be unnecessary. 

@author:
The Websocket sometimes "abnormally" disconnects, for example when my computer sleeps with the bot running; this accounts for the "abnormal disconnections" when there's a CancelledError but the WebSocket is still intended to be connected; this replaces the self._client with a connected WebSocket. Should I figure out a workaround or just remove it?

@user1:
Does the `CancelledError` raise on those situations, like when the websocket abnormally disconnects? From what I understand the CancelledError is raised only when `cancel()` is called, and it is a way to wrap up and finish the task. If you need to reconnect, it's fine but I don't think this is the right place for it.

@author:
Made a mistake. Wrong location; I put it in ConnectionClosedError from WebSockets instead. This is for when connectionclosed errors codes are not 1000 (OK) or 1001 ("Going Away"). Trying to figure out if there's a way to check connection another way in the fetch loop

@author:
@user1 pushed up changes, what do you think?

diff_hunk: @@ -0,0 +1,169 @@
+import websockets
+from web3 import Web3
+from web3.exceptions import BlockNotFound
+from websockets.exceptions import ConnectionClosedOK
+import logging
+import ujson
+import asyncio
+
+from hexbytes import HexBytes
+from web3.datastructures import AttributeDict
+from cachetools import TTLCache
+
+from typing import Optional, Dict
+
+from hummingbot.core.utils.async_utils import safe_ensure_future
+from hummingbot.wallet.ethereum.watcher.base_watcher import BaseWatcher
+from hummingbot.logger import HummingbotLogger
+from hummingbot.core.event.events import NewBlocksWatcherEvent
+
+
+class WSNewBlocksWatcher(BaseWatcher):
+    def __init__(self, w3: Web3, websocket_url):
+        super().__init__(w3)
+        self._network_on = False
+        self._nonce: int = 0
+        self._current_block_number: int = -1
+        self._websocket_url = websocket_url
+        self._node_address = None
+        self._client: Optional[websockets.WebSocketClientProtocol] = None
+        self._fetch_new_blocks_task: Optional[asyncio.Task] = None
+        self._block_cache = TTLCache(maxsize=10, ttl=120)
+
+    _nbw_logger: Optional[HummingbotLogger] = None
+
+    @classmethod
+    def logger(cls) -> HummingbotLogger:
+        if cls._nbw_logger is None:
+            cls._nbw_logger = logging.getLogger(__name__)
+        return cls._nbw_logger
+
+    @property
+    def block_number(self) -> int:
+        return self._current_block_number
+
+    @property
+    def block_cache(self) -> Dict[HexBytes, AttributeDict]:
+        cache_dict: Dict[HexBytes, AttributeDict] = dict([(key, self._block_cache[key])
+                                                          for key in self._block_cache.keys()])
+        return cache_dict
+
+    async def start_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.stop_network()
+        else:
+            try:
+                self._current_block_number = await self.call_async(getattr, self._w3.eth, "blockNumber")
+            except asyncio.CancelledError:
+                raise
+            except Exception:
+                self.logger().network("Error fetching newest Ethereum block number.",
+                                      app_warning_msg="Error fetching newest Ethereum block number. "
+                                                      "Check Ethereum node connection",
+                                      exc_info=True)
+            await self.connect()
+            await self.subscribe(["newHeads"])
+            self._fetch_new_blocks_task: asyncio.Task = safe_ensure_future(self.fetch_new_blocks_loop())
+            self._network_on = True
+
+    async def stop_network(self):
+        if self._fetch_new_blocks_task is not None:
+            await self.disconnect()
+            self._fetch_new_blocks_task.cancel()
+            self._fetch_new_blocks_task = None
+            self._network_on = False
+
+    async def connect(self):
+        try:
+            self._client = await websockets.connect(uri=self._websocket_url)
+            return self._client
+        except Exception as e:
+            self.logger().network(f"ERROR in connection: {e}")
+
+    async def disconnect(self):
+        try:
+            await self._client.close()
+        except Exception as e:
+            self.logger().network(f"ERROR in disconnection: {e}")
+
+    async def _send(self, emit_data) -> int:
+        self._nonce += 1
+        emit_data["id"] = self._nonce
+        await self._client.send(ujson.dumps(emit_data))
+        return self._nonce
+
+    async def subscribe(self, params) -> bool:
+        emit_data = {
+            "method": "eth_subscribe",
+            "params": params
+        }
+        nonce = await self._send(emit_data)
+        raw_message = await self._client.recv()
+        if raw_message is not None:
+            resp = ujson.loads(raw_message)
+            if resp.get("id", None) == nonce:
+                self._node_address = resp.get("result")
+                return True
+        return False
+
+    async def unsubscribe(self, params) -> bool:
path: hummingbot/wallet/ethereum/watcher/websocket_watcher.py
text: @user1:
Do we need this? Is it used anywhere? 

@author:
It's not used, but would be necessary if this became generalized for handling WebSockets. I can remove it