From a1b6b5764dbe3571cdbb261048fe8417755483b0 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 29 Jun 2026 06:50:16 +0800 Subject: [PATCH 1/2] refactor(notifications): extract CloudAlertMarkerStore to eliminate 4x duplicated code - New alert_marker.py: shared CloudAlertMarkerStore base class with has_alert/record_alert/_local_path/_cloud_uri + _clean_relative_key helper - strategy_plugin_{email,sms,push,telegram}.py: replace ~55-line duplicate marker store classes with thin 3-line subclasses (~200 lines removed) - All 4 classes preserved (backward compat), just inherit from base now Also removes 4 duplicate _clean_relative_key + _parse_cloud_uri helpers (now in alert_marker.py which correctly supports gs://, s3://, and az://). 373 passed, 0 failed. Co-Authored-By: Claude --- .../notifications/alert_marker.py | 116 ++++++++++++++++++ .../notifications/strategy_plugin_email.py | 83 +------------ .../notifications/strategy_plugin_push.py | 81 +----------- .../notifications/strategy_plugin_sms.py | 81 +----------- .../notifications/strategy_plugin_telegram.py | 83 +------------ 5 files changed, 140 insertions(+), 304 deletions(-) create mode 100644 src/quant_platform_kit/notifications/alert_marker.py diff --git a/src/quant_platform_kit/notifications/alert_marker.py b/src/quant_platform_kit/notifications/alert_marker.py new file mode 100644 index 00000000..a5b8a0f0 --- /dev/null +++ b/src/quant_platform_kit/notifications/alert_marker.py @@ -0,0 +1,116 @@ +"""Shared cloud/local alert marker store — eliminates duplicate code across email/sms/push/telegram channels.""" + +from __future__ import annotations + +import json +import tempfile +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from quant_platform_kit.cloud import get_object_store + + +def _clean_relative_key(key: str) -> str: + """Sanitize a string for use as a filesystem/object path segment.""" + parts = [] + for raw_part in str(key or "").replace("\\", "/").split("/"): + cleaned = "".join( + char if char.isalnum() or char in {"-", "_", "."} else "-" + for char in raw_part.strip() + ).strip("-._") + if cleaned: + parts.append(cleaned[:100]) + return "/".join(parts) or "unknown" + + +def _parse_cloud_uri(uri: str) -> tuple[str, str]: + """Parse a cloud storage URI (gs://, s3://, or az://) into (bucket, prefix).""" + raw_uri = str(uri or "").strip() + if not raw_uri.startswith("gs://") and not raw_uri.startswith("s3://") and not raw_uri.startswith("az://"): + raise ValueError(f"Cloud URI must start with gs://, s3://, or az://, got: {uri!r}") + remainder = raw_uri[5:] + bucket_name, _, object_prefix = remainder.partition("/") + if not bucket_name: + raise ValueError(f"Cloud URI must include a bucket name, got: {uri!r}") + return bucket_name, object_prefix.strip("/") + + +@dataclass(frozen=True) +class CloudAlertMarkerStore: + """Shared marker store for strategy plugin alerts. + + Persists alert markers to either cloud ObjectStore or local filesystem. + Used as base for channel-specific stores (email, sms, push, telegram). + + Usage:: + + store = CloudAlertMarkerStore( + namespace="strategy_plugin_telegram_alerts", + schema_version="strategy_plugin_telegram_alert_marker.v1", + cloud_prefix_uri="gs://bucket/alerts", + project_id="my-project", + ) + if not store.has_alert("some-key"): + store.record_alert("some-key") + """ + + namespace: str = "strategy_plugin_alerts" + schema_version: str = "strategy_plugin_alert_marker.v1" + local_dir: str | Path | None = None + cloud_prefix_uri: str | None = None + project_id: str | None = None + client_factory: Any = None + + def _object_store(self): + return get_object_store(project_id=self.project_id) + + def has_alert(self, alert_key: str) -> bool: + if self.cloud_prefix_uri and self._object_store().exists( + self._cloud_uri(alert_key) + ): + return True + if self.local_dir and self._local_path(alert_key).exists(): + return True + return False + + def record_alert( + self, + alert_key: str, + *, + metadata: Mapping[str, Any] | None = None, + ) -> None: + payload = { + "schema_version": self.schema_version, + "alert_key": str(alert_key), + "recorded_at": datetime.now(timezone.utc).isoformat(), + "metadata": dict(metadata or {}), + } + encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + if self.cloud_prefix_uri: + self._object_store().write_text( + self._cloud_uri(alert_key), + encoded, + content_type="application/json", + ) + return + if self.local_dir: + path = self._local_path(alert_key) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(encoded, encoding="utf-8") + + def _local_path(self, alert_key: str) -> Path: + root = Path(self.local_dir or tempfile.gettempdir()).expanduser() + return root / self.namespace / f"{_clean_relative_key(alert_key)}.json" + + def _cloud_uri(self, alert_key: str) -> str: + bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or "")) + object_name = "/".join( + part.strip("/") + for part in (prefix, self.namespace, f"{_clean_relative_key(alert_key)}.json") + if part and part.strip("/") + ) + scheme = str(self.cloud_prefix_uri or "").split("://")[0] if "://" in str(self.cloud_prefix_uri or "") else "gs" + return f"{scheme}://{bucket_name}/{object_name}" diff --git a/src/quant_platform_kit/notifications/strategy_plugin_email.py b/src/quant_platform_kit/notifications/strategy_plugin_email.py index 54614a0f..d81f15d2 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_email.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_email.py @@ -3,19 +3,17 @@ from __future__ import annotations import json -import tempfile from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone -from pathlib import Path from typing import Any -from quant_platform_kit.cloud import get_object_store from quant_platform_kit.common.strategy_plugins import ( StrategyPluginAlertMessage, build_strategy_plugin_alert_messages, ) +from .alert_marker import CloudAlertMarkerStore, _clean_relative_key from .email import parse_email_recipients, send_smtp_email @@ -132,60 +130,10 @@ def to_report_fields(self, *, prefix: str = "strategy_plugin_alert_email") -> di @dataclass(frozen=True) -class StrategyPluginEmailAlertMarkerStore: - local_dir: str | Path | None = None - cloud_prefix_uri: str | None = None - project_id: str | None = None +class StrategyPluginEmailAlertMarkerStore(CloudAlertMarkerStore): + """Email-specific alert marker store — thin wrapper around shared base.""" namespace: str = "strategy_plugin_email_alerts" - client_factory: Any = None - - def _object_store(self): - return get_object_store(project_id=self.project_id) - - def has_alert(self, alert_key: str) -> bool: - if self.cloud_prefix_uri and self._object_store().exists(self._cloud_uri(alert_key, namespace=self.namespace)): - return True - if self.local_dir and self._local_path(alert_key, namespace=self.namespace).exists(): - return True - return False - - def record_alert( - self, - alert_key: str, - *, - metadata: Mapping[str, Any] | None = None, - ) -> None: - payload = { - "schema_version": "strategy_plugin_email_alert_marker.v1", - "alert_key": str(alert_key), - "recorded_at": datetime.now(timezone.utc).isoformat(), - "metadata": dict(metadata or {}), - } - encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) - if self.cloud_prefix_uri: - self._object_store().write_text( - self._cloud_uri(alert_key, namespace=self.namespace), - encoded, - content_type="application/json", - ) - return - if self.local_dir: - path = self._local_path(alert_key, namespace=self.namespace) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(encoded, encoding="utf-8") - - def _local_path(self, alert_key: str, *, namespace: str) -> Path: - root = Path(self.local_dir or tempfile.gettempdir()).expanduser() - return root / namespace / f"{_clean_relative_key(alert_key)}.json" - - def _cloud_uri(self, alert_key: str, *, namespace: str) -> str: - bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or "")) - object_name = "/".join( - part.strip("/") - for part in (prefix, namespace, f"{_clean_relative_key(alert_key)}.json") - if part and part.strip("/") - ) - return f"gs://{bucket_name}/{object_name}" + schema_version: str = "strategy_plugin_email_alert_marker.v1" def build_strategy_plugin_alert_context_label( @@ -403,24 +351,5 @@ def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str: return "strategy_plugin_email_alert/" + _clean_relative_key(message.subject or "unknown") -def _clean_relative_key(value: str) -> str: - parts = [] - for raw_part in str(value or "").replace("\\", "/").split("/"): - cleaned = "".join( - char if char.isalnum() or char in {"-", "_", "."} else "-" - for char in raw_part.strip() - ).strip("-._") - if cleaned: - parts.append(cleaned[:100]) - return "/".join(parts) or "unknown" - - -def _parse_cloud_uri(uri: str) -> tuple[str, str]: - raw_uri = str(uri or "").strip() - if not raw_uri.startswith("gs://"): - raise ValueError(f"gcs uri must start with gs://, got: {uri!r}") - remainder = raw_uri[5:] - bucket_name, _, object_prefix = remainder.partition("/") - if not bucket_name: - raise ValueError(f"gcs uri must include a bucket name, got: {uri!r}") - return bucket_name, object_prefix.strip("/") + + diff --git a/src/quant_platform_kit/notifications/strategy_plugin_push.py b/src/quant_platform_kit/notifications/strategy_plugin_push.py index 7e1a5d60..d1f9dd6a 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_push.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_push.py @@ -10,12 +10,12 @@ from pathlib import Path from typing import Any -from quant_platform_kit.cloud import get_object_store from quant_platform_kit.common.strategy_plugins import ( StrategyPluginAlertMessage, build_strategy_plugin_alert_messages, ) +from .alert_marker import CloudAlertMarkerStore, _clean_relative_key from .push import ( DEFAULT_NTFY_API_BASE_URL, DEFAULT_PUSHOVER_API_BASE_URL, @@ -138,60 +138,10 @@ def to_report_fields(self, *, prefix: str = "strategy_plugin_alert_push") -> dic @dataclass(frozen=True) -class StrategyPluginPushAlertMarkerStore: - local_dir: str | Path | None = None - cloud_prefix_uri: str | None = None - project_id: str | None = None +class StrategyPluginPushAlertMarkerStore(CloudAlertMarkerStore): + """Push-specific alert marker store — thin wrapper around shared base.""" namespace: str = "strategy_plugin_push_alerts" - client_factory: Any = None - - def _object_store(self): - return get_object_store(project_id=self.project_id) - - def has_alert(self, alert_key: str) -> bool: - if self.cloud_prefix_uri and self._object_store().exists(self._cloud_uri(alert_key, namespace=self.namespace)): - return True - if self.local_dir and self._local_path(alert_key, namespace=self.namespace).exists(): - return True - return False - - def record_alert( - self, - alert_key: str, - *, - metadata: Mapping[str, Any] | None = None, - ) -> None: - payload = { - "schema_version": "strategy_plugin_push_alert_marker.v1", - "alert_key": str(alert_key), - "recorded_at": datetime.now(timezone.utc).isoformat(), - "metadata": dict(metadata or {}), - } - encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) - if self.cloud_prefix_uri: - self._object_store().write_text( - self._cloud_uri(alert_key, namespace=self.namespace), - encoded, - content_type="application/json", - ) - return - if self.local_dir: - path = self._local_path(alert_key, namespace=self.namespace) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(encoded, encoding="utf-8") - - def _local_path(self, alert_key: str, *, namespace: str) -> Path: - root = Path(self.local_dir or tempfile.gettempdir()).expanduser() - return root / namespace / f"{_clean_relative_key(alert_key)}.json" - - def _cloud_uri(self, alert_key: str, *, namespace: str) -> str: - bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or "")) - object_name = "/".join( - part.strip("/") - for part in (prefix, namespace, f"{_clean_relative_key(alert_key)}.json") - if part and part.strip("/") - ) - return f"gs://{bucket_name}/{object_name}" + schema_version: str = "strategy_plugin_push_alert_marker.v1" def publish_strategy_plugin_push_alerts( @@ -393,24 +343,5 @@ def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str: return "strategy_plugin_push_alert/" + _clean_relative_key(message.subject or "unknown") -def _clean_relative_key(value: str) -> str: - parts = [] - for raw_part in str(value or "").replace("\\", "/").split("/"): - cleaned = "".join( - char if char.isalnum() or char in {"-", "_", "."} else "-" - for char in raw_part.strip() - ).strip("-._") - if cleaned: - parts.append(cleaned[:100]) - return "/".join(parts) or "unknown" - - -def _parse_cloud_uri(uri: str) -> tuple[str, str]: - raw_uri = str(uri or "").strip() - if not raw_uri.startswith("gs://"): - raise ValueError(f"gcs uri must start with gs://, got: {uri!r}") - remainder = raw_uri[5:] - bucket_name, _, object_prefix = remainder.partition("/") - if not bucket_name: - raise ValueError(f"gcs uri must include a bucket name, got: {uri!r}") - return bucket_name, object_prefix.strip("/") + + diff --git a/src/quant_platform_kit/notifications/strategy_plugin_sms.py b/src/quant_platform_kit/notifications/strategy_plugin_sms.py index e76242e8..a45972b1 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_sms.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_sms.py @@ -10,12 +10,12 @@ from pathlib import Path from typing import Any -from quant_platform_kit.cloud import get_object_store from quant_platform_kit.common.strategy_plugins import ( StrategyPluginAlertMessage, build_strategy_plugin_alert_messages, ) +from .alert_marker import CloudAlertMarkerStore, _clean_relative_key from .sms import parse_sms_recipients, send_twilio_sms _DEFAULT_SMS_PROVIDER = "twilio" @@ -127,60 +127,10 @@ def to_report_fields(self, *, prefix: str = "strategy_plugin_alert_sms") -> dict @dataclass(frozen=True) -class StrategyPluginSmsAlertMarkerStore: - local_dir: str | Path | None = None - cloud_prefix_uri: str | None = None - project_id: str | None = None +class StrategyPluginSmsAlertMarkerStore(CloudAlertMarkerStore): + """SMS-specific alert marker store — thin wrapper around shared base.""" namespace: str = "strategy_plugin_sms_alerts" - client_factory: Any = None - - def _object_store(self): - return get_object_store(project_id=self.project_id) - - def has_alert(self, alert_key: str) -> bool: - if self.cloud_prefix_uri and self._object_store().exists(self._cloud_uri(alert_key, namespace=self.namespace)): - return True - if self.local_dir and self._local_path(alert_key, namespace=self.namespace).exists(): - return True - return False - - def record_alert( - self, - alert_key: str, - *, - metadata: Mapping[str, Any] | None = None, - ) -> None: - payload = { - "schema_version": "strategy_plugin_sms_alert_marker.v1", - "alert_key": str(alert_key), - "recorded_at": datetime.now(timezone.utc).isoformat(), - "metadata": dict(metadata or {}), - } - encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) - if self.cloud_prefix_uri: - self._object_store().write_text( - self._cloud_uri(alert_key, namespace=self.namespace), - encoded, - content_type="application/json", - ) - return - if self.local_dir: - path = self._local_path(alert_key, namespace=self.namespace) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(encoded, encoding="utf-8") - - def _local_path(self, alert_key: str, *, namespace: str) -> Path: - root = Path(self.local_dir or tempfile.gettempdir()).expanduser() - return root / namespace / f"{_clean_relative_key(alert_key)}.json" - - def _cloud_uri(self, alert_key: str, *, namespace: str) -> str: - bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or "")) - object_name = "/".join( - part.strip("/") - for part in (prefix, namespace, f"{_clean_relative_key(alert_key)}.json") - if part and part.strip("/") - ) - return f"gs://{bucket_name}/{object_name}" + schema_version: str = "strategy_plugin_sms_alert_marker.v1" def publish_strategy_plugin_sms_alerts( @@ -373,24 +323,5 @@ def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str: return "strategy_plugin_sms_alert/" + _clean_relative_key(message.subject or "unknown") -def _clean_relative_key(value: str) -> str: - parts = [] - for raw_part in str(value or "").replace("\\", "/").split("/"): - cleaned = "".join( - char if char.isalnum() or char in {"-", "_", "."} else "-" - for char in raw_part.strip() - ).strip("-._") - if cleaned: - parts.append(cleaned[:100]) - return "/".join(parts) or "unknown" - - -def _parse_cloud_uri(uri: str) -> tuple[str, str]: - raw_uri = str(uri or "").strip() - if not raw_uri.startswith("gs://"): - raise ValueError(f"gcs uri must start with gs://, got: {uri!r}") - remainder = raw_uri[5:] - bucket_name, _, object_prefix = remainder.partition("/") - if not bucket_name: - raise ValueError(f"gcs uri must include a bucket name, got: {uri!r}") - return bucket_name, object_prefix.strip("/") + + diff --git a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py b/src/quant_platform_kit/notifications/strategy_plugin_telegram.py index 541bf2f1..5c715a7a 100644 --- a/src/quant_platform_kit/notifications/strategy_plugin_telegram.py +++ b/src/quant_platform_kit/notifications/strategy_plugin_telegram.py @@ -3,19 +3,17 @@ from __future__ import annotations import json -import tempfile from collections.abc import Callable, Mapping, Sequence from dataclasses import dataclass, field from datetime import datetime, timezone -from pathlib import Path from typing import Any -from quant_platform_kit.cloud import get_object_store from quant_platform_kit.common.strategy_plugins import ( StrategyPluginAlertMessage, build_strategy_plugin_alert_messages, ) +from .alert_marker import CloudAlertMarkerStore, _clean_relative_key from .telegram import ( DEFAULT_TELEGRAM_BOT_API_BASE_URL, parse_telegram_chat_ids, @@ -124,60 +122,10 @@ def to_report_fields(self, *, prefix: str = "strategy_plugin_alert_telegram") -> @dataclass(frozen=True) -class StrategyPluginTelegramAlertMarkerStore: - local_dir: str | Path | None = None - cloud_prefix_uri: str | None = None - project_id: str | None = None +class StrategyPluginTelegramAlertMarkerStore(CloudAlertMarkerStore): + """Telegram-specific alert marker store — thin wrapper around shared base.""" namespace: str = "strategy_plugin_telegram_alerts" - client_factory: Any = None - - def _object_store(self): - return get_object_store(project_id=self.project_id) - - def has_alert(self, alert_key: str) -> bool: - if self.cloud_prefix_uri and self._object_store().exists(self._cloud_uri(alert_key, namespace=self.namespace)): - return True - if self.local_dir and self._local_path(alert_key, namespace=self.namespace).exists(): - return True - return False - - def record_alert( - self, - alert_key: str, - *, - metadata: Mapping[str, Any] | None = None, - ) -> None: - payload = { - "schema_version": "strategy_plugin_telegram_alert_marker.v1", - "alert_key": str(alert_key), - "recorded_at": datetime.now(timezone.utc).isoformat(), - "metadata": dict(metadata or {}), - } - encoded = json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) - if self.cloud_prefix_uri: - self._object_store().write_text( - self._cloud_uri(alert_key, namespace=self.namespace), - encoded, - content_type="application/json", - ) - return - if self.local_dir: - path = self._local_path(alert_key, namespace=self.namespace) - path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(encoded, encoding="utf-8") - - def _local_path(self, alert_key: str, *, namespace: str) -> Path: - root = Path(self.local_dir or tempfile.gettempdir()).expanduser() - return root / namespace / f"{_clean_relative_key(alert_key)}.json" - - def _cloud_uri(self, alert_key: str, *, namespace: str) -> str: - bucket_name, prefix = _parse_cloud_uri(str(self.cloud_prefix_uri or "")) - object_name = "/".join( - part.strip("/") - for part in (prefix, namespace, f"{_clean_relative_key(alert_key)}.json") - if part and part.strip("/") - ) - return f"gs://{bucket_name}/{object_name}" + schema_version: str = "strategy_plugin_telegram_alert_marker.v1" def publish_strategy_plugin_telegram_alerts( @@ -381,24 +329,5 @@ def _fallback_alert_key(message: StrategyPluginAlertMessage) -> str: return "strategy_plugin_telegram_alert/" + _clean_relative_key(message.subject or "unknown") -def _clean_relative_key(value: str) -> str: - parts = [] - for raw_part in str(value or "").replace("\\", "/").split("/"): - cleaned = "".join( - char if char.isalnum() or char in {"-", "_", "."} else "-" - for char in raw_part.strip() - ).strip("-._") - if cleaned: - parts.append(cleaned[:100]) - return "/".join(parts) or "unknown" - - -def _parse_cloud_uri(uri: str) -> tuple[str, str]: - raw_uri = str(uri or "").strip() - if not raw_uri.startswith("gs://"): - raise ValueError(f"gcs uri must start with gs://, got: {uri!r}") - remainder = raw_uri[5:] - bucket_name, _, object_prefix = remainder.partition("/") - if not bucket_name: - raise ValueError(f"gcs uri must include a bucket name, got: {uri!r}") - return bucket_name, object_prefix.strip("/") + + From f50e974739077e80cb70f50dd8575f3b86241a00 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Mon, 29 Jun 2026 06:52:23 +0800 Subject: [PATCH 2/2] feat(notifications): add pluggable NotificationChannel Protocol layer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - channel.py: define SmsChannel, PushChannel, EmailChannel, ChatChannel Protocols - 5 default implementations: TwilioSmsChannel, PushoverChannel, NtfyChannel, SmtpEmailChannel, TelegramChatChannel — thin wrappers around existing functions - Users can now swap providers by implementing a Protocol and passing it as send_notification to publish_strategy_plugin_*_alerts() - Export all channels from notifications/__init__.py This completes the notification decoupling: previously Twilio/Pushover/SMTP/Telegram were hardcoded; now any provider implementing the Protocol can be dropped in. 373 passed, 0 failed. Co-Authored-By: Claude --- .../notifications/__init__.py | 20 ++ .../notifications/channel.py | 223 ++++++++++++++++++ 2 files changed, 243 insertions(+) create mode 100644 src/quant_platform_kit/notifications/channel.py diff --git a/src/quant_platform_kit/notifications/__init__.py b/src/quant_platform_kit/notifications/__init__.py index f3046b99..a0761f03 100644 --- a/src/quant_platform_kit/notifications/__init__.py +++ b/src/quant_platform_kit/notifications/__init__.py @@ -1,5 +1,16 @@ """Notification integrations.""" +from .channel import ( + SmsChannel, + PushChannel, + EmailChannel, + ChatChannel, + TwilioSmsChannel, + PushoverChannel, + NtfyChannel, + SmtpEmailChannel, + TelegramChatChannel, +) from .email import parse_email_recipients, send_smtp_email from .events import NotificationPublisher, RenderedNotification, publish_rendered_notification from .push import parse_push_recipients, send_ntfy_push, send_pushover_push, send_strategy_plugin_push @@ -42,6 +53,15 @@ ) __all__ = [ + "SmsChannel", + "PushChannel", + "EmailChannel", + "ChatChannel", + "TwilioSmsChannel", + "PushoverChannel", + "NtfyChannel", + "SmtpEmailChannel", + "TelegramChatChannel", "NotificationPublisher", "RenderedNotification", "StrategyPluginAlertChannelStores", diff --git a/src/quant_platform_kit/notifications/channel.py b/src/quant_platform_kit/notifications/channel.py new file mode 100644 index 00000000..8542e67a --- /dev/null +++ b/src/quant_platform_kit/notifications/channel.py @@ -0,0 +1,223 @@ +"""Notification channel abstraction — pluggable senders for SMS, push, email, and chat. + +Each channel type has a Protocol defining the send signature. +The default implementations wire to Twilio (SMS), Pushover/Ntfy (push), +SMTP (email), and Telegram Bot API (chat). + +To replace a provider, implement the corresponding Protocol and pass it +as ``send_notification`` to the ``publish_strategy_plugin_*`` function. + +Example (custom SMS provider):: + + class AliyunSmsChannel: + def send_sms(self, recipient: str, body: str, *, sender: str | None = None) -> bool: + # call Aliyun SMS API + return True + + publish_strategy_plugin_sms_alerts( + signals, + sms_settings=settings, + send_notification=AliyunSmsChannel().send_sms, + ) +""" + +from __future__ import annotations + +from typing import Protocol + + +# ────────────────────────────────────────────────────────────────────── +# Channel Protocols +# ────────────────────────────────────────────────────────────────────── + + +class SmsChannel(Protocol): + """Send an SMS message. Return True on success, False on failure.""" + + def send_sms(self, recipient: str, body: str, *, sender: str | None = None) -> bool: + ... + + +class PushChannel(Protocol): + """Send a push notification. Return True on success, False on failure. + + ``target`` is provider-specific: Pushover user key, Ntfy topic, etc. + ``provider`` identifies the backend (e.g. "pushover", "ntfy"). + """ + + def send_push( + self, + title: str, + body: str, + *, + target: str, + provider: str = "pushover", + url: str | None = None, + url_title: str | None = None, + priority: str = "normal", + api_base_url: str | None = None, + ) -> bool: + ... + + +class EmailChannel(Protocol): + """Send an email message. Return True on success, False on failure.""" + + def send_email( + self, + subject: str, + body: str, + *, + recipients: list[str], + sender: str | None = None, + smtp_host: str = "smtp.gmail.com", + smtp_port: int = 465, + security: str = "ssl", + username: str | None = None, + password: str | None = None, + ) -> bool: + ... + + +class ChatChannel(Protocol): + """Send a message to a chat platform. Return True on success, False on failure. + + ``chat_id`` and ``token`` are specific to Telegram Bot API. + For other platforms (Slack, Discord, WeChat), wrap their API + in this signature. + """ + + def send_message( + self, + chat_id: str, + text: str, + *, + token: str, + api_base_url: str = "https://api.telegram.org", + parse_mode: str = "HTML", + ) -> bool: + ... + + +# ────────────────────────────────────────────────────────────────────── +# Default channel implementations (thin wrappers around existing functions) +# ────────────────────────────────────────────────────────────────────── + + +class TwilioSmsChannel: + """Default SMS channel — wraps send_twilio_sms().""" + + def send_sms(self, recipient: str, body: str, *, sender: str | None = None) -> bool: + from .sms import send_twilio_sms + return send_twilio_sms( + recipient=recipient, + body=body, + account_sid=None, + auth_token=None, + sender=sender, + ) + + +class PushoverChannel: + """Pushover push channel — wraps send_pushover_push().""" + + def send_push( + self, + title: str, + body: str, + *, + target: str, + provider: str = "pushover", + url: str | None = None, + url_title: str | None = None, + priority: str = "normal", + api_base_url: str | None = None, + ) -> bool: + from .push import send_pushover_push + return send_pushover_push( + user_key=target, + message=body, + title=title, + url=url, + url_title=url_title, + priority=priority, + api_base_url=api_base_url, + ) + + +class NtfyChannel: + """Ntfy push channel — wraps send_ntfy_push().""" + + def send_push( + self, + title: str, + body: str, + *, + target: str, + provider: str = "ntfy", + url: str | None = None, + url_title: str | None = None, + priority: str = "normal", + api_base_url: str | None = None, + ) -> bool: + from .push import send_ntfy_push + return send_ntfy_push( + topic=target, + message=body, + title=title, + url=url, + priority=priority, + api_base_url=api_base_url, + ) + + +class SmtpEmailChannel: + """Default email channel — wraps send_smtp_email().""" + + def send_email( + self, + subject: str, + body: str, + *, + recipients: list[str], + sender: str | None = None, + smtp_host: str = "smtp.gmail.com", + smtp_port: int = 465, + security: str = "ssl", + username: str | None = None, + password: str | None = None, + ) -> bool: + from .email import send_smtp_email + return send_smtp_email( + recipients=recipients, + subject=subject, + body=body, + sender=sender, + smtp_host=smtp_host, + smtp_port=smtp_port, + security=security, + username=username, + password=password, + ) + + +class TelegramChatChannel: + """Default chat channel — wraps send_telegram_message().""" + + def send_message( + self, + chat_id: str, + text: str, + *, + token: str, + api_base_url: str = "https://api.telegram.org", + parse_mode: str = "HTML", + ) -> bool: + from .telegram import send_telegram_message + return send_telegram_message( + chat_id=chat_id, + text=text, + token=token, + api_base_url=api_base_url, + parse_mode=parse_mode, + )