Matthias
452 lines
15 KiB
Python
452 lines
15 KiB
Python
import asyncio
|
|
import logging
|
|
import threading
|
|
from datetime import timedelta
|
|
from time import sleep
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from ccxt import NotSupported
|
|
|
|
from freqtrade.enums import CandleType
|
|
from freqtrade.exceptions import TemporaryError
|
|
from freqtrade.exchange.exchange_ws import ExchangeWS
|
|
from ft_client.test_client.test_rest_client import log_has_re
|
|
|
|
|
|
def test_exchangews_init(mocker):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
sleep(0.1)
|
|
|
|
assert exchange_ws.config == config
|
|
assert exchange_ws._ccxt_object == ccxt_object
|
|
assert exchange_ws._thread.name == "ccxt_ws"
|
|
assert exchange_ws._background_tasks == set()
|
|
assert exchange_ws._klines_watching == set()
|
|
assert exchange_ws._klines_scheduled == set()
|
|
assert exchange_ws._klines_last_refresh == {}
|
|
assert exchange_ws._klines_last_request == {}
|
|
# Cleanup
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_cleanup_error(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
ccxt_object.close = AsyncMock(side_effect=Exception("Test"))
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
patch_eventloop_threading(exchange_ws)
|
|
|
|
sleep(0.1)
|
|
exchange_ws.reset_connections()
|
|
|
|
assert log_has_re("Exception in _cleanup_async", caplog)
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_reset_connections_timeout_and_exception(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws._loop = MagicMock()
|
|
exchange_ws._loop.is_closed.return_value = False
|
|
|
|
timeout_future = MagicMock()
|
|
timeout_future.result.side_effect = TimeoutError("timed out")
|
|
|
|
error_future = MagicMock()
|
|
error_future.result.side_effect = RuntimeError("broken future")
|
|
|
|
def fake_run_coroutine_threadsafe(coro, loop):
|
|
# Avoid coroutine warnings since we don't execute it in this unit test.
|
|
coro.close()
|
|
fake_run_coroutine_threadsafe.calls += 1
|
|
return timeout_future if fake_run_coroutine_threadsafe.calls == 1 else error_future
|
|
|
|
fake_run_coroutine_threadsafe.calls = 0
|
|
|
|
mock_run = mocker.patch(
|
|
"freqtrade.exchange.exchange_ws.asyncio.run_coroutine_threadsafe",
|
|
side_effect=fake_run_coroutine_threadsafe,
|
|
)
|
|
|
|
exchange_ws.reset_connections()
|
|
assert log_has_re("Timed out while resetting websocket connections", caplog)
|
|
assert log_has_re("Resetting exchange WS connections", caplog)
|
|
assert mock_run.call_count == 1
|
|
|
|
exchange_ws.reset_connections(cleanup=True)
|
|
|
|
assert mock_run.call_count == 2
|
|
assert log_has_re("Exception while resetting websocket connections", caplog)
|
|
assert log_has_re("Cleaning up exchange WS connections", caplog)
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_cleanup_thread_timeout_warning(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws._loop = MagicMock()
|
|
exchange_ws._loop.is_closed.return_value = True
|
|
|
|
thread_mock = MagicMock()
|
|
thread_mock.is_alive.return_value = True
|
|
exchange_ws._thread = thread_mock
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
thread_mock.join.assert_called_once_with(timeout=5)
|
|
assert log_has_re("Websocket loop thread did not stop within timeout", caplog)
|
|
|
|
|
|
def test_exchangews_schedule_ohlcv_loop_not_ready(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
run_threadsafe = mocker.patch("freqtrade.exchange.exchange_ws.asyncio.run_coroutine_threadsafe")
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
|
|
|
|
assert exchange_ws._klines_watching == set()
|
|
assert exchange_ws._klines_last_request == {}
|
|
assert run_threadsafe.call_count == 0
|
|
assert log_has_re("Websocket loop not ready. Could not schedule ETH/BTC, 1m", caplog)
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def patch_eventloop_threading(exchange):
|
|
init_event = threading.Event()
|
|
|
|
def thread_func():
|
|
exchange._loop = asyncio.new_event_loop()
|
|
init_event.set()
|
|
exchange._loop.run_forever()
|
|
|
|
x = threading.Thread(target=thread_func, daemon=True)
|
|
x.start()
|
|
# Wait for thread to be properly initialized with timeout
|
|
if not init_event.wait(timeout=5.0):
|
|
raise RuntimeError("Failed to initialize event loop thread")
|
|
|
|
|
|
async def test_exchangews_ohlcv(mocker, time_machine, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
caplog.set_level(logging.DEBUG)
|
|
|
|
async def controlled_sleeper(*args, **kwargs):
|
|
# Sleep to pass control back to the event loop
|
|
await asyncio.sleep(0.1)
|
|
return MagicMock()
|
|
|
|
async def wait_for_condition(condition_func, timeout_=5.0, check_interval=0.01):
|
|
"""Wait for a condition to be true with timeout."""
|
|
try:
|
|
async with asyncio.timeout(timeout_):
|
|
while True:
|
|
if condition_func():
|
|
return True
|
|
await asyncio.sleep(check_interval)
|
|
except TimeoutError:
|
|
return False
|
|
|
|
ccxt_object.un_watch_ohlcv_for_symbols = AsyncMock(side_effect=[NotSupported, ValueError])
|
|
ccxt_object.watch_ohlcv = AsyncMock(side_effect=controlled_sleeper)
|
|
ccxt_object.has = {"unWatchOHLCVForSymbols": True}
|
|
ccxt_object.close = AsyncMock()
|
|
time_machine.move_to("2024-11-01 01:00:02 +00:00")
|
|
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
patch_eventloop_threading(exchange_ws)
|
|
try:
|
|
assert exchange_ws._klines_watching == set()
|
|
assert exchange_ws._klines_scheduled == set()
|
|
|
|
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
|
|
exchange_ws.schedule_ohlcv("XRP/BTC", "1m", CandleType.SPOT)
|
|
|
|
# Wait for both pairs to be properly scheduled and watching
|
|
await wait_for_condition(
|
|
lambda: (
|
|
len(exchange_ws._klines_watching) == 2 and len(exchange_ws._klines_scheduled) == 2
|
|
),
|
|
timeout_=2.0,
|
|
)
|
|
|
|
assert exchange_ws._klines_watching == {
|
|
("ETH/BTC", "1m", CandleType.SPOT),
|
|
("XRP/BTC", "1m", CandleType.SPOT),
|
|
}
|
|
assert exchange_ws._klines_scheduled == {
|
|
("ETH/BTC", "1m", CandleType.SPOT),
|
|
("XRP/BTC", "1m", CandleType.SPOT),
|
|
}
|
|
|
|
# Wait for the expected number of watch calls
|
|
await wait_for_condition(lambda: ccxt_object.watch_ohlcv.call_count >= 6, timeout_=3.0)
|
|
assert ccxt_object.watch_ohlcv.call_count >= 6
|
|
ccxt_object.watch_ohlcv.reset_mock()
|
|
|
|
time_machine.shift(timedelta(minutes=5))
|
|
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
|
|
|
|
# Wait for log message
|
|
await wait_for_condition(
|
|
lambda: log_has_re("un_watch_ohlcv_for_symbols not supported: ", caplog), timeout_=2.0
|
|
)
|
|
assert log_has_re("un_watch_ohlcv_for_symbols not supported: ", caplog)
|
|
|
|
# XRP/BTC should be cleaned up.
|
|
assert exchange_ws._klines_watching == {
|
|
("ETH/BTC", "1m", CandleType.SPOT),
|
|
}
|
|
|
|
# Cleanup happened.
|
|
exchange_ws.schedule_ohlcv("ETH/BTC", "1m", CandleType.SPOT)
|
|
|
|
# Verify final state
|
|
assert exchange_ws._klines_watching == {
|
|
("ETH/BTC", "1m", CandleType.SPOT),
|
|
}
|
|
assert exchange_ws._klines_scheduled == {
|
|
("ETH/BTC", "1m", CandleType.SPOT),
|
|
}
|
|
|
|
# Triggers 2nd call to un_watch_ohlcv_for_symbols which raises ValueError
|
|
exchange_ws._klines_watching.discard(("ETH/BTC", "1m", CandleType.SPOT))
|
|
await wait_for_condition(
|
|
lambda: log_has_re("Exception in _unwatch_ohlcv", caplog), timeout_=2.0
|
|
)
|
|
assert log_has_re("Exception in _unwatch_ohlcv", caplog)
|
|
|
|
finally:
|
|
# Cleanup
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
async def test_exchangews_get_ohlcv(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
ccxt_object.ohlcvs = {
|
|
"ETH/USDT": {
|
|
"1m": [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
],
|
|
"5m": [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840300000, 105, 201, 301, 401, 501],
|
|
[1635840600000, 102, 202, 302, 402, 502],
|
|
],
|
|
}
|
|
}
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws._klines_last_refresh = {
|
|
("ETH/USDT", "1m", CandleType.SPOT): 1635840120000,
|
|
("ETH/USDT", "5m", CandleType.SPOT): 1635840600000,
|
|
}
|
|
|
|
# Matching last candle time - drop hint is true
|
|
resp = await exchange_ws.get_ohlcv("ETH/USDT", "1m", CandleType.SPOT, 1635840120000)
|
|
assert resp[0] == "ETH/USDT"
|
|
assert resp[1] == "1m"
|
|
assert resp[3] == [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
]
|
|
assert resp[4] is True
|
|
|
|
# expected time > last candle time - drop hint is false
|
|
resp = await exchange_ws.get_ohlcv("ETH/USDT", "1m", CandleType.SPOT, 1635840180000)
|
|
assert resp[0] == "ETH/USDT"
|
|
assert resp[1] == "1m"
|
|
assert resp[3] == [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
]
|
|
assert resp[4] is False
|
|
|
|
# Change "received" times to be before the candle starts.
|
|
# This should trigger the "time sync" warning.
|
|
exchange_ws._klines_last_refresh = {
|
|
("ETH/USDT", "1m", CandleType.SPOT): 1635840110000,
|
|
("ETH/USDT", "5m", CandleType.SPOT): 1635840600000,
|
|
}
|
|
msg = r".*Candle date > last refresh.*"
|
|
assert not log_has_re(msg, caplog)
|
|
resp = await exchange_ws.get_ohlcv("ETH/USDT", "1m", CandleType.SPOT, 1635840120000)
|
|
assert resp[0] == "ETH/USDT"
|
|
assert resp[1] == "1m"
|
|
assert resp[3] == [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
]
|
|
assert resp[4] is True
|
|
|
|
assert log_has_re(msg, caplog)
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
async def test_exchangews_get_ohlcv_missing_refresh_date(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
ccxt_object.ohlcvs = {
|
|
"ETH/USDT": {
|
|
"1m": [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
]
|
|
}
|
|
}
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws._klines_last_refresh = {}
|
|
|
|
# No refresh-date entry should not raise KeyError.
|
|
resp = await exchange_ws.get_ohlcv("ETH/USDT", "1m", CandleType.SPOT, 1635840120000)
|
|
assert resp[0] == "ETH/USDT"
|
|
assert resp[1] == "1m"
|
|
assert resp[4] is True
|
|
assert not log_has_re(r".*Candle date > last refresh.*", caplog)
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_ohlcvs_deepcopy_and_retry(mocker):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
ccxt_object.ohlcvs = {
|
|
"ETH/USDT": {
|
|
"1m": [[1, 2, 3, 4, 5, 6]],
|
|
}
|
|
}
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
|
|
call_count = {"count": 0}
|
|
|
|
def deepcopy_side_effect(value):
|
|
call_count["count"] += 1
|
|
if call_count["count"] < 3:
|
|
raise RuntimeError("copy failed")
|
|
return [candle.copy() for candle in value]
|
|
|
|
mocker.patch("freqtrade.exchange.exchange_ws.deepcopy", deepcopy_side_effect)
|
|
|
|
result = exchange_ws.ohlcvs("ETH/USDT", "1m")
|
|
|
|
assert call_count["count"] == 3
|
|
assert result == [[1, 2, 3, 4, 5, 6]]
|
|
assert result is not ccxt_object.ohlcvs["ETH/USDT"]["1m"]
|
|
|
|
# Fail all the time
|
|
mocker.patch("freqtrade.exchange.exchange_ws.deepcopy", side_effect=RuntimeError("copy failed"))
|
|
with pytest.raises(TemporaryError, match=r"Error deepcopying: copy failed"):
|
|
exchange_ws.ohlcvs("ETH/USDT", "1m")
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_get_ohlcv_with_refresh(mocker):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
ohlcvs_mock = mocker.patch.object(
|
|
exchange_ws, "ohlcvs", return_value=[[10, 11, 12, 13, 14, 15]]
|
|
)
|
|
|
|
paircomb = ("ETH/USDT", "1m", CandleType.SPOT)
|
|
exchange_ws._klines_last_refresh[paircomb] = 123456789
|
|
|
|
candles, refresh = exchange_ws.get_ohlcv_with_refresh("ETH/USDT", "1m", CandleType.SPOT)
|
|
|
|
ohlcvs_mock.assert_called_once_with("ETH/USDT", "1m")
|
|
assert candles == [[10, 11, 12, 13, 14, 15]]
|
|
assert refresh == 123456789
|
|
|
|
candles, refresh = exchange_ws.get_ohlcv_with_refresh("ETH/USDT", "5m", CandleType.SPOT)
|
|
assert candles == [[10, 11, 12, 13, 14, 15]]
|
|
assert refresh == 0
|
|
|
|
exchange_ws.cleanup()
|
|
|
|
|
|
def test_exchangews_continuous_stopped_task_exception(mocker, caplog):
|
|
config = MagicMock()
|
|
ccxt_object = MagicMock()
|
|
ccxt_object.ohlcvs = {
|
|
"ETH/USDT": {
|
|
"1m": [
|
|
[1635840000000, 100, 200, 300, 400, 500],
|
|
[1635840060000, 101, 201, 301, 401, 501],
|
|
[1635840120000, 102, 202, 302, 402, 502],
|
|
]
|
|
}
|
|
}
|
|
mocker.patch("freqtrade.exchange.exchange_ws.ExchangeWS._start_forever", MagicMock())
|
|
|
|
exchange_ws = ExchangeWS(config, ccxt_object)
|
|
exchange_ws._loop = MagicMock()
|
|
exchange_ws._loop.is_closed.return_value = False
|
|
|
|
paircomb = ("ETH/USDT", "1m", CandleType.SPOT)
|
|
exchange_ws._klines_scheduled.add(paircomb)
|
|
exchange_ws._klines_last_refresh[paircomb] = 1
|
|
|
|
task = MagicMock()
|
|
task.cancelled.return_value = False
|
|
task.result.side_effect = RuntimeError("unexpected")
|
|
exchange_ws._background_tasks.add(task)
|
|
|
|
completed_future = MagicMock()
|
|
completed_future.result.return_value = None
|
|
|
|
def side_effect(coro, loop):
|
|
coro.close()
|
|
return completed_future
|
|
|
|
run_threadsafe = mocker.patch(
|
|
"freqtrade.exchange.exchange_ws.asyncio.run_coroutine_threadsafe",
|
|
side_effect=side_effect,
|
|
)
|
|
|
|
exchange_ws._continuous_stopped(task, "ETH/USDT", "1m", CandleType.SPOT)
|
|
|
|
assert task not in exchange_ws._background_tasks
|
|
assert paircomb not in exchange_ws._klines_scheduled
|
|
assert paircomb not in exchange_ws._klines_last_refresh
|
|
assert ccxt_object.ohlcvs["ETH/USDT"].get("1m") is None
|
|
assert run_threadsafe.call_count == 1
|
|
assert log_has_re("Unhandled exception in watch task callback for ETH/USDT, 1m", caplog)
|
|
|
|
exchange_ws.cleanup()
|