freqtrade_origin/tests/plugins/test_remotepairlist.py
2024-05-13 07:10:24 +02:00

344 lines
11 KiB
Python

import json
from unittest.mock import MagicMock, PropertyMock
import pytest
import requests
from freqtrade.exceptions import OperationalException
from freqtrade.plugins.pairlist.RemotePairList import RemotePairList
from freqtrade.plugins.pairlistmanager import PairListManager
from tests.conftest import EXMS, get_patched_exchange, get_patched_freqtradebot, log_has
@pytest.fixture(scope="function")
def rpl_config(default_conf):
default_conf["stake_currency"] = "USDT"
default_conf["exchange"]["pair_whitelist"] = [
"ETH/USDT",
"XRP/USDT",
]
default_conf["exchange"]["pair_blacklist"] = ["BLK/USDT"]
return default_conf
def test_gen_pairlist_with_local_file(mocker, rpl_config):
mock_file = MagicMock()
mock_file.read.return_value = '{"pairs": ["TKN/USDT","ETH/USDT","NANO/USDT"]}'
mocker.patch("freqtrade.plugins.pairlist.RemotePairList.open", return_value=mock_file)
mock_file_path = mocker.patch("freqtrade.plugins.pairlist.RemotePairList.Path")
mock_file_path.exists.return_value = True
jsonparse = json.loads(mock_file.read.return_value)
mocker.patch("freqtrade.plugins.pairlist.RemotePairList.rapidjson.load", return_value=jsonparse)
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"number_assets": 2,
"refresh_period": 1800,
"keep_pairlist_on_failure": True,
"pairlist_url": "file:///pairlist.json",
"bearer_token": "",
"read_timeout": 60,
}
]
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][0], 0
)
result = remote_pairlist.gen_pairlist([])
assert result == ["TKN/USDT", "ETH/USDT"]
def test_fetch_pairlist_mock_response_html(mocker, rpl_config):
mock_response = MagicMock()
mock_response.headers = {"content-type": "text/html"}
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"pairlist_url": "http://example.com/pairlist",
"number_assets": 10,
"read_timeout": 10,
"keep_pairlist_on_failure": True,
}
]
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
mocker.patch(
"freqtrade.plugins.pairlist.RemotePairList.requests.get", return_value=mock_response
)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][0], 0
)
with pytest.raises(OperationalException, match="RemotePairList is not of type JSON."):
remote_pairlist.fetch_pairlist()
def test_fetch_pairlist_timeout_keep_last_pairlist(mocker, rpl_config, caplog):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"pairlist_url": "http://example.com/pairlist",
"number_assets": 10,
"read_timeout": 10,
"keep_pairlist_on_failure": True,
}
]
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
mocker.patch(
"freqtrade.plugins.pairlist.RemotePairList.requests.get",
side_effect=requests.exceptions.RequestException,
)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][0], 0
)
remote_pairlist._last_pairlist = ["BTC/USDT", "ETH/USDT", "LTC/USDT"]
remote_pairlist._init_done = True
pairlist_url = rpl_config["pairlists"][0]["pairlist_url"]
pairs, _time_elapsed = remote_pairlist.fetch_pairlist()
assert log_has(f"Error: Was not able to fetch pairlist from: " f"{pairlist_url}", caplog)
assert log_has("Keeping last fetched pairlist", caplog)
assert pairs == ["BTC/USDT", "ETH/USDT", "LTC/USDT"]
def test_remote_pairlist_init_no_pairlist_url(mocker, rpl_config):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"number_assets": 10,
"keep_pairlist_on_failure": True,
}
]
get_patched_exchange(mocker, rpl_config)
with pytest.raises(
OperationalException,
match=r"`pairlist_url` not specified."
r' Please check your configuration for "pairlist.config.pairlist_url"',
):
get_patched_freqtradebot(mocker, rpl_config)
def test_remote_pairlist_init_no_number_assets(mocker, rpl_config):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"pairlist_url": "http://example.com/pairlist",
"keep_pairlist_on_failure": True,
}
]
get_patched_exchange(mocker, rpl_config)
with pytest.raises(
OperationalException,
match=r"`number_assets` not specified. "
'Please check your configuration for "pairlist.config.number_assets"',
):
get_patched_freqtradebot(mocker, rpl_config)
def test_fetch_pairlist_mock_response_valid(mocker, rpl_config):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"pairlist_url": "http://example.com/pairlist",
"number_assets": 10,
"refresh_period": 10,
"read_timeout": 10,
"keep_pairlist_on_failure": True,
}
]
mock_response = MagicMock()
mock_response.json.return_value = {
"pairs": ["ETH/USDT", "XRP/USDT", "LTC/USDT", "EOS/USDT"],
"refresh_period": 60,
}
mock_response.headers = {"content-type": "application/json"}
mock_response.elapsed.total_seconds.return_value = 0.4
mocker.patch(
"freqtrade.plugins.pairlist.RemotePairList.requests.get", return_value=mock_response
)
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][0], 0
)
pairs, time_elapsed = remote_pairlist.fetch_pairlist()
assert pairs == ["ETH/USDT", "XRP/USDT", "LTC/USDT", "EOS/USDT"]
assert time_elapsed == 0.4
assert remote_pairlist._refresh_period == 60
def test_remote_pairlist_init_wrong_mode(mocker, rpl_config):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"mode": "blacklis",
"number_assets": 20,
"pairlist_url": "http://example.com/pairlist",
"keep_pairlist_on_failure": True,
}
]
with pytest.raises(
OperationalException,
match=r'`mode` not configured correctly. Supported Modes are "whitelist","blacklist"',
):
get_patched_freqtradebot(mocker, rpl_config)
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"mode": "blacklist",
"number_assets": 20,
"pairlist_url": "http://example.com/pairlist",
"keep_pairlist_on_failure": True,
}
]
with pytest.raises(
OperationalException, match=r"A `blacklist` mode RemotePairList can not be.*first.*"
):
get_patched_freqtradebot(mocker, rpl_config)
def test_remote_pairlist_init_wrong_proc_mode(mocker, rpl_config):
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"processing_mode": "filler",
"mode": "whitelist",
"number_assets": 20,
"pairlist_url": "http://example.com/pairlist",
"keep_pairlist_on_failure": True,
}
]
get_patched_exchange(mocker, rpl_config)
with pytest.raises(
OperationalException,
match=r'`processing_mode` not configured correctly. Supported Modes are "filter","append"',
):
get_patched_freqtradebot(mocker, rpl_config)
def test_remote_pairlist_blacklist(mocker, rpl_config, caplog, markets, tickers):
mock_response = MagicMock()
mock_response.json.return_value = {"pairs": ["XRP/USDT"], "refresh_period": 60}
mock_response.headers = {"content-type": "application/json"}
rpl_config["pairlists"] = [
{
"method": "StaticPairList",
},
{
"method": "RemotePairList",
"mode": "blacklist",
"pairlist_url": "http://example.com/pairlist",
"number_assets": 3,
},
]
mocker.patch.multiple(
EXMS,
markets=PropertyMock(return_value=markets),
exchange_has=MagicMock(return_value=True),
get_tickers=tickers,
)
mocker.patch(
"freqtrade.plugins.pairlist.RemotePairList.requests.get", return_value=mock_response
)
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][1], 1
)
pairs, _time_elapsed = remote_pairlist.fetch_pairlist()
assert pairs == ["XRP/USDT"]
whitelist = remote_pairlist.filter_pairlist(rpl_config["exchange"]["pair_whitelist"], {})
assert whitelist == ["ETH/USDT"]
assert log_has(f"Blacklist - Filtered out pairs: {pairs}", caplog)
@pytest.mark.parametrize("processing_mode", ["filter", "append"])
def test_remote_pairlist_whitelist(mocker, rpl_config, processing_mode, markets, tickers):
mock_response = MagicMock()
mock_response.json.return_value = {"pairs": ["XRP/USDT"], "refresh_period": 60}
mock_response.headers = {"content-type": "application/json"}
rpl_config["pairlists"] = [
{
"method": "StaticPairList",
},
{
"method": "RemotePairList",
"mode": "whitelist",
"processing_mode": processing_mode,
"pairlist_url": "http://example.com/pairlist",
"number_assets": 3,
},
]
mocker.patch.multiple(
EXMS,
markets=PropertyMock(return_value=markets),
exchange_has=MagicMock(return_value=True),
get_tickers=tickers,
)
mocker.patch(
"freqtrade.plugins.pairlist.RemotePairList.requests.get", return_value=mock_response
)
exchange = get_patched_exchange(mocker, rpl_config)
pairlistmanager = PairListManager(exchange, rpl_config)
remote_pairlist = RemotePairList(
exchange, pairlistmanager, rpl_config, rpl_config["pairlists"][1], 1
)
pairs, _time_elapsed = remote_pairlist.fetch_pairlist()
assert pairs == ["XRP/USDT"]
whitelist = remote_pairlist.filter_pairlist(rpl_config["exchange"]["pair_whitelist"], {})
assert whitelist == (["XRP/USDT"] if processing_mode == "filter" else ["ETH/USDT", "XRP/USDT"])