gitadmin/dzentra_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

07.4.4.1.9.5 Execution Confidence Engine

Browse Source
This commit is contained in:
Sergey
2026-05-12 14:09:58 +03:00
parent 2be7d92660
commit 1aa8f6c407
6 changed files with 692 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
app/src/telegram/handlers/auto/ui.py
View File

@@ -148,6 +148,29 @@ def _execution_semantic_line(state) -> str:
return str(message)
def _execution_confidence_line(state) -> str:
signal = (state.last_signal or "HOLD").upper()
if signal not in {"BUY", "SELL"}:
return ""
score = getattr(state, "execution_confidence_score", None)
level = getattr(state, "execution_confidence_level", None)
if score is None:
return ""
percent = int(round(float(score) * 100))
if level == "HIGH":
return f"🧠 Уверенность входа · {percent}% высокая"
if level == "NORMAL":
return f"🧠 Уверенность входа · {percent}% нормальная"
return f"🧠 Уверенность входа · {percent}% низкая"
def _build_waiting_text(state) -> str:
price = _signal_entry_price(state)
@@ -164,6 +187,7 @@ def _build_waiting_text(state) -> str:
signal_lines = [
_signal_line(state),
_signal_confirmation_line(state),
_execution_confidence_line(state),
_market_semantic_line(state),
_entry_block_line(state),
_execution_semantic_line(state),

180
app/src/trading/auto/service.py
View File

@@ -29,6 +29,8 @@ class AutoTradeService:
# минимальная уверенность для готовности к будущему execution
_ready_confidence = 0.3
# минимальный итоговый execution confidence для допуска входа
_execution_confidence_required_score = 0.55
_signal_ttl_seconds = 90
_market_analysis_ttl_seconds = 180
@@ -376,6 +378,14 @@ class AutoTradeService:
state.execution_semantic_status = None
state.execution_semantic_message = None
state.execution_semantic_reason = None
state.execution_quality = None
state.execution_quality_reason = None
state.execution_quality_message = None
state.execution_confidence_score = None
state.execution_confidence_level = None
state.execution_confidence_required_score = self._execution_confidence_required_score
state.execution_confidence_reason = None
state.execution_confidence_factors = None
state.signal_started_at = None
state.signal_updated_at = None
state.market_state = None
@@ -395,9 +405,6 @@ class AutoTradeService:
state.runtime_expired_message = None
state.snapshot_age_seconds = None
state.spread_percent = None
state.execution_quality = None
state.execution_quality_reason = None
state.execution_quality_message = None
# собрать контекст для стратегии
def _build_strategy_context(self) -> StrategyContext:
@@ -515,6 +522,24 @@ class AutoTradeService:
f"{confidence:.2f} < {self._ready_confidence:.2f}."
)
return
self._sync_execution_confidence_state(
state=state,
signal=signal,
confidence=confidence,
)
if (
state.execution_confidence_score is not None
and state.execution_confidence_score < self._execution_confidence_required_score
):
state.decision_status = "BLOCKED"
state.decision_reason = (
f"Execution confidence низкий: "
f"{state.execution_confidence_score:.2f} < "
f"{self._execution_confidence_required_score:.2f}."
)
return
state.is_signal_ready = True
state.signal_confirmation_progress = 1.0
@@ -1271,6 +1296,140 @@ class AutoTradeService:
except Exception:
pass
def _sync_execution_confidence_state(
self,
*,
state: AutoTradeState,
signal: str,
confidence: float,
) -> None:
if signal not in {"BUY", "SELL"}:
state.execution_confidence_score = None
state.execution_confidence_level = None
state.execution_confidence_required_score = self._execution_confidence_required_score
state.execution_confidence_reason = None
state.execution_confidence_factors = None
return
signal_score = self._clamp_score(confidence)
confirmation_score = self._clamp_score(state.signal_confirmation_progress)
market_score = self._market_confidence_score(state)
execution_score = self._execution_quality_confidence_score(state)
score = (
signal_score * 0.35
+ confirmation_score * 0.20
+ market_score * 0.25
+ execution_score * 0.20
)
score = round(self._clamp_score(score), 3)
state.execution_confidence_score = score
state.execution_confidence_required_score = self._execution_confidence_required_score
state.execution_confidence_level = self._execution_confidence_level(score)
state.execution_confidence_reason = self._execution_confidence_reason(state)
state.execution_confidence_factors = {
"signal_score": round(signal_score, 3),
"confirmation_score": round(confirmation_score, 3),
"market_score": round(market_score, 3),
"execution_score": round(execution_score, 3),
"required_score": self._execution_confidence_required_score,
"market_state": state.market_state,
"market_trend": state.market_trend,
"market_trend_strength": state.market_trend_strength,
"market_trend_quality": state.market_trend_quality,
"market_phase": state.market_phase,
"execution_quality": state.execution_quality,
"execution_quality_reason": state.execution_quality_reason,
"spread_percent": state.spread_percent,
}
def _market_confidence_score(self, state: AutoTradeState) -> float:
market_state = state.market_state
strength = state.market_trend_strength
quality = state.market_trend_quality
phase = state.market_phase
if market_state in {"HIGH_VOLATILITY", "LOW_VOLATILITY", "RANGE", "UNKNOWN", None}:
return 0.25
score = 0.65
if strength == "STRONG":
score += 0.2
elif strength == "NORMAL":
score += 0.1
elif strength == "WEAK":
score -= 0.25
if quality == "CLEAN":
score += 0.1
elif quality == "NOISY":
score -= 0.25
if phase == "IMPULSE":
score += 0.1
elif phase == "PULLBACK":
score -= 0.25
elif phase in {"RANGE", "SQUEEZE"}:
score -= 0.3
return self._clamp_score(score)
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
def _execution_confidence_level(self, score: float) -> str:
if score >= 0.75:
return "HIGH"
if score >= self._execution_confidence_required_score:
return "NORMAL"
return "LOW"
def _execution_confidence_reason(self, state: AutoTradeState) -> str:
score = state.execution_confidence_score
if score is None:
return "execution confidence не рассчитан"
if score < self._execution_confidence_required_score:
return "низкая совокупная уверенность входа"
if state.execution_confidence_level == "HIGH":
return "высокая совокупная уверенность входа"
return "достаточная совокупная уверенность входа"
def _clamp_score(self, value: float | int | None) -> float:
if value is None:
return 0.0
return max(0.0, min(1.0, float(value)))
def _sync_execution_semantic_state(self, state: AutoTradeState) -> None:
if state.execution_quality == "BLOCKED":
state.execution_semantic_status = "BLOCKED"
@@ -1278,6 +1437,21 @@ class AutoTradeService:
state.execution_semantic_reason = state.execution_quality_reason
return
if state.decision_status == "BLOCKED":
state.execution_semantic_status = "BLOCKED"
if (
state.execution_confidence_score is not None
and state.execution_confidence_score < self._execution_confidence_required_score
):
state.execution_semantic_message = "⛔ Исполнение · низкая уверенность"
state.execution_semantic_reason = state.execution_confidence_reason
return
state.execution_semantic_message = "⛔ Исполнение · сигнал заблокирован"
state.execution_semantic_reason = state.decision_reason
return
if state.position_side != "NONE":
state.execution_semantic_status = "POSITION_OPEN"
state.execution_semantic_message = "📌 Исполнение · позиция открыта"

17
app/src/trading/auto/state.py
View File

@@ -192,4 +192,19 @@ class AutoTradeState:
execution_semantic_message: str | None = None
# техническая детализация для логов / отладки
execution_semantic_reason: str | None = None
execution_semantic_reason: str | None = None
# итоговая execution confidence от 0.0 до 1.0
execution_confidence_score: float | None = None
# уровень confidence: LOW / NORMAL / HIGH
execution_confidence_level: str | None = None
# минимальный score для допуска execution
execution_confidence_required_score: float | None = None
# человекочитаемая причина confidence-оценки
execution_confidence_reason: str | None = None
# детализация факторов confidence для логов / отладки
execution_confidence_factors: dict | None = None