Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 2f89945ce1 |
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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)
|
||||
@@ -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
|
||||
|
||||
|
||||
Reference in New Issue
Block a user