diff --git a/src/polymarket_insider_tracker/ingestor/__init__.py b/src/polymarket_insider_tracker/ingestor/__init__.py index c53bc12..8edff2c 100644 --- a/src/polymarket_insider_tracker/ingestor/__init__.py +++ b/src/polymarket_insider_tracker/ingestor/__init__.py @@ -5,16 +5,24 @@ from polymarket_insider_tracker.ingestor.clob_client import ( ClobClientError, RetryError, ) +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.websocket import ( ConnectionState, - StreamStats, + StreamStats as WebSocketStreamStats, TradeStreamError, TradeStreamHandler, ) @@ -24,15 +32,22 @@ __all__ = [ "ClobClient", "ClobClientError", "RetryError", + # Metadata Sync + "MarketMetadataSync", + "MetadataSyncError", + "SyncState", + "SyncStats", # Models "Market", + "MarketMetadata", "Orderbook", "OrderbookLevel", "Token", "TradeEvent", + "derive_category", # WebSocket "ConnectionState", - "StreamStats", + "WebSocketStreamStats", "TradeStreamError", "TradeStreamHandler", ] diff --git a/src/polymarket_insider_tracker/ingestor/metadata_sync.py b/src/polymarket_insider_tracker/ingestor/metadata_sync.py new file mode 100644 index 0000000..8aa9293 --- /dev/null +++ b/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() diff --git a/src/polymarket_insider_tracker/ingestor/models.py b/src/polymarket_insider_tracker/ingestor/models.py index 9024bf8..f605ea4 100644 --- a/src/polymarket_insider_tracker/ingestor/models.py +++ b/src/polymarket_insider_tracker/ingestor/models.py @@ -1,7 +1,7 @@ """Data models for the ingestor module.""" from dataclasses import dataclass, field -from datetime import datetime, timezone +from datetime import UTC, datetime from decimal import Decimal from typing import Any, Literal @@ -186,9 +186,9 @@ class TradeEvent: # 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=timezone.utc) + timestamp = datetime.fromtimestamp(raw_timestamp, tz=UTC) else: - timestamp = datetime.now(timezone.utc) + timestamp = datetime.now(UTC) # Parse side - normalize to uppercase side_raw = str(data.get("side", "BUY")).upper() @@ -226,3 +226,268 @@ class TradeEvent: 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, + ) diff --git a/tests/ingestor/test_metadata_sync.py b/tests/ingestor/test_metadata_sync.py new file mode 100644 index 0000000..45f5ee9 --- /dev/null +++ b/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