feat: implement fresh wallet detection algorithm (#14)

Add FreshWalletDetector class that identifies trades from fresh wallets
and generates alert signals with confidence scores.

Features:
- FreshWalletSignal dataclass with trade/wallet data and confidence
- Configurable thresholds for nonce, age, and minimum trade size
- Confidence scoring based on wallet freshness and trade characteristics
- Batch analysis support with parallel processing
- Integration with WalletAnalyzer from profiler module

Confidence scoring:
- Base: 0.5 (fresh wallet detected)
- +0.2 if nonce == 0 (brand new)
- +0.1 if age < 2 hours (very young)
- +0.1 if trade size > $10,000 (large trade)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,118 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+*.py[cod]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# PyInstaller
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py,cover
+.hypothesis/
+.pytest_cache/
+
+# Translations
+*.mo
+*.pot
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+.python-version
+
+# PEP 582
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# Environment
+.env
+.env.local
+.venv
+env/
+venv/
+ENV/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# Ruff
+.ruff_cache/
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
+*~
+
+# OS
+.DS_Store
+Thumbs.db
+
+# Local development
+*.local
+.direnv/

.pre-commit-config.yaml

@@ -0,0 +1,23 @@
+repos:
+  - repo: https://github.com/pre-commit/pre-commit-hooks
+    rev: v4.5.0
+    hooks:
+      - id: trailing-whitespace
+      - id: end-of-file-fixer
+      - id: check-yaml
+      - id: check-added-large-files
+      - id: check-merge-conflict
+
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    rev: v0.1.9
+    hooks:
+      - id: ruff
+        args: [--fix]
+      - id: ruff-format
+
+  - repo: https://github.com/pre-commit/mirrors-mypy
+    rev: v1.8.0
+    hooks:
+      - id: mypy
+        additional_dependencies:
+          - pydantic>=2.0.0

pyproject.toml

@@ -0,0 +1,99 @@
+[project]
+name = "polymarket-insider-tracker"
+version = "0.1.0"
+description = "Detect insider activity on Polymarket"
+readme = "README.md"
+requires-python = ">=3.11"
+dependencies = [
+    "py-clob-client>=0.1.0",
+    "web3>=6.0.0",
+    "httpx>=0.25.0",
+    "redis>=5.0.0",
+    "sqlalchemy>=2.0.0",
+    "alembic>=1.13.0",
+    "pydantic>=2.0.0",
+    "python-dotenv>=1.0.0",
+    "websockets>=12.0",
+    "prometheus-client>=0.19.0",
+    "aiohttp>=3.9.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=7.0.0",
+    "pytest-asyncio>=0.21.0",
+    "pytest-cov>=4.0.0",
+    "ruff>=0.1.0",
+    "mypy>=1.7.0",
+    "pre-commit>=3.0.0",
+]
+
+[build-system]
+requires = ["setuptools>=61.0"]
+build-backend = "setuptools.build_meta"
+
+[tool.setuptools.packages.find]
+where = ["src"]
+
+[tool.ruff]
+line-length = 100
+target-version = "py311"
+src = ["src", "tests"]
+
+[tool.ruff.lint]
+select = [
+    "E",      # pycodestyle errors
+    "W",      # pycodestyle warnings
+    "F",      # Pyflakes
+    "I",      # isort
+    "B",      # flake8-bugbear
+    "C4",     # flake8-comprehensions
+    "UP",     # pyupgrade
+    "ARG",    # flake8-unused-arguments
+    "SIM",    # flake8-simplify
+]
+ignore = [
+    "E501",   # line too long (handled by formatter)
+]
+
+[tool.ruff.lint.isort]
+known-first-party = ["polymarket_insider_tracker"]
+
+[tool.mypy]
+python_version = "3.11"
+strict = true
+warn_return_any = true
+warn_unused_ignores = true
+disallow_untyped_defs = true
+disallow_incomplete_defs = true
+check_untyped_defs = true
+disallow_untyped_decorators = true
+no_implicit_optional = true
+warn_redundant_casts = true
+warn_unused_configs = true
+plugins = ["pydantic.mypy"]
+
+[[tool.mypy.overrides]]
+module = ["py_clob_client.*", "web3.*", "redis.*"]
+ignore_missing_imports = true
+
+[tool.pytest.ini_options]
+asyncio_mode = "auto"
+testpaths = ["tests"]
+addopts = "-v --tb=short"
+markers = [
+    "slow: marks tests as slow (deselect with '-m \"not slow\"')",
+    "integration: marks tests as integration tests",
+]
+
+[tool.coverage.run]
+source = ["src"]
+branch = true
+
+[tool.coverage.report]
+exclude_lines = [
+    "pragma: no cover",
+    "def __repr__",
+    "raise NotImplementedError",
+    "if TYPE_CHECKING:",
+]

src/polymarket_insider_tracker/__init__.py

@@ -0,0 +1,3 @@
+"""Polymarket Insider Tracker - Detect insider activity on Polymarket."""
+
+__version__ = "0.1.0"

src/polymarket_insider_tracker/alerter/__init__.py

@@ -0,0 +1 @@
+"""Alerting layer - Real-time notification delivery."""

src/polymarket_insider_tracker/detector/__init__.py

@@ -0,0 +1,6 @@
+"""Anomaly detection layer - Suspicious activity identification."""
+
+from polymarket_insider_tracker.detector.fresh_wallet import FreshWalletDetector
+from polymarket_insider_tracker.detector.models import FreshWalletSignal
+
+__all__ = ["FreshWalletDetector", "FreshWalletSignal"]

src/polymarket_insider_tracker/detector/fresh_wallet.py

@@ -0,0 +1,238 @@
+"""Fresh wallet detection algorithm.
+
+This module provides the FreshWalletDetector class that identifies trades
+from fresh wallets and generates alert signals with confidence scores.
+"""
+
+import logging
+from decimal import Decimal
+
+from polymarket_insider_tracker.detector.models import FreshWalletSignal
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.profiler.analyzer import WalletAnalyzer
+from polymarket_insider_tracker.profiler.models import WalletProfile
+
+logger = logging.getLogger(__name__)
+
+# Default configuration
+DEFAULT_MIN_TRADE_SIZE = Decimal("1000")  # $1,000 minimum trade size
+DEFAULT_MAX_NONCE = 5  # Max nonce to be considered fresh
+DEFAULT_MAX_AGE_HOURS = 48.0  # Max age in hours to be considered fresh
+
+# Confidence scoring constants
+BASE_CONFIDENCE = 0.5
+BRAND_NEW_BONUS = 0.2  # nonce == 0
+VERY_YOUNG_BONUS = 0.1  # age < 2 hours
+LARGE_TRADE_BONUS = 0.1  # trade size > $10,000
+LARGE_TRADE_THRESHOLD = Decimal("10000")
+
+
+class FreshWalletDetector:
+    """Detector for fresh wallet trading patterns.
+
+    This detector analyzes trade events for fresh wallet signals. A trade
+    is flagged as suspicious if:
+    1. The wallet meets freshness criteria (low nonce, recent activity)
+    2. The trade size meets minimum threshold
+
+    The detector produces confidence scores based on multiple factors:
+    - Wallet nonce (brand new = higher confidence)
+    - Wallet age (very young = higher confidence)
+    - Trade size (larger = higher confidence)
+
+    Example:
+        ```python
+        analyzer = WalletAnalyzer(polygon_client, redis=redis)
+        detector = FreshWalletDetector(analyzer)
+
+        # Analyze a single trade
+        signal = await detector.analyze(trade_event)
+        if signal is not None:
+            print(f"Fresh wallet detected! Confidence: {signal.confidence}")
+        ```
+    """
+
+    def __init__(
+        self,
+        wallet_analyzer: WalletAnalyzer,
+        *,
+        min_trade_size: Decimal = DEFAULT_MIN_TRADE_SIZE,
+        max_nonce: int = DEFAULT_MAX_NONCE,
+        max_age_hours: float = DEFAULT_MAX_AGE_HOURS,
+    ) -> None:
+        """Initialize the fresh wallet detector.
+
+        Args:
+            wallet_analyzer: WalletAnalyzer instance for wallet profiling.
+            min_trade_size: Minimum trade size to analyze (default $1,000).
+            max_nonce: Maximum nonce to consider wallet fresh (default 5).
+            max_age_hours: Maximum age in hours to consider fresh (default 48).
+        """
+        self._analyzer = wallet_analyzer
+        self._min_trade_size = min_trade_size
+        self._max_nonce = max_nonce
+        self._max_age_hours = max_age_hours
+
+    async def analyze(self, trade: TradeEvent) -> FreshWalletSignal | None:
+        """Analyze a trade event for fresh wallet signals.
+
+        This method:
+        1. Filters out trades below the minimum size threshold
+        2. Analyzes the trader's wallet profile
+        3. Determines if the wallet is fresh
+        4. Calculates confidence score based on multiple factors
+
+        Args:
+            trade: TradeEvent to analyze.
+
+        Returns:
+            FreshWalletSignal if the trade is from a fresh wallet,
+            None otherwise.
+        """
+        # Filter by minimum trade size
+        if trade.notional_value < self._min_trade_size:
+            logger.debug(
+                "Trade %s below minimum size: %s < %s",
+                trade.trade_id,
+                trade.notional_value,
+                self._min_trade_size,
+            )
+            return None
+
+        # Get wallet profile
+        try:
+            profile = await self._analyzer.analyze(trade.wallet_address)
+        except Exception as e:
+            logger.warning(
+                "Failed to analyze wallet %s for trade %s: %s",
+                trade.wallet_address,
+                trade.trade_id,
+                e,
+            )
+            return None
+
+        # Check if wallet is fresh
+        if not self._is_wallet_fresh(profile):
+            logger.debug(
+                "Wallet %s is not fresh (nonce=%d, age=%s)",
+                trade.wallet_address,
+                profile.nonce,
+                profile.age_hours,
+            )
+            return None
+
+        # Calculate confidence score
+        confidence, factors = self.calculate_confidence(profile, trade)
+
+        logger.info(
+            "Fresh wallet signal: wallet=%s, market=%s, size=%s, confidence=%.2f",
+            trade.wallet_address[:10] + "...",
+            trade.market_id[:10] + "...",
+            trade.notional_value,
+            confidence,
+        )
+
+        return FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=confidence,
+            factors=factors,
+        )
+
+    def _is_wallet_fresh(self, profile: WalletProfile) -> bool:
+        """Check if wallet meets freshness criteria.
+
+        A wallet is considered fresh if:
+        1. Nonce is at or below max_nonce threshold
+        2. Age is unknown OR within max_age_hours
+
+        Args:
+            profile: Wallet profile to check.
+
+        Returns:
+            True if wallet is fresh.
+        """
+        # Must have few transactions
+        if profile.nonce > self._max_nonce:
+            return False
+
+        # If age is known, must be recent
+        return not (profile.age_hours is not None and profile.age_hours > self._max_age_hours)
+
+    def calculate_confidence(
+        self,
+        profile: WalletProfile,
+        trade: TradeEvent,
+    ) -> tuple[float, dict[str, float]]:
+        """Calculate confidence score based on multiple factors.
+
+        Confidence scoring:
+        - Base: 0.5 (fresh wallet detected)
+        - +0.2 if nonce == 0 (brand new wallet)
+        - +0.1 if age < 2 hours (very young)
+        - +0.1 if trade size > $10,000 (large trade)
+
+        Final confidence is clamped to [0.0, 1.0].
+
+        Args:
+            profile: Wallet profile with nonce and age data.
+            trade: Trade event with size data.
+
+        Returns:
+            Tuple of (confidence_score, factors_dict).
+        """
+        factors: dict[str, float] = {"base": BASE_CONFIDENCE}
+        confidence = BASE_CONFIDENCE
+
+        # Brand new wallet bonus
+        if profile.nonce == 0:
+            factors["brand_new"] = BRAND_NEW_BONUS
+            confidence += BRAND_NEW_BONUS
+
+        # Very young wallet bonus
+        if profile.age_hours is not None and profile.age_hours < 2.0:
+            factors["very_young"] = VERY_YOUNG_BONUS
+            confidence += VERY_YOUNG_BONUS
+
+        # Large trade bonus
+        if trade.notional_value > LARGE_TRADE_THRESHOLD:
+            factors["large_trade"] = LARGE_TRADE_BONUS
+            confidence += LARGE_TRADE_BONUS
+
+        # Clamp to valid range
+        confidence = max(0.0, min(1.0, confidence))
+
+        return confidence, factors
+
+    async def analyze_batch(
+        self,
+        trades: list[TradeEvent],
+    ) -> list[FreshWalletSignal]:
+        """Analyze multiple trades for fresh wallet signals.
+
+        Processes trades in parallel for efficiency.
+
+        Args:
+            trades: List of trades to analyze.
+
+        Returns:
+            List of FreshWalletSignal for trades from fresh wallets.
+        """
+        import asyncio
+
+        tasks = [self.analyze(trade) for trade in trades]
+        results = await asyncio.gather(*tasks, return_exceptions=True)
+
+        signals: list[FreshWalletSignal] = []
+        for trade, result in zip(trades, results, strict=True):
+            if isinstance(result, BaseException):
+                logger.warning(
+                    "Failed to analyze trade %s: %s",
+                    trade.trade_id,
+                    result,
+                )
+                continue
+            if result is not None:
+                signals.append(result)
+
+        return signals

src/polymarket_insider_tracker/detector/models.py

@@ -0,0 +1,73 @@
+"""Data models for the detector module."""
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from decimal import Decimal
+
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.profiler.models import WalletProfile
+
+
+@dataclass(frozen=True)
+class FreshWalletSignal:
+    """Signal emitted when a fresh wallet makes a suspicious trade.
+
+    This signal combines trade event data with wallet profile analysis
+    to produce a confidence score indicating the likelihood of suspicious
+    activity.
+
+    Attributes:
+        trade_event: The original trade event that triggered this signal.
+        wallet_profile: Analyzed profile of the trader's wallet.
+        confidence: Overall confidence score (0.0 to 1.0).
+        factors: Individual factor scores contributing to confidence.
+        timestamp: When this signal was generated.
+    """
+
+    trade_event: TradeEvent
+    wallet_profile: WalletProfile
+    confidence: float
+    factors: dict[str, float]
+    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+    @property
+    def wallet_address(self) -> str:
+        """Return the wallet address from the trade event."""
+        return self.trade_event.wallet_address
+
+    @property
+    def market_id(self) -> str:
+        """Return the market ID from the trade event."""
+        return self.trade_event.market_id
+
+    @property
+    def trade_size_usdc(self) -> Decimal:
+        """Return the trade size in USDC (notional value)."""
+        return self.trade_event.notional_value
+
+    @property
+    def is_high_confidence(self) -> bool:
+        """Return True if confidence exceeds 0.7."""
+        return self.confidence >= 0.7
+
+    @property
+    def is_very_high_confidence(self) -> bool:
+        """Return True if confidence exceeds 0.85."""
+        return self.confidence >= 0.85
+
+    def to_dict(self) -> dict[str, object]:
+        """Serialize to dictionary for Redis stream publishing."""
+        return {
+            "wallet_address": self.wallet_address,
+            "market_id": self.market_id,
+            "trade_id": self.trade_event.trade_id,
+            "trade_size": str(self.trade_size_usdc),
+            "trade_side": self.trade_event.side,
+            "trade_price": str(self.trade_event.price),
+            "wallet_nonce": self.wallet_profile.nonce,
+            "wallet_age_hours": self.wallet_profile.age_hours,
+            "wallet_is_fresh": self.wallet_profile.is_fresh,
+            "confidence": self.confidence,
+            "factors": self.factors,
+            "timestamp": self.timestamp.isoformat(),
+        }

src/polymarket_insider_tracker/ingestor/__init__.py

@@ -0,0 +1,77 @@
+"""Data ingestion layer - Real-time Polymarket trade streaming."""
+
+from polymarket_insider_tracker.ingestor.clob_client import (
+    ClobClient,
+    ClobClientError,
+    RetryError,
+)
+from polymarket_insider_tracker.ingestor.health import (
+    HealthMonitor,
+    HealthReport,
+    HealthStatus,
+    StreamHealth,
+    StreamStatus,
+)
+from polymarket_insider_tracker.ingestor.metadata_sync import (
+    MarketMetadataSync,
+    MetadataSyncError,
+    SyncState,
+    SyncStats,
+)
+from polymarket_insider_tracker.ingestor.models import (
+    Market,
+    MarketMetadata,
+    Orderbook,
+    OrderbookLevel,
+    Token,
+    TradeEvent,
+    derive_category,
+)
+from polymarket_insider_tracker.ingestor.publisher import (
+    ConsumerGroupExistsError,
+    EventPublisher,
+    PublisherError,
+    StreamEntry,
+)
+from polymarket_insider_tracker.ingestor.websocket import (
+    ConnectionState,
+    StreamStats as WebSocketStreamStats,
+    TradeStreamError,
+    TradeStreamHandler,
+)
+
+__all__ = [
+    # CLOB Client
+    "ClobClient",
+    "ClobClientError",
+    "RetryError",
+    # Health Monitor
+    "HealthMonitor",
+    "HealthReport",
+    "HealthStatus",
+    "StreamHealth",
+    "StreamStatus",
+    # Metadata Sync
+    "MarketMetadataSync",
+    "MetadataSyncError",
+    "SyncState",
+    "SyncStats",
+    # Models
+    "Market",
+    "MarketMetadata",
+    "Orderbook",
+    "OrderbookLevel",
+    "Token",
+    "TradeEvent",
+    "derive_category",
+    # Publisher
+    "ConsumerGroupExistsError",
+    "EventPublisher",
+    "PublisherError",
+    "StreamEntry",
+    # WebSocket
+    "ConnectionState",
+    "WebSocketStreamStats",
+    "TradeStreamError",
+    "TradeStreamHandler",
+]

src/polymarket_insider_tracker/ingestor/clob_client.py

@@ -0,0 +1,340 @@
+"""Wrapper around py-clob-client with rate limiting and retry logic."""
+
+import asyncio
+import logging
+import os
+import time
+from collections.abc import Callable
+from functools import wraps
+from typing import Any, ParamSpec, TypeVar
+
+from py_clob_client.client import ClobClient as BaseClobClient
+from py_clob_client.clob_types import BookParams
+
+from polymarket_insider_tracker.ingestor.models import Market, Orderbook
+
+logger = logging.getLogger(__name__)
+
+P = ParamSpec("P")
+T = TypeVar("T")
+
+# Constants
+DEFAULT_HOST = "https://clob.polymarket.com"
+MAX_REQUESTS_PER_SECOND = 10
+MIN_REQUEST_INTERVAL = 1.0 / MAX_REQUESTS_PER_SECOND  # 0.1 seconds
+
+DEFAULT_MAX_RETRIES = 3
+DEFAULT_RETRY_BASE_DELAY = 1.0
+RETRY_STATUS_CODES = (429, 500, 502, 503, 504)
+
+
+class RateLimiter:
+    """Token bucket rate limiter for API requests."""
+
+    def __init__(self, max_requests_per_second: float = MAX_REQUESTS_PER_SECOND) -> None:
+        """Initialize the rate limiter.
+
+        Args:
+            max_requests_per_second: Maximum requests allowed per second.
+        """
+        self._min_interval = 1.0 / max_requests_per_second
+        self._last_request_time: float = 0.0
+        self._lock = asyncio.Lock()
+
+    async def acquire(self) -> None:
+        """Wait until a request slot is available."""
+        async with self._lock:
+            now = time.monotonic()
+            elapsed = now - self._last_request_time
+            if elapsed < self._min_interval:
+                wait_time = self._min_interval - elapsed
+                await asyncio.sleep(wait_time)
+            self._last_request_time = time.monotonic()
+
+    def acquire_sync(self) -> None:
+        """Synchronous version of acquire for sync operations."""
+        now = time.monotonic()
+        elapsed = now - self._last_request_time
+        if elapsed < self._min_interval:
+            wait_time = self._min_interval - elapsed
+            time.sleep(wait_time)
+        self._last_request_time = time.monotonic()
+
+
+class RetryError(Exception):
+    """Raised when all retry attempts are exhausted."""
+
+    def __init__(self, message: str, last_exception: Exception | None = None) -> None:
+        super().__init__(message)
+        self.last_exception = last_exception
+
+
+def with_retry(
+    max_retries: int = DEFAULT_MAX_RETRIES,
+    base_delay: float = DEFAULT_RETRY_BASE_DELAY,
+    retry_on: tuple[type[Exception], ...] = (Exception,),
+) -> Callable[[Callable[P, T]], Callable[P, T]]:
+    """Decorator for adding retry logic with exponential backoff.
+
+    Args:
+        max_retries: Maximum number of retry attempts.
+        base_delay: Base delay in seconds (doubles with each retry).
+        retry_on: Tuple of exception types to retry on.
+
+    Returns:
+        Decorated function with retry logic.
+    """
+
+    def decorator(func: Callable[P, T]) -> Callable[P, T]:
+        @wraps(func)
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+            last_exception: Exception | None = None
+
+            for attempt in range(max_retries + 1):
+                try:
+                    return func(*args, **kwargs)
+                except retry_on as e:
+                    last_exception = e
+                    if attempt == max_retries:
+                        break
+
+                    delay = base_delay * (2**attempt)
+                    logger.warning(
+                        "Attempt %d/%d failed: %s. Retrying in %.1f seconds...",
+                        attempt + 1,
+                        max_retries + 1,
+                        str(e),
+                        delay,
+                    )
+                    time.sleep(delay)
+
+            raise RetryError(
+                f"All {max_retries + 1} attempts failed for {func.__name__}",
+                last_exception=last_exception,
+            )
+
+        return wrapper
+
+    return decorator
+
+
+class ClobClientError(Exception):
+    """Base exception for ClobClient errors."""
+
+
+class ClobClient:
+    """Wrapper around py-clob-client with rate limiting and retry logic.
+
+    This client provides a clean interface for querying Polymarket CLOB data
+    with built-in rate limiting (10 requests/second) and automatic retry
+    with exponential backoff on transient errors.
+
+    Example:
+        >>> client = ClobClient()  # Uses POLYMARKET_API_KEY env var
+        >>> markets = client.get_markets()
+        >>> orderbook = client.get_orderbook("token_id_here")
+    """
+
+    def __init__(
+        self,
+        api_key: str | None = None,
+        host: str = DEFAULT_HOST,
+        max_retries: int = DEFAULT_MAX_RETRIES,
+        requests_per_second: float = MAX_REQUESTS_PER_SECOND,
+    ) -> None:
+        """Initialize the CLOB client.
+
+        Args:
+            api_key: Polymarket API key. If not provided, reads from
+                POLYMARKET_API_KEY environment variable.
+            host: CLOB API endpoint URL.
+            max_retries: Maximum retry attempts for failed requests.
+            requests_per_second: Rate limit for API requests.
+        """
+        self._api_key = api_key or os.environ.get("POLYMARKET_API_KEY")
+        self._host = host
+        self._max_retries = max_retries
+        self._rate_limiter = RateLimiter(requests_per_second)
+
+        # Initialize the underlying client (read-only, no auth needed for queries)
+        self._client = BaseClobClient(host)
+
+        logger.info(
+            "Initialized ClobClient with host=%s, rate_limit=%.1f req/s",
+            host,
+            requests_per_second,
+        )
+
+    def _with_rate_limit(self, func: Callable[P, T]) -> Callable[P, T]:
+        """Wrap a function with rate limiting."""
+
+        @wraps(func)
+        def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
+            self._rate_limiter.acquire_sync()
+            return func(*args, **kwargs)
+
+        return wrapper
+
+    @with_retry()
+    def get_markets(self, active_only: bool = True) -> list[Market]:
+        """Fetch all markets from the CLOB.
+
+        Args:
+            active_only: If True, only return active (non-closed) markets.
+
+        Returns:
+            List of Market objects.
+        """
+        self._rate_limiter.acquire_sync()
+
+        all_markets: list[Market] = []
+        cursor: str | None = None
+
+        while True:
+            if cursor:
+                response = self._client.get_simplified_markets(cursor)
+            else:
+                response = self._client.get_simplified_markets()
+
+            data = response.get("data", [])
+            for market_data in data:
+                market = Market.from_dict(market_data)
+                if active_only and market.closed:
+                    continue
+                all_markets.append(market)
+
+            next_cursor = response.get("next_cursor")
+            if not next_cursor or next_cursor == "LTE=":
+                break
+            cursor = next_cursor
+
+            # Rate limit between pagination requests
+            self._rate_limiter.acquire_sync()
+
+        logger.debug("Fetched %d markets", len(all_markets))
+        return all_markets
+
+    @with_retry()
+    def get_market(self, condition_id: str) -> Market:
+        """Fetch a specific market by its condition ID.
+
+        Args:
+            condition_id: The market's condition ID.
+
+        Returns:
+            Market object.
+
+        Raises:
+            ClobClientError: If the market is not found.
+        """
+        self._rate_limiter.acquire_sync()
+
+        try:
+            response = self._client.get_market(condition_id)
+            return Market.from_dict(response)
+        except Exception as e:
+            raise ClobClientError(f"Failed to fetch market {condition_id}: {e}") from e
+
+    @with_retry()
+    def get_orderbook(self, token_id: str) -> Orderbook:
+        """Fetch the orderbook for a specific token.
+
+        Args:
+            token_id: The token ID to fetch the orderbook for.
+
+        Returns:
+            Orderbook object with bids, asks, and spread information.
+        """
+        self._rate_limiter.acquire_sync()
+
+        try:
+            orderbook = self._client.get_order_book(token_id)
+            return Orderbook.from_clob_orderbook(orderbook)
+        except Exception as e:
+            raise ClobClientError(f"Failed to fetch orderbook for {token_id}: {e}") from e
+
+    @with_retry()
+    def get_orderbooks(self, token_ids: list[str]) -> list[Orderbook]:
+        """Fetch orderbooks for multiple tokens in a single request.
+
+        Args:
+            token_ids: List of token IDs to fetch orderbooks for.
+
+        Returns:
+            List of Orderbook objects.
+        """
+        self._rate_limiter.acquire_sync()
+
+        params = [BookParams(token_id=tid) for tid in token_ids]
+
+        try:
+            orderbooks = self._client.get_order_books(params)
+            return [Orderbook.from_clob_orderbook(ob) for ob in orderbooks]
+        except Exception as e:
+            raise ClobClientError(f"Failed to fetch orderbooks: {e}") from e
+
+    @with_retry()
+    def get_midpoint(self, token_id: str) -> str | None:
+        """Fetch the midpoint price for a token.
+
+        Args:
+            token_id: The token ID.
+
+        Returns:
+            Midpoint price as a string, or None if unavailable.
+        """
+        self._rate_limiter.acquire_sync()
+
+        try:
+            response = self._client.get_midpoint(token_id)
+            return response.get("mid")
+        except Exception as e:
+            logger.warning("Failed to get midpoint for %s: %s", token_id, e)
+            return None
+
+    @with_retry()
+    def get_price(self, token_id: str, side: str = "BUY") -> str | None:
+        """Fetch the best price for a token on a given side.
+
+        Args:
+            token_id: The token ID.
+            side: Either "BUY" or "SELL".
+
+        Returns:
+            Best price as a string, or None if unavailable.
+        """
+        self._rate_limiter.acquire_sync()
+
+        try:
+            response = self._client.get_price(token_id, side=side)
+            return response.get("price")
+        except Exception as e:
+            logger.warning("Failed to get %s price for %s: %s", side, token_id, e)
+            return None
+
+    def health_check(self) -> bool:
+        """Check if the CLOB API is reachable.
+
+        Returns:
+            True if the API responds with "OK", False otherwise.
+        """
+        try:
+            self._rate_limiter.acquire_sync()
+            result = self._client.get_ok()
+            return result == "OK"
+        except Exception as e:
+            logger.error("Health check failed: %s", e)
+            return False
+
+    def get_server_time(self) -> int | None:
+        """Get the server timestamp.
+
+        Returns:
+            Server timestamp in milliseconds, or None on error.
+        """
+        try:
+            self._rate_limiter.acquire_sync()
+            return self._client.get_server_time()
+        except Exception as e:
+            logger.error("Failed to get server time: %s", e)
+            return None

src/polymarket_insider_tracker/ingestor/health.py

@@ -0,0 +1,511 @@
+"""Connection health monitor with metrics and HTTP endpoints.
+
+This module provides health monitoring for the data ingestion layer,
+tracking connection states, event throughput, and staleness detection.
+"""
+
+import asyncio
+import contextlib
+import copy
+import logging
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+from aiohttp import web
+from prometheus_client import Counter, Gauge, Histogram, generate_latest
+
+logger = logging.getLogger(__name__)
+
+# Default configuration
+DEFAULT_STALE_THRESHOLD_SECONDS = 60  # No events for 60s = stale
+DEFAULT_HEALTH_CHECK_INTERVAL = 5  # seconds
+DEFAULT_HTTP_PORT = 8080
+
+
+class HealthStatus(Enum):
+    """Overall health status."""
+
+    HEALTHY = "healthy"
+    DEGRADED = "degraded"
+    UNHEALTHY = "unhealthy"
+
+
+class StreamStatus(Enum):
+    """Status of an individual stream."""
+
+    ACTIVE = "active"
+    STALE = "stale"
+    DISCONNECTED = "disconnected"
+
+
+@dataclass
+class StreamHealth:
+    """Health status for an individual stream."""
+
+    name: str
+    status: StreamStatus = StreamStatus.DISCONNECTED
+    last_event_time: float | None = None
+    events_received: int = 0
+    events_per_second: float = 0.0
+    connected_since: float | None = None
+    last_error: str | None = None
+
+
+@dataclass
+class HealthReport:
+    """Comprehensive health report for all streams."""
+
+    status: HealthStatus
+    streams: dict[str, StreamHealth] = field(default_factory=dict)
+    total_events_received: int = 0
+    total_events_per_second: float = 0.0
+    uptime_seconds: float = 0.0
+    timestamp: float = field(default_factory=time.time)
+
+
+# Type aliases
+HealthCallback = Callable[[HealthReport], Awaitable[None]]
+
+
+# Prometheus metrics
+EVENTS_TOTAL = Counter(
+    "polymarket_events_total",
+    "Total number of events received",
+    ["stream"],
+)
+
+EVENTS_PER_SECOND = Gauge(
+    "polymarket_events_per_second",
+    "Current events per second rate",
+    ["stream"],
+)
+
+STREAM_STATUS = Gauge(
+    "polymarket_stream_status",
+    "Stream status (1=active, 0.5=stale, 0=disconnected)",
+    ["stream"],
+)
+
+LAST_EVENT_TIMESTAMP = Gauge(
+    "polymarket_last_event_timestamp",
+    "Unix timestamp of last event received",
+    ["stream"],
+)
+
+EVENT_LATENCY = Histogram(
+    "polymarket_event_latency_seconds",
+    "Event processing latency in seconds",
+    ["stream"],
+    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0],
+)
+
+HEALTH_STATUS = Gauge(
+    "polymarket_health_status",
+    "Overall health status (1=healthy, 0.5=degraded, 0=unhealthy)",
+)
+
+
+class HealthMonitor:
+    """Monitor connection health and expose metrics.
+
+    This class tracks the health of multiple streams, calculates throughput,
+    detects stale streams, and exposes Prometheus-compatible metrics.
+
+    Example:
+        ```python
+        monitor = HealthMonitor(stale_threshold_seconds=60)
+        await monitor.start()
+
+        # Record events
+        monitor.record_event("trades", processing_time=0.001)
+
+        # Update connection state
+        monitor.set_stream_connected("trades")
+
+        # Get health report
+        report = monitor.get_health_report()
+
+        # HTTP endpoints: /health and /metrics
+        # Start HTTP server with monitor.start_http_server(port=8080)
+        ```
+    """
+
+    def __init__(
+        self,
+        *,
+        stale_threshold_seconds: float = DEFAULT_STALE_THRESHOLD_SECONDS,
+        health_check_interval: float = DEFAULT_HEALTH_CHECK_INTERVAL,
+        on_health_change: HealthCallback | None = None,
+    ) -> None:
+        """Initialize the health monitor.
+
+        Args:
+            stale_threshold_seconds: Seconds without events before stream is stale.
+            health_check_interval: Seconds between health check updates.
+            on_health_change: Optional callback when health status changes.
+        """
+        self._stale_threshold = stale_threshold_seconds
+        self._health_check_interval = health_check_interval
+        self._on_health_change = on_health_change
+
+        self._streams: dict[str, StreamHealth] = {}
+        self._start_time: float | None = None
+        self._running = False
+        self._health_task: asyncio.Task[None] | None = None
+        self._last_health_status: HealthStatus | None = None
+
+        # For throughput calculation
+        self._event_windows: dict[str, list[float]] = {}
+        self._window_duration = 10.0  # 10 second sliding window
+
+        # HTTP server
+        self._app: web.Application | None = None
+        self._runner: web.AppRunner | None = None
+
+    @property
+    def is_running(self) -> bool:
+        """Return True if the monitor is running."""
+        return self._running
+
+    def register_stream(self, name: str) -> None:
+        """Register a stream for monitoring.
+
+        Args:
+            name: Unique name for the stream.
+        """
+        if name not in self._streams:
+            self._streams[name] = StreamHealth(name=name)
+            self._event_windows[name] = []
+            logger.info("Registered stream for monitoring: %s", name)
+
+    def set_stream_connected(self, name: str) -> None:
+        """Mark a stream as connected.
+
+        Args:
+            name: Stream name.
+        """
+        self.register_stream(name)
+        stream = self._streams[name]
+        stream.status = StreamStatus.ACTIVE
+        stream.connected_since = time.time()
+        stream.last_error = None
+        STREAM_STATUS.labels(stream=name).set(1.0)
+        logger.debug("Stream connected: %s", name)
+
+    def set_stream_disconnected(self, name: str, error: str | None = None) -> None:
+        """Mark a stream as disconnected.
+
+        Args:
+            name: Stream name.
+            error: Optional error message.
+        """
+        self.register_stream(name)
+        stream = self._streams[name]
+        stream.status = StreamStatus.DISCONNECTED
+        stream.connected_since = None
+        stream.last_error = error
+        STREAM_STATUS.labels(stream=name).set(0.0)
+        logger.debug("Stream disconnected: %s (error: %s)", name, error)
+
+    def record_event(
+        self,
+        stream_name: str,
+        *,
+        processing_time: float | None = None,
+    ) -> None:
+        """Record an event received from a stream.
+
+        Args:
+            stream_name: Name of the stream.
+            processing_time: Optional processing latency in seconds.
+        """
+        self.register_stream(stream_name)
+        now = time.time()
+
+        stream = self._streams[stream_name]
+        stream.events_received += 1
+        stream.last_event_time = now
+        stream.status = StreamStatus.ACTIVE
+
+        # Update metrics
+        EVENTS_TOTAL.labels(stream=stream_name).inc()
+        LAST_EVENT_TIMESTAMP.labels(stream=stream_name).set(now)
+        STREAM_STATUS.labels(stream=stream_name).set(1.0)
+
+        if processing_time is not None:
+            EVENT_LATENCY.labels(stream=stream_name).observe(processing_time)
+
+        # Add to sliding window for throughput
+        window = self._event_windows[stream_name]
+        window.append(now)
+
+        # Clean old entries from window
+        cutoff = now - self._window_duration
+        self._event_windows[stream_name] = [t for t in window if t > cutoff]
+
+    def _calculate_throughput(self, stream_name: str) -> float:
+        """Calculate events per second for a stream.
+
+        Args:
+            stream_name: Name of the stream.
+
+        Returns:
+            Events per second rate.
+        """
+        window = self._event_windows.get(stream_name, [])
+        if not window:
+            return 0.0
+
+        now = time.time()
+        cutoff = now - self._window_duration
+
+        # Count events in window
+        recent_events = [t for t in window if t > cutoff]
+        if not recent_events:
+            return 0.0
+
+        # Calculate rate based on window
+        window_span = now - cutoff
+        return len(recent_events) / window_span if window_span > 0 else 0.0
+
+    def _check_stream_staleness(self) -> None:
+        """Check all streams for staleness."""
+        now = time.time()
+
+        for name, stream in self._streams.items():
+            if stream.status == StreamStatus.DISCONNECTED:
+                continue
+
+            if stream.last_event_time is None:
+                # Connected but no events yet - check connection time
+                if stream.connected_since:
+                    since_connect = now - stream.connected_since
+                    if since_connect > self._stale_threshold:
+                        stream.status = StreamStatus.STALE
+                        STREAM_STATUS.labels(stream=name).set(0.5)
+            else:
+                since_event = now - stream.last_event_time
+                if since_event > self._stale_threshold:
+                    stream.status = StreamStatus.STALE
+                    STREAM_STATUS.labels(stream=name).set(0.5)
+                else:
+                    stream.status = StreamStatus.ACTIVE
+                    STREAM_STATUS.labels(stream=name).set(1.0)
+
+    def _determine_overall_status(self) -> HealthStatus:
+        """Determine overall health status based on stream states.
+
+        Returns:
+            Overall health status.
+        """
+        if not self._streams:
+            return HealthStatus.HEALTHY  # No streams = healthy (nothing to monitor)
+
+        statuses = [s.status for s in self._streams.values()]
+
+        if all(s == StreamStatus.DISCONNECTED for s in statuses):
+            return HealthStatus.UNHEALTHY
+
+        if any(s == StreamStatus.DISCONNECTED for s in statuses):
+            return HealthStatus.DEGRADED
+
+        if any(s == StreamStatus.STALE for s in statuses):
+            return HealthStatus.DEGRADED
+
+        return HealthStatus.HEALTHY
+
+    def get_health_report(self) -> HealthReport:
+        """Generate a comprehensive health report.
+
+        Returns:
+            HealthReport with current status of all streams.
+        """
+        self._check_stream_staleness()
+
+        # Update throughput metrics
+        total_eps = 0.0
+        for name, stream in self._streams.items():
+            eps = self._calculate_throughput(name)
+            stream.events_per_second = eps
+            EVENTS_PER_SECOND.labels(stream=name).set(eps)
+            total_eps += eps
+
+        overall_status = self._determine_overall_status()
+        HEALTH_STATUS.set(
+            1.0 if overall_status == HealthStatus.HEALTHY
+            else 0.5 if overall_status == HealthStatus.DEGRADED
+            else 0.0
+        )
+
+        uptime = 0.0
+        if self._start_time:
+            uptime = time.time() - self._start_time
+
+        # Deep copy streams to prevent mutations affecting internal state
+        streams_copy = {name: copy.copy(stream) for name, stream in self._streams.items()}
+
+        return HealthReport(
+            status=overall_status,
+            streams=streams_copy,
+            total_events_received=sum(s.events_received for s in self._streams.values()),
+            total_events_per_second=total_eps,
+            uptime_seconds=uptime,
+        )
+
+    async def _health_check_loop(self) -> None:
+        """Background task for periodic health checks."""
+        while self._running:
+            try:
+                report = self.get_health_report()
+
+                # Notify on status change
+                if (
+                    self._on_health_change
+                    and report.status != self._last_health_status
+                ):
+                    self._last_health_status = report.status
+                    try:
+                        await self._on_health_change(report)
+                    except Exception as e:
+                        logger.error("Error in health change callback: %s", e)
+
+                await asyncio.sleep(self._health_check_interval)
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error("Error in health check loop: %s", e)
+                await asyncio.sleep(1)
+
+    async def start(self) -> None:
+        """Start the health monitor.
+
+        Begins periodic health checks and staleness detection.
+        """
+        if self._running:
+            return
+
+        self._running = True
+        self._start_time = time.time()
+        self._health_task = asyncio.create_task(self._health_check_loop())
+        logger.info("Health monitor started")
+
+    async def stop(self) -> None:
+        """Stop the health monitor."""
+        if not self._running:
+            return
+
+        self._running = False
+
+        if self._health_task:
+            self._health_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._health_task
+            self._health_task = None
+
+        await self.stop_http_server()
+        logger.info("Health monitor stopped")
+
+    # HTTP Server methods
+
+    async def _handle_health(self, _request: web.Request) -> web.Response:
+        """Handle /health endpoint."""
+        report = self.get_health_report()
+
+        status_code = 200 if report.status == HealthStatus.HEALTHY else 503
+
+        body: dict[str, Any] = {
+            "status": report.status.value,
+            "uptime_seconds": report.uptime_seconds,
+            "total_events_received": report.total_events_received,
+            "total_events_per_second": round(report.total_events_per_second, 2),
+            "streams": {},
+        }
+
+        for name, stream in report.streams.items():
+            body["streams"][name] = {
+                "status": stream.status.value,
+                "events_received": stream.events_received,
+                "events_per_second": round(stream.events_per_second, 2),
+                "last_event_time": stream.last_event_time,
+                "last_error": stream.last_error,
+            }
+
+        return web.json_response(body, status=status_code)
+
+    async def _handle_metrics(self, _request: web.Request) -> web.Response:
+        """Handle /metrics endpoint (Prometheus format)."""
+        # Ensure latest values are calculated
+        self.get_health_report()
+
+        metrics = generate_latest()
+        return web.Response(
+            body=metrics,
+            content_type="text/plain",
+            charset="utf-8",
+        )
+
+    async def _handle_ready(self, _request: web.Request) -> web.Response:
+        """Handle /ready endpoint for k8s readiness probe."""
+        report = self.get_health_report()
+
+        if report.status == HealthStatus.UNHEALTHY:
+            return web.json_response(
+                {"ready": False, "reason": "unhealthy"},
+                status=503,
+            )
+
+        return web.json_response({"ready": True}, status=200)
+
+    async def _handle_live(self, _request: web.Request) -> web.Response:
+        """Handle /live endpoint for k8s liveness probe."""
+        # Always return 200 if the server is running
+        return web.json_response({"live": True}, status=200)
+
+    def _create_app(self) -> web.Application:
+        """Create the aiohttp application."""
+        app = web.Application()
+        app.router.add_get("/health", self._handle_health)
+        app.router.add_get("/metrics", self._handle_metrics)
+        app.router.add_get("/ready", self._handle_ready)
+        app.router.add_get("/live", self._handle_live)
+        return app
+
+    async def start_http_server(self, port: int = DEFAULT_HTTP_PORT) -> None:
+        """Start the HTTP server for health and metrics endpoints.
+
+        Args:
+            port: Port to listen on.
+        """
+        if self._runner:
+            logger.warning("HTTP server already running")
+            return
+
+        self._app = self._create_app()
+        self._runner = web.AppRunner(self._app)
+        await self._runner.setup()
+
+        site = web.TCPSite(self._runner, "0.0.0.0", port)
+        await site.start()
+
+        logger.info("Health HTTP server started on port %d", port)
+
+    async def stop_http_server(self) -> None:
+        """Stop the HTTP server."""
+        if self._runner:
+            await self._runner.cleanup()
+            self._runner = None
+            self._app = None
+            logger.info("Health HTTP server stopped")
+
+    async def __aenter__(self) -> "HealthMonitor":
+        """Async context manager entry."""
+        await self.start()
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        """Async context manager exit."""
+        await self.stop()

src/polymarket_insider_tracker/ingestor/metadata_sync.py

@@ -0,0 +1,362 @@
+"""Market metadata synchronizer with Redis caching.
+
+This module provides a background sync service that keeps market metadata
+up-to-date in Redis, with cache-first lookups for fast access.
+"""
+
+import asyncio
+import contextlib
+import json
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from enum import Enum
+
+from redis.asyncio import Redis
+
+from .clob_client import ClobClient
+from .models import MarketMetadata
+
+logger = logging.getLogger(__name__)
+
+
+# Default configuration
+DEFAULT_SYNC_INTERVAL_SECONDS = 300  # 5 minutes
+DEFAULT_CACHE_TTL_SECONDS = 600  # 10 minutes
+DEFAULT_REDIS_KEY_PREFIX = "polymarket:market:"
+
+
+class SyncState(str, Enum):
+    """State of the metadata synchronizer."""
+
+    STOPPED = "stopped"
+    STARTING = "starting"
+    SYNCING = "syncing"
+    IDLE = "idle"
+    STOPPING = "stopping"
+    ERROR = "error"
+
+
+@dataclass
+class SyncStats:
+    """Statistics for the metadata sync process."""
+
+    total_syncs: int = 0
+    successful_syncs: int = 0
+    failed_syncs: int = 0
+    markets_cached: int = 0
+    last_sync_time: datetime | None = None
+    last_sync_duration_seconds: float = 0.0
+    last_error: str | None = None
+
+
+# Type aliases for callbacks
+StateCallback = Callable[[SyncState], None]
+SyncCallback = Callable[[SyncStats], None]
+
+
+class MetadataSyncError(Exception):
+    """Base exception for metadata sync errors."""
+
+    pass
+
+
+class MarketMetadataSync:
+    """Background service that syncs market metadata to Redis.
+
+    This service:
+    - Fetches all markets from the CLOB API on startup
+    - Refreshes the cache every sync_interval_seconds (default: 5 minutes)
+    - Stores market metadata in Redis with TTL-based expiration
+    - Provides cache-first lookups via get_market()
+
+    Example:
+        ```python
+        redis = Redis.from_url("redis://localhost:6379")
+        clob = ClobClient()
+
+        sync = MarketMetadataSync(redis=redis, clob_client=clob)
+        await sync.start()
+
+        # Get market metadata (cache-first)
+        metadata = await sync.get_market("0x1234...")
+
+        await sync.stop()
+        ```
+    """
+
+    def __init__(
+        self,
+        redis: Redis,
+        clob_client: ClobClient,
+        *,
+        sync_interval_seconds: int = DEFAULT_SYNC_INTERVAL_SECONDS,
+        cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS,
+        key_prefix: str = DEFAULT_REDIS_KEY_PREFIX,
+        on_state_change: StateCallback | None = None,
+        on_sync_complete: SyncCallback | None = None,
+    ) -> None:
+        """Initialize the metadata sync service.
+
+        Args:
+            redis: Redis async client for caching.
+            clob_client: CLOB API client for fetching markets.
+            sync_interval_seconds: Interval between syncs (default: 300 / 5 min).
+            cache_ttl_seconds: TTL for cached entries (default: 600 / 10 min).
+            key_prefix: Redis key prefix for market data.
+            on_state_change: Callback for state changes.
+            on_sync_complete: Callback after each sync completes.
+        """
+        self._redis = redis
+        self._clob = clob_client
+        self._sync_interval = sync_interval_seconds
+        self._cache_ttl = cache_ttl_seconds
+        self._key_prefix = key_prefix
+        self._on_state_change = on_state_change
+        self._on_sync_complete = on_sync_complete
+
+        self._state = SyncState.STOPPED
+        self._stats = SyncStats()
+        self._sync_task: asyncio.Task[None] | None = None
+        self._stop_event = asyncio.Event()
+
+    @property
+    def state(self) -> SyncState:
+        """Current sync state."""
+        return self._state
+
+    @property
+    def stats(self) -> SyncStats:
+        """Current sync statistics."""
+        return self._stats
+
+    def _set_state(self, new_state: SyncState) -> None:
+        """Update state and notify callback."""
+        old_state = self._state
+        self._state = new_state
+        if self._on_state_change and old_state != new_state:
+            try:
+                self._on_state_change(new_state)
+            except Exception as e:
+                logger.warning(f"State change callback failed: {e}")
+
+    async def start(self) -> None:
+        """Start the background sync service.
+
+        This will:
+        1. Perform an initial sync of all markets
+        2. Start a background task to periodically refresh
+        """
+        if self._state != SyncState.STOPPED:
+            logger.warning(f"Cannot start sync: already in state {self._state}")
+            return
+
+        self._set_state(SyncState.STARTING)
+        self._stop_event.clear()
+
+        # Perform initial sync
+        try:
+            await self._sync_all_markets()
+        except Exception as e:
+            logger.error(f"Initial sync failed: {e}")
+            self._set_state(SyncState.ERROR)
+            self._stats.last_error = str(e)
+            raise MetadataSyncError(f"Failed to start: initial sync failed: {e}") from e
+
+        # Start background sync loop
+        self._sync_task = asyncio.create_task(self._sync_loop())
+        self._set_state(SyncState.IDLE)
+        logger.info("Market metadata sync started")
+
+    async def stop(self) -> None:
+        """Stop the background sync service."""
+        if self._state == SyncState.STOPPED:
+            return
+
+        self._set_state(SyncState.STOPPING)
+        self._stop_event.set()
+
+        if self._sync_task:
+            self._sync_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._sync_task
+            self._sync_task = None
+
+        self._set_state(SyncState.STOPPED)
+        logger.info("Market metadata sync stopped")
+
+    async def _sync_loop(self) -> None:
+        """Background loop that periodically syncs markets."""
+        while not self._stop_event.is_set():
+            try:
+                # Wait for next sync interval or stop event
+                try:
+                    await asyncio.wait_for(
+                        self._stop_event.wait(),
+                        timeout=self._sync_interval,
+                    )
+                    # Stop event was set
+                    break
+                except TimeoutError:
+                    # Timeout - time to sync
+                    pass
+
+                if self._stop_event.is_set():
+                    break
+
+                await self._sync_all_markets()
+
+            except asyncio.CancelledError:
+                break
+            except Exception as e:
+                logger.error(f"Sync loop error: {e}")
+                self._stats.failed_syncs += 1
+                self._stats.last_error = str(e)
+                self._set_state(SyncState.ERROR)
+                # Continue running - will retry on next interval
+
+    async def _sync_all_markets(self) -> None:
+        """Fetch all markets and cache them in Redis."""
+        self._set_state(SyncState.SYNCING)
+        start_time = datetime.now(UTC)
+        self._stats.total_syncs += 1
+
+        try:
+            # Fetch markets from CLOB API (runs in thread pool for sync API)
+            markets = await asyncio.to_thread(self._clob.get_markets, True)
+
+            # Cache each market in Redis
+            cached_count = 0
+            for market in markets:
+                try:
+                    metadata = MarketMetadata.from_market(market)
+                    await self._cache_market(metadata)
+                    cached_count += 1
+                except Exception as e:
+                    logger.warning(f"Failed to cache market {market.condition_id}: {e}")
+
+            # Update stats
+            end_time = datetime.now(UTC)
+            self._stats.successful_syncs += 1
+            self._stats.markets_cached = cached_count
+            self._stats.last_sync_time = end_time
+            self._stats.last_sync_duration_seconds = (end_time - start_time).total_seconds()
+            self._stats.last_error = None
+
+            self._set_state(SyncState.IDLE)
+            logger.info(
+                f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s"
+            )
+
+            # Notify callback
+            if self._on_sync_complete:
+                try:
+                    self._on_sync_complete(self._stats)
+                except Exception as e:
+                    logger.warning(f"Sync complete callback failed: {e}")
+
+        except Exception as e:
+            self._stats.failed_syncs += 1
+            self._stats.last_error = str(e)
+            self._set_state(SyncState.ERROR)
+            logger.error(f"Market sync failed: {e}")
+            raise
+
+    async def _cache_market(self, metadata: MarketMetadata) -> None:
+        """Cache a single market metadata in Redis.
+
+        Args:
+            metadata: The market metadata to cache.
+        """
+        key = f"{self._key_prefix}{metadata.condition_id}"
+        value = json.dumps(metadata.to_dict())
+        await self._redis.setex(key, self._cache_ttl, value)
+
+    async def get_market(self, condition_id: str) -> MarketMetadata | None:
+        """Get market metadata with cache-first lookup.
+
+        This first checks Redis cache. If not found or expired,
+        it fetches from the CLOB API and caches the result.
+
+        Args:
+            condition_id: The market condition ID.
+
+        Returns:
+            MarketMetadata if found, None otherwise.
+        """
+        # Try cache first
+        key = f"{self._key_prefix}{condition_id}"
+        cached = await self._redis.get(key)
+
+        if cached:
+            try:
+                data = json.loads(cached)
+                return MarketMetadata.from_dict(data)
+            except (json.JSONDecodeError, KeyError) as e:
+                logger.warning(f"Failed to parse cached market {condition_id}: {e}")
+
+        # Cache miss - fetch from API
+        try:
+            market = await asyncio.to_thread(self._clob.get_market, condition_id)
+            if market:
+                metadata = MarketMetadata.from_market(market)
+                await self._cache_market(metadata)
+                return metadata
+        except Exception as e:
+            logger.warning(f"Failed to fetch market {condition_id}: {e}")
+
+        return None
+
+    async def get_markets_by_category(self, category: str) -> list[MarketMetadata]:
+        """Get all cached markets of a specific category.
+
+        Note: This scans all cached markets. For large datasets,
+        consider using a Redis set or secondary index.
+
+        Args:
+            category: The category to filter by.
+
+        Returns:
+            List of matching MarketMetadata.
+        """
+        results: list[MarketMetadata] = []
+        pattern = f"{self._key_prefix}*"
+
+        cursor = 0
+        while True:
+            cursor, keys = await self._redis.scan(cursor, match=pattern, count=100)
+            for key in keys:
+                cached = await self._redis.get(key)
+                if cached:
+                    try:
+                        data = json.loads(cached)
+                        if data.get("category") == category:
+                            results.append(MarketMetadata.from_dict(data))
+                    except (json.JSONDecodeError, KeyError):
+                        pass
+            if cursor == 0:
+                break
+
+        return results
+
+    async def invalidate_market(self, condition_id: str) -> bool:
+        """Invalidate (delete) a cached market.
+
+        Args:
+            condition_id: The market condition ID to invalidate.
+
+        Returns:
+            True if the key was deleted, False if it didn't exist.
+        """
+        key = f"{self._key_prefix}{condition_id}"
+        deleted = await self._redis.delete(key)
+        return deleted > 0
+
+    async def force_sync(self) -> None:
+        """Force an immediate sync of all markets.
+
+        This can be called to refresh the cache outside of the
+        normal sync interval.
+        """
+        await self._sync_all_markets()

src/polymarket_insider_tracker/ingestor/models.py

@@ -0,0 +1,493 @@
+"""Data models for the ingestor module."""
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from decimal import Decimal
+from typing import Any, Literal
+
+
+@dataclass(frozen=True)
+class Token:
+    """Represents a token in a Polymarket market."""
+
+    token_id: str
+    outcome: str
+    price: Decimal | None = None
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "Token":
+        """Create a Token from a dictionary."""
+        price = data.get("price")
+        return cls(
+            token_id=str(data["token_id"]),
+            outcome=str(data["outcome"]),
+            price=Decimal(str(price)) if price is not None else None,
+        )
+
+
+@dataclass(frozen=True)
+class Market:
+    """Represents a Polymarket prediction market."""
+
+    condition_id: str
+    question: str
+    description: str
+    tokens: tuple[Token, ...]
+    end_date: datetime | None = None
+    active: bool = True
+    closed: bool = False
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "Market":
+        """Create a Market from a dictionary response."""
+        tokens_data = data.get("tokens", [])
+        tokens = tuple(Token.from_dict(t) for t in tokens_data)
+
+        end_date = None
+        end_date_iso = data.get("end_date_iso")
+        if end_date_iso:
+            try:
+                end_date = datetime.fromisoformat(end_date_iso.replace("Z", "+00:00"))
+            except (ValueError, AttributeError):
+                pass
+
+        return cls(
+            condition_id=str(data["condition_id"]),
+            question=str(data.get("question", "")),
+            description=str(data.get("description", "")),
+            tokens=tokens,
+            end_date=end_date,
+            active=bool(data.get("active", True)),
+            closed=bool(data.get("closed", False)),
+        )
+
+
+@dataclass(frozen=True)
+class OrderbookLevel:
+    """Represents a single price level in an orderbook."""
+
+    price: Decimal
+    size: Decimal
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "OrderbookLevel":
+        """Create an OrderbookLevel from a dictionary."""
+        return cls(
+            price=Decimal(str(data["price"])),
+            size=Decimal(str(data["size"])),
+        )
+
+
+@dataclass(frozen=True)
+class Orderbook:
+    """Represents an orderbook for a Polymarket token."""
+
+    market: str
+    asset_id: str
+    bids: tuple[OrderbookLevel, ...]
+    asks: tuple[OrderbookLevel, ...]
+    tick_size: Decimal
+    timestamp: datetime = field(default_factory=datetime.utcnow)
+
+    @classmethod
+    def from_clob_orderbook(cls, orderbook: Any) -> "Orderbook":
+        """Create an Orderbook from a py-clob-client orderbook object."""
+        bids = tuple(
+            OrderbookLevel(
+                price=Decimal(str(bid.price)),
+                size=Decimal(str(bid.size)),
+            )
+            for bid in (orderbook.bids or [])
+        )
+        asks = tuple(
+            OrderbookLevel(
+                price=Decimal(str(ask.price)),
+                size=Decimal(str(ask.size)),
+            )
+            for ask in (orderbook.asks or [])
+        )
+
+        return cls(
+            market=str(orderbook.market),
+            asset_id=str(orderbook.asset_id),
+            bids=bids,
+            asks=asks,
+            tick_size=Decimal(str(orderbook.tick_size)),
+        )
+
+    @property
+    def best_bid(self) -> Decimal | None:
+        """Return the best bid price, or None if no bids."""
+        return self.bids[0].price if self.bids else None
+
+    @property
+    def best_ask(self) -> Decimal | None:
+        """Return the best ask price, or None if no asks."""
+        return self.asks[0].price if self.asks else None
+
+    @property
+    def spread(self) -> Decimal | None:
+        """Return the bid-ask spread, or None if missing data."""
+        if self.best_bid is not None and self.best_ask is not None:
+            return self.best_ask - self.best_bid
+        return None
+
+    @property
+    def midpoint(self) -> Decimal | None:
+        """Return the midpoint price, or None if missing data."""
+        if self.best_bid is not None and self.best_ask is not None:
+            return (self.best_bid + self.best_ask) / 2
+        return None
+
+
+@dataclass(frozen=True)
+class TradeEvent:
+    """Represents a trade event from the Polymarket WebSocket feed.
+
+    This captures all the information about a single trade execution,
+    including the market, wallet, trade details, and metadata.
+    """
+
+    # Core trade identifiers
+    market_id: str  # conditionId - the market/CTF condition ID
+    trade_id: str  # transactionHash - unique trade identifier
+    wallet_address: str  # proxyWallet - trader's wallet address
+
+    # Trade details
+    side: Literal["BUY", "SELL"]
+    outcome: str  # Human-readable outcome (e.g., "Yes", "No")
+    outcome_index: int  # Index of the outcome (0 or 1)
+    price: Decimal
+    size: Decimal  # Number of shares traded
+    timestamp: datetime
+
+    # Asset information
+    asset_id: str  # ERC1155 token ID
+
+    # Market metadata
+    market_slug: str = ""
+    event_slug: str = ""
+    event_title: str = ""
+
+    # Trader metadata (optional - may not be available for all trades)
+    trader_name: str = ""
+    trader_pseudonym: str = ""
+
+    @classmethod
+    def from_websocket_message(cls, data: dict[str, Any]) -> "TradeEvent":
+        """Create a TradeEvent from a WebSocket activity/trade message.
+
+        Args:
+            data: The payload from a WebSocket trade message.
+
+        Returns:
+            TradeEvent instance.
+        """
+        # Parse timestamp - it's a Unix timestamp in seconds
+        raw_timestamp = data.get("timestamp", 0)
+        if isinstance(raw_timestamp, int):
+            timestamp = datetime.fromtimestamp(raw_timestamp, tz=UTC)
+        else:
+            timestamp = datetime.now(UTC)
+
+        # Parse side - normalize to uppercase
+        side_raw = str(data.get("side", "BUY")).upper()
+        side: Literal["BUY", "SELL"] = "BUY" if side_raw == "BUY" else "SELL"
+
+        return cls(
+            market_id=str(data.get("conditionId", "")),
+            trade_id=str(data.get("transactionHash", "")),
+            wallet_address=str(data.get("proxyWallet", "")),
+            side=side,
+            outcome=str(data.get("outcome", "")),
+            outcome_index=int(data.get("outcomeIndex", 0)),
+            price=Decimal(str(data.get("price", 0))),
+            size=Decimal(str(data.get("size", 0))),
+            timestamp=timestamp,
+            asset_id=str(data.get("asset", "")),
+            market_slug=str(data.get("slug", "")),
+            event_slug=str(data.get("eventSlug", "")),
+            event_title=str(data.get("title", "")),
+            trader_name=str(data.get("name", "")),
+            trader_pseudonym=str(data.get("pseudonym", "")),
+        )
+
+    @property
+    def is_buy(self) -> bool:
+        """Return True if this is a buy trade."""
+        return self.side == "BUY"
+
+    @property
+    def is_sell(self) -> bool:
+        """Return True if this is a sell trade."""
+        return self.side == "SELL"
+
+    @property
+    def notional_value(self) -> Decimal:
+        """Return the notional value of the trade (price * size)."""
+        return self.price * self.size
+
+
+# Category keywords for market classification
+_CATEGORY_KEYWORDS: dict[str, list[str]] = {
+    "politics": [
+        "election",
+        "president",
+        "congress",
+        "senate",
+        "house",
+        "governor",
+        "mayor",
+        "vote",
+        "ballot",
+        "democrat",
+        "republican",
+        "trump",
+        "biden",
+        "political",
+        "party",
+        "campaign",
+        "poll",
+        "primary",
+        "caucus",
+    ],
+    "crypto": [
+        "bitcoin",
+        "ethereum",
+        "crypto",
+        "btc",
+        "eth",
+        "blockchain",
+        "token",
+        "defi",
+        "nft",
+        "altcoin",
+        "solana",
+        "cardano",
+        "dogecoin",
+    ],
+    "sports": [
+        "nfl",
+        "nba",
+        "mlb",
+        "nhl",
+        "soccer",
+        "football",
+        "basketball",
+        "baseball",
+        "hockey",
+        "tennis",
+        "golf",
+        "ufc",
+        "boxing",
+        "olympics",
+        "championship",
+        "super bowl",
+        "world cup",
+        "playoffs",
+        "finals",
+    ],
+    "entertainment": [
+        "movie",
+        "film",
+        "oscar",
+        "grammy",
+        "emmy",
+        "album",
+        "song",
+        "celebrity",
+        "netflix",
+        "disney",
+        "streaming",
+        "box office",
+        "tv show",
+        "series",
+        "actor",
+        "actress",
+        "music",
+    ],
+    "finance": [
+        "stock",
+        "market",
+        "fed",
+        "interest rate",
+        "inflation",
+        "gdp",
+        "unemployment",
+        "recession",
+        "economy",
+        "s&p",
+        "nasdaq",
+        "dow",
+        "treasury",
+        "bond",
+        "forex",
+        "gold",
+        "oil",
+        "commodity",
+    ],
+    "tech": [
+        "apple",
+        "google",
+        "microsoft",
+        "amazon",
+        "meta",
+        "tesla",
+        "ai",
+        "artificial intelligence",
+        "chatgpt",
+        "openai",
+        "semiconductor",
+        "iphone",
+        "android",
+        "software",
+        "hardware",
+        "startup",
+    ],
+    "science": [
+        "nasa",
+        "space",
+        "climate",
+        "weather",
+        "vaccine",
+        "covid",
+        "fda",
+        "drug",
+        "trial",
+        "research",
+        "study",
+        "discovery",
+    ],
+}
+
+
+def derive_category(title: str) -> str:
+    """Derive a market category from the market title.
+
+    Args:
+        title: The market question or title.
+
+    Returns:
+        Category string, or "other" if no match found.
+    """
+    title_lower = title.lower()
+
+    for category, keywords in _CATEGORY_KEYWORDS.items():
+        for keyword in keywords:
+            if keyword in title_lower:
+                return category
+
+    return "other"
+
+
+@dataclass(frozen=True)
+class MarketMetadata:
+    """Extended market metadata with derived fields and caching support.
+
+    This combines the core Market data with derived metadata like category
+    and is designed for efficient caching in Redis.
+    """
+
+    # Core market data
+    condition_id: str
+    question: str
+    description: str
+    tokens: tuple[Token, ...]
+    end_date: datetime | None = None
+    active: bool = True
+    closed: bool = False
+
+    # Derived metadata
+    category: str = "other"
+
+    # Cache metadata
+    last_updated: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+    @classmethod
+    def from_market(cls, market: Market) -> "MarketMetadata":
+        """Create MarketMetadata from a Market object.
+
+        Args:
+            market: The source Market object.
+
+        Returns:
+            MarketMetadata with derived fields populated.
+        """
+        return cls(
+            condition_id=market.condition_id,
+            question=market.question,
+            description=market.description,
+            tokens=market.tokens,
+            end_date=market.end_date,
+            active=market.active,
+            closed=market.closed,
+            category=derive_category(market.question),
+            last_updated=datetime.now(UTC),
+        )
+
+    def to_dict(self) -> dict[str, Any]:
+        """Serialize to a dictionary for Redis storage.
+
+        Returns:
+            Dictionary representation suitable for JSON serialization.
+        """
+        return {
+            "condition_id": self.condition_id,
+            "question": self.question,
+            "description": self.description,
+            "tokens": [
+                {
+                    "token_id": t.token_id,
+                    "outcome": t.outcome,
+                    "price": str(t.price) if t.price is not None else None,
+                }
+                for t in self.tokens
+            ],
+            "end_date": self.end_date.isoformat() if self.end_date else None,
+            "active": self.active,
+            "closed": self.closed,
+            "category": self.category,
+            "last_updated": self.last_updated.isoformat(),
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> "MarketMetadata":
+        """Deserialize from a dictionary (from Redis storage).
+
+        Args:
+            data: Dictionary from Redis.
+
+        Returns:
+            MarketMetadata instance.
+        """
+        tokens_data = data.get("tokens", [])
+        tokens = tuple(Token.from_dict(t) for t in tokens_data)
+
+        end_date = None
+        end_date_str = data.get("end_date")
+        if end_date_str:
+            try:
+                end_date = datetime.fromisoformat(end_date_str)
+            except (ValueError, AttributeError):
+                pass
+
+        last_updated_str = data.get("last_updated")
+        if last_updated_str:
+            try:
+                last_updated = datetime.fromisoformat(last_updated_str)
+            except (ValueError, AttributeError):
+                last_updated = datetime.now(UTC)
+        else:
+            last_updated = datetime.now(UTC)
+
+        return cls(
+            condition_id=str(data["condition_id"]),
+            question=str(data.get("question", "")),
+            description=str(data.get("description", "")),
+            tokens=tokens,
+            end_date=end_date,
+            active=bool(data.get("active", True)),
+            closed=bool(data.get("closed", False)),
+            category=str(data.get("category", "other")),
+            last_updated=last_updated,
+        )

src/polymarket_insider_tracker/ingestor/publisher.py

@@ -0,0 +1,419 @@
+"""Redis Streams event publisher for trade events.
+
+This module provides an event publisher that writes normalized trade events
+to Redis Streams, enabling downstream consumers to process events independently.
+"""
+
+import logging
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from decimal import Decimal
+from typing import Any, Literal
+
+from redis.asyncio import Redis
+from redis.exceptions import ResponseError
+
+from .models import TradeEvent
+
+logger = logging.getLogger(__name__)
+
+
+# Default configuration
+DEFAULT_STREAM_NAME = "trades"
+DEFAULT_MAX_LEN = 100_000  # 100k events
+DEFAULT_BLOCK_MS = 1000
+DEFAULT_COUNT = 10
+
+
+class PublisherError(Exception):
+    """Base exception for publisher errors."""
+
+    pass
+
+
+class ConsumerGroupExistsError(PublisherError):
+    """Raised when trying to create a consumer group that already exists."""
+
+    pass
+
+
+@dataclass
+class StreamEntry:
+    """Represents an entry read from a Redis Stream."""
+
+    entry_id: str
+    event: TradeEvent
+
+
+def _serialize_trade_event(event: TradeEvent) -> dict[str, str]:
+    """Serialize a TradeEvent to a dict suitable for Redis Streams.
+
+    Redis Streams require string key-value pairs, so we convert all
+    values to strings.
+
+    Args:
+        event: The TradeEvent to serialize.
+
+    Returns:
+        Dictionary with string keys and values.
+    """
+    return {
+        "market_id": event.market_id,
+        "trade_id": event.trade_id,
+        "wallet_address": event.wallet_address,
+        "side": event.side,
+        "outcome": event.outcome,
+        "outcome_index": str(event.outcome_index),
+        "price": str(event.price),
+        "size": str(event.size),
+        "timestamp": event.timestamp.isoformat(),
+        "asset_id": event.asset_id,
+        "market_slug": event.market_slug,
+        "event_slug": event.event_slug,
+        "event_title": event.event_title,
+        "trader_name": event.trader_name,
+        "trader_pseudonym": event.trader_pseudonym,
+    }
+
+
+def _deserialize_trade_event(data: dict[bytes | str, bytes | str]) -> TradeEvent:
+    """Deserialize a TradeEvent from Redis Stream data.
+
+    Args:
+        data: The raw data from Redis Stream (may have bytes keys/values).
+
+    Returns:
+        TradeEvent instance.
+    """
+    # Convert bytes to strings if needed
+    decoded: dict[str, str] = {}
+    for k, v in data.items():
+        key = k.decode() if isinstance(k, bytes) else k
+        value = v.decode() if isinstance(v, bytes) else v
+        decoded[key] = value
+
+    # Parse timestamp
+    timestamp_str = decoded.get("timestamp", "")
+    try:
+        timestamp = datetime.fromisoformat(timestamp_str)
+    except (ValueError, TypeError):
+        timestamp = datetime.now(UTC)
+
+    # Parse side
+    side_raw = decoded.get("side", "BUY").upper()
+    side: Literal["BUY", "SELL"] = "BUY" if side_raw == "BUY" else "SELL"
+
+    return TradeEvent(
+        market_id=decoded.get("market_id", ""),
+        trade_id=decoded.get("trade_id", ""),
+        wallet_address=decoded.get("wallet_address", ""),
+        side=side,
+        outcome=decoded.get("outcome", ""),
+        outcome_index=int(decoded.get("outcome_index", "0")),
+        price=Decimal(decoded.get("price", "0")),
+        size=Decimal(decoded.get("size", "0")),
+        timestamp=timestamp,
+        asset_id=decoded.get("asset_id", ""),
+        market_slug=decoded.get("market_slug", ""),
+        event_slug=decoded.get("event_slug", ""),
+        event_title=decoded.get("event_title", ""),
+        trader_name=decoded.get("trader_name", ""),
+        trader_pseudonym=decoded.get("trader_pseudonym", ""),
+    )
+
+
+class EventPublisher:
+    """Event publisher using Redis Streams.
+
+    This class wraps Redis Streams to provide:
+    - Publishing single or batch trade events
+    - Consumer group management
+    - Event reading for consumers
+
+    Example:
+        ```python
+        redis = Redis.from_url("redis://localhost:6379")
+        publisher = EventPublisher(redis)
+
+        # Publish events
+        event_id = await publisher.publish(trade_event)
+
+        # Create consumer group for downstream processing
+        await publisher.create_consumer_group("wallet-profiler")
+
+        # Read events as consumer
+        entries = await publisher.read_events(
+            group_name="wallet-profiler",
+            consumer_name="worker-1"
+        )
+        for entry in entries:
+            process(entry.event)
+            await publisher.ack(entry.entry_id)
+        ```
+    """
+
+    def __init__(
+        self,
+        redis: Redis,
+        stream_name: str = DEFAULT_STREAM_NAME,
+        *,
+        max_len: int = DEFAULT_MAX_LEN,
+    ) -> None:
+        """Initialize the event publisher.
+
+        Args:
+            redis: Redis async client.
+            stream_name: Name of the Redis Stream.
+            max_len: Maximum number of entries to keep in stream.
+        """
+        self._redis = redis
+        self._stream_name = stream_name
+        self._max_len = max_len
+
+    @property
+    def stream_name(self) -> str:
+        """Return the stream name."""
+        return self._stream_name
+
+    async def publish(self, event: TradeEvent) -> str:
+        """Publish a single trade event to the stream.
+
+        Args:
+            event: The TradeEvent to publish.
+
+        Returns:
+            The entry ID assigned by Redis.
+        """
+        data = _serialize_trade_event(event)
+        entry_id = await self._redis.xadd(
+            self._stream_name,
+            data,
+            maxlen=self._max_len,
+        )
+        # entry_id may be bytes or str
+        if isinstance(entry_id, bytes):
+            return entry_id.decode()
+        return str(entry_id)
+
+    async def publish_batch(self, events: Sequence[TradeEvent]) -> list[str]:
+        """Publish multiple trade events atomically.
+
+        Uses a Redis pipeline for efficiency.
+
+        Args:
+            events: Sequence of TradeEvents to publish.
+
+        Returns:
+            List of entry IDs assigned by Redis.
+        """
+        if not events:
+            return []
+
+        pipe = self._redis.pipeline()
+        for event in events:
+            data = _serialize_trade_event(event)
+            pipe.xadd(self._stream_name, data, maxlen=self._max_len)
+
+        results = await pipe.execute()
+
+        entry_ids: list[str] = []
+        for entry_id in results:
+            if isinstance(entry_id, bytes):
+                entry_ids.append(entry_id.decode())
+            else:
+                entry_ids.append(str(entry_id))
+
+        return entry_ids
+
+    async def create_consumer_group(
+        self,
+        group_name: str,
+        start_id: str = "0",
+        *,
+        mkstream: bool = True,
+    ) -> None:
+        """Create a consumer group for the stream.
+
+        Args:
+            group_name: Name of the consumer group.
+            start_id: ID to start reading from ("0" = beginning, "$" = new only).
+            mkstream: Create the stream if it doesn't exist.
+
+        Raises:
+            ConsumerGroupExistsError: If the group already exists.
+        """
+        try:
+            await self._redis.xgroup_create(
+                self._stream_name,
+                group_name,
+                id=start_id,
+                mkstream=mkstream,
+            )
+            logger.info(f"Created consumer group '{group_name}' on stream '{self._stream_name}'")
+        except ResponseError as e:
+            if "BUSYGROUP" in str(e):
+                raise ConsumerGroupExistsError(
+                    f"Consumer group '{group_name}' already exists"
+                ) from e
+            raise
+
+    async def ensure_consumer_group(
+        self,
+        group_name: str,
+        start_id: str = "0",
+    ) -> bool:
+        """Ensure a consumer group exists, creating it if needed.
+
+        Args:
+            group_name: Name of the consumer group.
+            start_id: ID to start reading from if creating.
+
+        Returns:
+            True if the group was created, False if it already existed.
+        """
+        try:
+            await self.create_consumer_group(group_name, start_id)
+            return True
+        except ConsumerGroupExistsError:
+            return False
+
+    async def read_events(
+        self,
+        group_name: str,
+        consumer_name: str,
+        *,
+        count: int = DEFAULT_COUNT,
+        block_ms: int = DEFAULT_BLOCK_MS,
+    ) -> list[StreamEntry]:
+        """Read events from the stream as a consumer.
+
+        Args:
+            group_name: Consumer group name.
+            consumer_name: Name of this consumer within the group.
+            count: Maximum number of entries to read.
+            block_ms: Milliseconds to block waiting for new entries.
+
+        Returns:
+            List of StreamEntry with entry ID and TradeEvent.
+        """
+        # Read new entries (> means entries not delivered to this consumer)
+        results = await self._redis.xreadgroup(
+            group_name,
+            consumer_name,
+            {self._stream_name: ">"},
+            count=count,
+            block=block_ms,
+        )
+
+        entries: list[StreamEntry] = []
+        if not results:
+            return entries
+
+        # Results format: [[stream_name, [(entry_id, data), ...]]]
+        for _stream_name, stream_entries in results:
+            for entry_id, data in stream_entries:
+                # Decode entry_id
+                entry_id_str = entry_id.decode() if isinstance(entry_id, bytes) else str(entry_id)
+
+                try:
+                    event = _deserialize_trade_event(data)
+                    entries.append(StreamEntry(entry_id=entry_id_str, event=event))
+                except Exception as e:
+                    logger.warning(f"Failed to deserialize entry {entry_id_str}: {e}")
+
+        return entries
+
+    async def read_pending(
+        self,
+        group_name: str,
+        consumer_name: str,
+        *,
+        count: int = DEFAULT_COUNT,
+    ) -> list[StreamEntry]:
+        """Read pending (unacknowledged) entries for a consumer.
+
+        This is useful for recovering from crashes - entries that were
+        delivered but not acknowledged will be re-read.
+
+        Args:
+            group_name: Consumer group name.
+            consumer_name: Name of this consumer.
+            count: Maximum number of entries to read.
+
+        Returns:
+            List of pending StreamEntry.
+        """
+        # Read pending entries (0 means all pending entries)
+        results = await self._redis.xreadgroup(
+            group_name,
+            consumer_name,
+            {self._stream_name: "0"},
+            count=count,
+        )
+
+        entries: list[StreamEntry] = []
+        if not results:
+            return entries
+
+        for _stream_name, stream_entries in results:
+            for entry_id, data in stream_entries:
+                entry_id_str = entry_id.decode() if isinstance(entry_id, bytes) else str(entry_id)
+
+                # Skip entries with no data (already acked)
+                if not data:
+                    continue
+
+                try:
+                    event = _deserialize_trade_event(data)
+                    entries.append(StreamEntry(entry_id=entry_id_str, event=event))
+                except Exception as e:
+                    logger.warning(f"Failed to deserialize pending entry {entry_id_str}: {e}")
+
+        return entries
+
+    async def ack(self, group_name: str, *entry_ids: str) -> int:
+        """Acknowledge that entries have been processed.
+
+        Args:
+            group_name: Consumer group name.
+            *entry_ids: Entry IDs to acknowledge.
+
+        Returns:
+            Number of entries acknowledged.
+        """
+        if not entry_ids:
+            return 0
+        return await self._redis.xack(self._stream_name, group_name, *entry_ids)
+
+    async def get_stream_info(self) -> dict[str, Any]:
+        """Get information about the stream.
+
+        Returns:
+            Dictionary with stream info (length, groups, etc.).
+        """
+        try:
+            info = await self._redis.xinfo_stream(self._stream_name)
+            return dict(info) if info else {}
+        except ResponseError:
+            return {}
+
+    async def get_stream_length(self) -> int:
+        """Get the current length of the stream.
+
+        Returns:
+            Number of entries in the stream.
+        """
+        return await self._redis.xlen(self._stream_name)
+
+    async def trim_stream(self, max_len: int | None = None) -> int:
+        """Trim the stream to a maximum length.
+
+        Args:
+            max_len: Maximum entries to keep (uses default if not specified).
+
+        Returns:
+            Number of entries removed.
+        """
+        length = max_len or self._max_len
+        return await self._redis.xtrim(self._stream_name, maxlen=length)

src/polymarket_insider_tracker/ingestor/websocket.py

@@ -0,0 +1,338 @@
+"""WebSocket client for streaming Polymarket trade events."""
+
+import asyncio
+import json
+import logging
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+import websockets
+from websockets.asyncio.client import ClientConnection
+
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+
+logger = logging.getLogger(__name__)
+
+# Constants
+DEFAULT_WS_HOST = "wss://ws-live-data.polymarket.com"
+DEFAULT_PING_INTERVAL = 30  # seconds
+DEFAULT_MAX_RECONNECT_DELAY = 30  # seconds
+DEFAULT_INITIAL_RECONNECT_DELAY = 1  # seconds
+
+
+class ConnectionState(Enum):
+    """WebSocket connection states."""
+
+    DISCONNECTED = "disconnected"
+    CONNECTING = "connecting"
+    CONNECTED = "connected"
+    RECONNECTING = "reconnecting"
+
+
+@dataclass
+class StreamStats:
+    """Statistics about the trade stream."""
+
+    trades_received: int = 0
+    reconnect_count: int = 0
+    last_trade_time: float | None = None
+    connected_since: float | None = None
+    last_error: str | None = None
+
+
+class TradeStreamError(Exception):
+    """Base exception for trade stream errors."""
+
+
+class ConnectionError(TradeStreamError):
+    """Raised when connection to WebSocket fails."""
+
+
+TradeCallback = Callable[[TradeEvent], Awaitable[None]]
+StateCallback = Callable[[ConnectionState], Awaitable[None]]
+
+
+class TradeStreamHandler:
+    """WebSocket client for streaming Polymarket trade events.
+
+    This handler maintains a persistent connection to Polymarket's real-time
+    trade feed, automatically reconnecting on disconnection with exponential
+    backoff.
+
+    Example:
+        >>> async def on_trade(trade: TradeEvent):
+        ...     print(f"Trade: {trade.side} {trade.size} @ {trade.price}")
+        ...
+        >>> handler = TradeStreamHandler(on_trade=on_trade)
+        >>> await handler.start()  # Blocks until stop() is called
+
+    Attributes:
+        state: Current connection state.
+        stats: Statistics about the stream (trades received, reconnects, etc.).
+    """
+
+    def __init__(
+        self,
+        on_trade: TradeCallback,
+        *,
+        host: str = DEFAULT_WS_HOST,
+        on_state_change: StateCallback | None = None,
+        ping_interval: int = DEFAULT_PING_INTERVAL,
+        max_reconnect_delay: int = DEFAULT_MAX_RECONNECT_DELAY,
+        initial_reconnect_delay: int = DEFAULT_INITIAL_RECONNECT_DELAY,
+        event_filter: str | None = None,
+        market_filter: str | None = None,
+    ) -> None:
+        """Initialize the trade stream handler.
+
+        Args:
+            on_trade: Async callback invoked for each trade event.
+            host: WebSocket endpoint URL.
+            on_state_change: Optional callback for connection state changes.
+            ping_interval: Seconds between heartbeat pings.
+            max_reconnect_delay: Maximum delay between reconnection attempts.
+            initial_reconnect_delay: Initial delay for reconnection backoff.
+            event_filter: Optional event slug to filter trades by event.
+            market_filter: Optional market slug to filter trades by market.
+        """
+        self._on_trade = on_trade
+        self._on_state_change = on_state_change
+        self._host = host
+        self._ping_interval = ping_interval
+        self._max_reconnect_delay = max_reconnect_delay
+        self._initial_reconnect_delay = initial_reconnect_delay
+        self._event_filter = event_filter
+        self._market_filter = market_filter
+
+        self._state = ConnectionState.DISCONNECTED
+        self._stats = StreamStats()
+        self._ws: ClientConnection | None = None
+        self._running = False
+        self._stop_event: asyncio.Event | None = None
+
+    @property
+    def state(self) -> ConnectionState:
+        """Current connection state."""
+        return self._state
+
+    @property
+    def stats(self) -> StreamStats:
+        """Stream statistics."""
+        return self._stats
+
+    async def _set_state(self, new_state: ConnectionState) -> None:
+        """Update state and notify callback."""
+        if self._state != new_state:
+            old_state = self._state
+            self._state = new_state
+            logger.info("Connection state: %s -> %s", old_state.value, new_state.value)
+
+            if self._on_state_change:
+                try:
+                    await self._on_state_change(new_state)
+                except Exception as e:
+                    logger.error("Error in state change callback: %s", e)
+
+    def _build_subscription_message(self) -> dict[str, Any]:
+        """Build the WebSocket subscription message."""
+        subscription: dict[str, Any] = {
+            "topic": "activity",
+            "type": "trades",
+        }
+
+        # Add filters if specified
+        if self._event_filter:
+            subscription["filters"] = json.dumps({"event_slug": self._event_filter})
+        elif self._market_filter:
+            subscription["filters"] = json.dumps({"market_slug": self._market_filter})
+
+        return {"subscriptions": [subscription]}
+
+    async def _connect(self) -> ClientConnection:
+        """Establish WebSocket connection."""
+        await self._set_state(ConnectionState.CONNECTING)
+
+        try:
+            ws = await websockets.connect(
+                self._host,
+                ping_interval=self._ping_interval,
+                ping_timeout=self._ping_interval * 2,
+            )
+
+            # Send subscription message
+            subscribe_msg = self._build_subscription_message()
+            await ws.send(json.dumps(subscribe_msg))
+
+            logger.info("Connected to %s and subscribed to trades", self._host)
+            await self._set_state(ConnectionState.CONNECTED)
+            self._stats.connected_since = time.time()
+
+            return ws
+
+        except Exception as e:
+            logger.error("Failed to connect: %s", e)
+            self._stats.last_error = str(e)
+            raise ConnectionError(f"Failed to connect to {self._host}: {e}") from e
+
+    async def _handle_message(self, message: str) -> None:
+        """Parse and process an incoming WebSocket message."""
+        try:
+            data = json.loads(message)
+
+            # Check if this is a trade message
+            topic = data.get("topic")
+            msg_type = data.get("type")
+
+            if topic == "activity" and msg_type == "trades":
+                payload = data.get("payload", {})
+                trade = TradeEvent.from_websocket_message(payload)
+
+                self._stats.trades_received += 1
+                self._stats.last_trade_time = time.time()
+
+                logger.debug(
+                    "Trade: %s %s @ %s on %s",
+                    trade.side,
+                    trade.size,
+                    trade.price,
+                    trade.market_slug,
+                )
+
+                try:
+                    await self._on_trade(trade)
+                except Exception as e:
+                    logger.error("Error in trade callback: %s", e)
+
+            else:
+                # Log other message types for debugging
+                logger.debug("Received message: topic=%s type=%s", topic, msg_type)
+
+        except json.JSONDecodeError as e:
+            logger.warning("Invalid JSON message: %s", e)
+        except Exception as e:
+            logger.error("Error processing message: %s", e)
+
+    async def _listen(self, ws: ClientConnection) -> None:
+        """Listen for messages on the WebSocket."""
+        try:
+            async for message in ws:
+                if not self._running:
+                    break
+
+                if isinstance(message, str):
+                    await self._handle_message(message)
+                else:
+                    logger.debug("Received binary message (%d bytes)", len(message))
+
+        except websockets.ConnectionClosed as e:
+            logger.warning("Connection closed: %s", e)
+            raise
+        except Exception as e:
+            logger.error("Error in message loop: %s", e)
+            raise
+
+    async def _reconnect_loop(self) -> None:
+        """Handle reconnection with exponential backoff."""
+        delay = self._initial_reconnect_delay
+
+        while self._running:
+            try:
+                await self._set_state(ConnectionState.RECONNECTING)
+
+                logger.info("Reconnecting in %.1f seconds...", delay)
+                await asyncio.sleep(delay)
+
+                if not self._running:
+                    break
+
+                self._ws = await self._connect()
+                self._stats.reconnect_count += 1
+                delay = self._initial_reconnect_delay  # Reset delay on success
+                return
+
+            except Exception as e:
+                logger.error("Reconnection failed: %s", e)
+                self._stats.last_error = str(e)
+
+                # Exponential backoff with jitter
+                delay = min(delay * 2, self._max_reconnect_delay)
+
+    async def start(self) -> None:
+        """Connect and begin streaming trades.
+
+        This method blocks until stop() is called. It automatically
+        handles reconnection on disconnection.
+
+        Raises:
+            ConnectionError: If initial connection fails.
+        """
+        if self._running:
+            logger.warning("Handler already running")
+            return
+
+        self._running = True
+        self._stop_event = asyncio.Event()
+
+        try:
+            # Initial connection
+            self._ws = await self._connect()
+
+            # Main loop
+            while self._running:
+                try:
+                    await self._listen(self._ws)
+                except (websockets.ConnectionClosed, Exception) as e:
+                    if not self._running:
+                        break
+
+                    logger.warning("Connection lost: %s", e)
+                    await self._set_state(ConnectionState.DISCONNECTED)
+
+                    # Attempt reconnection
+                    await self._reconnect_loop()
+
+                    if not self._running or self._ws is None:
+                        break
+
+        finally:
+            await self._cleanup()
+
+    async def stop(self) -> None:
+        """Gracefully disconnect from the WebSocket.
+
+        This signals the handler to stop and cleanly close the connection.
+        """
+        if not self._running:
+            return
+
+        logger.info("Stopping trade stream handler...")
+        self._running = False
+
+        if self._stop_event:
+            self._stop_event.set()
+
+        await self._cleanup()
+
+    async def _cleanup(self) -> None:
+        """Clean up resources."""
+        if self._ws:
+            try:
+                await self._ws.close()
+            except Exception as e:
+                logger.debug("Error closing WebSocket: %s", e)
+            finally:
+                self._ws = None
+
+        await self._set_state(ConnectionState.DISCONNECTED)
+        logger.info("Trade stream handler stopped")
+
+    async def __aenter__(self) -> "TradeStreamHandler":
+        """Async context manager entry."""
+        return self
+
+    async def __aexit__(self, *args: Any) -> None:
+        """Async context manager exit."""
+        await self.stop()

src/polymarket_insider_tracker/profiler/__init__.py

@@ -0,0 +1,30 @@
+"""Wallet profiler - Blockchain analysis for trader intelligence."""
+
+from polymarket_insider_tracker.profiler.analyzer import (
+    WalletAnalyzer,
+)
+from polymarket_insider_tracker.profiler.chain import (
+    PolygonClient,
+    PolygonClientError,
+    RateLimitError,
+    RPCError,
+)
+from polymarket_insider_tracker.profiler.models import (
+    Transaction,
+    WalletInfo,
+    WalletProfile,
+)
+
+__all__ = [
+    # Analyzer
+    "WalletAnalyzer",
+    # Polygon Client
+    "PolygonClient",
+    "PolygonClientError",
+    "RateLimitError",
+    "RPCError",
+    # Models
+    "Transaction",
+    "WalletInfo",
+    "WalletProfile",
+]

src/polymarket_insider_tracker/profiler/analyzer.py

@@ -0,0 +1,283 @@
+"""Wallet analysis for fresh wallet detection.
+
+This module provides wallet analysis capabilities to identify potentially
+suspicious wallets based on their on-chain activity patterns.
+"""
+
+import json
+import logging
+from datetime import UTC, datetime
+from decimal import Decimal
+
+from redis.asyncio import Redis
+
+from polymarket_insider_tracker.profiler.chain import PolygonClient
+from polymarket_insider_tracker.profiler.models import WalletProfile
+
+logger = logging.getLogger(__name__)
+
+# USDC contract address on Polygon
+USDC_POLYGON_ADDRESS = "0x3c499c542cEF5E3811e1192ce70d8cC03d5c3359"
+
+# Default configuration
+DEFAULT_FRESH_THRESHOLD = 5  # Max nonce to be considered fresh
+DEFAULT_PROFILE_CACHE_TTL = 300  # 5 minutes
+
+
+class WalletAnalyzer:
+    """Analyzes wallets to detect fresh wallet patterns.
+
+    This class provides wallet analysis functionality including:
+    - Fresh wallet detection based on transaction count
+    - Wallet age calculation from first transaction
+    - Balance queries for MATIC and USDC
+    - Caching of analysis results
+
+    Example:
+        ```python
+        client = PolygonClient("https://polygon-rpc.com", redis=redis)
+        analyzer = WalletAnalyzer(client, redis=redis)
+
+        # Full analysis
+        profile = await analyzer.analyze("0x...")
+        print(f"Fresh: {profile.is_fresh}, Score: {profile.freshness_score}")
+
+        # Quick check
+        is_fresh = await analyzer.is_fresh("0x...")
+        ```
+    """
+
+    def __init__(
+        self,
+        polygon_client: PolygonClient,
+        *,
+        redis: Redis | None = None,
+        fresh_threshold: int = DEFAULT_FRESH_THRESHOLD,
+        cache_ttl_seconds: int = DEFAULT_PROFILE_CACHE_TTL,
+        usdc_address: str = USDC_POLYGON_ADDRESS,
+    ) -> None:
+        """Initialize the wallet analyzer.
+
+        Args:
+            polygon_client: PolygonClient for blockchain queries.
+            redis: Optional Redis client for caching profiles.
+            fresh_threshold: Maximum nonce to be considered fresh.
+            cache_ttl_seconds: How long to cache analysis results.
+            usdc_address: USDC token contract address on Polygon.
+        """
+        self._client = polygon_client
+        self._redis = redis
+        self._fresh_threshold = fresh_threshold
+        self._cache_ttl = cache_ttl_seconds
+        self._usdc_address = usdc_address
+        self._cache_prefix = "wallet_profile:"
+
+    def _cache_key(self, address: str) -> str:
+        """Generate cache key for wallet profile."""
+        return f"{self._cache_prefix}{address.lower()}"
+
+    async def _get_cached_profile(self, address: str) -> WalletProfile | None:
+        """Get cached profile if available."""
+        if not self._redis:
+            return None
+
+        try:
+            key = self._cache_key(address)
+            cached = await self._redis.get(key)
+            if cached is None:
+                return None
+
+            data = json.loads(cached if isinstance(cached, str) else cached.decode())
+            return WalletProfile(
+                address=data["address"],
+                nonce=data["nonce"],
+                first_seen=datetime.fromisoformat(data["first_seen"]) if data["first_seen"] else None,
+                age_hours=data["age_hours"],
+                is_fresh=data["is_fresh"],
+                total_tx_count=data["total_tx_count"],
+                matic_balance=Decimal(data["matic_balance"]),
+                usdc_balance=Decimal(data["usdc_balance"]),
+                analyzed_at=datetime.fromisoformat(data["analyzed_at"]),
+                fresh_threshold=data["fresh_threshold"],
+            )
+        except Exception as e:
+            logger.warning("Failed to get cached profile for %s: %s", address, e)
+            return None
+
+    async def _cache_profile(self, profile: WalletProfile) -> None:
+        """Cache a wallet profile."""
+        if not self._redis:
+            return
+
+        try:
+            key = self._cache_key(profile.address)
+            data = {
+                "address": profile.address,
+                "nonce": profile.nonce,
+                "first_seen": profile.first_seen.isoformat() if profile.first_seen else None,
+                "age_hours": profile.age_hours,
+                "is_fresh": profile.is_fresh,
+                "total_tx_count": profile.total_tx_count,
+                "matic_balance": str(profile.matic_balance),
+                "usdc_balance": str(profile.usdc_balance),
+                "analyzed_at": profile.analyzed_at.isoformat(),
+                "fresh_threshold": profile.fresh_threshold,
+            }
+            await self._redis.set(key, json.dumps(data), ex=self._cache_ttl)
+        except Exception as e:
+            logger.warning("Failed to cache profile for %s: %s", profile.address, e)
+
+    async def analyze(
+        self,
+        address: str,
+        *,
+        force_refresh: bool = False,
+    ) -> WalletProfile:
+        """Analyze a wallet and return its profile.
+
+        This method performs a comprehensive analysis of the wallet including:
+        - Transaction count (nonce)
+        - First transaction timestamp and wallet age
+        - MATIC and USDC balances
+        - Fresh wallet determination
+
+        Args:
+            address: Wallet address to analyze.
+            force_refresh: If True, bypass cache and re-analyze.
+
+        Returns:
+            WalletProfile with analysis results.
+        """
+        address = address.lower()
+
+        # Check cache unless force refresh
+        if not force_refresh:
+            cached = await self._get_cached_profile(address)
+            if cached is not None:
+                logger.debug("Using cached profile for %s", address)
+                return cached
+
+        # Get wallet info from blockchain
+        wallet_info = await self._client.get_wallet_info(address)
+
+        # Get USDC balance
+        try:
+            usdc_balance = await self._client.get_token_balance(address, self._usdc_address)
+        except Exception as e:
+            logger.warning("Failed to get USDC balance for %s: %s", address, e)
+            usdc_balance = Decimal(0)
+
+        # Calculate age from first transaction
+        first_seen: datetime | None = None
+        age_hours: float | None = None
+
+        if wallet_info.first_transaction is not None:
+            first_seen = wallet_info.first_transaction.timestamp
+            now = datetime.now(UTC)
+            delta = now - first_seen
+            age_hours = delta.total_seconds() / 3600
+
+        # Determine if fresh
+        is_fresh = self._is_wallet_fresh(wallet_info.transaction_count, age_hours)
+
+        # Build profile
+        profile = WalletProfile(
+            address=address,
+            nonce=wallet_info.transaction_count,
+            first_seen=first_seen,
+            age_hours=age_hours,
+            is_fresh=is_fresh,
+            total_tx_count=wallet_info.transaction_count,
+            matic_balance=wallet_info.balance_wei,
+            usdc_balance=usdc_balance,
+            fresh_threshold=self._fresh_threshold,
+        )
+
+        # Cache the result
+        await self._cache_profile(profile)
+
+        return profile
+
+    def _is_wallet_fresh(self, nonce: int, age_hours: float | None) -> bool:
+        """Determine if wallet should be considered fresh.
+
+        A wallet is fresh if:
+        - Transaction count (nonce) is below the threshold
+        - AND either age is unknown OR age is less than 48 hours
+
+        Args:
+            nonce: Transaction count.
+            age_hours: Wallet age in hours, or None if unknown.
+
+        Returns:
+            True if wallet is fresh.
+        """
+        # Must have few transactions
+        if nonce >= self._fresh_threshold:
+            return False
+
+        # If age is known, must be recent (within 48 hours)
+        return not (age_hours is not None and age_hours > 48)
+
+    async def is_fresh(self, address: str) -> bool:
+        """Quick check if wallet is fresh.
+
+        This is a convenience method that returns just the freshness status.
+        It uses cached data if available.
+
+        Args:
+            address: Wallet address to check.
+
+        Returns:
+            True if wallet is fresh.
+        """
+        profile = await self.analyze(address)
+        return profile.is_fresh
+
+    async def analyze_batch(
+        self,
+        addresses: list[str],
+        *,
+        force_refresh: bool = False,
+    ) -> dict[str, WalletProfile]:
+        """Analyze multiple wallets.
+
+        Analyzes wallets in parallel for efficiency.
+
+        Args:
+            addresses: List of wallet addresses to analyze.
+            force_refresh: If True, bypass cache for all wallets.
+
+        Returns:
+            Dictionary mapping address (lowercase) to WalletProfile.
+        """
+        import asyncio
+
+        results: dict[str, WalletProfile] = {}
+
+        # Analyze all in parallel
+        tasks = [self.analyze(addr, force_refresh=force_refresh) for addr in addresses]
+        profiles = await asyncio.gather(*tasks, return_exceptions=True)
+
+        for addr, profile in zip(addresses, profiles, strict=True):
+            if isinstance(profile, BaseException):
+                logger.warning("Failed to analyze %s: %s", addr, profile)
+                continue
+            results[addr.lower()] = profile
+
+        return results
+
+    async def get_fresh_wallets(
+        self,
+        addresses: list[str],
+    ) -> list[str]:
+        """Filter addresses to only return fresh wallets.
+
+        Args:
+            addresses: List of wallet addresses to check.
+
+        Returns:
+            List of addresses that are fresh wallets.
+        """
+        profiles = await self.analyze_batch(addresses)
+        return [addr for addr, profile in profiles.items() if profile.is_fresh]

src/polymarket_insider_tracker/profiler/chain.py

@@ -0,0 +1,527 @@
+"""Polygon blockchain client with connection pooling and caching.
+
+This module provides a Polygon client for wallet data queries with:
+- Connection pooling for concurrent requests
+- Redis caching to avoid redundant RPC calls
+- Retry logic with exponential backoff
+- Rate limiting to respect provider limits
+- Failover to secondary RPC URL
+"""
+
+import asyncio
+import json
+import logging
+import time
+from collections.abc import Sequence
+from dataclasses import dataclass
+from datetime import datetime
+from decimal import Decimal
+from typing import Any, cast
+
+from redis.asyncio import Redis
+from web3 import AsyncWeb3
+from web3.exceptions import Web3Exception
+from web3.providers import AsyncHTTPProvider
+
+from polymarket_insider_tracker.profiler.models import Transaction, WalletInfo
+
+logger = logging.getLogger(__name__)
+
+# Default configuration
+DEFAULT_CACHE_TTL_SECONDS = 300  # 5 minutes
+DEFAULT_MAX_REQUESTS_PER_SECOND = 25
+DEFAULT_MAX_RETRIES = 3
+DEFAULT_RETRY_DELAY_SECONDS = 1.0
+DEFAULT_CONNECTION_POOL_SIZE = 10
+DEFAULT_REQUEST_TIMEOUT = 30
+
+
+class PolygonClientError(Exception):
+    """Base exception for Polygon client errors."""
+
+
+class RPCError(PolygonClientError):
+    """Raised when RPC call fails."""
+
+
+class RateLimitError(PolygonClientError):
+    """Raised when rate limit is exceeded."""
+
+
+@dataclass
+class RateLimiter:
+    """Token bucket rate limiter."""
+
+    max_tokens: float
+    refill_rate: float  # tokens per second
+    tokens: float
+    last_refill: float
+
+    @classmethod
+    def create(cls, max_requests_per_second: float) -> "RateLimiter":
+        """Create a rate limiter with specified max requests per second."""
+        return cls(
+            max_tokens=max_requests_per_second,
+            refill_rate=max_requests_per_second,
+            tokens=max_requests_per_second,
+            last_refill=time.monotonic(),
+        )
+
+    def _refill(self) -> None:
+        """Refill tokens based on elapsed time."""
+        now = time.monotonic()
+        elapsed = now - self.last_refill
+        self.tokens = min(self.max_tokens, self.tokens + elapsed * self.refill_rate)
+        self.last_refill = now
+
+    async def acquire(self, tokens: float = 1.0) -> None:
+        """Acquire tokens, waiting if necessary."""
+        while True:
+            self._refill()
+            if self.tokens >= tokens:
+                self.tokens -= tokens
+                return
+            # Wait for tokens to refill
+            wait_time = (tokens - self.tokens) / self.refill_rate
+            await asyncio.sleep(wait_time)
+
+
+class PolygonClient:
+    """Polygon blockchain client with caching and rate limiting.
+
+    Provides efficient access to wallet data with:
+    - Connection pooling for concurrent requests
+    - Redis caching with configurable TTL
+    - Rate limiting to respect provider limits
+    - Retry logic with exponential backoff
+    - Failover to secondary RPC
+
+    Example:
+        ```python
+        redis = Redis.from_url("redis://localhost:6379")
+        client = PolygonClient(
+            rpc_url="https://polygon-rpc.com",
+            fallback_rpc_url="https://polygon-bor.publicnode.com",
+            redis=redis,
+        )
+
+        # Get single wallet info
+        nonce = await client.get_transaction_count("0x...")
+
+        # Batch query multiple wallets
+        nonces = await client.get_transaction_counts(["0x...", "0x..."])
+        ```
+    """
+
+    def __init__(
+        self,
+        rpc_url: str,
+        *,
+        fallback_rpc_url: str | None = None,
+        redis: Redis | None = None,
+        cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS,
+        max_requests_per_second: float = DEFAULT_MAX_REQUESTS_PER_SECOND,
+        max_retries: int = DEFAULT_MAX_RETRIES,
+        retry_delay_seconds: float = DEFAULT_RETRY_DELAY_SECONDS,
+    ) -> None:
+        """Initialize the Polygon client.
+
+        Args:
+            rpc_url: Primary Polygon RPC endpoint URL.
+            fallback_rpc_url: Optional fallback RPC URL for failover.
+            redis: Optional Redis client for caching.
+            cache_ttl_seconds: Cache TTL in seconds.
+            max_requests_per_second: Rate limit for RPC calls.
+            max_retries: Maximum retry attempts on failure.
+            retry_delay_seconds: Initial delay between retries.
+        """
+        self._rpc_url = rpc_url
+        self._fallback_rpc_url = fallback_rpc_url
+        self._redis = redis
+        self._cache_ttl = cache_ttl_seconds
+        self._max_retries = max_retries
+        self._retry_delay = retry_delay_seconds
+
+        # Create web3 instances
+        self._w3 = AsyncWeb3(AsyncHTTPProvider(rpc_url))
+        self._w3_fallback: AsyncWeb3[AsyncHTTPProvider] | None = None
+        if fallback_rpc_url:
+            self._w3_fallback = AsyncWeb3(AsyncHTTPProvider(fallback_rpc_url))
+
+        # Rate limiter
+        self._rate_limiter = RateLimiter.create(max_requests_per_second)
+
+        # Track primary RPC health
+        self._primary_healthy = True
+        self._last_primary_check = 0.0
+        self._primary_recovery_interval = 60.0  # Try primary again after 60s
+
+        # Cache key prefix
+        self._cache_prefix = "polygon:"
+
+    def _cache_key(self, key_type: str, address: str) -> str:
+        """Generate a cache key."""
+        return f"{self._cache_prefix}{key_type}:{address.lower()}"
+
+    async def _get_cached(self, key: str) -> str | None:
+        """Get value from cache."""
+        if not self._redis:
+            return None
+        try:
+            value = await self._redis.get(key)
+            if isinstance(value, bytes):
+                return value.decode()
+            return str(value) if value is not None else None
+        except Exception as e:
+            logger.warning("Cache get failed: %s", e)
+            return None
+
+    async def _set_cached(self, key: str, value: str, ttl: int | None = None) -> None:
+        """Set value in cache."""
+        if not self._redis:
+            return
+        try:
+            await self._redis.set(key, value, ex=ttl or self._cache_ttl)
+        except Exception as e:
+            logger.warning("Cache set failed: %s", e)
+
+    def _should_try_primary(self) -> bool:
+        """Check if we should try the primary RPC."""
+        if self._primary_healthy:
+            return True
+        # Periodically retry primary
+        now = time.monotonic()
+        if now - self._last_primary_check > self._primary_recovery_interval:
+            self._last_primary_check = now
+            return True
+        return False
+
+    async def _execute_with_retry(
+        self,
+        func_name: str,
+        *args: Any,
+        **kwargs: Any,
+    ) -> Any:
+        """Execute an RPC call with retry and failover logic.
+
+        Args:
+            func_name: Name of the web3.eth method to call.
+            *args: Positional arguments for the method.
+            **kwargs: Keyword arguments for the method.
+
+        Returns:
+            Result from the RPC call.
+
+        Raises:
+            RPCError: If all retries and failover fail.
+        """
+        await self._rate_limiter.acquire()
+
+        last_error: Exception | None = None
+        delay = self._retry_delay
+
+        # Try primary RPC
+        if self._should_try_primary():
+            for attempt in range(self._max_retries):
+                try:
+                    method = getattr(self._w3.eth, func_name)
+                    result = await method(*args, **kwargs)
+                    self._primary_healthy = True
+                    return result
+                except Web3Exception as e:
+                    last_error = e
+                    logger.warning(
+                        "Primary RPC %s failed (attempt %d/%d): %s",
+                        func_name,
+                        attempt + 1,
+                        self._max_retries,
+                        e,
+                    )
+                    if attempt < self._max_retries - 1:
+                        await asyncio.sleep(delay)
+                        delay *= 2  # Exponential backoff
+
+            # Mark primary as unhealthy
+            self._primary_healthy = False
+            self._last_primary_check = time.monotonic()
+
+        # Try fallback RPC
+        if self._w3_fallback:
+            delay = self._retry_delay
+            for attempt in range(self._max_retries):
+                try:
+                    method = getattr(self._w3_fallback.eth, func_name)
+                    result = await method(*args, **kwargs)
+                    logger.info("Fallback RPC succeeded for %s", func_name)
+                    return result
+                except Web3Exception as e:
+                    last_error = e
+                    logger.warning(
+                        "Fallback RPC %s failed (attempt %d/%d): %s",
+                        func_name,
+                        attempt + 1,
+                        self._max_retries,
+                        e,
+                    )
+                    if attempt < self._max_retries - 1:
+                        await asyncio.sleep(delay)
+                        delay *= 2
+
+        raise RPCError(f"RPC call {func_name} failed after all retries: {last_error}")
+
+    async def get_transaction_count(self, address: str) -> int:
+        """Get wallet transaction count (nonce).
+
+        Args:
+            address: Wallet address.
+
+        Returns:
+            Transaction count.
+        """
+        cache_key = self._cache_key("nonce", address)
+
+        # Check cache
+        cached = await self._get_cached(cache_key)
+        if cached is not None:
+            return int(cached)
+
+        # Query blockchain
+        count = await self._execute_with_retry(
+            "get_transaction_count",
+            AsyncWeb3.to_checksum_address(address),
+        )
+
+        # Cache result
+        await self._set_cached(cache_key, str(count))
+
+        return int(count)
+
+    async def get_transaction_counts(
+        self,
+        addresses: Sequence[str],
+    ) -> dict[str, int]:
+        """Batch get transaction counts for multiple addresses.
+
+        Args:
+            addresses: List of wallet addresses.
+
+        Returns:
+            Dictionary mapping address to transaction count.
+        """
+        if not addresses:
+            return {}
+
+        results: dict[str, int] = {}
+        uncached: list[str] = []
+
+        # Check cache for each address
+        for address in addresses:
+            cache_key = self._cache_key("nonce", address)
+            cached = await self._get_cached(cache_key)
+            if cached is not None:
+                results[address.lower()] = int(cached)
+            else:
+                uncached.append(address)
+
+        # Query uncached addresses concurrently
+        if uncached:
+            tasks = [self.get_transaction_count(addr) for addr in uncached]
+            counts = await asyncio.gather(*tasks, return_exceptions=True)
+
+            for addr, count in zip(uncached, counts, strict=True):
+                if isinstance(count, BaseException):
+                    logger.warning("Failed to get nonce for %s: %s", addr, count)
+                    results[addr.lower()] = 0
+                else:
+                    results[addr.lower()] = count
+
+        return results
+
+    async def get_balance(self, address: str) -> Decimal:
+        """Get wallet MATIC balance in Wei.
+
+        Args:
+            address: Wallet address.
+
+        Returns:
+            Balance in Wei as Decimal.
+        """
+        cache_key = self._cache_key("balance", address)
+
+        # Check cache
+        cached = await self._get_cached(cache_key)
+        if cached is not None:
+            return Decimal(cached)
+
+        # Query blockchain
+        balance = await self._execute_with_retry(
+            "get_balance",
+            AsyncWeb3.to_checksum_address(address),
+        )
+
+        # Cache result
+        await self._set_cached(cache_key, str(balance))
+
+        return Decimal(balance)
+
+    async def get_token_balance(
+        self,
+        address: str,
+        token_address: str,
+    ) -> Decimal:
+        """Get ERC20 token balance.
+
+        Args:
+            address: Wallet address.
+            token_address: ERC20 token contract address.
+
+        Returns:
+            Token balance in smallest unit as Decimal.
+        """
+        cache_key = self._cache_key(f"token:{token_address.lower()}", address)
+
+        # Check cache
+        cached = await self._get_cached(cache_key)
+        if cached is not None:
+            return Decimal(cached)
+
+        # ERC20 balanceOf ABI
+        erc20_abi = [
+            {
+                "constant": True,
+                "inputs": [{"name": "_owner", "type": "address"}],
+                "name": "balanceOf",
+                "outputs": [{"name": "balance", "type": "uint256"}],
+                "type": "function",
+            }
+        ]
+
+        await self._rate_limiter.acquire()
+
+        try:
+            w3 = self._w3 if self._primary_healthy else (self._w3_fallback or self._w3)
+            contract = w3.eth.contract(
+                address=AsyncWeb3.to_checksum_address(token_address),
+                abi=erc20_abi,
+            )
+            balance = await contract.functions.balanceOf(
+                AsyncWeb3.to_checksum_address(address)
+            ).call()
+        except Web3Exception as e:
+            raise RPCError(f"Failed to get token balance: {e}") from e
+
+        # Cache result
+        await self._set_cached(cache_key, str(balance))
+
+        return Decimal(balance)
+
+    async def get_block(self, block_number: int) -> dict[str, Any]:
+        """Get block by number.
+
+        Args:
+            block_number: Block number.
+
+        Returns:
+            Block data dictionary.
+        """
+        cache_key = f"{self._cache_prefix}block:{block_number}"
+
+        # Check cache
+        cached = await self._get_cached(cache_key)
+        if cached is not None:
+            return cast(dict[str, Any], json.loads(cached))
+
+        block = await self._execute_with_retry("get_block", block_number)
+
+        # Convert to serializable dict
+        block_dict = dict(block)
+        block_dict["timestamp"] = int(block_dict["timestamp"])
+
+        # Cache result (blocks are immutable, use longer TTL)
+        await self._set_cached(cache_key, json.dumps(block_dict), ttl=3600)
+
+        return dict(block_dict)
+
+    async def get_first_transaction(self, address: str) -> Transaction | None:
+        """Get the first transaction for a wallet.
+
+        This is useful for determining wallet age. Note: This is an expensive
+        operation as it may require scanning transaction history.
+
+        Args:
+            address: Wallet address.
+
+        Returns:
+            First transaction or None if no transactions.
+        """
+        cache_key = self._cache_key("first_tx", address)
+
+        # Check cache
+        cached = await self._get_cached(cache_key)
+        if cached is not None:
+            if cached == "null":
+                return None
+            data = json.loads(cached)
+            return Transaction(
+                hash=data["hash"],
+                block_number=data["block_number"],
+                timestamp=datetime.fromisoformat(data["timestamp"]),
+                from_address=data["from_address"],
+                to_address=data["to_address"],
+                value=Decimal(data["value"]),
+                gas_used=data["gas_used"],
+                gas_price=Decimal(data["gas_price"]),
+            )
+
+        # Check if wallet has any transactions
+        nonce = await self.get_transaction_count(address)
+        if nonce == 0:
+            await self._set_cached(cache_key, "null", ttl=60)  # Short TTL for empty
+            return None
+
+        # Note: Getting the actual first transaction requires using an indexer
+        # or scanning blocks, which is expensive. For now, we'll return None
+        # and recommend using an indexer service for production.
+        logger.warning(
+            "get_first_transaction requires an indexer service for %s (nonce=%d)",
+            address,
+            nonce,
+        )
+        return None
+
+    async def get_wallet_info(self, address: str) -> WalletInfo:
+        """Get aggregated wallet information.
+
+        Args:
+            address: Wallet address.
+
+        Returns:
+            WalletInfo with transaction count, balance, and first transaction.
+        """
+        # Fetch data concurrently
+        nonce_task = self.get_transaction_count(address)
+        balance_task = self.get_balance(address)
+        first_tx_task = self.get_first_transaction(address)
+
+        nonce, balance, first_tx = await asyncio.gather(
+            nonce_task, balance_task, first_tx_task
+        )
+
+        return WalletInfo(
+            address=address.lower(),
+            transaction_count=nonce,
+            balance_wei=balance,
+            first_transaction=first_tx,
+        )
+
+    async def health_check(self) -> bool:
+        """Check if the client can connect to the RPC.
+
+        Returns:
+            True if healthy, False otherwise.
+        """
+        try:
+            await self._execute_with_retry("block_number")
+            return True
+        except RPCError:
+            return False

src/polymarket_insider_tracker/profiler/models.py

@@ -0,0 +1,133 @@
+"""Data models for the profiler module."""
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from decimal import Decimal
+
+
+@dataclass(frozen=True)
+class Transaction:
+    """Represents a blockchain transaction."""
+
+    hash: str
+    block_number: int
+    timestamp: datetime
+    from_address: str
+    to_address: str | None
+    value: Decimal  # In Wei
+    gas_used: int
+    gas_price: Decimal  # In Wei
+
+    @property
+    def value_matic(self) -> Decimal:
+        """Return value in MATIC (10^18 Wei = 1 MATIC)."""
+        return self.value / Decimal("1000000000000000000")
+
+    @property
+    def gas_cost_wei(self) -> Decimal:
+        """Return total gas cost in Wei."""
+        return Decimal(self.gas_used) * self.gas_price
+
+    @property
+    def gas_cost_matic(self) -> Decimal:
+        """Return total gas cost in MATIC."""
+        return self.gas_cost_wei / Decimal("1000000000000000000")
+
+
+@dataclass(frozen=True)
+class WalletInfo:
+    """Aggregated wallet information from blockchain queries."""
+
+    address: str
+    transaction_count: int  # Nonce
+    balance_wei: Decimal
+    first_transaction: Transaction | None = None
+
+    @property
+    def balance_matic(self) -> Decimal:
+        """Return balance in MATIC."""
+        return self.balance_wei / Decimal("1000000000000000000")
+
+    @property
+    def is_fresh(self) -> bool:
+        """Return True if wallet has very few transactions (potential fresh wallet)."""
+        return self.transaction_count < 10
+
+    @property
+    def wallet_age_days(self) -> float | None:
+        """Return wallet age in days based on first transaction."""
+        if self.first_transaction is None:
+            return None
+        delta = datetime.now(tz=self.first_transaction.timestamp.tzinfo) - self.first_transaction.timestamp
+        return delta.total_seconds() / 86400
+
+
+@dataclass(frozen=True)
+class WalletProfile:
+    """Complete wallet analysis profile.
+
+    This is the result of analyzing a wallet's on-chain activity to determine
+    if it exhibits suspicious behavior patterns like fresh wallet trading.
+
+    Attributes:
+        address: The wallet address (lowercase).
+        nonce: Transaction count (number of outgoing transactions).
+        first_seen: Timestamp of first transaction, if available.
+        age_hours: Wallet age in hours since first transaction.
+        is_fresh: True if wallet meets fresh wallet criteria.
+        total_tx_count: Total number of transactions (same as nonce for now).
+        matic_balance: MATIC balance in Wei.
+        usdc_balance: USDC balance in smallest unit (6 decimals).
+        analyzed_at: Timestamp when this profile was created.
+        fresh_threshold: The threshold used to determine freshness.
+    """
+
+    address: str
+    nonce: int
+    first_seen: datetime | None
+    age_hours: float | None
+    is_fresh: bool
+    total_tx_count: int
+    matic_balance: Decimal
+    usdc_balance: Decimal
+    analyzed_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+    fresh_threshold: int = 5
+
+    @property
+    def age_days(self) -> float | None:
+        """Return wallet age in days."""
+        if self.age_hours is None:
+            return None
+        return self.age_hours / 24.0
+
+    @property
+    def matic_balance_formatted(self) -> Decimal:
+        """Return MATIC balance in human-readable format (18 decimals)."""
+        return self.matic_balance / Decimal("1000000000000000000")
+
+    @property
+    def usdc_balance_formatted(self) -> Decimal:
+        """Return USDC balance in human-readable format (6 decimals)."""
+        return self.usdc_balance / Decimal("1000000")
+
+    @property
+    def is_brand_new(self) -> bool:
+        """Return True if wallet has never transacted (nonce = 0)."""
+        return self.nonce == 0
+
+    @property
+    def freshness_score(self) -> float:
+        """Return a 0-1 score where 1 is maximally fresh.
+
+        Score is based on:
+        - Nonce (fewer = fresher)
+        - Age (younger = fresher)
+        """
+        # Nonce component: 1.0 at 0, 0.0 at threshold or higher
+        nonce_score = max(0.0, 1.0 - (self.nonce / self.fresh_threshold))
+
+        # Age component: 1.0 at 0 hours, 0.0 at 48 hours or more
+        age_score = 1.0 if self.age_hours is None else max(0.0, 1.0 - self.age_hours / 48.0)
+
+        # Weighted average: nonce is slightly more important
+        return 0.6 * nonce_score + 0.4 * age_score

src/polymarket_insider_tracker/storage/__init__.py

@@ -0,0 +1 @@
+"""Storage layer - Database schemas and repositories."""

tests/__init__.py

@@ -0,0 +1 @@
+"""Test suite for Polymarket Insider Tracker."""

tests/alerter/__init__.py

@@ -0,0 +1 @@
+"""Tests for alerter module."""

tests/conftest.py

@@ -0,0 +1,9 @@
+"""Pytest configuration and fixtures."""
+
+import pytest
+
+
+@pytest.fixture
+def sample_market_id() -> str:
+    """Sample market ID for testing."""
+    return "0x1234567890abcdef1234567890abcdef12345678"

tests/detector/__init__.py

@@ -0,0 +1 @@
+"""Tests for detector module."""

tests/detector/test_fresh_wallet.py

@@ -0,0 +1,608 @@
+"""Tests for the fresh wallet detector."""
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from unittest.mock import AsyncMock
+
+import pytest
+
+from polymarket_insider_tracker.detector.fresh_wallet import (
+    BASE_CONFIDENCE,
+    BRAND_NEW_BONUS,
+    DEFAULT_MAX_AGE_HOURS,
+    DEFAULT_MAX_NONCE,
+    DEFAULT_MIN_TRADE_SIZE,
+    LARGE_TRADE_BONUS,
+    VERY_YOUNG_BONUS,
+    FreshWalletDetector,
+)
+from polymarket_insider_tracker.detector.models import FreshWalletSignal
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.profiler.models import WalletProfile
+
+
+# Test fixtures
+@pytest.fixture
+def mock_wallet_analyzer():
+    """Create a mock WalletAnalyzer."""
+    return AsyncMock()
+
+
+@pytest.fixture
+def detector(mock_wallet_analyzer):
+    """Create a FreshWalletDetector with mocked analyzer."""
+    return FreshWalletDetector(mock_wallet_analyzer)
+
+
+def create_trade_event(
+    *,
+    wallet_address: str = "0x1234567890123456789012345678901234567890",
+    market_id: str = "market123",
+    trade_id: str = "trade123",
+    price: Decimal = Decimal("0.5"),
+    size: Decimal = Decimal("2000"),
+    side: str = "BUY",
+) -> TradeEvent:
+    """Create a TradeEvent for testing."""
+    return TradeEvent(
+        market_id=market_id,
+        trade_id=trade_id,
+        wallet_address=wallet_address,
+        side=side,
+        outcome="Yes",
+        outcome_index=0,
+        price=price,
+        size=size,
+        timestamp=datetime.now(UTC),
+        asset_id="asset123",
+    )
+
+
+def create_wallet_profile(
+    *,
+    address: str = "0x1234567890123456789012345678901234567890",
+    nonce: int = 2,
+    age_hours: float | None = 24.0,
+    is_fresh: bool = True,
+) -> WalletProfile:
+    """Create a WalletProfile for testing."""
+    first_seen = None
+    if age_hours is not None:
+        first_seen = datetime.now(UTC) - timedelta(hours=age_hours)
+
+    return WalletProfile(
+        address=address.lower(),
+        nonce=nonce,
+        first_seen=first_seen,
+        age_hours=age_hours,
+        is_fresh=is_fresh,
+        total_tx_count=nonce,
+        matic_balance=Decimal("1000000000000000000"),  # 1 MATIC
+        usdc_balance=Decimal("1000000000"),  # 1000 USDC
+        fresh_threshold=5,
+    )
+
+
+# === Test FreshWalletSignal Model ===
+
+
+class TestFreshWalletSignal:
+    """Tests for FreshWalletSignal dataclass."""
+
+    def test_wallet_address_property(self):
+        """Test wallet_address property returns trade wallet."""
+        trade = create_trade_event(wallet_address="0xabc123")
+        profile = create_wallet_profile(address="0xabc123")
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.7,
+            factors={"base": 0.5},
+        )
+        assert signal.wallet_address == "0xabc123"
+
+    def test_market_id_property(self):
+        """Test market_id property returns trade market."""
+        trade = create_trade_event(market_id="market456")
+        profile = create_wallet_profile()
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.7,
+            factors={"base": 0.5},
+        )
+        assert signal.market_id == "market456"
+
+    def test_trade_size_usdc_property(self):
+        """Test trade_size_usdc returns notional value."""
+        trade = create_trade_event(price=Decimal("0.5"), size=Decimal("2000"))
+        profile = create_wallet_profile()
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.7,
+            factors={"base": 0.5},
+        )
+        assert signal.trade_size_usdc == Decimal("1000")
+
+    def test_is_high_confidence(self):
+        """Test is_high_confidence threshold."""
+        trade = create_trade_event()
+        profile = create_wallet_profile()
+
+        # At threshold
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.7,
+            factors={},
+        )
+        assert signal.is_high_confidence is True
+
+        # Below threshold
+        signal_low = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.69,
+            factors={},
+        )
+        assert signal_low.is_high_confidence is False
+
+    def test_is_very_high_confidence(self):
+        """Test is_very_high_confidence threshold."""
+        trade = create_trade_event()
+        profile = create_wallet_profile()
+
+        # At threshold
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.85,
+            factors={},
+        )
+        assert signal.is_very_high_confidence is True
+
+        # Below threshold
+        signal_low = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.84,
+            factors={},
+        )
+        assert signal_low.is_very_high_confidence is False
+
+    def test_to_dict(self):
+        """Test serialization to dictionary."""
+        trade = create_trade_event(
+            wallet_address="0xtest",
+            market_id="market1",
+            trade_id="tx1",
+            price=Decimal("0.6"),
+            size=Decimal("5000"),
+            side="BUY",
+        )
+        profile = create_wallet_profile(address="0xtest", nonce=0, age_hours=1.0)
+        signal = FreshWalletSignal(
+            trade_event=trade,
+            wallet_profile=profile,
+            confidence=0.8,
+            factors={"base": 0.5, "brand_new": 0.2},
+        )
+
+        result = signal.to_dict()
+
+        assert result["wallet_address"] == "0xtest"
+        assert result["market_id"] == "market1"
+        assert result["trade_id"] == "tx1"
+        assert result["trade_size"] == "3000.0"
+        assert result["trade_side"] == "BUY"
+        assert result["trade_price"] == "0.6"
+        assert result["wallet_nonce"] == 0
+        assert result["wallet_age_hours"] == 1.0
+        assert result["wallet_is_fresh"] is True
+        assert result["confidence"] == 0.8
+        assert result["factors"] == {"base": 0.5, "brand_new": 0.2}
+        assert "timestamp" in result
+
+
+# === Test FreshWalletDetector Initialization ===
+
+
+class TestFreshWalletDetectorInit:
+    """Tests for FreshWalletDetector initialization."""
+
+    def test_default_config(self, mock_wallet_analyzer):
+        """Test detector initializes with default config."""
+        detector = FreshWalletDetector(mock_wallet_analyzer)
+        assert detector._min_trade_size == DEFAULT_MIN_TRADE_SIZE
+        assert detector._max_nonce == DEFAULT_MAX_NONCE
+        assert detector._max_age_hours == DEFAULT_MAX_AGE_HOURS
+
+    def test_custom_config(self, mock_wallet_analyzer):
+        """Test detector accepts custom config."""
+        detector = FreshWalletDetector(
+            mock_wallet_analyzer,
+            min_trade_size=Decimal("5000"),
+            max_nonce=10,
+            max_age_hours=72.0,
+        )
+        assert detector._min_trade_size == Decimal("5000")
+        assert detector._max_nonce == 10
+        assert detector._max_age_hours == 72.0
+
+
+# === Test FreshWalletDetector.analyze ===
+
+
+class TestFreshWalletDetectorAnalyze:
+    """Tests for FreshWalletDetector.analyze method."""
+
+    async def test_filters_small_trades(self, detector, mock_wallet_analyzer):
+        """Test that trades below minimum size are filtered out."""
+        trade = create_trade_event(
+            price=Decimal("0.1"),
+            size=Decimal("100"),  # notional = $10
+        )
+
+        result = await detector.analyze(trade)
+
+        assert result is None
+        mock_wallet_analyzer.analyze.assert_not_called()
+
+    async def test_detects_fresh_wallet(self, detector, mock_wallet_analyzer):
+        """Test detection of fresh wallet trade."""
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("2000"),  # notional = $1000
+        )
+        profile = create_wallet_profile(nonce=2, age_hours=24.0, is_fresh=True)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is not None
+        assert isinstance(result, FreshWalletSignal)
+        assert result.trade_event == trade
+        assert result.wallet_profile == profile
+        assert result.confidence >= BASE_CONFIDENCE
+
+    async def test_filters_non_fresh_wallet(self, detector, mock_wallet_analyzer):
+        """Test that non-fresh wallets are filtered out."""
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("4000"),  # notional = $2000
+        )
+        profile = create_wallet_profile(nonce=10, age_hours=100.0, is_fresh=False)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is None
+
+    async def test_handles_analyzer_error(self, detector, mock_wallet_analyzer):
+        """Test graceful handling of analyzer errors."""
+        trade = create_trade_event()
+        mock_wallet_analyzer.analyze.side_effect = Exception("RPC error")
+
+        result = await detector.analyze(trade)
+
+        assert result is None
+
+    async def test_wallet_at_nonce_threshold(self, detector, mock_wallet_analyzer):
+        """Test wallet exactly at max_nonce threshold."""
+        trade = create_trade_event(size=Decimal("3000"))
+        profile = create_wallet_profile(nonce=5, age_hours=24.0)  # At threshold
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is not None
+
+    async def test_wallet_above_nonce_threshold(self, detector, mock_wallet_analyzer):
+        """Test wallet above max_nonce threshold."""
+        trade = create_trade_event(size=Decimal("3000"))
+        profile = create_wallet_profile(nonce=6, age_hours=24.0)  # Above threshold
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is None
+
+    async def test_wallet_at_age_threshold(self, detector, mock_wallet_analyzer):
+        """Test wallet exactly at max_age_hours threshold."""
+        trade = create_trade_event(size=Decimal("3000"))
+        profile = create_wallet_profile(nonce=2, age_hours=48.0)  # At threshold
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is not None
+
+    async def test_wallet_above_age_threshold(self, detector, mock_wallet_analyzer):
+        """Test wallet above max_age_hours threshold."""
+        trade = create_trade_event(size=Decimal("3000"))
+        profile = create_wallet_profile(nonce=2, age_hours=49.0)  # Above threshold
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        assert result is None
+
+    async def test_wallet_with_unknown_age(self, detector, mock_wallet_analyzer):
+        """Test wallet with unknown age (None)."""
+        trade = create_trade_event(size=Decimal("3000"))
+        profile = create_wallet_profile(nonce=2, age_hours=None)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        # Should pass since age is unknown
+        assert result is not None
+
+
+# === Test Confidence Scoring ===
+
+
+class TestConfidenceScoring:
+    """Tests for confidence score calculation."""
+
+    def test_base_confidence_only(self, detector):
+        """Test base confidence for fresh wallet."""
+        profile = create_wallet_profile(nonce=3, age_hours=24.0)
+        trade = create_trade_event(size=Decimal("2000"))  # $1000 notional
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        assert confidence == BASE_CONFIDENCE
+        assert factors == {"base": BASE_CONFIDENCE}
+
+    def test_brand_new_bonus(self, detector):
+        """Test bonus for brand new wallet (nonce=0)."""
+        profile = create_wallet_profile(nonce=0, age_hours=24.0)
+        trade = create_trade_event(size=Decimal("2000"))
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        expected = BASE_CONFIDENCE + BRAND_NEW_BONUS
+        assert confidence == expected
+        assert factors["brand_new"] == BRAND_NEW_BONUS
+
+    def test_very_young_bonus(self, detector):
+        """Test bonus for very young wallet (age < 2 hours)."""
+        profile = create_wallet_profile(nonce=3, age_hours=1.5)
+        trade = create_trade_event(size=Decimal("2000"))
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        expected = BASE_CONFIDENCE + VERY_YOUNG_BONUS
+        assert confidence == expected
+        assert factors["very_young"] == VERY_YOUNG_BONUS
+
+    def test_large_trade_bonus(self, detector):
+        """Test bonus for large trade (> $10,000)."""
+        profile = create_wallet_profile(nonce=3, age_hours=24.0)
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("25000"),  # $12,500 notional
+        )
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        expected = BASE_CONFIDENCE + LARGE_TRADE_BONUS
+        assert confidence == expected
+        assert factors["large_trade"] == LARGE_TRADE_BONUS
+
+    def test_all_bonuses_combined(self, detector):
+        """Test all bonuses stacking."""
+        profile = create_wallet_profile(nonce=0, age_hours=0.5)
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("30000"),  # $15,000 notional
+        )
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        expected = BASE_CONFIDENCE + BRAND_NEW_BONUS + VERY_YOUNG_BONUS + LARGE_TRADE_BONUS
+        assert confidence == expected
+        assert factors["brand_new"] == BRAND_NEW_BONUS
+        assert factors["very_young"] == VERY_YOUNG_BONUS
+        assert factors["large_trade"] == LARGE_TRADE_BONUS
+
+    def test_confidence_clamped_to_max(self, detector):
+        """Test confidence is clamped to 1.0 max."""
+        # Create scenario where total would exceed 1.0
+        # BASE=0.5 + BRAND_NEW=0.2 + VERY_YOUNG=0.1 + LARGE_TRADE=0.1 = 0.9
+        # This is less than 1.0, so let's verify clamping works
+        profile = create_wallet_profile(nonce=0, age_hours=0.5)
+        trade = create_trade_event(size=Decimal("30000"))
+
+        confidence, _ = detector.calculate_confidence(profile, trade)
+
+        assert confidence <= 1.0
+
+    def test_no_bonus_at_age_boundary(self, detector):
+        """Test no bonus when age exactly at 2 hours."""
+        profile = create_wallet_profile(nonce=3, age_hours=2.0)
+        trade = create_trade_event(size=Decimal("2000"))
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        assert "very_young" not in factors
+        assert confidence == BASE_CONFIDENCE
+
+    def test_no_bonus_at_trade_size_boundary(self, detector):
+        """Test no bonus when trade exactly at $10,000."""
+        profile = create_wallet_profile(nonce=3, age_hours=24.0)
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("20000"),  # $10,000 notional exactly
+        )
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        # Exactly at threshold should NOT get bonus
+        assert "large_trade" not in factors
+        assert confidence == BASE_CONFIDENCE
+
+    def test_unknown_age_no_young_bonus(self, detector):
+        """Test no young bonus when age is None."""
+        profile = create_wallet_profile(nonce=3, age_hours=None)
+        trade = create_trade_event(size=Decimal("2000"))
+
+        confidence, factors = detector.calculate_confidence(profile, trade)
+
+        assert "very_young" not in factors
+        assert confidence == BASE_CONFIDENCE
+
+
+# === Test Batch Analysis ===
+
+
+class TestBatchAnalysis:
+    """Tests for batch trade analysis."""
+
+    async def test_analyze_batch_success(self, detector, mock_wallet_analyzer):
+        """Test analyzing multiple trades."""
+        trades = [
+            create_trade_event(trade_id="trade1", size=Decimal("3000")),
+            create_trade_event(trade_id="trade2", size=Decimal("4000")),
+        ]
+        profile = create_wallet_profile(nonce=2, age_hours=24.0)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        results = await detector.analyze_batch(trades)
+
+        assert len(results) == 2
+        assert all(isinstance(r, FreshWalletSignal) for r in results)
+
+    async def test_analyze_batch_filters_small(self, detector, mock_wallet_analyzer):
+        """Test batch analysis filters small trades."""
+        trades = [
+            create_trade_event(trade_id="trade1", size=Decimal("3000")),  # Above threshold
+            create_trade_event(trade_id="trade2", size=Decimal("100")),  # Below threshold
+        ]
+        profile = create_wallet_profile(nonce=2, age_hours=24.0)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        results = await detector.analyze_batch(trades)
+
+        assert len(results) == 1
+        assert results[0].trade_event.trade_id == "trade1"
+
+    async def test_analyze_batch_handles_errors(self, detector, mock_wallet_analyzer):
+        """Test batch analysis handles individual errors gracefully."""
+        trades = [
+            create_trade_event(trade_id="trade1", size=Decimal("3000")),
+            create_trade_event(trade_id="trade2", size=Decimal("4000")),
+        ]
+
+        # First call succeeds, second fails
+        profile = create_wallet_profile(nonce=2, age_hours=24.0)
+        mock_wallet_analyzer.analyze.side_effect = [profile, Exception("Error")]
+
+        results = await detector.analyze_batch(trades)
+
+        assert len(results) == 1
+
+    async def test_analyze_batch_empty_list(self, detector):
+        """Test batch analysis with empty list."""
+        results = await detector.analyze_batch([])
+
+        assert results == []
+
+    async def test_analyze_batch_all_filtered(self, detector, mock_wallet_analyzer):
+        """Test batch analysis when all trades are filtered."""
+        trades = [
+            create_trade_event(trade_id="trade1", size=Decimal("100")),
+            create_trade_event(trade_id="trade2", size=Decimal("50")),
+        ]
+
+        results = await detector.analyze_batch(trades)
+
+        assert results == []
+        mock_wallet_analyzer.analyze.assert_not_called()
+
+
+# === Integration-Style Tests ===
+
+
+class TestDetectorIntegration:
+    """Integration-style tests for complete detector flow."""
+
+    async def test_full_detection_flow(self, mock_wallet_analyzer):
+        """Test complete detection flow from trade to signal."""
+        detector = FreshWalletDetector(mock_wallet_analyzer)
+
+        # Create a suspicious trade
+        trade = create_trade_event(
+            wallet_address="0xfresh123",
+            market_id="suspicious_market",
+            price=Decimal("0.65"),
+            size=Decimal("20000"),  # $13,000 notional
+        )
+
+        # Create a fresh wallet profile
+        profile = create_wallet_profile(
+            address="0xfresh123",
+            nonce=0,  # Brand new
+            age_hours=0.5,  # Very young
+        )
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        # Analyze
+        signal = await detector.analyze(trade)
+
+        # Verify signal
+        assert signal is not None
+        assert signal.wallet_address == "0xfresh123"
+        assert signal.market_id == "suspicious_market"
+        assert signal.is_high_confidence is True
+        assert "brand_new" in signal.factors
+        assert "very_young" in signal.factors
+        assert "large_trade" in signal.factors
+
+    async def test_custom_thresholds(self, mock_wallet_analyzer):
+        """Test detection with custom thresholds."""
+        detector = FreshWalletDetector(
+            mock_wallet_analyzer,
+            min_trade_size=Decimal("5000"),
+            max_nonce=3,
+            max_age_hours=24.0,
+        )
+
+        # Trade that would pass default but fails custom thresholds
+        trade = create_trade_event(size=Decimal("8000"))  # $4000 notional
+        profile = create_wallet_profile(nonce=4, age_hours=30.0)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        # Should fail min_trade_size check
+        result = await detector.analyze(trade)
+        assert result is None
+
+    async def test_edge_case_exact_min_trade_size(self, detector, mock_wallet_analyzer):
+        """Test trade exactly at minimum size threshold."""
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("2000"),  # $1000 notional exactly at default
+        )
+        profile = create_wallet_profile(nonce=2)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        # At threshold should pass
+        assert result is not None
+
+    async def test_edge_case_below_min_trade_size(self, detector, mock_wallet_analyzer):
+        """Test trade just below minimum size threshold."""
+        trade = create_trade_event(
+            price=Decimal("0.5"),
+            size=Decimal("1999"),  # $999.50 - just under $1000
+        )
+        profile = create_wallet_profile(nonce=2)
+        mock_wallet_analyzer.analyze.return_value = profile
+
+        result = await detector.analyze(trade)
+
+        # Below threshold should fail
+        assert result is None

tests/ingestor/__init__.py

@@ -0,0 +1 @@
+"""Tests for ingestor module."""

tests/ingestor/test_clob_client.py

@@ -0,0 +1,356 @@
+"""Tests for ClobClient wrapper."""
+
+import time
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from polymarket_insider_tracker.ingestor.clob_client import (
+    ClobClient,
+    ClobClientError,
+    RateLimiter,
+    RetryError,
+    with_retry,
+)
+from polymarket_insider_tracker.ingestor.models import Market, Orderbook
+
+
+class TestRateLimiter:
+    """Tests for RateLimiter."""
+
+    def test_acquire_sync_no_wait_first_call(self) -> None:
+        """First call should not wait."""
+        limiter = RateLimiter(max_requests_per_second=10)
+        start = time.monotonic()
+        limiter.acquire_sync()
+        elapsed = time.monotonic() - start
+
+        # Should be nearly instant
+        assert elapsed < 0.05
+
+    def test_acquire_sync_enforces_rate(self) -> None:
+        """Subsequent calls should be rate limited."""
+        limiter = RateLimiter(max_requests_per_second=10)  # 100ms between calls
+
+        # First call
+        limiter.acquire_sync()
+
+        # Second call should wait
+        start = time.monotonic()
+        limiter.acquire_sync()
+        elapsed = time.monotonic() - start
+
+        # Should wait at least 90ms (allowing some tolerance)
+        assert elapsed >= 0.08
+
+
+class TestWithRetry:
+    """Tests for retry decorator."""
+
+    def test_success_first_try(self) -> None:
+        """Function succeeds on first try."""
+        call_count = 0
+
+        @with_retry(max_retries=3)
+        def succeed() -> str:
+            nonlocal call_count
+            call_count += 1
+            return "success"
+
+        result = succeed()
+
+        assert result == "success"
+        assert call_count == 1
+
+    def test_success_after_retries(self) -> None:
+        """Function succeeds after some retries."""
+        call_count = 0
+
+        @with_retry(max_retries=3, base_delay=0.01)
+        def succeed_eventually() -> str:
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                raise ValueError("Not yet")
+            return "success"
+
+        result = succeed_eventually()
+
+        assert result == "success"
+        assert call_count == 3
+
+    def test_exhausted_retries(self) -> None:
+        """Raises RetryError after exhausting retries."""
+
+        @with_retry(max_retries=2, base_delay=0.01)
+        def always_fails() -> str:
+            raise ValueError("Always fails")
+
+        with pytest.raises(RetryError) as exc_info:
+            always_fails()
+
+        assert "3 attempts failed" in str(exc_info.value)
+        assert isinstance(exc_info.value.last_exception, ValueError)
+
+    def test_specific_exception_types(self) -> None:
+        """Only retries on specified exception types."""
+        call_count = 0
+
+        @with_retry(max_retries=3, base_delay=0.01, retry_on=(ValueError,))
+        def raise_type_error() -> str:
+            nonlocal call_count
+            call_count += 1
+            raise TypeError("Not retried")
+
+        with pytest.raises(TypeError):
+            raise_type_error()
+
+        # Should only be called once since TypeError is not in retry_on
+        assert call_count == 1
+
+
+class TestClobClient:
+    """Tests for ClobClient wrapper."""
+
+    @pytest.fixture
+    def mock_base_client(self) -> MagicMock:
+        """Create a mock base CLOB client."""
+        with patch(
+            "polymarket_insider_tracker.ingestor.clob_client.BaseClobClient"
+        ) as mock:
+            yield mock.return_value
+
+    def test_init_defaults(self, mock_base_client: MagicMock) -> None:
+        """Test client initialization with defaults."""
+        client = ClobClient()
+
+        assert client._host == "https://clob.polymarket.com"
+        assert client._max_retries == 3
+
+    def test_init_with_env_api_key(self, mock_base_client: MagicMock) -> None:
+        """Test client reads API key from environment."""
+        with patch.dict("os.environ", {"POLYMARKET_API_KEY": "test-key"}):
+            client = ClobClient()
+            assert client._api_key == "test-key"
+
+    def test_init_with_explicit_api_key(self, mock_base_client: MagicMock) -> None:
+        """Test client uses explicitly provided API key."""
+        client = ClobClient(api_key="explicit-key")
+        assert client._api_key == "explicit-key"
+
+    def test_health_check_success(self, mock_base_client: MagicMock) -> None:
+        """Test health check returns True when API responds OK."""
+        mock_base_client.get_ok.return_value = "OK"
+
+        client = ClobClient()
+        result = client.health_check()
+
+        assert result is True
+        mock_base_client.get_ok.assert_called_once()
+
+    def test_health_check_failure(self, mock_base_client: MagicMock) -> None:
+        """Test health check returns False on error."""
+        mock_base_client.get_ok.side_effect = Exception("Connection failed")
+
+        client = ClobClient()
+        result = client.health_check()
+
+        assert result is False
+
+    def test_get_server_time(self, mock_base_client: MagicMock) -> None:
+        """Test getting server time."""
+        mock_base_client.get_server_time.return_value = 1704067200000
+
+        client = ClobClient()
+        result = client.get_server_time()
+
+        assert result == 1704067200000
+
+    def test_get_markets(self, mock_base_client: MagicMock) -> None:
+        """Test fetching markets."""
+        mock_base_client.get_simplified_markets.return_value = {
+            "data": [
+                {
+                    "condition_id": "0x123",
+                    "question": "Test market?",
+                    "tokens": [],
+                    "closed": False,
+                },
+            ],
+            "next_cursor": "LTE=",
+        }
+
+        client = ClobClient()
+        markets = client.get_markets()
+
+        assert len(markets) == 1
+        assert isinstance(markets[0], Market)
+        assert markets[0].condition_id == "0x123"
+
+    def test_get_markets_filters_closed(self, mock_base_client: MagicMock) -> None:
+        """Test that closed markets are filtered when active_only=True."""
+        mock_base_client.get_simplified_markets.return_value = {
+            "data": [
+                {"condition_id": "0x1", "closed": False},
+                {"condition_id": "0x2", "closed": True},
+            ],
+            "next_cursor": "LTE=",
+        }
+
+        client = ClobClient()
+        markets = client.get_markets(active_only=True)
+
+        assert len(markets) == 1
+        assert markets[0].condition_id == "0x1"
+
+    def test_get_markets_includes_closed(self, mock_base_client: MagicMock) -> None:
+        """Test that closed markets are included when active_only=False."""
+        mock_base_client.get_simplified_markets.return_value = {
+            "data": [
+                {"condition_id": "0x1", "closed": False},
+                {"condition_id": "0x2", "closed": True},
+            ],
+            "next_cursor": "LTE=",
+        }
+
+        client = ClobClient()
+        markets = client.get_markets(active_only=False)
+
+        assert len(markets) == 2
+
+    def test_get_markets_pagination(self, mock_base_client: MagicMock) -> None:
+        """Test that pagination is handled correctly."""
+        mock_base_client.get_simplified_markets.side_effect = [
+            {
+                "data": [{"condition_id": "0x1"}],
+                "next_cursor": "cursor2",
+            },
+            {
+                "data": [{"condition_id": "0x2"}],
+                "next_cursor": "LTE=",
+            },
+        ]
+
+        client = ClobClient()
+        markets = client.get_markets()
+
+        assert len(markets) == 2
+        assert mock_base_client.get_simplified_markets.call_count == 2
+
+    def test_get_market(self, mock_base_client: MagicMock) -> None:
+        """Test fetching a single market."""
+        mock_base_client.get_market.return_value = {
+            "condition_id": "0xabc",
+            "question": "Will it happen?",
+            "tokens": [
+                {"token_id": "t1", "outcome": "Yes"},
+                {"token_id": "t2", "outcome": "No"},
+            ],
+        }
+
+        client = ClobClient()
+        market = client.get_market("0xabc")
+
+        assert isinstance(market, Market)
+        assert market.condition_id == "0xabc"
+        assert len(market.tokens) == 2
+
+    def test_get_market_not_found(self, mock_base_client: MagicMock) -> None:
+        """Test error handling when market not found."""
+        mock_base_client.get_market.side_effect = Exception("Not found")
+
+        client = ClobClient()
+
+        with pytest.raises(ClobClientError) as exc_info:
+            client.get_market("0xnotfound")
+
+        assert "0xnotfound" in str(exc_info.value)
+
+    def test_get_orderbook(self, mock_base_client: MagicMock) -> None:
+        """Test fetching an orderbook."""
+        mock_bid = MagicMock()
+        mock_bid.price = "0.50"
+        mock_bid.size = "100"
+
+        mock_ask = MagicMock()
+        mock_ask.price = "0.52"
+        mock_ask.size = "150"
+
+        mock_orderbook = MagicMock()
+        mock_orderbook.market = "0xmarket"
+        mock_orderbook.asset_id = "token123"
+        mock_orderbook.tick_size = "0.01"
+        mock_orderbook.bids = [mock_bid]
+        mock_orderbook.asks = [mock_ask]
+
+        mock_base_client.get_order_book.return_value = mock_orderbook
+
+        client = ClobClient()
+        orderbook = client.get_orderbook("token123")
+
+        assert isinstance(orderbook, Orderbook)
+        assert orderbook.asset_id == "token123"
+        assert len(orderbook.bids) == 1
+        assert len(orderbook.asks) == 1
+
+    def test_get_orderbooks(self, mock_base_client: MagicMock) -> None:
+        """Test fetching multiple orderbooks."""
+        mock_ob1 = MagicMock()
+        mock_ob1.market = "m1"
+        mock_ob1.asset_id = "t1"
+        mock_ob1.tick_size = "0.01"
+        mock_ob1.bids = []
+        mock_ob1.asks = []
+
+        mock_ob2 = MagicMock()
+        mock_ob2.market = "m2"
+        mock_ob2.asset_id = "t2"
+        mock_ob2.tick_size = "0.01"
+        mock_ob2.bids = []
+        mock_ob2.asks = []
+
+        mock_base_client.get_order_books.return_value = [mock_ob1, mock_ob2]
+
+        client = ClobClient()
+        orderbooks = client.get_orderbooks(["t1", "t2"])
+
+        assert len(orderbooks) == 2
+        assert all(isinstance(ob, Orderbook) for ob in orderbooks)
+
+    def test_get_midpoint(self, mock_base_client: MagicMock) -> None:
+        """Test fetching midpoint price."""
+        mock_base_client.get_midpoint.return_value = {"mid": "0.55"}
+
+        client = ClobClient()
+        result = client.get_midpoint("token123")
+
+        assert result == "0.55"
+
+    def test_get_midpoint_error(self, mock_base_client: MagicMock) -> None:
+        """Test midpoint returns None on error."""
+        mock_base_client.get_midpoint.side_effect = Exception("API error")
+
+        client = ClobClient()
+        result = client.get_midpoint("token123")
+
+        assert result is None
+
+    def test_get_price_buy(self, mock_base_client: MagicMock) -> None:
+        """Test fetching buy price."""
+        mock_base_client.get_price.return_value = {"price": "0.53"}
+
+        client = ClobClient()
+        result = client.get_price("token123", side="BUY")
+
+        assert result == "0.53"
+        mock_base_client.get_price.assert_called_with("token123", side="BUY")
+
+    def test_get_price_sell(self, mock_base_client: MagicMock) -> None:
+        """Test fetching sell price."""
+        mock_base_client.get_price.return_value = {"price": "0.51"}
+
+        client = ClobClient()
+        result = client.get_price("token123", side="SELL")
+
+        assert result == "0.51"
+        mock_base_client.get_price.assert_called_with("token123", side="SELL")

tests/ingestor/test_health.py

@@ -0,0 +1,617 @@
+"""Tests for the connection health monitor."""
+
+import asyncio
+import time
+from unittest.mock import AsyncMock
+
+import pytest
+from aiohttp import web
+
+from polymarket_insider_tracker.ingestor.health import (
+    DEFAULT_HEALTH_CHECK_INTERVAL,
+    DEFAULT_STALE_THRESHOLD_SECONDS,
+    HealthMonitor,
+    HealthReport,
+    HealthStatus,
+    StreamHealth,
+    StreamStatus,
+)
+
+
+class TestStreamHealth:
+    """Tests for the StreamHealth dataclass."""
+
+    def test_stream_health_defaults(self) -> None:
+        """Test default values."""
+        health = StreamHealth(name="test-stream")
+
+        assert health.name == "test-stream"
+        assert health.status == StreamStatus.DISCONNECTED
+        assert health.last_event_time is None
+        assert health.events_received == 0
+        assert health.events_per_second == 0.0
+        assert health.connected_since is None
+        assert health.last_error is None
+
+    def test_stream_health_custom_values(self) -> None:
+        """Test with custom values."""
+        now = time.time()
+        health = StreamHealth(
+            name="trades",
+            status=StreamStatus.ACTIVE,
+            last_event_time=now,
+            events_received=100,
+            events_per_second=5.0,
+            connected_since=now - 3600,
+            last_error=None,
+        )
+
+        assert health.name == "trades"
+        assert health.status == StreamStatus.ACTIVE
+        assert health.events_received == 100
+
+
+class TestHealthReport:
+    """Tests for the HealthReport dataclass."""
+
+    def test_health_report_defaults(self) -> None:
+        """Test default values."""
+        report = HealthReport(status=HealthStatus.HEALTHY)
+
+        assert report.status == HealthStatus.HEALTHY
+        assert report.streams == {}
+        assert report.total_events_received == 0
+        assert report.total_events_per_second == 0.0
+        assert report.uptime_seconds == 0.0
+        assert report.timestamp > 0
+
+    def test_health_report_with_streams(self) -> None:
+        """Test with stream data."""
+        stream = StreamHealth(name="trades", events_received=100)
+        report = HealthReport(
+            status=HealthStatus.DEGRADED,
+            streams={"trades": stream},
+            total_events_received=100,
+            total_events_per_second=5.0,
+            uptime_seconds=3600.0,
+        )
+
+        assert report.status == HealthStatus.DEGRADED
+        assert "trades" in report.streams
+        assert report.total_events_received == 100
+
+
+class TestHealthMonitor:
+    """Tests for the HealthMonitor class."""
+
+    def test_init(self) -> None:
+        """Test initialization."""
+        monitor = HealthMonitor()
+
+        assert monitor._stale_threshold == DEFAULT_STALE_THRESHOLD_SECONDS
+        assert monitor._health_check_interval == DEFAULT_HEALTH_CHECK_INTERVAL
+        assert not monitor.is_running
+
+    def test_init_custom_config(self) -> None:
+        """Test initialization with custom config."""
+        monitor = HealthMonitor(
+            stale_threshold_seconds=30,
+            health_check_interval=10,
+        )
+
+        assert monitor._stale_threshold == 30
+        assert monitor._health_check_interval == 10
+
+    def test_register_stream(self) -> None:
+        """Test registering a stream."""
+        monitor = HealthMonitor()
+
+        monitor.register_stream("trades")
+
+        assert "trades" in monitor._streams
+        assert monitor._streams["trades"].name == "trades"
+        assert monitor._streams["trades"].status == StreamStatus.DISCONNECTED
+
+    def test_register_stream_idempotent(self) -> None:
+        """Test that registering the same stream twice is idempotent."""
+        monitor = HealthMonitor()
+
+        monitor.register_stream("trades")
+        monitor.record_event("trades")  # Adds an event
+        monitor.register_stream("trades")  # Should not reset
+
+        assert monitor._streams["trades"].events_received == 1
+
+    def test_set_stream_connected(self) -> None:
+        """Test marking a stream as connected."""
+        monitor = HealthMonitor()
+
+        monitor.set_stream_connected("trades")
+
+        assert monitor._streams["trades"].status == StreamStatus.ACTIVE
+        assert monitor._streams["trades"].connected_since is not None
+        assert monitor._streams["trades"].last_error is None
+
+    def test_set_stream_disconnected(self) -> None:
+        """Test marking a stream as disconnected."""
+        monitor = HealthMonitor()
+
+        monitor.set_stream_connected("trades")
+        monitor.set_stream_disconnected("trades", error="Connection reset")
+
+        assert monitor._streams["trades"].status == StreamStatus.DISCONNECTED
+        assert monitor._streams["trades"].connected_since is None
+        assert monitor._streams["trades"].last_error == "Connection reset"
+
+    def test_record_event(self) -> None:
+        """Test recording an event."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+
+        stream = monitor._streams["trades"]
+        assert stream.events_received == 1
+        assert stream.last_event_time is not None
+        assert stream.status == StreamStatus.ACTIVE
+
+    def test_record_event_multiple(self) -> None:
+        """Test recording multiple events."""
+        monitor = HealthMonitor()
+
+        for _ in range(10):
+            monitor.record_event("trades")
+
+        assert monitor._streams["trades"].events_received == 10
+
+    def test_record_event_with_processing_time(self) -> None:
+        """Test recording event with processing time."""
+        monitor = HealthMonitor()
+
+        # Should not raise
+        monitor.record_event("trades", processing_time=0.001)
+
+        assert monitor._streams["trades"].events_received == 1
+
+    def test_calculate_throughput_empty(self) -> None:
+        """Test throughput calculation with no events."""
+        monitor = HealthMonitor()
+
+        rate = monitor._calculate_throughput("nonexistent")
+
+        assert rate == 0.0
+
+    def test_calculate_throughput(self) -> None:
+        """Test throughput calculation."""
+        monitor = HealthMonitor()
+
+        # Add events
+        for _ in range(10):
+            monitor.record_event("trades")
+
+        rate = monitor._calculate_throughput("trades")
+
+        # Should have ~10 events in the window
+        assert rate > 0
+
+    def test_check_stream_staleness_active(self) -> None:
+        """Test that active stream is not marked stale."""
+        monitor = HealthMonitor(stale_threshold_seconds=60)
+
+        monitor.record_event("trades")
+        monitor._check_stream_staleness()
+
+        assert monitor._streams["trades"].status == StreamStatus.ACTIVE
+
+    def test_check_stream_staleness_stale(self) -> None:
+        """Test that stream becomes stale after threshold."""
+        monitor = HealthMonitor(stale_threshold_seconds=1)
+
+        monitor.record_event("trades")
+        # Simulate time passing
+        monitor._streams["trades"].last_event_time = time.time() - 2
+
+        monitor._check_stream_staleness()
+
+        assert monitor._streams["trades"].status == StreamStatus.STALE
+
+    def test_check_stream_staleness_connected_no_events(self) -> None:
+        """Test staleness when connected but no events received."""
+        monitor = HealthMonitor(stale_threshold_seconds=1)
+
+        monitor.set_stream_connected("trades")
+        # Simulate time passing since connection
+        monitor._streams["trades"].connected_since = time.time() - 2
+
+        monitor._check_stream_staleness()
+
+        assert monitor._streams["trades"].status == StreamStatus.STALE
+
+    def test_determine_overall_status_no_streams(self) -> None:
+        """Test overall status with no streams."""
+        monitor = HealthMonitor()
+
+        status = monitor._determine_overall_status()
+
+        assert status == HealthStatus.HEALTHY
+
+    def test_determine_overall_status_all_active(self) -> None:
+        """Test overall status with all active streams."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+        monitor.record_event("orderbook")
+
+        status = monitor._determine_overall_status()
+
+        assert status == HealthStatus.HEALTHY
+
+    def test_determine_overall_status_some_stale(self) -> None:
+        """Test overall status with some stale streams."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+        monitor.register_stream("orderbook")
+        monitor._streams["orderbook"].status = StreamStatus.STALE
+
+        status = monitor._determine_overall_status()
+
+        assert status == HealthStatus.DEGRADED
+
+    def test_determine_overall_status_some_disconnected(self) -> None:
+        """Test overall status with some disconnected streams."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+        monitor.set_stream_disconnected("orderbook")
+
+        status = monitor._determine_overall_status()
+
+        assert status == HealthStatus.DEGRADED
+
+    def test_determine_overall_status_all_disconnected(self) -> None:
+        """Test overall status with all disconnected streams."""
+        monitor = HealthMonitor()
+
+        monitor.set_stream_disconnected("trades")
+        monitor.set_stream_disconnected("orderbook")
+
+        status = monitor._determine_overall_status()
+
+        assert status == HealthStatus.UNHEALTHY
+
+    def test_get_health_report(self) -> None:
+        """Test getting a health report."""
+        monitor = HealthMonitor()
+        monitor._start_time = time.time() - 100
+
+        monitor.record_event("trades")
+        monitor.record_event("trades")
+
+        report = monitor.get_health_report()
+
+        assert report.status == HealthStatus.HEALTHY
+        assert "trades" in report.streams
+        assert report.total_events_received == 2
+        assert report.uptime_seconds >= 100
+        assert report.timestamp > 0
+
+    def test_get_health_report_calculates_throughput(self) -> None:
+        """Test that health report calculates throughput."""
+        monitor = HealthMonitor()
+
+        for _ in range(10):
+            monitor.record_event("trades")
+
+        report = monitor.get_health_report()
+
+        assert report.streams["trades"].events_per_second > 0
+        assert report.total_events_per_second > 0
+
+    @pytest.mark.asyncio
+    async def test_start_stop(self) -> None:
+        """Test starting and stopping the monitor."""
+        monitor = HealthMonitor()
+
+        await monitor.start()
+        assert monitor.is_running
+        assert monitor._health_task is not None
+
+        await monitor.stop()
+        assert not monitor.is_running
+        assert monitor._health_task is None
+
+    @pytest.mark.asyncio
+    async def test_start_idempotent(self) -> None:
+        """Test that starting twice is safe."""
+        monitor = HealthMonitor()
+
+        await monitor.start()
+        await monitor.start()  # Should not raise
+
+        assert monitor.is_running
+
+        await monitor.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_when_not_running(self) -> None:
+        """Test that stopping when not running is safe."""
+        monitor = HealthMonitor()
+
+        await monitor.stop()  # Should not raise
+
+    @pytest.mark.asyncio
+    async def test_context_manager(self) -> None:
+        """Test async context manager."""
+        async with HealthMonitor() as monitor:
+            assert monitor.is_running
+
+        assert not monitor.is_running
+
+    @pytest.mark.asyncio
+    async def test_health_check_loop_updates_report(self) -> None:
+        """Test that health check loop updates the report."""
+        monitor = HealthMonitor(health_check_interval=0.1)
+
+        await monitor.start()
+        monitor.record_event("trades")
+
+        await asyncio.sleep(0.2)
+
+        # Health should have been checked
+        report = monitor.get_health_report()
+        assert report.status == HealthStatus.HEALTHY
+
+        await monitor.stop()
+
+    @pytest.mark.asyncio
+    async def test_health_change_callback(self) -> None:
+        """Test that health change callback is invoked."""
+        callback = AsyncMock()
+        monitor = HealthMonitor(
+            health_check_interval=0.1,
+            on_health_change=callback,
+        )
+
+        await monitor.start()
+        monitor.record_event("trades")
+
+        # Wait for health check
+        await asyncio.sleep(0.2)
+
+        await monitor.stop()
+
+        # Callback should have been called at least once
+        assert callback.called
+
+    @pytest.mark.asyncio
+    async def test_health_change_callback_error_handling(self) -> None:
+        """Test that callback errors don't crash the loop."""
+        callback = AsyncMock(side_effect=ValueError("test error"))
+        monitor = HealthMonitor(
+            health_check_interval=0.1,
+            on_health_change=callback,
+        )
+
+        await monitor.start()
+        monitor.record_event("trades")
+
+        # Should not crash
+        await asyncio.sleep(0.2)
+
+        await monitor.stop()
+
+
+class TestHealthMonitorHTTPEndpoints:
+    """Tests for HTTP endpoints."""
+
+    @pytest.fixture
+    def monitor(self) -> HealthMonitor:
+        """Create a monitor instance."""
+        return HealthMonitor()
+
+    @pytest.fixture
+    def app(self, monitor: HealthMonitor) -> web.Application:
+        """Create the aiohttp application."""
+        return monitor._create_app()
+
+    @pytest.mark.asyncio
+    async def test_health_endpoint_healthy(
+        self, monitor: HealthMonitor, app: web.Application
+    ) -> None:
+        """Test /health endpoint when healthy."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        monitor.record_event("trades")
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/health")
+            assert resp.status == 200
+
+            data = await resp.json()
+            assert data["status"] == "healthy"
+            assert "trades" in data["streams"]
+
+    @pytest.mark.asyncio
+    async def test_health_endpoint_unhealthy(
+        self, monitor: HealthMonitor, app: web.Application
+    ) -> None:
+        """Test /health endpoint when unhealthy."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        monitor.set_stream_disconnected("trades")
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/health")
+            assert resp.status == 503
+
+            data = await resp.json()
+            assert data["status"] == "unhealthy"
+
+    @pytest.mark.asyncio
+    async def test_metrics_endpoint(
+        self, monitor: HealthMonitor, app: web.Application
+    ) -> None:
+        """Test /metrics endpoint returns Prometheus format."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        monitor.record_event("trades")
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/metrics")
+            assert resp.status == 200
+
+            content_type = resp.headers.get("Content-Type", "")
+            assert "text/plain" in content_type
+
+            text = await resp.text()
+            assert "polymarket_events_total" in text
+            assert "polymarket_health_status" in text
+
+    @pytest.mark.asyncio
+    async def test_ready_endpoint_ready(
+        self, monitor: HealthMonitor, app: web.Application
+    ) -> None:
+        """Test /ready endpoint when ready."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        monitor.record_event("trades")
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/ready")
+            assert resp.status == 200
+
+            data = await resp.json()
+            assert data["ready"] is True
+
+    @pytest.mark.asyncio
+    async def test_ready_endpoint_not_ready(
+        self, monitor: HealthMonitor, app: web.Application
+    ) -> None:
+        """Test /ready endpoint when not ready."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        monitor.set_stream_disconnected("trades")
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/ready")
+            assert resp.status == 503
+
+            data = await resp.json()
+            assert data["ready"] is False
+
+    @pytest.mark.asyncio
+    async def test_live_endpoint(self, app: web.Application) -> None:
+        """Test /live endpoint always returns 200."""
+        from aiohttp.test_utils import TestClient, TestServer
+
+        async with TestClient(TestServer(app)) as client:
+            resp = await client.get("/live")
+            assert resp.status == 200
+
+            data = await resp.json()
+            assert data["live"] is True
+
+    @pytest.mark.asyncio
+    async def test_start_stop_http_server(self, monitor: HealthMonitor) -> None:
+        """Test starting and stopping HTTP server."""
+        await monitor.start_http_server(port=18080)
+        assert monitor._runner is not None
+
+        await monitor.stop_http_server()
+        assert monitor._runner is None
+
+    @pytest.mark.asyncio
+    async def test_start_http_server_idempotent(self, monitor: HealthMonitor) -> None:
+        """Test that starting HTTP server twice is safe."""
+        await monitor.start_http_server(port=18081)
+        await monitor.start_http_server(port=18081)  # Should not raise
+
+        await monitor.stop_http_server()
+
+
+class TestPrometheusMetrics:
+    """Tests for Prometheus metric updates."""
+
+    def test_events_total_incremented(self) -> None:
+        """Test that events_total counter is incremented."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("test-metrics")
+        monitor.record_event("test-metrics")
+
+        # Counter should have been incremented
+        # (We can't easily test prometheus metrics directly, but at least verify no errors)
+
+    def test_stream_status_updated(self) -> None:
+        """Test that stream_status gauge is updated."""
+        monitor = HealthMonitor()
+
+        monitor.set_stream_connected("test-status")
+        # Gauge should be 1.0
+
+        monitor.set_stream_disconnected("test-status")
+        # Gauge should be 0.0
+
+    def test_health_status_updated(self) -> None:
+        """Test that health_status gauge is updated."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("test-health")
+        report = monitor.get_health_report()
+
+        assert report.status == HealthStatus.HEALTHY
+
+
+class TestEdgeCases:
+    """Tests for edge cases and error handling."""
+
+    def test_throughput_with_old_events(self) -> None:
+        """Test throughput calculation ignores old events."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+        # Manually add old event to window
+        monitor._event_windows["trades"].append(time.time() - 100)
+
+        rate = monitor._calculate_throughput("trades")
+
+        # Old event should be filtered out
+        # Rate should only count recent events
+        assert rate >= 0
+
+    def test_multiple_streams_independent(self) -> None:
+        """Test that multiple streams are tracked independently."""
+        monitor = HealthMonitor()
+
+        monitor.record_event("trades")
+        monitor.record_event("trades")
+        monitor.set_stream_disconnected("orderbook")
+
+        assert monitor._streams["trades"].events_received == 2
+        assert monitor._streams["trades"].status == StreamStatus.ACTIVE
+        assert monitor._streams["orderbook"].events_received == 0
+        assert monitor._streams["orderbook"].status == StreamStatus.DISCONNECTED
+
+    def test_report_streams_are_copied(self) -> None:
+        """Test that report streams are a copy."""
+        monitor = HealthMonitor()
+        monitor.record_event("trades")
+
+        report = monitor.get_health_report()
+
+        # Modifying report should not affect monitor
+        report.streams["trades"].events_received = 999
+        assert monitor._streams["trades"].events_received == 1
+
+    @pytest.mark.asyncio
+    async def test_stop_cleans_up_http_server(self) -> None:
+        """Test that stop() also stops HTTP server."""
+        monitor = HealthMonitor()
+
+        await monitor.start()
+        await monitor.start_http_server(port=18082)
+
+        await monitor.stop()
+
+        assert not monitor.is_running
+        assert monitor._runner is None

tests/ingestor/test_metadata_sync.py

@@ -0,0 +1,443 @@
+"""Tests for the market metadata synchronizer."""
+
+import json
+from datetime import UTC, datetime
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+from polymarket_insider_tracker.ingestor.clob_client import ClobClient
+from polymarket_insider_tracker.ingestor.metadata_sync import (
+    DEFAULT_CACHE_TTL_SECONDS,
+    DEFAULT_REDIS_KEY_PREFIX,
+    DEFAULT_SYNC_INTERVAL_SECONDS,
+    MarketMetadataSync,
+    MetadataSyncError,
+    SyncState,
+    SyncStats,
+)
+from polymarket_insider_tracker.ingestor.models import (
+    Market,
+    MarketMetadata,
+    Token,
+    derive_category,
+)
+
+
+# Test fixtures
+@pytest.fixture
+def sample_token() -> Token:
+    """Create a sample token."""
+    return Token(token_id="token123", outcome="Yes", price=Decimal("0.65"))
+
+
+@pytest.fixture
+def sample_market(sample_token: Token) -> Market:
+    """Create a sample market."""
+    return Market(
+        condition_id="cond123",
+        question="Will Bitcoin exceed $100k in 2026?",
+        description="Market on BTC price",
+        tokens=(sample_token,),
+        end_date=datetime(2026, 12, 31, tzinfo=UTC),
+        active=True,
+        closed=False,
+    )
+
+
+@pytest.fixture
+def sample_metadata(sample_market: Market) -> MarketMetadata:
+    """Create sample metadata from market."""
+    return MarketMetadata.from_market(sample_market)
+
+
+@pytest.fixture
+def mock_redis() -> AsyncMock:
+    """Create a mock Redis client."""
+    redis = AsyncMock()
+    redis.get = AsyncMock(return_value=None)
+    redis.setex = AsyncMock()
+    redis.delete = AsyncMock(return_value=1)
+    redis.scan = AsyncMock(return_value=(0, []))
+    return redis
+
+
+@pytest.fixture
+def mock_clob(sample_market: Market) -> MagicMock:
+    """Create a mock CLOB client."""
+    clob = MagicMock(spec=ClobClient)
+    clob.get_markets = MagicMock(return_value=[sample_market])
+    clob.get_market = MagicMock(return_value=sample_market)
+    return clob
+
+
+class TestDeriveCategory:
+    """Tests for the derive_category function."""
+
+    def test_politics_keywords(self) -> None:
+        """Test political category detection."""
+        assert derive_category("Will Trump win the 2024 election?") == "politics"
+        assert derive_category("Who will be the next president?") == "politics"
+        assert derive_category("Senate majority party after midterms?") == "politics"
+
+    def test_crypto_keywords(self) -> None:
+        """Test crypto category detection."""
+        assert derive_category("Will Bitcoin hit $100k?") == "crypto"
+        assert derive_category("Ethereum price by end of year?") == "crypto"
+        assert derive_category("Next altcoin to moon?") == "crypto"
+
+    def test_sports_keywords(self) -> None:
+        """Test sports category detection."""
+        assert derive_category("Who will win the Super Bowl?") == "sports"
+        assert derive_category("NBA Finals champion?") == "sports"
+        assert derive_category("Next UFC heavyweight champion?") == "sports"
+
+    def test_entertainment_keywords(self) -> None:
+        """Test entertainment category detection."""
+        assert derive_category("Best Picture Oscar winner?") == "entertainment"
+        assert derive_category("Next Grammy Album of the Year?") == "entertainment"
+        assert derive_category("Highest box office movie this summer?") == "entertainment"
+
+    def test_finance_keywords(self) -> None:
+        """Test finance category detection."""
+        assert derive_category("Fed interest rate decision?") == "finance"
+        assert derive_category("Will we enter a recession?") == "finance"
+        assert derive_category("S&P 500 by year end?") == "finance"
+
+    def test_tech_keywords(self) -> None:
+        """Test tech category detection."""
+        assert derive_category("Will Apple release a new iPhone?") == "tech"
+        assert derive_category("Next major AI breakthrough?") == "tech"
+        assert derive_category("Tesla vehicle deliveries?") == "tech"
+
+    def test_science_keywords(self) -> None:
+        """Test science category detection."""
+        assert derive_category("NASA Mars mission timeline?") == "science"
+        assert derive_category("FDA approval for new drug?") == "science"
+        assert derive_category("Climate change targets met?") == "science"
+
+    def test_other_category(self) -> None:
+        """Test fallback to 'other' category."""
+        assert derive_category("Random obscure question?") == "other"
+        assert derive_category("Will it be sunny tomorrow?") == "other"
+
+    def test_case_insensitive(self) -> None:
+        """Test case insensitivity."""
+        assert derive_category("BITCOIN PRICE") == "crypto"
+        assert derive_category("bitcoin price") == "crypto"
+        assert derive_category("Bitcoin Price") == "crypto"
+
+
+class TestMarketMetadata:
+    """Tests for the MarketMetadata dataclass."""
+
+    def test_from_market(self, sample_market: Market) -> None:
+        """Test creating metadata from a market."""
+        metadata = MarketMetadata.from_market(sample_market)
+
+        assert metadata.condition_id == sample_market.condition_id
+        assert metadata.question == sample_market.question
+        assert metadata.description == sample_market.description
+        assert metadata.tokens == sample_market.tokens
+        assert metadata.end_date == sample_market.end_date
+        assert metadata.active == sample_market.active
+        assert metadata.closed == sample_market.closed
+        assert metadata.category == "crypto"  # "Bitcoin" in question
+        assert metadata.last_updated is not None
+
+    def test_to_dict(self, sample_metadata: MarketMetadata) -> None:
+        """Test serialization to dict."""
+        data = sample_metadata.to_dict()
+
+        assert data["condition_id"] == sample_metadata.condition_id
+        assert data["question"] == sample_metadata.question
+        assert data["category"] == "crypto"
+        assert len(data["tokens"]) == 1
+        assert data["tokens"][0]["token_id"] == "token123"
+
+    def test_from_dict(self, sample_metadata: MarketMetadata) -> None:
+        """Test deserialization from dict."""
+        data = sample_metadata.to_dict()
+        restored = MarketMetadata.from_dict(data)
+
+        assert restored.condition_id == sample_metadata.condition_id
+        assert restored.question == sample_metadata.question
+        assert restored.category == sample_metadata.category
+        assert len(restored.tokens) == 1
+
+    def test_roundtrip(self, sample_metadata: MarketMetadata) -> None:
+        """Test serialization roundtrip."""
+        data = sample_metadata.to_dict()
+        json_str = json.dumps(data)
+        parsed = json.loads(json_str)
+        restored = MarketMetadata.from_dict(parsed)
+
+        assert restored.condition_id == sample_metadata.condition_id
+        assert restored.question == sample_metadata.question
+
+
+class TestSyncStats:
+    """Tests for the SyncStats dataclass."""
+
+    def test_defaults(self) -> None:
+        """Test default values."""
+        stats = SyncStats()
+
+        assert stats.total_syncs == 0
+        assert stats.successful_syncs == 0
+        assert stats.failed_syncs == 0
+        assert stats.markets_cached == 0
+        assert stats.last_sync_time is None
+        assert stats.last_error is None
+
+
+class TestMarketMetadataSync:
+    """Tests for the MarketMetadataSync class."""
+
+    def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test initialization."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        assert sync.state == SyncState.STOPPED
+        assert sync.stats.total_syncs == 0
+        assert sync._sync_interval == DEFAULT_SYNC_INTERVAL_SECONDS
+        assert sync._cache_ttl == DEFAULT_CACHE_TTL_SECONDS
+        assert sync._key_prefix == DEFAULT_REDIS_KEY_PREFIX
+
+    def test_init_custom_config(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test initialization with custom config."""
+        sync = MarketMetadataSync(
+            redis=mock_redis,
+            clob_client=mock_clob,
+            sync_interval_seconds=60,
+            cache_ttl_seconds=120,
+            key_prefix="custom:",
+        )
+
+        assert sync._sync_interval == 60
+        assert sync._cache_ttl == 120
+        assert sync._key_prefix == "custom:"
+
+    @pytest.mark.asyncio
+    async def test_start_stop(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test starting and stopping the sync service."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        # Start
+        await sync.start()
+        assert sync.state == SyncState.IDLE
+        assert sync.stats.total_syncs == 1
+        assert sync.stats.successful_syncs == 1
+
+        # Stop
+        await sync.stop()
+        assert sync.state == SyncState.STOPPED
+
+    @pytest.mark.asyncio
+    async def test_start_performs_initial_sync(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock
+    ) -> None:
+        """Test that start performs an initial sync."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        await sync.start()
+
+        # Should have called get_markets
+        mock_clob.get_markets.assert_called_once_with(True)
+
+        # Should have cached the market
+        mock_redis.setex.assert_called()
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_start_failure(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test start failure handling."""
+        mock_clob.get_markets.side_effect = Exception("API error")
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        with pytest.raises(MetadataSyncError, match="initial sync failed"):
+            await sync.start()
+
+        assert sync.state == SyncState.ERROR
+        assert sync.stats.last_error == "API error"
+
+    @pytest.mark.asyncio
+    async def test_get_market_cache_hit(
+        self,
+        mock_redis: AsyncMock,
+        mock_clob: MagicMock,
+        sample_metadata: MarketMetadata,
+    ) -> None:
+        """Test get_market with cache hit."""
+        # Setup cache hit
+        cached_data = json.dumps(sample_metadata.to_dict())
+        mock_redis.get = AsyncMock(return_value=cached_data)
+
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        await sync.start()
+
+        result = await sync.get_market("cond123")
+
+        assert result is not None
+        assert result.condition_id == "cond123"
+        # Should not have called API
+        mock_clob.get_market.assert_not_called()
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_get_market_cache_miss(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test get_market with cache miss."""
+        # Setup cache miss
+        mock_redis.get = AsyncMock(return_value=None)
+
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        await sync.start()
+
+        result = await sync.get_market("cond123")
+
+        assert result is not None
+        assert result.condition_id == "cond123"
+        # Should have called API
+        mock_clob.get_market.assert_called_with("cond123")
+        # Should have cached the result
+        assert mock_redis.setex.call_count >= 2  # Initial sync + cache miss
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_get_market_not_found(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test get_market when market doesn't exist."""
+        mock_redis.get = AsyncMock(return_value=None)
+        mock_clob.get_market.return_value = None
+
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        await sync.start()
+
+        result = await sync.get_market("nonexistent")
+
+        assert result is None
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_invalidate_market(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test cache invalidation."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        await sync.start()
+
+        result = await sync.invalidate_market("cond123")
+
+        assert result is True
+        mock_redis.delete.assert_called_with(f"{DEFAULT_REDIS_KEY_PREFIX}cond123")
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_force_sync(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test forced sync."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        await sync.start()
+
+        # Initial sync
+        assert sync.stats.total_syncs == 1
+
+        # Force sync
+        await sync.force_sync()
+
+        assert sync.stats.total_syncs == 2
+        assert sync.stats.successful_syncs == 2
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_state_change_callback(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test state change callback."""
+        states: list[SyncState] = []
+
+        def on_state_change(state: SyncState) -> None:
+            states.append(state)
+
+        sync = MarketMetadataSync(
+            redis=mock_redis,
+            clob_client=mock_clob,
+            on_state_change=on_state_change,
+        )
+
+        await sync.start()
+        await sync.stop()
+
+        assert SyncState.STARTING in states
+        assert SyncState.SYNCING in states
+        assert SyncState.IDLE in states
+        assert SyncState.STOPPING in states
+        assert SyncState.STOPPED in states
+
+    @pytest.mark.asyncio
+    async def test_sync_complete_callback(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock
+    ) -> None:
+        """Test sync complete callback."""
+        sync_stats: list[SyncStats] = []
+
+        def on_sync_complete(stats: SyncStats) -> None:
+            sync_stats.append(stats)
+
+        sync = MarketMetadataSync(
+            redis=mock_redis,
+            clob_client=mock_clob,
+            on_sync_complete=on_sync_complete,
+        )
+
+        await sync.start()
+
+        assert len(sync_stats) == 1
+        assert sync_stats[0].successful_syncs == 1
+        assert sync_stats[0].markets_cached == 1
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_get_markets_by_category(
+        self, mock_redis: AsyncMock, mock_clob: MagicMock, sample_metadata: MarketMetadata
+    ) -> None:
+        """Test getting markets by category."""
+        # Setup scan to return keys
+        key = f"{DEFAULT_REDIS_KEY_PREFIX}cond123"
+        mock_redis.scan = AsyncMock(return_value=(0, [key]))
+
+        # Setup get to return cached data
+        cached_data = json.dumps(sample_metadata.to_dict())
+        mock_redis.get = AsyncMock(return_value=cached_data)
+
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+        # Don't start to avoid initial sync complexity
+        sync._state = SyncState.IDLE
+
+        results = await sync.get_markets_by_category("crypto")
+
+        assert len(results) == 1
+        assert results[0].category == "crypto"
+
+    @pytest.mark.asyncio
+    async def test_cannot_start_twice(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test that starting twice doesn't double-start."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        await sync.start()
+        await sync.start()  # Should be a no-op
+
+        assert sync.stats.total_syncs == 1  # Only one initial sync
+
+        await sync.stop()
+
+    @pytest.mark.asyncio
+    async def test_stop_when_stopped(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None:
+        """Test stopping when already stopped."""
+        sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob)
+
+        await sync.stop()  # Should be a no-op
+
+        assert sync.state == SyncState.STOPPED

tests/ingestor/test_models.py

@@ -0,0 +1,420 @@
+"""Tests for ingestor data models."""
+
+from datetime import datetime, timezone
+from decimal import Decimal
+from unittest.mock import MagicMock
+
+import pytest
+
+from polymarket_insider_tracker.ingestor.models import (
+    Market,
+    Orderbook,
+    OrderbookLevel,
+    Token,
+    TradeEvent,
+)
+
+
+class TestToken:
+    """Tests for Token model."""
+
+    def test_from_dict_with_price(self) -> None:
+        """Test creating Token from dict with price."""
+        data = {
+            "token_id": "123abc",
+            "outcome": "Yes",
+            "price": "0.65",
+        }
+        token = Token.from_dict(data)
+
+        assert token.token_id == "123abc"
+        assert token.outcome == "Yes"
+        assert token.price == Decimal("0.65")
+
+    def test_from_dict_without_price(self) -> None:
+        """Test creating Token from dict without price."""
+        data = {
+            "token_id": "456def",
+            "outcome": "No",
+        }
+        token = Token.from_dict(data)
+
+        assert token.token_id == "456def"
+        assert token.outcome == "No"
+        assert token.price is None
+
+    def test_frozen(self) -> None:
+        """Test that Token is immutable."""
+        token = Token(token_id="123", outcome="Yes", price=Decimal("0.5"))
+        with pytest.raises(AttributeError):
+            token.token_id = "456"  # type: ignore[misc]
+
+
+class TestMarket:
+    """Tests for Market model."""
+
+    def test_from_dict_full(self) -> None:
+        """Test creating Market from complete dict."""
+        data = {
+            "condition_id": "0xabc123",
+            "question": "Will it rain tomorrow?",
+            "description": "Market for weather prediction",
+            "tokens": [
+                {"token_id": "t1", "outcome": "Yes", "price": "0.7"},
+                {"token_id": "t2", "outcome": "No", "price": "0.3"},
+            ],
+            "end_date_iso": "2024-12-31T23:59:59Z",
+            "active": True,
+            "closed": False,
+        }
+        market = Market.from_dict(data)
+
+        assert market.condition_id == "0xabc123"
+        assert market.question == "Will it rain tomorrow?"
+        assert market.description == "Market for weather prediction"
+        assert len(market.tokens) == 2
+        assert market.tokens[0].outcome == "Yes"
+        assert market.tokens[1].outcome == "No"
+        assert market.end_date == datetime(2024, 12, 31, 23, 59, 59, tzinfo=timezone.utc)
+        assert market.active is True
+        assert market.closed is False
+
+    def test_from_dict_minimal(self) -> None:
+        """Test creating Market from minimal dict."""
+        data = {
+            "condition_id": "0xdef456",
+        }
+        market = Market.from_dict(data)
+
+        assert market.condition_id == "0xdef456"
+        assert market.question == ""
+        assert market.description == ""
+        assert market.tokens == ()
+        assert market.end_date is None
+        assert market.active is True
+        assert market.closed is False
+
+    def test_from_dict_invalid_date(self) -> None:
+        """Test that invalid date is handled gracefully."""
+        data = {
+            "condition_id": "0x123",
+            "end_date_iso": "not-a-valid-date",
+        }
+        market = Market.from_dict(data)
+
+        assert market.end_date is None
+
+    def test_frozen(self) -> None:
+        """Test that Market is immutable."""
+        market = Market(
+            condition_id="0x123",
+            question="Test?",
+            description="",
+            tokens=(),
+        )
+        with pytest.raises(AttributeError):
+            market.condition_id = "0x456"  # type: ignore[misc]
+
+
+class TestOrderbookLevel:
+    """Tests for OrderbookLevel model."""
+
+    def test_from_dict(self) -> None:
+        """Test creating OrderbookLevel from dict."""
+        data = {"price": "0.55", "size": "100.5"}
+        level = OrderbookLevel.from_dict(data)
+
+        assert level.price == Decimal("0.55")
+        assert level.size == Decimal("100.5")
+
+    def test_frozen(self) -> None:
+        """Test that OrderbookLevel is immutable."""
+        level = OrderbookLevel(price=Decimal("0.5"), size=Decimal("10"))
+        with pytest.raises(AttributeError):
+            level.price = Decimal("0.6")  # type: ignore[misc]
+
+
+class TestOrderbook:
+    """Tests for Orderbook model."""
+
+    def test_from_clob_orderbook(self) -> None:
+        """Test creating Orderbook from py-clob-client response."""
+        # Create mock bid/ask objects
+        mock_bid = MagicMock()
+        mock_bid.price = "0.50"
+        mock_bid.size = "100"
+
+        mock_ask = MagicMock()
+        mock_ask.price = "0.52"
+        mock_ask.size = "150"
+
+        mock_orderbook = MagicMock()
+        mock_orderbook.market = "0xmarket123"
+        mock_orderbook.asset_id = "token123"
+        mock_orderbook.tick_size = "0.01"
+        mock_orderbook.bids = [mock_bid]
+        mock_orderbook.asks = [mock_ask]
+
+        orderbook = Orderbook.from_clob_orderbook(mock_orderbook)
+
+        assert orderbook.market == "0xmarket123"
+        assert orderbook.asset_id == "token123"
+        assert orderbook.tick_size == Decimal("0.01")
+        assert len(orderbook.bids) == 1
+        assert len(orderbook.asks) == 1
+        assert orderbook.bids[0].price == Decimal("0.50")
+        assert orderbook.asks[0].price == Decimal("0.52")
+
+    def test_from_clob_orderbook_empty(self) -> None:
+        """Test creating Orderbook with empty bids/asks."""
+        mock_orderbook = MagicMock()
+        mock_orderbook.market = "0xmarket"
+        mock_orderbook.asset_id = "token"
+        mock_orderbook.tick_size = "0.01"
+        mock_orderbook.bids = None
+        mock_orderbook.asks = []
+
+        orderbook = Orderbook.from_clob_orderbook(mock_orderbook)
+
+        assert orderbook.bids == ()
+        assert orderbook.asks == ()
+
+    def test_best_bid(self) -> None:
+        """Test best_bid property."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(
+                OrderbookLevel(Decimal("0.50"), Decimal("100")),
+                OrderbookLevel(Decimal("0.49"), Decimal("50")),
+            ),
+            asks=(),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.best_bid == Decimal("0.50")
+
+    def test_best_bid_empty(self) -> None:
+        """Test best_bid with no bids."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(),
+            asks=(),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.best_bid is None
+
+    def test_best_ask(self) -> None:
+        """Test best_ask property."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(),
+            asks=(
+                OrderbookLevel(Decimal("0.52"), Decimal("100")),
+                OrderbookLevel(Decimal("0.53"), Decimal("50")),
+            ),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.best_ask == Decimal("0.52")
+
+    def test_spread(self) -> None:
+        """Test spread calculation."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(OrderbookLevel(Decimal("0.50"), Decimal("100")),),
+            asks=(OrderbookLevel(Decimal("0.52"), Decimal("100")),),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.spread == Decimal("0.02")
+
+    def test_spread_missing_data(self) -> None:
+        """Test spread with missing bid or ask."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(OrderbookLevel(Decimal("0.50"), Decimal("100")),),
+            asks=(),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.spread is None
+
+    def test_midpoint(self) -> None:
+        """Test midpoint calculation."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(OrderbookLevel(Decimal("0.50"), Decimal("100")),),
+            asks=(OrderbookLevel(Decimal("0.52"), Decimal("100")),),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.midpoint == Decimal("0.51")
+
+    def test_midpoint_missing_data(self) -> None:
+        """Test midpoint with missing data."""
+        orderbook = Orderbook(
+            market="0x",
+            asset_id="t",
+            bids=(),
+            asks=(OrderbookLevel(Decimal("0.52"), Decimal("100")),),
+            tick_size=Decimal("0.01"),
+        )
+
+        assert orderbook.midpoint is None
+
+
+class TestTradeEvent:
+    """Tests for TradeEvent model."""
+
+    def test_from_websocket_message_full(self) -> None:
+        """Test creating TradeEvent from a full WebSocket message."""
+        data = {
+            "conditionId": "0xmarket123",
+            "transactionHash": "0xtx456",
+            "proxyWallet": "0xwallet789",
+            "side": "BUY",
+            "outcome": "Yes",
+            "outcomeIndex": 0,
+            "price": 0.65,
+            "size": 100,
+            "timestamp": 1704067200,  # 2024-01-01 00:00:00 UTC
+            "asset": "token123",
+            "slug": "will-it-rain",
+            "eventSlug": "weather-markets",
+            "title": "Weather Predictions",
+            "name": "Alice",
+            "pseudonym": "AliceTrader",
+        }
+        trade = TradeEvent.from_websocket_message(data)
+
+        assert trade.market_id == "0xmarket123"
+        assert trade.trade_id == "0xtx456"
+        assert trade.wallet_address == "0xwallet789"
+        assert trade.side == "BUY"
+        assert trade.outcome == "Yes"
+        assert trade.outcome_index == 0
+        assert trade.price == Decimal("0.65")
+        assert trade.size == Decimal("100")
+        assert trade.timestamp == datetime(2024, 1, 1, 0, 0, 0, tzinfo=timezone.utc)
+        assert trade.asset_id == "token123"
+        assert trade.market_slug == "will-it-rain"
+        assert trade.event_slug == "weather-markets"
+        assert trade.event_title == "Weather Predictions"
+        assert trade.trader_name == "Alice"
+        assert trade.trader_pseudonym == "AliceTrader"
+
+    def test_from_websocket_message_minimal(self) -> None:
+        """Test creating TradeEvent with minimal data."""
+        data = {
+            "conditionId": "0x123",
+            "transactionHash": "0xtx",
+            "proxyWallet": "0xwallet",
+            "side": "SELL",
+            "outcome": "No",
+            "price": 0.35,
+            "size": 50,
+        }
+        trade = TradeEvent.from_websocket_message(data)
+
+        assert trade.market_id == "0x123"
+        assert trade.side == "SELL"
+        assert trade.outcome == "No"
+        assert trade.price == Decimal("0.35")
+        assert trade.size == Decimal("50")
+        assert trade.market_slug == ""
+        assert trade.trader_name == ""
+
+    def test_from_websocket_message_lowercase_side(self) -> None:
+        """Test that lowercase side is normalized."""
+        data = {
+            "side": "buy",
+            "price": 0.5,
+            "size": 10,
+        }
+        trade = TradeEvent.from_websocket_message(data)
+
+        assert trade.side == "BUY"
+
+    def test_from_websocket_message_sell_side(self) -> None:
+        """Test SELL side handling."""
+        data = {
+            "side": "sell",
+            "price": 0.5,
+            "size": 10,
+        }
+        trade = TradeEvent.from_websocket_message(data)
+
+        assert trade.side == "SELL"
+
+    def test_is_buy(self) -> None:
+        """Test is_buy property."""
+        buy_trade = TradeEvent(
+            market_id="",
+            trade_id="",
+            wallet_address="",
+            side="BUY",
+            outcome="",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("10"),
+            timestamp=datetime.now(timezone.utc),
+            asset_id="",
+        )
+        sell_trade = TradeEvent(
+            market_id="",
+            trade_id="",
+            wallet_address="",
+            side="SELL",
+            outcome="",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("10"),
+            timestamp=datetime.now(timezone.utc),
+            asset_id="",
+        )
+
+        assert buy_trade.is_buy is True
+        assert buy_trade.is_sell is False
+        assert sell_trade.is_buy is False
+        assert sell_trade.is_sell is True
+
+    def test_notional_value(self) -> None:
+        """Test notional value calculation."""
+        trade = TradeEvent(
+            market_id="",
+            trade_id="",
+            wallet_address="",
+            side="BUY",
+            outcome="",
+            outcome_index=0,
+            price=Decimal("0.65"),
+            size=Decimal("100"),
+            timestamp=datetime.now(timezone.utc),
+            asset_id="",
+        )
+
+        assert trade.notional_value == Decimal("65")
+
+    def test_frozen(self) -> None:
+        """Test that TradeEvent is immutable."""
+        trade = TradeEvent(
+            market_id="0x123",
+            trade_id="0xtx",
+            wallet_address="0xwallet",
+            side="BUY",
+            outcome="Yes",
+            outcome_index=0,
+            price=Decimal("0.5"),
+            size=Decimal("10"),
+            timestamp=datetime.now(timezone.utc),
+            asset_id="token",
+        )
+        with pytest.raises(AttributeError):
+            trade.market_id = "0x456"  # type: ignore[misc]

tests/ingestor/test_publisher.py

@@ -0,0 +1,470 @@
+"""Tests for the Redis Streams event publisher."""
+
+from datetime import UTC, datetime
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from redis.exceptions import ResponseError
+
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.ingestor.publisher import (
+    DEFAULT_BLOCK_MS,
+    DEFAULT_COUNT,
+    DEFAULT_MAX_LEN,
+    DEFAULT_STREAM_NAME,
+    ConsumerGroupExistsError,
+    EventPublisher,
+    StreamEntry,
+    _deserialize_trade_event,
+    _serialize_trade_event,
+)
+
+
+# Test fixtures
+@pytest.fixture
+def sample_trade_event() -> TradeEvent:
+    """Create a sample trade event."""
+    return TradeEvent(
+        market_id="0xmarket123",
+        trade_id="0xtx456",
+        wallet_address="0xwallet789",
+        side="BUY",
+        outcome="Yes",
+        outcome_index=0,
+        price=Decimal("0.65"),
+        size=Decimal("1000"),
+        timestamp=datetime(2026, 1, 4, 12, 0, 0, tzinfo=UTC),
+        asset_id="token123",
+        market_slug="will-it-rain",
+        event_slug="weather-markets",
+        event_title="Weather Predictions",
+        trader_name="Alice",
+        trader_pseudonym="AliceTrader",
+    )
+
+
+@pytest.fixture
+def mock_redis() -> AsyncMock:
+    """Create a mock Redis client."""
+    redis = AsyncMock()
+    redis.xadd = AsyncMock(return_value="1704369600000-0")
+    redis.xreadgroup = AsyncMock(return_value=[])
+    redis.xack = AsyncMock(return_value=1)
+    redis.xlen = AsyncMock(return_value=100)
+    redis.xtrim = AsyncMock(return_value=0)
+    redis.xinfo_stream = AsyncMock(return_value={"length": 100})
+    redis.xgroup_create = AsyncMock()
+    redis.pipeline = MagicMock()
+    return redis
+
+
+class TestSerializationFunctions:
+    """Tests for serialization helper functions."""
+
+    def test_serialize_trade_event(self, sample_trade_event: TradeEvent) -> None:
+        """Test serializing a trade event."""
+        data = _serialize_trade_event(sample_trade_event)
+
+        assert data["market_id"] == "0xmarket123"
+        assert data["trade_id"] == "0xtx456"
+        assert data["wallet_address"] == "0xwallet789"
+        assert data["side"] == "BUY"
+        assert data["outcome"] == "Yes"
+        assert data["outcome_index"] == "0"
+        assert data["price"] == "0.65"
+        assert data["size"] == "1000"
+        assert data["timestamp"] == "2026-01-04T12:00:00+00:00"
+        assert data["asset_id"] == "token123"
+        assert data["market_slug"] == "will-it-rain"
+        assert data["trader_name"] == "Alice"
+
+    def test_serialize_all_values_are_strings(self, sample_trade_event: TradeEvent) -> None:
+        """Test that all serialized values are strings."""
+        data = _serialize_trade_event(sample_trade_event)
+
+        for key, value in data.items():
+            assert isinstance(key, str), f"Key {key} is not a string"
+            assert isinstance(value, str), f"Value for {key} is not a string"
+
+    def test_deserialize_trade_event(self, sample_trade_event: TradeEvent) -> None:
+        """Test deserializing a trade event."""
+        data = _serialize_trade_event(sample_trade_event)
+        restored = _deserialize_trade_event(data)
+
+        assert restored.market_id == sample_trade_event.market_id
+        assert restored.trade_id == sample_trade_event.trade_id
+        assert restored.wallet_address == sample_trade_event.wallet_address
+        assert restored.side == sample_trade_event.side
+        assert restored.outcome == sample_trade_event.outcome
+        assert restored.outcome_index == sample_trade_event.outcome_index
+        assert restored.price == sample_trade_event.price
+        assert restored.size == sample_trade_event.size
+        assert restored.timestamp == sample_trade_event.timestamp
+        assert restored.asset_id == sample_trade_event.asset_id
+
+    def test_deserialize_with_bytes_keys(self, sample_trade_event: TradeEvent) -> None:
+        """Test deserializing with bytes keys/values (as returned by Redis)."""
+        data = _serialize_trade_event(sample_trade_event)
+        # Convert to bytes like Redis returns
+        bytes_data = {k.encode(): v.encode() for k, v in data.items()}
+
+        restored = _deserialize_trade_event(bytes_data)
+
+        assert restored.market_id == sample_trade_event.market_id
+        assert restored.side == sample_trade_event.side
+
+    def test_deserialize_with_invalid_timestamp(self) -> None:
+        """Test deserializing with invalid timestamp falls back to now."""
+        data = {
+            "market_id": "0x123",
+            "timestamp": "not-a-timestamp",
+            "side": "BUY",
+            "price": "0.5",
+            "size": "100",
+        }
+
+        event = _deserialize_trade_event(data)
+
+        assert event.market_id == "0x123"
+        # Timestamp should be recent (within last minute)
+        assert (datetime.now(UTC) - event.timestamp).total_seconds() < 60
+
+    def test_deserialize_with_missing_fields(self) -> None:
+        """Test deserializing with missing fields uses defaults."""
+        data = {
+            "market_id": "0x123",
+            "side": "SELL",
+        }
+
+        event = _deserialize_trade_event(data)
+
+        assert event.market_id == "0x123"
+        assert event.side == "SELL"
+        assert event.price == Decimal("0")
+        assert event.outcome == ""
+
+
+class TestEventPublisher:
+    """Tests for the EventPublisher class."""
+
+    def test_init(self, mock_redis: AsyncMock) -> None:
+        """Test initialization."""
+        publisher = EventPublisher(mock_redis)
+
+        assert publisher.stream_name == DEFAULT_STREAM_NAME
+        assert publisher._max_len == DEFAULT_MAX_LEN
+
+    def test_init_custom_config(self, mock_redis: AsyncMock) -> None:
+        """Test initialization with custom config."""
+        publisher = EventPublisher(
+            mock_redis,
+            stream_name="custom-stream",
+            max_len=50_000,
+        )
+
+        assert publisher.stream_name == "custom-stream"
+        assert publisher._max_len == 50_000
+
+    @pytest.mark.asyncio
+    async def test_publish(self, mock_redis: AsyncMock, sample_trade_event: TradeEvent) -> None:
+        """Test publishing a single event."""
+        publisher = EventPublisher(mock_redis)
+
+        entry_id = await publisher.publish(sample_trade_event)
+
+        assert entry_id == "1704369600000-0"
+        mock_redis.xadd.assert_called_once()
+        call_args = mock_redis.xadd.call_args
+        assert call_args[0][0] == DEFAULT_STREAM_NAME
+        assert call_args[1]["maxlen"] == DEFAULT_MAX_LEN
+
+    @pytest.mark.asyncio
+    async def test_publish_returns_decoded_bytes(
+        self, mock_redis: AsyncMock, sample_trade_event: TradeEvent
+    ) -> None:
+        """Test that publish handles bytes entry IDs."""
+        mock_redis.xadd = AsyncMock(return_value=b"1704369600000-0")
+        publisher = EventPublisher(mock_redis)
+
+        entry_id = await publisher.publish(sample_trade_event)
+
+        assert entry_id == "1704369600000-0"
+        assert isinstance(entry_id, str)
+
+    @pytest.mark.asyncio
+    async def test_publish_batch(
+        self, mock_redis: AsyncMock, sample_trade_event: TradeEvent
+    ) -> None:
+        """Test batch publishing."""
+        mock_pipeline = AsyncMock()
+        mock_pipeline.xadd = MagicMock()
+        mock_pipeline.execute = AsyncMock(return_value=["1704369600000-0", "1704369600000-1"])
+        mock_redis.pipeline.return_value = mock_pipeline
+
+        publisher = EventPublisher(mock_redis)
+        events = [sample_trade_event, sample_trade_event]
+
+        entry_ids = await publisher.publish_batch(events)
+
+        assert len(entry_ids) == 2
+        assert entry_ids[0] == "1704369600000-0"
+        assert entry_ids[1] == "1704369600000-1"
+        assert mock_pipeline.xadd.call_count == 2
+
+    @pytest.mark.asyncio
+    async def test_publish_batch_empty(self, mock_redis: AsyncMock) -> None:
+        """Test batch publishing with empty list."""
+        publisher = EventPublisher(mock_redis)
+
+        entry_ids = await publisher.publish_batch([])
+
+        assert entry_ids == []
+        mock_redis.pipeline.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_create_consumer_group(self, mock_redis: AsyncMock) -> None:
+        """Test creating a consumer group."""
+        publisher = EventPublisher(mock_redis)
+
+        await publisher.create_consumer_group("test-group")
+
+        mock_redis.xgroup_create.assert_called_once_with(
+            DEFAULT_STREAM_NAME,
+            "test-group",
+            id="0",
+            mkstream=True,
+        )
+
+    @pytest.mark.asyncio
+    async def test_create_consumer_group_custom_start_id(self, mock_redis: AsyncMock) -> None:
+        """Test creating a consumer group with custom start ID."""
+        publisher = EventPublisher(mock_redis)
+
+        await publisher.create_consumer_group("test-group", start_id="$")
+
+        call_args = mock_redis.xgroup_create.call_args
+        assert call_args[1]["id"] == "$"
+
+    @pytest.mark.asyncio
+    async def test_create_consumer_group_already_exists(self, mock_redis: AsyncMock) -> None:
+        """Test creating a consumer group that already exists."""
+        mock_redis.xgroup_create.side_effect = ResponseError(
+            "BUSYGROUP Consumer Group name already exists"
+        )
+        publisher = EventPublisher(mock_redis)
+
+        with pytest.raises(ConsumerGroupExistsError):
+            await publisher.create_consumer_group("existing-group")
+
+    @pytest.mark.asyncio
+    async def test_ensure_consumer_group_creates(self, mock_redis: AsyncMock) -> None:
+        """Test ensure_consumer_group creates if not exists."""
+        publisher = EventPublisher(mock_redis)
+
+        created = await publisher.ensure_consumer_group("new-group")
+
+        assert created is True
+        mock_redis.xgroup_create.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_ensure_consumer_group_exists(self, mock_redis: AsyncMock) -> None:
+        """Test ensure_consumer_group returns False if exists."""
+        mock_redis.xgroup_create.side_effect = ResponseError("BUSYGROUP")
+        publisher = EventPublisher(mock_redis)
+
+        created = await publisher.ensure_consumer_group("existing-group")
+
+        assert created is False
+
+    @pytest.mark.asyncio
+    async def test_read_events(self, mock_redis: AsyncMock, sample_trade_event: TradeEvent) -> None:
+        """Test reading events from stream."""
+        serialized = _serialize_trade_event(sample_trade_event)
+        mock_redis.xreadgroup = AsyncMock(
+            return_value=[
+                (
+                    "trades",
+                    [
+                        ("1704369600000-0", serialized),
+                    ],
+                )
+            ]
+        )
+        publisher = EventPublisher(mock_redis)
+
+        entries = await publisher.read_events("test-group", "worker-1")
+
+        assert len(entries) == 1
+        assert entries[0].entry_id == "1704369600000-0"
+        assert entries[0].event.market_id == sample_trade_event.market_id
+        mock_redis.xreadgroup.assert_called_once_with(
+            "test-group",
+            "worker-1",
+            {DEFAULT_STREAM_NAME: ">"},
+            count=DEFAULT_COUNT,
+            block=DEFAULT_BLOCK_MS,
+        )
+
+    @pytest.mark.asyncio
+    async def test_read_events_empty(self, mock_redis: AsyncMock) -> None:
+        """Test reading when no events available."""
+        mock_redis.xreadgroup = AsyncMock(return_value=None)
+        publisher = EventPublisher(mock_redis)
+
+        entries = await publisher.read_events("test-group", "worker-1")
+
+        assert entries == []
+
+    @pytest.mark.asyncio
+    async def test_read_events_with_bytes(
+        self, mock_redis: AsyncMock, sample_trade_event: TradeEvent
+    ) -> None:
+        """Test reading events with bytes data (as from real Redis)."""
+        serialized = _serialize_trade_event(sample_trade_event)
+        bytes_data = {k.encode(): v.encode() for k, v in serialized.items()}
+        mock_redis.xreadgroup = AsyncMock(
+            return_value=[
+                (
+                    b"trades",
+                    [
+                        (b"1704369600000-0", bytes_data),
+                    ],
+                )
+            ]
+        )
+        publisher = EventPublisher(mock_redis)
+
+        entries = await publisher.read_events("test-group", "worker-1")
+
+        assert len(entries) == 1
+        assert entries[0].entry_id == "1704369600000-0"
+
+    @pytest.mark.asyncio
+    async def test_read_pending(
+        self, mock_redis: AsyncMock, sample_trade_event: TradeEvent
+    ) -> None:
+        """Test reading pending events."""
+        serialized = _serialize_trade_event(sample_trade_event)
+        mock_redis.xreadgroup = AsyncMock(
+            return_value=[
+                (
+                    "trades",
+                    [
+                        ("1704369600000-0", serialized),
+                    ],
+                )
+            ]
+        )
+        publisher = EventPublisher(mock_redis)
+
+        entries = await publisher.read_pending("test-group", "worker-1")
+
+        assert len(entries) == 1
+        # Should read from "0" not ">"
+        call_args = mock_redis.xreadgroup.call_args
+        assert call_args[0][2] == {DEFAULT_STREAM_NAME: "0"}
+
+    @pytest.mark.asyncio
+    async def test_read_pending_skips_empty_data(self, mock_redis: AsyncMock) -> None:
+        """Test that read_pending skips entries with no data (already acked)."""
+        mock_redis.xreadgroup = AsyncMock(
+            return_value=[
+                (
+                    "trades",
+                    [
+                        ("1704369600000-0", {}),  # Empty = already acked
+                        ("1704369600000-1", None),  # None = already acked
+                    ],
+                )
+            ]
+        )
+        publisher = EventPublisher(mock_redis)
+
+        entries = await publisher.read_pending("test-group", "worker-1")
+
+        assert entries == []
+
+    @pytest.mark.asyncio
+    async def test_ack(self, mock_redis: AsyncMock) -> None:
+        """Test acknowledging entries."""
+        publisher = EventPublisher(mock_redis)
+
+        count = await publisher.ack("test-group", "1704369600000-0", "1704369600000-1")
+
+        assert count == 1  # Mocked return value
+        mock_redis.xack.assert_called_once_with(
+            DEFAULT_STREAM_NAME,
+            "test-group",
+            "1704369600000-0",
+            "1704369600000-1",
+        )
+
+    @pytest.mark.asyncio
+    async def test_ack_empty(self, mock_redis: AsyncMock) -> None:
+        """Test ack with no entry IDs."""
+        publisher = EventPublisher(mock_redis)
+
+        count = await publisher.ack("test-group")
+
+        assert count == 0
+        mock_redis.xack.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_get_stream_info(self, mock_redis: AsyncMock) -> None:
+        """Test getting stream info."""
+        publisher = EventPublisher(mock_redis)
+
+        info = await publisher.get_stream_info()
+
+        assert info["length"] == 100
+        mock_redis.xinfo_stream.assert_called_once_with(DEFAULT_STREAM_NAME)
+
+    @pytest.mark.asyncio
+    async def test_get_stream_info_not_exists(self, mock_redis: AsyncMock) -> None:
+        """Test getting stream info when stream doesn't exist."""
+        mock_redis.xinfo_stream.side_effect = ResponseError("ERR no such key")
+        publisher = EventPublisher(mock_redis)
+
+        info = await publisher.get_stream_info()
+
+        assert info == {}
+
+    @pytest.mark.asyncio
+    async def test_get_stream_length(self, mock_redis: AsyncMock) -> None:
+        """Test getting stream length."""
+        publisher = EventPublisher(mock_redis)
+
+        length = await publisher.get_stream_length()
+
+        assert length == 100
+        mock_redis.xlen.assert_called_once_with(DEFAULT_STREAM_NAME)
+
+    @pytest.mark.asyncio
+    async def test_trim_stream(self, mock_redis: AsyncMock) -> None:
+        """Test trimming stream."""
+        publisher = EventPublisher(mock_redis)
+
+        await publisher.trim_stream(50_000)
+
+        mock_redis.xtrim.assert_called_once_with(DEFAULT_STREAM_NAME, maxlen=50_000)
+
+    @pytest.mark.asyncio
+    async def test_trim_stream_default(self, mock_redis: AsyncMock) -> None:
+        """Test trimming stream with default max_len."""
+        publisher = EventPublisher(mock_redis)
+
+        await publisher.trim_stream()
+
+        mock_redis.xtrim.assert_called_once_with(DEFAULT_STREAM_NAME, maxlen=DEFAULT_MAX_LEN)
+
+
+class TestStreamEntry:
+    """Tests for the StreamEntry dataclass."""
+
+    def test_stream_entry(self, sample_trade_event: TradeEvent) -> None:
+        """Test creating a StreamEntry."""
+        entry = StreamEntry(entry_id="1704369600000-0", event=sample_trade_event)
+
+        assert entry.entry_id == "1704369600000-0"
+        assert entry.event == sample_trade_event

tests/ingestor/test_websocket.py

@@ -0,0 +1,343 @@
+"""Tests for WebSocket trade stream handler."""
+
+import asyncio
+import json
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from polymarket_insider_tracker.ingestor.models import TradeEvent
+from polymarket_insider_tracker.ingestor.websocket import (
+    ConnectionState,
+    StreamStats,
+    TradeStreamHandler,
+)
+
+
+class TestStreamStats:
+    """Tests for StreamStats."""
+
+    def test_defaults(self) -> None:
+        """Test default values."""
+        stats = StreamStats()
+
+        assert stats.trades_received == 0
+        assert stats.reconnect_count == 0
+        assert stats.last_trade_time is None
+        assert stats.connected_since is None
+        assert stats.last_error is None
+
+
+class TestTradeStreamHandler:
+    """Tests for TradeStreamHandler."""
+
+    @pytest.fixture
+    def on_trade_mock(self) -> AsyncMock:
+        """Create mock trade callback."""
+        return AsyncMock()
+
+    @pytest.fixture
+    def on_state_change_mock(self) -> AsyncMock:
+        """Create mock state change callback."""
+        return AsyncMock()
+
+    @pytest.fixture
+    def handler(
+        self, on_trade_mock: AsyncMock, on_state_change_mock: AsyncMock
+    ) -> TradeStreamHandler:
+        """Create handler with mocks."""
+        return TradeStreamHandler(
+            on_trade=on_trade_mock,
+            on_state_change=on_state_change_mock,
+            initial_reconnect_delay=0.01,  # Fast reconnect for tests
+            max_reconnect_delay=0.1,
+        )
+
+    def test_init_defaults(self, on_trade_mock: AsyncMock) -> None:
+        """Test handler initialization with defaults."""
+        handler = TradeStreamHandler(on_trade=on_trade_mock)
+
+        assert handler.state == ConnectionState.DISCONNECTED
+        assert handler.stats.trades_received == 0
+        assert handler._host == "wss://ws-live-data.polymarket.com"
+
+    def test_init_custom_host(self, on_trade_mock: AsyncMock) -> None:
+        """Test handler with custom host."""
+        handler = TradeStreamHandler(
+            on_trade=on_trade_mock,
+            host="wss://custom.example.com",
+        )
+
+        assert handler._host == "wss://custom.example.com"
+
+    def test_init_with_event_filter(self, on_trade_mock: AsyncMock) -> None:
+        """Test handler with event filter."""
+        handler = TradeStreamHandler(
+            on_trade=on_trade_mock,
+            event_filter="presidential-election-2024",
+        )
+
+        assert handler._event_filter == "presidential-election-2024"
+
+    def test_build_subscription_message_no_filter(
+        self, handler: TradeStreamHandler
+    ) -> None:
+        """Test building subscription message without filters."""
+        msg = handler._build_subscription_message()
+
+        assert msg == {
+            "subscriptions": [{"topic": "activity", "type": "trades"}]
+        }
+
+    def test_build_subscription_message_with_event_filter(
+        self, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test building subscription message with event filter."""
+        handler = TradeStreamHandler(
+            on_trade=on_trade_mock,
+            event_filter="test-event",
+        )
+        msg = handler._build_subscription_message()
+
+        assert msg["subscriptions"][0]["filters"] == json.dumps(
+            {"event_slug": "test-event"}
+        )
+
+    def test_build_subscription_message_with_market_filter(
+        self, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test building subscription message with market filter."""
+        handler = TradeStreamHandler(
+            on_trade=on_trade_mock,
+            market_filter="test-market",
+        )
+        msg = handler._build_subscription_message()
+
+        assert msg["subscriptions"][0]["filters"] == json.dumps(
+            {"market_slug": "test-market"}
+        )
+
+    @pytest.mark.asyncio
+    async def test_handle_message_trade(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test handling a valid trade message."""
+        message = json.dumps(
+            {
+                "topic": "activity",
+                "type": "trades",
+                "payload": {
+                    "conditionId": "0xmarket",
+                    "transactionHash": "0xtx",
+                    "proxyWallet": "0xwallet",
+                    "side": "BUY",
+                    "outcome": "Yes",
+                    "price": 0.65,
+                    "size": 100,
+                    "timestamp": 1704067200,
+                    "asset": "token123",
+                },
+            }
+        )
+
+        await handler._handle_message(message)
+
+        on_trade_mock.assert_called_once()
+        trade: TradeEvent = on_trade_mock.call_args[0][0]
+        assert trade.market_id == "0xmarket"
+        assert trade.side == "BUY"
+        assert trade.price == Decimal("0.65")
+        assert handler.stats.trades_received == 1
+
+    @pytest.mark.asyncio
+    async def test_handle_message_non_trade(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test handling a non-trade message."""
+        message = json.dumps(
+            {
+                "topic": "comments",
+                "type": "comment_created",
+                "payload": {"body": "Hello"},
+            }
+        )
+
+        await handler._handle_message(message)
+
+        on_trade_mock.assert_not_called()
+        assert handler.stats.trades_received == 0
+
+    @pytest.mark.asyncio
+    async def test_handle_message_invalid_json(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test handling invalid JSON message."""
+        await handler._handle_message("not valid json")
+
+        on_trade_mock.assert_not_called()
+        assert handler.stats.trades_received == 0
+
+    @pytest.mark.asyncio
+    async def test_handle_message_callback_error(
+        self, handler: TradeStreamHandler, on_trade_mock: AsyncMock
+    ) -> None:
+        """Test that callback errors don't crash the handler."""
+        on_trade_mock.side_effect = ValueError("Callback error")
+
+        message = json.dumps(
+            {
+                "topic": "activity",
+                "type": "trades",
+                "payload": {
+                    "conditionId": "0x",
+                    "transactionHash": "0x",
+                    "proxyWallet": "0x",
+                    "side": "BUY",
+                    "price": 0.5,
+                    "size": 10,
+                },
+            }
+        )
+
+        # Should not raise
+        await handler._handle_message(message)
+
+        # Trade was still counted
+        assert handler.stats.trades_received == 1
+
+    @pytest.mark.asyncio
+    async def test_set_state_calls_callback(
+        self, handler: TradeStreamHandler, on_state_change_mock: AsyncMock
+    ) -> None:
+        """Test that state changes trigger callback."""
+        await handler._set_state(ConnectionState.CONNECTING)
+
+        on_state_change_mock.assert_called_once_with(ConnectionState.CONNECTING)
+        assert handler.state == ConnectionState.CONNECTING
+
+    @pytest.mark.asyncio
+    async def test_set_state_same_state_no_callback(
+        self, handler: TradeStreamHandler, on_state_change_mock: AsyncMock
+    ) -> None:
+        """Test that same state doesn't trigger callback."""
+        handler._state = ConnectionState.CONNECTED
+
+        await handler._set_state(ConnectionState.CONNECTED)
+
+        on_state_change_mock.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_stop_when_not_running(self, handler: TradeStreamHandler) -> None:
+        """Test stop when handler is not running."""
+        # Should not raise
+        await handler.stop()
+
+    @pytest.mark.asyncio
+    async def test_connect_sends_subscription(
+        self, handler: TradeStreamHandler, on_state_change_mock: AsyncMock
+    ) -> None:
+        """Test that connection sends subscription message."""
+        mock_ws = AsyncMock()
+        mock_ws.send = AsyncMock()
+
+        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
+            ws = await handler._connect()
+
+            assert ws is mock_ws
+            mock_ws.send.assert_called_once()
+
+            # Verify subscription message
+            sent_msg = json.loads(mock_ws.send.call_args[0][0])
+            assert "subscriptions" in sent_msg
+            assert sent_msg["subscriptions"][0]["topic"] == "activity"
+            assert sent_msg["subscriptions"][0]["type"] == "trades"
+
+    @pytest.mark.asyncio
+    async def test_cleanup_closes_websocket(
+        self, handler: TradeStreamHandler
+    ) -> None:
+        """Test that cleanup closes the WebSocket."""
+        mock_ws = AsyncMock()
+        mock_ws.close = AsyncMock()
+        handler._ws = mock_ws
+
+        await handler._cleanup()
+
+        mock_ws.close.assert_called_once()
+        assert handler._ws is None
+        assert handler.state == ConnectionState.DISCONNECTED
+
+    @pytest.mark.asyncio
+    async def test_context_manager(self, on_trade_mock: AsyncMock) -> None:
+        """Test async context manager."""
+        handler = TradeStreamHandler(on_trade=on_trade_mock)
+
+        async with handler:
+            pass
+
+        # Should be stopped after exiting context
+        assert handler._running is False
+
+
+class TestTradeStreamHandlerIntegration:
+    """Integration tests for TradeStreamHandler.
+
+    These tests verify the full message flow with mocked WebSocket.
+    """
+
+    @pytest.mark.asyncio
+    async def test_start_and_receive_trades(self) -> None:
+        """Test starting handler and receiving trades."""
+        received_trades: list[TradeEvent] = []
+
+        async def on_trade(trade: TradeEvent) -> None:
+            received_trades.append(trade)
+
+        handler = TradeStreamHandler(
+            on_trade=on_trade,
+            initial_reconnect_delay=0.01,
+        )
+
+        # Create mock WebSocket that sends one trade then closes
+        mock_ws = MagicMock()
+        mock_ws.send = AsyncMock()
+        mock_ws.close = AsyncMock()
+
+        trade_message = json.dumps(
+            {
+                "topic": "activity",
+                "type": "trades",
+                "payload": {
+                    "conditionId": "0xtest",
+                    "transactionHash": "0xtx",
+                    "proxyWallet": "0xwallet",
+                    "side": "BUY",
+                    "outcome": "Yes",
+                    "price": 0.75,
+                    "size": 50,
+                    "timestamp": 1704067200,
+                    "asset": "token",
+                },
+            }
+        )
+
+        # Mock async iteration
+        async def mock_iter() -> None:
+            yield trade_message
+            await handler.stop()  # Stop after first message
+
+        mock_ws.__aiter__ = mock_iter
+
+        with patch("websockets.connect", AsyncMock(return_value=mock_ws)):
+            # Run with timeout to prevent hanging
+            try:
+                await asyncio.wait_for(handler.start(), timeout=1.0)
+            except asyncio.TimeoutError:
+                await handler.stop()
+
+        # Verify trade was received
+        assert len(received_trades) == 1
+        assert received_trades[0].market_id == "0xtest"
+        assert received_trades[0].side == "BUY"
+        assert received_trades[0].price == Decimal("0.75")

tests/profiler/__init__.py

@@ -0,0 +1 @@
+"""Tests for profiler module."""

tests/profiler/test_analyzer.py

@@ -0,0 +1,395 @@
+"""Tests for the wallet analyzer."""
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+from unittest.mock import AsyncMock
+
+import pytest
+
+from polymarket_insider_tracker.profiler.analyzer import (
+    DEFAULT_FRESH_THRESHOLD,
+    USDC_POLYGON_ADDRESS,
+    WalletAnalyzer,
+)
+from polymarket_insider_tracker.profiler.models import Transaction, WalletInfo
+
+# Valid Ethereum addresses for testing
+VALID_ADDRESS = "0x742d35Cc6634C0532925a3b844Bc9e7595f5eaE2"
+VALID_ADDRESS_2 = "0x8ba1f109551bD432803012645Ac136ddd64DBA72"
+
+
+class TestWalletAnalyzerInit:
+    """Tests for WalletAnalyzer initialization."""
+
+    def test_init_default(self) -> None:
+        """Test initialization with defaults."""
+        client = AsyncMock()
+        analyzer = WalletAnalyzer(client)
+
+        assert analyzer._client is client
+        assert analyzer._redis is None
+        assert analyzer._fresh_threshold == DEFAULT_FRESH_THRESHOLD
+        assert analyzer._usdc_address == USDC_POLYGON_ADDRESS
+
+    def test_init_with_redis(self) -> None:
+        """Test initialization with Redis."""
+        client = AsyncMock()
+        redis = AsyncMock()
+        analyzer = WalletAnalyzer(client, redis=redis)
+
+        assert analyzer._redis is redis
+
+    def test_init_custom_threshold(self) -> None:
+        """Test initialization with custom threshold."""
+        client = AsyncMock()
+        analyzer = WalletAnalyzer(client, fresh_threshold=10)
+
+        assert analyzer._fresh_threshold == 10
+
+
+class TestWalletAnalyzerAnalyze:
+    """Tests for the analyze method."""
+
+    @pytest.fixture
+    def mock_client(self) -> AsyncMock:
+        """Create a mock PolygonClient."""
+        client = AsyncMock()
+        client.get_wallet_info = AsyncMock()
+        client.get_token_balance = AsyncMock(return_value=Decimal("1000000"))
+        return client
+
+    @pytest.fixture
+    def mock_redis(self) -> AsyncMock:
+        """Create a mock Redis client."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+        return redis
+
+    @pytest.mark.asyncio
+    async def test_analyze_fresh_wallet(self, mock_client: AsyncMock) -> None:
+        """Test analyzing a fresh wallet."""
+        first_tx = Transaction(
+            hash="0xabc",
+            block_number=1000,
+            timestamp=datetime.now(UTC) - timedelta(hours=12),
+            from_address="0xfaucet",
+            to_address=VALID_ADDRESS.lower(),
+            value=Decimal("1000000000000000000"),
+            gas_used=21000,
+            gas_price=Decimal("50000000000"),
+        )
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=3,
+            balance_wei=Decimal("5000000000000000000"),
+            first_transaction=first_tx,
+        )
+
+        analyzer = WalletAnalyzer(mock_client)
+        profile = await analyzer.analyze(VALID_ADDRESS)
+
+        assert profile.address == VALID_ADDRESS.lower()
+        assert profile.nonce == 3
+        assert profile.is_fresh is True
+        assert profile.first_seen is not None
+        assert profile.age_hours is not None
+        assert 11 < profile.age_hours < 13
+
+    @pytest.mark.asyncio
+    async def test_analyze_old_wallet(self, mock_client: AsyncMock) -> None:
+        """Test analyzing an old wallet with many transactions."""
+        first_tx = Transaction(
+            hash="0xabc",
+            block_number=1000,
+            timestamp=datetime.now(UTC) - timedelta(days=365),
+            from_address="0xfaucet",
+            to_address=VALID_ADDRESS.lower(),
+            value=Decimal("1000000000000000000"),
+            gas_used=21000,
+            gas_price=Decimal("50000000000"),
+        )
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=500,
+            balance_wei=Decimal("100000000000000000000"),
+            first_transaction=first_tx,
+        )
+
+        analyzer = WalletAnalyzer(mock_client)
+        profile = await analyzer.analyze(VALID_ADDRESS)
+
+        assert profile.nonce == 500
+        assert profile.is_fresh is False
+        assert profile.age_hours is not None
+        assert profile.age_hours > 8000  # Over 365 days in hours
+
+    @pytest.mark.asyncio
+    async def test_analyze_brand_new_wallet(self, mock_client: AsyncMock) -> None:
+        """Test analyzing a wallet with no transactions."""
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=0,
+            balance_wei=Decimal("1000000000000000000"),
+            first_transaction=None,
+        )
+
+        analyzer = WalletAnalyzer(mock_client)
+        profile = await analyzer.analyze(VALID_ADDRESS)
+
+        assert profile.nonce == 0
+        assert profile.is_fresh is True
+        assert profile.is_brand_new is True
+        assert profile.first_seen is None
+        assert profile.age_hours is None
+
+    @pytest.mark.asyncio
+    async def test_analyze_uses_cache(
+        self, mock_client: AsyncMock, mock_redis: AsyncMock
+    ) -> None:
+        """Test that analyze uses cached data."""
+        cached_data = {
+            "address": VALID_ADDRESS.lower(),
+            "nonce": 2,
+            "first_seen": datetime.now(UTC).isoformat(),
+            "age_hours": 6.0,
+            "is_fresh": True,
+            "total_tx_count": 2,
+            "matic_balance": "1000000000000000000",
+            "usdc_balance": "500000",
+            "analyzed_at": datetime.now(UTC).isoformat(),
+            "fresh_threshold": 5,
+        }
+        mock_redis.get = AsyncMock(return_value=str(cached_data).replace("'", '"').encode())
+
+        # Actually mock it properly with json
+        import json
+        mock_redis.get = AsyncMock(return_value=json.dumps(cached_data).encode())
+
+        analyzer = WalletAnalyzer(mock_client, redis=mock_redis)
+        profile = await analyzer.analyze(VALID_ADDRESS)
+
+        assert profile.address == VALID_ADDRESS.lower()
+        assert profile.nonce == 2
+        mock_client.get_wallet_info.assert_not_called()
+
+    @pytest.mark.asyncio
+    async def test_analyze_force_refresh(
+        self, mock_client: AsyncMock, mock_redis: AsyncMock
+    ) -> None:
+        """Test that force_refresh bypasses cache."""
+        import json
+
+        cached_data = {
+            "address": VALID_ADDRESS.lower(),
+            "nonce": 1,
+            "first_seen": None,
+            "age_hours": None,
+            "is_fresh": True,
+            "total_tx_count": 1,
+            "matic_balance": "1000",
+            "usdc_balance": "0",
+            "analyzed_at": datetime.now(UTC).isoformat(),
+            "fresh_threshold": 5,
+        }
+        mock_redis.get = AsyncMock(return_value=json.dumps(cached_data).encode())
+
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=10,
+            balance_wei=Decimal("5000000000000000000"),
+            first_transaction=None,
+        )
+
+        analyzer = WalletAnalyzer(mock_client, redis=mock_redis)
+        profile = await analyzer.analyze(VALID_ADDRESS, force_refresh=True)
+
+        assert profile.nonce == 10  # From fresh query, not cache
+        mock_client.get_wallet_info.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_analyze_handles_usdc_error(self, mock_client: AsyncMock) -> None:
+        """Test that USDC balance error is handled gracefully."""
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=1,
+            balance_wei=Decimal("1000000000000000000"),
+            first_transaction=None,
+        )
+        mock_client.get_token_balance.side_effect = Exception("Token query failed")
+
+        analyzer = WalletAnalyzer(mock_client)
+        profile = await analyzer.analyze(VALID_ADDRESS)
+
+        assert profile.usdc_balance == Decimal(0)
+
+
+class TestWalletAnalyzerIsFresh:
+    """Tests for the is_fresh method."""
+
+    @pytest.fixture
+    def mock_client(self) -> AsyncMock:
+        """Create a mock PolygonClient."""
+        client = AsyncMock()
+        client.get_wallet_info = AsyncMock()
+        client.get_token_balance = AsyncMock(return_value=Decimal("0"))
+        return client
+
+    @pytest.mark.asyncio
+    async def test_is_fresh_true(self, mock_client: AsyncMock) -> None:
+        """Test is_fresh returns True for fresh wallet."""
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=2,
+            balance_wei=Decimal("1000000000000000000"),
+            first_transaction=None,
+        )
+
+        analyzer = WalletAnalyzer(mock_client)
+        result = await analyzer.is_fresh(VALID_ADDRESS)
+
+        assert result is True
+
+    @pytest.mark.asyncio
+    async def test_is_fresh_false(self, mock_client: AsyncMock) -> None:
+        """Test is_fresh returns False for old wallet."""
+        mock_client.get_wallet_info.return_value = WalletInfo(
+            address=VALID_ADDRESS.lower(),
+            transaction_count=100,
+            balance_wei=Decimal("1000000000000000000"),
+            first_transaction=None,
+        )
+
+        analyzer = WalletAnalyzer(mock_client)
+        result = await analyzer.is_fresh(VALID_ADDRESS)
+
+        assert result is False
+
+
+class TestWalletAnalyzerFreshnessLogic:
+    """Tests for freshness determination logic."""
+
+    @pytest.fixture
+    def mock_client(self) -> AsyncMock:
+        """Create a mock PolygonClient."""
+        return AsyncMock()
+
+    def test_is_wallet_fresh_low_nonce_no_age(self, mock_client: AsyncMock) -> None:
+        """Test fresh wallet with low nonce and unknown age."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+        result = analyzer._is_wallet_fresh(nonce=2, age_hours=None)
+        assert result is True
+
+    def test_is_wallet_fresh_low_nonce_young_age(self, mock_client: AsyncMock) -> None:
+        """Test fresh wallet with low nonce and young age."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+        result = analyzer._is_wallet_fresh(nonce=2, age_hours=12.0)
+        assert result is True
+
+    def test_is_wallet_fresh_low_nonce_old_age(self, mock_client: AsyncMock) -> None:
+        """Test not fresh when nonce is low but age is old."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+        result = analyzer._is_wallet_fresh(nonce=2, age_hours=100.0)
+        assert result is False
+
+    def test_is_wallet_fresh_high_nonce(self, mock_client: AsyncMock) -> None:
+        """Test not fresh when nonce is high."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+        result = analyzer._is_wallet_fresh(nonce=10, age_hours=12.0)
+        assert result is False
+
+    def test_is_wallet_fresh_at_threshold(self, mock_client: AsyncMock) -> None:
+        """Test not fresh when nonce equals threshold."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+        result = analyzer._is_wallet_fresh(nonce=5, age_hours=12.0)
+        assert result is False
+
+    def test_is_wallet_fresh_at_age_boundary(self, mock_client: AsyncMock) -> None:
+        """Test at 48 hour boundary."""
+        analyzer = WalletAnalyzer(mock_client, fresh_threshold=5)
+
+        result_under = analyzer._is_wallet_fresh(nonce=2, age_hours=47.9)
+        assert result_under is True
+
+        result_over = analyzer._is_wallet_fresh(nonce=2, age_hours=48.1)
+        assert result_over is False
+
+
+class TestWalletAnalyzerBatch:
+    """Tests for batch analysis."""
+
+    @pytest.fixture
+    def mock_client(self) -> AsyncMock:
+        """Create a mock PolygonClient."""
+        client = AsyncMock()
+        client.get_wallet_info = AsyncMock()
+        client.get_token_balance = AsyncMock(return_value=Decimal("0"))
+        return client
+
+    @pytest.mark.asyncio
+    async def test_analyze_batch(self, mock_client: AsyncMock) -> None:
+        """Test batch analysis."""
+        mock_client.get_wallet_info.side_effect = [
+            WalletInfo(
+                address=VALID_ADDRESS.lower(),
+                transaction_count=2,
+                balance_wei=Decimal("1000"),
+                first_transaction=None,
+            ),
+            WalletInfo(
+                address=VALID_ADDRESS_2.lower(),
+                transaction_count=100,
+                balance_wei=Decimal("2000"),
+                first_transaction=None,
+            ),
+        ]
+
+        analyzer = WalletAnalyzer(mock_client)
+        profiles = await analyzer.analyze_batch([VALID_ADDRESS, VALID_ADDRESS_2])
+
+        assert len(profiles) == 2
+        assert profiles[VALID_ADDRESS.lower()].is_fresh is True
+        assert profiles[VALID_ADDRESS_2.lower()].is_fresh is False
+
+    @pytest.mark.asyncio
+    async def test_analyze_batch_handles_errors(self, mock_client: AsyncMock) -> None:
+        """Test batch analysis handles individual failures."""
+        mock_client.get_wallet_info.side_effect = [
+            WalletInfo(
+                address=VALID_ADDRESS.lower(),
+                transaction_count=2,
+                balance_wei=Decimal("1000"),
+                first_transaction=None,
+            ),
+            Exception("RPC error"),
+        ]
+
+        analyzer = WalletAnalyzer(mock_client)
+        profiles = await analyzer.analyze_batch([VALID_ADDRESS, VALID_ADDRESS_2])
+
+        assert len(profiles) == 1
+        assert VALID_ADDRESS.lower() in profiles
+
+    @pytest.mark.asyncio
+    async def test_get_fresh_wallets(self, mock_client: AsyncMock) -> None:
+        """Test filtering to only fresh wallets."""
+        mock_client.get_wallet_info.side_effect = [
+            WalletInfo(
+                address=VALID_ADDRESS.lower(),
+                transaction_count=2,
+                balance_wei=Decimal("1000"),
+                first_transaction=None,
+            ),
+            WalletInfo(
+                address=VALID_ADDRESS_2.lower(),
+                transaction_count=100,
+                balance_wei=Decimal("2000"),
+                first_transaction=None,
+            ),
+        ]
+
+        analyzer = WalletAnalyzer(mock_client)
+        fresh = await analyzer.get_fresh_wallets([VALID_ADDRESS, VALID_ADDRESS_2])
+
+        assert len(fresh) == 1
+        assert VALID_ADDRESS.lower() in fresh

tests/profiler/test_chain.py

@@ -0,0 +1,500 @@
+"""Tests for the Polygon blockchain client."""
+
+import asyncio
+from decimal import Decimal
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from web3.exceptions import Web3Exception
+
+from polymarket_insider_tracker.profiler.chain import (
+    DEFAULT_CACHE_TTL_SECONDS,
+    PolygonClient,
+    RateLimiter,
+    RPCError,
+)
+
+# Valid Ethereum addresses for testing
+VALID_ADDRESS = "0x742d35Cc6634C0532925a3b844Bc9e7595f5eaE2"
+VALID_ADDRESS_2 = "0x8ba1f109551bD432803012645Ac136ddd64DBA72"
+VALID_ADDRESS_3 = "0x1234567890AbCdEf1234567890ABcDeF12345678"
+VALID_TOKEN = "0x7D1AfA7B718fb893dB30A3aBc0Cfc608AaCfeBB0"  # MATIC token
+
+
+class TestRateLimiter:
+    """Tests for the RateLimiter class."""
+
+    def test_create(self) -> None:
+        """Test creating a rate limiter."""
+        limiter = RateLimiter.create(10.0)
+
+        assert limiter.max_tokens == 10.0
+        assert limiter.refill_rate == 10.0
+        assert limiter.tokens == 10.0
+
+    @pytest.mark.asyncio
+    async def test_acquire_available(self) -> None:
+        """Test acquiring when tokens are available."""
+        limiter = RateLimiter.create(10.0)
+
+        await limiter.acquire(1.0)
+
+        assert limiter.tokens < 10.0
+
+    @pytest.mark.asyncio
+    async def test_acquire_multiple(self) -> None:
+        """Test acquiring multiple tokens."""
+        limiter = RateLimiter.create(10.0)
+
+        for _ in range(5):
+            await limiter.acquire(1.0)
+
+        assert limiter.tokens < 6.0
+
+    @pytest.mark.asyncio
+    async def test_acquire_waits_when_empty(self) -> None:
+        """Test that acquire waits when tokens are depleted."""
+        limiter = RateLimiter.create(2.0)
+
+        # Deplete tokens
+        await limiter.acquire(2.0)
+
+        # This should wait briefly for refill
+        start = asyncio.get_event_loop().time()
+        await limiter.acquire(0.5)
+        elapsed = asyncio.get_event_loop().time() - start
+
+        # Should have waited some time
+        assert elapsed >= 0.1
+
+
+class TestPolygonClient:
+    """Tests for the PolygonClient class."""
+
+    @pytest.fixture
+    def mock_redis(self) -> AsyncMock:
+        """Create a mock Redis client."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+        return redis
+
+    @pytest.fixture
+    def mock_w3(self) -> MagicMock:
+        """Create a mock Web3 instance."""
+        w3 = MagicMock()
+        w3.eth = MagicMock()
+        w3.eth.get_transaction_count = AsyncMock(return_value=42)
+        w3.eth.get_balance = AsyncMock(return_value=1000000000000000000)
+        w3.eth.get_block = AsyncMock(return_value={"timestamp": 1704369600})
+        w3.eth.block_number = AsyncMock(return_value=50000000)
+        return w3
+
+    def test_init(self) -> None:
+        """Test initialization."""
+        client = PolygonClient("https://polygon-rpc.com")
+
+        assert client._rpc_url == "https://polygon-rpc.com"
+        assert client._fallback_rpc_url is None
+        assert client._cache_ttl == DEFAULT_CACHE_TTL_SECONDS
+
+    def test_init_with_fallback(self) -> None:
+        """Test initialization with fallback RPC."""
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            fallback_rpc_url="https://fallback.com",
+        )
+
+        assert client._fallback_rpc_url == "https://fallback.com"
+        assert client._w3_fallback is not None
+
+    def test_init_custom_config(self) -> None:
+        """Test initialization with custom config."""
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            cache_ttl_seconds=600,
+            max_requests_per_second=50,
+            max_retries=5,
+        )
+
+        assert client._cache_ttl == 600
+        assert client._max_retries == 5
+        assert client._rate_limiter.max_tokens == 50
+
+    def test_cache_key(self) -> None:
+        """Test cache key generation."""
+        client = PolygonClient("https://polygon-rpc.com")
+
+        key = client._cache_key("nonce", "0xAbC123")
+
+        assert key == "polygon:nonce:0xabc123"
+
+    @pytest.mark.asyncio
+    async def test_get_cached_miss(self, mock_redis: AsyncMock) -> None:
+        """Test cache miss."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        result = await client._get_cached("test:key")
+
+        assert result is None
+        mock_redis.get.assert_called_once_with("test:key")
+
+    @pytest.mark.asyncio
+    async def test_get_cached_hit(self, mock_redis: AsyncMock) -> None:
+        """Test cache hit."""
+        mock_redis.get = AsyncMock(return_value=b"cached_value")
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        result = await client._get_cached("test:key")
+
+        assert result == "cached_value"
+
+    @pytest.mark.asyncio
+    async def test_get_cached_error_handling(self, mock_redis: AsyncMock) -> None:
+        """Test that cache errors are handled gracefully."""
+        mock_redis.get = AsyncMock(side_effect=Exception("Redis error"))
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        result = await client._get_cached("test:key")
+
+        assert result is None  # Should not raise
+
+    @pytest.mark.asyncio
+    async def test_set_cached(self, mock_redis: AsyncMock) -> None:
+        """Test setting cache."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        await client._set_cached("test:key", "value")
+
+        mock_redis.set.assert_called_once_with(
+            "test:key", "value", ex=DEFAULT_CACHE_TTL_SECONDS
+        )
+
+    @pytest.mark.asyncio
+    async def test_set_cached_custom_ttl(self, mock_redis: AsyncMock) -> None:
+        """Test setting cache with custom TTL."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        await client._set_cached("test:key", "value", ttl=3600)
+
+        mock_redis.set.assert_called_once_with("test:key", "value", ex=3600)
+
+    @pytest.mark.asyncio
+    async def test_get_transaction_count_cached(self, mock_redis: AsyncMock) -> None:
+        """Test getting transaction count from cache."""
+        mock_redis.get = AsyncMock(return_value=b"42")
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        count = await client.get_transaction_count(VALID_ADDRESS)
+
+        assert count == 42
+        mock_redis.get.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_get_transaction_count_uncached(self, mock_redis: AsyncMock) -> None:
+        """Test getting transaction count from blockchain."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "_execute_with_retry", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = 42
+
+            count = await client.get_transaction_count(VALID_ADDRESS)
+
+            assert count == 42
+            mock_exec.assert_called_once()
+            mock_redis.set.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_get_transaction_counts_batch(self, mock_redis: AsyncMock) -> None:
+        """Test batch getting transaction counts."""
+        mock_redis.get = AsyncMock(side_effect=[b"10", None, b"30"])
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "get_transaction_count", new_callable=AsyncMock) as mock_get:
+            mock_get.return_value = 20
+
+            addresses = [VALID_ADDRESS, VALID_ADDRESS_2, VALID_ADDRESS_3]
+            counts = await client.get_transaction_counts(addresses)
+
+            assert counts[VALID_ADDRESS.lower()] == 10  # From cache
+            assert counts[VALID_ADDRESS_2.lower()] == 20  # From blockchain
+            assert counts[VALID_ADDRESS_3.lower()] == 30  # From cache
+
+    @pytest.mark.asyncio
+    async def test_get_transaction_counts_empty(self, mock_redis: AsyncMock) -> None:
+        """Test batch with empty list."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        counts = await client.get_transaction_counts([])
+
+        assert counts == {}
+
+    @pytest.mark.asyncio
+    async def test_get_balance_cached(self, mock_redis: AsyncMock) -> None:
+        """Test getting balance from cache."""
+        mock_redis.get = AsyncMock(return_value=b"1000000000000000000")
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        balance = await client.get_balance(VALID_ADDRESS)
+
+        assert balance == Decimal("1000000000000000000")
+
+    @pytest.mark.asyncio
+    async def test_get_balance_uncached(self, mock_redis: AsyncMock) -> None:
+        """Test getting balance from blockchain."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "_execute_with_retry", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = 2000000000000000000
+
+            balance = await client.get_balance(VALID_ADDRESS)
+
+            assert balance == Decimal("2000000000000000000")
+            mock_redis.set.assert_called_once()
+
+    @pytest.mark.asyncio
+    async def test_get_wallet_info(self, mock_redis: AsyncMock) -> None:
+        """Test getting aggregated wallet info."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with (
+            patch.object(client, "get_transaction_count", new_callable=AsyncMock) as mock_nonce,
+            patch.object(client, "get_balance", new_callable=AsyncMock) as mock_balance,
+            patch.object(client, "get_first_transaction", new_callable=AsyncMock) as mock_tx,
+        ):
+            mock_nonce.return_value = 42
+            mock_balance.return_value = Decimal("1000000000000000000")
+            mock_tx.return_value = None
+
+            info = await client.get_wallet_info(VALID_ADDRESS)
+
+            assert info.address == VALID_ADDRESS.lower()
+            assert info.transaction_count == 42
+            assert info.balance_wei == Decimal("1000000000000000000")
+            assert info.first_transaction is None
+
+    @pytest.mark.asyncio
+    async def test_get_first_transaction_no_transactions(
+        self, mock_redis: AsyncMock
+    ) -> None:
+        """Test get_first_transaction when wallet has no transactions."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "get_transaction_count", new_callable=AsyncMock) as mock_nonce:
+            mock_nonce.return_value = 0
+
+            tx = await client.get_first_transaction(VALID_ADDRESS)
+
+            assert tx is None
+
+    @pytest.mark.asyncio
+    async def test_health_check_success(self, mock_redis: AsyncMock) -> None:
+        """Test successful health check."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "_execute_with_retry", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = 50000000
+
+            healthy = await client.health_check()
+
+            assert healthy is True
+
+    @pytest.mark.asyncio
+    async def test_health_check_failure(self, mock_redis: AsyncMock) -> None:
+        """Test failed health check."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "_execute_with_retry", new_callable=AsyncMock) as mock_exec:
+            mock_exec.side_effect = RPCError("Connection failed")
+
+            healthy = await client.health_check()
+
+            assert healthy is False
+
+
+class TestPolygonClientRetryLogic:
+    """Tests for retry and failover logic."""
+
+    @pytest.fixture
+    def mock_redis(self) -> AsyncMock:
+        """Create a mock Redis client."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+        return redis
+
+    @pytest.mark.asyncio
+    async def test_retry_on_failure(self, mock_redis: AsyncMock) -> None:
+        """Test that client retries on RPC failure."""
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            redis=mock_redis,
+            max_retries=3,
+            retry_delay_seconds=0.01,
+        )
+
+        call_count = 0
+
+        async def mock_get_tx_count(*_args: object, **_kwargs: object) -> int:
+            nonlocal call_count
+            call_count += 1
+            if call_count < 3:
+                raise Web3Exception("Temporary error")
+            return 42
+
+        client._w3.eth.get_transaction_count = mock_get_tx_count
+
+        count = await client.get_transaction_count(VALID_ADDRESS)
+
+        assert count == 42
+        assert call_count == 3
+
+    @pytest.mark.asyncio
+    async def test_failover_to_secondary(self, mock_redis: AsyncMock) -> None:
+        """Test failover to secondary RPC."""
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            fallback_rpc_url="https://fallback.com",
+            redis=mock_redis,
+            max_retries=1,
+            retry_delay_seconds=0.01,
+        )
+
+        # Primary always fails
+        async def primary_fail(*_args: object, **_kwargs: object) -> int:
+            raise Web3Exception("Primary down")
+
+        client._w3.eth.get_transaction_count = primary_fail
+
+        # Fallback works
+        client._w3_fallback.eth.get_transaction_count = AsyncMock(return_value=42)
+
+        count = await client.get_transaction_count(VALID_ADDRESS)
+
+        assert count == 42
+        assert not client._primary_healthy
+
+    @pytest.mark.asyncio
+    async def test_all_retries_exhausted(self, mock_redis: AsyncMock) -> None:
+        """Test error when all retries are exhausted."""
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            redis=mock_redis,
+            max_retries=2,
+            retry_delay_seconds=0.01,
+        )
+
+        async def always_fail(*_args: object, **_kwargs: object) -> int:
+            raise Web3Exception("Always fails")
+
+        client._w3.eth.get_transaction_count = always_fail
+
+        with pytest.raises(RPCError):
+            await client.get_transaction_count(VALID_ADDRESS)
+
+
+class TestPolygonClientRateLimiting:
+    """Tests for rate limiting."""
+
+    @pytest.mark.asyncio
+    async def test_rate_limiting_enforced(self) -> None:
+        """Test that rate limiting delays requests."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+
+        client = PolygonClient(
+            "https://polygon-rpc.com",
+            redis=redis,
+            max_requests_per_second=5.0,
+        )
+
+        # Deplete rate limit
+        client._rate_limiter.tokens = 0
+
+        with patch.object(client._w3.eth, "get_transaction_count", new_callable=AsyncMock) as mock:
+            mock.return_value = 42
+
+            start = asyncio.get_event_loop().time()
+            await client.get_transaction_count(VALID_ADDRESS)
+            elapsed = asyncio.get_event_loop().time() - start
+
+            # Should have waited for token refill
+            assert elapsed >= 0.1
+
+
+class TestPolygonClientTokenBalance:
+    """Tests for ERC20 token balance queries."""
+
+    @pytest.fixture
+    def mock_redis(self) -> AsyncMock:
+        """Create a mock Redis client."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+        return redis
+
+    @pytest.mark.asyncio
+    async def test_get_token_balance_cached(self, mock_redis: AsyncMock) -> None:
+        """Test getting token balance from cache."""
+        mock_redis.get = AsyncMock(return_value=b"1000000")
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        balance = await client.get_token_balance(VALID_ADDRESS, VALID_TOKEN)
+
+        assert balance == Decimal("1000000")
+
+    @pytest.mark.asyncio
+    async def test_get_token_balance_uncached(self, mock_redis: AsyncMock) -> None:
+        """Test getting token balance from blockchain."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        # Mock the contract call
+        mock_contract = MagicMock()
+        mock_contract.functions.balanceOf.return_value.call = AsyncMock(
+            return_value=5000000
+        )
+        client._w3.eth.contract = MagicMock(return_value=mock_contract)
+
+        balance = await client.get_token_balance(VALID_ADDRESS, VALID_TOKEN)
+
+        assert balance == Decimal("5000000")
+        mock_redis.set.assert_called_once()
+
+
+class TestPolygonClientBlock:
+    """Tests for block queries."""
+
+    @pytest.fixture
+    def mock_redis(self) -> AsyncMock:
+        """Create a mock Redis client."""
+        redis = AsyncMock()
+        redis.get = AsyncMock(return_value=None)
+        redis.set = AsyncMock()
+        return redis
+
+    @pytest.mark.asyncio
+    async def test_get_block_cached(self, mock_redis: AsyncMock) -> None:
+        """Test getting block from cache."""
+        mock_redis.get = AsyncMock(return_value=b'{"timestamp": 1704369600}')
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        block = await client.get_block(50000000)
+
+        assert block["timestamp"] == 1704369600
+
+    @pytest.mark.asyncio
+    async def test_get_block_uncached(self, mock_redis: AsyncMock) -> None:
+        """Test getting block from blockchain."""
+        client = PolygonClient("https://polygon-rpc.com", redis=mock_redis)
+
+        with patch.object(client, "_execute_with_retry", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = {"timestamp": 1704369600, "number": 50000000}
+
+            block = await client.get_block(50000000)
+
+            assert block["timestamp"] == 1704369600
+            # Block cache uses 1 hour TTL
+            mock_redis.set.assert_called_once()
+            call_args = mock_redis.set.call_args
+            assert call_args[1]["ex"] == 3600

tests/profiler/test_models.py

@@ -0,0 +1,427 @@
+"""Tests for the profiler data models."""
+
+from datetime import UTC, datetime, timedelta
+from decimal import Decimal
+
+import pytest
+
+from polymarket_insider_tracker.profiler.models import Transaction, WalletInfo, WalletProfile
+
+
+class TestTransaction:
+    """Tests for the Transaction dataclass."""
+
+    @pytest.fixture
+    def sample_transaction(self) -> Transaction:
+        """Create a sample transaction."""
+        return Transaction(
+            hash="0xabc123",
+            block_number=50000000,
+            timestamp=datetime(2026, 1, 4, 12, 0, 0, tzinfo=UTC),
+            from_address="0xsender",
+            to_address="0xreceiver",
+            value=Decimal("1000000000000000000"),  # 1 MATIC
+            gas_used=21000,
+            gas_price=Decimal("50000000000"),  # 50 Gwei
+        )
+
+    def test_transaction_creation(self, sample_transaction: Transaction) -> None:
+        """Test creating a transaction."""
+        assert sample_transaction.hash == "0xabc123"
+        assert sample_transaction.block_number == 50000000
+        assert sample_transaction.from_address == "0xsender"
+        assert sample_transaction.to_address == "0xreceiver"
+
+    def test_value_matic(self, sample_transaction: Transaction) -> None:
+        """Test value in MATIC conversion."""
+        assert sample_transaction.value_matic == Decimal("1")
+
+    def test_value_matic_fractional(self) -> None:
+        """Test fractional MATIC value."""
+        tx = Transaction(
+            hash="0x123",
+            block_number=1,
+            timestamp=datetime.now(UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("500000000000000000"),  # 0.5 MATIC
+            gas_used=21000,
+            gas_price=Decimal("50000000000"),
+        )
+
+        assert tx.value_matic == Decimal("0.5")
+
+    def test_gas_cost_wei(self, sample_transaction: Transaction) -> None:
+        """Test gas cost in Wei."""
+        expected = 21000 * 50000000000
+        assert sample_transaction.gas_cost_wei == Decimal(expected)
+
+    def test_gas_cost_matic(self, sample_transaction: Transaction) -> None:
+        """Test gas cost in MATIC."""
+        # 21000 * 50 Gwei = 1050000 Gwei = 0.00105 MATIC
+        expected = Decimal("21000") * Decimal("50000000000") / Decimal("1000000000000000000")
+        assert sample_transaction.gas_cost_matic == expected
+
+    def test_transaction_frozen(self, sample_transaction: Transaction) -> None:
+        """Test that transaction is immutable."""
+        with pytest.raises(AttributeError):
+            sample_transaction.hash = "0xnew"  # type: ignore[misc]
+
+    def test_transaction_no_recipient(self) -> None:
+        """Test transaction with no recipient (contract creation)."""
+        tx = Transaction(
+            hash="0x123",
+            block_number=1,
+            timestamp=datetime.now(UTC),
+            from_address="0x1",
+            to_address=None,
+            value=Decimal("0"),
+            gas_used=100000,
+            gas_price=Decimal("50000000000"),
+        )
+
+        assert tx.to_address is None
+
+
+class TestWalletInfo:
+    """Tests for the WalletInfo dataclass."""
+
+    @pytest.fixture
+    def sample_wallet(self) -> WalletInfo:
+        """Create a sample wallet info."""
+        return WalletInfo(
+            address="0xwallet123",
+            transaction_count=100,
+            balance_wei=Decimal("5000000000000000000"),  # 5 MATIC
+            first_transaction=None,
+        )
+
+    @pytest.fixture
+    def wallet_with_transaction(self) -> WalletInfo:
+        """Create a wallet with first transaction."""
+        first_tx = Transaction(
+            hash="0xfirst",
+            block_number=1000000,
+            timestamp=datetime.now(UTC) - timedelta(days=365),
+            from_address="0xfaucet",
+            to_address="0xwallet123",
+            value=Decimal("1000000000000000000"),
+            gas_used=21000,
+            gas_price=Decimal("50000000000"),
+        )
+        return WalletInfo(
+            address="0xwallet123",
+            transaction_count=100,
+            balance_wei=Decimal("5000000000000000000"),
+            first_transaction=first_tx,
+        )
+
+    def test_wallet_creation(self, sample_wallet: WalletInfo) -> None:
+        """Test creating wallet info."""
+        assert sample_wallet.address == "0xwallet123"
+        assert sample_wallet.transaction_count == 100
+        assert sample_wallet.balance_wei == Decimal("5000000000000000000")
+
+    def test_balance_matic(self, sample_wallet: WalletInfo) -> None:
+        """Test balance in MATIC conversion."""
+        assert sample_wallet.balance_matic == Decimal("5")
+
+    def test_is_fresh_false(self, sample_wallet: WalletInfo) -> None:
+        """Test that wallet with many transactions is not fresh."""
+        assert sample_wallet.is_fresh is False
+
+    def test_is_fresh_true(self) -> None:
+        """Test that wallet with few transactions is fresh."""
+        wallet = WalletInfo(
+            address="0xnewwallet",
+            transaction_count=5,
+            balance_wei=Decimal("1000000000000000000"),
+        )
+
+        assert wallet.is_fresh is True
+
+    def test_is_fresh_boundary(self) -> None:
+        """Test fresh wallet boundary (10 transactions)."""
+        wallet_9 = WalletInfo(
+            address="0x1",
+            transaction_count=9,
+            balance_wei=Decimal("0"),
+        )
+        wallet_10 = WalletInfo(
+            address="0x2",
+            transaction_count=10,
+            balance_wei=Decimal("0"),
+        )
+
+        assert wallet_9.is_fresh is True
+        assert wallet_10.is_fresh is False
+
+    def test_wallet_age_days_no_transaction(self, sample_wallet: WalletInfo) -> None:
+        """Test wallet age when no first transaction."""
+        assert sample_wallet.wallet_age_days is None
+
+    def test_wallet_age_days_with_transaction(
+        self, wallet_with_transaction: WalletInfo
+    ) -> None:
+        """Test wallet age calculation."""
+        age = wallet_with_transaction.wallet_age_days
+
+        assert age is not None
+        # Should be approximately 365 days
+        assert 364 < age < 366
+
+    def test_wallet_frozen(self, sample_wallet: WalletInfo) -> None:
+        """Test that wallet info is immutable."""
+        with pytest.raises(AttributeError):
+            sample_wallet.address = "0xnew"  # type: ignore[misc]
+
+    def test_wallet_zero_balance(self) -> None:
+        """Test wallet with zero balance."""
+        wallet = WalletInfo(
+            address="0xempty",
+            transaction_count=0,
+            balance_wei=Decimal("0"),
+        )
+
+        assert wallet.balance_matic == Decimal("0")
+        assert wallet.is_fresh is True
+
+    def test_wallet_very_small_balance(self) -> None:
+        """Test wallet with very small balance."""
+        wallet = WalletInfo(
+            address="0xdust",
+            transaction_count=1,
+            balance_wei=Decimal("1"),  # 1 Wei
+        )
+
+        # Should be a very small fraction
+        expected = Decimal("1") / Decimal("1000000000000000000")
+        assert wallet.balance_matic == expected
+
+
+class TestTransactionEquality:
+    """Tests for transaction equality and hashing."""
+
+    def test_equal_transactions(self) -> None:
+        """Test that identical transactions are equal."""
+        tx1 = Transaction(
+            hash="0xabc",
+            block_number=1,
+            timestamp=datetime(2026, 1, 1, tzinfo=UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("1000"),
+            gas_used=21000,
+            gas_price=Decimal("50"),
+        )
+        tx2 = Transaction(
+            hash="0xabc",
+            block_number=1,
+            timestamp=datetime(2026, 1, 1, tzinfo=UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("1000"),
+            gas_used=21000,
+            gas_price=Decimal("50"),
+        )
+
+        assert tx1 == tx2
+
+    def test_different_transactions(self) -> None:
+        """Test that different transactions are not equal."""
+        tx1 = Transaction(
+            hash="0xabc",
+            block_number=1,
+            timestamp=datetime(2026, 1, 1, tzinfo=UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("1000"),
+            gas_used=21000,
+            gas_price=Decimal("50"),
+        )
+        tx2 = Transaction(
+            hash="0xdef",  # Different hash
+            block_number=1,
+            timestamp=datetime(2026, 1, 1, tzinfo=UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("1000"),
+            gas_used=21000,
+            gas_price=Decimal("50"),
+        )
+
+        assert tx1 != tx2
+
+    def test_transaction_hashable(self) -> None:
+        """Test that transactions can be used in sets."""
+        tx = Transaction(
+            hash="0xabc",
+            block_number=1,
+            timestamp=datetime(2026, 1, 1, tzinfo=UTC),
+            from_address="0x1",
+            to_address="0x2",
+            value=Decimal("1000"),
+            gas_used=21000,
+            gas_price=Decimal("50"),
+        )
+
+        tx_set = {tx}
+        assert tx in tx_set
+
+
+class TestWalletProfile:
+    """Tests for the WalletProfile dataclass."""
+
+    @pytest.fixture
+    def fresh_profile(self) -> WalletProfile:
+        """Create a fresh wallet profile."""
+        return WalletProfile(
+            address="0xfresh",
+            nonce=2,
+            first_seen=datetime.now(UTC) - timedelta(hours=6),
+            age_hours=6.0,
+            is_fresh=True,
+            total_tx_count=2,
+            matic_balance=Decimal("1000000000000000000"),  # 1 MATIC
+            usdc_balance=Decimal("1000000"),  # 1 USDC
+            fresh_threshold=5,
+        )
+
+    @pytest.fixture
+    def old_profile(self) -> WalletProfile:
+        """Create an old wallet profile."""
+        return WalletProfile(
+            address="0xold",
+            nonce=500,
+            first_seen=datetime.now(UTC) - timedelta(days=365),
+            age_hours=365 * 24,
+            is_fresh=False,
+            total_tx_count=500,
+            matic_balance=Decimal("100000000000000000000"),  # 100 MATIC
+            usdc_balance=Decimal("10000000000"),  # 10000 USDC
+            fresh_threshold=5,
+        )
+
+    def test_profile_creation(self, fresh_profile: WalletProfile) -> None:
+        """Test creating a wallet profile."""
+        assert fresh_profile.address == "0xfresh"
+        assert fresh_profile.nonce == 2
+        assert fresh_profile.is_fresh is True
+        assert fresh_profile.age_hours == 6.0
+
+    def test_age_days(self, fresh_profile: WalletProfile) -> None:
+        """Test age_days property."""
+        assert fresh_profile.age_days == 0.25  # 6 hours = 0.25 days
+
+    def test_age_days_none(self) -> None:
+        """Test age_days when age_hours is None."""
+        profile = WalletProfile(
+            address="0xnew",
+            nonce=0,
+            first_seen=None,
+            age_hours=None,
+            is_fresh=True,
+            total_tx_count=0,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+        )
+        assert profile.age_days is None
+
+    def test_matic_balance_formatted(self, fresh_profile: WalletProfile) -> None:
+        """Test MATIC balance formatting."""
+        assert fresh_profile.matic_balance_formatted == Decimal("1")
+
+    def test_usdc_balance_formatted(self, fresh_profile: WalletProfile) -> None:
+        """Test USDC balance formatting."""
+        assert fresh_profile.usdc_balance_formatted == Decimal("1")
+
+    def test_is_brand_new(self) -> None:
+        """Test is_brand_new property."""
+        brand_new = WalletProfile(
+            address="0xnew",
+            nonce=0,
+            first_seen=None,
+            age_hours=None,
+            is_fresh=True,
+            total_tx_count=0,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+        )
+        assert brand_new.is_brand_new is True
+
+        not_brand_new = WalletProfile(
+            address="0xold",
+            nonce=1,
+            first_seen=datetime.now(UTC),
+            age_hours=1.0,
+            is_fresh=True,
+            total_tx_count=1,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+        )
+        assert not_brand_new.is_brand_new is False
+
+    def test_freshness_score_brand_new(self) -> None:
+        """Test freshness score for brand new wallet."""
+        profile = WalletProfile(
+            address="0xnew",
+            nonce=0,
+            first_seen=None,
+            age_hours=None,
+            is_fresh=True,
+            total_tx_count=0,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+            fresh_threshold=5,
+        )
+        # nonce_score = 1.0 (0/5 = 0, 1-0 = 1)
+        # age_score = 1.0 (None = assumed new)
+        # score = 0.6 * 1.0 + 0.4 * 1.0 = 1.0
+        assert profile.freshness_score == 1.0
+
+    def test_freshness_score_old_wallet(self, old_profile: WalletProfile) -> None:
+        """Test freshness score for old wallet."""
+        # nonce_score = max(0, 1 - 500/5) = 0
+        # age_score = max(0, 1 - 8760/48) = 0
+        # score = 0
+        assert old_profile.freshness_score == 0.0
+
+    def test_freshness_score_moderate(self) -> None:
+        """Test freshness score for moderately fresh wallet."""
+        profile = WalletProfile(
+            address="0xmoderate",
+            nonce=2,
+            first_seen=datetime.now(UTC) - timedelta(hours=24),
+            age_hours=24.0,
+            is_fresh=True,
+            total_tx_count=2,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+            fresh_threshold=5,
+        )
+        # nonce_score = 1 - 2/5 = 0.6
+        # age_score = 1 - 24/48 = 0.5
+        # score = 0.6 * 0.6 + 0.4 * 0.5 = 0.36 + 0.2 = 0.56
+        assert profile.freshness_score == pytest.approx(0.56)
+
+    def test_profile_frozen(self, fresh_profile: WalletProfile) -> None:
+        """Test that wallet profile is immutable."""
+        with pytest.raises(AttributeError):
+            fresh_profile.nonce = 100  # type: ignore[misc]
+
+    def test_analyzed_at_default(self) -> None:
+        """Test that analyzed_at has a default."""
+        before = datetime.now(UTC)
+        profile = WalletProfile(
+            address="0x1",
+            nonce=0,
+            first_seen=None,
+            age_hours=None,
+            is_fresh=True,
+            total_tx_count=0,
+            matic_balance=Decimal("0"),
+            usdc_balance=Decimal("0"),
+        )
+        after = datetime.now(UTC)
+
+        assert before <= profile.analyzed_at <= after

tests/storage/__init__.py

@@ -0,0 +1 @@
+"""Tests for storage module."""

tests/test_setup.py

@@ -0,0 +1,24 @@
+"""Test that the project setup is working correctly."""
+
+import polymarket_insider_tracker
+
+
+def test_version() -> None:
+    """Test that version is defined."""
+    assert polymarket_insider_tracker.__version__ == "0.1.0"
+
+
+def test_import_modules() -> None:
+    """Test that all submodules can be imported."""
+    from polymarket_insider_tracker import ingestor
+    from polymarket_insider_tracker import profiler
+    from polymarket_insider_tracker import detector
+    from polymarket_insider_tracker import alerter
+    from polymarket_insider_tracker import storage
+
+    # Just verify imports work
+    assert ingestor is not None
+    assert profiler is not None
+    assert detector is not None
+    assert alerter is not None
+    assert storage is not None