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

07.4.4.1.11 — Advanced Trend Quality & EMA Distance Layer

Browse Source
This commit is contained in:
Sergey
2026-05-20 21:15:00 +03:00
parent 2c75f95b46
commit 06ea376cb5
36 changed files with 6260 additions and 2092 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -5,6 +5,7 @@ from __future__ import annotations
from src.notifications.models import NotificationMessage
from src.runtime_events.event_types import RuntimeEventType
from src.runtime_events.models import RuntimeEvent
from src.core.numbers import safe_float
def build_execution_notification(event: RuntimeEvent) -> NotificationMessage | None:
@@ -27,8 +28,9 @@ def _build_position_opened(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
strategy = str(payload.get("strategy") or "—")
side = str(payload.get("side") or "—").upper()
strategy = str(payload.get("strategy") or "—").title()
side_raw = str(payload.get("side") or "—").upper()
side = side_raw.title()
leverage = _format_leverage(payload.get("leverage"))
entry_price = _format_price(payload.get("entry_price"))
size = _format_size(payload.get("size"))
@@ -39,13 +41,13 @@ def _build_position_opened(event: RuntimeEvent) -> NotificationMessage:
)
semantic_lines = payload.get("semantic_lines") or []
side_icon = "🟢" if side == "LONG" else "🔴"
side_icon = "🟢" if side_raw == "LONG" else "🔴"
lines = [
"<b>🧾 Позиция открыта</b>",
"",
f"{side_icon} {symbol} · {strategy} · {side} {leverage}",
f"Вход: {entry_price}",
f"Вход: ${entry_price}",
f"Размер: {size}",
f"Объём: {_format_notional(entry_price=payload.get('entry_price'), size=payload.get('size'))}",
"",
@@ -71,7 +73,7 @@ def _build_position_closed(event: RuntimeEvent) -> NotificationMessage:
payload = event.payload
symbol = _format_symbol(payload.get("symbol"))
side = str(payload.get("side") or "—").upper()
side = str(payload.get("side") or "—").title()
leverage = _format_leverage(payload.get("leverage"))
entry_price = _format_price(payload.get("entry_price"))
@@ -91,8 +93,8 @@ def _build_position_closed(event: RuntimeEvent) -> NotificationMessage:
f"{pnl_icon} {pnl_label} · {pnl_text}",
"",
f"{symbol} · {side} {leverage}",
f"Вход: $ {entry_price}",
f"Выход: $ {exit_price}",
f"Вход: ${entry_price}",
f"Выход: ${exit_price}",
f"Размер: {size}",
]
@@ -138,8 +140,13 @@ def _build_position_flipped(event: RuntimeEvent) -> NotificationMessage:
symbol = _format_symbol(payload.get("symbol"))
strategy = str(payload.get("strategy") or "—").title()
old_side = str(payload.get("old_side") or "—").upper()
new_side = str(payload.get("new_side") or payload.get("side") or "—").upper()
old_side_raw = str(payload.get("old_side") or "—").upper()
new_side_raw = str(
payload.get("new_side") or payload.get("side") or "—"
).upper()
old_side = old_side_raw.title()
new_side = new_side_raw.title()
old_leverage = _format_leverage(
payload.get("old_leverage")
@@ -161,8 +168,8 @@ def _build_position_flipped(event: RuntimeEvent) -> NotificationMessage:
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 "🔴"
old_icon = "🟢" if old_side_raw == "LONG" else "🔴"
new_icon = "🟢" if new_side_raw == "LONG" else "🔴"
confidence = float(payload.get("confidence") or 0.0)
repeat_count = int(payload.get("repeat_count") or 0)
@@ -178,12 +185,12 @@ def _build_position_flipped(event: RuntimeEvent) -> NotificationMessage:
f"{symbol} · {strategy} {old_icon} {old_side} → {new_icon} {new_side}",
"",
f"Закрыта {old_side} {old_leverage}",
f"Вход: $ {entry_price}",
f"Выход: $ {exit_price}",
f"Вход: ${entry_price}",
f"Выход: ${exit_price}",
f"Размер: {old_size}",
"",
f"Открыта {new_side} {new_leverage}",
f"Вход: $ {new_entry_price}",
f"Вход: ${new_entry_price}",
f"Размер: {new_size}",
(
"Объём: "
@@ -215,9 +222,9 @@ def _build_flip_blocked(event: RuntimeEvent) -> NotificationMessage:
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()
position_side = str(payload.get("position_side") or "—").title()
target_side = "LONG" if signal == "BUY" else "SHORT" if signal == "SELL" else "—"
target_side = "Long" if signal == "BUY" else "Short" if signal == "SELL" else "—"
icon = "🟢" if target_side == "LONG" else "🔴" if target_side == "SHORT" else ""
text = (
@@ -247,43 +254,30 @@ def _format_symbol(value: object) -> str:
def _format_leverage(value: object) -> str:
try:
return f"x{float(value):g}"
except (TypeError, ValueError):
number = safe_float(value)
if number is None:
return "—"
return f"x{number:g}"
def _format_price(value: object) -> str:
try:
number = float(value)
except (TypeError, ValueError):
number = safe_float(value)
if number is None:
return "—"
return f"{number:,.2f}".replace(",", " ")
def _format_size(value: object) -> str:
try:
return f"{float(value):.8f}".rstrip("0").rstrip(".")
except (TypeError, ValueError):
number = safe_float(value)
if number is None:
return "—"
def _format_pnl(value: object) -> str:
try:
number = float(value)
except (TypeError, ValueError):
return "—"
amount = f"$ {abs(number):,.2f}".replace(",", " ").rstrip("0").rstrip(".")
if number > 0:
return f"🟢 +{amount}"
if number < 0:
return f"🔴 −{amount}"
return "$ 0"
return f"{number:.8f}".rstrip("0").rstrip(".")
def _alert_priority(*, confidence: float, repeat_count: int) -> str:
@@ -319,9 +313,12 @@ def _format_notional(
entry_price: object,
size: object,
) -> str:
try:
value = float(entry_price) * float(size)
except (TypeError, ValueError):
entry = safe_float(entry_price)
amount = safe_float(size)
if entry is None or amount is None:
return "—"
value = entry * amount
return f"$ {value:,.2f}".replace(",", " ").rstrip("0").rstrip(".")

99
app/src/notifications/templates/signal.py
View File

@@ -2,6 +2,8 @@
from __future__ import annotations
from src.core.numbers import safe_float
from src.core.types import JsonDict, JsonList, NumericLike
from src.notifications.models import NotificationMessage
from src.runtime_events.event_types import RuntimeEventType
from src.runtime_events.models import RuntimeEvent
@@ -11,34 +13,49 @@ def build_signal_notification(event: RuntimeEvent) -> NotificationMessage | None
if event.event_type != RuntimeEventType.AUTO_SIGNAL_READY:
return None
payload = event.payload
payload: JsonDict = event.payload
signal = str(payload.get("signal") or "—").upper()
symbol = _format_symbol(str(payload.get("symbol") or "—"))
confidence = float(payload.get("confidence") or 0.0)
position_context = str(payload.get("position_context") or "NONE").upper()
semantic_lines = payload.get("semantic_lines") or []
confidence = safe_float(payload.get("confidence")) or 0.0
repeat_count = int(safe_float(payload.get("repeat_count")) or 0)
priority = str(event.priority or _alert_priority(
confidence=confidence,
repeat_count=int(payload.get("repeat_count") or 0),
)).upper()
position_context = str(payload.get("position_context") or "NONE").upper()
semantic_lines = _as_json_list(payload.get("semantic_lines"))
bid_price = payload.get("bid_price")
ask_price = payload.get("ask_price")
priority = str(
event.priority
or _alert_priority(
confidence=confidence,
repeat_count=repeat_count,
)
).upper()
direction = _signal_direction(signal)
direction_key = direction.upper()
icon = _direction_icon(direction)
strength = _strength_label(priority)
strength_bar = _strength_bar(priority)
market_price_line = _market_price_line(
direction=direction_key,
bid_price=bid_price,
ask_price=ask_price,
)
lines = [
f"<b>Сигнал {icon} {symbol} · {direction}</b>",
"",
]
if position_context not in {"NONE", "—", ""} and position_context != direction:
lines.extend([
"⚠️ ПРОТИВ ПОЗИЦИИ",
"",
])
if market_price_line:
lines.extend([market_price_line, ""])
if position_context not in {"NONE", "—", ""} and position_context != direction_key:
lines.extend(["⚠️ ПРОТИВ ПОЗИЦИИ", ""])
lines.append(f"{strength_bar} {strength} · {confidence:.2f}")
@@ -87,19 +104,21 @@ def _strength_bar(priority: str) -> str:
def _signal_direction(signal: str) -> str:
if signal == "BUY":
return "LONG"
return "Long"
if signal == "SELL":
return "SHORT"
return "Short"
return "—"
def _direction_icon(direction: str) -> str:
if direction == "LONG":
normalized = direction.upper()
if normalized == "LONG":
return "🟢"
if direction == "SHORT":
if normalized == "SHORT":
return "🔴"
return ""
@@ -112,14 +131,39 @@ def _format_symbol(symbol: str) -> str:
return symbol.split("_", 1)[0].split("/", 1)[0].upper()
def _format_leverage(leverage: object) -> str:
if isinstance(leverage, (int, float)):
return f"x{leverage:g}"
def _market_price_line(
*,
direction: str,
bid_price: NumericLike | None,
ask_price: NumericLike | None,
) -> str:
bid = _format_price(bid_price)
ask = _format_price(ask_price)
return "—"
if bid == "—" and ask == "—":
return ""
if direction == "LONG":
return f"Цена входа: Ask ${ask} / Bid ${bid}"
if direction == "SHORT":
return f"Цена входа: Bid ${bid} / Ask ${ask}"
return f"Цена рынка: Bid ${bid} / Ask ${ask}"
def _dedupe_key(payload: dict) -> str:
def _format_price(value: NumericLike | None) -> str:
number = safe_float(value)
if number is None:
return "—"
return f"{number:,.2f}".replace(",", " ")
def _dedupe_key(payload: JsonDict) -> str:
confidence = safe_float(payload.get("confidence")) or 0.0
return (
f"auto_signal_ready:"
f"{payload.get('position_context')}:"
@@ -127,7 +171,14 @@ def _dedupe_key(payload: dict) -> str:
f"{payload.get('strategy')}:"
f"{payload.get('signal')}:"
f"{payload.get('repeat_count')}:"
f"{float(payload.get('confidence') or 0.0):.2f}:"
f"{confidence:.2f}:"
f"{payload.get('decision_status')}:"
f"{payload.get('reason')}"
)
)
def _as_json_list(value: object) -> JsonList:
if isinstance(value, list):
return value
return []