From c28bb5982c8dbdbe531af854d0051949ea73e02a Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Fri, 25 Sep 2026 00:13:42 +0000 Subject: [PATCH 1/4] Support live Lighter WebSocket channels; release 1.11.0 lighter_orderbook, lighter_trades, lighter_open_interest and lighter_funding now accept live subscriptions. lighter_candles and lighter_l3_orderbook stay replay-only, and replay of all six channels is unchanged. - subscribe() and subscribe_async() take an interval_ms keyword for lighter_orderbook (100 to 5000 ms, one book a second when omitted). It is validated before anything is sent and re-sent on reconnect. - subscribe_lighter_* / unsubscribe_lighter_* helpers. - on_lighter_orderbook, on_lighter_trades and on_lighter_market_context handlers, so Lighter BTC is not mixed up with Hyperliquid BTC. Lighter books and trades still reach on_orderbook / on_trades when the dedicated handlers are not set. - Typed live payloads: LighterLiveTrade, LighterMarketContext and LighterMarketContextUpdate. Live trade legs map the Lighter account index to the new Trade.account_index field, which Lighter REST trades also carry. - symbol on WsSubscribed, WsUnsubscribed and WsData. - README WebSocket section, CHANGELOG and version 1.11.0. --- CHANGELOG.md | 43 +++ README.md | 97 +++++- oxarchive/__init__.py | 10 +- oxarchive/types.py | 151 ++++++++- oxarchive/websocket.py | 316 ++++++++++++++++-- pyproject.toml | 2 +- tests/test_websocket_capabilities.py | 465 ++++++++++++++++++++++++++- 7 files changed, 1025 insertions(+), 59 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 386910e..df73778 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,49 @@ 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.11.0] - 2026-09-25 + +### Added +- Live Lighter WebSocket subscriptions for `lighter_orderbook`, + `lighter_trades`, `lighter_open_interest`, and `lighter_funding` on + `wss://api.0xarchive.io/ws` (the client default). New helpers + `subscribe_lighter_orderbook()`, `subscribe_lighter_trades()`, + `subscribe_lighter_open_interest()`, `subscribe_lighter_funding()` and their + `unsubscribe_*` counterparts. +- `interval_ms` keyword on `subscribe()`, `subscribe_async()` and + `subscribe_lighter_orderbook()`. Live Lighter books default to one per + second; pass 100 to 5000 to choose the rate. Each book sent is one metered + message. It is accepted on `lighter_orderbook` only, is checked before + anything is sent, and is re-sent on reconnect. +- `on_lighter_orderbook()`, `on_lighter_trades()` and + `on_lighter_market_context()` handlers. The first two take precedence over + `on_orderbook()` and `on_trades()` for Lighter messages, so Lighter `BTC` + is not mixed up with Hyperliquid `BTC`; without them, Lighter books and + trades still reach the generic handlers. +- Typed live payloads: `LighterLiveTrade` (one trade leg, with the Lighter + account index in `users`) and `LighterMarketContext` / + `LighterMarketContextUpdate` (the message shared by `lighter_open_interest` + and `lighter_funding`). Live books decode to the existing `OrderBook`. +- `Trade.account_index`: the Lighter account index of a fill's owner, set on + live Lighter trade legs and on Lighter REST trades. +- `symbol` on `WsSubscribed`, `WsUnsubscribed` and `WsData`. +- `LIGHTER_LIVE_CHANNELS`, `LIGHTER_REPLAY_ONLY_CHANNELS` and the + `LIGHTER_BOOK_INTERVAL_*` constants in `oxarchive.websocket`. + +### Changed +- Live subscribe calls no longer raise for the four live Lighter channels. + `lighter_candles` and `lighter_l3_orderbook` remain replay-only and still + raise `ValueError` with `LIGHTER_SUBSCRIPTION_ERROR`, whose text now names + those two channels. +- Live Lighter trade legs decode with their own mapping instead of the + Hyperliquid one: `users` becomes `Trade.account_index` (not + `maker_address`), and `order_id`, `crossed` and `start_position` are kept. + Each trade arrives as two legs sharing `trade_id`; `fee`, `fee_token`, + `closed_pnl` and `direction` are `None` in live messages. Live trades are + preliminary; `client.lighter.trades.list()` serves the reconciled record. +- Replay of all six Lighter channels is unchanged and keeps its historical + row shapes, which differ from the live messages. + ## [1.10.0] - 2026-09-23 Versions 1.8.0, 1.9.0 and 1.9.1 are not available on PyPI. This release diff --git a/README.md b/README.md index dee47da..5c9943b 100644 --- a/README.md +++ b/README.md @@ -1254,9 +1254,9 @@ trades = client.trades.list("BTC", start=..., end=...) ## WebSocket Client -The WebSocket client supports live subscriptions for supported Hyperliquid channels and historical replay. For file-based historical exports, use the [Data Catalog](https://www.0xarchive.io/data). +The WebSocket client supports live subscriptions for supported Hyperliquid and Lighter.xyz channels and historical replay. For file-based historical exports, use the [Data Catalog](https://www.0xarchive.io/data). -> Lighter channels support historical replay but not live subscriptions through the 0xArchive WebSocket. Use REST for current data and REST, WebSocket replay, or exports for historical data. +> Lighter supports live subscriptions on `lighter_orderbook`, `lighter_trades`, `lighter_open_interest`, and `lighter_funding` at `wss://api.0xarchive.io/ws` (the client default). `lighter_candles` and `lighter_l3_orderbook` remain replay-only. All six Lighter channels support historical replay. ```python import asyncio @@ -1304,6 +1304,73 @@ async def main(): asyncio.run(main()) ``` +### Live Lighter.xyz Data + +Live Lighter messages use the same `data` envelope as Hyperliquid live data. Symbols are the ones returned by `client.lighter.instruments.list()`; they are case-insensitive on subscribe and echoed uppercase. Live Lighter data is served at `wss://api.0xarchive.io/ws`, the client default; `wss://stream.0xarchive.io/ws` does not serve Lighter channels and answers a Lighter subscribe with an error pointing to `wss://api.0xarchive.io/ws`. It is available on every plan and metered per message like Hyperliquid live data, with the same per-plan subscription and connection limits and the limit of 10 subscribe operations per second. + +```python +import asyncio +from oxarchive import OxArchiveWs, WsError, WsOptions + +def on_book(coin, book): + print(f"Lighter {coin} mid {book.mid_price} ({len(book.bids)} bids, {len(book.asks)} asks)") + +def on_trades(coin, legs): + # Two legs per trade: count distinct trade ids, not legs + print(f"Lighter {coin}: {len({leg.trade_id for leg in legs})} trades") + +def on_context(channel, coin, ctx): + print(f"Lighter {coin} OI {ctx.open_interest} funding {ctx.funding_rate} mark {ctx.mark_price}") + +def on_message(msg): + if isinstance(msg, WsError): # includes lag notices + print(f"Server: {msg.message}") + +async def main(): + ws = OxArchiveWs(WsOptions(api_key="0xa_your_api_key")) + + # Dedicated Lighter handlers keep Lighter BTC apart from Hyperliquid BTC + ws.on_lighter_orderbook(on_book) + ws.on_lighter_trades(on_trades) + ws.on_lighter_market_context(on_context) + ws.on_message(on_message) + + await ws.connect() + + ws.subscribe_lighter_orderbook("BTC") # one book per second (default) + ws.subscribe_lighter_orderbook("ETH", interval_ms=250) # at most one book every 250 ms + ws.subscribe_lighter_trades("BTC") + ws.subscribe_lighter_funding("BTC") # same message as lighter_open_interest + + await asyncio.sleep(60) + await ws.disconnect() + +asyncio.run(main()) +``` + +The generic form works too: `ws.subscribe("lighter_orderbook", "BTC", interval_ms=500)`. + +**`lighter_orderbook`**: every message is a full top-20 book per side, not a diff. Bids and asks are ordered best first, `px` and `sz` are decimal strings exactly as Lighter publishes them, and `n` is always 1 because Lighter does not publish per-level order counts. The server sends the newest book at most once per interval: once per second by default, or at the `interval_ms` you pass (100 to 5000, `lighter_orderbook` only; the SDK raises `ValueError` otherwise). Each book sent is one metered message. On subscribe, the current book is sent right away when one is available; illiquid markets can go minutes without a change. Books arrive on `on_lighter_orderbook` when it is set, otherwise on `on_orderbook`, as `OrderBook` records. + +**`lighter_trades`**: each trade arrives as two legs, one per side, sharing `trade_id`. `side` is `"A"` for the ask side and `"B"` for the bid side, `crossed=True` marks the taker leg, `account_index` is that side's Lighter account index (a string), `order_id` is that side's order id, and `start_position` is that account's signed position before the trade. `fee`, `fee_token`, `closed_pnl`, and `direction` are always `None` in live messages. Count trades by distinct `trade_id`, not by list length, and compute volume from one leg per `trade_id`. Trades arrive on `on_lighter_trades` when it is set, otherwise on `on_trades`, as `Trade` records; the raw legs are modelled by `LighterLiveTrade`. Live trades are preliminary. The finalized record, with fields the live stream does not carry (such as fees), is served by `client.lighter.trades.list()`, which returns reconciled trades only; `client.lighter.trades.recent()` serves the preliminary tier. + +**`lighter_open_interest` and `lighter_funding`**: both channels carry the same message, updated as Lighter publishes it (about once per second per market), with the latest values sent on subscribe when available. `on_lighter_market_context` receives `(channel, coin, LighterMarketContext)`: + +| Field | Wire key | Meaning | +|-------|----------|---------| +| `open_interest` | `openInterest` | Lighter's reported open interest (same value as `client.lighter.open_interest.current()`) | +| `funding_rate` | `funding` | Current funding rate as a fraction (Lighter publishes percent; divided by 100, same as REST `funding_rate`) | +| `premium` | `premium` | Premium as a fraction | +| `mark_price` | `markPx` | Mark price | +| `oracle_price` | `oraclePx` | Lighter's index price | +| `mid_price` | `midPx` | Mid price | +| `day_ntl_volume` | `dayNtlVlm` | 24h quote volume | +| `day_base_volume` | `dayBaseVlm` | 24h base volume | +| `prev_day_price` | `prevDayPx` | Derived from the last trade price and Lighter's 24h percent change | +| `impact_prices` | `impactPxs` | Always `None` (Lighter has no impact prices) | + +**Falling behind**: if your connection falls behind `lighter_trades`, `lighter_open_interest`, or `lighter_funding`, the server sends an error notice (for example `Dropped ~N live lighter_trades messages for BTC: ...`) and continues. If the lag persists, it stops that subscription with `Stopped the lighter_trades stream for BTC: your connection is too slow to keep up. Re-subscribe to resume.`; call the subscribe method again to resume. `lighter_orderbook` always sends the newest book; a book skipped between intervals loses nothing because each book is a full state. + ### Historical Replay Replay historical data with timing preserved. Perfect for backtesting. @@ -1343,8 +1410,8 @@ async def main(): speed=10 # Optional, defaults to 1x ) - # Lighter.xyz replay with granularity. Lighter channels are replay-only; - # use the Lighter REST resources for current data. + # Lighter.xyz replay with granularity. Replay rows keep their historical + # shapes, which differ from the live Lighter messages. await ws.replay( "lighter_orderbook", "BTC", start=int(time.time() * 1000) - 86400000, @@ -1519,16 +1586,16 @@ ws.subscribe_spot_trades("HYPE-USDC") #### Lighter.xyz Channels -| Channel | Description | Requires Coin | Live Subscription | Historical Replay | Current Data Path | -|---------|-------------|---------------|-------------------|-------------------|-------------------| -| `lighter_orderbook` | Lighter L2 order book (reconstructed) | Yes | No | Yes | Lighter REST | -| `lighter_trades` | Lighter trade/fill updates | Yes | No | Yes | Lighter REST | -| `lighter_candles` | Lighter OHLCV candle data | Yes | No | Yes | Lighter REST | -| `lighter_open_interest` | Lighter open interest snapshots | Yes | No | Yes | Lighter REST | -| `lighter_funding` | Lighter funding rate records | Yes | No | Yes | Lighter REST | -| `lighter_l3_orderbook` | Lighter L3 order-level orderbook | Yes | No | Yes | Lighter REST | +| Channel | Description | Requires Coin | Live Subscription | Historical Replay | +|---------|-------------|---------------|-------------------|-------------------| +| `lighter_orderbook` | Lighter L2 order book (live: full top-20 book per side, 1 per second by default, `interval_ms` 100 to 5000) | Yes | Yes | Yes | +| `lighter_trades` | Lighter trade/fill updates (live: two legs per trade) | Yes | Yes | Yes | +| `lighter_candles` | Lighter OHLCV candle data | Yes | No | Yes | +| `lighter_open_interest` | Lighter open interest (live: market context, same message as `lighter_funding`) | Yes | Yes | Yes | +| `lighter_funding` | Lighter funding rates (live: market context, same message as `lighter_open_interest`) | Yes | Yes | Yes | +| `lighter_l3_orderbook` | Lighter L3 order-level orderbook | Yes | No | Yes | -Current Lighter data is available through the REST resources. Historical Lighter data is available through REST, WebSocket replay, or exports. +Live Lighter subscriptions are served at `wss://api.0xarchive.io/ws`. Replay of all six channels keeps its historical row shapes, which differ from the live messages described under Live Lighter.xyz Data above. Current Lighter data is also available through the REST resources, and historical Lighter data through REST, WebSocket replay, or exports. #### Candle Replay @@ -1575,7 +1642,7 @@ await ws.replay( #### Open Interest / Funding Replay -The `open_interest`, `funding`, `lighter_open_interest`, `lighter_funding`, `hip3_open_interest`, and `hip3_funding` channels are **historical only** (replay). They do not support real-time subscriptions. +The `open_interest`, `funding`, `hip3_open_interest`, and `hip3_funding` channels are **historical only** (replay). They do not support real-time subscriptions. `lighter_open_interest` and `lighter_funding` support both replay and live subscriptions (see Live Lighter.xyz Data above). ```python # Replay open interest at 50x speed @@ -1741,7 +1808,7 @@ from oxarchive import Client, LighterGranularity from oxarchive.types import ( OrderBook, Trade, Instrument, LighterInstrument, FundingRate, OpenInterest, Candle, Liquidation, LiquidationVolume, CoinFreshness, CoinSummary, PriceSnapshot, - WsReplaySnapshot, + WsReplaySnapshot, LighterLiveTrade, LighterMarketContext, LighterMarketContextUpdate, ) from oxarchive.resources.trades import CursorResponse diff --git a/oxarchive/__init__.py b/oxarchive/__init__.py index 9e3f35b..4712797 100644 --- a/oxarchive/__init__.py +++ b/oxarchive/__init__.py @@ -96,6 +96,10 @@ WsData, WsL4Snapshot, WsL4Batch, + # Live Lighter payload types + LighterLiveTrade, + LighterMarketContext, + LighterMarketContextUpdate, # Replay types (Option B) WsReplayStarted, WsReplayPaused, @@ -124,7 +128,7 @@ OxArchiveWs = None # type: ignore WsOptions = None # type: ignore -__version__ = "1.10.0" +__version__ = "1.11.0" __all__ = [ # Client @@ -203,6 +207,10 @@ "WsData", "WsL4Snapshot", "WsL4Batch", + # Live Lighter Payload Types + "LighterLiveTrade", + "LighterMarketContext", + "LighterMarketContextUpdate", # Replay Types (Option B) "WsReplayStarted", "WsReplayPaused", diff --git a/oxarchive/types.py b/oxarchive/types.py index 7f94b44..e467e7c 100644 --- a/oxarchive/types.py +++ b/oxarchive/types.py @@ -138,6 +138,10 @@ class Trade(BaseModel): user_address: Optional[str] = None """User's wallet address (for fill-level data).""" + account_index: Optional[str] = None + """Lighter account index of this fill's owner, as a string. Present on Lighter + fills (REST trades and live ``lighter_trades`` legs); ``None`` for Hyperliquid.""" + maker_address: Optional[str] = None """Maker's wallet address (for market-level WebSocket trades).""" @@ -1098,10 +1102,13 @@ class Candle(BaseModel): - ticker/all_tickers are real-time only. - liquidations and hip3_liquidations now stream live (realtime + replay). Each item shares the trades wire shape (a fill row with ``is_liquidation: true``). -- open_interest, funding, lighter_open_interest, lighter_funding, - hip3_open_interest, hip3_funding are historical only (replay/stream). -- all six ``lighter_*`` channels support historical replay, not live - subscriptions. Use Lighter REST for current data. +- open_interest, funding, hip3_open_interest, hip3_funding are historical + only (replay/stream). +- lighter_orderbook, lighter_trades, lighter_open_interest and lighter_funding + support live subscriptions and historical replay. Live messages use the + Hyperliquid-style shapes described on :class:`LighterLiveTrade` and + :class:`LighterMarketContext`; replay rows keep their historical shapes. +- lighter_candles and lighter_l3_orderbook support historical replay only. - l4_diffs, l4_orders: Hyperliquid core L4 order-level data. Historical replay emits one ``l4_snapshot`` followed by ordered ``l4_batch`` messages. - hip3_l4_diffs, hip3_l4_orders: HIP-3 L4 order-level data (live-only). @@ -1126,6 +1133,8 @@ class WsSubscribed(BaseModel): type: Literal["subscribed"] channel: WsChannel coin: Optional[str] = None + symbol: Optional[str] = None + """Symbol as the server echoes it (Lighter symbols are echoed uppercase).""" class WsUnsubscribed(BaseModel): @@ -1134,6 +1143,8 @@ class WsUnsubscribed(BaseModel): type: Literal["unsubscribed"] channel: WsChannel coin: Optional[str] = None + symbol: Optional[str] = None + """Symbol as the server echoes it.""" class WsPong(BaseModel): @@ -1155,11 +1166,15 @@ class WsData(BaseModel): Note: The `data` field can be either a dict (for orderbook) or a list (for trades). - Orderbook: dict with 'levels', 'time', etc. - Trades: list of trade objects with 'coin', 'side', 'px', 'sz', etc. + - lighter_open_interest / lighter_funding: dict with 'coin' and 'ctx' + (see :class:`LighterMarketContextUpdate`). """ type: Literal["data"] channel: WsChannel coin: str + symbol: Optional[str] = None + """Symbol as the server echoes it.""" data: Union[dict[str, Any], list[dict[str, Any]]] @@ -1195,6 +1210,134 @@ class WsL4Batch(BaseModel): """Events in server order; each event carries its own block/sequence data.""" +# ============================================================================= +# WebSocket Live Lighter Payload Types +# ============================================================================= +# +# Live Lighter channels use the same ``data`` envelope as Hyperliquid live data. +# ``lighter_orderbook`` books share the Hyperliquid live book shape +# (``{coin, time, levels: [bids, asks]}``) and decode to :class:`OrderBook`. +# Trades and market context carry Lighter-specific fields and are modelled here. +# Replay of the same channels keeps its historical row shapes. + + +class LighterLiveTrade(BaseModel): + """One leg of a live ``lighter_trades`` trade, as sent in ``WsData.data``. + + Each trade arrives as two legs, one per side, that share ``tid``. Count + trades by distinct ``tid``, not by list length, and compute volume from + ``sz`` over one leg per ``tid``. Live legs are preliminary: the finalized + record, with fields the live stream does not carry (such as fees), is served + by ``client.lighter.trades.list()``. + """ + + coin: str + """Market symbol.""" + + side: Literal["A", "B"] + """``'A'`` for the ask side, ``'B'`` for the bid side.""" + + px: str + """Price as a decimal string, exactly as Lighter publishes it.""" + + sz: str + """Size as a decimal string, exactly as Lighter publishes it.""" + + time: int + """Trade time in Unix milliseconds.""" + + hash: Optional[str] = None + """Lighter transaction hash.""" + + tid: int + """Trade id, shared by both legs of the trade.""" + + oid: Optional[int] = None + """This side's order id.""" + + crossed: bool + """``True`` for the taker leg, ``False`` for the maker leg.""" + + dir: Optional[str] = None + """Always ``None`` in live messages; Lighter's live stream does not carry it.""" + + fee: Optional[str] = None + """Always ``None`` in live messages; Lighter's live stream does not carry it.""" + + fee_token: Optional[str] = None + """Always ``None`` in live messages; Lighter's live stream does not carry it.""" + + closed_pnl: Optional[str] = None + """Always ``None`` in live messages; Lighter's live stream does not carry it.""" + + start_position: Optional[str] = None + """This account's signed position before the trade.""" + + users: list[str] = Field(default_factory=list) + """``[account_index]``: this side's Lighter account index as a string.""" + + @property + def account_index(self) -> Optional[str]: + """This side's Lighter account index, or ``None`` when absent.""" + return self.users[0] if self.users else None + + +class LighterMarketContext(BaseModel): + """Market context from a live ``lighter_open_interest`` or ``lighter_funding`` message. + + Both channels deliver the same message, about once per second per market as + Lighter publishes it. Attributes use the SDK's snake_case names; the wire + keys are Hyperliquid-style camelCase (``openInterest``, ``markPx``, ...) and + are accepted as aliases. Values are decimal strings, and a field is ``None`` + when Lighter has not reported it. + """ + + model_config = {"populate_by_name": True} + + open_interest: Optional[str] = Field(default=None, alias="openInterest") + """Lighter's reported open interest (wire ``openInterest``). Same value as + ``open_interest`` from ``client.lighter.open_interest.current()``.""" + + funding_rate: Optional[str] = Field(default=None, alias="funding") + """Current funding rate as a fraction (wire ``funding``). Lighter publishes + percent; this value is already divided by 100, matching REST ``funding_rate``.""" + + premium: Optional[str] = None + """Premium as a fraction.""" + + mark_price: Optional[str] = Field(default=None, alias="markPx") + """Mark price (wire ``markPx``).""" + + oracle_price: Optional[str] = Field(default=None, alias="oraclePx") + """Lighter's index price (wire ``oraclePx``).""" + + mid_price: Optional[str] = Field(default=None, alias="midPx") + """Mid price (wire ``midPx``).""" + + day_ntl_volume: Optional[str] = Field(default=None, alias="dayNtlVlm") + """24-hour quote volume (wire ``dayNtlVlm``).""" + + day_base_volume: Optional[str] = Field(default=None, alias="dayBaseVlm") + """24-hour base volume (wire ``dayBaseVlm``).""" + + prev_day_price: Optional[str] = Field(default=None, alias="prevDayPx") + """Price 24 hours ago (wire ``prevDayPx``), derived from the last trade price + and Lighter's 24-hour percent change.""" + + impact_prices: Optional[list[str]] = Field(default=None, alias="impactPxs") + """Always ``None`` (wire ``impactPxs``); Lighter has no impact prices.""" + + +class LighterMarketContextUpdate(BaseModel): + """Payload of a live ``lighter_open_interest`` or ``lighter_funding`` message.""" + + coin: str + """Market symbol.""" + + ctx: LighterMarketContext + """Market context values.""" + + # ============================================================================= # WebSocket Replay Types (Historical Replay Mode) # ============================================================================= diff --git a/oxarchive/websocket.py b/oxarchive/websocket.py index 2b77225..b9576ab 100644 --- a/oxarchive/websocket.py +++ b/oxarchive/websocket.py @@ -8,6 +8,11 @@ >>> ws.on_orderbook(lambda coin, ob: print(f"{coin}: {ob.mid_price}")) >>> ws.subscribe_orderbook("BTC") + Live Lighter data (books default to one per second): + >>> ws.on_lighter_orderbook(lambda coin, ob: print(f"Lighter {coin}: {ob.mid_price}")) + >>> ws.subscribe_lighter_orderbook("BTC", interval_ms=250) + >>> ws.subscribe_lighter_trades("BTC") + Historical replay (like Tardis.dev): >>> ws = OxArchiveWs(WsOptions(api_key="ox_...")) >>> await ws.connect() @@ -51,6 +56,9 @@ ) from .types import ( + LighterLiveTrade, + LighterMarketContext, + LighterMarketContextUpdate, Liquidation, OrderBook, OrderbookDelta, @@ -90,23 +98,41 @@ DEFAULT_RECONNECT_DELAY = 1.0 DEFAULT_MAX_RECONNECT_ATTEMPTS = 10 -LIGHTER_REPLAY_CHANNELS: frozenset[str] = frozenset( +LIGHTER_LIVE_CHANNELS: frozenset[str] = frozenset( { "lighter_orderbook", "lighter_trades", - "lighter_candles", "lighter_open_interest", "lighter_funding", + } +) +"""Lighter channels that support live subscriptions (and historical replay).""" + +LIGHTER_REPLAY_ONLY_CHANNELS: frozenset[str] = frozenset( + { + "lighter_candles", "lighter_l3_orderbook", } ) """Lighter channels available through historical replay, not live subscription.""" +LIGHTER_REPLAY_CHANNELS: frozenset[str] = LIGHTER_LIVE_CHANNELS | LIGHTER_REPLAY_ONLY_CHANNELS +"""All six Lighter channels; each supports historical replay.""" + LIGHTER_SUBSCRIPTION_ERROR = ( - "Lighter WebSocket channels support replay, not live subscriptions. " + "lighter_candles and lighter_l3_orderbook support replay, not live subscriptions. " "Use REST for current data or a replay request for stored history." ) +LIGHTER_BOOK_INTERVAL_MIN_MS = 100 +"""Smallest ``interval_ms`` accepted on a ``lighter_orderbook`` subscription.""" + +LIGHTER_BOOK_INTERVAL_MAX_MS = 5000 +"""Largest ``interval_ms`` accepted on a ``lighter_orderbook`` subscription.""" + +LIGHTER_BOOK_INTERVAL_DEFAULT_MS = 1000 +"""Book interval the server uses when ``interval_ms`` is left out.""" + CORE_L4_REPLAY_CHANNELS: frozenset[str] = frozenset({"l4_diffs", "l4_orders"}) """Hyperliquid core L4 channels whose replay starts with a snapshot.""" @@ -134,10 +160,29 @@ def _validate_historical_l4_channel(channel: WsChannel) -> None: raise ValueError(L4_LIVE_ONLY_ERROR) -def _validate_live_subscription(channel: WsChannel) -> None: - """Reject live subscriptions for channels that only support replay.""" - if channel in LIGHTER_REPLAY_CHANNELS: +def _validate_live_subscription(channel: WsChannel, interval_ms: Optional[int] = None) -> None: + """Reject live subscriptions the server would refuse, before any state changes. + + ``lighter_candles`` and ``lighter_l3_orderbook`` are replay-only, and + ``interval_ms`` is accepted only on ``lighter_orderbook``, between + ``LIGHTER_BOOK_INTERVAL_MIN_MS`` and ``LIGHTER_BOOK_INTERVAL_MAX_MS``. + """ + if channel in LIGHTER_REPLAY_ONLY_CHANNELS: raise ValueError(LIGHTER_SUBSCRIPTION_ERROR) + if interval_ms is None: + return + if channel != "lighter_orderbook": + raise ValueError("interval_ms is only supported on lighter_orderbook.") + if ( + isinstance(interval_ms, bool) + or not isinstance(interval_ms, int) + or not LIGHTER_BOOK_INTERVAL_MIN_MS <= interval_ms <= LIGHTER_BOOK_INTERVAL_MAX_MS + ): + raise ValueError( + f"interval_ms must be between {LIGHTER_BOOK_INTERVAL_MIN_MS} and " + f"{LIGHTER_BOOK_INTERVAL_MAX_MS} for lighter_orderbook (got {interval_ms!r}). " + "Leave it out for one book a second." + ) # Server idle timeout is 60 seconds. The SDK sends pings every 30 seconds @@ -189,6 +234,7 @@ class WsOptions: TradesHandler = Callable[[str, list[Trade]], None] LiquidationsHandler = Callable[[str, list[Liquidation]], None] OutcomeSettledHandler = Callable[[WsOutcomeSettled], None] +LighterMarketContextHandler = Callable[[WsChannel, str, LighterMarketContext], None] StateHandler = Callable[[WsConnectionState], None] ErrorHandler = Callable[[Exception], None] @@ -276,6 +322,54 @@ def _transform_trades(coin: str, raw_data: list) -> list[Trade]: return [_transform_trade(coin, t) for t in raw_data] +def _transform_lighter_trade(coin: str, raw: dict[str, Any]) -> Trade: + """Map one live ``lighter_trades`` leg to the SDK Trade type. + + Each leg is one side of a trade: ``users`` holds that side's Lighter + account index, ``oid`` its order id and ``crossed`` marks the taker leg. + Lighter's live stream carries no fee, fee token, realized PnL or direction, + so those stay ``None``. + """ + from datetime import datetime, timezone + + leg = LighterLiveTrade.model_validate({**raw, "coin": raw.get("coin") or coin}) + return Trade( + coin=leg.coin, + side=leg.side, + price=leg.px, + size=leg.sz, + timestamp=datetime.fromtimestamp(leg.time / 1000, tz=timezone.utc), + tx_hash=leg.hash, + trade_id=leg.tid, + order_id=leg.oid, + crossed=leg.crossed, + fee=leg.fee, + fee_token=leg.fee_token, + closed_pnl=leg.closed_pnl, + direction=leg.dir, + start_position=leg.start_position, + account_index=leg.account_index, + ) + + +def _transform_lighter_trades(coin: str, raw_data: Any) -> list[Trade]: + """Map a live ``lighter_trades`` payload (a list of legs) to Trade records.""" + if isinstance(raw_data, dict): + return [_transform_lighter_trade(coin, raw_data)] + if not isinstance(raw_data, list): + return [] + return [_transform_lighter_trade(coin, leg) for leg in raw_data if isinstance(leg, dict)] + + +def _transform_lighter_market_context(coin: str, raw: Any) -> LighterMarketContextUpdate: + """Parse a live ``lighter_open_interest`` / ``lighter_funding`` payload.""" + if not isinstance(raw, dict): + raise ValueError("expected an object with 'coin' and 'ctx'") + return LighterMarketContextUpdate.model_validate( + {"coin": raw.get("coin") or coin, "ctx": raw.get("ctx") or {}} + ) + + def _transform_liquidation(coin: str, raw: dict) -> Liquidation: """Transform a single liquidation fill row to a Liquidation. @@ -410,6 +504,8 @@ def __init__(self, options: WsOptions): self._ws: Optional[ClientConnection] = None self._state: WsConnectionState = "disconnected" self._subscriptions: Set[str] = set() + # Per-subscription options re-sent on reconnect (e.g. interval_ms). + self._subscription_options: dict[str, dict[str, Any]] = {} self._reconnect_attempts = 0 self._running = False self._ping_task: Optional[asyncio.Task] = None @@ -423,6 +519,9 @@ def __init__(self, options: WsOptions): self._on_l4_snapshot: Optional[Callable[[str, str, dict], None]] = None self._on_l4_batch: Optional[Callable[[str, str, list], None]] = None self._on_outcome_settled: Optional[OutcomeSettledHandler] = None + self._on_lighter_orderbook: Optional[OrderbookHandler] = None + self._on_lighter_trades: Optional[TradesHandler] = None + self._on_lighter_market_context: Optional[LighterMarketContextHandler] = None self._on_state_change: Optional[StateHandler] = None self._on_error: Optional[ErrorHandler] = None self._on_open: Optional[Callable[[], None]] = None @@ -508,36 +607,52 @@ async def disconnect(self) -> None: await self._ws.close(1000, "Client disconnect") self._ws = None - def subscribe(self, channel: WsChannel, coin: Optional[str] = None) -> None: + def subscribe( + self, + channel: WsChannel, + coin: Optional[str] = None, + *, + interval_ms: Optional[int] = None, + ) -> None: """Subscribe to a supported live channel. Args: channel: Channel type coin: Coin symbol (required for coin-specific channels) + interval_ms: ``lighter_orderbook`` only. Send the newest book at + most once per this many milliseconds (100 to 5000). Leave it + out for one book a second. Each book sent is one metered + message. Raises: - ValueError: If ``channel`` is a Lighter channel. Lighter supports - replay over WebSocket and current data through REST instead. + ValueError: If ``channel`` is ``lighter_candles`` or + ``lighter_l3_orderbook`` (replay-only), or if ``interval_ms`` is + passed for another channel or is out of range. """ - _validate_live_subscription(channel) - key = self._subscription_key(channel, coin) - self._subscriptions.add(key) + _validate_live_subscription(channel, interval_ms) + self._remember_subscription(channel, coin, interval_ms) if self.is_connected: - asyncio.create_task(self._send_subscribe(channel, coin)) + asyncio.create_task(self._send_subscribe(channel, coin, interval_ms)) - async def subscribe_async(self, channel: WsChannel, coin: Optional[str] = None) -> None: + async def subscribe_async( + self, + channel: WsChannel, + coin: Optional[str] = None, + *, + interval_ms: Optional[int] = None, + ) -> None: """Subscribe asynchronously to a supported live channel. - Lighter channels are replay-only over WebSocket; use REST for current - data or :meth:`replay` for stored history. + Takes the same arguments as :meth:`subscribe`. ``lighter_candles`` and + ``lighter_l3_orderbook`` are replay-only; use REST for current data or + :meth:`replay` for stored history. """ - _validate_live_subscription(channel) - key = self._subscription_key(channel, coin) - self._subscriptions.add(key) + _validate_live_subscription(channel, interval_ms) + self._remember_subscription(channel, coin, interval_ms) if self.is_connected: - await self._send_subscribe(channel, coin) + await self._send_subscribe(channel, coin, interval_ms) def subscribe_orderbook(self, coin: str) -> None: """Subscribe to order book updates for a coin.""" @@ -559,6 +674,7 @@ def unsubscribe(self, channel: WsChannel, coin: Optional[str] = None) -> None: """Unsubscribe from a channel.""" key = self._subscription_key(channel, coin) self._subscriptions.discard(key) + self._subscription_options.pop(key, None) if self.is_connected: asyncio.create_task(self._send_unsubscribe(channel, coin)) @@ -567,6 +683,7 @@ async def unsubscribe_async(self, channel: WsChannel, coin: Optional[str] = None """Unsubscribe from a channel (async version).""" key = self._subscription_key(channel, coin) self._subscriptions.discard(key) + self._subscription_options.pop(key, None) if self.is_connected: await self._send_unsubscribe(channel, coin) @@ -715,6 +832,72 @@ def unsubscribe_spot_twap(self, coin: str) -> None: """Unsubscribe from live spot TWAP status updates for a pair.""" self.unsubscribe("spot_twap", coin) + # -- Lighter.xyz (live) --------------------------------------------------- + + def subscribe_lighter_orderbook(self, coin: str, interval_ms: Optional[int] = None) -> None: + """Subscribe to live Lighter L2 books for a market. + + Every message is a full top-20 book per side (not a diff), bids and asks + best first, with ``px``/``sz`` exactly as Lighter publishes them and + ``n`` always 1 (Lighter does not publish per-level order counts). The + server sends the newest book at most once per interval: one second by + default, or ``interval_ms`` between 100 and 5000. Each book sent is one + metered message. The current book is sent right after subscribing when + one is available. + + Books arrive on :meth:`on_lighter_orderbook` when it is set, otherwise + on :meth:`on_orderbook`. Symbols are the Lighter instrument symbols, + case-insensitive (the server echoes them uppercase). + """ + self.subscribe("lighter_orderbook", coin, interval_ms=interval_ms) + + def unsubscribe_lighter_orderbook(self, coin: str) -> None: + """Unsubscribe from live Lighter L2 books for a market.""" + self.unsubscribe("lighter_orderbook", coin) + + def subscribe_lighter_trades(self, coin: str) -> None: + """Subscribe to live Lighter trades for a market. + + Each trade arrives as two legs (one per side) sharing ``trade_id``; + ``crossed`` marks the taker leg and ``account_index`` is that side's + Lighter account. Fee, fee token, realized PnL and direction are ``None`` + in live messages. Live trades are preliminary; the finalized record is + served by ``client.lighter.trades.list()``. + + Trades arrive on :meth:`on_lighter_trades` when it is set, otherwise on + :meth:`on_trades`. + """ + self.subscribe("lighter_trades", coin) + + def unsubscribe_lighter_trades(self, coin: str) -> None: + """Unsubscribe from live Lighter trades for a market.""" + self.unsubscribe("lighter_trades", coin) + + def subscribe_lighter_open_interest(self, coin: str) -> None: + """Subscribe to live Lighter market context on ``lighter_open_interest``. + + ``lighter_open_interest`` and ``lighter_funding`` carry the same + message; subscribe to one of them unless you want each update twice. + Updates arrive on :meth:`on_lighter_market_context`. + """ + self.subscribe("lighter_open_interest", coin) + + def unsubscribe_lighter_open_interest(self, coin: str) -> None: + """Unsubscribe from live Lighter ``lighter_open_interest`` updates.""" + self.unsubscribe("lighter_open_interest", coin) + + def subscribe_lighter_funding(self, coin: str) -> None: + """Subscribe to live Lighter market context on ``lighter_funding``. + + Carries the same message as ``lighter_open_interest``. Updates arrive + on :meth:`on_lighter_market_context`. + """ + self.subscribe("lighter_funding", coin) + + def unsubscribe_lighter_funding(self, coin: str) -> None: + """Unsubscribe from live Lighter ``lighter_funding`` updates.""" + self.unsubscribe("lighter_funding", coin) + # ========================================================================= # Historical Replay (Option B) - Like Tardis.dev # ========================================================================= @@ -734,9 +917,9 @@ async def replay( Hyperliquid core ``l4_diffs`` and ``l4_orders`` replay as one typed ``l4_snapshot`` followed by ordered ``l4_batch`` messages. HIP-3, HIP-4, and Hyperliquid Spot L4 channels are live-only and are rejected - here. All six ``lighter_*`` channels continue to support bounded - historical replay and do not support live subscriptions; use Lighter - REST for current data. + here. All six ``lighter_*`` channels support bounded historical replay. + Replay rows keep their historical shapes, which differ from the live + Lighter messages. Args: channel: Data channel to replay @@ -985,6 +1168,32 @@ def on_outcome_settled(self, handler: OutcomeSettledHandler) -> None: """ self._on_outcome_settled = handler + def on_lighter_orderbook(self, handler: OrderbookHandler) -> None: + """Set handler for live ``lighter_orderbook`` books. + + Handler receives ``(coin, OrderBook)``. When set, Lighter books go here + instead of :meth:`on_orderbook`, so a Lighter ``BTC`` book is never + mixed up with a Hyperliquid ``BTC`` book. + """ + self._on_lighter_orderbook = handler + + def on_lighter_trades(self, handler: TradesHandler) -> None: + """Set handler for live ``lighter_trades`` legs. + + Handler receives ``(coin, [Trade, ...])``, two legs per trade sharing + ``trade_id``. When set, Lighter trades go here instead of + :meth:`on_trades`. Count trades by distinct ``trade_id``. + """ + self._on_lighter_trades = handler + + def on_lighter_market_context(self, handler: LighterMarketContextHandler) -> None: + """Set handler for live ``lighter_open_interest`` / ``lighter_funding`` updates. + + Handler receives ``(channel, coin, LighterMarketContext)``. Both channels + carry the same message; ``channel`` says which subscription it came from. + """ + self._on_lighter_market_context = handler + def on_state_change(self, handler: StateHandler) -> None: """Set handler for state changes.""" self._on_state_change = handler @@ -1119,13 +1328,35 @@ def _subscription_key(self, channel: WsChannel, coin: Optional[str]) -> str: """Create subscription key.""" return f"{channel}:{coin}" if coin else channel - async def _send_subscribe(self, channel: WsChannel, coin: Optional[str]) -> None: + def _remember_subscription( + self, channel: WsChannel, coin: Optional[str], interval_ms: Optional[int] + ) -> None: + """Record a subscription and its options so a reconnect re-sends them. + + A new subscribe to the same channel and coin replaces the previous one + on the server, so options left out now are cleared here too. + """ + key = self._subscription_key(channel, coin) + self._subscriptions.add(key) + if interval_ms is not None: + self._subscription_options[key] = {"interval_ms": interval_ms} + else: + self._subscription_options.pop(key, None) + + async def _send_subscribe( + self, + channel: WsChannel, + coin: Optional[str], + interval_ms: Optional[int] = None, + ) -> None: """Send subscribe message. Wire field is `symbol`; `coin` is the deprecated alias kept on the SDK surface for backward compatibility.""" if self._ws: - msg = {"op": "subscribe", "channel": channel} + msg: dict[str, Any] = {"op": "subscribe", "channel": channel} if coin: msg["symbol"] = coin + if interval_ms is not None: + msg["interval_ms"] = interval_ms await self._ws.send(json.dumps(msg)) async def _send_unsubscribe(self, channel: WsChannel, coin: Optional[str]) -> None: @@ -1138,11 +1369,12 @@ async def _send_unsubscribe(self, channel: WsChannel, coin: Optional[str]) -> No async def _resubscribe(self) -> None: """Resubscribe to all channels.""" - for key in self._subscriptions: + for key in list(self._subscriptions): parts = key.split(":", 1) channel = parts[0] coin = parts[1] if len(parts) > 1 else None - await self._send_subscribe(channel, coin) # type: ignore + options = self._subscription_options.get(key, {}) + await self._send_subscribe(channel, coin, **options) # type: ignore async def _ping_loop(self) -> None: """Send periodic pings.""" @@ -1215,12 +1447,33 @@ def _handle_message(self, raw: str) -> None: coin = data.get("coin", "") raw_data = data.get("data", {}) + # Live Lighter channels. Books share the Hyperliquid wire shape; + # trades carry Lighter account indexes; OI and funding carry the + # same market-context message. The dedicated Lighter handlers take + # precedence over the generic ones, since a Lighter coin such as + # "BTC" is otherwise indistinguishable from the Hyperliquid one. + if channel == "lighter_orderbook" and self._on_lighter_orderbook: + self._on_lighter_orderbook(coin, _transform_orderbook(coin, raw_data)) + + elif channel == "lighter_trades": + trades_handler = self._on_lighter_trades or self._on_trades + if trades_handler: + trades_handler(coin, _transform_lighter_trades(coin, raw_data)) + + elif ( + channel in ("lighter_open_interest", "lighter_funding") + and self._on_lighter_market_context + ): + update = _transform_lighter_market_context(coin, raw_data) + self._on_lighter_market_context(channel, coin, update.ctx) + # Map venue-prefixed channel variants to the same typed handler. # All orderbook variants (Hyperliquid/HIP-3/HIP-4/Lighter) share - # the same wire shape; same for trades and liquidations. Users - # should not have to fall back to on_message just because they - # subscribed to ``hip4_orderbook`` instead of ``orderbook``. - if channel in ( + # the same wire shape; same for trades (Lighter trades are + # handled above) and liquidations. Users should not have to fall + # back to on_message just because they subscribed to + # ``hip4_orderbook`` instead of ``orderbook``. + elif channel in ( "orderbook", "hip3_orderbook", "hip4_orderbook", @@ -1235,7 +1488,6 @@ def _handle_message(self, raw: str) -> None: "trades", "hip3_trades", "hip4_trades", - "lighter_trades", "spot_trades", ) and self._on_trades: # Transform raw Hyperliquid format to SDK Trade type diff --git a/pyproject.toml b/pyproject.toml index ff2fa38..f4951e0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "oxarchive" -version = "1.10.0" +version = "1.11.0" description = "Python SDK for the 0xArchive market data API for Hyperliquid and Lighter" readme = "README.md" license = "MIT" diff --git a/tests/test_websocket_capabilities.py b/tests/test_websocket_capabilities.py index 9c9bf9f..b7b82d6 100644 --- a/tests/test_websocket_capabilities.py +++ b/tests/test_websocket_capabilities.py @@ -1,21 +1,146 @@ import asyncio import json import re +from datetime import datetime, timezone from typing import Any, cast import pytest - -from oxarchive.types import WsChannel, WsL4Batch, WsL4Snapshot +from websockets.protocol import State as WsState + +from oxarchive.types import ( + LighterLiveTrade, + LighterMarketContext, + LighterMarketContextUpdate, + OrderBook, + Trade, + WsChannel, + WsData, + WsError, + WsL4Batch, + WsL4Snapshot, + WsSubscribed, +) from oxarchive.websocket import ( + LIGHTER_LIVE_CHANNELS, LIGHTER_REPLAY_CHANNELS, + LIGHTER_REPLAY_ONLY_CHANNELS, LIGHTER_SUBSCRIPTION_ERROR, OxArchiveWs, WsOptions, ) +# Frames captured from the production WebSocket (orderbook truncated to three +# levels per side; live frames carry up to 20). +LIGHTER_BOOK_FRAME: dict[str, Any] = { + "coin": "BTC", + "time": 1790294171459, + "levels": [ + [ + {"px": "84368.7", "sz": "0.00020", "n": 1}, + {"px": "84368.6", "sz": "0.00020", "n": 1}, + {"px": "84368.3", "sz": "0.00010", "n": 1}, + ], + [ + {"px": "84368.8", "sz": "0.05720", "n": 1}, + {"px": "84368.9", "sz": "0.14223", "n": 1}, + {"px": "84369.1", "sz": "0.01198", "n": 1}, + ], + ], +} + +LIGHTER_TRADES_FRAME: list[dict[str, Any]] = [ + { + "coin": "BTC", + "side": "A", + "px": "84367.9", + "sz": "0.00003", + "time": 1790294182211, + "hash": "0000001dc8774b28000001a0d5d94943000000000000000000000000000000000000000000000000", + "tid": 31944180930, + "oid": 562953419896990, + "crossed": False, + "dir": None, + "fee": None, + "fee_token": None, + "closed_pnl": None, + "start_position": "109.79011", + "users": ["281474976623827"], + }, + { + "coin": "BTC", + "side": "B", + "px": "84367.9", + "sz": "0.00003", + "time": 1790294182211, + "hash": "0000001dc8774b28000001a0d5d94943000000000000000000000000000000000000000000000000", + "tid": 31944180930, + "oid": 844421425107071, + "crossed": True, + "dir": None, + "fee": None, + "fee_token": None, + "closed_pnl": None, + "start_position": "0.03940", + "users": ["713845"], + }, +] + +LIGHTER_CTX_FRAME: dict[str, Any] = { + "coin": "BTC", + "ctx": { + "openInterest": "172706178.266310", + "funding": "0.000012", + "premium": "-0.000327", + "markPx": "84363.5", + "oraclePx": "84397.0", + "midPx": "84368.8", + "dayNtlVlm": "908611371.550746", + "dayBaseVlm": "10808.97087", + "prevDayPx": "84285.9", + "impactPxs": None, + }, +} + + +def _data_message(channel: str, payload: Any) -> str: + return json.dumps( + {"type": "data", "channel": channel, "coin": "BTC", "symbol": "BTC", "data": payload} + ) -@pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_CHANNELS)) -def test_lighter_channels_reject_live_subscription_before_state_change(channel: str) -> None: + +class FakeSocket: + """Stands in for an open connection and records what the client sends.""" + + def __init__(self) -> None: + self.state = WsState.OPEN + self.sent: list[dict[str, Any]] = [] + + async def send(self, raw: str) -> None: + self.sent.append(json.loads(raw)) + + +def _connected_client() -> tuple[OxArchiveWs, FakeSocket]: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + socket = FakeSocket() + setattr(ws, "_ws", socket) + return ws, socket + + +def test_lighter_channel_sets_split_live_and_replay_only() -> None: + assert LIGHTER_LIVE_CHANNELS == { + "lighter_orderbook", + "lighter_trades", + "lighter_open_interest", + "lighter_funding", + } + assert LIGHTER_REPLAY_ONLY_CHANNELS == {"lighter_candles", "lighter_l3_orderbook"} + assert LIGHTER_REPLAY_CHANNELS == LIGHTER_LIVE_CHANNELS | LIGHTER_REPLAY_ONLY_CHANNELS + + +@pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_ONLY_CHANNELS)) +def test_replay_only_lighter_channels_reject_live_subscription_before_state_change( + channel: str, +) -> None: channel = cast(WsChannel, channel) ws = OxArchiveWs(WsOptions(api_key="test-key")) @@ -25,8 +150,10 @@ def test_lighter_channels_reject_live_subscription_before_state_change(channel: assert ws._subscriptions == set() -@pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_CHANNELS)) -def test_lighter_channels_reject_async_live_subscription_before_state_change(channel: str) -> None: +@pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_ONLY_CHANNELS)) +def test_replay_only_lighter_channels_reject_async_live_subscription_before_state_change( + channel: str, +) -> None: channel = cast(WsChannel, channel) ws = OxArchiveWs(WsOptions(api_key="test-key")) @@ -38,6 +165,159 @@ async def run() -> None: assert ws._subscriptions == set() +@pytest.mark.parametrize("channel", sorted(LIGHTER_LIVE_CHANNELS)) +def test_live_lighter_channels_send_a_plain_subscribe(channel: str) -> None: + channel = cast(WsChannel, channel) + ws, socket = _connected_client() + + asyncio.run(ws.subscribe_async(channel, "BTC")) + + assert ws._subscriptions == {f"{channel}:BTC"} + assert socket.sent == [{"op": "subscribe", "channel": channel, "symbol": "BTC"}] + + +def test_lighter_convenience_helpers_send_subscribe_and_unsubscribe() -> None: + ws, socket = _connected_client() + + async def run() -> None: + ws.subscribe_lighter_orderbook("BTC") + ws.subscribe_lighter_orderbook("ETH", interval_ms=250) + ws.subscribe_lighter_trades("BTC") + ws.subscribe_lighter_open_interest("BTC") + ws.subscribe_lighter_funding("BTC") + await asyncio.sleep(0) + ws.unsubscribe_lighter_orderbook("ETH") + ws.unsubscribe_lighter_trades("BTC") + ws.unsubscribe_lighter_open_interest("BTC") + ws.unsubscribe_lighter_funding("BTC") + await asyncio.sleep(0) + + asyncio.run(run()) + + assert socket.sent == [ + {"op": "subscribe", "channel": "lighter_orderbook", "symbol": "BTC"}, + {"op": "subscribe", "channel": "lighter_orderbook", "symbol": "ETH", "interval_ms": 250}, + {"op": "subscribe", "channel": "lighter_trades", "symbol": "BTC"}, + {"op": "subscribe", "channel": "lighter_open_interest", "symbol": "BTC"}, + {"op": "subscribe", "channel": "lighter_funding", "symbol": "BTC"}, + {"op": "unsubscribe", "channel": "lighter_orderbook", "symbol": "ETH"}, + {"op": "unsubscribe", "channel": "lighter_trades", "symbol": "BTC"}, + {"op": "unsubscribe", "channel": "lighter_open_interest", "symbol": "BTC"}, + {"op": "unsubscribe", "channel": "lighter_funding", "symbol": "BTC"}, + ] + assert ws._subscriptions == {"lighter_orderbook:BTC"} + assert ws._subscription_options == {} + + +@pytest.mark.parametrize("interval_ms", [100, 250, 1000, 5000]) +def test_lighter_orderbook_interval_ms_is_sent_on_the_wire(interval_ms: int) -> None: + ws, socket = _connected_client() + + asyncio.run(ws.subscribe_async("lighter_orderbook", "btc", interval_ms=interval_ms)) + + assert socket.sent == [ + { + "op": "subscribe", + "channel": "lighter_orderbook", + "symbol": "btc", + "interval_ms": interval_ms, + } + ] + + +@pytest.mark.parametrize("interval_ms", [50, 99, 5001, 0, -1]) +def test_lighter_orderbook_interval_ms_out_of_range_is_rejected(interval_ms: int) -> None: + ws, socket = _connected_client() + expected = ( + f"interval_ms must be between 100 and 5000 for lighter_orderbook (got {interval_ms}). " + "Leave it out for one book a second." + ) + + with pytest.raises(ValueError, match=re.escape(expected)): + ws.subscribe_lighter_orderbook("BTC", interval_ms=interval_ms) + + assert ws._subscriptions == set() + assert socket.sent == [] + + +@pytest.mark.parametrize("interval_ms", [True, 250.0, "250"]) +def test_lighter_orderbook_interval_ms_must_be_an_integer(interval_ms: Any) -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + + with pytest.raises(ValueError, match="interval_ms must be between 100 and 5000"): + ws.subscribe("lighter_orderbook", "BTC", interval_ms=interval_ms) + + assert ws._subscriptions == set() + + +@pytest.mark.parametrize( + "channel", ["lighter_trades", "lighter_open_interest", "lighter_funding", "orderbook"] +) +def test_interval_ms_is_only_accepted_on_lighter_orderbook(channel: str) -> None: + channel = cast(WsChannel, channel) + ws, socket = _connected_client() + + async def run() -> None: + with pytest.raises( + ValueError, match=re.escape("interval_ms is only supported on lighter_orderbook.") + ): + await ws.subscribe_async(channel, "BTC", interval_ms=250) + + asyncio.run(run()) + assert ws._subscriptions == set() + assert socket.sent == [] + + +def test_reconnect_resends_interval_ms_and_a_resubscribe_replaces_it() -> None: + ws, socket = _connected_client() + + async def run() -> None: + await ws.subscribe_async("lighter_orderbook", "BTC", interval_ms=250) + await ws.subscribe_async("lighter_trades", "BTC") + socket.sent.clear() + await ws._resubscribe() + + asyncio.run(run()) + assert sorted(socket.sent, key=lambda m: m["channel"]) == [ + {"op": "subscribe", "channel": "lighter_orderbook", "symbol": "BTC", "interval_ms": 250}, + {"op": "subscribe", "channel": "lighter_trades", "symbol": "BTC"}, + ] + + async def resubscribe_without_interval() -> None: + await ws.subscribe_async("lighter_orderbook", "BTC") + socket.sent.clear() + await ws._resubscribe() + + asyncio.run(resubscribe_without_interval()) + assert {"op": "subscribe", "channel": "lighter_orderbook", "symbol": "BTC"} in socket.sent + assert all("interval_ms" not in message for message in socket.sent) + + +@pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_CHANNELS)) +def test_every_lighter_channel_still_replays(channel: str) -> None: + channel = cast(WsChannel, channel) + ws = OxArchiveWs(WsOptions(api_key="test-key")) + sent: list[dict[str, Any]] = [] + + async def fake_send(message: dict[str, Any]) -> None: + sent.append(message) + + setattr(ws, "_send", fake_send) + + asyncio.run(ws.replay(channel, "BTC", start=1_757_000_000_000, end=1_757_003_600_000)) + + assert sent == [ + { + "op": "replay", + "channel": channel, + "symbol": "BTC", + "start": 1_757_000_000_000, + "speed": 1.0, + "end": 1_757_003_600_000, + } + ] + + def test_lighter_channels_are_allowed_for_bounded_replay() -> None: ws = OxArchiveWs(WsOptions(api_key="test-key")) sent: list[dict[str, Any]] = [] @@ -68,6 +348,179 @@ async def fake_send(message: dict[str, Any]) -> None: ] +def test_lighter_subscribe_ack_and_data_envelope_are_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": "lighter_orderbook", "coin": "BTC", "symbol": "BTC"} + ) + ) + ws._handle_message(_data_message("lighter_orderbook", LIGHTER_BOOK_FRAME)) + + ack, data = messages + assert isinstance(ack, WsSubscribed) + assert ack.channel == "lighter_orderbook" + assert ack.symbol == "BTC" + assert isinstance(data, WsData) + assert data.channel == "lighter_orderbook" + assert data.symbol == "BTC" + assert isinstance(data.data, dict) + assert data.data["levels"][0][0] == {"px": "84368.7", "sz": "0.00020", "n": 1} + + +def test_lighter_orderbook_frame_decodes_to_orderbook() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + lighter_books: list[tuple[str, OrderBook]] = [] + generic_books: list[tuple[str, OrderBook]] = [] + ws.on_lighter_orderbook(lambda coin, book: lighter_books.append((coin, book))) + ws.on_orderbook(lambda coin, book: generic_books.append((coin, book))) + + ws._handle_message(_data_message("lighter_orderbook", LIGHTER_BOOK_FRAME)) + + assert generic_books == [] + ((coin, book),) = lighter_books + assert coin == "BTC" + assert [level.px for level in book.bids] == ["84368.7", "84368.6", "84368.3"] + assert [level.px for level in book.asks] == ["84368.8", "84368.9", "84369.1"] + assert book.bids[0].sz == "0.00020" + assert {level.n for level in book.bids + book.asks} == {1} + assert book.timestamp == datetime.fromtimestamp(1790294171459 / 1000, tz=timezone.utc) + assert book.mid_price is not None + assert float(book.mid_price) == pytest.approx(84368.75) + + +def test_lighter_orderbook_falls_back_to_on_orderbook() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + books: list[tuple[str, OrderBook]] = [] + ws.on_orderbook(lambda coin, book: books.append((coin, book))) + + ws._handle_message(_data_message("lighter_orderbook", LIGHTER_BOOK_FRAME)) + + assert len(books) == 1 + assert books[0][1].asks[0].px == "84368.8" + + +def test_lighter_trades_frame_decodes_to_two_legs_per_trade() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + lighter_trades: list[tuple[str, list[Trade]]] = [] + generic_trades: list[tuple[str, list[Trade]]] = [] + ws.on_lighter_trades(lambda coin, trades: lighter_trades.append((coin, trades))) + ws.on_trades(lambda coin, trades: generic_trades.append((coin, trades))) + + ws._handle_message(_data_message("lighter_trades", LIGHTER_TRADES_FRAME)) + + assert generic_trades == [] + ((coin, (ask_leg, bid_leg)),) = lighter_trades + assert coin == "BTC" + assert ask_leg.trade_id == bid_leg.trade_id == 31944180930 + assert (ask_leg.side, ask_leg.crossed, ask_leg.account_index) == ("A", False, "281474976623827") + assert (bid_leg.side, bid_leg.crossed, bid_leg.account_index) == ("B", True, "713845") + assert ask_leg.order_id == 562953419896990 + assert bid_leg.order_id == 844421425107071 + assert ask_leg.start_position == "109.79011" + assert bid_leg.start_position == "0.03940" + assert ask_leg.price == bid_leg.price == "84367.9" + assert ask_leg.size == "0.00003" + assert ask_leg.tx_hash == LIGHTER_TRADES_FRAME[0]["hash"] + assert ask_leg.timestamp == datetime.fromtimestamp(1790294182211 / 1000, tz=timezone.utc) + for leg in (ask_leg, bid_leg): + assert leg.fee is None + assert leg.fee_token is None + assert leg.closed_pnl is None + assert leg.direction is None + assert leg.user_address is None + assert leg.maker_address is None + assert leg.taker_address is None + assert len({leg.trade_id for leg in (ask_leg, bid_leg)}) == 1 + + +def test_lighter_trades_fall_back_to_on_trades() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + received: list[list[Trade]] = [] + ws.on_trades(lambda _coin, trades: received.append(trades)) + + ws._handle_message(_data_message("lighter_trades", LIGHTER_TRADES_FRAME)) + + assert [leg.account_index for leg in received[0]] == ["281474976623827", "713845"] + + +def test_lighter_live_trade_model_parses_the_wire_leg() -> None: + legs = [LighterLiveTrade.model_validate(raw) for raw in LIGHTER_TRADES_FRAME] + + assert [leg.account_index for leg in legs] == ["281474976623827", "713845"] + assert [leg.crossed for leg in legs] == [False, True] + assert all(leg.fee is None and leg.dir is None for leg in legs) + assert LighterLiveTrade.model_validate({**LIGHTER_TRADES_FRAME[0], "users": []}).account_index is None + + +@pytest.mark.parametrize("channel", ["lighter_open_interest", "lighter_funding"]) +def test_lighter_market_context_frame_decodes_on_both_channels(channel: str) -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + received: list[tuple[str, str, LighterMarketContext]] = [] + ws.on_lighter_market_context(lambda ch, coin, ctx: received.append((ch, coin, ctx))) + + ws._handle_message(_data_message(channel, LIGHTER_CTX_FRAME)) + + ((received_channel, coin, ctx),) = received + assert received_channel == channel + assert coin == "BTC" + assert ctx.open_interest == "172706178.266310" + assert ctx.funding_rate == "0.000012" + assert ctx.premium == "-0.000327" + assert ctx.mark_price == "84363.5" + assert ctx.oracle_price == "84397.0" + assert ctx.mid_price == "84368.8" + assert ctx.day_ntl_volume == "908611371.550746" + assert ctx.day_base_volume == "10808.97087" + assert ctx.prev_day_price == "84285.9" + assert ctx.impact_prices is None + + +def test_lighter_market_context_round_trips_wire_keys() -> None: + update = LighterMarketContextUpdate.model_validate(LIGHTER_CTX_FRAME) + + assert update.coin == "BTC" + assert update.ctx.model_dump(by_alias=True) == LIGHTER_CTX_FRAME["ctx"] + assert LighterMarketContext.model_validate({"funding_rate": "0.0001"}).funding_rate == "0.0001" + assert LighterMarketContext.model_validate({}).open_interest is None + + +def test_lighter_lag_notices_reach_on_message_as_errors() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + messages: list[object] = [] + ws.on_message(messages.append) + notice = ( + "Dropped ~12 live lighter_trades messages for BTC: your connection fell behind " + "the Lighter stream, and those trades were not delivered." + ) + + ws._handle_message(json.dumps({"type": "error", "message": notice})) + + assert isinstance(messages[0], WsError) + assert messages[0].message == notice + + +def test_trade_model_accepts_rest_lighter_account_index() -> None: + trade = Trade.model_validate( + { + "coin": "BTC", + "side": "B", + "price": "84367.9", + "size": "0.00003", + "timestamp": "2026-09-24T21:56:22.211Z", + "trade_id": 31944180930, + "crossed": True, + "account_index": "713845", + } + ) + + assert trade.account_index == "713845" + assert trade.user_address is None + + def test_hyperliquid_live_subscription_remains_allowed() -> None: ws = OxArchiveWs(WsOptions(api_key="test-key")) From d3ddf7e07218114fd9b1f594c13ba17bbb2e0f96 Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Fri, 25 Sep 2026 00:34:51 +0000 Subject: [PATCH 2/4] Track Lighter symbols case-insensitively; clearer interval_ms type error - Live Lighter subscriptions are keyed by the uppercase symbol, as the server treats them. Subscribing as btc and unsubscribing as BTC now removes the subscription and its interval_ms, so a reconnect does not restore it, and a resubscribe in another case replaces the earlier interval. - A non-integer interval_ms (for example 250.0 or True) raises "interval_ms must be an integer number of milliseconds" instead of the out-of-range message. Out-of-range integers keep the server's wording. --- CHANGELOG.md | 7 +++-- oxarchive/websocket.py | 26 +++++++++++------ tests/test_websocket_capabilities.py | 42 +++++++++++++++++++++++++++- 3 files changed, 63 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index df73778..9f359a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,8 +17,11 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - `interval_ms` keyword on `subscribe()`, `subscribe_async()` and `subscribe_lighter_orderbook()`. Live Lighter books default to one per second; pass 100 to 5000 to choose the rate. Each book sent is one metered - message. It is accepted on `lighter_orderbook` only, is checked before - anything is sent, and is re-sent on reconnect. + message. It is accepted on `lighter_orderbook` only, must be an integer, is + checked before anything is sent, and is re-sent on reconnect. +- Live Lighter subscriptions are tracked by the uppercase symbol, matching the + server, so subscribing as `btc` and unsubscribing as `BTC` removes the + subscription and it is not restored on reconnect. - `on_lighter_orderbook()`, `on_lighter_trades()` and `on_lighter_market_context()` handlers. The first two take precedence over `on_orderbook()` and `on_trades()` for Lighter messages, so Lighter `BTC` diff --git a/oxarchive/websocket.py b/oxarchive/websocket.py index b9576ab..aaead0d 100644 --- a/oxarchive/websocket.py +++ b/oxarchive/websocket.py @@ -164,8 +164,8 @@ def _validate_live_subscription(channel: WsChannel, interval_ms: Optional[int] = """Reject live subscriptions the server would refuse, before any state changes. ``lighter_candles`` and ``lighter_l3_orderbook`` are replay-only, and - ``interval_ms`` is accepted only on ``lighter_orderbook``, between - ``LIGHTER_BOOK_INTERVAL_MIN_MS`` and ``LIGHTER_BOOK_INTERVAL_MAX_MS``. + ``interval_ms`` is accepted only on ``lighter_orderbook``, as an integer + between ``LIGHTER_BOOK_INTERVAL_MIN_MS`` and ``LIGHTER_BOOK_INTERVAL_MAX_MS``. """ if channel in LIGHTER_REPLAY_ONLY_CHANNELS: raise ValueError(LIGHTER_SUBSCRIPTION_ERROR) @@ -173,11 +173,11 @@ def _validate_live_subscription(channel: WsChannel, interval_ms: Optional[int] = return if channel != "lighter_orderbook": raise ValueError("interval_ms is only supported on lighter_orderbook.") - if ( - isinstance(interval_ms, bool) - or not isinstance(interval_ms, int) - or not LIGHTER_BOOK_INTERVAL_MIN_MS <= interval_ms <= LIGHTER_BOOK_INTERVAL_MAX_MS - ): + if isinstance(interval_ms, bool) or not isinstance(interval_ms, int): + raise ValueError( + f"interval_ms must be an integer number of milliseconds (got {interval_ms!r})." + ) + if not LIGHTER_BOOK_INTERVAL_MIN_MS <= interval_ms <= LIGHTER_BOOK_INTERVAL_MAX_MS: raise ValueError( f"interval_ms must be between {LIGHTER_BOOK_INTERVAL_MIN_MS} and " f"{LIGHTER_BOOK_INTERVAL_MAX_MS} for lighter_orderbook (got {interval_ms!r}). " @@ -627,7 +627,8 @@ def subscribe( Raises: ValueError: If ``channel`` is ``lighter_candles`` or ``lighter_l3_orderbook`` (replay-only), or if ``interval_ms`` is - passed for another channel or is out of range. + passed for another channel, is not an integer, or is out of + range. """ _validate_live_subscription(channel, interval_ms) self._remember_subscription(channel, coin, interval_ms) @@ -1325,7 +1326,14 @@ def _set_state(self, state: WsConnectionState) -> None: self._on_state_change(state) def _subscription_key(self, channel: WsChannel, coin: Optional[str]) -> str: - """Create subscription key.""" + """Create subscription key. + + Live Lighter symbols are case-insensitive on the server, which echoes + them uppercase, so their keys use the uppercase symbol. ``btc`` and + ``BTC`` then name one subscription here, as they do on the server. + """ + if coin and channel in LIGHTER_LIVE_CHANNELS: + coin = coin.upper() return f"{channel}:{coin}" if coin else channel def _remember_subscription( diff --git a/tests/test_websocket_capabilities.py b/tests/test_websocket_capabilities.py index b7b82d6..3f07e02 100644 --- a/tests/test_websocket_capabilities.py +++ b/tests/test_websocket_capabilities.py @@ -243,8 +243,9 @@ def test_lighter_orderbook_interval_ms_out_of_range_is_rejected(interval_ms: int @pytest.mark.parametrize("interval_ms", [True, 250.0, "250"]) def test_lighter_orderbook_interval_ms_must_be_an_integer(interval_ms: Any) -> None: ws = OxArchiveWs(WsOptions(api_key="test-key")) + expected = f"interval_ms must be an integer number of milliseconds (got {interval_ms!r})." - with pytest.raises(ValueError, match="interval_ms must be between 100 and 5000"): + with pytest.raises(ValueError, match=re.escape(expected)): ws.subscribe("lighter_orderbook", "BTC", interval_ms=interval_ms) assert ws._subscriptions == set() @@ -293,6 +294,45 @@ async def resubscribe_without_interval() -> None: assert all("interval_ms" not in message for message in socket.sent) +def test_lighter_symbols_are_tracked_case_insensitively() -> None: + ws, socket = _connected_client() + + async def run() -> None: + await ws.subscribe_async("lighter_orderbook", "btc", interval_ms=250) + await ws.subscribe_async("lighter_trades", "eth") + await ws.unsubscribe_async("lighter_orderbook", "BTC") + socket.sent.clear() + await ws._resubscribe() + + asyncio.run(run()) + assert ws._subscriptions == {"lighter_trades:ETH"} + assert ws._subscription_options == {} + assert socket.sent == [{"op": "subscribe", "channel": "lighter_trades", "symbol": "ETH"}] + + +def test_a_differently_cased_lighter_resubscribe_replaces_the_interval() -> None: + ws, socket = _connected_client() + + async def run() -> None: + await ws.subscribe_async("lighter_orderbook", "btc", interval_ms=250) + await ws.subscribe_async("lighter_orderbook", "BTC") + socket.sent.clear() + await ws._resubscribe() + + asyncio.run(run()) + assert ws._subscriptions == {"lighter_orderbook:BTC"} + assert ws._subscription_options == {} + assert socket.sent == [{"op": "subscribe", "channel": "lighter_orderbook", "symbol": "BTC"}] + + +def test_hyperliquid_symbols_keep_their_case() -> None: + ws, _ = _connected_client() + + asyncio.run(ws.subscribe_async("trades", "kPEPE")) + + assert ws._subscriptions == {"trades:kPEPE"} + + @pytest.mark.parametrize("channel", sorted(LIGHTER_REPLAY_CHANNELS)) def test_every_lighter_channel_still_replays(channel: str) -> None: channel = cast(WsChannel, channel) From 5218adb16c606fcf8d3c846f103b97637d831bd7 Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Fri, 25 Sep 2026 01:46:50 +0000 Subject: [PATCH 3/4] Document live Hyperliquid open interest and funding; note unpublished versions The README listed the Hyperliquid open_interest and funding WebSocket channels as replay only. Both stream live as well, so the channel table, the replay section and the real-time streaming section now say so and show how to subscribe. HIP-3 wording is unchanged. The 1.11.0 changelog entry now states that 1.7.1 through 1.10.0 were not published to PyPI, so upgrading from PyPI goes straight from 1.7.0 to 1.11.0. --- CHANGELOG.md | 9 +++++++++ README.md | 8 +++++--- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9f359a6..680d60f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm ## [1.11.0] - 2026-09-25 +Versions 1.7.1, 1.8.0, 1.9.0, 1.9.1 and 1.10.0 were not published to PyPI. +This release includes their changes, listed in the sections below. The last +release on PyPI is 1.7.0, so upgrading from PyPI goes straight from 1.7.0 to +1.11.0. + ### Added - Live Lighter WebSocket subscriptions for `lighter_orderbook`, `lighter_trades`, `lighter_open_interest`, and `lighter_funding` on @@ -51,6 +56,10 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm - Replay of all six Lighter channels is unchanged and keeps its historical row shapes, which differ from the live messages. +### Fixed +- The README now lists Hyperliquid `open_interest` and `funding` as available + for live subscription as well as replay. + ## [1.10.0] - 2026-09-23 Versions 1.8.0, 1.9.0 and 1.9.1 are not available on PyPI. This release diff --git a/README.md b/README.md index 5c9943b..5e30b2a 100644 --- a/README.md +++ b/README.md @@ -1304,6 +1304,8 @@ async def main(): asyncio.run(main()) ``` +Hyperliquid `open_interest` and `funding` also stream live, in addition to historical replay. Subscribe with `ws.subscribe("open_interest", "BTC")` or `ws.subscribe("funding", "BTC")`. Each update's `data` holds `coin` and a `ctx` object with fields such as `openInterest`, `funding`, `markPx`, and `oraclePx`. These channels have no typed callback, so their updates arrive on `on_message` as `WsData` messages. + ### Live Lighter.xyz Data Live Lighter messages use the same `data` envelope as Hyperliquid live data. Symbols are the ones returned by `client.lighter.instruments.list()`; they are case-insensitive on subscribe and echoed uppercase. Live Lighter data is served at `wss://api.0xarchive.io/ws`, the client default; `wss://stream.0xarchive.io/ws` does not serve Lighter channels and answers a Lighter subscribe with an error pointing to `wss://api.0xarchive.io/ws`. It is available on every plan and metered per message like Hyperliquid live data, with the same per-plan subscription and connection limits and the limit of 10 subscribe operations per second. @@ -1516,8 +1518,8 @@ ws = OxArchiveWs(WsOptions( | `trades` | Trade/fill updates | Yes | Yes | Yes | | `candles` | OHLCV candle data | Yes | No | Yes | | `liquidations` | Liquidation events (May 2025+) | Yes | Yes | Yes | -| `open_interest` | Open interest snapshots | Yes | No | Yes | -| `funding` | Funding rate records | Yes | No | Yes | +| `open_interest` | Open interest snapshots | Yes | Yes | Yes | +| `funding` | Funding rate records | Yes | Yes | Yes | | `ticker` | Price and 24h volume | Yes | Yes | No | | `all_tickers` | All market tickers | No | Yes | No | | `l4_diffs` | L4 orderbook diffs with user attribution | Yes | Yes | Yes | @@ -1642,7 +1644,7 @@ await ws.replay( #### Open Interest / Funding Replay -The `open_interest`, `funding`, `hip3_open_interest`, and `hip3_funding` channels are **historical only** (replay). They do not support real-time subscriptions. `lighter_open_interest` and `lighter_funding` support both replay and live subscriptions (see Live Lighter.xyz Data above). +The Hyperliquid `open_interest` and `funding` channels support both replay and live subscriptions (see Real-time Streaming above). The `hip3_open_interest` and `hip3_funding` channels are **historical only** (replay). They do not support real-time subscriptions. `lighter_open_interest` and `lighter_funding` support both replay and live subscriptions (see Live Lighter.xyz Data above). ```python # Replay open interest at 50x speed From 9e9b8b9f798d94eff0bcf6e707ace15ec9701a23 Mon Sep 17 00:00:00 2001 From: 0xFantomMenace <93462858+0xFantomMenace@users.noreply.github.com> Date: Fri, 25 Sep 2026 02:18:14 +0000 Subject: [PATCH 4/4] Deprecate WebSocket bulk streaming; mark Hyperliquid OI and funding live The server has discontinued WebSocket bulk streaming and answers {"op": "stream"} with an error that points to the S3 Parquet bulk export. stream(), multi_stream() and stream_stop() stay for compatibility but now emit a DeprecationWarning, and their docstrings say the call returns the server's error and that large downloads use the bulk export at https://0xarchive.io/data. The bulk stream handler setters and message models are marked deprecated with them, and the bulk streaming example is gone from the module docstring. The WsChannel docstring now lists Hyperliquid open_interest and funding as live and replayable. The HIP-3 wording is unchanged. The README drops the "replay/stream" wording and notes that bulk streaming is discontinued. --- CHANGELOG.md | 14 ++- README.md | 4 +- oxarchive/__init__.py | 4 +- oxarchive/types.py | 50 +++++++++-- oxarchive/websocket.py | 124 +++++++++++++++++++-------- tests/test_websocket_capabilities.py | 104 +++++++++++++++++++++- 6 files changed, 253 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 680d60f..2985b3f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,9 +56,19 @@ release on PyPI is 1.7.0, so upgrading from PyPI goes straight from 1.7.0 to - Replay of all six Lighter channels is unchanged and keeps its historical row shapes, which differ from the live messages. +### Deprecated +- `OxArchiveWs.stream()`, `multi_stream()` and `stream_stop()`. The server has + discontinued WebSocket bulk streaming and answers these requests with an + error message instead of data. Each call now emits a `DeprecationWarning`. + For large dataset downloads, use the S3 Parquet bulk export at + https://0xarchive.io/data. The `on_batch()`, `on_stream_start()`, + `on_stream_progress()` and `on_stream_complete()` handler setters and the + bulk stream message models are deprecated with them. + ### Fixed -- The README now lists Hyperliquid `open_interest` and `funding` as available - for live subscription as well as replay. +- The README and the `WsChannel` docstring now list Hyperliquid + `open_interest` and `funding` as available for live subscription as well as + replay. ## [1.10.0] - 2026-09-23 diff --git a/README.md b/README.md index 5e30b2a..2e679ba 100644 --- a/README.md +++ b/README.md @@ -1256,6 +1256,8 @@ trades = client.trades.list("BTC", start=..., end=...) The WebSocket client supports live subscriptions for supported Hyperliquid and Lighter.xyz channels and historical replay. For file-based historical exports, use the [Data Catalog](https://www.0xarchive.io/data). +> WebSocket bulk streaming has been discontinued. For large dataset downloads, use the S3 Parquet bulk export in the [Data Catalog](https://www.0xarchive.io/data). The `stream()`, `multi_stream()`, and `stream_stop()` methods remain for compatibility but are deprecated: each call emits a `DeprecationWarning`, and the server answers with an error message (a `WsError` on `on_message`) instead of data. The `on_batch`, `on_stream_start`, `on_stream_progress`, and `on_stream_complete` handler setters are deprecated too: setting one emits a `DeprecationWarning`, and the handler is never called. + > Lighter supports live subscriptions on `lighter_orderbook`, `lighter_trades`, `lighter_open_interest`, and `lighter_funding` at `wss://api.0xarchive.io/ws` (the client default). `lighter_candles` and `lighter_l3_orderbook` remain replay-only. All six Lighter channels support historical replay. ```python @@ -1469,7 +1471,7 @@ from oxarchive import OxArchiveWs, WsOptions async def main(): ws = OxArchiveWs(WsOptions(api_key="ox_...")) - # Handle gap notifications during replay/stream + # Handle gap notifications during replay def handle_gap(channel, coin, gap_start, gap_end, duration_minutes): print(f"Gap detected in {channel}/{coin}:") print(f" From: {gap_start}") diff --git a/oxarchive/__init__.py b/oxarchive/__init__.py index 4712797..66f50f6 100644 --- a/oxarchive/__init__.py +++ b/oxarchive/__init__.py @@ -108,7 +108,7 @@ WsReplayStopped, WsHistoricalData, WsReplaySnapshot, - # Stream types (Option D) + # Bulk stream types (deprecated: the server has discontinued bulk streaming) WsStreamStarted, WsStreamProgress, WsHistoricalBatch, @@ -219,7 +219,7 @@ "WsReplayStopped", "WsHistoricalData", "WsReplaySnapshot", - # Stream Types (Option D) + # Bulk stream types (deprecated: the server has discontinued bulk streaming) "WsStreamStarted", "WsStreamProgress", "WsHistoricalBatch", diff --git a/oxarchive/types.py b/oxarchive/types.py index e467e7c..c9b51ab 100644 --- a/oxarchive/types.py +++ b/oxarchive/types.py @@ -1102,8 +1102,9 @@ class Candle(BaseModel): - ticker/all_tickers are real-time only. - liquidations and hip3_liquidations now stream live (realtime + replay). Each item shares the trades wire shape (a fill row with ``is_liquidation: true``). -- open_interest, funding, hip3_open_interest, hip3_funding are historical - only (replay/stream). +- open_interest and funding (Hyperliquid core) support live subscriptions + and historical replay. +- hip3_open_interest, hip3_funding are historical only (replay). - lighter_orderbook, lighter_trades, lighter_open_interest and lighter_funding support live subscriptions and historical replay. Live messages use the Hyperliquid-style shapes described on :class:`LighterLiveTrade` and @@ -1464,13 +1465,23 @@ class WsHistoricalTickData(BaseModel): # ============================================================================= -# WebSocket Bulk Stream Types (Data Catalog Mode) +# WebSocket Bulk Stream Types (deprecated) +# +# The server has discontinued WebSocket bulk streaming and no longer sends +# these messages. The models stay exported for backward compatibility. For +# large dataset downloads, use the S3 Parquet bulk export at +# https://0xarchive.io/data. # ============================================================================= class WsStreamStarted(BaseModel): """Stream started response. + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming and no longer sends this + message. For large dataset downloads, use the S3 Parquet bulk export + at https://0xarchive.io/data. + In multi-channel mode, ``channels`` lists all channels being streamed. """ @@ -1487,21 +1498,37 @@ class WsStreamStarted(BaseModel): class WsStreamProgress(BaseModel): - """Stream progress response (sent every ~2 seconds).""" + """Stream progress response. + + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming and no longer sends this + message. + """ type: Literal["stream_progress"] snapshots_sent: int class TimestampedRecord(BaseModel): - """A record with timestamp for batched data.""" + """A record with timestamp for batched data. + + .. deprecated:: 1.11.0 + Only used by :class:`WsHistoricalBatch`, which the server no longer + sends because it has discontinued bulk streaming. + """ timestamp: int data: dict[str, Any] class WsHistoricalBatch(BaseModel): - """Batch of historical data (bulk streaming).""" + """Batch of historical data (bulk streaming). + + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming and no longer sends this + message. For large dataset downloads, use the S3 Parquet bulk export + at https://0xarchive.io/data. + """ type: Literal["historical_batch"] channel: WsChannel @@ -1512,6 +1539,10 @@ class WsHistoricalBatch(BaseModel): class WsStreamCompleted(BaseModel): """Stream completed response. + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming and no longer sends this + message. + In multi-channel mode, ``channels`` lists all channels that were streamed. """ @@ -1525,7 +1556,12 @@ class WsStreamCompleted(BaseModel): class WsStreamStopped(BaseModel): - """Stream stopped response.""" + """Stream stopped response. + + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming and no longer sends this + message. + """ type: Literal["stream_stopped"] snapshots_sent: int diff --git a/oxarchive/websocket.py b/oxarchive/websocket.py index aaead0d..6aeac15 100644 --- a/oxarchive/websocket.py +++ b/oxarchive/websocket.py @@ -1,5 +1,10 @@ """ -WebSocket client for 0xarchive real-time streaming, replay, and bulk download. +WebSocket client for 0xarchive real-time streaming and historical replay. + +For large dataset downloads, use the S3 Parquet bulk export at +https://0xarchive.io/data. The server has discontinued WebSocket bulk +streaming, so the deprecated ``stream()``, ``multi_stream()`` and +``stream_stop()`` methods only return the server's error. Examples: Real-time streaming: @@ -28,13 +33,6 @@ ... ["orderbook", "trades", "funding"], "BTC", ... start=time.time()*1000 - 86400000, speed=10 ... ) - - Bulk streaming (like Databento): - >>> ws = OxArchiveWs(WsOptions(api_key="ox_...")) - >>> await ws.connect() - >>> batches = [] - >>> ws.on_batch(lambda coin, records: batches.extend(records)) - >>> await ws.stream("orderbook", "ETH", start=..., end=..., batch_size=1000) """ from __future__ import annotations @@ -42,6 +40,7 @@ import asyncio import json import logging +import warnings from dataclasses import dataclass from typing import Any, Callable, Literal, Optional, Set, Union @@ -154,6 +153,19 @@ ) +# Large dataset downloads: the S3 Parquet bulk export. +_BULK_EXPORT_URL = "https://0xarchive.io/data" + + +def _bulk_stream_deprecation(name: str) -> str: + """Deprecation message for the discontinued WebSocket bulk streaming API.""" + return ( + f"OxArchiveWs.{name}() is deprecated because the server has discontinued " + "WebSocket bulk streaming. For large dataset downloads, use the S3 Parquet " + f"bulk export at {_BULK_EXPORT_URL}." + ) + + def _validate_historical_l4_channel(channel: WsChannel) -> None: """Reject historical operations for L4 channels that are live-only.""" if channel in L4_LIVE_ONLY_CHANNELS: @@ -245,7 +257,7 @@ class WsOptions: ReplayStartHandler = Callable[[WsChannel, str, int, int, float], None] # channel, coin, start, end, speed ReplayCompleteHandler = Callable[[WsChannel, str, int], None] # channel, coin, snapshots_sent -# Stream handlers +# Bulk stream handlers (deprecated: the server has discontinued bulk streaming) BatchHandler = Callable[[str, list[TimestampedRecord]], None] StreamStartHandler = Callable[[WsChannel, str, int, int], None] # channel, coin, start, end StreamProgressHandler = Callable[[int], None] # snapshots_sent @@ -534,7 +546,7 @@ def __init__(self, options: WsOptions): self._on_replay_start: Optional[ReplayStartHandler] = None self._on_replay_complete: Optional[ReplayCompleteHandler] = None - # Stream handlers (Option D) + # Bulk stream handlers (deprecated: the server has discontinued bulk streaming) self._on_batch: Optional[BatchHandler] = None self._on_stream_start: Optional[StreamStartHandler] = None self._on_stream_progress: Optional[StreamProgressHandler] = None @@ -1023,7 +1035,7 @@ async def multi_replay( await self._send(msg) # ========================================================================= - # Bulk Streaming (Option D) - Like Databento + # Bulk Streaming (deprecated: the server has discontinued it) # ========================================================================= async def stream( @@ -1036,7 +1048,17 @@ async def stream( granularity: Optional[str] = None, interval: Optional[str] = None, ) -> None: - """Start bulk streaming for fast data download. + """Request a bulk stream. Deprecated: the server has discontinued bulk streaming. + + .. deprecated:: 1.11.0 + The server no longer serves WebSocket bulk streaming. This method + still sends the request, and the server answers with a + :class:`~oxarchive.types.WsError` delivered to the ``on_message`` + handler; no data is streamed. For large dataset downloads, use the + S3 Parquet bulk export at https://0xarchive.io/data. For timed + playback of history, use :meth:`replay`. + + Calling it emits a :class:`DeprecationWarning`. Args: channel: Data channel to stream @@ -1046,11 +1068,8 @@ async def stream( batch_size: Records per batch message granularity: Data resolution for Lighter orderbook ('checkpoint', '30s', '10s', '1s', 'tick') interval: Candle interval for candles channel ('1m', '5m', '15m', '30m', '1h', '4h', '1d', '1w') - - Example: - >>> await ws.stream("orderbook", "ETH", start=..., end=..., batch_size=1000) - >>> await ws.stream("candles", "BTC", start=..., end=..., interval="1h") """ + warnings.warn(_bulk_stream_deprecation("stream"), DeprecationWarning, stacklevel=2) _validate_historical_l4_channel(channel) msg = { "op": "stream", @@ -1067,7 +1086,18 @@ async def stream( await self._send(msg) async def stream_stop(self) -> None: - """Stop the current bulk stream.""" + """Stop a bulk stream. Deprecated: the server has discontinued bulk streaming. + + .. deprecated:: 1.11.0 + No bulk stream can be running, so there is nothing to stop. This + method still sends the request, and the server answers with a + :class:`~oxarchive.types.WsError` delivered to the ``on_message`` + handler. For large dataset downloads, use the S3 Parquet bulk + export at https://0xarchive.io/data. + + Calling it emits a :class:`DeprecationWarning`. + """ + warnings.warn(_bulk_stream_deprecation("stream_stop"), DeprecationWarning, stacklevel=2) await self._send({"op": "stream.stop"}) async def multi_stream( @@ -1078,12 +1108,17 @@ async def multi_stream( end: int, batch_size: int = 1000, ) -> None: - """Start a multi-channel bulk stream for fast data download. + """Request a multi-channel bulk stream. Deprecated: the server has discontinued it. + + .. deprecated:: 1.11.0 + The server no longer serves WebSocket bulk streaming. This method + still sends the request, and the server answers with a + :class:`~oxarchive.types.WsError` delivered to the ``on_message`` + handler; no data is streamed. For large dataset downloads, use the + S3 Parquet bulk export at https://0xarchive.io/data. For timed + playback of several channels together, use :meth:`multi_replay`. - All channels are streamed together in a single interleaved timeline. - Data arrives in batches without timing delays. The ``channel`` field on - each ``historical_batch`` message indicates which channel the batch - belongs to. + Calling it emits a :class:`DeprecationWarning`. Args: channels: List of channels to stream together (e.g., @@ -1092,16 +1127,8 @@ async def multi_stream( start: Start timestamp (Unix ms). end: End timestamp (Unix ms). batch_size: Records per batch message. - - Example: - >>> await ws.multi_stream( - ... ["orderbook", "trades", "funding"], - ... "BTC", - ... start=int(time.time() * 1000) - 3600000, - ... end=int(time.time() * 1000), - ... batch_size=1000, - ... ) """ + warnings.warn(_bulk_stream_deprecation("multi_stream"), DeprecationWarning, stacklevel=2) for channel in channels: _validate_historical_l4_channel(channel) @@ -1265,38 +1292,66 @@ def on_replay_complete(self, handler: ReplayCompleteHandler) -> None: """ self._on_replay_complete = handler - # Stream event handlers (Option D) + # Bulk stream event handlers (deprecated: the server has discontinued bulk streaming) def on_batch(self, handler: BatchHandler) -> None: """Set handler for batched data (bulk stream mode). + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming, so this handler is + never called. For large dataset downloads, use the S3 Parquet bulk + export at https://0xarchive.io/data. Calling it emits a + :class:`DeprecationWarning`. + Handler receives: (coin, records) where records is list of TimestampedRecord """ + warnings.warn(_bulk_stream_deprecation("on_batch"), DeprecationWarning, stacklevel=2) self._on_batch = handler def on_stream_start(self, handler: StreamStartHandler) -> None: """Set handler for stream started event. + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming, so this handler is + never called. Calling it emits a :class:`DeprecationWarning`. + Handler receives: (channel, coin, start, end) """ + warnings.warn( + _bulk_stream_deprecation("on_stream_start"), DeprecationWarning, stacklevel=2 + ) self._on_stream_start = handler def on_stream_progress(self, handler: StreamProgressHandler) -> None: """Set handler for stream progress event. + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming, so this handler is + never called. Calling it emits a :class:`DeprecationWarning`. + Handler receives: (snapshots_sent) """ + warnings.warn( + _bulk_stream_deprecation("on_stream_progress"), DeprecationWarning, stacklevel=2 + ) self._on_stream_progress = handler def on_stream_complete(self, handler: StreamCompleteHandler) -> None: """Set handler for stream completed event. + .. deprecated:: 1.11.0 + The server has discontinued bulk streaming, so this handler is + never called. Calling it emits a :class:`DeprecationWarning`. + Handler receives: (channel, coin, snapshots_sent) """ + warnings.warn( + _bulk_stream_deprecation("on_stream_complete"), DeprecationWarning, stacklevel=2 + ) self._on_stream_complete = handler def on_gap(self, handler: GapHandler) -> None: - """Set handler for gap detected events during replay or streaming. + """Set handler for gap detected events during replay. Called when there's a gap in the historical data exceeding the threshold. Thresholds: 2 minutes for orderbook/candles/liquidations, 60 minutes for trades. @@ -1552,7 +1607,8 @@ def _handle_message(self, raw: str) -> None: channel = data.get("channel") or (data.get("channels", [None])[0]) self._on_replay_complete(channel, data["coin"], data["snapshots_sent"]) - # Stream messages (Option D) + # Bulk stream messages (deprecated: the server has discontinued bulk + # streaming and no longer sends these; kept for compatibility) elif msg_type == "stream_started" and self._on_stream_start: channel = data.get("channel") or (data.get("channels", [None])[0]) self._on_stream_start(channel, data["coin"], data["start"], data["end"]) diff --git a/tests/test_websocket_capabilities.py b/tests/test_websocket_capabilities.py index 3f07e02..e06edb1 100644 --- a/tests/test_websocket_capabilities.py +++ b/tests/test_websocket_capabilities.py @@ -1,8 +1,9 @@ import asyncio import json import re +import warnings from datetime import datetime, timezone -from typing import Any, cast +from typing import Any, Callable, cast import pytest from websockets.protocol import State as WsState @@ -687,3 +688,104 @@ def test_l4_replay_frames_are_typed_and_batch_order_is_preserved() -> None: assert messages[1].data[1]["seq"] == 2 assert snapshots[0]["last_block_number"] == 100 assert [record["seq"] for record in batches[0]] == [1, 2] + + +# The server has discontinued WebSocket bulk streaming. The bulk stream methods +# stay for compatibility, warn on use, and still send the request, which the +# server answers with this error. +SERVER_STREAM_ERROR = ( + "Bulk streaming has been discontinued. For large dataset downloads, use our S3 " + "Parquet bulk export at https://0xarchive.io/data or contact enterprise@0xarchive.io." +) + + +def _offline_client() -> tuple[OxArchiveWs, list[dict[str, Any]]]: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + sent: list[dict[str, Any]] = [] + + async def fake_send(message: dict[str, Any]) -> None: + sent.append(message) + + setattr(ws, "_send", fake_send) + return ws, sent + + +def test_stream_emits_a_deprecation_warning_and_still_sends_the_request() -> None: + ws, sent = _offline_client() + + with pytest.warns(DeprecationWarning, match=r"stream\(\) is deprecated") as record: + asyncio.run(ws.stream("orderbook", "BTC", start=1_757_000_000_000, end=1_757_003_600_000)) + + message = str(record[0].message) + assert "discontinued" in message + assert "S3 Parquet bulk export" in message + assert "https://0xarchive.io/data" in message + assert sent == [ + { + "op": "stream", + "channel": "orderbook", + "symbol": "BTC", + "start": 1_757_000_000_000, + "end": 1_757_003_600_000, + "batch_size": 1000, + } + ] + + +def test_stream_deprecation_warning_points_at_the_caller() -> None: + ws, _ = _offline_client() + + async def caller() -> None: + await ws.stream("trades", "ETH", start=1_757_000_000_000, end=1_757_003_600_000) + + with pytest.warns(DeprecationWarning) as record: + asyncio.run(caller()) + + assert record[0].filename == __file__ + + +def test_multi_stream_and_stream_stop_emit_deprecation_warnings() -> None: + ws, sent = _offline_client() + + with pytest.warns(DeprecationWarning, match=r"multi_stream\(\) is deprecated"): + asyncio.run( + ws.multi_stream( + ["orderbook", "trades"], "BTC", start=1_757_000_000_000, end=1_757_003_600_000 + ) + ) + with pytest.warns(DeprecationWarning, match=r"stream_stop\(\) is deprecated"): + asyncio.run(ws.stream_stop()) + + assert [message["op"] for message in sent] == ["stream", "stream.stop"] + + +@pytest.mark.parametrize( + "setter", + ["on_batch", "on_stream_start", "on_stream_progress", "on_stream_complete"], +) +def test_bulk_stream_handler_setters_emit_deprecation_warnings(setter: str) -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + set_handler: Callable[[Callable[..., None]], None] = getattr(ws, setter) + + with pytest.warns(DeprecationWarning, match=rf"{setter}\(\) is deprecated"): + set_handler(lambda *args: None) + + +def test_replay_and_live_handlers_do_not_emit_deprecation_warnings() -> None: + ws, _ = _offline_client() + + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + asyncio.run(ws.replay("orderbook", "BTC", start=1_757_000_000_000)) + ws.on_orderbook(lambda coin, book: None) + + +def test_the_bulk_stream_rejection_reaches_on_message_as_an_error() -> None: + ws = OxArchiveWs(WsOptions(api_key="test-key")) + messages: list[object] = [] + ws.on_message(messages.append) + + ws._handle_message(json.dumps({"type": "error", "message": SERVER_STREAM_ERROR})) + + assert isinstance(messages[0], WsError) + assert messages[0].message == SERVER_STREAM_ERROR