Compare commits

...

1 Commits

Author SHA1 Message Date
schrodinger01 2f89945ce1 feat(ingestor): enrich market metadata with gamma-api volume/liquidity
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.
2026-06-14 19:12:27 +00:00
5 changed files with 580 additions and 31 deletions
@@ -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
@@ -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
@@ -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,
)
+249
View File
@@ -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)
+67 -26
View File
@@ -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