adelorenzo/stable
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Don't blindly create coroutines, but fire them off in batches

Browse Source
This commit is contained in:
Matthias 2021-09-10 05:52:48 +00:00
parent 7251a3ab19
commit a19c33ba54
1 changed files with 13 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
freqtrade/exchange/exchange.py
View File

@ -28,7 +28,7 @@ from freqtrade.exceptions import (DDosProtection, ExchangeError, InsufficientFun
from freqtrade.exchange.common import (API_FETCH_ORDER_RETRY_COUNT, BAD_EXCHANGES, from freqtrade.exchange.common import (API_FETCH_ORDER_RETRY_COUNT, BAD_EXCHANGES,
EXCHANGE_HAS_OPTIONAL, EXCHANGE_HAS_REQUIRED, retrier, EXCHANGE_HAS_OPTIONAL, EXCHANGE_HAS_REQUIRED, retrier,
retrier_async) retrier_async)
from freqtrade.misc import deep_merge_dicts, safe_value_fallback2 from freqtrade.misc import chunks, deep_merge_dicts, safe_value_fallback2
from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist from freqtrade.plugins.pairlist.pairlist_helpers import expand_pairlist
@ -1238,19 +1238,20 @@ class Exchange:
input_coroutines = [self._async_get_candle_history( input_coroutines = [self._async_get_candle_history(
pair, timeframe, since) for since in pair, timeframe, since) for since in
range(since_ms, arrow.utcnow().int_timestamp * 1000, one_call)] range(since_ms, arrow.utcnow().int_timestamp * 1000, one_call)]
results = await asyncio.gather(*input_coroutines, return_exceptions=True)
# Combine gathered results # Combine gathered results
data: List = [] data: List = []
for res in results: for input_coro in chunks(input_coroutines, 100):
if isinstance(res, Exception):
logger.warning("Async code raised an exception: %s", res.__class__.__name__) results = await asyncio.gather(*input_coro, return_exceptions=True)
continue for res in results:
# Deconstruct tuple if it's not an exception if isinstance(res, Exception):
p, _, new_data = res logger.warning("Async code raised an exception: %s", res.__class__.__name__)
if p == pair: continue
data.extend(new_data) # Deconstruct tuple if it's not an exception
p, _, new_data = res
if p == pair:
data.extend(new_data)
# Sort data again after extending the result - above calls return in "async order" # Sort data again after extending the result - above calls return in "async order"
data = sorted(data, key=lambda x: x[0]) data = sorted(data, key=lambda x: x[0])
logger.info("Downloaded data for %s with length %s.", pair, len(data)) logger.info("Downloaded data for %s with length %s.", pair, len(data))