Add Polymarket Gamma API support for instrument loading

nautilus_trader/adapters/databento/providers.py

@@ -71,7 +71,7 @@ def __init__(
         super().__init__(config=config)
 
         self._clock = clock
-        self._config = config
+        self._config = config or InstrumentProviderConfig()
         self._live_api_key = live_api_key or http_client.key
         self._live_gateway = live_gateway
 

nautilus_trader/adapters/polymarket/common/gamma_markets.py

@@ -0,0 +1,293 @@
+# -------------------------------------------------------------------------------------------------
+#  Copyright (C) 2015-2025 Nautech Systems Pty Ltd. All rights reserved.
+#  https://nautechsystems.io
+#
+#  Licensed under the GNU Lesser General Public License Version 3.0 (the "License");
+#  You may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at https://www.gnu.org/licenses/lgpl-3.0.en.html
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+# -------------------------------------------------------------------------------------------------
+"""
+Thin Gamma Markets API client utilities for Polymarket.
+
+Provides functions to fetch markets using server-side filters, returning
+raw market dictionaries ready for further client-side filtering.
+
+References
+----------
+- Gamma Get Markets docs: https://docs.polymarket.com/developers/gamma-markets-api/get-markets
+
+"""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Generator
+from typing import Any
+
+import requests
+
+
+DEFAULT_GAMMA_BASE_URL = os.getenv("GAMMA_API_URL", "https://gamma-api.polymarket.com")
+
+
+def _normalize_base_url(base_url: str | None) -> str:
+    url = base_url or DEFAULT_GAMMA_BASE_URL
+    return url[:-1] if url.endswith("/") else url
+
+
+def build_markets_query(filters: dict[str, Any] | None = None) -> dict[str, Any]:
+    """
+    Build query params for Gamma Get Markets from a generic filter dict.
+
+    Supported keys (passed through if present):
+    - active, archived, closed, limit, offset, order, ascending, id, slug,
+      clob_token_ids, condition_ids,
+      liquidity_num_min, liquidity_num_max,
+      volume_num_min, volume_num_max,
+      start_date_min, start_date_max,
+      end_date_min, end_date_max,
+      tag_id, related_tags
+
+    Special handling:
+    - is_active=True implies active=true, archived=false, closed=false
+    - next_cursor: will be added separately by the fetch function
+
+    """
+    params: dict[str, Any] = {}
+    if not filters:
+        return params
+
+    if filters.get("is_active") is True:
+        params["active"] = "true"
+        params["archived"] = "false"
+        params["closed"] = "false"
+
+    passthrough_keys = (
+        "active",
+        "archived",
+        "closed",
+        "limit",
+        "offset",
+        "order",
+        "ascending",
+        "id",
+        "slug",
+        "clob_token_ids",
+        "condition_ids",
+        "liquidity_num_min",
+        "liquidity_num_max",
+        "volume_num_min",
+        "volume_num_max",
+        "start_date_min",
+        "start_date_max",
+        "end_date_min",
+        "end_date_max",
+        "tag_id",
+        "related_tags",
+    )
+    for key in passthrough_keys:
+        if key in filters and filters[key] is not None:
+            params[key] = filters[key]
+
+    return params
+
+
+def _request_markets_page(
+    session: requests.Session,
+    base_url: str,
+    params: dict[str, Any],
+    offset: int,
+    limit: int,
+    timeout: float,
+) -> list[dict[str, Any]]:
+    """
+    Fetch a single page of markets using limit/offset pagination.
+
+    Returns a list of market dicts.
+
+    """
+    url = f"{base_url}/markets"
+    effective_params = dict(params)
+    effective_params["limit"] = limit
+    effective_params["offset"] = offset
+
+    resp = session.get(url, params=effective_params, timeout=timeout)
+    if resp.status_code != 200:
+        raise RuntimeError(f"Gamma Get Markets failed: {resp.status_code} for url {url} with params {effective_params} and body {resp.text}")
+
+    data = resp.json()
+    if isinstance(data, list):
+        return data
+    if isinstance(data, dict) and "data" in data:
+        return data.get("data", []) or []
+
+    raise RuntimeError("Unrecognized response schema from Gamma Get Markets")
+
+
+def iter_markets(
+    filters: dict[str, Any] | None = None,
+    base_url: str | None = None,
+    timeout: float = 10.0,
+) -> Generator[dict[str, Any], None, None]:
+    """
+    Iterate markets that pass server-side filters, yielding raw market dicts.
+    """
+    base = _normalize_base_url(base_url)
+    params = build_markets_query(filters)
+    limit = int(filters.get("limit", 500)) if filters else 500
+    offset = int(filters.get("offset", 0)) if filters else 0
+
+    with requests.Session() as session:
+        while True:
+            markets = _request_markets_page(
+                session=session,
+                base_url=base,
+                params=params,
+                offset=offset,
+                limit=limit,
+                timeout=timeout,
+            )
+            if not markets:
+                break
+            yield from markets
+            if len(markets) < limit:
+                break
+            offset += limit
+
+
+def normalize_gamma_market_to_clob_format(gamma_market: dict[str, Any]) -> dict[str, Any]:
+    """
+    Normalize Gamma API market format to CLOB API format.
+
+    Gamma API uses camelCase field names, while the CLOB API and parsing code
+    expects snake_case field names.
+
+    Parameters
+    ----------
+    gamma_market : dict[str, Any]
+        Market data from Gamma API in camelCase format.
+
+    Returns
+    -------
+    dict[str, Any]
+        Market data normalized to CLOB API format with snake_case fields.
+
+    """
+    import json
+
+    # Handle rewards field
+    rewards = gamma_market.get("clobRewards", [])
+    rewards_dict = None
+    if rewards and len(rewards) > 0:
+        reward = rewards[0]
+        rewards_dict = {
+            "rates": reward.get("rewardsDailyRate"),
+            "min_size": gamma_market.get("rewardsMinSize"),
+            "max_spread": gamma_market.get("rewardsMaxSpread"),
+        }
+
+    # Build tokens array from clobTokenIds and outcomes
+    tokens = []
+    clob_token_ids = gamma_market.get("clobTokenIds", [])
+    outcomes = gamma_market.get("outcomes", [])
+    outcome_prices = gamma_market.get("outcomePrices", [])
+
+    # Parse JSON strings if needed
+    if isinstance(clob_token_ids, str):
+        clob_token_ids = json.loads(clob_token_ids)
+    if isinstance(outcomes, str):
+        outcomes = json.loads(outcomes)
+    if isinstance(outcome_prices, str):
+        outcome_prices = json.loads(outcome_prices)
+
+    # Create tokens array in CLOB format
+    for i, (token_id, outcome) in enumerate(zip(clob_token_ids, outcomes)):
+        token_entry = {
+            "token_id": token_id,
+            "outcome": outcome,
+            "price": float(outcome_prices[i]) if i < len(outcome_prices) else 0.5,
+            "winner": False,  # Default value
+        }
+        tokens.append(token_entry)
+
+    normalized = {
+        # Core identifiers
+        "condition_id": gamma_market.get("conditionId"),
+        "question_id": gamma_market.get("questionID"),
+        "question": gamma_market.get("question"),
+        "description": gamma_market.get("description"),
+        "market_slug": gamma_market.get("slug"),
+
+        # Order book and trading settings
+        "enable_order_book": gamma_market.get("enableOrderBook", True),
+        "minimum_tick_size": gamma_market.get("orderPriceMinTickSize", 0.001),
+        "minimum_order_size": gamma_market.get("orderMinSize", 5),
+        "accepting_orders": gamma_market.get("acceptingOrders", True),
+        "accepting_order_timestamp": gamma_market.get("acceptingOrdersTimestamp"),
+        "seconds_delay": gamma_market.get("secondsDelay", 0),
+
+        # Market status flags
+        "active": gamma_market.get("active", False),
+        "closed": gamma_market.get("closed", False),
+        "archived": gamma_market.get("archived", False),
+
+        # Dates
+        "end_date_iso": gamma_market.get("endDateIso"),
+        "game_start_time": gamma_market.get("startDateIso"),
+
+        # Fee structure
+        "maker_base_fee": 0,  # Gamma API doesn't provide fees directly, use known defaults
+        "taker_base_fee": 0,  # Gamma API doesn't provide fees directly, use known defaults
+        "fpmm": gamma_market.get("marketMakerAddress", ""),
+
+        # Negative risk settings
+        "neg_risk": gamma_market.get("negRisk", False),
+        "neg_risk_market_id": gamma_market.get("negRiskMarketID"),
+        "neg_risk_request_id": gamma_market.get("negRiskRequestID"),
+
+        # Media
+        "icon": gamma_market.get("icon"),
+        "image": gamma_market.get("image"),
+
+        # Rewards and notifications
+        "rewards": rewards_dict,
+        "notifications_enabled": True,  # Default value
+
+        # Outcome pricing flag
+        # "is_50_50_outcome": False,  # Gamma API doesn't provide this flag
+
+        # Tokens array (CLOB API format)
+        "tokens": tokens,
+
+        # Preserve original data for reference
+        "_gamma_original": gamma_market,
+    }
+    return normalized
+
+
+def list_markets(
+    filters: dict[str, Any] | None = None,
+    base_url: str | None = None,
+    timeout: float = 10.0,
+    max_results: int | None = None,
+) -> list[dict[str, Any]]:
+    """
+    Collect markets into a list.
+
+    Use `max_results` to cap total items fetched.
+
+    """
+    results: list[dict[str, Any]] = []
+    count = 0
+    for market in iter_markets(filters=filters, base_url=base_url, timeout=timeout):
+        results.append(market)
+        count += 1
+        if max_results is not None and len(results) >= max_results:
+            break
+    return results

nautilus_trader/adapters/polymarket/providers.py

@@ -20,6 +20,8 @@
 from py_clob_client.client import ClobClient
 
 from nautilus_trader.adapters.polymarket.common.constants import POLYMARKET_VENUE
+from nautilus_trader.adapters.polymarket.common.gamma_markets import list_markets
+from nautilus_trader.adapters.polymarket.common.gamma_markets import normalize_gamma_market_to_clob_format
 from nautilus_trader.adapters.polymarket.common.parsing import parse_polymarket_instrument
 from nautilus_trader.adapters.polymarket.common.symbol import get_polymarket_condition_id
 from nautilus_trader.adapters.polymarket.common.symbol import get_polymarket_token_id
@@ -63,6 +65,59 @@ def __init__(
     async def load_all_async(self, filters: dict | None = None) -> None:
         await self._load_markets([], filters)
 
+    def _load_ids_using_gamma_markets(
+        self,
+        instrument_ids: list[InstrumentId],
+        filters: dict | None = None,
+    ) -> None:
+        """
+        Load instruments using Gamma API markets.
+        """
+        condition_ids = [get_polymarket_condition_id(inst_id) for inst_id in instrument_ids]
+
+        if filters is None:
+            filters = {}
+
+        if len(condition_ids) <= 100:  # We can filter directly by condition_id, but there is an API limit of max 100 condition_ids in the query string
+            self._log.info(f"Loading {len(condition_ids)} instruments, using direct condition_id filtering")
+            filters["condition_ids"] = condition_ids
+        else:
+            self._log.info(f"Loading {len(condition_ids)} instruments, using bulk load of all markets")
+
+        markets = list_markets(filters=filters)  # Usually, you would use filters={"is_active": True} to skip archived markets
+        self._log.info(f"Loaded {len(markets)} markets using Gamma API")
+        for market in markets:
+            condition_id = market.get("conditionId")
+            if not condition_id:
+                continue
+
+            if condition_ids and condition_id not in condition_ids:
+                continue
+
+            normalized_market = normalize_gamma_market_to_clob_format(market)
+
+            # Use the normalized tokens array
+            for token_info in normalized_market.get("tokens", []):
+                token_id = token_info["token_id"]
+                outcome = token_info["outcome"]
+                self._load_instrument(normalized_market, token_id, outcome)
+
+    async def _load_ids_using_clob_api(
+        self,
+        instrument_ids: list[InstrumentId],
+        filters: dict | None = None,
+    ) -> None:
+        """
+        Load instruments using CLOB API.
+        """
+        if len(instrument_ids) > 200:
+            self._log.warning(
+                f"Loading {len(instrument_ids)} instruments, using bulk load of all markets as a faster alternative",
+            )
+            await self._load_markets(instrument_ids, filters)
+        else:
+            await self._load_markets_seq(instrument_ids, filters)
+
     async def load_ids_async(
         self,
         instrument_ids: list[InstrumentId],
@@ -81,13 +136,10 @@ async def load_ids_async(
                 "POLYMARKET",
             )
 
-        if len(instrument_ids) > 200:
-            self._log.warning(
-                f"Loading {len(instrument_ids)} instruments, using bulk load of all markets as a faster alternative",
-            )
-            await self._load_markets(instrument_ids, filters)
+        if self._config.use_gamma_markets:
+            self._load_ids_using_gamma_markets(instrument_ids, filters)
         else:
-            await self._load_markets_seq(instrument_ids, filters)
+            await self._load_ids_using_clob_api(instrument_ids, filters)
 
     async def load_async(self, instrument_id: InstrumentId, filters: dict | None = None) -> None:
         PyCondition.not_none(instrument_id, "instrument_id")