feat: add market metadata synchronizer with Redis caching (#4)
Implement MarketMetadataSync class for background synchronization of market metadata with Redis-based caching. Key features: - MarketMetadata dataclass with derived category field - Automatic category derivation from market title (politics, crypto, sports, entertainment, finance, tech, science, other) - Background sync loop with configurable interval (default: 5 min) - Redis caching with TTL-based expiration (default: 10 min) - Cache-first lookups via get_market() method - State management with callbacks for monitoring - Comprehensive test suite (29 tests) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.5
parent
20d272b6c3
commit
8db2516001
@@ -5,16 +5,24 @@ from polymarket_insider_tracker.ingestor.clob_client import (
|
|||||||
ClobClientError,
|
ClobClientError,
|
||||||
RetryError,
|
RetryError,
|
||||||
)
|
)
|
||||||
|
from polymarket_insider_tracker.ingestor.metadata_sync import (
|
||||||
|
MarketMetadataSync,
|
||||||
|
MetadataSyncError,
|
||||||
|
SyncState,
|
||||||
|
SyncStats,
|
||||||
|
)
|
||||||
from polymarket_insider_tracker.ingestor.models import (
|
from polymarket_insider_tracker.ingestor.models import (
|
||||||
Market,
|
Market,
|
||||||
|
MarketMetadata,
|
||||||
Orderbook,
|
Orderbook,
|
||||||
OrderbookLevel,
|
OrderbookLevel,
|
||||||
Token,
|
Token,
|
||||||
TradeEvent,
|
TradeEvent,
|
||||||
|
derive_category,
|
||||||
)
|
)
|
||||||
from polymarket_insider_tracker.ingestor.websocket import (
|
from polymarket_insider_tracker.ingestor.websocket import (
|
||||||
ConnectionState,
|
ConnectionState,
|
||||||
StreamStats,
|
StreamStats as WebSocketStreamStats,
|
||||||
TradeStreamError,
|
TradeStreamError,
|
||||||
TradeStreamHandler,
|
TradeStreamHandler,
|
||||||
)
|
)
|
||||||
@@ -24,15 +32,22 @@ __all__ = [
|
|||||||
"ClobClient",
|
"ClobClient",
|
||||||
"ClobClientError",
|
"ClobClientError",
|
||||||
"RetryError",
|
"RetryError",
|
||||||
|
# Metadata Sync
|
||||||
|
"MarketMetadataSync",
|
||||||
|
"MetadataSyncError",
|
||||||
|
"SyncState",
|
||||||
|
"SyncStats",
|
||||||
# Models
|
# Models
|
||||||
"Market",
|
"Market",
|
||||||
|
"MarketMetadata",
|
||||||
"Orderbook",
|
"Orderbook",
|
||||||
"OrderbookLevel",
|
"OrderbookLevel",
|
||||||
"Token",
|
"Token",
|
||||||
"TradeEvent",
|
"TradeEvent",
|
||||||
|
"derive_category",
|
||||||
# WebSocket
|
# WebSocket
|
||||||
"ConnectionState",
|
"ConnectionState",
|
||||||
"StreamStats",
|
"WebSocketStreamStats",
|
||||||
"TradeStreamError",
|
"TradeStreamError",
|
||||||
"TradeStreamHandler",
|
"TradeStreamHandler",
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -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()
|
||||||
@@ -1,7 +1,7 @@
|
|||||||
"""Data models for the ingestor module."""
|
"""Data models for the ingestor module."""
|
||||||
|
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from datetime import datetime, timezone
|
from datetime import UTC, datetime
|
||||||
from decimal import Decimal
|
from decimal import Decimal
|
||||||
from typing import Any, Literal
|
from typing import Any, Literal
|
||||||
|
|
||||||
@@ -186,9 +186,9 @@ class TradeEvent:
|
|||||||
# Parse timestamp - it's a Unix timestamp in seconds
|
# Parse timestamp - it's a Unix timestamp in seconds
|
||||||
raw_timestamp = data.get("timestamp", 0)
|
raw_timestamp = data.get("timestamp", 0)
|
||||||
if isinstance(raw_timestamp, int):
|
if isinstance(raw_timestamp, int):
|
||||||
timestamp = datetime.fromtimestamp(raw_timestamp, tz=timezone.utc)
|
timestamp = datetime.fromtimestamp(raw_timestamp, tz=UTC)
|
||||||
else:
|
else:
|
||||||
timestamp = datetime.now(timezone.utc)
|
timestamp = datetime.now(UTC)
|
||||||
|
|
||||||
# Parse side - normalize to uppercase
|
# Parse side - normalize to uppercase
|
||||||
side_raw = str(data.get("side", "BUY")).upper()
|
side_raw = str(data.get("side", "BUY")).upper()
|
||||||
@@ -226,3 +226,268 @@ class TradeEvent:
|
|||||||
def notional_value(self) -> Decimal:
|
def notional_value(self) -> Decimal:
|
||||||
"""Return the notional value of the trade (price * size)."""
|
"""Return the notional value of the trade (price * size)."""
|
||||||
return self.price * self.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,
|
||||||
|
)
|
||||||
|
|||||||
@@ -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
|
||||||
Reference in New Issue
Block a user