CTYPTO/freqtrade
1
0
Fork 0
mirror of https://github.com/freqtrade/freqtrade.git synced 2025-11-29 00:23:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve ccxt.ws live test

Browse Source
This commit is contained in:
Matthias
2023-11-21 06:23:26 +01:00
parent f9ce0bb9ab
commit bd9ebe4a72
1 changed files with 15 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
tests/exchange_online/test_ccxt_ws_compat.py
View File

@@ -6,12 +6,14 @@ However, these tests aim to test ccxt compatibility, specifically regarding webs
import logging
from datetime import timedelta
from time import sleep
import pytest
from freqtrade.enums import CandleType
from freqtrade.exchange.exchange_utils import (timeframe_to_minutes, timeframe_to_next_date,
timeframe_to_prev_date)
from freqtrade.loggers.set_log_levels import set_loggers
from freqtrade.util.datetime_helpers import dt_now
from tests.conftest import log_has_re
from tests.exchange_online.conftest import EXCHANGE_FIXTURE_TYPE, EXCHANGES
@@ -20,25 +22,31 @@ from tests.exchange_online.conftest import EXCHANGE_FIXTURE_TYPE, EXCHANGES
@pytest.mark.longrun
class TestCCXTExchangeWs:
def test_ccxt_ohlcv(self, exchange_ws: EXCHANGE_FIXTURE_TYPE, caplog):
def test_ccxt_ohlcv(self, exchange_ws: EXCHANGE_FIXTURE_TYPE, caplog, mocker):
exch, exchangename = exchange_ws
assert exch._ws_async is not None
pair = EXCHANGES[exchangename]['pair']
timeframe = '1m'
pair_tf = (pair, timeframe, CandleType.SPOT)
m_hist = mocker.spy(exch, '_async_get_historic_ohlcv')
m_cand = mocker.spy(exch, '_async_get_candle_history')
res = exch.refresh_latest_ohlcv([pair_tf])
now = dt_now() - timedelta(minutes=(timeframe_to_minutes(timeframe) * 1.1))
assert m_cand.call_count == 1
# Currently open candle
next_candle = timeframe_to_prev_date(timeframe, dt_now())
now = next_candle - timedelta(seconds=1)
# Currently closed candle
curr_candle = timeframe_to_prev_date(timeframe, now)
# Currently open candle
next_candle = timeframe_to_next_date(timeframe, now)
assert pair_tf in exch._exchange_ws._klines_watching
assert pair_tf in exch._exchange_ws._klines_scheduled
assert res[pair_tf] is not None
df1 = res[pair_tf]
caplog.set_level(logging.DEBUG)
set_loggers(1)
assert df1.iloc[-1]['date'] == curr_candle
# Wait until the next candle (might be up to 1 minute).
@@ -50,6 +58,8 @@ class TestCCXTExchangeWs:
if df2.iloc[-1]['date'] == next_candle:
break
assert df2.iloc[-1]['date'] == curr_candle
sleep(1)
assert m_hist.call_count == 0
assert m_cand.call_count == 1
assert log_has_re(r"watch result.*", caplog)