diff --git a/src/polymarket_insider_tracker/alerter/__init__.py b/src/polymarket_insider_tracker/alerter/__init__.py index cadaca3..93e9981 100644 --- a/src/polymarket_insider_tracker/alerter/__init__.py +++ b/src/polymarket_insider_tracker/alerter/__init__.py @@ -1 +1,9 @@ """Alerting layer - Real-time notification delivery.""" + +from polymarket_insider_tracker.alerter.formatter import AlertFormatter +from polymarket_insider_tracker.alerter.models import FormattedAlert + +__all__ = [ + "AlertFormatter", + "FormattedAlert", +] diff --git a/src/polymarket_insider_tracker/alerter/formatter.py b/src/polymarket_insider_tracker/alerter/formatter.py new file mode 100644 index 0000000..9f7f184 --- /dev/null +++ b/src/polymarket_insider_tracker/alerter/formatter.py @@ -0,0 +1,378 @@ +"""Alert message formatter for multi-channel delivery. + +This module transforms RiskAssessment objects into human-readable, +actionable alert messages optimized for Discord, Telegram, and plain text. +""" + +from __future__ import annotations + +from decimal import Decimal +from typing import Literal + +from polymarket_insider_tracker.alerter.models import FormattedAlert +from polymarket_insider_tracker.detector.models import RiskAssessment + +# Polymarket URLs +POLYMARKET_MARKET_URL = "https://polymarket.com/event/{slug}" +POLYGONSCAN_ADDRESS_URL = "https://polygonscan.com/address/{address}" + +# Discord embed colors (decimal values) +COLOR_HIGH_RISK = 15158332 # Red (#E74C3C) +COLOR_MEDIUM_RISK = 15105570 # Orange (#E67E22) +COLOR_LOW_RISK = 16776960 # Yellow (#FFFF00) + +# Risk level thresholds +HIGH_RISK_THRESHOLD = 0.7 +MEDIUM_RISK_THRESHOLD = 0.5 + + +def truncate_address(address: str, chars: int = 4) -> str: + """Truncate an Ethereum address to 0x1234...5678 format.""" + if len(address) < chars * 2 + 4: + return address + return f"{address[:chars+2]}...{address[-chars:]}" + + +def format_usdc(amount: Decimal) -> str: + """Format a USDC amount with commas and 2 decimal places.""" + return f"${amount:,.2f}" + + +def get_risk_level(score: float) -> str: + """Get human-readable risk level from score.""" + if score >= HIGH_RISK_THRESHOLD: + return "HIGH" + if score >= MEDIUM_RISK_THRESHOLD: + return "MEDIUM" + return "LOW" + + +def get_risk_color(score: float) -> int: + """Get Discord embed color based on risk score.""" + if score >= HIGH_RISK_THRESHOLD: + return COLOR_HIGH_RISK + if score >= MEDIUM_RISK_THRESHOLD: + return COLOR_MEDIUM_RISK + return COLOR_LOW_RISK + + +def get_triggered_signals(assessment: RiskAssessment) -> list[str]: + """Get list of triggered signal names.""" + signals = [] + if assessment.fresh_wallet_signal: + signals.append("Fresh Wallet") + if assessment.size_anomaly_signal: + signals.append("Large Position") + if assessment.size_anomaly_signal.is_niche_market: + signals.append("Niche Market") + return signals + + +class AlertFormatter: + """Formats RiskAssessments into multi-channel alert messages. + + Supports two verbosity levels: + - compact: Essential info only (wallet, score, market) + - detailed: Full context (all signals, links, trade details) + """ + + def __init__( + self, + verbosity: Literal["compact", "detailed"] = "detailed", + ) -> None: + """Initialize the formatter. + + Args: + verbosity: Level of detail in formatted messages. + """ + self.verbosity = verbosity + + def format(self, assessment: RiskAssessment) -> FormattedAlert: + """Format a risk assessment into a multi-channel alert. + + Args: + assessment: The risk assessment to format. + + Returns: + FormattedAlert with all channel formats. + """ + # Build common data + wallet_short = truncate_address(assessment.wallet_address) + risk_level = get_risk_level(assessment.weighted_score) + signals = get_triggered_signals(assessment) + + # Build links + links = self._build_links(assessment) + + # Build title + title = f"🚨 Suspicious Activity Detected - {risk_level} Risk" + + # Build body based on verbosity + body = self._build_body(assessment, wallet_short, risk_level, signals) + + # Build channel-specific formats + discord_embed = self._build_discord_embed( + assessment, wallet_short, risk_level, signals, links + ) + telegram_md = self._build_telegram_markdown( + assessment, wallet_short, risk_level, signals, links + ) + plain_text = self._build_plain_text( + assessment, wallet_short, risk_level, signals, links + ) + + return FormattedAlert( + title=title, + body=body, + discord_embed=discord_embed, + telegram_markdown=telegram_md, + plain_text=plain_text, + links=links, + ) + + def _build_links(self, assessment: RiskAssessment) -> dict[str, str]: + """Build dictionary of relevant links.""" + trade = assessment.trade_event + links = { + "wallet": POLYGONSCAN_ADDRESS_URL.format(address=assessment.wallet_address), + } + + # Add market link if we have the slug + if trade.market_slug: + links["market"] = POLYMARKET_MARKET_URL.format(slug=trade.market_slug) + + return links + + def _build_body( + self, + assessment: RiskAssessment, + wallet_short: str, + risk_level: str, + signals: list[str], + ) -> str: + """Build the main body text.""" + trade = assessment.trade_event + + if self.verbosity == "compact": + return ( + f"Wallet {wallet_short} made a {trade.side} trade " + f"({format_usdc(trade.notional_value)}) with risk score " + f"{assessment.weighted_score:.2f} ({risk_level})" + ) + + # Detailed body + lines = [ + f"Wallet: {wallet_short}", + f"Risk Score: {assessment.weighted_score:.2f} ({risk_level})", + f"Trade: {trade.side} {trade.outcome} @ ${trade.price:.3f}", + f"Size: {format_usdc(trade.notional_value)}", + ] + + if signals: + lines.append(f"Signals: {', '.join(signals)}") + + if trade.event_title: + lines.append(f"Market: {trade.event_title}") + + return "\n".join(lines) + + def _build_discord_embed( + self, + assessment: RiskAssessment, + wallet_short: str, + risk_level: str, + signals: list[str], + links: dict[str, str], + ) -> dict[str, object]: + """Build Discord-optimized embed format.""" + trade = assessment.trade_event + color = get_risk_color(assessment.weighted_score) + + # Get wallet age if available + wallet_age_str = "" + if assessment.fresh_wallet_signal: + age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours + if age_hours < 1: + wallet_age_str = f" (Age: {int(age_hours * 60)}m)" + else: + wallet_age_str = f" (Age: {age_hours:.0f}h)" + + fields: list[dict[str, object]] = [ + { + "name": "Wallet", + "value": f"`{wallet_short}`{wallet_age_str}", + "inline": True, + }, + { + "name": "Risk Score", + "value": f"{assessment.weighted_score:.2f} ({risk_level})", + "inline": True, + }, + ] + + # Market field + market_title = trade.event_title or trade.market_slug or "Unknown Market" + market_value = market_title + if "market" in links: + market_value = f"[{market_title}]({links['market']})" + fields.append({"name": "Market", "value": market_value, "inline": False}) + + # Trade details + trade_detail = ( + f"{trade.side} {trade.outcome} @ ${trade.price:.3f} | " + f"{format_usdc(trade.notional_value)}" + ) + fields.append({"name": "Trade", "value": trade_detail, "inline": False}) + + # Signals (if any) + if signals: + fields.append({ + "name": "Signals", + "value": ", ".join(signals), + "inline": False, + }) + + # Add detailed info for detailed verbosity + if self.verbosity == "detailed": + # Add confidence breakdown + confidences = [] + if assessment.fresh_wallet_signal: + conf = assessment.fresh_wallet_signal.confidence + confidences.append(f"Fresh Wallet: {conf:.0%}") + if assessment.size_anomaly_signal: + conf = assessment.size_anomaly_signal.confidence + confidences.append(f"Size Anomaly: {conf:.0%}") + + if confidences: + fields.append({ + "name": "Confidence", + "value": " | ".join(confidences), + "inline": False, + }) + + embed: dict[str, object] = { + "title": "🚨 Suspicious Activity Detected", + "color": color, + "fields": fields, + "footer": {"text": "Polymarket Insider Tracker"}, + } + + # Add wallet link as URL if available + if "wallet" in links: + embed["url"] = links["wallet"] + + return embed + + def _build_telegram_markdown( + self, + assessment: RiskAssessment, + wallet_short: str, + risk_level: str, + signals: list[str], + links: dict[str, str], + ) -> str: + """Build Telegram-optimized markdown format.""" + trade = assessment.trade_event + + lines = ["🚨 *Suspicious Activity Detected*", ""] + + # Wallet with link + wallet_line = f"*Wallet:* `{wallet_short}`" + if assessment.fresh_wallet_signal: + age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours + if age_hours < 1: + wallet_line += f" \\(Age: {int(age_hours * 60)}m\\)" + else: + wallet_line += f" \\(Age: {age_hours:.0f}h\\)" + lines.append(wallet_line) + + # Risk score + lines.append(f"*Risk Score:* {assessment.weighted_score:.2f} \\({risk_level}\\)") + + # Market + market_title = trade.event_title or trade.market_slug or "Unknown Market" + # Escape special Telegram markdown characters + market_title_escaped = self._escape_telegram_markdown(market_title) + if "market" in links: + lines.append(f"*Market:* [{market_title_escaped}]({links['market']})") + else: + lines.append(f"*Market:* {market_title_escaped}") + + # Trade details + usdc_value = format_usdc(trade.notional_value).replace("$", "\\$") + lines.append( + f"*Trade:* {trade.side} {trade.outcome} @ \\${trade.price:.3f} \\| {usdc_value}" + ) + + # Signals + if signals: + lines.append(f"*Signals:* {', '.join(signals)}") + + # Links + lines.append("") + if "wallet" in links: + lines.append(f"[View Wallet]({links['wallet']})") + if "market" in links: + lines.append(f"[View Market]({links['market']})") + + return "\n".join(lines) + + def _escape_telegram_markdown(self, text: str) -> str: + """Escape special Telegram MarkdownV2 characters.""" + special_chars = ["_", "*", "[", "]", "(", ")", "~", "`", ">", "#", "+", "-", "=", "|", "{", "}", ".", "!"] + for char in special_chars: + text = text.replace(char, f"\\{char}") + return text + + def _build_plain_text( + self, + assessment: RiskAssessment, + wallet_short: str, + risk_level: str, + signals: list[str], + links: dict[str, str], + ) -> str: + """Build plain text format for generic channels.""" + trade = assessment.trade_event + + lines = [ + "SUSPICIOUS ACTIVITY DETECTED", + "=" * 30, + "", + ] + + # Wallet info + wallet_line = f"Wallet: {wallet_short}" + if assessment.fresh_wallet_signal: + age_hours = assessment.fresh_wallet_signal.wallet_profile.age_hours + if age_hours < 1: + wallet_line += f" (Age: {int(age_hours * 60)}m)" + else: + wallet_line += f" (Age: {age_hours:.0f}h)" + lines.append(wallet_line) + + # Risk + lines.append(f"Risk Score: {assessment.weighted_score:.2f} ({risk_level})") + + # Market + market_title = trade.event_title or trade.market_slug or "Unknown Market" + lines.append(f"Market: {market_title}") + + # Trade + lines.append( + f"Trade: {trade.side} {trade.outcome} @ ${trade.price:.3f} | " + f"{format_usdc(trade.notional_value)}" + ) + + # Signals + if signals: + lines.append(f"Signals: {', '.join(signals)}") + + # Links + lines.append("") + if "wallet" in links: + lines.append(f"Wallet: {links['wallet']}") + if "market" in links: + lines.append(f"Market: {links['market']}") + + return "\n".join(lines) diff --git a/src/polymarket_insider_tracker/alerter/models.py b/src/polymarket_insider_tracker/alerter/models.py new file mode 100644 index 0000000..73958b7 --- /dev/null +++ b/src/polymarket_insider_tracker/alerter/models.py @@ -0,0 +1,26 @@ +"""Data models for the alerter module.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class FormattedAlert: + """A formatted alert message ready for delivery across multiple channels. + + Attributes: + title: Short alert title/headline. + body: Main alert body text. + discord_embed: Discord-optimized embed dictionary. + telegram_markdown: Telegram-formatted markdown string. + plain_text: Plain text fallback for other channels. + links: Dictionary of relevant links (e.g., market, wallet explorer). + """ + + title: str + body: str + discord_embed: dict[str, object] + telegram_markdown: str + plain_text: str + links: dict[str, str] = field(default_factory=dict) diff --git a/tests/alerter/__init__.py b/tests/alerter/__init__.py index e1c2200..3fb5f34 100644 --- a/tests/alerter/__init__.py +++ b/tests/alerter/__init__.py @@ -1 +1 @@ -"""Tests for alerter module.""" +"""Tests for the alerter module.""" diff --git a/tests/alerter/test_formatter.py b/tests/alerter/test_formatter.py new file mode 100644 index 0000000..fdb76f3 --- /dev/null +++ b/tests/alerter/test_formatter.py @@ -0,0 +1,621 @@ +"""Tests for alert message formatter.""" + +from datetime import UTC, datetime +from decimal import Decimal + +import pytest + +from polymarket_insider_tracker.alerter.formatter import ( + COLOR_HIGH_RISK, + COLOR_LOW_RISK, + COLOR_MEDIUM_RISK, + AlertFormatter, + format_usdc, + get_risk_color, + get_risk_level, + get_triggered_signals, + truncate_address, +) +from polymarket_insider_tracker.alerter.models import FormattedAlert +from polymarket_insider_tracker.detector.models import ( + FreshWalletSignal, + RiskAssessment, + SizeAnomalySignal, +) +from polymarket_insider_tracker.ingestor.models import MarketMetadata, Token, TradeEvent +from polymarket_insider_tracker.profiler.models import WalletProfile + +# ============================================================================ +# Fixtures +# ============================================================================ + + +@pytest.fixture +def sample_trade() -> TradeEvent: + """Create a sample trade event.""" + return TradeEvent( + market_id="market_abc123", + trade_id="tx_001", + wallet_address="0x1234567890abcdef1234567890abcdef12345678", + side="BUY", + outcome="Yes", + outcome_index=0, + price=Decimal("0.075"), + size=Decimal("200000"), + timestamp=datetime.now(UTC), + asset_id="token_123", + market_slug="will-x-happen", + event_title="Will X happen by Y?", + ) + + +@pytest.fixture +def sample_wallet_profile() -> WalletProfile: + """Create a sample wallet profile.""" + return WalletProfile( + address="0x1234567890abcdef1234567890abcdef12345678", + nonce=2, + first_seen=datetime.now(UTC), + age_hours=2.0, + is_fresh=True, + total_tx_count=2, + matic_balance=Decimal("1000000000000000000"), + usdc_balance=Decimal("1000000"), + ) + + +@pytest.fixture +def sample_metadata() -> MarketMetadata: + """Create sample market metadata.""" + return MarketMetadata( + condition_id="market_abc123", + question="Will X happen by Y?", + description="Test market description", + tokens=(Token(token_id="token_123", outcome="Yes", price=Decimal("0.075")),), + category="other", + ) + + +@pytest.fixture +def fresh_wallet_signal( + sample_trade: TradeEvent, sample_wallet_profile: WalletProfile +) -> FreshWalletSignal: + """Create a sample fresh wallet signal.""" + return FreshWalletSignal( + trade_event=sample_trade, + wallet_profile=sample_wallet_profile, + confidence=0.8, + factors={"base": 0.5, "brand_new_bonus": 0.2, "large_trade_bonus": 0.1}, + ) + + +@pytest.fixture +def size_anomaly_signal( + sample_trade: TradeEvent, sample_metadata: MarketMetadata +) -> SizeAnomalySignal: + """Create a sample size anomaly signal.""" + return SizeAnomalySignal( + trade_event=sample_trade, + market_metadata=sample_metadata, + volume_impact=0.10, + book_impact=0.15, + is_niche_market=True, + confidence=0.7, + factors={"volume_impact": 0.4, "book_impact": 0.3}, + ) + + +@pytest.fixture +def high_risk_assessment( + sample_trade: TradeEvent, + fresh_wallet_signal: FreshWalletSignal, + size_anomaly_signal: SizeAnomalySignal, +) -> RiskAssessment: + """Create a high-risk assessment with multiple signals.""" + return RiskAssessment( + trade_event=sample_trade, + wallet_address=sample_trade.wallet_address, + market_id=sample_trade.market_id, + fresh_wallet_signal=fresh_wallet_signal, + size_anomaly_signal=size_anomaly_signal, + signals_triggered=2, + weighted_score=0.82, + should_alert=True, + ) + + +@pytest.fixture +def medium_risk_assessment( + sample_trade: TradeEvent, + fresh_wallet_signal: FreshWalletSignal, +) -> RiskAssessment: + """Create a medium-risk assessment with one signal.""" + return RiskAssessment( + trade_event=sample_trade, + wallet_address=sample_trade.wallet_address, + market_id=sample_trade.market_id, + fresh_wallet_signal=fresh_wallet_signal, + size_anomaly_signal=None, + signals_triggered=1, + weighted_score=0.55, + should_alert=True, + ) + + +@pytest.fixture +def low_risk_assessment(sample_trade: TradeEvent) -> RiskAssessment: + """Create a low-risk assessment with no signals.""" + return RiskAssessment( + trade_event=sample_trade, + wallet_address=sample_trade.wallet_address, + market_id=sample_trade.market_id, + fresh_wallet_signal=None, + size_anomaly_signal=None, + signals_triggered=0, + weighted_score=0.25, + should_alert=False, + ) + + +# ============================================================================ +# Helper Function Tests +# ============================================================================ + + +class TestTruncateAddress: + """Tests for truncate_address helper.""" + + def test_truncate_standard_address(self) -> None: + """Test truncating a standard Ethereum address.""" + address = "0x1234567890abcdef1234567890abcdef12345678" + result = truncate_address(address) + assert result == "0x1234...5678" + + def test_truncate_with_custom_length(self) -> None: + """Test truncating with custom character count.""" + address = "0x1234567890abcdef1234567890abcdef12345678" + result = truncate_address(address, chars=6) + assert result == "0x123456...345678" + + def test_short_address_not_truncated(self) -> None: + """Test that short addresses are not truncated.""" + address = "0x1234" + result = truncate_address(address) + assert result == "0x1234" + + +class TestFormatUsdc: + """Tests for format_usdc helper.""" + + def test_format_whole_dollars(self) -> None: + """Test formatting whole dollar amounts.""" + result = format_usdc(Decimal("15000")) + assert result == "$15,000.00" + + def test_format_with_cents(self) -> None: + """Test formatting with decimal places.""" + result = format_usdc(Decimal("1234.56")) + assert result == "$1,234.56" + + def test_format_large_amount(self) -> None: + """Test formatting large amounts.""" + result = format_usdc(Decimal("1000000")) + assert result == "$1,000,000.00" + + +class TestGetRiskLevel: + """Tests for get_risk_level helper.""" + + def test_high_risk(self) -> None: + """Test high risk threshold.""" + assert get_risk_level(0.85) == "HIGH" + assert get_risk_level(0.70) == "HIGH" + + def test_medium_risk(self) -> None: + """Test medium risk threshold.""" + assert get_risk_level(0.65) == "MEDIUM" + assert get_risk_level(0.50) == "MEDIUM" + + def test_low_risk(self) -> None: + """Test low risk threshold.""" + assert get_risk_level(0.40) == "LOW" + assert get_risk_level(0.10) == "LOW" + + +class TestGetRiskColor: + """Tests for get_risk_color helper.""" + + def test_high_risk_color(self) -> None: + """Test high risk returns red color.""" + assert get_risk_color(0.85) == COLOR_HIGH_RISK + assert get_risk_color(0.70) == COLOR_HIGH_RISK + + def test_medium_risk_color(self) -> None: + """Test medium risk returns orange color.""" + assert get_risk_color(0.65) == COLOR_MEDIUM_RISK + assert get_risk_color(0.50) == COLOR_MEDIUM_RISK + + def test_low_risk_color(self) -> None: + """Test low risk returns yellow color.""" + assert get_risk_color(0.40) == COLOR_LOW_RISK + + +class TestGetTriggeredSignals: + """Tests for get_triggered_signals helper.""" + + def test_no_signals(self, low_risk_assessment: RiskAssessment) -> None: + """Test assessment with no signals.""" + signals = get_triggered_signals(low_risk_assessment) + assert signals == [] + + def test_fresh_wallet_only(self, medium_risk_assessment: RiskAssessment) -> None: + """Test assessment with only fresh wallet signal.""" + signals = get_triggered_signals(medium_risk_assessment) + assert "Fresh Wallet" in signals + assert "Large Position" not in signals + + def test_both_signals(self, high_risk_assessment: RiskAssessment) -> None: + """Test assessment with both signals.""" + signals = get_triggered_signals(high_risk_assessment) + assert "Fresh Wallet" in signals + assert "Large Position" in signals + assert "Niche Market" in signals # From size anomaly with is_niche_market=True + + +# ============================================================================ +# AlertFormatter Tests +# ============================================================================ + + +class TestAlertFormatterInit: + """Tests for AlertFormatter initialization.""" + + def test_default_verbosity(self) -> None: + """Test default verbosity is detailed.""" + formatter = AlertFormatter() + assert formatter.verbosity == "detailed" + + def test_compact_verbosity(self) -> None: + """Test setting compact verbosity.""" + formatter = AlertFormatter(verbosity="compact") + assert formatter.verbosity == "compact" + + +class TestAlertFormatterFormat: + """Tests for AlertFormatter.format method.""" + + def test_format_returns_formatted_alert( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that format returns a FormattedAlert.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert isinstance(result, FormattedAlert) + + def test_format_includes_all_fields( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that all fields are populated.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + + assert result.title != "" + assert result.body != "" + assert result.discord_embed != {} + assert result.telegram_markdown != "" + assert result.plain_text != "" + assert result.links != {} + + def test_format_title_includes_risk_level( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that title includes risk level.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "HIGH" in result.title + + def test_format_includes_wallet_link( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that wallet explorer link is included.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "wallet" in result.links + assert "polygonscan.com" in result.links["wallet"] + + def test_format_includes_market_link( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that market link is included when slug available.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "market" in result.links + assert "polymarket.com" in result.links["market"] + + +class TestDiscordEmbed: + """Tests for Discord embed format.""" + + def test_embed_has_required_fields( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that embed has required Discord fields.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + embed = result.discord_embed + + assert "title" in embed + assert "color" in embed + assert "fields" in embed + assert "footer" in embed + + def test_embed_color_reflects_risk( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that embed color matches risk level.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert result.discord_embed["color"] == COLOR_HIGH_RISK + + def test_embed_includes_wallet_field( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that embed includes wallet field.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + fields = result.discord_embed["fields"] + + wallet_field = next((f for f in fields if f["name"] == "Wallet"), None) + assert wallet_field is not None + assert "0x1234" in wallet_field["value"] + + def test_embed_includes_wallet_age( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that wallet age is shown when fresh wallet signal present.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + fields = result.discord_embed["fields"] + + wallet_field = next((f for f in fields if f["name"] == "Wallet"), None) + assert "Age:" in wallet_field["value"] + + def test_embed_includes_trade_details( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that trade details are in embed.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + fields = result.discord_embed["fields"] + + trade_field = next((f for f in fields if f["name"] == "Trade"), None) + assert trade_field is not None + assert "BUY" in trade_field["value"] + assert "Yes" in trade_field["value"] + + def test_embed_includes_signals_field( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that signals are listed in embed.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + fields = result.discord_embed["fields"] + + signals_field = next((f for f in fields if f["name"] == "Signals"), None) + assert signals_field is not None + assert "Fresh Wallet" in signals_field["value"] + + def test_detailed_embed_includes_confidence( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that detailed mode includes confidence breakdown.""" + formatter = AlertFormatter(verbosity="detailed") + result = formatter.format(high_risk_assessment) + fields = result.discord_embed["fields"] + + conf_field = next((f for f in fields if f["name"] == "Confidence"), None) + assert conf_field is not None + + +class TestTelegramMarkdown: + """Tests for Telegram markdown format.""" + + def test_telegram_includes_header( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that Telegram message has header.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "*Suspicious Activity Detected*" in result.telegram_markdown + + def test_telegram_includes_wallet( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that Telegram message includes wallet.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "`0x1234...5678`" in result.telegram_markdown + + def test_telegram_includes_risk_score( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that Telegram message includes risk score.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "0.82" in result.telegram_markdown + assert "HIGH" in result.telegram_markdown + + def test_telegram_includes_links( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that Telegram message includes links.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "[View Wallet]" in result.telegram_markdown + assert "[View Market]" in result.telegram_markdown + + +class TestPlainText: + """Tests for plain text format.""" + + def test_plain_text_header(self, high_risk_assessment: RiskAssessment) -> None: + """Test plain text has header.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "SUSPICIOUS ACTIVITY DETECTED" in result.plain_text + + def test_plain_text_wallet(self, high_risk_assessment: RiskAssessment) -> None: + """Test plain text includes wallet.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "Wallet:" in result.plain_text + assert "0x1234...5678" in result.plain_text + + def test_plain_text_trade(self, high_risk_assessment: RiskAssessment) -> None: + """Test plain text includes trade details.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "Trade:" in result.plain_text + assert "BUY" in result.plain_text + + def test_plain_text_signals(self, high_risk_assessment: RiskAssessment) -> None: + """Test plain text includes signals.""" + formatter = AlertFormatter() + result = formatter.format(high_risk_assessment) + assert "Signals:" in result.plain_text + assert "Fresh Wallet" in result.plain_text + + +class TestCompactVerbosity: + """Tests for compact verbosity mode.""" + + def test_compact_body_is_shorter( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that compact mode produces shorter body.""" + detailed_formatter = AlertFormatter(verbosity="detailed") + compact_formatter = AlertFormatter(verbosity="compact") + + detailed_result = detailed_formatter.format(high_risk_assessment) + compact_result = compact_formatter.format(high_risk_assessment) + + assert len(compact_result.body) < len(detailed_result.body) + + def test_compact_body_includes_essential_info( + self, high_risk_assessment: RiskAssessment + ) -> None: + """Test that compact mode still has essential info.""" + formatter = AlertFormatter(verbosity="compact") + result = formatter.format(high_risk_assessment) + + assert "0x1234...5678" in result.body + assert "0.82" in result.body + assert "HIGH" in result.body + + +class TestEdgeCases: + """Tests for edge cases and special scenarios.""" + + def test_no_market_slug(self, sample_trade: TradeEvent) -> None: + """Test formatting when market slug is empty.""" + trade = TradeEvent( + market_id=sample_trade.market_id, + trade_id=sample_trade.trade_id, + wallet_address=sample_trade.wallet_address, + side=sample_trade.side, + outcome=sample_trade.outcome, + outcome_index=sample_trade.outcome_index, + price=sample_trade.price, + size=sample_trade.size, + timestamp=sample_trade.timestamp, + asset_id=sample_trade.asset_id, + market_slug="", # Empty slug + event_title="", # Empty title + ) + assessment = RiskAssessment( + trade_event=trade, + wallet_address=trade.wallet_address, + market_id=trade.market_id, + fresh_wallet_signal=None, + size_anomaly_signal=None, + signals_triggered=0, + weighted_score=0.5, + should_alert=False, + ) + + formatter = AlertFormatter() + result = formatter.format(assessment) + + # Should not have market link + assert "market" not in result.links + # Should have fallback text + assert "Unknown Market" in result.plain_text + + def test_very_short_wallet_age( + self, + sample_trade: TradeEvent, + sample_wallet_profile: WalletProfile, + ) -> None: + """Test formatting with wallet age less than 1 hour.""" + profile = WalletProfile( + address=sample_wallet_profile.address, + nonce=sample_wallet_profile.nonce, + first_seen=datetime.now(UTC), + age_hours=0.5, # 30 minutes + is_fresh=True, + total_tx_count=1, + matic_balance=sample_wallet_profile.matic_balance, + usdc_balance=sample_wallet_profile.usdc_balance, + ) + signal = FreshWalletSignal( + trade_event=sample_trade, + wallet_profile=profile, + confidence=0.9, + factors={}, + ) + assessment = RiskAssessment( + trade_event=sample_trade, + wallet_address=sample_trade.wallet_address, + market_id=sample_trade.market_id, + fresh_wallet_signal=signal, + size_anomaly_signal=None, + signals_triggered=1, + weighted_score=0.75, + should_alert=True, + ) + + formatter = AlertFormatter() + result = formatter.format(assessment) + + # Should show age in minutes + assert "30m" in result.plain_text or "Age: 30m" in result.plain_text + + def test_size_anomaly_without_niche( + self, + sample_trade: TradeEvent, + sample_metadata: MarketMetadata, + ) -> None: + """Test size anomaly signal without niche market flag.""" + signal = SizeAnomalySignal( + trade_event=sample_trade, + market_metadata=sample_metadata, + volume_impact=0.10, + book_impact=0.15, + is_niche_market=False, # Not niche + confidence=0.7, + factors={}, + ) + assessment = RiskAssessment( + trade_event=sample_trade, + wallet_address=sample_trade.wallet_address, + market_id=sample_trade.market_id, + fresh_wallet_signal=None, + size_anomaly_signal=signal, + signals_triggered=1, + weighted_score=0.6, + should_alert=True, + ) + + signals = get_triggered_signals(assessment) + assert "Large Position" in signals + assert "Niche Market" not in signals