07.4.3.18.2 — Runtime Notification Migration

app/src/notifications/templates/execution.py

@@ -8,47 +8,165 @@ from src.runtime_events.models import RuntimeEvent
 
 
 def build_execution_notification(event: RuntimeEvent) -> NotificationMessage | None:
-    if event.event_type not in {
-        RuntimeEventType.POSITION_OPENED,
-        RuntimeEventType.POSITION_CLOSED,
-        RuntimeEventType.POSITION_FLIPPED,
-    }:
-        return None
+    if event.event_type == RuntimeEventType.POSITION_OPENED:
+        return _build_position_opened(event)
 
+    if event.event_type == RuntimeEventType.POSITION_CLOSED:
+        return _build_position_closed(event)
+
+    if event.event_type == RuntimeEventType.POSITION_FLIPPED:
+        return _build_position_flipped(event)
+
+    return None
+
+
+def _build_position_opened(event: RuntimeEvent) -> NotificationMessage:
     payload = event.payload
 
-    symbol = str(payload.get("symbol") or "—")
-    side = str(payload.get("side") or payload.get("new_side") or "—")
-    entry_price = payload.get("entry_price") or payload.get("new_entry_price")
-    exit_price = payload.get("exit_price")
-    pnl = payload.get("pnl")
+    symbol = _format_symbol(payload.get("symbol"))
+    side = str(payload.get("side") or "—")
+    leverage = _format_leverage(payload.get("leverage"))
+    entry_price = _format_price(payload.get("entry_price"))
+    size = _format_size(payload.get("size"))
 
-    if event.event_type == RuntimeEventType.POSITION_OPENED:
-        title = "📄 Position opened"
-    elif event.event_type == RuntimeEventType.POSITION_CLOSED:
-        title = "✅ Position closed"
-    else:
-        title = "🔁 Position flipped"
+    side_icon = "🟢" if side == "LONG" else "🔴"
 
-    lines = [
-        f"<b>{title}</b>",
-        "",
-        f"{symbol} · {side}",
-    ]
-
-    if entry_price is not None:
-        lines.append(f"Entry: $ {entry_price}")
-
-    if exit_price is not None:
-        lines.append(f"Exit: $ {exit_price}")
-
-    if pnl is not None:
-        lines.append("")
-        lines.append(f"PnL: {pnl}")
+    text = (
+        f"<b>📄 Paper position opened {side_icon} {side}</b>\n\n"
+        f"{symbol} · {leverage}\n"
+        f"Entry: $ {entry_price}\n"
+        f"Size: {size}"
+    )
 
     return NotificationMessage(
         title=event.title,
-        text="\n".join(lines),
+        text=text,
         priority=event.priority,
         dedupe_key=event.dedupe_key,
-    )
+    )
+
+
+def _build_position_closed(event: RuntimeEvent) -> NotificationMessage:
+    payload = event.payload
+
+    symbol = _format_symbol(payload.get("symbol"))
+    side = str(payload.get("side") or "—")
+    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 ""
+
+    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}"
+    )
+
+    return NotificationMessage(
+        title=event.title,
+        text=text,
+        priority=event.priority,
+        dedupe_key=event.dedupe_key,
+    )
+
+
+def _build_position_flipped(event: RuntimeEvent) -> NotificationMessage:
+    payload = event.payload
+
+    symbol = _format_symbol(payload.get("symbol"))
+    leverage = _format_leverage(payload.get("leverage"))
+
+    old_side = str(payload.get("old_side") or "—")
+    new_side = str(payload.get("new_side") or payload.get("side") or "—")
+
+    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"))
+
+    old_icon = "🟢" if old_side == "LONG" else "🔴"
+    new_icon = "🟢" if new_side == "LONG" 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}"
+    )
+
+    return NotificationMessage(
+        title=event.title,
+        text=text,
+        priority=event.priority,
+        dedupe_key=event.dedupe_key,
+    )
+
+
+def _format_symbol(value: object) -> str:
+    symbol = str(value or "—")
+
+    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
+
+
+def _format_leverage(value: object) -> str:
+    try:
+        return f"x{float(value):g}"
+    except (TypeError, ValueError):
+        return "—"
+
+
+def _format_price(value: object) -> str:
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return "—"
+
+    return f"{number:,.2f}".replace(",", " ")
+
+
+def _format_size(value: object) -> str:
+    try:
+        return f"{float(value):.8f}".rstrip("0").rstrip(".")
+    except (TypeError, ValueError):
+        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"