541 lines
19 KiB
Python
541 lines
19 KiB
Python
# app/src/trading/auto/execution_quality.py
|
|
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
|
|
from src.core.numbers import safe_float
|
|
from src.core.types import NumericLike
|
|
from src.integrations.exchange.service import ExchangeService
|
|
from src.integrations.exchange.status import (
|
|
ExchangeRuntimeStatus,
|
|
ExchangeStatusCode,
|
|
build_exchange_error_status,
|
|
)
|
|
from src.trading.auto.state import AutoTradeState
|
|
from src.trading.journal.service import JournalService
|
|
|
|
|
|
class AutoExecutionQualityMixin:
|
|
_spread_thresholds_by_asset: dict[str, dict[str, float]]
|
|
_default_spread_thresholds: dict[str, float]
|
|
|
|
_max_snapshot_age_seconds: float
|
|
_warning_snapshot_age_seconds: float
|
|
|
|
_last_logged_execution_quality_key: str | None
|
|
|
|
# получить базовый asset из symbol для spread thresholds
|
|
def _asset_symbol(self, symbol: str | None) -> str:
|
|
if not symbol:
|
|
return ""
|
|
|
|
base = str(symbol).split("_", 1)[0].upper()
|
|
|
|
if "/" in base:
|
|
return base.split("/", 1)[0]
|
|
|
|
for suffix in ("USDT", "USD", "EUR", "BTC"):
|
|
if base.endswith(suffix) and len(base) > len(suffix):
|
|
return base[: -len(suffix)]
|
|
|
|
return base
|
|
|
|
# получить spread thresholds для конкретного инструмента
|
|
def _spread_thresholds(self, symbol: str | None) -> dict[str, float]:
|
|
asset = self._asset_symbol(symbol)
|
|
|
|
return self._spread_thresholds_by_asset.get(
|
|
asset,
|
|
self._default_spread_thresholds,
|
|
)
|
|
|
|
# синхронизировать единый статус биржи/торговой сессии в AutoTradeState
|
|
def _sync_market_availability_state(self, state: AutoTradeState) -> bool:
|
|
try:
|
|
status = ExchangeService().get_symbol_runtime_status(state.symbol)
|
|
except Exception as exc:
|
|
status = build_exchange_error_status(exc)
|
|
|
|
state.market_is_open = status.is_open
|
|
state.market_status = status.code.value
|
|
state.market_status_message = status.ui_line
|
|
state.market_status_updated_at = time.monotonic()
|
|
|
|
if status.is_open:
|
|
self._clear_exchange_block_state(state)
|
|
return True
|
|
|
|
self._apply_exchange_block_state(
|
|
state=state,
|
|
status=status,
|
|
)
|
|
|
|
return False
|
|
|
|
# очистить старую блокировку биржи, если рынок снова доступен
|
|
def _clear_exchange_block_state(self, state: AutoTradeState) -> None:
|
|
if state.execution_quality_reason not in {
|
|
"MARKET_BREAK",
|
|
"EXCHANGE_UNAVAILABLE",
|
|
"AUTH_ERROR",
|
|
"TIME_ERROR",
|
|
"INVALID_SYMBOL",
|
|
"MARKET_CLOSED",
|
|
}:
|
|
return
|
|
|
|
state.execution_quality = None
|
|
state.execution_quality_reason = None
|
|
state.execution_quality_message = None
|
|
state.execution_block_reason = None
|
|
state.market_runtime_degraded = False
|
|
|
|
state.entry_block_reason = None
|
|
state.entry_block_message = None
|
|
|
|
# применить блокировку execution по единому ExchangeRuntimeStatus
|
|
def _apply_exchange_block_state(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
status: ExchangeRuntimeStatus,
|
|
) -> None:
|
|
reason = self._exchange_execution_reason(status)
|
|
message = status.ui_line or status.message
|
|
|
|
state.execution_quality = "BLOCKED"
|
|
state.execution_quality_reason = reason
|
|
state.execution_quality_message = message
|
|
state.execution_block_reason = message
|
|
state.market_runtime_degraded = True
|
|
|
|
state.entry_block_reason = reason
|
|
state.entry_block_message = message
|
|
|
|
state.decision_status = "WAITING"
|
|
state.decision_reason = message
|
|
state.is_signal_confirmed = False
|
|
state.is_signal_ready = False
|
|
|
|
self._log_exchange_availability_if_changed(
|
|
state=state,
|
|
status=status,
|
|
reason=reason,
|
|
)
|
|
|
|
# преобразовать typed exchange status в код причины execution layer
|
|
def _exchange_execution_reason(self, status: ExchangeRuntimeStatus) -> str:
|
|
if status.code == ExchangeStatusCode.BREAK:
|
|
return "MARKET_BREAK"
|
|
|
|
if status.code == ExchangeStatusCode.AUTH_ERROR:
|
|
return "AUTH_ERROR"
|
|
|
|
if status.code == ExchangeStatusCode.TIME_ERROR:
|
|
return "TIME_ERROR"
|
|
|
|
if status.code == ExchangeStatusCode.INVALID_SYMBOL:
|
|
return "INVALID_SYMBOL"
|
|
|
|
if status.code == ExchangeStatusCode.EXCHANGE_UNAVAILABLE:
|
|
return "EXCHANGE_UNAVAILABLE"
|
|
|
|
return "MARKET_BREAK"
|
|
|
|
# собрать payload изменения доступности биржи без изменения состояния
|
|
def _build_exchange_availability_payload(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
status: ExchangeRuntimeStatus,
|
|
reason: str,
|
|
) -> dict[str, object]:
|
|
return {
|
|
"status": state.status,
|
|
"symbol": state.symbol,
|
|
"strategy": state.strategy,
|
|
"exchange_status_code": status.code.value,
|
|
"exchange_reason": status.reason,
|
|
"execution_reason": reason,
|
|
"is_open": status.is_open,
|
|
"is_available": status.is_available,
|
|
"is_auth_ok": status.is_auth_ok,
|
|
"message": status.message,
|
|
"raw_status": status.raw_status,
|
|
"raw_error": status.raw_error,
|
|
}
|
|
|
|
# залогировать изменение доступности биржи/рынка
|
|
def _log_exchange_availability_if_changed(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
status: ExchangeRuntimeStatus,
|
|
reason: str,
|
|
) -> None:
|
|
key = (
|
|
f"{state.status}:{state.symbol}:{state.strategy}:"
|
|
f"{status.code.value}:{reason}:{status.ui_line}"
|
|
)
|
|
|
|
if key == type(self)._last_logged_execution_quality_key:
|
|
return
|
|
|
|
type(self)._last_logged_execution_quality_key = key
|
|
|
|
try:
|
|
JournalService().log_ui_warning(
|
|
event_type="exchange_availability_changed",
|
|
message=status.ui_line,
|
|
screen="auto",
|
|
action="exchange_status",
|
|
payload=self._build_exchange_availability_payload(
|
|
state=state,
|
|
status=status,
|
|
reason=reason,
|
|
),
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# рассчитать качество исполнения на основе spread
|
|
def _spread_execution_quality(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
spread_percent: NumericLike | None,
|
|
) -> tuple[str | None, str | None, str | None, bool]:
|
|
spread = safe_float(spread_percent)
|
|
|
|
if spread is None:
|
|
return None, None, None, False
|
|
|
|
thresholds = self._spread_thresholds(state.symbol)
|
|
|
|
warning_enter = thresholds["warning_enter"]
|
|
warning_exit = thresholds["warning_exit"]
|
|
block_enter = thresholds["block_enter"]
|
|
block_exit = thresholds["block_exit"]
|
|
|
|
previous_quality = state.execution_quality
|
|
previous_reason = state.execution_quality_reason
|
|
|
|
if previous_quality == "BLOCKED" and previous_reason == "HIGH_SPREAD":
|
|
if spread > block_exit:
|
|
return "BLOCKED", "HIGH_SPREAD", "высокий spread", False
|
|
|
|
if spread > warning_exit:
|
|
return "WARNING", "WIDE_SPREAD", "spread повышен", False
|
|
|
|
return "GOOD", "MARKET_OK", "рынок готов", False
|
|
|
|
if previous_quality == "WARNING" and previous_reason == "WIDE_SPREAD":
|
|
if spread >= block_enter:
|
|
return "BLOCKED", "HIGH_SPREAD", "высокий spread", False
|
|
|
|
if spread > warning_exit:
|
|
return "WARNING", "WIDE_SPREAD", "spread повышен", False
|
|
|
|
return "GOOD", "MARKET_OK", "рынок готов", False
|
|
|
|
if spread >= block_enter:
|
|
return "BLOCKED", "HIGH_SPREAD", "высокий spread", False
|
|
|
|
if spread >= warning_enter:
|
|
return "WARNING", "WIDE_SPREAD", "spread повышен", False
|
|
|
|
return "GOOD", "MARKET_OK", "рынок готов", False
|
|
|
|
# синхронизировать runtime quality исполнения
|
|
def _sync_execution_quality_state(self, state: AutoTradeState) -> None:
|
|
if state.market_is_open is False:
|
|
return
|
|
|
|
try:
|
|
snapshot = ExchangeService().get_market_snapshot(
|
|
state.symbol,
|
|
runtime_key="auto",
|
|
)
|
|
|
|
age_seconds = safe_float(snapshot.get("age_seconds"))
|
|
|
|
if (
|
|
age_seconds is not None
|
|
and age_seconds > self._warning_snapshot_age_seconds
|
|
):
|
|
try:
|
|
snapshot = ExchangeService().refresh_market_snapshot_cache(
|
|
state.symbol,
|
|
runtime_key="auto",
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
except Exception as exc:
|
|
fallback_price = None
|
|
|
|
try:
|
|
fallback_price = safe_float(
|
|
ExchangeService().get_price(
|
|
state.symbol,
|
|
runtime_key="auto",
|
|
).price
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# Snapshot недоступен — очищаем все pricing-поля,
|
|
# чтобы UI/execution не использовали старые bid/ask/last.
|
|
state.snapshot_age_seconds = None
|
|
state.spread_percent = None
|
|
state.execution_price_source = None
|
|
state.execution_price_age_seconds = None
|
|
state.execution_bid_price = None
|
|
state.execution_ask_price = None
|
|
state.execution_last_price = fallback_price
|
|
state.execution_price_freshness = "UNKNOWN"
|
|
|
|
if fallback_price is not None and fallback_price > 0:
|
|
state.execution_quality = "WARNING"
|
|
state.execution_quality_reason = "SNAPSHOT_UNAVAILABLE"
|
|
state.execution_quality_message = "нет depth snapshot"
|
|
state.execution_block_reason = None
|
|
state.market_runtime_degraded = True
|
|
else:
|
|
status = build_exchange_error_status(exc)
|
|
self._apply_exchange_block_state(
|
|
state=state,
|
|
status=status,
|
|
)
|
|
|
|
self._log_execution_quality_if_changed(
|
|
state=state,
|
|
payload={
|
|
"error": str(exc),
|
|
"error_type": type(exc).__name__,
|
|
"fallback_price_available": fallback_price is not None,
|
|
},
|
|
)
|
|
return
|
|
|
|
bid_price = safe_float(snapshot.get("bid_price"))
|
|
ask_price = safe_float(snapshot.get("ask_price"))
|
|
last_price = safe_float(snapshot.get("last_price"))
|
|
age_seconds = safe_float(snapshot.get("age_seconds"))
|
|
is_fresh = bool(snapshot.get("is_fresh", False))
|
|
source = str(snapshot.get("source") or "")
|
|
|
|
self._sync_execution_pricing_state(
|
|
state,
|
|
snapshot,
|
|
)
|
|
|
|
state.snapshot_age_seconds = age_seconds
|
|
state.spread_percent = self._spread_percent(
|
|
bid_price=bid_price,
|
|
ask_price=ask_price,
|
|
)
|
|
|
|
if age_seconds is not None and age_seconds > self._max_snapshot_age_seconds:
|
|
state.execution_quality = "BLOCKED"
|
|
state.execution_quality_reason = "STALE_SNAPSHOT"
|
|
state.execution_quality_message = "snapshot устарел"
|
|
state.market_runtime_degraded = True
|
|
|
|
elif age_seconds is not None and age_seconds > self._warning_snapshot_age_seconds:
|
|
state.execution_quality = "WARNING"
|
|
state.execution_quality_reason = "AGING_SNAPSHOT"
|
|
state.execution_quality_message = "snapshot стареет"
|
|
state.market_runtime_degraded = not is_fresh
|
|
|
|
elif state.spread_percent is not None:
|
|
(
|
|
state.execution_quality,
|
|
state.execution_quality_reason,
|
|
state.execution_quality_message,
|
|
state.market_runtime_degraded,
|
|
) = self._spread_execution_quality(
|
|
state=state,
|
|
spread_percent=state.spread_percent,
|
|
)
|
|
|
|
else:
|
|
state.execution_quality = "GOOD"
|
|
state.execution_quality_reason = "MARKET_OK"
|
|
state.execution_quality_message = "рынок готов"
|
|
state.market_runtime_degraded = False
|
|
|
|
if state.execution_quality == "BLOCKED":
|
|
state.execution_block_reason = state.execution_quality_message
|
|
|
|
elif state.execution_block_reason == state.execution_quality_message:
|
|
state.execution_block_reason = None
|
|
|
|
spread_thresholds = self._spread_thresholds(state.symbol)
|
|
|
|
self._log_execution_quality_if_changed(
|
|
state=state,
|
|
payload={
|
|
"symbol": state.symbol,
|
|
"strategy": state.strategy,
|
|
"bid_price": bid_price,
|
|
"ask_price": ask_price,
|
|
"last_price": last_price,
|
|
"snapshot_age_seconds": age_seconds,
|
|
"spread_percent": state.spread_percent,
|
|
"is_fresh": is_fresh,
|
|
"source": source,
|
|
"execution_quality": state.execution_quality,
|
|
"execution_quality_reason": state.execution_quality_reason,
|
|
"execution_quality_message": state.execution_quality_message,
|
|
"market_runtime_degraded": state.market_runtime_degraded,
|
|
"max_snapshot_age_seconds": self._max_snapshot_age_seconds,
|
|
"warning_snapshot_age_seconds": self._warning_snapshot_age_seconds,
|
|
"spread_asset": self._asset_symbol(state.symbol),
|
|
"spread_warning_enter_percent": spread_thresholds["warning_enter"],
|
|
"spread_warning_exit_percent": spread_thresholds["warning_exit"],
|
|
"spread_block_enter_percent": spread_thresholds["block_enter"],
|
|
"spread_block_exit_percent": spread_thresholds["block_exit"],
|
|
},
|
|
)
|
|
|
|
# рассчитать spread между bid/ask в процентах
|
|
def _spread_percent(
|
|
self,
|
|
*,
|
|
bid_price: NumericLike | None,
|
|
ask_price: NumericLike | None,
|
|
) -> float | None:
|
|
bid = safe_float(bid_price)
|
|
ask = safe_float(ask_price)
|
|
|
|
if bid is None or ask is None:
|
|
return None
|
|
|
|
if bid <= 0 or ask <= 0:
|
|
return None
|
|
|
|
mid_price = (bid + ask) / 2
|
|
|
|
if mid_price <= 0:
|
|
return None
|
|
|
|
spread = ask - bid
|
|
|
|
if spread < 0:
|
|
return None
|
|
|
|
return round((spread / mid_price) * 100, 5)
|
|
|
|
# синхронизировать execution pricing данные в state
|
|
def _sync_execution_pricing_state(
|
|
self,
|
|
state: AutoTradeState,
|
|
snapshot: dict[str, object],
|
|
) -> None:
|
|
age_seconds = safe_float(snapshot.get("age_seconds"))
|
|
|
|
state.execution_price_source = str(snapshot.get("source") or "")
|
|
state.execution_price_age_seconds = age_seconds
|
|
state.execution_bid_price = safe_float(snapshot.get("bid_price"))
|
|
state.execution_ask_price = safe_float(snapshot.get("ask_price"))
|
|
state.execution_last_price = safe_float(snapshot.get("last_price"))
|
|
|
|
if age_seconds is None:
|
|
state.execution_price_freshness = "UNKNOWN"
|
|
elif age_seconds <= 1:
|
|
state.execution_price_freshness = "FRESH"
|
|
elif age_seconds <= self._warning_snapshot_age_seconds:
|
|
state.execution_price_freshness = "AGING"
|
|
else:
|
|
state.execution_price_freshness = "STALE"
|
|
|
|
# собрать payload изменения качества исполнения без изменения состояния
|
|
def _build_execution_quality_payload(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
payload: dict[str, object],
|
|
) -> dict[str, object]:
|
|
return {
|
|
**payload,
|
|
"status": state.status,
|
|
"symbol": state.symbol,
|
|
"strategy": state.strategy,
|
|
}
|
|
|
|
# записать событие изменения execution quality
|
|
def _log_execution_quality_if_changed(
|
|
self,
|
|
*,
|
|
state: AutoTradeState,
|
|
payload: dict[str, object],
|
|
) -> None:
|
|
quality = state.execution_quality
|
|
reason = state.execution_quality_reason
|
|
message = state.execution_quality_message
|
|
|
|
if not quality or not reason or not message:
|
|
return
|
|
|
|
key = f"{state.status}:{state.symbol}:{state.strategy}:{quality}:{reason}:{message}"
|
|
|
|
if key == type(self)._last_logged_execution_quality_key:
|
|
return
|
|
|
|
type(self)._last_logged_execution_quality_key = key
|
|
|
|
if quality == "GOOD":
|
|
return
|
|
|
|
try:
|
|
log_payload = self._build_execution_quality_payload(
|
|
state=state,
|
|
payload=payload,
|
|
)
|
|
|
|
if quality == "BLOCKED":
|
|
JournalService().log_ui_warning(
|
|
event_type="execution_quality_changed",
|
|
message=f"Качество исполнения: {message}.",
|
|
screen="auto",
|
|
action="execution_quality",
|
|
payload=log_payload,
|
|
)
|
|
return
|
|
|
|
JournalService().log_ui_info(
|
|
event_type="execution_quality_changed",
|
|
message=f"Качество исполнения: {message}.",
|
|
screen="auto",
|
|
action="execution_quality",
|
|
payload=log_payload,
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# рассчитать confidence execution quality для общего execution confidence
|
|
def _execution_quality_confidence_score(self, state: AutoTradeState) -> float:
|
|
quality = state.execution_quality
|
|
reason = state.execution_quality_reason
|
|
|
|
if quality == "GOOD":
|
|
return 1.0
|
|
|
|
if quality == "WARNING":
|
|
if reason == "WIDE_SPREAD":
|
|
return 0.65
|
|
|
|
if reason == "AGING_SNAPSHOT":
|
|
return 0.6
|
|
|
|
if reason == "SNAPSHOT_UNAVAILABLE":
|
|
return 0.55
|
|
|
|
return 0.6
|
|
|
|
if quality == "BLOCKED":
|
|
return 0.0
|
|
|
|
return 0.5 |