feat: add Discord and Telegram webhook dispatcher (#21)
Implements AlertDispatcher with multi-channel delivery support: - DiscordChannel using webhook URL with embed formatting - TelegramChannel using Bot API with MarkdownV2 - Rate limiting per channel (configurable) - Circuit breaker pattern for failing channels - Async delivery with retry and exponential backoff - 23 comprehensive tests 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
Patrick Selamy
co-authored by
Claude Opus 4.5
Claude Opus 4.5
parent
f91dbb6b6d
commit
b0314fee57
@@ -1,9 +1,23 @@
|
||||
"""Alerting layer - Real-time notification delivery."""
|
||||
|
||||
from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
|
||||
from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
|
||||
from polymarket_insider_tracker.alerter.dispatcher import (
|
||||
AlertChannel,
|
||||
AlertDispatcher,
|
||||
CircuitBreakerState,
|
||||
DispatchResult,
|
||||
)
|
||||
from polymarket_insider_tracker.alerter.formatter import AlertFormatter
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
__all__ = [
|
||||
"AlertChannel",
|
||||
"AlertDispatcher",
|
||||
"AlertFormatter",
|
||||
"CircuitBreakerState",
|
||||
"DiscordChannel",
|
||||
"DispatchResult",
|
||||
"FormattedAlert",
|
||||
"TelegramChannel",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,9 @@
|
||||
"""Alert channel implementations for various platforms."""
|
||||
|
||||
from polymarket_insider_tracker.alerter.channels.discord import DiscordChannel
|
||||
from polymarket_insider_tracker.alerter.channels.telegram import TelegramChannel
|
||||
|
||||
__all__ = [
|
||||
"DiscordChannel",
|
||||
"TelegramChannel",
|
||||
]
|
||||
@@ -0,0 +1,118 @@
|
||||
"""Discord webhook channel implementation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class DiscordChannel:
|
||||
"""Discord webhook channel for sending alerts.
|
||||
|
||||
Sends formatted alerts to Discord via webhook URL with rate limiting
|
||||
and retry support.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
webhook_url: str,
|
||||
*,
|
||||
rate_limit_per_minute: int = 30,
|
||||
max_retries: int = 3,
|
||||
retry_delay: float = 1.0,
|
||||
timeout: float = 10.0,
|
||||
) -> None:
|
||||
"""Initialize Discord channel.
|
||||
|
||||
Args:
|
||||
webhook_url: Discord webhook URL.
|
||||
rate_limit_per_minute: Maximum messages per minute (Discord limit is 30).
|
||||
max_retries: Maximum retry attempts on failure.
|
||||
retry_delay: Base delay between retries (exponential backoff).
|
||||
timeout: HTTP request timeout in seconds.
|
||||
"""
|
||||
self.webhook_url = webhook_url
|
||||
self.rate_limit_per_minute = rate_limit_per_minute
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
self.timeout = timeout
|
||||
self.name = "discord"
|
||||
|
||||
# Rate limiting state
|
||||
self._request_times: list[float] = []
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _wait_for_rate_limit(self) -> None:
|
||||
"""Wait if rate limit is exceeded."""
|
||||
async with self._lock:
|
||||
now = asyncio.get_event_loop().time()
|
||||
# Remove requests older than 1 minute
|
||||
self._request_times = [t for t in self._request_times if now - t < 60]
|
||||
|
||||
if len(self._request_times) >= self.rate_limit_per_minute:
|
||||
# Wait until the oldest request expires
|
||||
wait_time = 60 - (now - self._request_times[0])
|
||||
if wait_time > 0:
|
||||
logger.debug(f"Discord rate limit hit, waiting {wait_time:.2f}s")
|
||||
await asyncio.sleep(wait_time)
|
||||
|
||||
self._request_times.append(now)
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to Discord webhook.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert with discord_embed.
|
||||
|
||||
Returns:
|
||||
True if delivery succeeded, False otherwise.
|
||||
"""
|
||||
await self._wait_for_rate_limit()
|
||||
|
||||
payload = {
|
||||
"embeds": [alert.discord_embed],
|
||||
}
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.post(
|
||||
self.webhook_url,
|
||||
json=payload,
|
||||
)
|
||||
|
||||
if response.status_code == 204:
|
||||
logger.info("Discord alert delivered successfully")
|
||||
return True
|
||||
|
||||
if response.status_code == 429:
|
||||
# Rate limited by Discord
|
||||
retry_after = response.json().get("retry_after", 1.0)
|
||||
logger.warning(f"Discord rate limited, retry after {retry_after}s")
|
||||
await asyncio.sleep(retry_after)
|
||||
continue
|
||||
|
||||
logger.error(
|
||||
f"Discord webhook failed: {response.status_code} {response.text}"
|
||||
)
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.warning(f"Discord webhook timeout (attempt {attempt + 1})")
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Discord webhook error: {e}")
|
||||
|
||||
# Exponential backoff
|
||||
if attempt < self.max_retries - 1:
|
||||
delay = self.retry_delay * (2**attempt)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
logger.error("Discord delivery failed after all retries")
|
||||
return False
|
||||
@@ -0,0 +1,131 @@
|
||||
"""Telegram Bot API channel implementation."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import httpx
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
TELEGRAM_API_BASE = "https://api.telegram.org/bot{token}/sendMessage"
|
||||
|
||||
|
||||
class TelegramChannel:
|
||||
"""Telegram Bot API channel for sending alerts.
|
||||
|
||||
Sends formatted alerts to Telegram via Bot API with rate limiting
|
||||
and retry support.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
bot_token: str,
|
||||
chat_id: str,
|
||||
*,
|
||||
rate_limit_per_minute: int = 20,
|
||||
max_retries: int = 3,
|
||||
retry_delay: float = 1.0,
|
||||
timeout: float = 10.0,
|
||||
) -> None:
|
||||
"""Initialize Telegram channel.
|
||||
|
||||
Args:
|
||||
bot_token: Telegram bot token.
|
||||
chat_id: Target chat/channel ID.
|
||||
rate_limit_per_minute: Maximum messages per minute.
|
||||
max_retries: Maximum retry attempts on failure.
|
||||
retry_delay: Base delay between retries (exponential backoff).
|
||||
timeout: HTTP request timeout in seconds.
|
||||
"""
|
||||
self.bot_token = bot_token
|
||||
self.chat_id = chat_id
|
||||
self.rate_limit_per_minute = rate_limit_per_minute
|
||||
self.max_retries = max_retries
|
||||
self.retry_delay = retry_delay
|
||||
self.timeout = timeout
|
||||
self.name = "telegram"
|
||||
|
||||
self._api_url = TELEGRAM_API_BASE.format(token=bot_token)
|
||||
|
||||
# Rate limiting state
|
||||
self._request_times: list[float] = []
|
||||
self._lock = asyncio.Lock()
|
||||
|
||||
async def _wait_for_rate_limit(self) -> None:
|
||||
"""Wait if rate limit is exceeded."""
|
||||
async with self._lock:
|
||||
now = asyncio.get_event_loop().time()
|
||||
# Remove requests older than 1 minute
|
||||
self._request_times = [t for t in self._request_times if now - t < 60]
|
||||
|
||||
if len(self._request_times) >= self.rate_limit_per_minute:
|
||||
# Wait until the oldest request expires
|
||||
wait_time = 60 - (now - self._request_times[0])
|
||||
if wait_time > 0:
|
||||
logger.debug(f"Telegram rate limit hit, waiting {wait_time:.2f}s")
|
||||
await asyncio.sleep(wait_time)
|
||||
|
||||
self._request_times.append(now)
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to Telegram channel.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert with telegram_markdown.
|
||||
|
||||
Returns:
|
||||
True if delivery succeeded, False otherwise.
|
||||
"""
|
||||
await self._wait_for_rate_limit()
|
||||
|
||||
payload = {
|
||||
"chat_id": self.chat_id,
|
||||
"text": alert.telegram_markdown,
|
||||
"parse_mode": "MarkdownV2",
|
||||
"disable_web_page_preview": False,
|
||||
}
|
||||
|
||||
for attempt in range(self.max_retries):
|
||||
try:
|
||||
async with httpx.AsyncClient(timeout=self.timeout) as client:
|
||||
response = await client.post(
|
||||
self._api_url,
|
||||
json=payload,
|
||||
)
|
||||
|
||||
result = response.json()
|
||||
|
||||
if result.get("ok"):
|
||||
logger.info("Telegram alert delivered successfully")
|
||||
return True
|
||||
|
||||
error_code = result.get("error_code", 0)
|
||||
description = result.get("description", "Unknown error")
|
||||
|
||||
if error_code == 429:
|
||||
# Rate limited
|
||||
retry_after = result.get("parameters", {}).get("retry_after", 1)
|
||||
logger.warning(f"Telegram rate limited, retry after {retry_after}s")
|
||||
await asyncio.sleep(retry_after)
|
||||
continue
|
||||
|
||||
logger.error(f"Telegram API error: {error_code} - {description}")
|
||||
|
||||
except httpx.TimeoutException:
|
||||
logger.warning(f"Telegram API timeout (attempt {attempt + 1})")
|
||||
except httpx.HTTPError as e:
|
||||
logger.error(f"Telegram API error: {e}")
|
||||
|
||||
# Exponential backoff
|
||||
if attempt < self.max_retries - 1:
|
||||
delay = self.retry_delay * (2**attempt)
|
||||
await asyncio.sleep(delay)
|
||||
|
||||
logger.error("Telegram delivery failed after all retries")
|
||||
return False
|
||||
@@ -0,0 +1,239 @@
|
||||
"""Alert dispatcher for multi-channel delivery."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Protocol
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from polymarket_insider_tracker.alerter.models import FormattedAlert
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class AlertChannel(Protocol):
|
||||
"""Protocol for alert delivery channels."""
|
||||
|
||||
name: str
|
||||
|
||||
async def send(self, alert: FormattedAlert) -> bool:
|
||||
"""Send alert to channel. Returns True on success."""
|
||||
...
|
||||
|
||||
|
||||
@dataclass
|
||||
class CircuitBreakerState:
|
||||
"""State for circuit breaker pattern.
|
||||
|
||||
Tracks failures and manages open/closed state for a channel.
|
||||
"""
|
||||
|
||||
failure_count: int = 0
|
||||
last_failure_time: datetime | None = None
|
||||
is_open: bool = False
|
||||
half_open_attempts: int = 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class DispatchResult:
|
||||
"""Result of dispatching an alert to all channels."""
|
||||
|
||||
success_count: int
|
||||
failure_count: int
|
||||
channel_results: dict[str, bool] = field(default_factory=dict)
|
||||
timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
|
||||
|
||||
@property
|
||||
def all_succeeded(self) -> bool:
|
||||
"""Return True if all channels succeeded."""
|
||||
return self.failure_count == 0 and self.success_count > 0
|
||||
|
||||
|
||||
class AlertDispatcher:
|
||||
"""Dispatcher for sending alerts to multiple channels.
|
||||
|
||||
Manages concurrent delivery to all configured channels with
|
||||
circuit breaker protection for failing channels.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
channels: list[AlertChannel],
|
||||
*,
|
||||
failure_threshold: int = 5,
|
||||
recovery_timeout_seconds: int = 60,
|
||||
half_open_max_attempts: int = 3,
|
||||
) -> None:
|
||||
"""Initialize the dispatcher.
|
||||
|
||||
Args:
|
||||
channels: List of alert channels to dispatch to.
|
||||
failure_threshold: Number of consecutive failures before opening circuit.
|
||||
recovery_timeout_seconds: Time to wait before half-opening circuit.
|
||||
half_open_max_attempts: Number of test attempts in half-open state.
|
||||
"""
|
||||
self.channels = channels
|
||||
self.failure_threshold = failure_threshold
|
||||
self.recovery_timeout_seconds = recovery_timeout_seconds
|
||||
self.half_open_max_attempts = half_open_max_attempts
|
||||
|
||||
# Circuit breaker state per channel
|
||||
self._circuit_state: dict[str, CircuitBreakerState] = {
|
||||
ch.name: CircuitBreakerState() for ch in channels
|
||||
}
|
||||
|
||||
def _should_attempt(self, channel_name: str) -> bool:
|
||||
"""Check if we should attempt delivery to this channel."""
|
||||
state = self._circuit_state[channel_name]
|
||||
|
||||
if not state.is_open:
|
||||
return True
|
||||
|
||||
# Check if we should try half-open
|
||||
if state.last_failure_time:
|
||||
elapsed = (datetime.now(UTC) - state.last_failure_time).total_seconds()
|
||||
if (
|
||||
elapsed >= self.recovery_timeout_seconds
|
||||
and state.half_open_attempts < self.half_open_max_attempts
|
||||
):
|
||||
# Allow half-open attempt
|
||||
logger.info(
|
||||
f"Circuit half-open for {channel_name}, "
|
||||
f"attempt {state.half_open_attempts + 1}"
|
||||
)
|
||||
return True
|
||||
|
||||
return False
|
||||
|
||||
def _record_success(self, channel_name: str) -> None:
|
||||
"""Record a successful delivery."""
|
||||
state = self._circuit_state[channel_name]
|
||||
state.failure_count = 0
|
||||
state.is_open = False
|
||||
state.half_open_attempts = 0
|
||||
state.last_failure_time = None
|
||||
logger.debug(f"Circuit closed for {channel_name}")
|
||||
|
||||
def _record_failure(self, channel_name: str) -> None:
|
||||
"""Record a failed delivery."""
|
||||
state = self._circuit_state[channel_name]
|
||||
state.failure_count += 1
|
||||
state.last_failure_time = datetime.now(UTC)
|
||||
|
||||
if state.is_open:
|
||||
# Failed during half-open, increment attempts
|
||||
state.half_open_attempts += 1
|
||||
elif state.failure_count >= self.failure_threshold:
|
||||
# Open the circuit
|
||||
state.is_open = True
|
||||
logger.warning(
|
||||
f"Circuit opened for {channel_name} after "
|
||||
f"{state.failure_count} failures"
|
||||
)
|
||||
|
||||
async def _send_to_channel(
|
||||
self, channel: AlertChannel, alert: FormattedAlert
|
||||
) -> tuple[str, bool]:
|
||||
"""Send alert to a single channel with circuit breaker."""
|
||||
channel_name = channel.name
|
||||
|
||||
if not self._should_attempt(channel_name):
|
||||
logger.debug(f"Skipping {channel_name} - circuit open")
|
||||
return (channel_name, False)
|
||||
|
||||
try:
|
||||
success = await channel.send(alert)
|
||||
if success:
|
||||
self._record_success(channel_name)
|
||||
else:
|
||||
self._record_failure(channel_name)
|
||||
return (channel_name, success)
|
||||
except Exception as e:
|
||||
logger.error(f"Error sending to {channel_name}: {e}")
|
||||
self._record_failure(channel_name)
|
||||
return (channel_name, False)
|
||||
|
||||
async def dispatch(self, alert: FormattedAlert) -> DispatchResult:
|
||||
"""Dispatch alert to all channels concurrently.
|
||||
|
||||
Args:
|
||||
alert: Formatted alert to send.
|
||||
|
||||
Returns:
|
||||
DispatchResult with per-channel status.
|
||||
"""
|
||||
if not self.channels:
|
||||
logger.warning("No channels configured for dispatch")
|
||||
return DispatchResult(success_count=0, failure_count=0)
|
||||
|
||||
# Send to all channels concurrently
|
||||
tasks = [self._send_to_channel(ch, alert) for ch in self.channels]
|
||||
results = await asyncio.gather(*tasks)
|
||||
|
||||
# Aggregate results
|
||||
channel_results = dict(results)
|
||||
success_count = sum(1 for success in channel_results.values() if success)
|
||||
failure_count = len(channel_results) - success_count
|
||||
|
||||
result = DispatchResult(
|
||||
success_count=success_count,
|
||||
failure_count=failure_count,
|
||||
channel_results=channel_results,
|
||||
)
|
||||
|
||||
logger.info(
|
||||
f"Dispatch complete: {success_count}/{len(channel_results)} succeeded"
|
||||
)
|
||||
|
||||
return result
|
||||
|
||||
async def dispatch_batch(
|
||||
self, alerts: list[FormattedAlert]
|
||||
) -> list[DispatchResult]:
|
||||
"""Dispatch multiple alerts sequentially.
|
||||
|
||||
Args:
|
||||
alerts: List of formatted alerts to send.
|
||||
|
||||
Returns:
|
||||
List of DispatchResult for each alert.
|
||||
"""
|
||||
results = []
|
||||
for alert in alerts:
|
||||
result = await self.dispatch(alert)
|
||||
results.append(result)
|
||||
return results
|
||||
|
||||
def get_circuit_status(self) -> dict[str, dict[str, object]]:
|
||||
"""Get current circuit breaker status for all channels."""
|
||||
return {
|
||||
name: {
|
||||
"is_open": state.is_open,
|
||||
"failure_count": state.failure_count,
|
||||
"half_open_attempts": state.half_open_attempts,
|
||||
"last_failure": (
|
||||
state.last_failure_time.isoformat()
|
||||
if state.last_failure_time
|
||||
else None
|
||||
),
|
||||
}
|
||||
for name, state in self._circuit_state.items()
|
||||
}
|
||||
|
||||
def reset_circuit(self, channel_name: str) -> bool:
|
||||
"""Manually reset circuit breaker for a channel.
|
||||
|
||||
Args:
|
||||
channel_name: Name of channel to reset.
|
||||
|
||||
Returns:
|
||||
True if channel was found and reset.
|
||||
"""
|
||||
if channel_name in self._circuit_state:
|
||||
self._circuit_state[channel_name] = CircuitBreakerState()
|
||||
logger.info(f"Circuit reset for {channel_name}")
|
||||
return True
|
||||
return False
|
||||
Reference in New Issue
Block a user