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

07.4.4.1.10.3 — Telegram Diagnostic Screen

Browse Source
This commit is contained in:
Sergey
2026-05-16 09:23:37 +03:00
parent 8e1c09ad66
commit 2c75f95b46
16 changed files with 2902 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files

241
app/src/notifications/templates/execution.py
View File

@@ -16,6 +16,9 @@ def build_execution_notification(event: RuntimeEvent) -> NotificationMessage | N
if event.event_type == RuntimeEventType.POSITION_FLIPPED:
return _build_position_flipped(event)
if event.event_type == RuntimeEventType.POSITION_FLIP_BLOCKED:
return _build_flip_blocked(event)
return None
@@ -24,23 +27,41 @@ def _build_position_opened(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
side = str(payload.get("side") or "—")
strategy = str(payload.get("strategy") or "—")
side = str(payload.get("side") or "—").upper()
leverage = _format_leverage(payload.get("leverage"))
entry_price = _format_price(payload.get("entry_price"))
size = _format_size(payload.get("size"))
confidence = float(payload.get("confidence") or 0.0)
priority = _alert_priority(
confidence=confidence,
repeat_count=int(payload.get("repeat_count") or 0),
)
semantic_lines = payload.get("semantic_lines") or []
side_icon = "🟢" if side == "LONG" else "🔴"
text = (
f"<b>📄 Paper position opened {side_icon} {side}</b>\n\n"
f"{symbol} · {leverage}\n"
f"Entry: $ {entry_price}\n"
f"Size: {size}"
)
lines = [
"<b>🧾 Позиция открыта</b>",
"",
f"{side_icon} {symbol} · {strategy} · {side} {leverage}",
f"Вход: {entry_price}",
f"Размер: {size}",
f"Объём: {_format_notional(entry_price=payload.get('entry_price'), size=payload.get('size'))}",
"",
f"{_strength_bar(priority)} Сигнал {_strength_label(priority).lower()} · {confidence:.2f}",
]
if semantic_lines:
lines.extend(
str(line).strip().rstrip(".")
for line in semantic_lines
if str(line).strip()
)
return NotificationMessage(
title=event.title,
text=text,
text="\n".join(lines),
priority=event.priority,
dedupe_key=event.dedupe_key,
)
@@ -50,63 +71,162 @@ def _build_position_closed(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
side = str(payload.get("side") or "—")
side = str(payload.get("side") or "—").upper()
leverage = _format_leverage(payload.get("leverage"))
entry_price = _format_price(payload.get("entry_price"))
exit_price = _format_price(payload.get("exit_price"))
size = _format_size(payload.get("size"))
pnl = _format_pnl(payload.get("pnl"))
risk_reason = payload.get("risk_reason")
risk_line = f"\nRisk: {risk_reason}" if risk_reason else ""
pnl_value = float(payload.get("pnl") or 0.0)
pnl_text = _format_pnl_amount(pnl_value)
text = (
f"<b>✅ Paper position closed</b>\n\n"
f"{side} · {symbol} · {leverage}\n"
f"Entry: $ {entry_price}\n"
f"Exit: $ {exit_price}\n"
f"Size: {size}\n\n"
f"PnL: {pnl}"
f"{risk_line}"
)
risk_reason = _human_close_reason(payload.get("risk_reason"))
pnl_icon = "🟢" if pnl_value >= 0 else "🔴"
pnl_label = "Прибыль" if pnl_value >= 0 else "Убыток"
lines = [
"<b>🧾 Сделка закрыта</b>",
f"{pnl_icon} {pnl_label} · {pnl_text}",
"",
f"{symbol} · {side} {leverage}",
f"Вход: $ {entry_price}",
f"Выход: $ {exit_price}",
f"Размер: {size}",
]
if risk_reason:
lines.extend([
"",
f"Закрытие по {risk_reason}",
])
return NotificationMessage(
title=event.title,
text=text,
text="\n".join(lines),
priority=event.priority,
dedupe_key=event.dedupe_key,
)
def _format_pnl_amount(value: float) -> str:
amount = f"$ {abs(value):,.2f}".replace(",", " ").rstrip("0").rstrip(".")
if value > 0:
return f"+{amount}"
if value < 0:
return f"−{amount}"
return "$ 0"
def _human_close_reason(value: object) -> str:
mapping = {
"STOP_LOSS": "Stop Loss",
"TAKE_PROFIT": "Take Profit",
"MAX_LOSS": "Max Loss",
}
return mapping.get(str(value or ""), "")
def _build_position_flipped(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
leverage = _format_leverage(payload.get("leverage"))
strategy = str(payload.get("strategy") or "—").title()
old_side = str(payload.get("old_side") or "—")
new_side = str(payload.get("new_side") or payload.get("side") or "—")
old_side = str(payload.get("old_side") or "—").upper()
new_side = str(payload.get("new_side") or payload.get("side") or "—").upper()
old_leverage = _format_leverage(
payload.get("old_leverage")
if payload.get("old_leverage") is not None
else payload.get("leverage")
)
new_leverage = _format_leverage(payload.get("leverage"))
entry_price = _format_price(payload.get("entry_price"))
exit_price = _format_price(payload.get("exit_price"))
new_entry_price = _format_price(payload.get("new_entry_price"))
old_size = _format_size(payload.get("old_size"))
new_size = _format_size(payload.get("new_size"))
pnl = _format_pnl(payload.get("pnl"))
pnl_value = float(payload.get("pnl") or 0.0)
pnl_text = _format_pnl_amount(pnl_value)
pnl_icon = "🟢" if pnl_value >= 0 else "🔴"
pnl_label = "Прибыль" if pnl_value >= 0 else "Убыток"
old_icon = "🟢" if old_side == "LONG" else "🔴"
new_icon = "🟢" if new_side == "LONG" else "🔴"
confidence = float(payload.get("confidence") or 0.0)
repeat_count = int(payload.get("repeat_count") or 0)
priority = _alert_priority(
confidence=confidence,
repeat_count=repeat_count,
)
semantic_lines = payload.get("semantic_lines") or []
lines = [
"<b>🧾 Сделка развернута</b>",
f"{pnl_label} {pnl_icon} {pnl_text}",
f"{symbol} · {strategy} {old_icon} {old_side} → {new_icon} {new_side}",
"",
f"Закрыта {old_side} {old_leverage}",
f"Вход: $ {entry_price}",
f"Выход: $ {exit_price}",
f"Размер: {old_size}",
"",
f"Открыта {new_side} {new_leverage}",
f"Вход: $ {new_entry_price}",
f"Размер: {new_size}",
(
"Объём: "
f"{_format_notional(entry_price=payload.get('new_entry_price'), size=payload.get('new_size'))}"
),
"",
f"{_strength_bar(priority)} Сигнал {_strength_label(priority).lower()} · {confidence:.2f}",
]
if semantic_lines:
lines.extend(
str(line).strip().rstrip(".")
for line in semantic_lines
if str(line).strip()
)
return NotificationMessage(
title=event.title,
text="\n".join(lines),
priority=event.priority,
dedupe_key=event.dedupe_key,
)
def _build_flip_blocked(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
signal = str(payload.get("signal") or "").upper()
confidence = float(payload.get("confidence") or 0.0)
reason = str(payload.get("reason") or "Flip заблокирован")
position_side = str(payload.get("position_side") or "—").upper()
target_side = "LONG" if signal == "BUY" else "SHORT" if signal == "SELL" else "—"
icon = "🟢" if target_side == "LONG" else "🔴" if target_side == "SHORT" else ""
text = (
f"<b>🔁 Paper position flipped {old_icon} {old_side} → "
f"{new_icon} {new_side}</b>\n\n"
f"{symbol} · {leverage}\n\n"
f"Old entry: $ {entry_price}\n"
f"Exit: $ {exit_price}\n"
f"Old size: {old_size}\n\n"
f"New entry: $ {new_entry_price}\n"
f"New size: {new_size}\n\n"
f"PnL: {pnl}"
f"<b>⚠️ Flip отменён</b>\n\n"
f"{icon} {symbol} · {target_side}\n"
f"Текущая позиция: {position_side}\n\n"
f"Недостаточно условий для разворота\n"
f"{reason}\n"
f"Сила сигнала: {confidence:.2f}"
)
return NotificationMessage(
@@ -123,13 +243,7 @@ def _format_symbol(value: object) -> str:
if not symbol or symbol == "—":
return "—"
base_symbol = symbol.split("_", 1)[0]
parts = base_symbol.split("/", 1)
if len(parts) == 2:
return f"{parts[0]} / {parts[1]}"
return base_symbol
return symbol.split("_", 1)[0].split("/", 1)[0].upper()
def _format_leverage(value: object) -> str:
@@ -169,4 +283,45 @@ def _format_pnl(value: object) -> str:
if number < 0:
return f"🔴 −{amount}"
return "$ 0"
return "$ 0"
def _alert_priority(*, confidence: float, repeat_count: int) -> str:
if confidence >= 0.8 and repeat_count >= 3:
return "HIGH"
if confidence >= 0.6 or repeat_count >= 2:
return "MEDIUM"
return "LOW"
def _strength_label(priority: str) -> str:
mapping = {
"HIGH": "Сильный",
"MEDIUM": "Средний",
"LOW": "Слабый",
}
return mapping.get(priority.upper(), priority)
def _strength_bar(priority: str) -> str:
mapping = {
"HIGH": "●●●",
"MEDIUM": "●●○",
"LOW": "●○○",
}
return mapping.get(priority.upper(), "●○○")
def _format_notional(
*,
entry_price: object,
size: object,
) -> str:
try:
value = float(entry_price) * float(size)
except (TypeError, ValueError):
return "—"
return f"$ {value:,.2f}".replace(",", " ").rstrip("0").rstrip(".")