Revery concurrent_http changes and snapshot refresh changes due to bugs

Author: bmoscon

cryptofeed/exchanges/binance.py

@@ -4,14 +4,13 @@
 Please see the LICENSE file for the terms and conditions
 associated with this software.
 '''
-from asyncio import create_task
 from decimal import Decimal
 import logging
 from typing import List, Tuple, Callable, Dict
 
 from yapic import json
 
-from cryptofeed.connection import AsyncConnection, HTTPPoll, HTTPConcurrentPoll
+from cryptofeed.connection import AsyncConnection, HTTPPoll
 from cryptofeed.defines import BALANCES, BINANCE_FUTURES, BUY, FUNDING, LIMIT, LIQUIDATIONS, MARKET, OPEN_INTEREST, ORDER_INFO, POSITIONS, SELL
 from cryptofeed.exchanges.binance import Binance
 from cryptofeed.exchanges.mixins.binance_rest import BinanceFuturesRestMixin
@@ -98,7 +97,7 @@ def connect(self) -> List[Tuple[AsyncConnection, Callable[[None], None], Callabl
         ret = []
         if self.address:
             ret = super().connect()
-        PollCls = HTTPConcurrentPoll if self.concurrent_http else HTTPPoll
+        PollCls = HTTPPoll
         for chan in set(self.subscription):
             if chan == 'open_interest':
                 addrs = [f"{self.rest_endpoint}/openInterest?symbol={pair}" for pair in self.subscription[chan]]
@@ -244,10 +243,7 @@ async def message_handler(self, msg: str, conn: AsyncConnection, timestamp: floa
 
         # Handle REST endpoint messages first
         if 'openInterest' in msg:
-            coro = self._open_interest(msg, timestamp)
-            if self.concurrent_http:
-                return create_task(coro)
-            return await coro
+            return await self._open_interest(msg, timestamp)
 
         # Handle account updates from User Data Stream
         if self.requires_authentication:

cryptofeed/exchanges/binance_futures.py

@@ -2,7 +2,6 @@
 OrderBooks require a snapshot on initial subscription, hence connecting to a lot of symbols will eat up rate limits.
 
 Use a 'http_proxy' to bypass this limitation.
-Use 'concurrent_http' to make concurrent HTTP requests (used in polling and http GETs)
 
 Notes:
     1. 'http_proxy' will only be used for GET requests (not Websockets). For more information visit https://docs.aiohttp.org/en/stable/client_reference.html
@@ -34,8 +33,8 @@ def all_found(self):
                 return False
         return True
 
-    def callback(self, exchange, channel, symbols, concurrent_http):
-        concurrency = "[concurrent_http]" if concurrent_http else "[sync_http]"
+    def callback(self, exchange, channel, symbols):
+        concurrency = "[sync_http]"
         key = f'{exchange}:{channel} {concurrency}'
         self.counts[key] = defaultdict(int)
         self.total[key] = len(symbols)
@@ -83,28 +82,24 @@ def main(proxy):
     f = FeedHandler()
     counter = Counter(f)
     f.add_feed(Binance(depth_interval='1000ms',
-                       concurrent_http=False,
                        http_proxy=proxy,
                        max_depth=1,
                        symbols=book_symbols,
                        channels=[L2_BOOK],
                        callbacks={L2_BOOK: counter.callback(BINANCE, L2_BOOK, book_symbols, False)}))
     f.add_feed(Binance(depth_interval='1000ms',
-                       concurrent_http=True,
                        http_proxy=proxy,
                        max_depth=1,
                        symbols=book_symbols,
                        channels=[L2_BOOK],
                        callbacks={L2_BOOK: counter.callback(BINANCE, L2_BOOK, book_symbols, True)}))
     f.add_feed(BinanceFutures(http_proxy=proxy,
                               open_interest_interval=1.0,
-                              concurrent_http=False,
                               symbols=futures_symbols,
                               channels=[OPEN_INTEREST],
                               callbacks={OPEN_INTEREST: counter.callback(BINANCE_FUTURES, OPEN_INTEREST, futures_symbols, False)}))
     f.add_feed(BinanceFutures(http_proxy=proxy,
                               open_interest_interval=1.0,
-                              concurrent_http=True,
                               symbols=futures_symbols,
                               channels=[OPEN_INTEREST],
                               callbacks={OPEN_INTEREST: counter.callback(BINANCE_FUTURES, OPEN_INTEREST, futures_symbols, True)}))

examples/demo_concurrent_proxy.py

@@ -20,8 +20,8 @@
 
 lookup_table = {
     BEQUANT: {L2_BOOK: 978, TICKER: 1235, TRADES: 4040, CANDLES: 4040},
-    BINANCE: {L2_BOOK: 532, TICKER: 412, TRADES: 28, CANDLES: 17},
-    BINANCE_US: {TICKER: 241, TRADES: 3, L2_BOOK: 358, CANDLES: 3},
+    BINANCE: {L2_BOOK: 176, TICKER: 84, TRADES: 2, CANDLES: 2},
+    BINANCE_US: {TICKER: 128, TRADES: 11, L2_BOOK: 336, CANDLES: 5},
     BINANCE_FUTURES: {CANDLES: 67, L2_BOOK: 756, TICKER: 613, TRADES: 91},
     BINANCE_DELIVERY: {L2_BOOK: 1798, TICKER: 2240, TRADES: 51, CANDLES: 33},
     FMFW: {L2_BOOK: 1748, TICKER: 75, TRADES: 4, CANDLES: 4},