Fruther generalize json_rpc hook mechanic to allow for multi hook, Add new maybe_open_ticker_feed to stream greeks, iv, open interest of an instrument

piker/brokers/deribit/api.py

@@ -152,7 +152,7 @@ class Client:
     def __init__(
         self,
         json_rpc: Callable,
-        update_hooks: Callable,
+        append_hooks: Callable,
         update_types: Callable,
     ) -> None:
 
@@ -169,7 +169,7 @@ class Client:
             self._key_secret = None
 
         self.json_rpc = json_rpc
-        self.update_hooks = update_hooks
+        self.append_hooks = append_hooks
         self.update_types = update_types
 
     @property
@@ -490,6 +490,7 @@ async def open_price_feed(
                     }]
                 }
             ))
+            return True
 
         elif chan == book_chan:
             bid, bsize = data['bids'][0]
@@ -504,11 +505,14 @@ async def open_price_feed(
                     {'type': 'asize', 'price': ask, 'size': asize}
                 ]}
             ))
+            return True
+
+        return False
 
     async with open_cached_client('deribit') as client:
 
-        client.update_hooks({
+        client.append_hooks({
-            'request': sub_hook
+            'request': [sub_hook]
         })
         client.update_types({
             'request': JSONRPCSubRequest
@@ -517,12 +521,15 @@ async def open_price_feed(
         resp = await client.json_rpc(
             'private/subscribe', {'channels': channels})
 
-        assert resp.result == channels
+        assert not resp.error
 
         log.info(f'Subscribed to {channels}')
 
         yield recv_chann
 
+        resp = await client.json_rpc('private/unsubscribe', {'channels': channels})
+
+        assert not resp.error
 
 @acm
 async def maybe_open_price_feed(
@@ -543,71 +550,64 @@ async def maybe_open_price_feed(
             yield feed
 
 
-# TODO: order broker support: this is all draft code from @guilledk B)
+@acm
+async def open_ticker_feed(
+    instrument: str
+) -> trio.abc.ReceiveStream:
 
-# async def aio_order_feed_relay(
+    instrument_db = sym_fmt_piker_to_deribit(instrument)
-#     fh: FeedHandler,
-#     instrument: Symbol,
-#     from_trio: asyncio.Queue,
-#     to_trio: trio.abc.SendChannel,
 
-# ) -> None:
+    ticker_chan = f'incremental_ticker.{instrument_db}'
-#     async def _fill(data: dict, receipt_timestamp):
-#         breakpoint()
 
-#     async def _order_info(data: dict, receipt_timestamp):
+    channels = [ticker_chan]
-#         breakpoint()
 
-#     fh.add_feed(
+    send_chann, recv_chann = trio.open_memory_channel(0)
-#         DERIBIT,
+    async def sub_hook(msg):
-#         channels=[FILLS, ORDER_INFO],
+        chann = msg.params['channel']
-#         symbols=[instrument.upper()],
+        if chann == ticker_chan:
-#         callbacks={
+            data = msg.params['data']
-#             FILLS: _fill,
+            await send_chann.send((
-#             ORDER_INFO: _order_info,
+                'ticker', {
-#         })
+                    'symbol': instrument,
+                    'data': data
+                }
+            ))
+            return True
 
-#     if not fh.running:
+        return False
-#         fh.run(
-#             start_loop=False,
-#             install_signal_handlers=False)
 
-#     # sync with trio
+    async with open_cached_client('deribit') as client:
-#     to_trio.send_nowait(None)
 
-#     await asyncio.sleep(float('inf'))
+        client.append_hooks({
+            'request': [sub_hook]
+        })
 
+        resp = await client.json_rpc(
+            'private/subscribe', {'channels': channels})
 
-# @acm
+        assert not resp.error
-# async def open_order_feed(
-#     instrument: list[str]
-# ) -> trio.abc.ReceiveStream:
-#     async with maybe_open_feed_handler() as fh:
-#         async with to_asyncio.open_channel_from(
-#             partial(
-#                 aio_order_feed_relay,
-#                 fh,
-#                 instrument
-#             )
-#         ) as (first, chan):
-#             yield chan
 
+        log.info(f'Subscribed to {channels}')
 
-# @acm
+        yield recv_chann
-# async def maybe_open_order_feed(
-#     instrument: str
-# ) -> trio.abc.ReceiveStream:
 
-#     # TODO: add a predicate to maybe_open_context
+        resp = await client.json_rpc('private/unsubscribe', {'channels': channels})
-#     async with maybe_open_context(
+
-#         acm_func=open_order_feed,
+        assert not resp.error
-#         kwargs={
+
-#             'instrument': instrument,
+@acm
-#             'fh': fh
+async def maybe_open_ticker_feed(
-#         },
+    instrument: str
-#         key=f'{instrument}-order',
+) -> trio.abc.ReceiveStream:
-#     ) as (cache_hit, feed):
+
-#         if cache_hit:
+    async with maybe_open_context(
-#             yield broadcast_receiver(feed, 10)
+        acm_func=open_ticker_feed,
-#         else:
+        kwargs={
-#             yield feed
+            'instrument': instrument
+        },
+        key=f'{instrument}-ticker',
+    ) as (cache_hit, feed):
+        if cache_hit:
+            yield broadcast_receiver(feed, 10)
+        else:
+            yield feed

piker/data/_web_bs.py

@@ -209,9 +209,17 @@ async def open_jsonrpc_session(
         raise ValueError(
             'Need to path both a request_type and request_hook')
 
+    req_hooks = []
+    if request_hook:
+        req_hooks.append(request_hook)
+
+    err_hooks = []
+    if error_hook:
+        err_hooks.append(error_hook)
+
     hook_table = {
-        'request': request_hook,
+        'request': req_hooks,
-        'error': error_hook
+        'error': err_hooks
     }
 
     types_table = {
@@ -219,9 +227,10 @@ async def open_jsonrpc_session(
         'request': request_type
     }
 
-    def update_hooks(new_hooks: dict):
+    def append_hooks(new_hooks: dict):
         nonlocal hook_table
-        hook_table.update(new_hooks)
+        for htype, hooks in new_hooks.items():
+            hook_table[htype] += hooks
 
     def update_types(new_types: dict):
         nonlocal types_table
@@ -234,7 +243,7 @@ async def open_jsonrpc_session(
         rpc_id: Iterable = count(start_id)
         rpc_results: dict[int, dict] = {}
 
-        async def json_rpc(method: str, params: dict) -> dict:
+        async def json_rpc(method: str, params: dict = {}) -> dict:
             '''
             perform a json rpc call and wait for the result, raise exception in
             case of error field present on response
@@ -303,19 +312,25 @@ async def open_jsonrpc_session(
                         'params': _,
                     }:
                         log.info(f'Recieved\n{msg}')
-                        if hook_table['request']:
+                        if len(hook_table['request']) > 0:
-                            await hook_table['request'](types_table['request'](**msg))
+                            for hook in hook_table['request']:
+                                result = await hook(types_table['request'](**msg))
+                                if result:
+                                    break
 
                     case {
                         'error': error,
                     }:
                         log.warning(f'Recieved\n{error}')
-                        if hook_table['error']:
+                        if len(hook_table['error']) > 0:
-                            await hook_table['error'](types_table['response'](**msg))
+                            for hook in hook_table['error']:
+                                result = await hook(types_table['response'](**msg))
+                                if result:
+                                    break
 
                     case _:
                         log.warning(f'Unhandled JSON-RPC msg!?\n{msg}')
 
         n.start_soon(recv_task)
-        yield json_rpc, update_hooks, update_types
+        yield json_rpc, append_hooks, update_types
         n.cancel_scope.cancel()