refactor(notifications): deduplicate AlertMarkerStore + add pluggable channel Protocols (#121)

* 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 <noreply@anthropic.com>

* feat(notifications): add pluggable NotificationChannel Protocol layer

- 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 <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: Pigbibi

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",

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}"

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,
+        )