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

07.4.4.1.12 — Position Health & Runtime Risk Layer

Browse Source
This commit is contained in:
Sergey
2026-05-21 19:32:55 +03:00
parent 06ea376cb5
commit f9a25e7671
10 changed files with 3068 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,21 +41,21 @@ def auto_keyboard() -> InlineKeyboardMarkup:
status = (state.status or "").upper()
if status == "OFF":
builder.button(text="▶️ Start", callback_data="auto:start")
builder.button(text="👀 Watch", callback_data="auto:observe")
builder.button(text="▶️ Запустить", callback_data="auto:start")
builder.button(text="👀 Наблюдать", callback_data="auto:observe")
elif status == "RUNNING":
builder.button(text="👀 Watch", callback_data="auto:observe")
builder.button(text="🛑 Stop", callback_data="auto:stop")
builder.button(text="👀 Наблюдать", callback_data="auto:observe")
builder.button(text="🛑 Остановить", callback_data="auto:stop")
elif status == "OBSERVING":
builder.button(text="▶️ Start", callback_data="auto:start")
builder.button(text="🛑 Stop", callback_data="auto:stop")
builder.button(text="▶️ Запустить", callback_data="auto:start")
builder.button(text="🛑 Остановить", callback_data="auto:stop")
else:
builder.button(text="▶️ Start", callback_data="auto:start")
builder.button(text="👀 Watch", callback_data="auto:observe")
builder.button(text="🛑 Stop", callback_data="auto:stop")
builder.button(text="▶️ Запустить", callback_data="auto:start")
builder.button(text="👀 Наблюдать", callback_data="auto:observe")
builder.button(text="🛑 Остановить", callback_data="auto:stop")
builder.button(text="🛠️ Настройки", callback_data="settings:auto")
builder.button(text="🧯 Защита", callback_data="auto:risk")
@@ -192,7 +192,7 @@ def _settings_risk_percent_line(state) -> str:
ml = (
f"-{_format_money_compact(abs(state.max_loss_usd))}"
if state.max_loss_usd is not None
else "off"
else "выкл"
)
return f"SL {sl} · TP {tp} · ML {ml}"
@@ -205,15 +205,10 @@ def _build_waiting_text(state) -> str:
estimated_size = _estimated_size(state, price)
cycle_trades = int(getattr(state, "cycle_closed_trades", 0) or 0)
cycle_pnl = float(
getattr(state, "cycle_realized_pnl_usd", 0.0) or 0.0
)
cycle_pnl = float(getattr(state, "cycle_realized_pnl_usd", 0.0) or 0.0)
parts = [
(
f"{_status_text(state)}"
),
f"{_status_text(state)}",
_account_mode_line(),
"",
f"Доступно 💰 {_format_money_compact(available)}",
@@ -222,10 +217,7 @@ def _build_waiting_text(state) -> str:
if cycle_trades > 0:
parts.extend([
"",
(
f"🔄 {_cycle_number_text(state)} · "
f"{cycle_trades} {_trade_word(cycle_trades)}"
),
f"🔄 {_cycle_number_text(state)} · {cycle_trades} {_trade_word(cycle_trades)}",
_format_pnl_line(cycle_pnl),
])
@@ -236,37 +228,50 @@ def _build_waiting_text(state) -> str:
parts.extend([
"",
_signal_line(state),
"",
])
execution_runtime_line = _execution_runtime_line(state)
if execution_runtime_line:
parts.extend([
"",
execution_runtime_line,
])
block_title = (
"Прогноз сделки 🔮"
if state.status == "OBSERVING"
else "Подготовка ордера 🧾"
)
parts.extend([
order_lines = [
"",
block_title,
_order_header_line(state),
f"Цена · {_format_plain_or_dash(price)}",
_estimated_size_text(state, price),
_max_reserved_line(state, price),
_effective_risk_line(state),
]
execution_confidence_line = _execution_confidence_line(state)
if execution_confidence_line:
order_lines.append(execution_confidence_line)
order_lines.append(
_risk_summary_line(
state,
estimated_size,
entry_price_override=price,
),
])
)
)
parts.extend(order_lines)
adjustment_visible = _adaptive_adjustment_visible(state)
if adjustment_visible:
reason = getattr(state, "adaptive_size_reason", "") or ""
multiplier = float(
getattr(state, "adaptive_size_multiplier", 1.0) or 1.0
)
multiplier = float(getattr(state, "adaptive_size_multiplier", 1.0) or 1.0)
parts.extend([
"",
@@ -280,6 +285,50 @@ def _build_waiting_text(state) -> str:
return "\n".join(parts)
def _execution_runtime_line(state) -> str:
quality = str(
getattr(state, "execution_quality", "") or ""
).upper()
reason = str(
getattr(state, "execution_quality_reason", "") or ""
).upper()
freshness = _execution_freshness_text(state)
if quality == "GOOD":
return ""
if quality == "WARNING":
if reason == "WIDE_SPREAD":
return f"Исполнение ⚠️ Повышенный spread · {freshness}"
if reason == "AGING_SNAPSHOT":
return f"Исполнение ⚠️ Snapshot стареет · {freshness}"
if reason == "SNAPSHOT_UNAVAILABLE":
return f"Исполнение ⚠️ Нет стакана · {freshness}"
return f"Исполнение ⚠️ Предупреждение · {freshness}"
if quality == "BLOCKED":
if reason == "MARKET_CLOSED":
return "Исполнение ⏸️ Рынок закрыт"
if reason == "STALE_SNAPSHOT":
return f"Исполнение 🔴 Snapshot устарел · {freshness}"
if reason == "HIGH_SPREAD":
return f"Исполнение 🔴 Высокий spread · {freshness}"
if reason == "SNAPSHOT_ERROR":
return "Исполнение 🔴 Нет данных рынка"
return f"Исполнение 🔴 Заблокировано · {freshness}"
return ""
def _build_active_position_text(state) -> str:
current_price = _current_price(state.symbol)
price_for_calc = current_price or state.entry_price or 0.0
@@ -356,9 +405,14 @@ def _build_active_position_text(state) -> str:
),
f"Объём · {_format_usd_compact(notional)}",
_format_pnl_line(pnl),
_risk_summary_line(state, size),
])
execution_runtime_line = _execution_runtime_line(state)
if execution_runtime_line:
parts.append(execution_runtime_line)
parts.append(_risk_summary_line(state, size))
if pnl < 0:
reason_block = _position_warning_reason(state)
@@ -528,6 +582,40 @@ def _effective_risk_line(state) -> str:
return f"Риск · {_format_usd_compact(_target_risk_usd(state))}"
def _execution_confidence_line(state) -> str:
score = safe_float(
getattr(state, "execution_confidence_score", None)
)
if score is None:
return ""
level = str(
getattr(state, "execution_confidence_level", "") or ""
).upper()
if level == "HIGH":
icon = "🟢"
level_text = "Высокая"
elif level == "NORMAL":
icon = "🟡"
level_text = "Нормальная"
elif level == "LOW":
icon = "🔴"
level_text = "Низкая"
else:
icon = "⚪"
level_text = "Неизвестно"
return (
f"Уверенность исполнения "
f"{icon} {level_text} · {score:.2f}"
)
def _estimated_size(state, price: float | None) -> float | None:
if (
price is None
@@ -642,13 +730,13 @@ def _risk_summary_line(
key, value = enabled[0]
if key == "SL":
return f"Stop Loss -{_format_usd_compact(value)}"
return f"Stop Loss −{_format_usd_compact(value)}"
if key == "TP":
return f"Take Profit +{_format_usd_compact(value)}"
if key == "ML":
return f"Max Loss -{_format_usd_compact(value)}"
return f"Макс. убыток −{_format_usd_compact(value)}"
items: list[str] = []
@@ -662,7 +750,7 @@ def _risk_summary_line(
items.append(f"ML -{_format_usd_compact(ml_value)}")
if not items:
return "SL off · TP off · ML off"
return "SL выкл · TP выкл · ML выкл"
return " · ".join(items)
@@ -739,7 +827,7 @@ def _signal_text(signal: str) -> str:
mapping = {
"BUY": "Long",
"SELL": "Short",
"HOLD": "Hold",
"HOLD": "Ожидание",
}
return mapping.get(signal.upper(), signal.title())
@@ -757,7 +845,7 @@ def _signal_line(state) -> str:
state.decision_status == "READY"
or getattr(state, "is_signal_ready", False)
):
return f"{signal_text} · READY"
return f"{signal_text} · готов"
duration = _signal_duration_text(state)
@@ -845,6 +933,21 @@ def _signal_duration_text(state) -> str:
return f"{seconds}с"
def _execution_freshness_text(state) -> str:
freshness = str(
getattr(state, "execution_price_freshness", "") or ""
).upper()
mapping = {
"FRESH": "данные свежие",
"AGING": "данные стареют",
"STALE": "данные устарели",
"UNKNOWN": "нет данных",
}
return mapping.get(freshness, "нет данных")
def _status_text(state) -> str:
runtime = _cycle_runtime_text(state)
@@ -1012,7 +1115,7 @@ def _format_crypto_size(value: float | int | None) -> str:
def _format_percent(value: float | int | None) -> str:
if value is None:
return "off"
return "выкл"
number = float(value)
@@ -1085,24 +1188,21 @@ def _adaptive_adjustment_visible(state) -> bool:
return _adaptive_size_active(state) and _show_adaptive_banner(state)
def _short_adaptive_reason(reason: str, multiplier: float | None = None) -> str:
if multiplier is not None:
try:
value = float(multiplier)
except (TypeError, ValueError):
value = 1.0
def _short_adaptive_reason(
reason: str,
multiplier: float | None = None,
) -> str:
value = safe_float(multiplier)
if value is not None:
if value < 0.15:
return "Вход почти заблокирован"
if value < 0.40:
return "Размер сильно уменьшен"
return "Вход заблокирован"
if value < 0.75:
return "Размер уменьшен"
return "Риск снижен"
if value < 1.0:
return "Небольшая коррекция"
return "Размер уменьшен"
if value > 1.05:
return "Размер увеличен"
@@ -1122,6 +1222,6 @@ def _short_adaptive_reason(reason: str, multiplier: float | None = None) -> str:
return "Низкая уверенность"
if not reason:
return "Размер изменён"
return "Размер скорректирован"
return reason[:1].upper() + reason[1:]

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

@@ -526,6 +526,12 @@ class AutoTradeService:
state.execution_quality = None
state.execution_quality_reason = None
state.execution_quality_message = 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 = None
state.execution_price_freshness = None
state.execution_confidence_score = None
state.execution_confidence_level = None
state.execution_confidence_required_score = self._execution_confidence_required_score
@@ -579,6 +585,49 @@ class AutoTradeService:
state.snapshot_age_seconds = None
state.spread_percent = None
state.position_pnl_percent = None
state.position_hold_seconds = None
state.position_pressure = None
state.position_health_score = None
state.position_health_status = None
state.position_health_reason = None
state.position_risk_level = None
state.position_risk_reason = None
state.position_trend_alignment = None
state.position_adverse_momentum = False
state.position_exit_pressure = None
state.position_lifecycle_stage = None
state.position_hold_quality = None
state.position_decay_state = None
state.position_exit_confidence = None
state.position_exit_signal = None
state.position_intelligence_reason = None
state.position_recommended_action = None
state.position_peak_pnl_usd = None
state.position_peak_pnl_percent = None
state.position_mfe_percent = None
state.position_mae_percent = None
state.position_fatigue_score = None
state.position_fatigue_state = None
state.position_giveback_percent = None
state.position_conviction_state = None
state.position_exit_urgency = None
state.position_reversal_risk = None
state.autonomous_action = None
state.autonomous_action_reason = None
state.autonomous_action_confidence = None
state.autonomous_protection_required = False
state.autonomous_reduce_required = False
state.autonomous_exit_required = False
state.autonomous_last_action = None
state.autonomous_last_action_reason = None
state.autonomous_last_action_at = None
state.last_loss_monotonic_at = None
# собрать контекст для стратегии
def _build_strategy_context(self) -> StrategyContext:
state = self.get_state()
@@ -1394,6 +1443,11 @@ class AutoTradeService:
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,
@@ -1490,6 +1544,767 @@ class AutoTradeService:
return round((spread / mid_price) * 100, 5)
def _sync_execution_pricing_state(
self,
state: AutoTradeState,
snapshot: JsonDict,
) -> 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"
def _sync_position_health_state(self, state: AutoTradeState) -> None:
if state.position_side == "NONE" or state.entry_price is None:
state.position_pnl_percent = None
state.position_hold_seconds = None
state.position_pressure = None
state.position_health_score = None
state.position_health_status = None
state.position_health_reason = None
state.position_risk_level = None
state.position_risk_reason = None
state.position_trend_alignment = None
state.position_adverse_momentum = False
state.position_exit_pressure = None
return
pnl_percent = self._position_pnl_percent(state)
hold_seconds = self._position_hold_seconds(state)
trend_alignment = self._position_trend_alignment(state)
adverse_momentum = self._has_adverse_position_momentum(state)
pressure = self._position_pressure(
state=state,
pnl_percent=pnl_percent,
)
health_score = self._position_health_score(
state=state,
pnl_percent=pnl_percent,
trend_alignment=trend_alignment,
adverse_momentum=adverse_momentum,
)
risk_level, risk_reason = self._position_risk_level(
state=state,
pnl_percent=pnl_percent,
trend_alignment=trend_alignment,
adverse_momentum=adverse_momentum,
)
state.position_pnl_percent = pnl_percent
state.position_hold_seconds = hold_seconds
state.position_pressure = pressure
state.position_health_score = health_score
state.position_health_status = self._position_health_status(health_score)
state.position_health_reason = self._position_health_reason(
pressure=pressure,
trend_alignment=trend_alignment,
adverse_momentum=adverse_momentum,
)
state.position_risk_level = risk_level
state.position_risk_reason = risk_reason
state.position_trend_alignment = trend_alignment
state.position_adverse_momentum = adverse_momentum
state.position_exit_pressure = self._position_exit_pressure(
state=state,
pnl_percent=pnl_percent,
risk_level=risk_level,
)
def _position_pnl_percent(self, state: AutoTradeState) -> float | None:
entry_price = safe_float(state.entry_price)
size = safe_float(state.position_size)
pnl = safe_float(state.unrealized_pnl_usd)
if entry_price is None or entry_price <= 0:
return None
if size is None or size <= 0:
return None
if pnl is None:
return None
notional = entry_price * size
if notional <= 0:
return None
return round((pnl / notional) * 100, 4)
def _position_hold_seconds(self, state: AutoTradeState) -> int | None:
opened_at = getattr(state, "position_opened_monotonic_at", None)
if opened_at is None:
return None
opened = safe_float(opened_at)
if opened is None:
return None
return max(0, int(time.monotonic() - opened))
def _position_pressure(
self,
*,
state: AutoTradeState,
pnl_percent: float | None,
) -> str:
pnl = safe_float(state.unrealized_pnl_usd) or 0.0
if pnl_percent is None:
if pnl < 0:
return "LOSS"
if pnl > 0:
return "PROFIT"
return "FLAT"
if pnl_percent <= -0.8:
return "HIGH_LOSS"
if pnl_percent <= -0.3:
return "LOSS"
if pnl_percent >= 0.8:
return "STRONG_PROFIT"
if pnl_percent >= 0.3:
return "PROFIT"
return "FLAT"
def _position_trend_alignment(self, state: AutoTradeState) -> str:
side = str(state.position_side or "NONE").upper()
market_state = str(state.market_state or "").upper()
trend = str(state.market_trend or "").upper()
if side == "NONE":
return "NONE"
if side == "LONG":
if market_state == "TREND_UP" or trend == "UP":
return "ALIGNED"
if market_state == "TREND_DOWN" or trend == "DOWN":
return "AGAINST"
if side == "SHORT":
if market_state == "TREND_DOWN" or trend == "DOWN":
return "ALIGNED"
if market_state == "TREND_UP" or trend == "UP":
return "AGAINST"
return "NEUTRAL"
def _has_adverse_position_momentum(self, state: AutoTradeState) -> bool:
side = str(state.position_side or "NONE").upper()
momentum_direction = str(state.momentum_direction or "").upper()
momentum_state = str(state.momentum_state or "").upper()
if side == "LONG":
return (
momentum_direction == "DOWN"
or momentum_state in {"MOMENTUM_DOWN", "BREAKOUT_DOWN"}
)
if side == "SHORT":
return (
momentum_direction == "UP"
or momentum_state in {"MOMENTUM_UP", "BREAKOUT_UP"}
)
return False
def _position_health_score(
self,
*,
state: AutoTradeState,
pnl_percent: float | None,
trend_alignment: str,
adverse_momentum: bool,
) -> int:
score = 100
if pnl_percent is not None:
if pnl_percent <= -1.0:
score -= 35
elif pnl_percent <= -0.5:
score -= 22
elif pnl_percent < 0:
score -= 10
elif pnl_percent >= 0.8:
score += 5
if trend_alignment == "AGAINST":
score -= 25
elif trend_alignment == "NEUTRAL":
score -= 8
if adverse_momentum:
score -= 20
if state.execution_quality == "BLOCKED":
score -= 15
elif state.execution_quality == "WARNING":
score -= 8
if state.market_runtime_degraded:
score -= 10
return max(0, min(100, score))
def _position_health_status(self, score: int | None) -> str:
if score is None:
return "UNKNOWN"
if score >= 80:
return "HEALTHY"
if score >= 55:
return "WATCH"
if score >= 35:
return "PRESSURE"
return "DANGER"
def _position_health_reason(
self,
*,
pressure: str,
trend_alignment: str,
adverse_momentum: bool,
) -> str:
if trend_alignment == "AGAINST" and adverse_momentum:
return "тренд и momentum против позиции"
if trend_alignment == "AGAINST":
return "тренд против позиции"
if adverse_momentum:
return "momentum против позиции"
if pressure in {"HIGH_LOSS", "LOSS"}:
return "позиция под давлением"
if pressure in {"PROFIT", "STRONG_PROFIT"}:
return "позиция в прибыли"
return "позиция стабильна"
def _position_risk_level(
self,
*,
state: AutoTradeState,
pnl_percent: float | None,
trend_alignment: str,
adverse_momentum: bool,
) -> tuple[str, str]:
if state.execution_quality == "BLOCKED":
return "HIGH", "исполнение заблокировано"
if pnl_percent is not None and pnl_percent <= -1.0:
return "HIGH", "сильная просадка позиции"
if trend_alignment == "AGAINST" and adverse_momentum:
return "HIGH", "рынок движется против позиции"
if pnl_percent is not None and pnl_percent < 0:
if trend_alignment == "AGAINST" or adverse_momentum:
return "ELEVATED", "убыток усиливается рыночным контекстом"
return "MODERATE", "позиция в минусе"
if adverse_momentum:
return "MODERATE", "momentum против позиции"
return "LOW", "критичных рисков нет"
def _position_exit_pressure(
self,
*,
state: AutoTradeState,
pnl_percent: float | None,
risk_level: str,
) -> str:
if risk_level == "HIGH":
return "HIGH"
if risk_level == "ELEVATED":
return "WATCH"
if pnl_percent is not None and pnl_percent <= -0.5:
return "WATCH"
return "LOW"
def _sync_position_intelligence_state(self, state: AutoTradeState) -> None:
if state.position_side == "NONE" or state.entry_price is None:
state.position_lifecycle_stage = None
state.position_hold_quality = None
state.position_decay_state = None
state.position_exit_confidence = None
state.position_exit_signal = None
state.position_intelligence_reason = None
state.position_recommended_action = None
state.position_peak_pnl_usd = None
state.position_peak_pnl_percent = None
state.position_mfe_percent = None
state.position_mae_percent = None
state.position_fatigue_score = None
state.position_fatigue_state = None
state.position_giveback_percent = None
state.position_conviction_state = None
state.position_exit_urgency = None
state.position_reversal_risk = None
return
lifecycle_stage = self._position_lifecycle_stage(state)
hold_quality = self._position_hold_quality(state)
decay_state = self._position_decay_state(state)
self._sync_advanced_position_analytics(
state=state,
lifecycle_stage=lifecycle_stage,
hold_quality=hold_quality,
decay_state=decay_state,
)
exit_confidence = self._position_exit_confidence(
state=state,
hold_quality=hold_quality,
decay_state=decay_state,
)
exit_signal = self._position_exit_signal(exit_confidence)
state.position_lifecycle_stage = lifecycle_stage
state.position_hold_quality = hold_quality
state.position_decay_state = decay_state
state.position_exit_confidence = exit_confidence
state.position_exit_signal = exit_signal
state.position_intelligence_reason = self._position_intelligence_reason(
state=state,
hold_quality=hold_quality,
decay_state=decay_state,
exit_signal=exit_signal,
)
state.position_recommended_action = self._position_recommended_action(
exit_signal
)
def _position_lifecycle_stage(self, state: AutoTradeState) -> str:
hold_seconds = state.position_hold_seconds
if hold_seconds is None:
return "UNKNOWN"
if hold_seconds < 60:
return "NEW"
if hold_seconds < 300:
return "ACTIVE"
if hold_seconds < 900:
return "MATURE"
return "AGED"
def _position_hold_quality(self, state: AutoTradeState) -> str:
health_status = str(state.position_health_status or "").upper()
pressure = str(state.position_pressure or "").upper()
trend_alignment = str(state.position_trend_alignment or "").upper()
if health_status == "DANGER":
return "BAD"
if pressure == "HIGH_LOSS":
return "BAD"
if trend_alignment == "AGAINST" and state.position_adverse_momentum:
return "BAD"
if health_status == "PRESSURE":
return "WEAK"
if pressure == "LOSS":
return "WEAK"
if pressure in {"PROFIT", "STRONG_PROFIT"} and trend_alignment == "ALIGNED":
return "GOOD"
if health_status == "HEALTHY":
return "GOOD"
return "NEUTRAL"
def _position_decay_state(self, state: AutoTradeState) -> str:
pressure = str(state.position_pressure or "").upper()
trend_alignment = str(state.position_trend_alignment or "").upper()
lifecycle = str(state.position_lifecycle_stage or "").upper()
if pressure in {"HIGH_LOSS", "LOSS"} and state.position_adverse_momentum:
return "ACCELERATING_LOSS"
if trend_alignment == "AGAINST" and state.position_adverse_momentum:
return "CONTEXT_DECAY"
if pressure == "PROFIT" and state.position_adverse_momentum:
return "PROFIT_DECAY"
if lifecycle == "AGED" and pressure == "FLAT":
return "TIME_DECAY"
return "NONE"
def _position_exit_confidence(
self,
*,
state: AutoTradeState,
hold_quality: str,
decay_state: str,
) -> float:
score = 0.0
risk_level = str(state.position_risk_level or "").upper()
exit_pressure = str(state.position_exit_pressure or "").upper()
if risk_level == "HIGH":
score += 0.45
elif risk_level == "ELEVATED":
score += 0.30
elif risk_level == "MODERATE":
score += 0.15
if exit_pressure == "HIGH":
score += 0.30
elif exit_pressure == "WATCH":
score += 0.15
if hold_quality == "BAD":
score += 0.25
elif hold_quality == "WEAK":
score += 0.15
if decay_state in {"ACCELERATING_LOSS", "CONTEXT_DECAY"}:
score += 0.25
elif decay_state in {"PROFIT_DECAY", "TIME_DECAY"}:
score += 0.15
if state.execution_quality == "BLOCKED":
score += 0.10
return round(max(0.0, min(1.0, score)), 3)
def _position_exit_signal(self, exit_confidence: float | None) -> str:
if exit_confidence is None:
return "NONE"
if exit_confidence >= 0.75:
return "EXIT"
if exit_confidence >= 0.50:
return "REDUCE_OR_PROTECT"
if exit_confidence >= 0.30:
return "WATCH"
return "HOLD"
def _position_intelligence_reason(
self,
*,
state: AutoTradeState,
hold_quality: str,
decay_state: str,
exit_signal: str,
) -> str:
if exit_signal == "EXIT":
return "позиция требует выхода"
if exit_signal == "REDUCE_OR_PROTECT":
return "позицию нужно защитить или уменьшить"
if decay_state != "NONE":
return "качество удержания ухудшается"
if hold_quality == "GOOD":
return "позицию можно удерживать"
if hold_quality == "WEAK":
return "позиция требует наблюдения"
return "критичных признаков выхода нет"
def _position_recommended_action(self, exit_signal: str | None) -> str:
if exit_signal == "EXIT":
return "CLOSE"
if exit_signal == "REDUCE_OR_PROTECT":
return "PROTECT"
if exit_signal == "WATCH":
return "WATCH"
return "HOLD"
def _sync_advanced_position_analytics(
self,
*,
state: AutoTradeState,
lifecycle_stage: str,
hold_quality: str,
decay_state: str,
) -> None:
pnl = safe_float(state.unrealized_pnl_usd)
pnl_percent = safe_float(state.position_pnl_percent)
peak_pnl = safe_float(state.position_peak_pnl_usd)
peak_pnl_percent = safe_float(state.position_peak_pnl_percent)
if pnl is not None:
if peak_pnl is None or pnl > peak_pnl:
state.position_peak_pnl_usd = pnl
if pnl_percent is not None:
if peak_pnl_percent is None or pnl_percent > peak_pnl_percent:
state.position_peak_pnl_percent = pnl_percent
state.position_mfe_percent = self._position_mfe_percent(state)
state.position_mae_percent = self._position_mae_percent(state)
state.position_giveback_percent = self._position_giveback_percent(state)
fatigue_score = self._position_fatigue_score(
state=state,
lifecycle_stage=lifecycle_stage,
hold_quality=hold_quality,
decay_state=decay_state,
)
state.position_fatigue_score = fatigue_score
state.position_fatigue_state = self._position_fatigue_state(fatigue_score)
state.position_conviction_state = self._position_conviction_state(state)
state.position_exit_urgency = self._position_exit_urgency(state)
state.position_reversal_risk = self._position_reversal_risk(state)
def _position_mfe_percent(self, state: AutoTradeState) -> float | None:
peak = safe_float(state.position_peak_pnl_percent)
if peak is None:
return None
return round(max(0.0, peak), 4)
def _position_mae_percent(self, state: AutoTradeState) -> float | None:
current = safe_float(state.position_pnl_percent)
if current is None:
return None
return round(min(0.0, current), 4)
def _position_giveback_percent(self, state: AutoTradeState) -> float | None:
peak = safe_float(state.position_peak_pnl_percent)
current = safe_float(state.position_pnl_percent)
if peak is None or current is None:
return None
if peak <= 0:
return 0.0
giveback = peak - current
if giveback <= 0:
return 0.0
return round((giveback / peak) * 100, 2)
def _position_fatigue_score(
self,
*,
state: AutoTradeState,
lifecycle_stage: str,
hold_quality: str,
decay_state: str,
) -> float:
score = 0.0
giveback = safe_float(state.position_giveback_percent) or 0.0
hold_seconds = safe_float(state.position_hold_seconds) or 0.0
if lifecycle_stage == "AGED":
score += 0.25
elif lifecycle_stage == "MATURE":
score += 0.15
if hold_quality == "BAD":
score += 0.30
elif hold_quality == "WEAK":
score += 0.18
if decay_state in {"ACCELERATING_LOSS", "CONTEXT_DECAY"}:
score += 0.30
elif decay_state in {"PROFIT_DECAY", "TIME_DECAY"}:
score += 0.18
if giveback >= 70:
score += 0.30
elif giveback >= 45:
score += 0.20
elif giveback >= 25:
score += 0.10
if hold_seconds >= 1800:
score += 0.15
elif hold_seconds >= 900:
score += 0.08
if state.position_adverse_momentum:
score += 0.15
return round(max(0.0, min(1.0, score)), 3)
def _position_fatigue_state(self, score: float | None) -> str:
value = safe_float(score)
if value is None:
return "UNKNOWN"
if value >= 0.75:
return "EXHAUSTED"
if value >= 0.50:
return "TIRED"
if value >= 0.25:
return "WATCH"
return "FRESH"
def _position_conviction_state(self, state: AutoTradeState) -> str:
health = str(state.position_health_status or "").upper()
fatigue = str(state.position_fatigue_state or "").upper()
alignment = str(state.position_trend_alignment or "").upper()
if health == "DANGER" or fatigue == "EXHAUSTED":
return "BROKEN"
if alignment == "AGAINST" or fatigue == "TIRED":
return "WEAKENING"
if health == "HEALTHY" and alignment == "ALIGNED":
return "STRONG"
return "NEUTRAL"
def _position_exit_urgency(self, state: AutoTradeState) -> str:
exit_signal = str(state.position_exit_signal or "").upper()
fatigue = str(state.position_fatigue_state or "").upper()
risk = str(state.position_risk_level or "").upper()
if exit_signal == "EXIT" or risk == "HIGH":
return "IMMEDIATE"
if fatigue == "EXHAUSTED":
return "HIGH"
if exit_signal == "REDUCE_OR_PROTECT" or fatigue == "TIRED":
return "MEDIUM"
if exit_signal == "WATCH":
return "LOW"
return "NONE"
def _position_reversal_risk(self, state: AutoTradeState) -> str:
giveback = safe_float(state.position_giveback_percent) or 0.0
fatigue = str(state.position_fatigue_state or "").upper()
adverse = bool(state.position_adverse_momentum)
if adverse and giveback >= 45:
return "HIGH"
if fatigue in {"TIRED", "EXHAUSTED"} and giveback >= 25:
return "ELEVATED"
if adverse:
return "MODERATE"
return "LOW"
def _sync_autonomous_trade_management(
self,
state: AutoTradeState,
) -> None:
if state.position_side == "NONE":
state.autonomous_action = None
state.autonomous_action_reason = None
state.autonomous_action_confidence = None
state.autonomous_protection_required = False
state.autonomous_reduce_required = False
state.autonomous_exit_required = False
return
exit_signal = str(state.position_exit_signal or "HOLD").upper()
exit_confidence = safe_float(state.position_exit_confidence) or 0.0
action = "HOLD"
reason = "позиция удерживается"
protect_required = False
reduce_required = False
exit_required = False
if exit_signal == "WATCH":
action = "WATCH"
reason = "позиция требует наблюдения"
elif exit_signal == "REDUCE_OR_PROTECT":
if state.position_pressure in {"HIGH_LOSS", "LOSS"}:
action = "REDUCE"
reduce_required = True
reason = "позиция должна быть уменьшена"
else:
action = "PROTECT"
protect_required = True
reason = "позиция требует защиты"
elif exit_signal == "EXIT":
action = "EXIT"
exit_required = True
reason = "позиция требует закрытия"
if (
state.position_adverse_momentum
and state.position_trend_alignment == "AGAINST"
and exit_confidence >= 0.65
):
action = "EXIT"
exit_required = True
reason = "рынок агрессивно движется против позиции"
state.autonomous_action = action
state.autonomous_action_reason = reason
state.autonomous_action_confidence = exit_confidence
state.autonomous_protection_required = protect_required
state.autonomous_reduce_required = reduce_required
state.autonomous_exit_required = exit_required
def _log_execution_quality_if_changed(
self,
*,
@@ -1831,6 +2646,13 @@ class AutoTradeService:
if state.execution_quality != "BLOCKED":
ExecutionEngine().process(state)
self._sync_position_health_state(state)
self._sync_position_intelligence_state(state)
self._sync_autonomous_trade_management(state)
if state.execution_quality != "BLOCKED":
ExecutionEngine().process_runtime_action(state)
self._sync_execution_semantic_state(state)
return state

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

@@ -61,9 +61,88 @@ class AutoTradeState:
# размер позиции
position_size: float | None = None
# monotonic timestamp открытия текущей позиции
position_opened_monotonic_at: float | None = None
# нереализованный PnL
unrealized_pnl_usd: float | None = None
# position health / runtime risk
position_pnl_percent: float | None = None
position_hold_seconds: int | None = None
position_pressure: str | None = None
position_health_score: int | None = None
position_health_status: str | None = None
position_health_reason: str | None = None
position_risk_level: str | None = None
position_risk_reason: str | None = None
position_trend_alignment: str | None = None
position_adverse_momentum: bool = False
position_exit_pressure: str | None = None
# position intelligence layer
position_lifecycle_stage: str | None = None
position_hold_quality: str | None = None
position_decay_state: str | None = None
position_exit_confidence: float | None = None
position_exit_signal: str | None = None
position_intelligence_reason: str | None = None
position_recommended_action: str | None = None
# advanced lifecycle analytics
position_peak_pnl_usd: float | None = None
position_peak_pnl_percent: float | None = None
position_mfe_percent: float | None = None
position_mae_percent: float | None = None
position_fatigue_score: float | None = None
position_fatigue_state: str | None = None
position_giveback_percent: float | None = None
# position behavioral state
position_conviction_state: str | None = None
position_exit_urgency: str | None = None
position_reversal_risk: str | None = None
# autonomous trade management
autonomous_action: str | None = None
autonomous_action_reason: str | None = None
autonomous_action_confidence: float | None = None
autonomous_protection_required: bool = False
autonomous_reduce_required: bool = False
autonomous_exit_required: bool = False
# runtime position action execution
autonomous_last_action: str | None = None
autonomous_last_action_reason: str | None = None
autonomous_last_action_at: float | None = None
# loss cooldown tracking
last_loss_monotonic_at: float | None = None
# runtime protection engine
position_protection_status: str | None = None
position_protection_reason: str | None = None
# break-even protection
break_even_armed: bool = False
break_even_price: float | None = None
# trailing protection
trailing_stop_active: bool = False
trailing_stop_price: float | None = None
# locked profit protection
profit_lock_active: bool = False
profit_lock_price: float | None = None
# runtime protection execution
runtime_protection_action: str | None = None
runtime_protection_reason: str | None = None
runtime_protection_updated_at: float | None = None
# максимальная просадка
max_drawdown_usd: float | None = None
@@ -235,6 +314,25 @@ class AutoTradeState:
# человекочитаемое объяснение качества исполнения
execution_quality_message: str | None = None
# источник execution pricing snapshot
execution_price_source: str | None = None
# возраст execution snapshot
execution_price_age_seconds: float | None = None
# bid execution price
execution_bid_price: float | None = None
# ask execution price
execution_ask_price: float | None = None
# last execution price
execution_last_price: float | None = None
# pricing freshness status:
# FRESH / AGING / STALE / UNKNOWN
execution_price_freshness: str | None = None
# признак деградации runtime market data
market_runtime_degraded: bool = False

108
app/src/trading/diagnostics/formatter.py
View File

@@ -56,6 +56,9 @@ class SemanticDiagnosticFormatter:
if mode != "COMPACT":
if has_position:
sections.append(self._position_block(position))
position_health_block = self._position_health_block(position)
if position_health_block:
sections.append(position_health_block)
if self._has_adaptive_size(adaptive):
sections.append(self._adaptive_block(adaptive))
@@ -1035,6 +1038,111 @@ class SemanticDiagnosticFormatter:
return "\n".join(lines).strip()
def _position_health_block(self, data: JsonDict) -> str:
health_state = str(data.get("health_state") or "")
health_score = safe_float(data.get("health_score"))
health_message = str(data.get("health_message") or "").strip()
pressure_state = str(data.get("pressure_state") or "")
trend_alignment = str(data.get("trend_alignment") or "")
adverse_momentum = bool(data.get("adverse_momentum"))
risk_used = safe_float(data.get("risk_used_percent"))
price_move = safe_float(data.get("price_move_percent"))
opened_age = data.get("opened_age_seconds")
if not health_state or health_state in {"NONE", "UNKNOWN"}:
return ""
lines = [
(
f"{self._position_health_icon(health_state)} "
f"Здоровье позиции · {self._position_health_title(health_state)}"
),
]
if health_score is not None:
lines.append(f"• Score: {health_score:.0f}/100")
if price_move is not None:
lines.append(f"• Движение цены: {price_move:+.3f}%")
if risk_used is not None and risk_used > 0:
lines.append(f"• Использовано риска: {risk_used:.1f}%")
alignment_line = self._position_alignment_line(trend_alignment)
if alignment_line:
lines.append(alignment_line)
pressure_line = self._position_pressure_line(pressure_state)
if pressure_line:
lines.append(pressure_line)
if adverse_momentum:
lines.append("• Импульс: против позиции")
hold_time = self._duration(opened_age)
if hold_time != "—":
lines.append(f"• В позиции: {hold_time}")
if health_message:
lines.append(health_message)
return "\n".join(lines)
def _position_health_icon(self, value: object) -> str:
text = str(value or "")
if text == "HEALTHY":
return "🟢"
if text == "WATCH":
return "🟡"
if text == "PRESSURE":
return "🟠"
if text == "DANGER":
return "🔴"
return "⚪️"
def _position_health_title(self, value: object) -> str:
text = str(value or "")
mapping = {
"HEALTHY": "устойчива",
"WATCH": "под наблюдением",
"PRESSURE": "под давлением",
"DANGER": "высокий риск",
}
return mapping.get(text, "неизвестно")
def _position_alignment_line(self, value: object) -> str:
text = str(value or "")
if text == "ALIGNED":
return "• Рынок: по позиции"
if text == "AGAINST":
return "• Рынок: против позиции"
if text == "NEUTRAL":
return "• Рынок: нейтрально"
return ""
def _position_pressure_line(self, value: object) -> str:
text = str(value or "")
mapping = {
"PROFIT": "• Давление: нет",
"PROFIT_UNDER_PRESSURE": "• Давление: прибыль под риском",
"LOSS": "• Давление: умеренное",
"PRESSURE": "• Давление: повышенное",
"DANGER": "• Давление: критическое",
}
return mapping.get(text, "")
def _position_risk_line(self, data: JsonDict) -> str:
items: list[str] = []

302
app/src/trading/diagnostics/snapshot.py
View File

@@ -33,6 +33,11 @@ class SemanticDiagnosticSnapshotBuilder:
position_current_price = self._position_current_price(state)
position_health = self._position_health(
state=state,
current_price=position_current_price,
)
return {
"status": {
"status": state.status,
@@ -142,6 +147,18 @@ class SemanticDiagnosticSnapshotBuilder:
"stop_loss_usd": state.effective_target_risk_usd,
"take_profit_usd": self._take_profit_usd(state),
"max_loss_usd": state.max_loss_usd,
"opened_age_seconds": self._age_seconds(
now=now,
started_at=state.position_opened_monotonic_at,
),
"price_move_percent": position_health.get("price_move_percent"),
"risk_used_percent": position_health.get("risk_used_percent"),
"health_state": position_health.get("health_state"),
"health_score": position_health.get("health_score"),
"health_message": position_health.get("health_message"),
"pressure_state": position_health.get("pressure_state"),
"trend_alignment": position_health.get("trend_alignment"),
"adverse_momentum": position_health.get("adverse_momentum"),
},
"runtime_health": {
"health_score": health_score,
@@ -153,6 +170,8 @@ class SemanticDiagnosticSnapshotBuilder:
"runtime_expired_message": state.runtime_expired_message,
"has_market_data": state.market_state is not None,
"has_momentum_data": getattr(state, "momentum_state", None) is not None,
"position_health_state": position_health.get("health_state"),
"position_health_score": position_health.get("health_score"),
},
"summary": {
"health_score": health_score,
@@ -177,6 +196,7 @@ class SemanticDiagnosticSnapshotBuilder:
"momentum": getattr(state, "momentum_state", None),
"execution": state.execution_semantic_status,
"position": state.position_side,
"position_health": position_health.get("health_state"),
"is_ready": state.is_signal_ready,
"is_blocked": bool(blockers),
"blockers": blockers,
@@ -483,6 +503,288 @@ class SemanticDiagnosticSnapshotBuilder:
return result
def _position_health(
self,
*,
state: AutoTradeState,
current_price: float | None,
) -> dict[str, Any]:
if state.position_side == "NONE":
return {
"health_state": "NONE",
"health_score": None,
"health_message": None,
"price_move_percent": None,
"risk_used_percent": None,
"pressure_state": "NONE",
"trend_alignment": "NONE",
"adverse_momentum": False,
}
entry_price = safe_float(state.entry_price)
pnl = safe_float(state.unrealized_pnl_usd)
stop_loss_usd = safe_float(state.effective_target_risk_usd)
max_loss_usd = safe_float(state.max_loss_usd)
price_move_percent = self._position_price_move_percent(
side=state.position_side,
entry_price=entry_price,
current_price=current_price,
)
risk_used_percent = self._position_risk_used_percent(
pnl=pnl,
stop_loss_usd=stop_loss_usd,
max_loss_usd=max_loss_usd,
)
trend_alignment = self._position_trend_alignment(state)
adverse_momentum = self._has_adverse_momentum(state)
pressure_state = self._position_pressure_state(
pnl=pnl,
risk_used_percent=risk_used_percent,
adverse_momentum=adverse_momentum,
)
opened_age_seconds = self._age_seconds(
now=time.monotonic(),
started_at=state.position_opened_monotonic_at,
)
health_score = self._position_health_score(
pnl=pnl,
risk_used_percent=risk_used_percent,
trend_alignment=trend_alignment,
adverse_momentum=adverse_momentum,
pressure_state=pressure_state,
opened_age_seconds=opened_age_seconds,
)
health_state = self._position_health_state(health_score)
health_message = self._position_health_message(
health_state=health_state,
pressure_state=pressure_state,
trend_alignment=trend_alignment,
adverse_momentum=adverse_momentum,
)
return {
"health_state": health_state,
"health_score": health_score,
"health_message": health_message,
"price_move_percent": price_move_percent,
"risk_used_percent": risk_used_percent,
"pressure_state": pressure_state,
"trend_alignment": trend_alignment,
"adverse_momentum": adverse_momentum,
}
def _position_price_move_percent(
self,
*,
side: str | None,
entry_price: float | None,
current_price: float | None,
) -> float | None:
if entry_price is None or current_price is None:
return None
if entry_price <= 0 or current_price <= 0:
return None
normalized_side = str(side or "").upper()
if normalized_side == "LONG":
return round(((current_price - entry_price) / entry_price) * 100, 4)
if normalized_side == "SHORT":
return round(((entry_price - current_price) / entry_price) * 100, 4)
return None
def _position_risk_used_percent(
self,
*,
pnl: float | None,
stop_loss_usd: float | None,
max_loss_usd: float | None,
) -> float | None:
if pnl is None or pnl >= 0:
return 0.0
candidates = [
value
for value in [stop_loss_usd, max_loss_usd]
if value is not None and value > 0
]
if not candidates:
return None
risk_base = min(candidates)
if risk_base is None or risk_base <= 0:
return None
return round(min(999.0, (abs(pnl) / abs(risk_base)) * 100), 1)
def _position_trend_alignment(self, state: AutoTradeState) -> str:
side = str(state.position_side or "").upper()
market_state = str(state.market_state or "").upper()
trend = str(state.market_trend or "").upper()
if side == "LONG":
if market_state == "TREND_UP" or trend == "UP":
return "ALIGNED"
if market_state == "TREND_DOWN" or trend == "DOWN":
return "AGAINST"
if side == "SHORT":
if market_state == "TREND_DOWN" or trend == "DOWN":
return "ALIGNED"
if market_state == "TREND_UP" or trend == "UP":
return "AGAINST"
return "NEUTRAL"
def _has_adverse_momentum(self, state: AutoTradeState) -> bool:
side = str(state.position_side or "").upper()
momentum_direction = str(state.momentum_direction or "").upper()
momentum_state = str(state.momentum_state or "").upper()
if side == "LONG":
return (
momentum_direction == "DOWN"
or momentum_state in {"MOMENTUM_DOWN", "BREAKOUT_DOWN"}
)
if side == "SHORT":
return (
momentum_direction == "UP"
or momentum_state in {"MOMENTUM_UP", "BREAKOUT_UP"}
)
return False
def _position_pressure_state(
self,
*,
pnl: float | None,
risk_used_percent: float | None,
adverse_momentum: bool,
) -> str:
if pnl is None:
return "UNKNOWN"
if pnl >= 0:
if adverse_momentum:
return "PROFIT_UNDER_PRESSURE"
return "PROFIT"
if risk_used_percent is None:
return "LOSS"
if risk_used_percent >= 80:
return "DANGER"
if risk_used_percent >= 50:
return "PRESSURE"
return "LOSS"
def _position_health_score(
self,
*,
pnl: float | None,
risk_used_percent: float | None,
trend_alignment: str,
adverse_momentum: bool,
pressure_state: str,
opened_age_seconds: int | None,
) -> int:
score = 100
if pnl is not None:
if pnl < 0:
score -= 20
elif pnl > 0:
score += 5
if risk_used_percent is not None:
if risk_used_percent >= 80:
score -= 45
elif risk_used_percent >= 50:
score -= 30
elif risk_used_percent >= 25:
score -= 15
if trend_alignment == "AGAINST":
score -= 25
elif trend_alignment == "ALIGNED":
score += 8
if adverse_momentum:
score -= 20
if pressure_state == "DANGER":
score -= 20
elif pressure_state == "PRESSURE":
score -= 10
if opened_age_seconds is not None:
if opened_age_seconds >= 7200:
score -= 20
elif opened_age_seconds >= 3600:
score -= 10
return max(0, min(100, score))
def _position_health_state(self, score: int | None) -> str:
if score is None:
return "UNKNOWN"
if score >= 75:
return "HEALTHY"
if score >= 50:
return "WATCH"
if score >= 30:
return "PRESSURE"
return "DANGER"
def _position_health_message(
self,
*,
health_state: str,
pressure_state: str,
trend_alignment: str,
adverse_momentum: bool,
) -> str:
if health_state == "HEALTHY":
return "Позиция выглядит устойчиво."
if health_state == "WATCH":
return "Позиция требует наблюдения."
if health_state == "PRESSURE":
if trend_alignment == "AGAINST":
return "Позиция под давлением: рынок против направления."
if adverse_momentum:
return "Позиция под давлением: импульс против позиции."
return "Позиция под давлением."
if health_state == "DANGER":
return "Высокий риск по открытой позиции."
return "Состояние позиции не определено."
def _take_profit_usd(self, state: AutoTradeState) -> float | None:
take_profit_percent = safe_float(state.take_profit_percent)
position_size = safe_float(state.position_size)

1288
app/src/trading/execution/engine.py
View File

File diff suppressed because it is too large Load Diff

19
app/src/trading/position/state.py
View File

@@ -25,8 +25,27 @@ class PositionState:
# нереализованный PnL
unrealized_pnl_usd: float | None = None
# peak pnl tracking
peak_unrealized_pnl_usd: float | None = None
peak_pnl_percent: float | None = None
# lifecycle tracking
max_favorable_excursion_percent: float | None = None
max_adverse_excursion_percent: float | None = None
# runtime fatigue tracking
fatigue_score: float | None = None
fatigue_state: str | None = None
# position regime memory
best_price_seen: float | None = None
worst_price_seen: float | None = None
# время открытия позиции
opened_at: str | None = None
# monotonic timestamp открытия позиции
opened_monotonic_at: float | None = None
# время последнего обновления позиции
updated_at: str | None = None