Decouple notification delivery sink

Author: Pigbibi

application/rebalance_service.py

@@ -10,6 +10,7 @@
     build_reconciliation_record,
     write_reconciliation_record,
 )
+from notifications.events import NotificationPublisher, RenderedNotification
 
 
 _ZH_REASON_REPLACEMENTS = (
@@ -455,6 +456,10 @@ def run_strategy_core(
     reconciliation_output_path=None,
     result_hook=None,
 ):
+    notification_publisher = NotificationPublisher(
+        log_message=lambda message: print(message, flush=True),
+        send_message=send_tg_message,
+    )
     ib = None
     try:
         ib = connect_ib()
@@ -524,8 +529,12 @@ def run_strategy_core(
                 body_lines=[no_op_text],
                 dashboard_text=strategy_dashboard,
             )
-            print(detailed_message, flush=True)
-            send_tg_message(compact_message)
+            notification_publisher.publish(
+                RenderedNotification(
+                    detailed_text=detailed_message,
+                    compact_text=compact_message,
+                )
+            )
             if callable(result_hook):
                 result_hook(
                     {
@@ -614,8 +623,12 @@ def run_strategy_core(
                 dashboard_text=strategy_dashboard,
             )
 
-        print(detailed_message, flush=True)
-        send_tg_message(compact_message)
+        notification_publisher.publish(
+            RenderedNotification(
+                detailed_text=detailed_message,
+                compact_text=compact_message,
+            )
+        )
         if callable(result_hook):
             result_hook(
                 {

main.py

@@ -16,6 +16,7 @@
 except ImportError:
     compute_v1 = None
 
+from notifications.events import NotificationPublisher, RenderedNotification
 from notifications.telegram import build_strategy_display_name, build_translator, send_telegram_message
 from quant_platform_kit.common.models import OrderIntent
 from quant_platform_kit.common.runtime_reports import (
@@ -302,6 +303,19 @@ def send_tg_message(message):
     )
 
 
+def publish_notification(*, detailed_text, compact_text):
+    publisher = NotificationPublisher(
+        log_message=lambda message: print(message, flush=True),
+        send_message=send_tg_message,
+    )
+    publisher.publish(
+        RenderedNotification(
+            detailed_text=detailed_text,
+            compact_text=compact_text,
+        )
+    )
+
+
 def connect_ib():
     host = get_ib_host()
     last_error = None
@@ -573,8 +587,7 @@ def run_paper_liquidation_cycle():
             f"{SEPARATOR}\n"
             f"{_format_liquidation_orders(summary.get('orders_submitted'))}"
         )
-        send_tg_message(message)
-        print(message, flush=True)
+        publish_notification(detailed_text=message, compact_text=message)
         global LAST_CYCLE_DETAILS
         LAST_CYCLE_DETAILS = {"execution_summary": summary}
         return "OK"
@@ -716,8 +729,7 @@ def handle_request():
             error_message=str(exc),
         )
         error_msg = f"{t('error_title')}\n{traceback.format_exc()}"
-        send_tg_message(error_msg)
-        print(error_msg, flush=True)
+        publish_notification(detailed_text=error_msg, compact_text=error_msg)
         return "Error", 500
     finally:
         if lock_acquired:

notifications/events.py

@@ -0,0 +1,44 @@
+"""Notification event envelope and delivery helpers."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class RenderedNotification:
+    """Rendered notification payload split by sink."""
+
+    detailed_text: str
+    compact_text: str
+
+
+@dataclass(frozen=True)
+class NotificationPublisher:
+    """Publish rendered notifications to the configured sinks."""
+
+    log_message: Callable[[str], None]
+    send_message: Callable[[str], None]
+
+    def publish(self, notification: RenderedNotification) -> None:
+        publish_rendered_notification(
+            notification,
+            log_message=self.log_message,
+            send_message=self.send_message,
+        )
+
+
+def publish_rendered_notification(
+    notification: RenderedNotification,
+    *,
+    log_message: Callable[[str], None],
+    send_message: Callable[[str], None],
+) -> None:
+    """Write the detailed log copy and send the compact user notification."""
+    detailed = str(notification.detailed_text or "").strip()
+    compact = str(notification.compact_text or "").strip()
+    if detailed:
+        log_message(detailed)
+    if compact:
+        send_message(compact)

tests/test_notification_events.py

@@ -0,0 +1,55 @@
+from notifications.events import (
+    NotificationPublisher,
+    RenderedNotification,
+    publish_rendered_notification,
+)
+
+
+def test_publish_rendered_notification_splits_log_and_send_sinks():
+    logged = []
+    sent = []
+
+    publish_rendered_notification(
+        RenderedNotification(
+            detailed_text="detailed copy",
+            compact_text="compact copy",
+        ),
+        log_message=logged.append,
+        send_message=sent.append,
+    )
+
+    assert logged == ["detailed copy"]
+    assert sent == ["compact copy"]
+
+
+def test_publish_rendered_notification_skips_empty_sinks():
+    logged = []
+    sent = []
+
+    publish_rendered_notification(
+        RenderedNotification(detailed_text="  ", compact_text=""),
+        log_message=logged.append,
+        send_message=sent.append,
+    )
+
+    assert logged == []
+    assert sent == []
+
+
+def test_notification_publisher_uses_configured_sinks():
+    logged = []
+    sent = []
+    publisher = NotificationPublisher(
+        log_message=logged.append,
+        send_message=sent.append,
+    )
+
+    publisher.publish(
+        RenderedNotification(
+            detailed_text="detailed copy",
+            compact_text="compact copy",
+        )
+    )
+
+    assert logged == ["detailed copy"]
+    assert sent == ["compact copy"]