diff --git a/.gitignore b/.gitignore index 9fb810b..2790fa6 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ tests/* !tests/test_websocket_capabilities.py !tests/test_positions.py !tests/test_rh_lighter.py +!tests/test_mempool.py !tests/_mock_api.py !tests/test_timestamps.py !tests/test_spec_fixes.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 06d7c52..475be9c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,33 @@ All notable changes to the `oxarchive` Python SDK are documented here. The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [1.13.0] - 2026-10-08 + +### Added +- The `mempool` WebSocket channel: signed Hyperliquid transactions as our + Hyperliquid node receives them from its peers, before they are included in + a block, for every Hyperliquid product. It is live only (`replay()` raises + `ValueError` for it before sending), served only at + `wss://stream.0xarchive.io/ws`, and included with the Pro, Scale and + Enterprise plans. Every other channel stays on every plan. +- `subscribe_mempool(symbol=None)` and `unsubscribe_mempool(symbol=None)`. + Without a symbol they cover every pending transaction our Hyperliquid node + receives; with one, only the actions that reference that market. + `mempool` is the only channel whose symbol is optional. +- `on_mempool()` handler, receiving `(symbol, [MempoolItem, ...])`, with + `symbol` set to `None` on the unfiltered stream. Mempool messages reach + `on_message()` as `WsMempoolData`. +- Typed items: `MempoolItem` (`received_at`, `received_at_ms`, `symbols`, + `action`, `nonce`, `vault_address`, `expires_after_ms`, `signature`) and + `MempoolSignature`. `action` is kept exactly as signed, as a dict in the + order sent. +- `oxarchive.websocket.STREAM_WS_URL` (`wss://stream.0xarchive.io/ws`). Pass + it as `WsOptions(ws_url=...)` for a client that subscribes to `mempool`. +- `ws_endpoint` and `plans` on `WsChannelSpec` and `Capability`, mirroring + the fields `/v1/capabilities` sets on the `mempool` row only. For + `ws_endpoint`, `None` means the default endpoint, `wss://api.0xarchive.io/ws`. + For `plans`, `None` means every plan. + ## [1.12.0] - 2026-10-05 Builds on 1.11.0. Two venues, Hyperliquid and Lighter; Lighter now has two diff --git a/README.md b/README.md index b0c6165..138c7f6 100644 --- a/README.md +++ b/README.md @@ -1852,7 +1852,7 @@ Every method has an async version prefixed with `a`: `await client.webhooks.aest Get API keys programmatically using an Ethereum wallet. No browser or email required. -Free includes every market, route, schema, and served depth, with history limited to the most recent rolling 30 days and a maximum 30-day span per request or replay. Build and above keep the full retained archive. Plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. See [Pricing](https://www.0xarchive.io/pricing) for plan capacity. +Free includes every market, route, schema, and served depth, with history limited to the most recent rolling 30 days and a maximum 30-day span per request or replay. Build and above keep the full retained archive. Plans gate capacity and Free's 30-day history window, not route families, schemas, or served depth. The one exception is the live `mempool` WebSocket channel, included with the Pro, Scale and Enterprise plans. See [Pricing](https://www.0xarchive.io/pricing) for plan capacity. #### Free Tier (SIWE) @@ -1984,6 +1984,8 @@ The WebSocket client supports live subscriptions for supported Hyperliquid and L > Lighter on Robinhood Chain supports live subscriptions on `rh_lighter_orderbook`, `rh_lighter_trades`, `rh_lighter_open_interest`, and `rh_lighter_funding` at `wss://api.0xarchive.io/ws` only, with the same message shapes as the mainnet Lighter channels. `rh_lighter_candles` is replay-only. All five channels support historical replay. +> The `mempool` channel (pending Hyperliquid transactions) is served only at `wss://stream.0xarchive.io/ws` and is included with the Pro, Scale and Enterprise plans. See [Pending Transactions (Mempool)](#pending-transactions-mempool). + Every connection selects API version `2026-10-01` (the `version` connection parameter). Server errors arrive on `on_message` as `WsError` messages with a stable `error_code` (see [Error Handling](#error-handling)): for example `unsupported_for_venue` when a channel does not offer the requested mode, `slow_consumer` when the connection fell behind and messages were dropped, and `endpoint_unsupported` when the endpoint does not serve the channel. ```python @@ -2152,6 +2154,65 @@ ws.subscribe("rh_lighter_orderbook", "BTC", interval_ms=250) Live trades are preliminary; `client.rh_lighter.trades.list()` serves the finalized record. +### Pending Transactions (Mempool) + +The `mempool` channel streams signed Hyperliquid transactions (orders, cancels, modifies, TWAPs, leverage changes, transfers and every other action type) as our Hyperliquid node receives them from its peers, before they are included in a block. It covers every Hyperliquid product: perps, HIP-3, HIP-4 and spot. + +- **Live only.** Pending transactions are never stored. There is no replay, history, REST route or export; `replay()` raises `ValueError` for it before sending. +- **One endpoint.** It is served only at `wss://stream.0xarchive.io/ws` (`oxarchive.websocket.STREAM_WS_URL`), with the same API key and protocol. At `wss://api.0xarchive.io/ws`, the client default, a subscribe is answered with a `WsError` whose `error_code` is `endpoint_unsupported`. Use a separate client for the channels served at the default endpoint. +- **Plans.** It is included with the Pro, Scale and Enterprise plans; on other plans a subscribe is answered with `error_code == "forbidden"`. Every other channel stays on every plan, Free included. Each data message is metered like any other WebSocket message. + +```python +import asyncio +from oxarchive import OxArchiveWs, WsError, WsOptions +from oxarchive.websocket import STREAM_WS_URL + +async def main(): + ws = OxArchiveWs(WsOptions(api_key="0xa_your_api_key", ws_url=STREAM_WS_URL)) + + def on_mempool(symbol, items): + for item in items: + print(symbol, item.received_at, item.action["type"], item.symbols) + + def on_message(msg): + if isinstance(msg, WsError): + print(msg.error_code, msg.message) + + ws.on_mempool(on_mempool) + ws.on_message(on_message) + await ws.connect() + + ws.subscribe_mempool("BTC") # actions that reference BTC + ws.subscribe_mempool("xyz:TSLA") # HIP-3; spot "HYPE-USDC", HIP-4 "#49720" + # ws.subscribe_mempool() # pending transactions our node receives (unfiltered) + + await asyncio.sleep(60) + await ws.disconnect() + +asyncio.run(main()) +``` + +`symbol` is optional on this channel only. Without it you receive every pending transaction our Hyperliquid node receives; with it, every action whose asset ids include that market, whole (an order batch that touches `BTC` and `ETH` reaches both subscriptions). Symbols are spelled as everywhere else: perps `BTC`, HIP-3 `xyz:TSLA`, spot `HYPE-USDC` (`HYPE/USDC` is also accepted) and HIP-4 `#49720`. An unknown symbol is answered with `invalid_symbol`. The `subscribed` acknowledgement carries the canonical symbol, or `None` for the unfiltered stream. + +The server sends one message per batch of transactions as it arrives. `on_mempool` receives the subscription's symbol (`None` when unfiltered) and one `MempoolItem` per signed action; `on_message` receives the whole message as `WsMempoolData`. + +| Field | Description | +|-------|-------------| +| `received_at` | When our node received the transaction: an RFC 3339 UTC string with nanosecond precision. Not a block time. | +| `received_at_ms` | The same time in Unix milliseconds | +| `symbols` | Markets the action's asset ids reference, in first-seen order without repeats. Empty for actions with no market, such as transfers, `noop`, `scheduleCancel` and validator actions | +| `action` | The action exactly as signed, in Hyperliquid's exchange-action format: asset ids (`a` or `asset`) rather than symbols, prices and sizes as strings, keys in the order sent | +| `nonce` | The action's nonce | +| `vault_address` | The vault or subaccount the action acts for, or `None` | +| `expires_after_ms` | The action's `expiresAfter` in Unix milliseconds, or `None` | +| `signature` | `MempoolSignature` with `r`, `s` and `v`. The signer's address is not included | + +`action["type"]` names the action, for example `order`, `cancel`, `cancelByCloid`, `modify`, `batchModify`, `scheduleCancel`, `twapOrder`, `twapCancel`, `updateLeverage`, `updateIsolatedMargin`, `noop`, `evmRawTx`, or a transfer such as `usdSend`, `spotSend`, `usdClassTransfer` or `sendAsset`. Hyperliquid adds action types, so handle types you do not recognise. + +A pending transaction is not an executed one: it can still be rejected, expire or never land in a block. The same signed action can occasionally arrive twice; deduplicate on `signature` if that matters to you. + +**Volume and limits**: the unfiltered stream is several megabytes per second before compression. The client negotiates permessage-deflate compression by default; subscribe with a symbol where you can. Unfiltered subscriptions are limited server-wide, and when they are at capacity a subscribe without a symbol is answered with `rate_limited`; symbol subscriptions are not capped this way. A connection that reads too slowly is disconnected, as on any channel, and the usual per-connection limits apply (subscriptions per plan, 10 subscribe operations per second). If the feed is temporarily unavailable, a subscribe is answered with `upstream_unavailable`. + ### Historical Replay Replay historical data with timing preserved. Perfect for backtesting. @@ -2298,7 +2359,7 @@ Gap thresholds vary by channel: ```python ws = OxArchiveWs(WsOptions( api_key="0xa_your_api_key", - ws_url="wss://api.0xarchive.io/ws", # Optional + ws_url="wss://api.0xarchive.io/ws", # Optional; STREAM_WS_URL for mempool auto_reconnect=True, # Auto-reconnect on disconnect (default: True) reconnect_delay=1.0, # Initial reconnect delay in seconds (default: 1.0) max_reconnect_attempts=10, # Max reconnect attempts (default: 10) @@ -2315,6 +2376,9 @@ from oxarchive.websocket import WS_CHANNELS spec = WS_CHANNELS["spot_l4_diffs"] print(spec.venue, spec.live, spec.replay, spec.bulk_replay) # spot True True True + +spec = WS_CHANNELS["mempool"] +print(spec.ws_endpoint, spec.plans) # wss://stream.0xarchive.io/ws ('pro', 'scale', 'enterprise') ``` #### Hyperliquid Channels @@ -2332,6 +2396,7 @@ print(spec.venue, spec.live, spec.replay, spec.bulk_replay) # spot True True Tr | `l4_diffs` | L4 orderbook diffs with user attribution | Yes | Yes | Yes (bulk) | | `l4_orders` | Order lifecycle events with user attribution | Yes | Yes | Yes (bulk) | | `orderbook_full` | Full-depth L2 order book: every price level, then changed levels | Yes | Yes | Yes (bulk) | +| `mempool` | Pending transactions on every Hyperliquid product, before they are in a block (Pro, Scale and Enterprise plans; served only at `wss://stream.0xarchive.io/ws`) | No (optional filter) | Yes | No | Bulk replays open with an `l4_snapshot` and continue with ordered `l4_batch` messages; see [L4 and Full-Depth Replay](#l4-and-full-depth-replay). diff --git a/oxarchive/__init__.py b/oxarchive/__init__.py index 38f25be..af4fc7a 100644 --- a/oxarchive/__init__.py +++ b/oxarchive/__init__.py @@ -203,6 +203,10 @@ WsData, WsL4Snapshot, WsL4Batch, + # Mempool (pending transactions) types + MempoolItem, + MempoolSignature, + WsMempoolData, # Live Lighter payload types LighterLiveTrade, LighterMarketContext, @@ -244,7 +248,7 @@ WS_CHANNELS = None # type: ignore decode_lighter_payload = None # type: ignore -__version__ = "1.12.0" +__version__ = "1.13.0" __all__ = [ # Client @@ -408,6 +412,10 @@ "WsData", "WsL4Snapshot", "WsL4Batch", + # Mempool (Pending Transactions) Types + "MempoolItem", + "MempoolSignature", + "WsMempoolData", # Live Lighter Payload Types "LighterLiveTrade", "LighterMarketContext", diff --git a/oxarchive/types.py b/oxarchive/types.py index be9ea33..c0f3e0a 100644 --- a/oxarchive/types.py +++ b/oxarchive/types.py @@ -1272,6 +1272,7 @@ class Candle(BaseModel): "hip3_l4_diffs", "hip3_l4_orders", "orderbook_full", "hip3_orderbook_full", "spot_orderbook", "spot_trades", "spot_l4_diffs", "spot_l4_orders", "spot_twap", + "mempool", ] """Available WebSocket channels. @@ -1288,8 +1289,8 @@ class Candle(BaseModel): - Replay only: ``candles``, ``hip3_candles``, ``hip4_orderbook``, ``hip4_open_interest``, ``lighter_candles``, ``lighter_l3_orderbook`` and ``rh_lighter_candles``. -- Live only: ``ticker``, ``all_tickers``, ``spot_orderbook`` and - ``spot_trades``. +- Live only: ``ticker``, ``all_tickers``, ``spot_orderbook``, + ``spot_trades`` and ``mempool``. - Neither: ``spot_twap``. Spot TWAP statuses are served over REST only (``client.spot.twap``); subscribing raises ``ValueError``. @@ -1304,6 +1305,11 @@ class Candle(BaseModel): messages share one shape per channel (see :class:`LighterLiveTrade` and :class:`LighterMarketContext`). HIP-4 has no funding or liquidation channels, and spot has no funding, open interest or liquidation channels. + +``mempool`` carries pending Hyperliquid transactions (see +:class:`MempoolItem`). It is the one channel served only at +``wss://stream.0xarchive.io/ws`` and only with the Pro, Scale and Enterprise +plans, and the one channel whose symbol is optional. """ WsConnectionState = Literal["connecting", "connected", "disconnected", "reconnecting"] @@ -1345,8 +1351,10 @@ class WsError(BaseModel): for example ``"invalid_parameter"``, ``"unsupported_for_venue"`` (the channel does not offer that mode), ``"slow_consumer"`` (the connection fell behind and messages were dropped: re-subscribe or restart the replay - to resync) or ``"endpoint_unsupported"`` (this endpoint does not serve the - channel; the message names the one that does). + to resync), ``"endpoint_unsupported"`` (this endpoint does not serve the + channel; the message names the one that does) or ``"forbidden"`` (the + channel is not included with your plan, as for ``mempool`` on plans other + than Pro, Scale and Enterprise). """ model_config = ConfigDict(extra="allow") @@ -1407,6 +1415,101 @@ class WsL4Batch(BaseModel): """Events in server order; each event carries its own block/sequence data.""" +# ============================================================================= +# WebSocket Mempool Types +# ============================================================================= +# +# ``mempool`` streams signed Hyperliquid transactions as our Hyperliquid node +# receives them from its peers, before they are included in a block, for every +# Hyperliquid product (perps, HIP-3, HIP-4 and spot). It is live only: there is +# no replay, history or REST route. It is served only at +# ``wss://stream.0xarchive.io/ws``, with the Pro, Scale and Enterprise plans. + + +class MempoolSignature(BaseModel): + """The ``{r, s, v}`` signature of a pending action, as signed.""" + + model_config = ConfigDict(extra="allow") + + r: str + """``r`` as a hex string.""" + + s: str + """``s`` as a hex string.""" + + v: int + """Recovery id.""" + + +class MempoolItem(BaseModel): + """One signed action from a ``mempool`` message. + + A pending transaction is not an executed one: it can still be rejected, + expire or never land in a block. The same signed action can occasionally + arrive twice; deduplicate on :attr:`signature` if that matters to you. + """ + + model_config = ConfigDict(extra="allow") + + received_at: Optional[str] = None + """When our node received the transaction, as an RFC 3339 UTC string with + the node's nanosecond precision (``"2026-10-08T01:57:23.548737209Z"``). It + is not a block time. Kept as a string because ``datetime`` holds + microseconds only; :attr:`received_at_ms` has it as Unix milliseconds.""" + + received_at_ms: Optional[int] = None + """When our node received the transaction, in Unix milliseconds.""" + + symbols: list[str] = Field(default_factory=list) + """The markets the action's asset ids reference, spelled as everywhere else + (``BTC``, ``xyz:TSLA``, ``HYPE-USDC``, ``#49720``), in first-seen order + without repeats. Empty for actions with no market, such as transfers, + ``noop``, ``scheduleCancel`` and validator actions.""" + + action: dict[str, Any] + """The action exactly as signed, in Hyperliquid's exchange-action format: + asset ids (``a`` or ``asset``) rather than symbols, and prices and sizes as + strings, with keys in the order sent. ``action["type"]`` names it, for + example ``order``, ``cancel``, ``cancelByCloid``, ``modify``, + ``batchModify``, ``scheduleCancel``, ``twapOrder``, ``twapCancel``, + ``updateLeverage``, ``updateIsolatedMargin``, ``noop``, ``evmRawTx`` or a + transfer (``usdSend``, ``spotSend``, ``usdClassTransfer``, ``sendAsset``). + Hyperliquid adds action types, so handle types you do not recognise.""" + + nonce: Optional[int] = None + """The action's nonce.""" + + vault_address: Optional[str] = None + """The vault or subaccount the action acts for, or ``None``.""" + + expires_after_ms: Optional[int] = None + """The action's ``expiresAfter`` in Unix milliseconds, or ``None``.""" + + signature: Optional[MempoolSignature] = None + """The signature over the action. The signer's address is not included; + it can be recovered from the signature and the action.""" + + +class WsMempoolData(BaseModel): + """A ``mempool`` data message: one batch of pending transactions. + + The server sends one message per batch our node receives from a peer, as + soon as it arrives. ``coin`` and ``symbol`` are the subscription's symbol, + or ``None`` on the unfiltered stream. A symbol subscription receives every + action that references that market, whole: an order batch that touches + ``BTC`` and ``ETH`` reaches both subscriptions. + """ + + type: Literal["data"] + channel: Literal["mempool"] + coin: Optional[str] = None + """The subscription's symbol, or ``None`` on the unfiltered stream.""" + symbol: Optional[str] = None + """The subscription's symbol, or ``None`` on the unfiltered stream.""" + data: list[MempoolItem] + """One item per signed action in the batch.""" + + # ============================================================================= # WebSocket Live Lighter Payload Types # ============================================================================= @@ -2853,6 +2956,16 @@ class Capability(BaseModel): notes: Optional[str] = None """Anything else worth knowing, such as how its replay behaves.""" + ws_endpoint: Optional[str] = None + """The WebSocket endpoint that serves its channels, when it is not the + default endpoint (``"wss://stream.0xarchive.io/ws"`` for ``mempool``). + ``None`` means the default endpoint, ``wss://api.0xarchive.io/ws``.""" + + plans: Optional[list[str]] = None + """The plans that include it, when that is not every plan (``["pro", + "scale", "enterprise"]`` for ``mempool``). ``None`` means every plan, Free + included.""" + # ============================================================================= # Data Quality Types diff --git a/oxarchive/websocket.py b/oxarchive/websocket.py index d7f66b5..e3ce59c 100644 --- a/oxarchive/websocket.py +++ b/oxarchive/websocket.py @@ -27,6 +27,13 @@ >>> ws.subscribe_rh_lighter_orderbook("AAPL-USDG") >>> ws.subscribe_rh_lighter_trades("BTC") + Pending Hyperliquid transactions (Pro, Scale and Enterprise plans), served + only at ``wss://stream.0xarchive.io/ws``: + >>> mempool = OxArchiveWs(WsOptions(api_key="ox_...", ws_url=STREAM_WS_URL)) + >>> await mempool.connect() + >>> mempool.on_mempool(lambda symbol, items: print(symbol, len(items))) + >>> mempool.subscribe_mempool("BTC") + Historical replay: >>> ws = OxArchiveWs(WsOptions(api_key="ox_...")) >>> await ws.connect() @@ -71,6 +78,7 @@ LighterMarketContext, LighterMarketContextUpdate, Liquidation, + MempoolItem, OrderBook, OrderbookDelta, PriceLevel, @@ -81,6 +89,7 @@ WsL4Snapshot, WsL4Batch, WsError, + WsMempoolData, WsOutcomeSettled, WsPong, WsSubscribed, @@ -105,6 +114,16 @@ logger = logging.getLogger("oxarchive.websocket") DEFAULT_WS_URL = "wss://api.0xarchive.io/ws" +"""The client default: every channel except ``mempool``, and every replay.""" + +STREAM_WS_URL = "wss://stream.0xarchive.io/ws" +"""The only endpoint that serves ``mempool``, with the same API key and +protocol as :data:`DEFAULT_WS_URL`. It serves a subset of the live channels; +a channel it does not serve, and every replay, is answered with a +:class:`~oxarchive.types.WsError` naming :data:`DEFAULT_WS_URL`. ``mempool`` +has no replay on any endpoint. Use one +client per endpoint: ``WsOptions(api_key=..., ws_url=STREAM_WS_URL)``.""" + DEFAULT_PING_INTERVAL = 30 DEFAULT_RECONNECT_DELAY = 1.0 DEFAULT_MAX_RECONNECT_ATTEMPTS = 10 @@ -125,6 +144,13 @@ class WsChannelSpec: refused. It opens with an ``l4_snapshot`` from the nearest checkpoint at or before ``start`` and continues with ordered ``l4_batch`` messages. + ws_endpoint: The WebSocket endpoint that serves the channel, when it + is not the default endpoint (:data:`STREAM_WS_URL` for + ``mempool``). ``None`` means the default endpoint, + ``wss://api.0xarchive.io/ws``. + plans: The plans that include the channel, when that is not every + plan (``("pro", "scale", "enterprise")`` for ``mempool``). ``None`` + means every plan, Free included. """ venue: str @@ -132,14 +158,24 @@ class WsChannelSpec: live: bool replay: bool bulk_replay: bool = False + ws_endpoint: Optional[str] = None + plans: Optional[tuple[str, ...]] = None def _both(venue: str, datatype: str, *, bulk: bool = False) -> WsChannelSpec: return WsChannelSpec(venue, datatype, live=True, replay=True, bulk_replay=bulk) -def _live_only(venue: str, datatype: str) -> WsChannelSpec: - return WsChannelSpec(venue, datatype, live=True, replay=False) +def _live_only( + venue: str, + datatype: str, + *, + ws_endpoint: Optional[str] = None, + plans: Optional[tuple[str, ...]] = None, +) -> WsChannelSpec: + return WsChannelSpec( + venue, datatype, live=True, replay=False, ws_endpoint=ws_endpoint, plans=plans + ) def _replay_only(venue: str, datatype: str) -> WsChannelSpec: @@ -164,6 +200,12 @@ def _rest_only(venue: str, datatype: str) -> WsChannelSpec: "liquidations": _both("hyperliquid", "liquidations"), "ticker": _live_only("hyperliquid", "ticker"), "all_tickers": _live_only("hyperliquid", "ticker"), + "mempool": _live_only( + "hyperliquid", + "mempool", + ws_endpoint=STREAM_WS_URL, + plans=("pro", "scale", "enterprise"), + ), # HIP-3 "hip3_orderbook": _both("hip3", "l2_orderbook"), "hip3_orderbook_full": _both("hip3", "l2_full_depth", bulk=True), @@ -211,7 +253,13 @@ def _rest_only(venue: str, datatype: str) -> WsChannelSpec: ``hip4_orderbook`` and ``hip4_open_interest`` replay but do not stream live. ``spot_twap`` neither streams nor replays: capabilities lists spot TWAP as REST only (``client.spot.twap``), and it is kept here so that subscribing to it -fails with that explanation instead of waiting for messages that never come.""" +fails with that explanation instead of waiting for messages that never come. + +``mempool`` streams live only, and only at the endpoint in its ``ws_endpoint`` +(:data:`STREAM_WS_URL`) with the plans in its ``plans``. The server decides +both: elsewhere a subscribe is answered with a :class:`~oxarchive.types.WsError` +whose ``error_code`` is ``"endpoint_unsupported"``, and on another plan with +``"forbidden"``.""" LIVE_CHANNELS: frozenset[str] = frozenset(c for c, spec in WS_CHANNELS.items() if spec.live) """Channels a subscription streams live.""" @@ -436,7 +484,8 @@ class WsOptions: Attributes: api_key: Your 0xarchive API key - ws_url: WebSocket server URL (default: wss://api.0xarchive.io/ws) + ws_url: WebSocket server URL (default: wss://api.0xarchive.io/ws). The + ``mempool`` channel is served only at :data:`STREAM_WS_URL`. auto_reconnect: Automatically reconnect on disconnect (default: True) reconnect_delay: Initial reconnect delay in seconds (default: 1.0) max_reconnect_attempts: Maximum reconnection attempts (default: 10) @@ -462,6 +511,7 @@ class WsOptions: WsL4Snapshot, WsL4Batch, WsOutcomeSettled, + WsMempoolData, ] ], None, @@ -469,6 +519,7 @@ class WsOptions: OrderbookHandler = Callable[[str, OrderBook], None] TradesHandler = Callable[[str, list[Trade]], None] LiquidationsHandler = Callable[[str, list[Liquidation]], None] +MempoolHandler = Callable[[Optional[str], list[MempoolItem]], None] OutcomeSettledHandler = Callable[[WsOutcomeSettled], None] LighterMarketContextHandler = Callable[[WsChannel, str, LighterMarketContext], None] StateHandler = Callable[[WsConnectionState], None] @@ -792,6 +843,7 @@ def __init__(self, options: WsOptions): self._on_rh_lighter_orderbook: Optional[OrderbookHandler] = None self._on_rh_lighter_trades: Optional[TradesHandler] = None self._on_rh_lighter_market_context: Optional[LighterMarketContextHandler] = None + self._on_mempool: Optional[MempoolHandler] = None self._on_state_change: Optional[StateHandler] = None self._on_error: Optional[ErrorHandler] = None self._on_open: Optional[Callable[[], None]] = None @@ -895,7 +947,9 @@ def subscribe( Args: channel: Channel type - coin: Coin symbol (required for coin-specific channels) + coin: Coin symbol (required for coin-specific channels; optional + on ``mempool``, where leaving it out subscribes to every + pending transaction our Hyperliquid node receives) interval_ms: ``lighter_orderbook`` and ``rh_lighter_orderbook`` only. Send the newest book at most once per this many milliseconds (100 to 5000). Leave it out for one book a @@ -1288,6 +1342,40 @@ def unsubscribe_rh_lighter_funding(self, coin: str) -> None: """Unsubscribe from live ``rh_lighter_funding`` updates.""" self.unsubscribe("rh_lighter_funding", coin) + # -- Mempool (pending transactions, live only) --------------------------- + # + # Served only at ``wss://stream.0xarchive.io/ws`` (:data:`STREAM_WS_URL`), + # with the Pro, Scale and Enterprise plans. A subscribe sent to the default + # endpoint is answered with ``endpoint_unsupported``, and one on another + # plan with ``forbidden``. + + def subscribe_mempool(self, symbol: Optional[str] = None) -> None: + """Subscribe to pending Hyperliquid transactions, before they are in a block. + + Signed actions (orders, cancels, modifies, TWAPs, leverage changes, + transfers and every other action type) on every Hyperliquid product, + as our Hyperliquid node receives them from its peers. Live only: there + is no replay or history. + + Without ``symbol``, every pending transaction our Hyperliquid node + receives. With one (``BTC``, ``xyz:TSLA``, ``HYPE-USDC``, + ``#49720``), only actions that reference that market; an action that + touches several markets reaches each of their subscriptions whole. + The unfiltered stream is several megabytes a second before + compression, and the server limits unfiltered subscriptions: when they + are at capacity, a subscribe without a symbol is answered with + ``rate_limited``. Subscribe with a symbol where you can. + + Messages arrive on :meth:`on_mempool` and, as + :class:`~oxarchive.types.WsMempoolData`, on :meth:`on_message`. + Connect the client to :data:`STREAM_WS_URL` first. + """ + self.subscribe("mempool", symbol) + + def unsubscribe_mempool(self, symbol: Optional[str] = None) -> None: + """Unsubscribe from pending transactions: the unfiltered stream, or one symbol's.""" + self.unsubscribe("mempool", symbol) + # ========================================================================= # Historical Replay (Option B) - Like Tardis.dev # ========================================================================= @@ -1306,7 +1394,7 @@ async def replay( Every channel whose :data:`WS_CHANNELS` entry has ``replay`` replays; the others (``ticker``, ``all_tickers``, ``spot_orderbook``, - ``spot_trades`` and ``spot_twap``) are refused here with + ``spot_trades``, ``spot_twap`` and ``mempool``) are refused here with ``ValueError`` before anything is sent. L4 channels (``l4_diffs``, ``l4_orders`` and the ``hip3_``, ``hip4_`` @@ -1653,6 +1741,15 @@ def on_rh_lighter_market_context(self, handler: LighterMarketContextHandler) -> """ self._on_rh_lighter_market_context = handler + def on_mempool(self, handler: MempoolHandler) -> None: + """Set handler for ``mempool`` messages (pending transactions). + + Handler receives ``(symbol, [MempoolItem, ...])``: the subscription's + symbol, or ``None`` on the unfiltered stream, and one + :class:`~oxarchive.types.MempoolItem` per signed action. + """ + self._on_mempool = handler + def on_state_change(self, handler: StateHandler) -> None: """Set handler for state changes.""" self._on_state_change = handler @@ -1956,6 +2053,15 @@ def _handle_message(self, raw: str) -> None: if self._on_message: self._on_message(msg) + # Pending transactions. The unfiltered stream has no symbol, so + # these frames have their own model instead of WsData. + elif msg_type == "data" and data.get("channel") == "mempool": + mempool = WsMempoolData.model_validate(data) + if self._on_message: + self._on_message(mempool) + if self._on_mempool: + self._on_mempool(mempool.symbol, mempool.data) + elif msg_type == "data": msg = WsData(**data) if self._on_message: diff --git a/pyproject.toml b/pyproject.toml index 48bdd44..13c5322 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "oxarchive" -version = "1.12.0" +version = "1.13.0" description = "Python SDK for the 0xArchive market data API for Hyperliquid and Lighter" readme = "README.md" license = "MIT" diff --git a/tests/test_api_contract.py b/tests/test_api_contract.py index 08c7e6b..fc0fb30 100644 --- a/tests/test_api_contract.py +++ b/tests/test_api_contract.py @@ -1082,7 +1082,7 @@ def test_l3_history_granularity_stays_refused() -> None: # =========================================================================== # ``ws_channels``, ``live`` and ``replay`` of every row of GET /v1/capabilities -# that names a channel, as served on 2026-10-04. +# that names a channel, as served on 2026-10-04, plus the ``mempool`` row. CAPABILITY_CHANNELS: dict[str, tuple[str, bool, bool]] = { "orderbook": ("hyperliquid", True, True), "orderbook_full": ("hyperliquid", True, True), @@ -1095,6 +1095,7 @@ def test_l3_history_granularity_stays_refused() -> None: "liquidations": ("hyperliquid", True, True), "ticker": ("hyperliquid", True, False), "all_tickers": ("hyperliquid", True, False), + "mempool": ("hyperliquid", True, False), "hip3_orderbook": ("hip3", True, True), "hip3_orderbook_full": ("hip3", True, True), "hip3_l4_diffs": ("hip3", True, True), @@ -1142,6 +1143,19 @@ def test_channel_table_mirrors_capabilities() -> None: assert (spec.venue, spec.live, spec.replay) == (venue, live, replay), channel +def test_only_mempool_names_an_endpoint_and_plans() -> None: + # The mempool row of /v1/capabilities is the only one with ``ws_endpoint`` + # and ``plans``; every other channel is on the default endpoint and every plan. + restricted = { + channel: (spec.ws_endpoint, spec.plans) + for channel, spec in WS_CHANNELS.items() + if spec.ws_endpoint is not None or spec.plans is not None + } + assert restricted == { + "mempool": ("wss://stream.0xarchive.io/ws", ("pro", "scale", "enterprise")) + } + + def test_bulk_replay_channels_are_every_l4_and_full_depth_channel() -> None: assert L4_REPLAY_CHANNELS == {c for c in WS_CHANNELS if "_l4_" in c or c.startswith("l4_")} assert BULK_REPLAY_CHANNELS == L4_REPLAY_CHANNELS | {"orderbook_full", "hip3_orderbook_full"} diff --git a/tests/test_api_contract_live.py b/tests/test_api_contract_live.py index b26cf1e..0a118cf 100644 --- a/tests/test_api_contract_live.py +++ b/tests/test_api_contract_live.py @@ -78,11 +78,13 @@ def test_capabilities_match_the_websocket_channel_table(client: Client) -> None: by_datatype = {(row.venue, row.datatype): row for row in rows} for row in rows: for channel in row.ws_channels: - served[channel] = (row.venue, row.live, row.replay) + plans = tuple(row.plans) if row.plans is not None else None + served[channel] = (row.venue, row.live, row.replay, row.ws_endpoint, plans) assert served, "no row names a WebSocket channel" - for channel, (venue, live, replay) in served.items(): + for channel, (venue, live, replay, ws_endpoint, plans) in served.items(): spec = WS_CHANNELS[channel] assert (spec.venue, spec.live, spec.replay) == (venue, live, replay), channel + assert (spec.ws_endpoint, spec.plans) == (ws_endpoint, plans), channel # A channel no row names must be one its datatype's row says neither # streams nor replays (spot TWAP is served over REST only). for channel, spec in WS_CHANNELS.items(): diff --git a/tests/test_mempool.py b/tests/test_mempool.py new file mode 100644 index 0000000..7a32856 --- /dev/null +++ b/tests/test_mempool.py @@ -0,0 +1,350 @@ +"""The ``mempool`` WebSocket channel: pending Hyperliquid transactions.""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any, Optional + +import pytest +from _mock_api import envelope, mock_client +from test_websocket_capabilities import FakeSocket + +from oxarchive import MempoolItem, MempoolSignature, WsMempoolData +from oxarchive.types import WsChannel, WsData, WsError, WsSubscribed, WsUnsubscribed +from oxarchive.websocket import ( + DEFAULT_WS_URL, + LIVE_CHANNELS, + REPLAY_CHANNELS, + STREAM_WS_URL, + WS_CHANNELS, + OxArchiveWs, + WsOptions, +) + +T_START = 1791424000000 + +# Items shaped as the server sends them: an order on BTC, and a transfer that +# references no market. +ORDER_ITEM: dict[str, Any] = { + "received_at": "2026-10-08T01:57:23.548737209Z", + "received_at_ms": 1791424643548, + "symbols": ["BTC"], + "action": { + "type": "order", + "orders": [ + { + "a": 0, + "b": True, + "p": "83276", + "s": "0.40011", + "r": False, + "t": {"limit": {"tif": "Alo"}}, + "c": "0x7849acc2c6c2f6f0fe4bc80ef13d1504", + } + ], + "grouping": "na", + }, + "nonce": 1791424643400, + "vault_address": None, + "expires_after_ms": None, + "signature": {"r": "0x5afc", "s": "0x57e2", "v": 28}, +} + +TRANSFER_ITEM: dict[str, Any] = { + "received_at": "2026-10-08T01:57:23.548737209Z", + "received_at_ms": 1791424643548, + "symbols": [], + "action": { + "type": "usdSend", + "signatureChainId": "0xa4b1", + "hyperliquidChain": "Mainnet", + "destination": "0x0000000000000000000000000000000000000001", + "amount": "10", + "time": 1791424643400, + }, + "nonce": 1791424643400, + "vault_address": None, + "expires_after_ms": 1791424703400, + "signature": {"r": "0x01", "s": "0x02", "v": 27}, +} + + +def _mempool_frame(symbol: Optional[str], items: list[dict[str, Any]]) -> str: + return json.dumps( + {"type": "data", "channel": "mempool", "coin": symbol, "symbol": symbol, "data": items} + ) + + +def _connected_client() -> tuple[OxArchiveWs, FakeSocket]: + ws = OxArchiveWs(WsOptions(api_key="test-key", ws_url=STREAM_WS_URL)) + socket = FakeSocket() + setattr(ws, "_ws", socket) + return ws, socket + + +def test_mempool_is_live_only_on_the_stream_endpoint_for_pro_and_above() -> None: + spec = WS_CHANNELS["mempool"] + + assert (spec.venue, spec.datatype, spec.live, spec.replay) == ( + "hyperliquid", + "mempool", + True, + False, + ) + assert spec.bulk_replay is False + assert spec.ws_endpoint == STREAM_WS_URL == "wss://stream.0xarchive.io/ws" + assert spec.plans == ("pro", "scale", "enterprise") + assert "mempool" in LIVE_CHANNELS and "mempool" not in REPLAY_CHANNELS + assert DEFAULT_WS_URL == "wss://api.0xarchive.io/ws" + assert OxArchiveWs(WsOptions(api_key="k")).options.ws_url == DEFAULT_WS_URL + + +def test_subscribe_without_a_symbol_is_the_unfiltered_stream() -> None: + ws, socket = _connected_client() + + async def run() -> None: + ws.subscribe_mempool() + await asyncio.sleep(0) + ws.unsubscribe_mempool() + await asyncio.sleep(0) + + asyncio.run(run()) + + assert socket.sent == [ + {"op": "subscribe", "channel": "mempool"}, + {"op": "unsubscribe", "channel": "mempool"}, + ] + assert ws._subscriptions == set() + + +@pytest.mark.parametrize("symbol", ["BTC", "xyz:TSLA", "HYPE-USDC", "#49720"]) +def test_subscribe_with_a_symbol_filters_by_market(symbol: str) -> None: + ws, socket = _connected_client() + + async def run() -> None: + ws.subscribe_mempool(symbol) + await asyncio.sleep(0) + + asyncio.run(run()) + + assert socket.sent == [{"op": "subscribe", "channel": "mempool", "symbol": symbol}] + assert ws._subscriptions == {f"mempool:{symbol}"} + + +def test_unfiltered_and_symbol_subscriptions_are_kept_apart_and_resent_on_reconnect() -> None: + ws, socket = _connected_client() + + async def run() -> None: + await ws.subscribe_async("mempool") + await ws.subscribe_async("mempool", "xyz:TSLA") + socket.sent.clear() + await ws._resubscribe() + + asyncio.run(run()) + + assert ws._subscriptions == {"mempool", "mempool:xyz:TSLA"} + assert sorted(socket.sent, key=lambda m: m.get("symbol", "")) == [ + {"op": "subscribe", "channel": "mempool"}, + {"op": "subscribe", "channel": "mempool", "symbol": "xyz:TSLA"}, + ] + + asyncio.run(ws.unsubscribe_async("mempool", "xyz:TSLA")) + assert ws._subscriptions == {"mempool"} + assert socket.sent[-1] == {"op": "unsubscribe", "channel": "mempool", "symbol": "xyz:TSLA"} + + +def test_mempool_does_not_replay() -> None: + ws, socket = _connected_client() + + with pytest.raises(ValueError, match="does not support historical replay"): + asyncio.run(ws.replay("mempool", "BTC", start=T_START)) + with pytest.raises(ValueError, match="does not support historical replay"): + asyncio.run(ws.multi_replay(["trades", "mempool"], "BTC", start=T_START)) + + assert socket.sent == [] + + +def test_subscribe_ack_without_a_symbol_is_typed() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + messages: list[object] = [] + ws.on_message(messages.append) + + ws._handle_message( + json.dumps({"type": "subscribed", "channel": "mempool", "coin": None, "symbol": None}) + ) + ws._handle_message( + json.dumps({"type": "unsubscribed", "channel": "mempool", "coin": "BTC", "symbol": "BTC"}) + ) + + ack, nack = messages + assert isinstance(ack, WsSubscribed) + assert (ack.channel, ack.coin, ack.symbol) == ("mempool", None, None) + assert isinstance(nack, WsUnsubscribed) + assert (nack.channel, nack.symbol) == ("mempool", "BTC") + + +def test_unfiltered_frame_decodes_to_items() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + received: list[tuple[Optional[str], list[MempoolItem]]] = [] + messages: list[object] = [] + ws.on_mempool(lambda symbol, items: received.append((symbol, items))) + ws.on_message(messages.append) + + ws._handle_message(_mempool_frame(None, [ORDER_ITEM, TRANSFER_ITEM])) + + ((symbol, items),) = received + assert symbol is None + order, transfer = items + assert isinstance(order, MempoolItem) + assert order.received_at == "2026-10-08T01:57:23.548737209Z" + assert order.received_at_ms == 1791424643548 + assert order.symbols == ["BTC"] + assert order.action == ORDER_ITEM["action"] + assert list(order.action) == ["type", "orders", "grouping"] + assert list(order.action["orders"][0]) == ["a", "b", "p", "s", "r", "t", "c"] + assert order.nonce == 1791424643400 + assert order.vault_address is None and order.expires_after_ms is None + assert order.signature == MempoolSignature(r="0x5afc", s="0x57e2", v=28) + assert transfer.symbols == [] + assert transfer.action["type"] == "usdSend" + assert transfer.expires_after_ms == 1791424703400 + + (message,) = messages + assert isinstance(message, WsMempoolData) + assert not isinstance(message, WsData) + assert (message.channel, message.coin, message.symbol) == ("mempool", None, None) + assert message.data == items + + +def test_symbol_frame_carries_the_subscription_symbol() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + received: list[tuple[Optional[str], list[MempoolItem]]] = [] + ws.on_mempool(lambda symbol, items: received.append((symbol, items))) + + ws._handle_message(_mempool_frame("BTC", [ORDER_ITEM])) + + ((symbol, items),) = received + assert symbol == "BTC" + assert items[0].symbols == ["BTC"] + + +def test_mempool_frames_never_reach_the_trade_or_book_handlers() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + other: list[object] = [] + ws.on_trades(lambda coin, trades: other.append(trades)) + ws.on_orderbook(lambda coin, book: other.append(book)) + + ws._handle_message(_mempool_frame("BTC", [ORDER_ITEM])) + + assert other == [] + + +def test_item_tolerates_nulls_and_new_fields() -> None: + item = MempoolItem.model_validate( + { + "received_at": None, + "received_at_ms": None, + "symbols": [], + "action": {"type": "someFutureAction", "x": 1}, + "nonce": None, + "vault_address": "0x00000000000000000000000000000000000000aa", + "expires_after_ms": None, + "signature": None, + "new_field": "kept", + } + ) + + assert item.received_at is None and item.nonce is None and item.signature is None + assert item.action["type"] == "someFutureAction" + assert item.vault_address == "0x00000000000000000000000000000000000000aa" + assert item.model_extra == {"new_field": "kept"} + + +@pytest.mark.parametrize( + "frame", + [ + { + "type": "error", + "message": "The mempool channel is included with the Pro, Scale and Enterprise " + "plans. Upgrade at https://0xarchive.io/pricing.", + "error_code": "forbidden", + }, + { + "type": "error", + "message": "The 'mempool' channel is live only and served on " + "wss://stream.0xarchive.io/ws. Subscribe to it there.", + "error_code": "endpoint_unsupported", + }, + { + "type": "error", + "message": "The unfiltered mempool stream is at capacity. Subscribe with a " + "symbol, or try again later.", + "error_code": "rate_limited", + }, + { + "type": "error", + "message": "The mempool channel is temporarily unavailable. Please try again " + "shortly.", + "error_code": "upstream_unavailable", + }, + {"type": "error", "message": "Unknown symbol 'NOPE'.", "error_code": "invalid_symbol"}, + ], +) +def test_mempool_refusals_reach_on_message_with_their_error_code(frame: dict[str, Any]) -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + messages: list[object] = [] + ws.on_message(messages.append) + + ws._handle_message(json.dumps(frame)) + + (error,) = messages + assert isinstance(error, WsError) + assert (error.error_code, error.message) == (frame["error_code"], frame["message"]) + + +def test_capabilities_parse_the_endpoint_and_plans_of_the_mempool_row() -> None: + rows = [ + { + "venue": "hyperliquid", + "datatype": "mempool", + "rest_routes": [], + "ws_channels": ["mempool"], + "live": True, + "replay": False, + "available_from": None, + "cadence": "event", + "page_limit": None, + "intervals": [], + "notes": "Live only, on wss://stream.0xarchive.io/ws.", + "ws_endpoint": "wss://stream.0xarchive.io/ws", + "plans": ["pro", "scale", "enterprise"], + }, + { + "venue": "hyperliquid", + "datatype": "trades", + "rest_routes": ["/v1/hyperliquid/trades/{symbol}"], + "ws_channels": ["trades"], + "live": True, + "replay": True, + "available_from": None, + "cadence": "event", + "page_limit": 1000, + "intervals": [], + "notes": None, + }, + ] + client, _ = mock_client(lambda p, q: envelope(rows)) + + mempool, trades = client.capabilities() + + assert mempool.ws_endpoint == "wss://stream.0xarchive.io/ws" + assert mempool.plans == ["pro", "scale", "enterprise"] + assert trades.ws_endpoint is None and trades.plans is None + spec = WS_CHANNELS[mempool.ws_channels[0]] + assert (spec.ws_endpoint, list(spec.plans or ())) == (mempool.ws_endpoint, mempool.plans) + + +def test_mempool_is_a_typed_channel_name() -> None: + channel: WsChannel = "mempool" + assert WsSubscribed(type="subscribed", channel=channel).channel == "mempool"