Deribit api key changes introduce:

- `get_timestamp_int`: added this is the hack, so we can aboid use the custom deribit date format.

- `get_currencies`: added so we could get all deribit's available currencies.

- Also a couple of format fixes.

piker/data/_sampling.py

@@ -29,7 +29,6 @@ import time
 from typing import (
     Any,
     AsyncIterator,
-    Callable,
     TYPE_CHECKING,
 )
 
@@ -54,9 +53,6 @@ from ._util import (
     get_console_log,
 )
 from ..service import maybe_spawn_daemon
-from piker.log import (
-    mk_repr,
-)
 
 if TYPE_CHECKING:
     from ._sharedmem import (
@@ -565,6 +561,7 @@ async def open_sample_stream(
 
 
 async def sample_and_broadcast(
+
     bus: _FeedsBus,  # noqa
     rt_shm: ShmArray,
     hist_shm: ShmArray,
@@ -585,22 +582,11 @@ async def sample_and_broadcast(
 
     overruns = Counter()
 
-    # multiline nested `dict` formatter (since rn quote-msgs are
-    # just that).
-    pfmt: Callable[[str], str] = mk_repr()
-
     # iterate stream delivered by broker
     async for quotes in quote_stream:
+        # print(quotes)
 
-        # XXX WARNING XXX only enable for debugging bc ow can cost
+        # TODO: ``numba`` this!
-        # ALOT of perf with HF-feedz!!!
-        #
-        # log.info(
-        #     'Rx live quotes:\n'
-        #     f'{pfmt(quotes)}'
-        # )
-
-        # TODO: `numba` this!
         for broker_symbol, quote in quotes.items():
             # TODO: in theory you can send the IPC msg *before* writing
             # to the sharedmem array to decrease latency, however, that
@@ -673,18 +659,6 @@ async def sample_and_broadcast(
             sub_key: str = broker_symbol.lower()
             subs: set[Sub] = bus.get_subs(sub_key)
 
-            if not subs:
-                all_bs_fqmes: list[str] = list(
-                    bus._subscribers.keys()
-                )
-                log.warning(
-                    f'No subscribers for {brokername!r} live-quote ??\n'
-                    f'broker_symbol: {broker_symbol}\n\n'
-
-                    f'Maybe the backend-sys symbol does not match one of,\n'
-                    f'{pfmt(all_bs_fqmes)}\n'
-                )
-
             # NOTE: by default the broker backend doesn't append
             # it's own "name" into the fqme schema (but maybe it
             # should?) so we have to manually generate the correct

piker/data/_web_bs.py

@@ -359,8 +359,8 @@ async def open_autorecon_ws(
 
 
 '''
-JSONRPC response-request style machinery for transparent multiplexing
+JSONRPC response-request style machinery for transparent multiplexing of msgs
-of msgs over a `NoBsWs`.
+over a NoBsWs.
 
 '''
 
@@ -377,35 +377,19 @@ async def open_jsonrpc_session(
     url: str,
     start_id: int = 0,
     response_type: type = JSONRPCResult,
-    msg_recv_timeout: float = float('inf'),
+    request_type: Optional[type] = None,
-    # ^NOTE, since only `deribit` is using this jsonrpc stuff atm
+    request_hook: Optional[Callable] = None,
-    # and options mkts are generally "slow moving"..
+    error_hook: Optional[Callable] = None,
-    #
-    # FURTHER if we break the underlying ws connection then since we
-    # don't pass a `fixture` to the task that manages `NoBsWs`, i.e.
-    # `_reconnect_forever()`, the jsonrpc "transport pipe" get's
-    # broken and never restored with wtv init sequence is required to
-    # re-establish a working req-resp session.
-
-    # request_type: Optional[type] = None,
-    # request_hook: Optional[Callable] = None,
-    # error_hook: Optional[Callable] = None,
 ) -> Callable[[str, dict], dict]:
 
     async with (
         trio.open_nursery() as n,
-        open_autorecon_ws(
+        open_autorecon_ws(url) as ws
-            url=url,
-            msg_recv_timeout=msg_recv_timeout,
-        ) as ws
     ):
         rpc_id: Iterable = count(start_id)
         rpc_results: dict[int, dict] = {}
 
-        async def json_rpc(
+        async def json_rpc(method: str, params: dict) -> dict:
-            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

piker/data/feed.py

@@ -540,10 +540,7 @@ async def open_feed_bus(
         # subscription since the backend isn't (yet) expected to
         # append it's own name to the fqme, so we filter on keys
         # which *do not* include that name (e.g .ib) .
-        bus._subscribers.setdefault(
+        bus._subscribers.setdefault(bs_fqme, set())
-            bs_fqme,
-            set(),
-        )
 
     # sync feed subscribers with flume handles
     await ctx.started(

piker/log.py

@@ -18,11 +18,7 @@
 Log like a forester!
 """
 import logging
-import reprlib
 import json
-from typing import (
-    Callable,
-)
 
 import tractor
 from pygments import (
@@ -88,27 +84,3 @@ def colorize_json(
         # likeable styles: algol_nu, tango, monokai
         formatters.TerminalTrueColorFormatter(style=style)
     )
-
-
-def mk_repr(
-    **repr_kws,
-) -> Callable[[str], str]:
-    '''
-    Allocate and deliver a `repr.Repr` instance with provided input
-    settings using the std-lib's `reprlib` mod,
-     * https://docs.python.org/3/library/reprlib.html
-
-    ------ Ex. ------
-    An up to 6-layer-nested `dict` as multi-line:
-    - https://stackoverflow.com/a/79102479
-    - https://docs.python.org/3/library/reprlib.html#reprlib.Repr.maxlevel
-
-    '''
-    def_kws: dict[str, int] = dict(
-        indent=2,
-        maxlevel=6,  # recursion levels
-        maxstring=66,  # match editor line-len limit
-    )
-    def_kws |= repr_kws
-    reprr = reprlib.Repr(**def_kws)
-    return reprr.repr

piker/storage/nativedb.py

@@ -161,12 +161,7 @@ class NativeStorageClient:
 
     def index_files(self):
         for path in self._datadir.iterdir():
-            if (
+            if path.name in {'borked', 'expired',}:
-                path.name in {'borked', 'expired',}
-                or
-                '.parquet' not in str(path)
-            ):
-                # ignore all non-apache files (for now)
                 continue
 
             key: str = path.name.rstrip('.parquet')

piker/tsp/__init__.py

@@ -434,32 +434,21 @@ async def start_backfill(
                 # - some other unknown error (ib blocking the
                 #   history bc they don't want you seeing how they
                 #   cucked all the tinas..)
-                if (
+                if dur := frame_types.get(timeframe):
-                    frame_types
+                    # decrement by a frame's worth of duration and
-                    and
+                    # retry a few times.
-                    (dur := frame_types.get(timeframe))
+                    last_start_dt.subtract(
-                ):
-                    # decrement by a duration's (frame) worth of time
-                    # as maybe indicated by the backend to see if we
-                    # can get older data before this possible
-                    # "history gap".
-                    orig_last_start_dt = last_start_dt
-                    last_start_dt = last_start_dt.subtract(
                         seconds=dur.total_seconds()
                     )
                     log.warning(
                         f'{mod.name} -> EMPTY FRAME for end_dt?\n'
                         f'tf@fqme: {timeframe}@{mkt.fqme}\n'
-                        f'Decrementing `end_dt` by {dur} and retry..\n\n'
+                        'bf_until <- last_start_dt:\n'
-
+                        f'{backfill_until_dt} <- {last_start_dt}\n'
-                        f'orig_last_start_dt: {orig_last_start_dt}\n'
+                        f'Decrementing `end_dt` by {dur} and retry..\n'
-                        f'dur subtracted last_start_dt: {last_start_dt}\n'
-                        f'bf_until: {backfill_until_dt}\n'
                     )
                     continue
 
-                raise
-
             # broker says there never was or is no more history to pull
             except DataUnavailable:
                 log.warning(