fix: polymarket deprecated endpoint update

src/sources/polymarket.py

@@ -25,7 +25,7 @@
 
 logger = logging.getLogger(__name__)
 
-_GAMMA_API_URL = "https://gamma-api.polymarket.com/markets"
+_GAMMA_API_URL = "https://gamma-api.polymarket.com/markets/keyset"
 _CLOB_API_URL = "https://clob.polymarket.com/prices-history"
 _MIN_MARKET_LIQUIDITY = 25000
 
@@ -228,40 +228,49 @@ def update(
     def _fetch_active_markets_from_api(self) -> list[dict]:
         """Fetch active binary markets from the Gamma API with price history attached.
 
-        Paginates through all active, non-archived, non-closed markets ordered by liquidity,
-        keeps binary markets with sufficient liquidity that aren't catch-all ("other") markets,
+        Paginates through active, non-archived, non-closed markets above the liquidity floor,
+        keeps the binary ones that aren't catch-all ("other") markets,
         and attaches each qualifying market's price history.
         """
         all_markets: list[dict] = []
-        offset = 0
-        limit = 500  # max page size: 500
         n_markets_fetched = 0
-
+        after_cursor: str | None = None
+        seen_ids: set[str] = set()
+
+        # https://docs.polymarket.com/api-reference/markets/list-markets-keyset-pagination
+        # /markets/keyset uses opaque cursor pagination
+        # Note that docs for `order` param are wrong; they specify:
+        # "Comma-separated list of JSON field names to order by, e.g. volume_num,liquidity_num"
+        # but those snake_case fields are ignored; camelCase is correct
+        # Also, liquidity field == liquidityNum (latter is force to number)
         params: dict[str, Any] = {
-            "limit": limit,
+            "limit": 100,  # limit as per API docs
             "archived": False,
             "active": True,
             "closed": False,
             "order": "liquidity",
             "ascending": False,
+            "liquidity_num_min": _MIN_MARKET_LIQUIDITY,
         }
 
         while True:
-            params["offset"] = offset
+            if after_cursor:
+                params["after_cursor"] = after_cursor
             try:
-                logger.info(f"Fetching markets with offset {offset}.")
+                logger.info(f"Fetching markets (cursor={after_cursor}).")
                 response = requests.get(_GAMMA_API_URL, params=params)
                 response.raise_for_status()
-                markets = response.json()
-                if not markets:
-                    logger.info(
-                        f"Fetched total of {n_markets_fetched} markets, "
-                        f"{len(all_markets)} satisfy criteria."
-                    )
-                    break
+                payload = response.json()
+                markets = payload.get("markets", [])
 
                 n_markets_fetched += len(markets)
                 for market in markets:
+                    # Keyset orders by liquidity, which changes as trades land, so a market
+                    # can shift across the cursor and recur on a later page; dedupe by id.
+                    condition_id = market["conditionId"]
+                    if condition_id in seen_ids:
+                        continue
+                    seen_ids.add(condition_id)
                     binary_market = self._is_market_binary(market)
                     # Avoids questions like the following, which don't make sense without the other
                     # questions in the event:
@@ -282,15 +291,30 @@ def _fetch_active_markets_from_api(self) -> list[dict]:
                             market["price_history"] = price_history
                             all_markets.append(market)
 
+                next_cursor = payload.get("next_cursor")
             except requests.exceptions.RequestException as e:
                 logger.error(f"Error fetching markets: {e}")
                 break
 
-            time.sleep(1)
-            offset += limit
+            if not next_cursor:
+                logger.info(
+                    f"Fetched total of {n_markets_fetched} markets, "
+                    f"{len(all_markets)} satisfy criteria."
+                )
+                break
+            after_cursor = next_cursor
+            # cap at ~20 req/s, under the /arkets limit of 300 req/10s (30 req/s)
+            # https://docs.polymarket.com/api-reference/rate-limits
+            time.sleep(0.05)
 
         return all_markets
 
+    @backoff.on_exception(
+        backoff.expo,
+        requests.exceptions.RequestException,
+        max_time=20,
+        on_backoff=data_utils.print_error_info_handler,
+    )
     def _get_market(self, condition_id: str) -> dict:
         """Fetch a single market by condition ID, trying open then closed markets.
 
@@ -309,7 +333,7 @@ def _get_market(self, condition_id: str) -> dict:
         ]:
             response = requests.get(_GAMMA_API_URL, params=params_market)
             response.raise_for_status()
-            markets = response.json()
+            markets = response.json().get("markets", [])
             if len(markets) == 1:
                 return markets[0]
         logger.error(f"Problem getting market for condition id {condition_id}.")

src/tests/test_polymarket.py

@@ -9,6 +9,7 @@
 
 from _schemas import PolymarketFetchFrame, QuestionFrame, ResolutionFrame
 from sources.polymarket import (
+    _MIN_MARKET_LIQUIDITY,
     ConditionIdMarketNotFoundError,
     FailedConditionIdsError,
     PolymarketSource,
@@ -327,10 +328,10 @@ def test_invalid_market_skips_resolution_branch(self):
 class TestFetchActiveMarketsFromApi:
     """Tests for PolymarketSource._fetch_active_markets_from_api."""
 
-    def _mock_response(self, data):
+    def _mock_response(self, data, next_cursor=None):
         resp = Mock()
         resp.ok = True
-        resp.json.return_value = data
+        resp.json.return_value = {"markets": data, "next_cursor": next_cursor}
         resp.raise_for_status = Mock()
         return resp
 
@@ -340,10 +341,7 @@ def _mock_response(self, data):
     def test_basic_returns_qualifying(self, mock_get, mock_price, mock_sleep, polymarket_source):
         """Binary, liquid, non-catch-all markets are returned."""
         market = make_polymarket_api_market()
-        mock_get.side_effect = [
-            self._mock_response([market]),
-            self._mock_response([]),  # End pagination
-        ]
+        mock_get.return_value = self._mock_response([market])
         mock_price.return_value = [{"t": 1736380800, "p": 0.5}]
 
         result = polymarket_source._fetch_active_markets_from_api()
@@ -358,10 +356,7 @@ def test_basic_returns_qualifying(self, mock_get, mock_price, mock_sleep, polyma
     def test_filters_non_binary(self, mock_get, mock_price, mock_sleep, polymarket_source):
         """Non-binary markets are excluded."""
         market = make_polymarket_api_market(outcomes='["Over", "Under"]')
-        mock_get.side_effect = [
-            self._mock_response([market]),
-            self._mock_response([]),
-        ]
+        mock_get.return_value = self._mock_response([market])
 
         result = polymarket_source._fetch_active_markets_from_api()
 
@@ -374,10 +369,7 @@ def test_filters_non_binary(self, mock_get, mock_price, mock_sleep, polymarket_s
     def test_filters_low_liquidity(self, mock_get, mock_price, mock_sleep, polymarket_source):
         """Markets with liquidityNum below the threshold are excluded."""
         market = make_polymarket_api_market(liquidityNum=10000)
-        mock_get.side_effect = [
-            self._mock_response([market]),
-            self._mock_response([]),
-        ]
+        mock_get.return_value = self._mock_response([market])
 
         result = polymarket_source._fetch_active_markets_from_api()
 
@@ -389,10 +381,7 @@ def test_filters_low_liquidity(self, mock_get, mock_price, mock_sleep, polymarke
     def test_filters_catch_all(self, mock_get, mock_price, mock_sleep, polymarket_source):
         """Markets with 'other' in the slug are excluded."""
         market = make_polymarket_api_market(slug="who-will-win-other-candidates")
-        mock_get.side_effect = [
-            self._mock_response([market]),
-            self._mock_response([]),
-        ]
+        mock_get.return_value = self._mock_response([market])
 
         result = polymarket_source._fetch_active_markets_from_api()
 
@@ -406,28 +395,50 @@ def test_pagination(self, mock_get, mock_price, mock_sleep, polymarket_source):
         m1 = make_polymarket_api_market(conditionId="0x001")
         m2 = make_polymarket_api_market(conditionId="0x002")
         mock_get.side_effect = [
-            self._mock_response([m1]),
+            self._mock_response([m1], next_cursor="cursor1"),
             self._mock_response([m2]),
-            self._mock_response([]),
         ]
         mock_price.return_value = [{"t": 1736380800, "p": 0.5}]
 
         result = polymarket_source._fetch_active_markets_from_api()
 
         assert len(result) == 2
+        # The 2nd request must forward the cursor from the 1st response; offset is never sent.
+        assert mock_get.call_count == 2
+        second_params = mock_get.call_args_list[1].kwargs["params"]
+        assert second_params["after_cursor"] == "cursor1"
+        assert "offset" not in second_params
+        # Liquidity floor is pushed server-side (client-side check stays the authoritative cutoff).
+        assert second_params["liquidity_num_min"] == _MIN_MARKET_LIQUIDITY
 
     @patch("sources.polymarket.time.sleep")
     @patch.object(PolymarketSource, "_fetch_price_history")
     @patch("sources.polymarket.requests.get")
-    def test_skips_when_price_history_none(
+    def test_dedupes_market_recurring_across_pages(
         self, mock_get, mock_price, mock_sleep, polymarket_source
     ):
-        """Markets where _fetch_price_history returns None are excluded."""
-        market = make_polymarket_api_market()
+        """A market re-served on a later page (keyset orders by mutable liquidity) is kept once."""
+        market = make_polymarket_api_market(conditionId="0xdupe")
         mock_get.side_effect = [
+            self._mock_response([market], next_cursor="cursor1"),
             self._mock_response([market]),
-            self._mock_response([]),
         ]
+        mock_price.return_value = [{"t": 1736380800, "p": 0.5}]
+
+        result = polymarket_source._fetch_active_markets_from_api()
+
+        assert len(result) == 1
+        assert result[0]["conditionId"] == "0xdupe"
+
+    @patch("sources.polymarket.time.sleep")
+    @patch.object(PolymarketSource, "_fetch_price_history")
+    @patch("sources.polymarket.requests.get")
+    def test_skips_when_price_history_none(
+        self, mock_get, mock_price, mock_sleep, polymarket_source
+    ):
+        """Markets where _fetch_price_history returns None are excluded."""
+        market = make_polymarket_api_market()
+        mock_get.return_value = self._mock_response([market])
         mock_price.return_value = None
 
         result = polymarket_source._fetch_active_markets_from_api()
@@ -443,10 +454,7 @@ def test_filters_missing_liquidity_key(
         """Markets without a liquidityNum key are excluded."""
         market = make_polymarket_api_market()
         del market["liquidityNum"]
-        mock_get.side_effect = [
-            self._mock_response([market]),
-            self._mock_response([]),
-        ]
+        mock_get.return_value = self._mock_response([market])
 
         result = polymarket_source._fetch_active_markets_from_api()
 
@@ -511,7 +519,7 @@ class TestGetMarket:
     def _mock_response(self, data):
         resp = Mock()
         resp.ok = True
-        resp.json.return_value = data
+        resp.json.return_value = {"markets": data}
         resp.raise_for_status = Mock()
         return resp
 
@@ -564,15 +572,22 @@ def test_multiple_results_raises(self, mock_get, polymarket_source):
         with pytest.raises(ConditionIdMarketNotFoundError):
             polymarket_source._get_market("0x001")
 
+    @patch("sources.polymarket.time.sleep")
     @patch("sources.polymarket.requests.get")
-    def test_http_error_propagates(self, mock_get, polymarket_source):
-        """An HTTP error is not swallowed — it propagates."""
-        resp = Mock()
-        resp.raise_for_status.side_effect = requests.exceptions.HTTPError("500")
-        mock_get.return_value = resp
+    def test_transient_error_then_success(self, mock_get, _mock_sleep, polymarket_source):
+        """A transient HTTP error (e.g. a 500 blip) is retried via backoff and then succeeds."""
+        market = make_polymarket_api_market()
+        err_resp = Mock()
+        err_resp.raise_for_status.side_effect = requests.exceptions.HTTPError("500")
+        mock_get.side_effect = [
+            err_resp,  # first attempt: transient 500
+            self._mock_response([market]),  # retry: open-market match
+        ]
 
-        with pytest.raises(requests.exceptions.HTTPError):
-            polymarket_source._get_market("0xabc123")
+        result = polymarket_source._get_market("0xabc123")
+
+        assert result == market
+        assert mock_get.call_count == 2
 
 
 # ---------------------------------------------------------------------------