From f4bce022f8e0a9dfbd13e82b791b1f4687a3ab43 Mon Sep 17 00:00:00 2001 From: kartojal Date: Wed, 4 Nov 2026 20:00:36 +0100 Subject: [PATCH 1/2] feat(streams): support Binance price provider --- src/polymarket/models/price_events.py | 1 + src/polymarket/streams/_specs.py | 11 +++++++++-- tests/unit/test_realtime_prices.py | 15 ++++++++++++--- 3 files changed, 22 insertions(+), 5 deletions(-) diff --git a/src/polymarket/models/price_events.py b/src/polymarket/models/price_events.py index 14eb201f..7d47fe17 100644 --- a/src/polymarket/models/price_events.py +++ b/src/polymarket/models/price_events.py @@ -13,6 +13,7 @@ class KnownPriceSource(StrEnum): """Known price sources; additional sources are preserved as plain strings.""" + BINANCE = "binance" PYTH = "pyth" CHAINLINK = "chainlink" MASSIVE = "massive" diff --git a/src/polymarket/streams/_specs.py b/src/polymarket/streams/_specs.py index 29595630..c4bac9bb 100644 --- a/src/polymarket/streams/_specs.py +++ b/src/polymarket/streams/_specs.py @@ -38,7 +38,8 @@ def _validate_perps_instrument_id(instrument_id: object, *, optional: bool = Fal CryptoPricesChainlinkTwapWindowSeconds = Literal[30, 60] EquityPricesEventType = Literal["update", "subscribe"] PriceProviderInput: TypeAlias = ( - Literal["chainlink", "pyth"] | Literal[KnownPriceSource.CHAINLINK, KnownPriceSource.PYTH] + Literal["chainlink", "pyth", "binance"] + | Literal[KnownPriceSource.CHAINLINK, KnownPriceSource.PYTH, KnownPriceSource.BINANCE] ) @@ -50,7 +51,9 @@ def _normalize_price_provider(provider: PriceProviderInput | None) -> PriceProvi return "chainlink" if provider == "pyth": return "pyth" - raise UserInputError("provider must be 'chainlink' or 'pyth'") + if provider == "binance": + return "binance" + raise UserInputError("provider must be 'chainlink', 'pyth' or 'binance'") def _normalize_crypto_symbols(symbols: Sequence[str] | None) -> tuple[str, ...] | None: @@ -467,6 +470,10 @@ class CryptoPriceSpec: Use explicit canonical lowercase symbols such as ``btcusd`` and ``ethusd``. Requires ``AsyncSecureClient``; subscription waits for server acceptance. + + Set ``provider="binance"`` for Binance USDT spot closes under USD symbols. + Omit the provider to follow the default. An unavailable pin may fall back; + check the payload ``source`` for the provider actually serving prices. """ symbols: Sequence[str] diff --git a/tests/unit/test_realtime_prices.py b/tests/unit/test_realtime_prices.py index 97abf971..6be6ba3c 100644 --- a/tests/unit/test_realtime_prices.py +++ b/tests/unit/test_realtime_prices.py @@ -213,7 +213,16 @@ def test_price_specs_reject_unsupported_providers( ], ) @pytest.mark.parametrize( - "provider", [None, "pyth", "chainlink", KnownPriceSource.CHAINLINK, KnownPriceSource.PYTH] + "provider", + [ + None, + "pyth", + "chainlink", + "binance", + KnownPriceSource.CHAINLINK, + KnownPriceSource.PYTH, + KnownPriceSource.BINANCE, + ], ) def test_price_provider_subscribe_and_unsubscribe_requests( spec_type: Any, @@ -353,7 +362,7 @@ async def scenario() -> None: "channel", ["price.crypto", "price.crypto.twap", "price.equity", "price.equity.twap"] ) @pytest.mark.parametrize("snapshot", [False, True]) -@pytest.mark.parametrize("source", ["pyth", "chainlink", "massive", "future-vendor"]) +@pytest.mark.parametrize("source", ["pyth", "chainlink", "massive", "binance", "future-vendor"]) def test_price_frames_preserve_exact_values_and_public_types( channel: str, snapshot: bool, source: str ) -> None: @@ -406,7 +415,7 @@ def test_price_frames_reject_non_string_sources( @pytest.mark.parametrize( "channel", ["price.crypto", "price.crypto.twap", "price.equity", "price.equity.twap"] ) -@pytest.mark.parametrize("source", ["pyth", "chainlink", "massive", "future-vendor"]) +@pytest.mark.parametrize("source", ["pyth", "chainlink", "massive", "binance", "future-vendor"]) @pytest.mark.parametrize("empty", [False, True]) def test_snapshot_and_refreshed_history_preserve_source( channel: str, source: str, empty: bool From 05c1833892f9df364cd2f74b28cc3d4ea8d3ee12 Mon Sep 17 00:00:00 2001 From: kartojal Date: Wed, 4 Nov 2026 20:18:19 +0100 Subject: [PATCH 2/2] test(streams): make pending reconnect backoff deterministic --- tests/unit/test_streams_clob_market.py | 17 ++++++++++++++--- 1 file changed, 14 insertions(+), 3 deletions(-) diff --git a/tests/unit/test_streams_clob_market.py b/tests/unit/test_streams_clob_market.py index 080154d8..def8be85 100644 --- a/tests/unit/test_streams_clob_market.py +++ b/tests/unit/test_streams_clob_market.py @@ -426,11 +426,21 @@ async def run() -> None: asyncio.run(run()) -def test_subscribe_while_reconnect_pending_reopens_immediately() -> None: +def test_subscribe_while_reconnect_pending_reopens_immediately( + monkeypatch: pytest.MonkeyPatch, +) -> None: """If the socket has dropped and a reconnect timer is queued, a new subscribe() must call connect() and bring the socket back up rather than waiting behind the backoff. """ + from polymarket._internal.streams import reconnect + + def fixed_backoff(attempt: int, *, base_s: float, max_s: float) -> float: + return max_s + + # The configured backoff is a cap: jitter can otherwise reconnect before + # this test observes the close and adds the second subscription. + monkeypatch.setattr(reconnect, "jittered_backoff", fixed_backoff) connect_count = 0 initial_frames: list[dict[str, Any]] = [] @@ -454,12 +464,13 @@ async def run() -> tuple[int, list[dict[str, Any]]]: from polymarket._internal.streams.reconnect import ReconnectScheduler mgr = ClobMarketStreamManager(url=url) - mgr._scheduler = ReconnectScheduler(base_s=30.0, max_s=30.0) # pyright: ignore[reportPrivateUsage] + scheduler = ReconnectScheduler(base_s=30.0, max_s=30.0) + mgr._scheduler = scheduler # pyright: ignore[reportPrivateUsage] try: h1 = await mgr.subscribe(token_ids=["a"]) # Wait for the server-initiated close to register. deadline = asyncio.get_event_loop().time() + 2.0 - while mgr.is_open: + while mgr.is_open or not scheduler.is_pending: if asyncio.get_event_loop().time() > deadline: raise AssertionError("server-close not observed") await asyncio.sleep(0.02)