392 lines
9.2 KiB
Python
392 lines
9.2 KiB
Python
# app/tests/unit/integrations/exchange/test_service_candles.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast
|
|
|
|
import pytest
|
|
|
|
from src.integrations.exchange.exceptions import ExchangeError
|
|
from src.integrations.exchange.models import SymbolValidationResult
|
|
from src.integrations.exchange.service import ExchangeService
|
|
from src.market_data.acquisition.models.candle import Candle
|
|
|
|
|
|
def _set_test_attribute(
|
|
target: object,
|
|
name: str,
|
|
value: object,
|
|
) -> None:
|
|
setattr(cast(Any, target), name, value)
|
|
|
|
|
|
def _service(
|
|
*,
|
|
default_symbol: str = "BTC/USD_LEVERAGE",
|
|
exchange_enabled: bool = True,
|
|
) -> ExchangeService:
|
|
service = ExchangeService.__new__(ExchangeService)
|
|
_set_test_attribute(
|
|
service,
|
|
"settings",
|
|
SimpleNamespace(
|
|
default_symbol=default_symbol,
|
|
exchange_enabled=exchange_enabled,
|
|
),
|
|
)
|
|
return service
|
|
|
|
|
|
def _valid_symbol(
|
|
symbol: str = "BTC/USD_LEVERAGE",
|
|
) -> SymbolValidationResult:
|
|
return SymbolValidationResult(
|
|
requested_symbol=symbol,
|
|
normalized_symbol=symbol,
|
|
is_valid=True,
|
|
message="OK",
|
|
symbol_info=None,
|
|
)
|
|
|
|
|
|
def _invalid_symbol(
|
|
symbol: str = "UNKNOWN",
|
|
) -> SymbolValidationResult:
|
|
return SymbolValidationResult(
|
|
requested_symbol=symbol,
|
|
normalized_symbol=symbol,
|
|
is_valid=False,
|
|
message="Invalid symbol.",
|
|
symbol_info=None,
|
|
)
|
|
|
|
|
|
def _candle(
|
|
*,
|
|
symbol: str = "BTC/USD_LEVERAGE",
|
|
interval: str = "1m",
|
|
open_time_ms: int = 1_750_000_000_000,
|
|
) -> Candle:
|
|
return Candle(
|
|
symbol=symbol,
|
|
interval=interval,
|
|
open_time=datetime.fromtimestamp(
|
|
open_time_ms / 1000,
|
|
tz=timezone.utc,
|
|
),
|
|
open_price=Decimal("100.10"),
|
|
high_price=Decimal("110.20"),
|
|
low_price=Decimal("90.30"),
|
|
close_price=Decimal("105.40"),
|
|
volume=Decimal("12.50"),
|
|
source="rest_klines:bid",
|
|
)
|
|
|
|
|
|
def test_get_candles_uses_default_symbol() -> None:
|
|
service = _service(default_symbol="ETH/USD_LEVERAGE")
|
|
requested_symbols: list[str] = []
|
|
acquisition_calls: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: (
|
|
requested_symbols.append(symbol) or _valid_symbol(symbol)
|
|
),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: acquisition_calls.append(kwargs) or (),
|
|
)
|
|
|
|
result = service.get_candles()
|
|
|
|
assert result == ()
|
|
assert requested_symbols == ["ETH/USD_LEVERAGE"]
|
|
assert acquisition_calls == [
|
|
{
|
|
"symbol": "ETH/USD_LEVERAGE",
|
|
"interval": "1m",
|
|
"limit": 200,
|
|
"price_type": "bid",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize("limit", [0, -1, -100])
|
|
def test_get_candles_normalizes_non_positive_limit(limit: int) -> None:
|
|
service = _service()
|
|
captured: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: captured.append(kwargs) or (),
|
|
)
|
|
|
|
service.get_candles(limit=limit)
|
|
|
|
assert captured[0]["limit"] == 200
|
|
|
|
|
|
def test_get_candles_caps_limit_at_200() -> None:
|
|
service = _service()
|
|
captured: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: captured.append(kwargs) or (),
|
|
)
|
|
|
|
service.get_candles(limit=500)
|
|
|
|
assert captured[0]["limit"] == 200
|
|
|
|
|
|
@pytest.mark.parametrize("interval", ["1m", "5m", "15m", "1h"])
|
|
def test_get_candles_accepts_supported_intervals(interval: str) -> None:
|
|
service = _service()
|
|
captured: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: captured.append(kwargs) or (),
|
|
)
|
|
|
|
service.get_candles(interval=interval)
|
|
|
|
assert captured[0]["interval"] == interval
|
|
|
|
|
|
def test_get_candles_rejects_unsupported_interval() -> None:
|
|
service = _service()
|
|
|
|
with pytest.raises(
|
|
ExchangeError,
|
|
match="Unsupported kline interval",
|
|
):
|
|
service.get_candles(interval="4h")
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("price_type", "expected"),
|
|
[
|
|
("bid", "bid"),
|
|
("ask", "ask"),
|
|
(" BID ", "bid"),
|
|
(" AsK ", "ask"),
|
|
("unknown", "bid"),
|
|
("", "bid"),
|
|
],
|
|
)
|
|
def test_get_candles_normalizes_price_type(
|
|
price_type: str,
|
|
expected: str,
|
|
) -> None:
|
|
service = _service()
|
|
captured: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: captured.append(kwargs) or (),
|
|
)
|
|
|
|
service.get_candles(price_type=price_type)
|
|
|
|
assert captured[0]["price_type"] == expected
|
|
|
|
|
|
def test_get_candles_rejects_mock_mode() -> None:
|
|
service = _service(exchange_enabled=False)
|
|
|
|
with pytest.raises(
|
|
ExchangeError,
|
|
match="Candles are not available in mock exchange mode",
|
|
):
|
|
service.get_candles()
|
|
|
|
|
|
def test_get_candles_rejects_invalid_symbol() -> None:
|
|
service = _service()
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _invalid_symbol(symbol),
|
|
)
|
|
|
|
with pytest.raises(ExchangeError, match="Invalid symbol"):
|
|
service.get_candles("UNKNOWN")
|
|
|
|
|
|
def test_get_candles_calls_acquisition_with_exact_arguments() -> None:
|
|
service = _service()
|
|
captured: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol("BTC/USD_LEVERAGE"),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: captured.append(kwargs) or (),
|
|
)
|
|
|
|
service.get_candles(
|
|
" btc/usd ",
|
|
interval="5m",
|
|
limit=123,
|
|
price_type="ask",
|
|
)
|
|
|
|
assert captured == [
|
|
{
|
|
"symbol": "BTC/USD_LEVERAGE",
|
|
"interval": "5m",
|
|
"limit": 123,
|
|
"price_type": "ask",
|
|
}
|
|
]
|
|
|
|
|
|
def test_get_candles_preserves_result_identity() -> None:
|
|
service = _service()
|
|
candles = (_candle(),)
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: candles,
|
|
)
|
|
|
|
result = service.get_candles()
|
|
|
|
assert result is candles
|
|
|
|
|
|
def test_get_candles_preserves_order() -> None:
|
|
service = _service()
|
|
first = _candle(open_time_ms=1000)
|
|
second = _candle(open_time_ms=2000)
|
|
candles = (second, first)
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: candles,
|
|
)
|
|
|
|
result = service.get_candles()
|
|
|
|
assert result is candles
|
|
assert result == (second, first)
|
|
|
|
|
|
def test_get_candles_preserves_empty_tuple_identity() -> None:
|
|
service = _service()
|
|
candles: tuple[Candle, ...] = ()
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
lambda **kwargs: candles,
|
|
)
|
|
|
|
result = service.get_candles()
|
|
|
|
assert result is candles
|
|
|
|
|
|
def test_get_candles_logs_and_wraps_acquisition_error() -> None:
|
|
service = _service()
|
|
original_error = RuntimeError("candles unavailable")
|
|
log_calls: list[dict[str, object]] = []
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"validate_symbol",
|
|
lambda symbol: _valid_symbol(symbol),
|
|
)
|
|
|
|
def raise_error(**kwargs: object) -> tuple[Candle, ...]:
|
|
raise original_error
|
|
|
|
_set_test_attribute(
|
|
service,
|
|
"_load_candles_via_acquisition",
|
|
raise_error,
|
|
)
|
|
_set_test_attribute(
|
|
service,
|
|
"_log_exchange_error",
|
|
lambda **kwargs: log_calls.append(kwargs),
|
|
)
|
|
|
|
with pytest.raises(
|
|
ExchangeError,
|
|
match="candles unavailable",
|
|
) as error_info:
|
|
service.get_candles(
|
|
interval="15m",
|
|
limit=25,
|
|
price_type="ask",
|
|
)
|
|
|
|
assert error_info.value.__cause__ is original_error
|
|
assert log_calls == [
|
|
{
|
|
"endpoint": "klines",
|
|
"exc": original_error,
|
|
"symbol": "BTC/USD_LEVERAGE",
|
|
"extra_payload": {
|
|
"interval": "15m",
|
|
"limit": 25,
|
|
"price_type": "ask",
|
|
},
|
|
}
|
|
]
|