From e003e4ba4701ef3fc409d928ca472749ab9d6432 Mon Sep 17 00:00:00 2001 From: Patrick Selamy Date: Sun, 14 Jun 2026 15:17:46 -0400 Subject: [PATCH] feat(ingestor): enrich market metadata with gamma-api volume/liquidity (#107) The CLOB API does not expose 24h volume or order-book liquidity, so the size_anomaly detector currently has no real ratio to compare a trade against and falls back to the niche-base 0.2 confidence floor. That makes the volume_impact / book_impact thresholds essentially dead code. This change adds a small client for the public gamma-api markets endpoint and merges its volume24hr / liquidityNum snapshot into MarketMetadata during the existing periodic sync. The detector can now compute real volume and book impact ratios. Notes on the gamma client: - gamma-api enforces a server-side max of 100 markets per page and caps `offset` around 10000. The client paginates with bounded concurrency, sorted by `volume24hr desc`, so the most-traded markets (which is where size anomalies actually matter) are always covered. Markets beyond that window have negligible recent volume and the niche path handles them fine without a ratio. - Failures are swallowed: a degraded gamma endpoint must not stop CLOB metadata from being cached, since size_anomaly + the niche path remain functional with `daily_volume=None`. MarketMetadata gains three optional Decimal fields (daily_volume, weekly_volume, liquidity); to_dict / from_dict round-trip is preserved and older cache entries without these keys deserialize cleanly. Tests: 12 new tests for GammaClient (parsing, single-page, short-page stop, offset-cap clean stop, retry, malformed responses); existing metadata_sync tests updated to inject a mocked GammaClient so they don't hit the real network. Co-authored-by: schrodinger01 --- .../ingestor/gamma_client.py | 205 ++++++++++++++ .../ingestor/metadata_sync.py | 43 ++- .../ingestor/models.py | 21 ++ tests/ingestor/test_gamma_client.py | 249 ++++++++++++++++++ tests/ingestor/test_metadata_sync.py | 93 +++++-- 5 files changed, 580 insertions(+), 31 deletions(-) create mode 100644 src/polymarket_insider_tracker/ingestor/gamma_client.py create mode 100644 tests/ingestor/test_gamma_client.py diff --git a/src/polymarket_insider_tracker/ingestor/gamma_client.py b/src/polymarket_insider_tracker/ingestor/gamma_client.py new file mode 100644 index 0000000..47ce7a5 --- /dev/null +++ b/src/polymarket_insider_tracker/ingestor/gamma_client.py @@ -0,0 +1,205 @@ +"""Gamma API client for Polymarket market volume / liquidity data. + +The CLOB API does not expose 24h volume or liquidity. The public Gamma API +(https://gamma-api.polymarket.com) does, with no auth required. This module +fetches the volume/liquidity snapshot keyed by condition_id so the +size_anomaly detector can do real ratio math instead of falling back to +the niche-base 0.2 confidence floor. +""" + +from __future__ import annotations + +import asyncio +import logging +from dataclasses import dataclass +from decimal import Decimal, InvalidOperation + +import httpx + +logger = logging.getLogger(__name__) + +DEFAULT_HOST = "https://gamma-api.polymarket.com" +DEFAULT_TIMEOUT_SECONDS = 15.0 +# Gamma /markets enforces a server-side max of 100 per page even when a +# higher `limit` is sent. Using 100 lines our page size up with the actual +# response so pagination doesn't bail out after the first page. +DEFAULT_PAGE_LIMIT = 100 +# Gamma also caps `offset` around 10000 for this collection. Combined with +# the 100/page limit that gives ~10k markets max, sequential — way too slow +# at default sync interval. We sort by 24h volume desc and only walk the +# top N pages, since markets with zero recent volume don't need a real +# ratio anyway (the niche path handles them). +DEFAULT_MAX_PAGES = 50 # 50 * 100 = 5000 most-traded markets per sync +DEFAULT_PAGE_CONCURRENCY = 5 +DEFAULT_MAX_RETRIES = 3 +DEFAULT_RETRY_BASE_DELAY_SECONDS = 1.0 + + +@dataclass(frozen=True) +class GammaMarketStats: + """Volume / liquidity snapshot for a single market from gamma-api.""" + + condition_id: str + daily_volume: Decimal | None + weekly_volume: Decimal | None + monthly_volume: Decimal | None + total_volume: Decimal | None + liquidity: Decimal | None + + +def _to_decimal(value: object) -> Decimal | None: + if value is None or value == "": + return None + try: + return Decimal(str(value)) + except (InvalidOperation, ValueError, TypeError): + return None + + +def _parse_market(raw: dict[str, object]) -> GammaMarketStats | None: + cid = raw.get("conditionId") + if not cid or not isinstance(cid, str): + return None + return GammaMarketStats( + condition_id=cid, + daily_volume=_to_decimal(raw.get("volume24hr")), + weekly_volume=_to_decimal(raw.get("volume1wk")), + monthly_volume=_to_decimal(raw.get("volume1mo")), + total_volume=_to_decimal(raw.get("volumeNum") or raw.get("volume")), + liquidity=_to_decimal(raw.get("liquidityNum") or raw.get("liquidity")), + ) + + +class GammaClientError(Exception): + """Raised when gamma-api returns an unrecoverable error.""" + + +class GammaClient: + """Async client for the public gamma-api markets endpoint. + + Provides batched, paginated reads of every active market with their + 24h/weekly/monthly volume and current liquidity. Designed to be called + from MarketMetadataSync once per sync interval; results are merged into + Redis-cached MarketMetadata objects. + """ + + def __init__( + self, + *, + host: str = DEFAULT_HOST, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + page_limit: int = DEFAULT_PAGE_LIMIT, + max_pages: int = DEFAULT_MAX_PAGES, + page_concurrency: int = DEFAULT_PAGE_CONCURRENCY, + max_retries: int = DEFAULT_MAX_RETRIES, + retry_base_delay_seconds: float = DEFAULT_RETRY_BASE_DELAY_SECONDS, + ) -> None: + self._host = host.rstrip("/") + self._timeout = timeout_seconds + self._page_limit = page_limit + self._max_pages = max_pages + self._page_concurrency = page_concurrency + self._max_retries = max_retries + self._retry_base = retry_base_delay_seconds + + async def _get_with_retry( + self, + client: httpx.AsyncClient, + path: str, + params: dict[str, object], + ) -> list[dict[str, object]]: + last_exc: Exception | None = None + delay = self._retry_base + for attempt in range(self._max_retries): + try: + resp = await client.get(path, params=params) + resp.raise_for_status() + payload = resp.json() + if not isinstance(payload, list): + raise GammaClientError( + f"Unexpected gamma response shape for {path}: {type(payload).__name__}" + ) + return payload + except (httpx.HTTPError, ValueError) as exc: + last_exc = exc + logger.warning( + "gamma %s attempt %d/%d failed: %s", + path, + attempt + 1, + self._max_retries, + exc, + ) + if attempt < self._max_retries - 1: + await asyncio.sleep(delay) + delay *= 2 + raise GammaClientError( + f"gamma {path} failed after {self._max_retries} attempts: {last_exc}" + ) + + async def get_active_market_stats(self) -> dict[str, GammaMarketStats]: + """Fetch volume/liquidity for the most-traded active markets. + + Walks up to `max_pages` pages of `page_limit` markets each, sorted + by 24h volume descending, with bounded concurrency. Markets beyond + that window have effectively zero recent volume — the size_anomaly + niche path handles them without needing a ratio. + + Returns: + Mapping condition_id -> GammaMarketStats. + """ + results: dict[str, GammaMarketStats] = {} + sem = asyncio.Semaphore(self._page_concurrency) + stop = asyncio.Event() + + async with httpx.AsyncClient( + base_url=self._host, + timeout=self._timeout, + headers={"User-Agent": "polymarket-insider-tracker/0.1"}, + ) as client: + + async def fetch_page(page_index: int) -> list[dict[str, object]]: + if stop.is_set(): + return [] + params = { + "limit": self._page_limit, + "offset": page_index * self._page_limit, + "active": "true", + "closed": "false", + "order": "volume24hr", + "ascending": "false", + } + async with sem: + if stop.is_set(): + return [] + try: + return await self._get_with_retry(client, "/markets", params) + except GammaClientError as exc: + # Gamma rejects offsets past its hard cap with a + # validation error; treat that as a clean stop. + logger.debug("gamma stop at page %d: %s", page_index, exc) + stop.set() + return [] + + tasks = [asyncio.create_task(fetch_page(i)) for i in range(self._max_pages)] + pages = await asyncio.gather(*tasks) + + empty_streak = 0 + for page in pages: + if not page: + empty_streak += 1 + continue + empty_streak = 0 + for raw in page: + if not isinstance(raw, dict): + continue + parsed = _parse_market(raw) + if parsed is not None: + results[parsed.condition_id] = parsed + if len(page) < self._page_limit: + # short page — we walked past the end of the active set + empty_streak += 1 + if empty_streak >= 2: + break + + logger.info("gamma sync: fetched stats for %d active markets", len(results)) + return results diff --git a/src/polymarket_insider_tracker/ingestor/metadata_sync.py b/src/polymarket_insider_tracker/ingestor/metadata_sync.py index b85fae2..88c0260 100644 --- a/src/polymarket_insider_tracker/ingestor/metadata_sync.py +++ b/src/polymarket_insider_tracker/ingestor/metadata_sync.py @@ -9,13 +9,14 @@ import contextlib import json import logging from collections.abc import Callable -from dataclasses import dataclass +from dataclasses import dataclass, replace from datetime import UTC, datetime from enum import StrEnum from redis.asyncio import Redis from .clob_client import ClobClient +from .gamma_client import GammaClient, GammaClientError, GammaMarketStats from .models import MarketMetadata logger = logging.getLogger(__name__) @@ -91,6 +92,7 @@ class MarketMetadataSync: redis: Redis, clob_client: ClobClient, *, + gamma_client: GammaClient | None = None, sync_interval_seconds: int = DEFAULT_SYNC_INTERVAL_SECONDS, cache_ttl_seconds: int = DEFAULT_CACHE_TTL_SECONDS, key_prefix: str = DEFAULT_REDIS_KEY_PREFIX, @@ -102,6 +104,8 @@ class MarketMetadataSync: Args: redis: Redis async client for caching. clob_client: CLOB API client for fetching markets. + gamma_client: Optional gamma-api client for volume/liquidity + enrichment. Defaults to a fresh GammaClient() instance. 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. @@ -110,6 +114,7 @@ class MarketMetadataSync: """ self._redis = redis self._clob = clob_client + self._gamma = gamma_client or GammaClient() self._sync_interval = sync_interval_seconds self._cache_ttl = cache_ttl_seconds self._key_prefix = key_prefix @@ -216,6 +221,18 @@ class MarketMetadataSync: self._set_state(SyncState.ERROR) # Continue running - will retry on next interval + async def _fetch_gamma_stats(self) -> dict[str, GammaMarketStats]: + """Fetch volume/liquidity stats from gamma-api. + + Returns an empty dict on failure so a degraded gamma endpoint + does not stop CLOB metadata from being cached. + """ + try: + return await self._gamma.get_active_market_stats() + except (GammaClientError, Exception) as e: + logger.warning("gamma stats fetch failed (continuing without volume): %s", e) + return {} + async def _sync_all_markets(self) -> None: """Fetch all markets and cache them in Redis.""" self._set_state(SyncState.SYNCING) @@ -223,14 +240,27 @@ class MarketMetadataSync: 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) + # Fetch CLOB markets and gamma volume snapshot in parallel + markets, gamma_stats = await asyncio.gather( + asyncio.to_thread(self._clob.get_markets, True), + self._fetch_gamma_stats(), + ) - # Cache each market in Redis + # Cache each market in Redis, enriched with gamma volume/liquidity cached_count = 0 + enriched_count = 0 for market in markets: try: metadata = MarketMetadata.from_market(market) + stats = gamma_stats.get(metadata.condition_id) + if stats is not None: + metadata = replace( + metadata, + daily_volume=stats.daily_volume, + weekly_volume=stats.weekly_volume, + liquidity=stats.liquidity, + ) + enriched_count += 1 await self._cache_market(metadata) cached_count += 1 except Exception as e: @@ -246,7 +276,10 @@ class MarketMetadataSync: self._set_state(SyncState.IDLE) logger.info( - f"Synced {cached_count} markets in {self._stats.last_sync_duration_seconds:.2f}s" + "Synced %d markets (%d enriched with gamma volume) in %.2fs", + cached_count, + enriched_count, + self._stats.last_sync_duration_seconds, ) # Notify callback diff --git a/src/polymarket_insider_tracker/ingestor/models.py b/src/polymarket_insider_tracker/ingestor/models.py index 618ccba..7108d54 100644 --- a/src/polymarket_insider_tracker/ingestor/models.py +++ b/src/polymarket_insider_tracker/ingestor/models.py @@ -399,6 +399,12 @@ class MarketMetadata: # Derived metadata category: str = "other" + # Liquidity/volume snapshot (from gamma-api). All optional — older + # cache entries and CLOB-only sync results may not have these. + daily_volume: Decimal | None = None + weekly_volume: Decimal | None = None + liquidity: Decimal | None = None + # Cache metadata last_updated: datetime = field(default_factory=lambda: datetime.now(UTC)) @@ -446,6 +452,9 @@ class MarketMetadata: "active": self.active, "closed": self.closed, "category": self.category, + "daily_volume": str(self.daily_volume) if self.daily_volume is not None else None, + "weekly_volume": str(self.weekly_volume) if self.weekly_volume is not None else None, + "liquidity": str(self.liquidity) if self.liquidity is not None else None, "last_updated": self.last_updated.isoformat(), } @@ -477,6 +486,15 @@ class MarketMetadata: else: last_updated = datetime.now(UTC) + def _opt_dec(key: str) -> Decimal | None: + raw = data.get(key) + if raw is None or raw == "": + return None + try: + return Decimal(str(raw)) + except (ValueError, ArithmeticError): + return None + return cls( condition_id=str(data["condition_id"]), question=str(data.get("question", "")), @@ -486,5 +504,8 @@ class MarketMetadata: active=bool(data.get("active", True)), closed=bool(data.get("closed", False)), category=str(data.get("category", "other")), + daily_volume=_opt_dec("daily_volume"), + weekly_volume=_opt_dec("weekly_volume"), + liquidity=_opt_dec("liquidity"), last_updated=last_updated, ) diff --git a/tests/ingestor/test_gamma_client.py b/tests/ingestor/test_gamma_client.py new file mode 100644 index 0000000..d536f38 --- /dev/null +++ b/tests/ingestor/test_gamma_client.py @@ -0,0 +1,249 @@ +"""Tests for the gamma-api client.""" + +from __future__ import annotations + +from decimal import Decimal + +import httpx +import pytest + +from polymarket_insider_tracker.ingestor import gamma_client as gamma_module +from polymarket_insider_tracker.ingestor.gamma_client import ( + GammaClient, + GammaClientError, + GammaMarketStats, + _parse_market, +) + + +class TestParseMarket: + def test_parses_full_payload(self) -> None: + raw = { + "conditionId": "0xabc", + "volume24hr": "12345.67", + "volume1wk": "100000", + "volume1mo": "500000", + "volumeNum": "999999.5", + "liquidityNum": "42000", + } + stats = _parse_market(raw) + assert stats is not None + assert stats.condition_id == "0xabc" + assert stats.daily_volume == Decimal("12345.67") + assert stats.weekly_volume == Decimal("100000") + assert stats.monthly_volume == Decimal("500000") + assert stats.total_volume == Decimal("999999.5") + assert stats.liquidity == Decimal("42000") + + def test_falls_back_to_alternative_keys(self) -> None: + raw = { + "conditionId": "0x1", + "volume24hr": "1", + "volume": "777", + "liquidity": "55", + } + stats = _parse_market(raw) + assert stats is not None + assert stats.total_volume == Decimal("777") + assert stats.liquidity == Decimal("55") + + def test_handles_missing_numeric_fields(self) -> None: + stats = _parse_market({"conditionId": "0x2"}) + assert stats is not None + assert stats.daily_volume is None + assert stats.liquidity is None + + def test_drops_garbage_decimals(self) -> None: + stats = _parse_market( + {"conditionId": "0x3", "volume24hr": "not-a-number", "liquidityNum": ""} + ) + assert stats is not None + assert stats.daily_volume is None + assert stats.liquidity is None + + def test_rejects_missing_condition_id(self) -> None: + assert _parse_market({"volume24hr": "1"}) is None + assert _parse_market({"conditionId": ""}) is None + assert _parse_market({"conditionId": 123}) is None # type: ignore[arg-type] + + +def _make_client( + _handler: httpx.MockTransport, + *, + page_limit: int = 100, + max_pages: int = 5, + page_concurrency: int = 5, + max_retries: int = 1, +) -> GammaClient: + """Build a GammaClient that constructs httpx.AsyncClient with the given transport. + + GammaClient creates its own AsyncClient inside `get_active_market_stats`, + so we monkeypatch the AsyncClient factory in the module to inject the mock + transport. + """ + return GammaClient( + page_limit=page_limit, + max_pages=max_pages, + page_concurrency=page_concurrency, + max_retries=max_retries, + retry_base_delay_seconds=0.0, + ) + + +@pytest.fixture +def patch_async_client(monkeypatch: pytest.MonkeyPatch): + """Replace the AsyncClient used by gamma_client with one bound to a MockTransport.""" + + def _apply(handler: httpx.MockTransport) -> None: + original = gamma_module.httpx.AsyncClient + + def factory(*args: object, **kwargs: object) -> httpx.AsyncClient: + kwargs["transport"] = handler # type: ignore[index] + return original(*args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(gamma_module.httpx, "AsyncClient", factory) + + return _apply + + +@pytest.mark.asyncio +async def test_get_active_market_stats_single_page(patch_async_client) -> None: + page_one = [ + {"conditionId": "0xa", "volume24hr": "100", "liquidityNum": "10"}, + {"conditionId": "0xb", "volume24hr": "200", "liquidityNum": "20"}, + ] + calls: list[dict[str, str]] = [] + + def handler(request: httpx.Request) -> httpx.Response: + calls.append(dict(request.url.params)) + offset = int(request.url.params.get("offset", "0")) + if offset == 0: + return httpx.Response(200, json=page_one) + return httpx.Response(200, json=[]) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=2, max_pages=3) + + result = await client.get_active_market_stats() + + assert set(result.keys()) == {"0xa", "0xb"} + assert isinstance(result["0xa"], GammaMarketStats) + assert result["0xa"].daily_volume == Decimal("100") + assert calls[0]["limit"] == "2" + assert calls[0]["order"] == "volume24hr" + assert calls[0]["ascending"] == "false" + + +@pytest.mark.asyncio +async def test_get_active_market_stats_short_page_stops(patch_async_client) -> None: + """A page shorter than page_limit signals end-of-data after a small empty streak.""" + page_zero = [{"conditionId": f"0x{i}", "volume24hr": str(i)} for i in range(5)] + page_one_short = [{"conditionId": "0xshort", "volume24hr": "1"}] + + def handler(request: httpx.Request) -> httpx.Response: + offset = int(request.url.params.get("offset", "0")) + if offset == 0: + return httpx.Response(200, json=page_zero) + if offset == 5: + return httpx.Response(200, json=page_one_short) + return httpx.Response(200, json=[]) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=5, max_pages=10) + + result = await client.get_active_market_stats() + + assert "0xshort" in result + assert len(result) == 6 + + +@pytest.mark.asyncio +async def test_get_active_market_stats_offset_cap_clean_stop( + patch_async_client, +) -> None: + """Gamma rejects offsets past its hard cap; that error is swallowed cleanly.""" + + def handler(request: httpx.Request) -> httpx.Response: + offset = int(request.url.params.get("offset", "0")) + if offset == 0: + return httpx.Response(200, json=[{"conditionId": "0xa", "volume24hr": "1"}]) + return httpx.Response(400, json={"error": "offset too large"}) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=1, max_pages=4, max_retries=1) + + result = await client.get_active_market_stats() + assert "0xa" in result + + +@pytest.mark.asyncio +async def test_get_with_retry_recovers_after_transient_error( + patch_async_client, +) -> None: + """Transient HTTP errors retry up to max_retries before giving up.""" + state = {"attempts": 0} + + def handler(request: httpx.Request) -> httpx.Response: + offset = int(request.url.params.get("offset", "0")) + if offset == 0: + state["attempts"] += 1 + if state["attempts"] < 2: + return httpx.Response(503, json={"error": "transient"}) + return httpx.Response(200, json=[{"conditionId": "0xrecover", "volume24hr": "1"}]) + return httpx.Response(200, json=[]) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=1, max_pages=2, max_retries=3) + + result = await client.get_active_market_stats() + assert "0xrecover" in result + assert state["attempts"] == 2 + + +@pytest.mark.asyncio +async def test_unexpected_response_shape_is_handled(patch_async_client) -> None: + """A non-list payload becomes a clean stop, not a crash.""" + + def handler(_request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"unexpected": "shape"}) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=1, max_pages=2, max_retries=1) + + result = await client.get_active_market_stats() + assert result == {} + + +@pytest.mark.asyncio +async def test_skips_non_dict_entries(patch_async_client) -> None: + """Defensive: server returning mixed-type list items shouldn't crash.""" + + def handler(request: httpx.Request) -> httpx.Response: + offset = int(request.url.params.get("offset", "0")) + if offset == 0: + return httpx.Response( + 200, + json=[ + {"conditionId": "0xa", "volume24hr": "1"}, + "garbage", + None, + 42, + ], + ) + return httpx.Response(200, json=[]) + + transport = httpx.MockTransport(handler) + patch_async_client(transport) + client = _make_client(transport, page_limit=4, max_pages=2) + + result = await client.get_active_market_stats() + assert list(result.keys()) == ["0xa"] + + +def test_gamma_client_error_inherits_exception() -> None: + assert issubclass(GammaClientError, Exception) diff --git a/tests/ingestor/test_metadata_sync.py b/tests/ingestor/test_metadata_sync.py index 45f5ee9..b5a133f 100644 --- a/tests/ingestor/test_metadata_sync.py +++ b/tests/ingestor/test_metadata_sync.py @@ -8,6 +8,7 @@ from unittest.mock import AsyncMock, MagicMock import pytest from polymarket_insider_tracker.ingestor.clob_client import ClobClient +from polymarket_insider_tracker.ingestor.gamma_client import GammaClient from polymarket_insider_tracker.ingestor.metadata_sync import ( DEFAULT_CACHE_TTL_SECONDS, DEFAULT_REDIS_KEY_PREFIX, @@ -72,6 +73,18 @@ def mock_clob(sample_market: Market) -> MagicMock: return clob +@pytest.fixture +def mock_gamma() -> MagicMock: + """Create a mock GammaClient that returns empty volume stats. + + Without this, MarketMetadataSync would instantiate a default + GammaClient and hit the real gamma-api over HTTP during unit tests. + """ + gamma = MagicMock(spec=GammaClient) + gamma.get_active_market_stats = AsyncMock(return_value={}) + return gamma + + class TestDeriveCategory: """Tests for the derive_category function.""" @@ -195,9 +208,9 @@ class TestSyncStats: class TestMarketMetadataSync: """Tests for the MarketMetadataSync class.""" - def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + def test_init(self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock) -> None: """Test initialization.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) assert sync.state == SyncState.STOPPED assert sync.stats.total_syncs == 0 @@ -205,11 +218,14 @@ class TestMarketMetadataSync: 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: + def test_init_custom_config( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test initialization with custom config.""" sync = MarketMetadataSync( redis=mock_redis, clob_client=mock_clob, + gamma_client=mock_gamma, sync_interval_seconds=60, cache_ttl_seconds=120, key_prefix="custom:", @@ -220,9 +236,11 @@ class TestMarketMetadataSync: assert sync._key_prefix == "custom:" @pytest.mark.asyncio - async def test_start_stop(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_start_stop( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test starting and stopping the sync service.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) # Start await sync.start() @@ -236,10 +254,10 @@ class TestMarketMetadataSync: @pytest.mark.asyncio async def test_start_performs_initial_sync( - self, mock_redis: AsyncMock, mock_clob: MagicMock + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock ) -> None: """Test that start performs an initial sync.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() @@ -252,10 +270,12 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_start_failure(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_start_failure( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test start failure handling.""" mock_clob.get_markets.side_effect = Exception("API error") - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) with pytest.raises(MetadataSyncError, match="initial sync failed"): await sync.start() @@ -268,6 +288,7 @@ class TestMarketMetadataSync: self, mock_redis: AsyncMock, mock_clob: MagicMock, + mock_gamma: MagicMock, sample_metadata: MarketMetadata, ) -> None: """Test get_market with cache hit.""" @@ -275,7 +296,7 @@ class TestMarketMetadataSync: 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) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() result = await sync.get_market("cond123") @@ -288,12 +309,14 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_get_market_cache_miss(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_get_market_cache_miss( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: 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) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() result = await sync.get_market("cond123") @@ -308,12 +331,14 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_get_market_not_found(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_get_market_not_found( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: 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) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() result = await sync.get_market("nonexistent") @@ -323,9 +348,11 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_invalidate_market(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_invalidate_market( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test cache invalidation.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() result = await sync.invalidate_market("cond123") @@ -336,9 +363,11 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_force_sync(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_force_sync( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test forced sync.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() # Initial sync @@ -353,7 +382,9 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_state_change_callback(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_state_change_callback( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test state change callback.""" states: list[SyncState] = [] @@ -363,6 +394,7 @@ class TestMarketMetadataSync: sync = MarketMetadataSync( redis=mock_redis, clob_client=mock_clob, + gamma_client=mock_gamma, on_state_change=on_state_change, ) @@ -377,7 +409,7 @@ class TestMarketMetadataSync: @pytest.mark.asyncio async def test_sync_complete_callback( - self, mock_redis: AsyncMock, mock_clob: MagicMock + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock ) -> None: """Test sync complete callback.""" sync_stats: list[SyncStats] = [] @@ -388,6 +420,7 @@ class TestMarketMetadataSync: sync = MarketMetadataSync( redis=mock_redis, clob_client=mock_clob, + gamma_client=mock_gamma, on_sync_complete=on_sync_complete, ) @@ -401,7 +434,11 @@ class TestMarketMetadataSync: @pytest.mark.asyncio async def test_get_markets_by_category( - self, mock_redis: AsyncMock, mock_clob: MagicMock, sample_metadata: MarketMetadata + self, + mock_redis: AsyncMock, + mock_clob: MagicMock, + mock_gamma: MagicMock, + sample_metadata: MarketMetadata, ) -> None: """Test getting markets by category.""" # Setup scan to return keys @@ -412,7 +449,7 @@ class TestMarketMetadataSync: 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) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) # Don't start to avoid initial sync complexity sync._state = SyncState.IDLE @@ -422,9 +459,11 @@ class TestMarketMetadataSync: assert results[0].category == "crypto" @pytest.mark.asyncio - async def test_cannot_start_twice(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_cannot_start_twice( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test that starting twice doesn't double-start.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.start() await sync.start() # Should be a no-op @@ -434,9 +473,11 @@ class TestMarketMetadataSync: await sync.stop() @pytest.mark.asyncio - async def test_stop_when_stopped(self, mock_redis: AsyncMock, mock_clob: MagicMock) -> None: + async def test_stop_when_stopped( + self, mock_redis: AsyncMock, mock_clob: MagicMock, mock_gamma: MagicMock + ) -> None: """Test stopping when already stopped.""" - sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob) + sync = MarketMetadataSync(redis=mock_redis, clob_client=mock_clob, gamma_client=mock_gamma) await sync.stop() # Should be a no-op