mohammadian7/freqtrade
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: simplified test setup for compat tests

Browse Source
This commit is contained in:
Matthias
2026-02-05 20:10:52 +01:00
parent e5bd33061b
commit 315a1baf2c
2 changed files with 98 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/exchange_online/conftest.py
+18 -16
View File
@@ -1,5 +1,6 @@
from copy import deepcopy from copy import deepcopy
from pathlib import Path from pathlib import Path
from typing import Any
import pytest import pytest
@@ -9,7 +10,7 @@ from freqtrade.resolvers.exchange_resolver import ExchangeResolver
from tests.conftest import EXMS, get_default_conf_usdt from tests.conftest import EXMS, get_default_conf_usdt
EXCHANGE_FIXTURE_TYPE = tuple[Exchange, str] EXCHANGE_FIXTURE_TYPE = tuple[Exchange, str, dict[str, Any]]
EXCHANGE_WS_FIXTURE_TYPE = tuple[Exchange, str, str] EXCHANGE_WS_FIXTURE_TYPE = tuple[Exchange, str, str]
@@ -565,26 +566,25 @@ def set_test_proxy(config: Config, use_proxy: bool) -> Config:
def get_exchange(exchange_name, exchange_conf): def get_exchange(exchange_name, exchange_conf):
exchange_conf = set_test_proxy( exchange_params = EXCHANGES[exchange_name]
exchange_conf, EXCHANGES[exchange_name].get("use_ci_proxy", False) exchange_conf = set_test_proxy(exchange_conf, exchange_params.get("use_ci_proxy", False))
)
exchange_conf["exchange"]["name"] = exchange_name exchange_conf["exchange"]["name"] = exchange_name
exchange_conf["stake_currency"] = EXCHANGES[exchange_name]["stake_currency"] exchange_conf["stake_currency"] = exchange_params["stake_currency"]
exchange = ExchangeResolver.load_exchange( exchange = ExchangeResolver.load_exchange(
exchange_conf, validate=True, load_leverage_tiers=True exchange_conf, validate=True, load_leverage_tiers=True
) )
return exchange, exchange_name return exchange, exchange_name, exchange_params
def get_futures_exchange(exchange_name, exchange_conf, class_mocker): def get_futures_exchange(exchange_name, exchange_conf, class_mocker):
if EXCHANGES[exchange_name].get("futures") is not True: exchange_params = EXCHANGES[exchange_name]
if exchange_params.get("futures") is not True:
pytest.skip(f"Exchange {exchange_name} does not support futures.") pytest.skip(f"Exchange {exchange_name} does not support futures.")
else: else:
exchange_conf = deepcopy(exchange_conf) exchange_conf = deepcopy(exchange_conf)
exchange_conf = set_test_proxy( exchange_conf = set_test_proxy(exchange_conf, exchange_params.get("use_ci_proxy", False))
exchange_conf, EXCHANGES[exchange_name].get("use_ci_proxy", False)
)
exchange_conf["trading_mode"] = "futures" exchange_conf["trading_mode"] = "futures"
exchange_conf["margin_mode"] = "isolated" exchange_conf["margin_mode"] = "isolated"
@@ -600,15 +600,17 @@ def get_futures_exchange(exchange_name, exchange_conf, class_mocker):
@pytest.fixture(params=EXCHANGES, scope="class") @pytest.fixture(params=EXCHANGES, scope="class")
def exchange(request, exchange_conf, class_mocker): def exchange(request, exchange_conf, class_mocker):
class_mocker.patch(f"{EXMS}.ft_additional_exchange_init") class_mocker.patch(f"{EXMS}.ft_additional_exchange_init")
exchange, name = get_exchange(request.param, exchange_conf) exchange, name, exchange_params = get_exchange(request.param, exchange_conf)
yield exchange, name yield exchange, name, exchange_params
exchange.close() exchange.close()
@pytest.fixture(params=EXCHANGES_FUTURES, scope="class") @pytest.fixture(params=EXCHANGES_FUTURES, scope="class")
def exchange_futures(request, exchange_conf, class_mocker): def exchange_futures(request, exchange_conf, class_mocker):
exchange, name = get_futures_exchange(request.param, exchange_conf, class_mocker) exchange, name, exchange_params = get_futures_exchange(
yield exchange, name request.param, exchange_conf, class_mocker
)
yield exchange, name, exchange_params
exchange.close() exchange.close()
@@ -625,10 +627,10 @@ def exchange_ws(request, exchange_conf, exchange_mode, class_mocker):
if exchange_param.get("skip_ws_tests"): if exchange_param.get("skip_ws_tests"):
pytest.skip(f"{request.param} does not support websocket tests.") pytest.skip(f"{request.param} does not support websocket tests.")
if exchange_mode == "spot": if exchange_mode == "spot":
exchange, name = get_exchange(request.param, exchange_conf) exchange, name, _ = get_exchange(request.param, exchange_conf)
pair = exchange_param["pair"] pair = exchange_param["pair"]
elif exchange_param.get("futures"): elif exchange_param.get("futures"):
exchange, name = get_futures_exchange( exchange, name, _ = get_futures_exchange(
request.param, exchange_conf, class_mocker=class_mocker request.param, exchange_conf, class_mocker=class_mocker
) )
pair = exchange_param["futures_pair"] pair = exchange_param["futures_pair"]
tests/exchange_online/test_ccxt_compat.py
+80 -86
View File
@@ -13,21 +13,21 @@ from freqtrade.enums import CandleType
from freqtrade.exchange import timeframe_to_minutes, timeframe_to_prev_date from freqtrade.exchange import timeframe_to_minutes, timeframe_to_prev_date
from freqtrade.exchange.exchange import Exchange, timeframe_to_msecs from freqtrade.exchange.exchange import Exchange, timeframe_to_msecs
from freqtrade.util import dt_floor_day, dt_now, dt_ts from freqtrade.util import dt_floor_day, dt_now, dt_ts
from tests.exchange_online.conftest import EXCHANGE_FIXTURE_TYPE, EXCHANGES from tests.exchange_online.conftest import EXCHANGE_FIXTURE_TYPE
@pytest.mark.longrun @pytest.mark.longrun
class TestCCXTExchange: class TestCCXTExchange:
def test_load_markets(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_load_markets(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
markets = exch.markets markets = exch.markets
assert pair in markets assert pair in markets
assert isinstance(markets[pair], dict) assert isinstance(markets[pair], dict)
assert exch.market_is_spot(markets[pair]) assert exch.market_is_spot(markets[pair])
def test_has_validations(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_has_validations(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, exchangename, _ = exchange
exch.validate_ordertypes( exch.validate_ordertypes(
{ {
@@ -49,17 +49,17 @@ class TestCCXTExchange:
) )
def test_ohlcv_limit(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ohlcv_limit(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
expected_count = EXCHANGES[exchangename].get("candle_count") expected_count = exchange_params.get("candle_count")
if not expected_count: if not expected_count:
pytest.skip("No expected candle count for exchange") pytest.skip("No expected candle count for exchange")
assert exch.ohlcv_candle_limit("1m", CandleType.SPOT) == expected_count assert exch.ohlcv_candle_limit("1m", CandleType.SPOT) == expected_count
def test_ohlcv_limit_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ohlcv_limit_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange_futures exch, _, exchange_params = exchange_futures
expected_count = EXCHANGES[exchangename].get( expected_count = exchange_params.get(
"candle_count_futures", EXCHANGES[exchangename].get("candle_count") "candle_count_futures", exchange_params.get("candle_count")
) )
if not expected_count: if not expected_count:
pytest.skip("No expected candle count for exchange") pytest.skip("No expected candle count for exchange")
@@ -67,10 +67,10 @@ class TestCCXTExchange:
assert exch.ohlcv_candle_limit("1m", CandleType.FUTURES) == expected_count assert exch.ohlcv_candle_limit("1m", CandleType.FUTURES) == expected_count
def test_load_markets_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_load_markets_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exchange, exchangename = exchange_futures exchange, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
pair1 = EXCHANGES[exchangename].get("futures_pair", pair) pair1 = exchange_params.get("futures_pair", pair)
alternative_pairs = EXCHANGES[exchangename].get("futures_alt_pairs", []) alternative_pairs = exchange_params.get("futures_alt_pairs", [])
markets = exchange.markets markets = exchange.markets
for pair in [pair1] + alternative_pairs: for pair in [pair1] + alternative_pairs:
assert pair in markets, f"Futures pair {pair} not found in markets" assert pair in markets, f"Futures pair {pair} not found in markets"
@@ -79,8 +79,8 @@ class TestCCXTExchange:
assert exchange.market_is_future(markets[pair]) assert exchange.market_is_future(markets[pair])
def test_ccxt_order_parse(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_order_parse(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchange_name = exchange exch, exchangename, exchange_params = exchange
if orders := EXCHANGES[exchange_name].get("sample_order"): if orders := exchange_params.get("sample_order"):
for order in orders: for order in orders:
pair = order["pair"] pair = order["pair"]
exchange_response: dict = order["exchange_response"] exchange_response: dict = order["exchange_response"]
@@ -108,11 +108,11 @@ class TestCCXTExchange:
assert po["amount"] == expected["amount"] assert po["amount"] == expected["amount"]
assert isinstance(po["status"], str) assert isinstance(po["status"], str)
else: else:
pytest.skip(f"No sample order available for exchange {exchange_name}") pytest.skip(f"No sample order available for exchange {exchangename}")
def test_ccxt_my_trades_parse(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_my_trades_parse(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchange_name = exchange exch, exchangename, exchange_params = exchange
if trades := EXCHANGES[exchange_name].get("sample_my_trades"): if trades := exchange_params.get("sample_my_trades"):
pair = "SOL/USDT" pair = "SOL/USDT"
for trade in trades: for trade in trades:
po = exch._api.parse_trade(trade) po = exch._api.parse_trade(trade)
@@ -132,11 +132,11 @@ class TestCCXTExchange:
assert isinstance(fee["currency"], str) assert isinstance(fee["currency"], str)
else: else:
pytest.skip(f"No sample Trades available for exchange {exchange_name}") pytest.skip(f"No sample Trades available for exchange {exchangename}")
def test_ccxt_balances_parse(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_balances_parse(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchange_name = exchange exch, exchangename, exchange_params = exchange
if balance_response := EXCHANGES[exchange_name].get("sample_balances"): if balance_response := exchange_params.get("sample_balances"):
balances = exch._api.parse_balance(balance_response["exchange_response"]) balances = exch._api.parse_balance(balance_response["exchange_response"])
expected = balance_response["expected"] expected = balance_response["expected"]
for currency, balance in expected.items(): for currency, balance in expected.items():
@@ -145,31 +145,31 @@ class TestCCXTExchange:
assert balance == balances[currency] assert balance == balances[currency]
pass pass
else: else:
pytest.skip(f"No sample Balances available for exchange {exchange_name}") pytest.skip(f"No sample Balances available for exchange {exchangename}")
def test_ccxt_fetch_tickers(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_tickers(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
tickers = exch.get_tickers() tickers = exch.get_tickers()
assert pair in tickers assert pair in tickers
assert "ask" in tickers[pair] assert "ask" in tickers[pair]
assert "bid" in tickers[pair] assert "bid" in tickers[pair]
if EXCHANGES[exchangename].get("tickers_have_bid_ask"): if exchange_params.get("tickers_have_bid_ask"):
assert tickers[pair]["bid"] is not None assert tickers[pair]["bid"] is not None
assert tickers[pair]["ask"] is not None assert tickers[pair]["ask"] is not None
assert "quoteVolume" in tickers[pair] assert "quoteVolume" in tickers[pair]
if EXCHANGES[exchangename].get("hasQuoteVolume"): if exchange_params.get("hasQuoteVolume"):
assert tickers[pair]["quoteVolume"] is not None assert tickers[pair]["quoteVolume"] is not None
def test_ccxt_fetch_tickers_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_tickers_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange_futures exch, exchangename, exchange_params = exchange_futures
if not exch or exchangename in ("gate"): if not exch or exchangename in ("gate"):
# exchange_futures only returns values for supported exchanges # exchange_futures only returns values for supported exchanges
return return
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
pair = EXCHANGES[exchangename].get("futures_pair", pair) pair = exchange_params.get("futures_pair", pair)
tickers = exch.get_tickers() tickers = exch.get_tickers()
assert pair in tickers assert pair in tickers
@@ -178,28 +178,28 @@ class TestCCXTExchange:
assert "bid" in tickers[pair] assert "bid" in tickers[pair]
assert tickers[pair]["bid"] is not None assert tickers[pair]["bid"] is not None
assert "quoteVolume" in tickers[pair] assert "quoteVolume" in tickers[pair]
if EXCHANGES[exchangename].get("hasQuoteVolumeFutures"): if exchange_params.get("hasQuoteVolumeFutures"):
assert tickers[pair]["quoteVolume"] is not None assert tickers[pair]["quoteVolume"] is not None
def test_ccxt_fetch_ticker(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_ticker(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
ticker = exch.fetch_ticker(pair) ticker = exch.fetch_ticker(pair)
assert "ask" in ticker assert "ask" in ticker
assert "bid" in ticker assert "bid" in ticker
if EXCHANGES[exchangename].get("tickers_have_bid_ask"): if exchange_params.get("tickers_have_bid_ask"):
assert ticker["ask"] is not None assert ticker["ask"] is not None
assert ticker["bid"] is not None assert ticker["bid"] is not None
assert "quoteVolume" in ticker assert "quoteVolume" in ticker
if EXCHANGES[exchangename].get("hasQuoteVolume"): if exchange_params.get("hasQuoteVolume"):
assert ticker["quoteVolume"] is not None assert ticker["quoteVolume"] is not None
def test_ccxt_fetch_l2_orderbook(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_l2_orderbook(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, exchangename, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
l2 = exch.fetch_l2_order_book(pair) l2 = exch.fetch_l2_order_book(pair)
orderbook_max_entries = EXCHANGES[exchangename].get("orderbook_max_entries") orderbook_max_entries = exchange_params.get("orderbook_max_entries")
assert "asks" in l2 assert "asks" in l2
assert "bids" in l2 assert "bids" in l2
assert len(l2["asks"]) >= 1 assert len(l2["asks"]) >= 1
@@ -237,9 +237,9 @@ class TestCCXTExchange:
assert len(l2["asks"]) == next_limit assert len(l2["asks"]) == next_limit
def test_ccxt_fetch_ohlcv(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_ohlcv(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
timeframe = EXCHANGES[exchangename]["timeframe"] timeframe = exchange_params["timeframe"]
pair_tf = (pair, timeframe, CandleType.SPOT) pair_tf = (pair, timeframe, CandleType.SPOT)
@@ -259,8 +259,8 @@ class TestCCXTExchange:
""" """
Test that pair data starts at the provided startdate Test that pair data starts at the provided startdate
""" """
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
timeframe = "1d" timeframe = "1d"
pair_tf = (pair, timeframe, CandleType.SPOT) pair_tf = (pair, timeframe, CandleType.SPOT)
@@ -310,12 +310,12 @@ class TestCCXTExchange:
assert candles[0][0] == since_ms or (since_ms + timeframe_ms) assert candles[0][0] == since_ms or (since_ms + timeframe_ms)
def test_ccxt__async_get_candle_history(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt__async_get_candle_history(self, exchange: EXCHANGE_FIXTURE_TYPE):
exc, exchangename = exchange exc, _, exchange_params = exchange
if not exc._ft_has["ohlcv_has_history"]: if not exc._ft_has["ohlcv_has_history"]:
pytest.skip("Exchange does not support candle history") pytest.skip("Exchange does not support candle history")
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
timeframe = EXCHANGES[exchangename]["timeframe"] timeframe = exchange_params["timeframe"]
self._ccxt__async_get_candle_history(exc, pair, timeframe, CandleType.SPOT) self._ccxt__async_get_candle_history(exc, pair, timeframe, CandleType.SPOT)
@pytest.mark.parametrize( @pytest.mark.parametrize(
@@ -331,9 +331,9 @@ class TestCCXTExchange:
def test_ccxt__async_get_candle_history_futures( def test_ccxt__async_get_candle_history_futures(
self, exchange_futures: EXCHANGE_FIXTURE_TYPE, candle_type: CandleType self, exchange_futures: EXCHANGE_FIXTURE_TYPE, candle_type: CandleType
): ):
exchange, exchangename = exchange_futures exchange, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename].get("futures_pair", EXCHANGES[exchangename]["pair"]) pair = exchange_params.get("futures_pair", exchange_params["pair"])
timeframe = EXCHANGES[exchangename]["timeframe"] timeframe = exchange_params["timeframe"]
if candle_type == CandleType.FUNDING_RATE: if candle_type == CandleType.FUNDING_RATE:
timeframe = exchange._ft_has.get( timeframe = exchange._ft_has.get(
"funding_fee_timeframe", exchange._ft_has["mark_ohlcv_timeframe"] "funding_fee_timeframe", exchange._ft_has["mark_ohlcv_timeframe"]
@@ -350,9 +350,9 @@ class TestCCXTExchange:
) )
def test_ccxt_fetch_funding_rate_history(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_funding_rate_history(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exchange, exchangename = exchange_futures exchange, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename].get("futures_pair", EXCHANGES[exchangename]["pair"]) pair = exchange_params.get("futures_pair", exchange_params["pair"])
since = int((datetime.now(UTC) - timedelta(days=5)).timestamp() * 1000) since = int((datetime.now(UTC) - timedelta(days=5)).timestamp() * 1000)
timeframe_ff = exchange._ft_has.get( timeframe_ff = exchange._ft_has.get(
"funding_fee_timeframe", exchange._ft_has["mark_ohlcv_timeframe"] "funding_fee_timeframe", exchange._ft_has["mark_ohlcv_timeframe"]
@@ -399,8 +399,8 @@ class TestCCXTExchange:
) )
def test_ccxt_fetch_mark_price_history(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_fetch_mark_price_history(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exchange, exchangename = exchange_futures exchange, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename].get("futures_pair", EXCHANGES[exchangename]["pair"]) pair = exchange_params.get("futures_pair", exchange_params["pair"])
since = int((datetime.now(UTC) - timedelta(days=5)).timestamp() * 1000) since = int((datetime.now(UTC) - timedelta(days=5)).timestamp() * 1000)
candle_type = CandleType.from_string( candle_type = CandleType.from_string(
exchange.get_option("mark_ohlcv_price", default=CandleType.MARK) exchange.get_option("mark_ohlcv_price", default=CandleType.MARK)
@@ -424,8 +424,8 @@ class TestCCXTExchange:
assert mark_candles[mark_candles["date"] == this_hour].iloc[0]["open"] != 0.0 assert mark_candles[mark_candles["date"] == this_hour].iloc[0]["open"] != 0.0
def test_ccxt__calculate_funding_fees(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt__calculate_funding_fees(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exchange, exchangename = exchange_futures exchange, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename].get("futures_pair", EXCHANGES[exchangename]["pair"]) pair = exchange_params.get("futures_pair", exchange_params["pair"])
since = datetime.now(UTC) - timedelta(days=5) since = datetime.now(UTC) - timedelta(days=5)
funding_fee = exchange._fetch_and_calculate_funding_fees( funding_fee = exchange._fetch_and_calculate_funding_fees(
@@ -436,10 +436,10 @@ class TestCCXTExchange:
assert funding_fee != 0 assert funding_fee != 0
def test_ccxt__async_get_trade_history(self, exchange: EXCHANGE_FIXTURE_TYPE, mocker): def test_ccxt__async_get_trade_history(self, exchange: EXCHANGE_FIXTURE_TYPE, mocker):
exch, exchangename = exchange exch, exchangename, exchange_params = exchange
if not (lookback := EXCHANGES[exchangename].get("trades_lookback_hours")): if not (lookback := exchange_params.get("trades_lookback_hours")):
pytest.skip("test_fetch_trades not enabled for this exchange") pytest.skip("test_fetch_trades not enabled for this exchange")
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
since = int((datetime.now(UTC) - timedelta(hours=lookback)).timestamp() * 1000) since = int((datetime.now(UTC) - timedelta(hours=lookback)).timestamp() * 1000)
nvspy = mocker.spy(exch, "_get_trade_pagination_next_value") nvspy = mocker.spy(exch, "_get_trade_pagination_next_value")
res = exch.loop.run_until_complete(exch._async_get_trade_history(pair, since, None, None)) res = exch.loop.run_until_complete(exch._async_get_trade_history(pair, since, None, None))
@@ -466,50 +466,46 @@ class TestCCXTExchange:
assert 0 < exch.get_fee(pair, "market", "sell") < threshold assert 0 < exch.get_fee(pair, "market", "sell") < threshold
def test_ccxt_get_fee_spot(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_fee_spot(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
pair = EXCHANGES[exchangename]["pair"] pair = exchange_params["pair"]
self._ccxt_get_fee(exch, pair) self._ccxt_get_fee(exch, pair)
def test_ccxt_get_fee_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_fee_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange_futures exch, _, exchange_params = exchange_futures
pair = EXCHANGES[exchangename].get("futures_pair", EXCHANGES[exchangename]["pair"]) pair = exchange_params.get("futures_pair", exchange_params["pair"])
self._ccxt_get_fee(exch, pair) self._ccxt_get_fee(exch, pair)
def test_ccxt_get_max_leverage_spot(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_max_leverage_spot(self, exchange: EXCHANGE_FIXTURE_TYPE):
spot, spot_name = exchange spot, _, exchange_params = exchange
if spot: if spot:
leverage_in_market_spot = EXCHANGES[spot_name].get("leverage_in_spot_market") leverage_in_market_spot = exchange_params.get("leverage_in_spot_market")
if leverage_in_market_spot: if leverage_in_market_spot:
spot_pair = EXCHANGES[spot_name].get("pair", EXCHANGES[spot_name]["pair"]) spot_pair = exchange_params.get("pair", exchange_params["pair"])
spot_leverage = spot.get_max_leverage(spot_pair, 20) spot_leverage = spot.get_max_leverage(spot_pair, 20)
assert isinstance(spot_leverage, float) or isinstance(spot_leverage, int) assert isinstance(spot_leverage, float) or isinstance(spot_leverage, int)
assert spot_leverage >= 1.0 assert spot_leverage >= 1.0
def test_ccxt_get_max_leverage_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_max_leverage_futures(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
futures, futures_name = exchange_futures futures, _, exchange_params = exchange_futures
leverage_tiers_public = EXCHANGES[futures_name].get("leverage_tiers_public") leverage_tiers_public = exchange_params.get("leverage_tiers_public")
if leverage_tiers_public: if leverage_tiers_public:
futures_pair = EXCHANGES[futures_name].get( futures_pair = exchange_params.get("futures_pair", exchange_params["pair"])
"futures_pair", EXCHANGES[futures_name]["pair"]
)
futures_leverage = futures.get_max_leverage(futures_pair, 20) futures_leverage = futures.get_max_leverage(futures_pair, 20)
assert isinstance(futures_leverage, float) or isinstance(futures_leverage, int) assert isinstance(futures_leverage, float) or isinstance(futures_leverage, int)
assert futures_leverage >= 1.0 assert futures_leverage >= 1.0
def test_ccxt_get_contract_size(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_contract_size(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
futures, futures_name = exchange_futures futures, _, exchange_params = exchange_futures
futures_pair = EXCHANGES[futures_name].get("futures_pair", EXCHANGES[futures_name]["pair"]) futures_pair = exchange_params.get("futures_pair", exchange_params["pair"])
contract_size = futures.get_contract_size(futures_pair) contract_size = futures.get_contract_size(futures_pair)
assert isinstance(contract_size, float) or isinstance(contract_size, int) assert isinstance(contract_size, float) or isinstance(contract_size, int)
assert contract_size >= 0.0 assert contract_size >= 0.0
def test_ccxt_load_leverage_tiers(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_load_leverage_tiers(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
futures, futures_name = exchange_futures futures, _, exchange_params = exchange_futures
if EXCHANGES[futures_name].get("leverage_tiers_public"): if exchange_params.get("leverage_tiers_public"):
leverage_tiers = futures.load_leverage_tiers() leverage_tiers = futures.load_leverage_tiers()
futures_pair = EXCHANGES[futures_name].get( futures_pair = exchange_params.get("futures_pair", exchange_params["pair"])
"futures_pair", EXCHANGES[futures_name]["pair"]
)
assert isinstance(leverage_tiers, dict) assert isinstance(leverage_tiers, dict)
assert futures_pair in leverage_tiers assert futures_pair in leverage_tiers
pair_tiers = leverage_tiers[futures_pair] pair_tiers = leverage_tiers[futures_pair]
@@ -532,11 +528,9 @@ class TestCCXTExchange:
oldmaxNotional = tier["maxNotional"] oldmaxNotional = tier["maxNotional"]
def test_ccxt_dry_run_liquidation_price(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_dry_run_liquidation_price(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
futures, futures_name = exchange_futures futures, _, exchange_params = exchange_futures
if EXCHANGES[futures_name].get("leverage_tiers_public"): if exchange_params.get("leverage_tiers_public"):
futures_pair = EXCHANGES[futures_name].get( futures_pair = exchange_params.get("futures_pair", exchange_params["pair"])
"futures_pair", EXCHANGES[futures_name]["pair"]
)
liquidation_price = futures.dry_run_liquidation_price( liquidation_price = futures.dry_run_liquidation_price(
pair=futures_pair, pair=futures_pair,
@@ -565,19 +559,19 @@ class TestCCXTExchange:
assert liquidation_price >= 0.0 assert liquidation_price >= 0.0
def test_ccxt_get_max_pair_stake_amount(self, exchange_futures: EXCHANGE_FIXTURE_TYPE): def test_ccxt_get_max_pair_stake_amount(self, exchange_futures: EXCHANGE_FIXTURE_TYPE):
futures, futures_name = exchange_futures futures, _, exchange_params = exchange_futures
futures_pair = EXCHANGES[futures_name].get("futures_pair", EXCHANGES[futures_name]["pair"]) futures_pair = exchange_params.get("futures_pair", exchange_params["pair"])
max_stake_amount = futures.get_max_pair_stake_amount(futures_pair, 40000) max_stake_amount = futures.get_max_pair_stake_amount(futures_pair, 40000)
assert isinstance(max_stake_amount, float) assert isinstance(max_stake_amount, float)
assert max_stake_amount >= 0.0 assert max_stake_amount >= 0.0
def test_private_method_presence(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_private_method_presence(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, _, exchange_params = exchange
for method in EXCHANGES[exchangename].get("private_methods", []): for method in exchange_params.get("private_methods", []):
assert hasattr(exch._api, method) assert hasattr(exch._api, method)
def test_ccxt_bitget_ohlcv_candle_limit(self, exchange: EXCHANGE_FIXTURE_TYPE): def test_ccxt_bitget_ohlcv_candle_limit(self, exchange: EXCHANGE_FIXTURE_TYPE):
exch, exchangename = exchange exch, exchangename, _ = exchange
if exchangename != "bitget": if exchangename != "bitget":
pytest.skip("This test is only for the Bitget exchange") pytest.skip("This test is only for the Bitget exchange")