feat(streams): support Binance price provider

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"

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]

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

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)