Add stream RPC client

examples/cases/stream/subscribe_to_rpc_stream.py

@@ -0,0 +1,63 @@
+import asyncio
+import logging
+from asyncio import run
+from signal import SIGINT, SIGTERM
+
+from examples.utils import BTC_USD_MARKET, create_stream_rpc_client, init_env
+from x10.clients.streamrpc.subscription_params import (
+    CandlesParams,
+    PricesParams,
+    TradesParams,
+)
+from x10.config import get_config_by_name
+from x10.models.stream_rpc import StreamRpcResponseModel
+
+LOGGER = logging.getLogger()
+MARKET_NAME = BTC_USD_MARKET
+
+
+def on_message(message: StreamRpcResponseModel) -> None:
+    LOGGER.info("Received message: %s", message)
+
+
+async def subscribe_to_rpc_stream(stop_event: asyncio.Event):
+    env_config = init_env()
+    client_config = get_config_by_name(env_config.client_config_name)
+
+    async with create_stream_rpc_client(client_config) as client:
+        await client.ping()
+
+        subscriptions_before = await client.list_subscriptions()
+
+        LOGGER.info("Active subscriptions: %s", subscriptions_before)
+
+        await client.subscribe(params=TradesParams(market="BTC-USD"), handler=on_message)
+        await client.subscribe(params=TradesParams(market="ETH-USD"), handler=on_message)
+        await client.subscribe(params=PricesParams(price_type="index", market="ETH-USD"), handler=on_message)
+        await client.subscribe(
+            params=CandlesParams(candle_type="index", market="ETH-USD", interval="PT1M"), handler=on_message
+        )
+
+        subscriptions_after = await client.list_subscriptions()
+
+        LOGGER.info("Active subscriptions: %s", subscriptions_after)
+
+        await stop_event.wait()
+
+
+async def run_example():
+    stop_event = asyncio.Event()
+    loop = asyncio.get_running_loop()
+
+    def signal_handler():
+        LOGGER.info("Signal received, stopping...")
+        stop_event.set()
+
+    loop.add_signal_handler(SIGINT, signal_handler)
+    loop.add_signal_handler(SIGTERM, signal_handler)
+
+    await subscribe_to_rpc_stream(stop_event)
+
+
+if __name__ == "__main__":
+    run(main=run_example())

examples/utils.py

@@ -11,6 +11,7 @@
 from x10.clients.blocking import BlockingTradingClient
 from x10.clients.rest import RestApiClient
 from x10.clients.stream import StreamClient
+from x10.clients.streamrpc.streamrpc_client import StreamRpcClient
 from x10.config import get_config_by_name
 from x10.core.client_config import ClientConfig
 from x10.core.env_config import EnvConfig
@@ -71,6 +72,10 @@ def create_stream_client(config: ClientConfig):
     return StreamClient(api_url=config.endpoints.stream_url)
 
 
+def create_stream_rpc_client(config: ClientConfig):
+    return StreamRpcClient(api_url=config.endpoints.stream_rpc_url)
+
+
 def get_adjust_price_by_pct(config: TradingConfigModel):
     def adjust_price_by_pct(price: Decimal, pct: Decimal | int):
         return config.round_price(price + price * Decimal(pct) / 100)

tests/clients/test_rest_api_client.py

@@ -51,8 +51,8 @@ async def test_get_markets(aiohttp_server, create_btc_usd_market):
                 "collateralAssetName": "USD",
                 "collateralAssetPrecision": 6,
                 "active": True,
-                "isRfq": True,
-                "isOffHours": True,
+                "isRfq": False,
+                "isOffHours": False,
                 "marketStats": {
                     "dailyVolume": "2410800.768021",
                     "dailyVolumeBase": "37.94502",

tests/clients/test_streamrpc_client.py

@@ -0,0 +1,84 @@
+import asyncio
+import json
+
+import pytest
+import websockets
+from hamcrest import assert_that, equal_to
+from websockets import WebSocketServer
+
+
+def get_url_from_server(server: WebSocketServer):
+    host, port = server.sockets[0].getsockname()  # type: ignore[index]
+    return f"ws://{host}:{port}"
+
+
+@pytest.mark.asyncio
+async def test_candle_stream():
+    from tests.fixtures.candle import create_candle_stream_rpc_message
+    from x10.clients.streamrpc.streamrpc_client import StreamRpcClient
+    from x10.clients.streamrpc.subscription_params import CandlesParams
+
+    message_model = create_candle_stream_rpc_message()
+    received_messages: asyncio.Queue = asyncio.Queue()
+
+    async def subscription_handler(msg):
+        await received_messages.put(msg)
+
+    async def mock_server(websocket):
+        subscribe_msg_raw = await websocket.recv()
+        subscribe_msg = json.loads(subscribe_msg_raw)
+
+        assert_that(subscribe_msg["method"], equal_to("subscribe"))
+
+        await websocket.send(
+            json.dumps(
+                {
+                    "id": subscribe_msg["id"],
+                    "result": {"subscription": message_model.subscription},
+                }
+            )
+        )
+
+        await websocket.send(json.dumps(message_model.to_api_request_json()))
+
+        unsubscribe_msg_raw = await websocket.recv()
+        unsubscribe_msg = json.loads(unsubscribe_msg_raw)
+
+        assert_that(unsubscribe_msg["method"], equal_to("unsubscribe"))
+
+        await websocket.send(
+            json.dumps(
+                {
+                    "id": unsubscribe_msg["id"],
+                    "result": {"method": "unsubscribe", "status": "OK"},
+                }
+            )
+        )
+
+    async with websockets.serve(mock_server, "127.0.0.1", 0) as server:
+        client = StreamRpcClient(api_url=get_url_from_server(server))
+        await client.connect()
+
+        subscription_params = CandlesParams(candle_type="last", market="BTC-USD", interval="PT1M")
+        subscription_id = await client.subscribe(params=subscription_params, handler=subscription_handler)
+
+        msg = await asyncio.wait_for(received_messages.get(), timeout=5)
+
+        await client.unsubscribe(subscription_id)
+        await client.close()
+
+    assert_that(
+        msg.to_api_request_json(),
+        equal_to(
+            {
+                "type": "CANDLES",
+                "data": [
+                    {"o": "3458.64", "l": "3399.07", "h": "3476.89", "c": "3414.85", "v": "3.938", "T": 1721106000000}
+                ],
+                "error": None,
+                "ts": 1721283121979,
+                "seq": 1,
+                "subscription": "candles.last.BTC-USD.PT1M",
+            }
+        ),
+    )

tests/fixtures/candle.py

@@ -3,6 +3,7 @@
 
 from x10.models.candle import CandleModel
 from x10.models.http import WrappedStreamResponseModel
+from x10.models.stream_rpc import StreamRpcResponseModel
 
 
 def create_candle_stream_message():
@@ -20,3 +21,22 @@ def create_candle_stream_message():
         ts=1721283121979,
         seq=1,
     )
+
+
+def create_candle_stream_rpc_message():
+    return StreamRpcResponseModel(
+        type="CANDLES",
+        data=[
+            CandleModel(
+                open=Decimal("3458.64"),
+                low=Decimal("3399.07"),
+                high=Decimal("3476.89"),
+                close=Decimal("3414.85"),
+                volume=Decimal("3.938"),
+                timestamp=1721106000000,
+            )
+        ],
+        ts=1721283121979,
+        seq=1,
+        subscription="candles.last.BTC-USD.PT1M",
+    )

tests/fixtures/market.py

@@ -15,8 +15,8 @@ def get_btc_usd_market_json_data():
                 "collateralAssetName": "USD",
                 "collateralAssetPrecision": 6,
                 "active": true,
-                "isRfq": true,
-                "isOffHours": true,
+                "isRfq": false,
+                "isOffHours": false,
                 "marketStats": {
                     "dailyVolume": "2410800.768021",
                     "dailyVolumeBase": "37.94502",