mirror of
https://github.com/OpenByteInc/QuantDinger.git
synced 2026-09-28 15:22:56 +08:00
75 lines
2.7 KiB
Python
75 lines
2.7 KiB
Python
"""Tests for unified strategy auto-stop helpers."""
|
|
|
|
import pytest
|
|
|
|
from app.services.strategy_lifecycle import (
|
|
is_fatal_exchange_error,
|
|
is_recoverable_position_error,
|
|
)
|
|
|
|
|
|
def test_binance_auth_fatal():
|
|
assert is_fatal_exchange_error('Binance HTTP 401: {"code":-2015,"msg":"Invalid API-key"}')
|
|
|
|
|
|
def test_ibkr_connection_fatal():
|
|
assert is_fatal_exchange_error("Connect call failed ('127.0.0.1', 7497)")
|
|
|
|
|
|
def test_bitget_ip_fatal():
|
|
assert is_fatal_exchange_error("Bitget error 40018: Invalid IP")
|
|
|
|
|
|
def test_okx_missing_credentials_fatal():
|
|
assert is_fatal_exchange_error("Missing OKX api_key/secret_key/passphrase")
|
|
|
|
|
|
def test_internal_order_adapter_signature_error_is_fatal():
|
|
assert is_fatal_exchange_error(
|
|
"make_client_order_id() got an unexpected keyword argument 'market_type'"
|
|
)
|
|
|
|
|
|
def test_binance_reduce_only_conflict_is_recoverable():
|
|
assert is_recoverable_position_error(
|
|
'Binance HTTP 400: {"code":-2022,"msg":"ReduceOnly Order is rejected."}'
|
|
)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"reason",
|
|
[
|
|
"OKX error: {'sCode': '51137', 'sMsg': 'The highest price limit for the buy leg is 2,666.99.'}",
|
|
"OKX error: {'sCode': '51138', 'sMsg': 'The lowest price limit for the sell leg is 2,640.44.'}",
|
|
'Binance HTTP 400: {"code":-4016,"msg":"Price is higher than mark price multiplier cap."}',
|
|
'Binance HTTP 400: {"code":-4024,"msg":"Price is lower than mark price multiplier floor."}',
|
|
'Bybit error: {"retCode":110003,"retMsg":"Order price exceeds the allowable range."}',
|
|
"Bitget error 25206: ETH trading price cannot exceed 5%",
|
|
"Gate error PRICE_TOO_DEVIATED: Order price deviates too much from mark price",
|
|
],
|
|
)
|
|
def test_dynamic_price_band_rejection_is_recoverable(reason):
|
|
assert is_recoverable_position_error(reason)
|
|
|
|
|
|
def test_maybe_auto_stop_fatal():
|
|
from unittest.mock import patch
|
|
|
|
from app.services.strategy_lifecycle import maybe_auto_stop_on_exchange_error
|
|
|
|
with patch("app.services.strategy_lifecycle.auto_stop_live_strategy") as stop:
|
|
assert maybe_auto_stop_on_exchange_error(1, "Missing OKX api_key/secret_key/passphrase", source="test")
|
|
stop.assert_called_once()
|
|
|
|
|
|
def test_maybe_auto_stop_repeated():
|
|
from unittest.mock import patch
|
|
|
|
from app.services.strategy_lifecycle import maybe_auto_stop_on_exchange_error
|
|
|
|
with patch("app.services.strategy_lifecycle.auto_stop_live_strategy") as stop:
|
|
assert not maybe_auto_stop_on_exchange_error(1, "timeout", consecutive_failures=2, consecutive_threshold=5)
|
|
stop.assert_not_called()
|
|
assert maybe_auto_stop_on_exchange_error(1, "timeout", consecutive_failures=5, consecutive_threshold=5)
|
|
stop.assert_called_once()
|