84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
# app/tests/unit/trading/auto/test_execution_quality.py
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
import src.trading.auto.execution_quality as module
|
|
from src.integrations.exchange.models import ExecutionPriceSnapshot
|
|
from src.trading.auto.execution_quality import AutoExecutionQualityMixin
|
|
|
|
|
|
class Harness(AutoExecutionQualityMixin):
|
|
_spread_thresholds_by_asset = {}
|
|
_default_spread_thresholds = {
|
|
"warning_enter": 1.0,
|
|
"warning_exit": 0.8,
|
|
"block_enter": 2.0,
|
|
"block_exit": 1.5,
|
|
}
|
|
_max_snapshot_age_seconds = 5.0
|
|
_warning_snapshot_age_seconds = 2.0
|
|
_last_logged_execution_quality_key = None
|
|
|
|
def _log_execution_quality_if_changed(self, **_: object) -> None:
|
|
return None
|
|
|
|
def _apply_exchange_block_state(self, **_: object) -> None:
|
|
raise AssertionError("exchange block must not be applied")
|
|
|
|
|
|
def _state() -> SimpleNamespace:
|
|
return SimpleNamespace(
|
|
market_is_open=True,
|
|
symbol="BTC/USD_LEVERAGE",
|
|
strategy="trend",
|
|
status="RUNNING",
|
|
|
|
execution_quality=None,
|
|
execution_quality_reason=None,
|
|
execution_quality_message=None,
|
|
execution_block_reason=None,
|
|
|
|
market_runtime_degraded=False,
|
|
snapshot_age_seconds=None,
|
|
spread_percent=None,
|
|
|
|
execution_price_age_seconds=None,
|
|
execution_bid_price=None,
|
|
execution_ask_price=None,
|
|
execution_last_price=None,
|
|
execution_price_freshness=None,
|
|
)
|
|
|
|
def test_execution_quality_uses_typed_execution_snapshot(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
snapshot = ExecutionPriceSnapshot(
|
|
symbol="BTC/USD_LEVERAGE",
|
|
last_price=100.5,
|
|
bid_price=100.0,
|
|
ask_price=101.0,
|
|
updated_at="13.07.2026 15:00:00",
|
|
source="dzengi:fresh_cache",
|
|
is_fresh=True,
|
|
age_seconds=0.5,
|
|
)
|
|
|
|
class Service:
|
|
def get_execution_snapshot(self, *_: object, **__: object) -> ExecutionPriceSnapshot:
|
|
return snapshot
|
|
|
|
monkeypatch.setattr(module, "ExchangeService", Service)
|
|
|
|
state = _state()
|
|
Harness()._sync_execution_quality_state(state) # type: ignore[arg-type]
|
|
|
|
assert state.execution_bid_price == 100.0
|
|
assert state.execution_ask_price == 101.0
|
|
assert state.execution_last_price == 100.5
|
|
assert state.execution_price_source == "dzengi:fresh_cache"
|
|
assert state.snapshot_age_seconds == 0.5
|