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=r"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: {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_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"])
@pytest.mark.parametrize(
"number_assets, result",
[
(1, ["ETH/USDT"]),
(2, ["ETH/USDT", "XRP/USDT"]),
(500, ["ETH/USDT", "XRP/USDT"]),
(None, ["ETH/USDT", "XRP/USDT"]),
],
)
def test_remote_pairlist_whitelist_number_assets(
mocker, rpl_config, number_assets, result, markets, tickers
):
mock_response = MagicMock()
mock_response.json.return_value = {
"pairs": ["ETH/USDT", "XRP/USDT", "TKN/USDT"],
"refresh_period": 60,
}
mock_response.headers = {"content-type": "application/json"}
rpl_config["pairlists"] = [
{
"method": "RemotePairList",
"mode": "whitelist",
"pairlist_url": "http://example.com/pairlist",
"number_assets": number_assets,
},
]
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"][0], 0
)
pairs, _ = remote_pairlist.fetch_pairlist()
assert pairs == ["ETH/USDT", "XRP/USDT", "TKN/USDT"]
whitelist = remote_pairlist.filter_pairlist(rpl_config["exchange"]["pair_whitelist"], {})
assert whitelist == result