07.4.4.1.9.6 Adaptive Position Sizing + 07.4.4.1.9.6.1 Market Semantic Layer for Adaptive Sizing

This commit is contained in:
2026-05-12 20:25:10 +03:00
parent 1aa8f6c407
commit 8b83055e6a
12 changed files with 1298 additions and 29 deletions

View File

@@ -4,6 +4,7 @@ from __future__ import annotations
import math
import time
import re
from aiogram.types import InlineKeyboardMarkup
from aiogram.utils.keyboard import InlineKeyboardBuilder
@@ -62,6 +63,16 @@ def build_auto_text() -> str:
return _build_waiting_text(state)
def build_auto_semantic_text() -> str:
text = build_auto_text()
return re.sub(
r" · \d+с| · \d+м \d+с| · \d+ч \d+м",
"",
text,
)
def _build_not_configured_text(state) -> str:
symbol_ready = state.symbol is not None
strategy_ready = state.strategy is not None
@@ -124,7 +135,7 @@ def _build_stopped_without_position_text(state) -> str:
f"Цена · {_format_plain_or_dash(price)}",
_estimated_size_text(state, price),
_max_reserved_line(state, price),
f"Риск · {_format_money_compact(_target_risk_usd(state))}",
_effective_risk_line(state),
]
if rr_line or risk_line:
@@ -171,6 +182,30 @@ def _execution_confidence_line(state) -> str:
return f"🧠 Уверенность входа · {percent}% низкая"
def _adaptive_size_line(state) -> str:
multiplier = getattr(state, "adaptive_size_multiplier", None)
if multiplier is None:
return ""
percent = int(round(float(multiplier) * 100))
reason = getattr(state, "adaptive_size_reason", None)
if multiplier <= 0:
return "🧮 Размер · вход заблокирован"
if multiplier < 1:
return f"🧮 Размер · {percent}% адаптивно уменьшен"
if multiplier > 1:
return f"🧮 Размер · {percent}% адаптивно увеличен"
if multiplier == 1:
return "🧮 Размер · без корректировки"
return ""
def _build_waiting_text(state) -> str:
price = _signal_entry_price(state)
@@ -206,15 +241,19 @@ def _build_waiting_text(state) -> str:
if signal_lines:
parts.extend(["", *signal_lines])
parts.extend([
"",
order_lines = [
"Подготовка ордера 🧾",
_order_header_line(state),
f"{_price_label_for_signal(state)} · {_format_plain_or_dash(price)}",
_estimated_size_text(state, price),
_adaptive_size_line(state),
_max_reserved_line(state, price),
f"Риск · {_format_money_compact(_target_risk_usd(state))}",
])
_effective_risk_line(state),
]
order_lines = [line for line in order_lines if line]
parts.extend(["", *order_lines])
if rr_line or risk_line:
parts.append("")
@@ -271,6 +310,7 @@ def _build_active_position_text(state) -> str:
"",
f"Размер · {_format_crypto_size(size)}",
f"Позиция · {_format_money_compact(notional)}",
_adaptive_size_line(state),
f"Вход · {_format_plain_or_dash(state.entry_price)}",
f"Цена · {_format_plain_or_dash(price_for_calc)}",
"",
@@ -487,6 +527,15 @@ def _target_risk_usd(state) -> float:
return _allocated_balance(state) * (state.risk_percent / 100)
def _effective_risk_line(state) -> str:
effective_risk_usd = getattr(state, "effective_target_risk_usd", None)
if effective_risk_usd is not None:
return f"Риск · {_format_money_compact(effective_risk_usd)}"
return f"Риск · {_format_money_compact(_target_risk_usd(state))}"
def _estimated_size(state, price: float | None) -> float | None:
if (
price is None
@@ -499,16 +548,24 @@ def _estimated_size(state, price: float | None) -> float | None:
return None
stop_loss_distance_usd = price * (state.stop_loss_percent / 100)
if stop_loss_distance_usd <= 0:
return None
risk_size = _target_risk_usd(state) / stop_loss_distance_usd
multiplier = getattr(state, "adaptive_size_multiplier", None)
if multiplier is not None:
risk_size *= float(multiplier)
max_percent = getattr(state, "max_reserved_balance_percent", None)
if max_percent is None or max_percent <= 0:
return _round_size(risk_size)
leverage = state.leverage or 1.0
if leverage <= 0:
return _round_size(risk_size)