piker/brokers/deribit/api.py

@@ -28,8 +28,6 @@ from decimal import (
     Decimal,
 )
 from functools import partial
-from pathlib import Path
-from pprint import pformat
 import time
 from typing import (
     Any,
@@ -39,6 +37,8 @@ from typing import (
 
 from pendulum import now
 import trio
+from trio_typing import TaskStatus
+from rapidfuzz import process as fuzzy
 import numpy as np
 from tractor.trionics import (
     broadcast_receiver,
@@ -52,16 +52,11 @@ from cryptofeed import FeedHandler
 from cryptofeed.defines import (
     DERIBIT,
     L1_BOOK, TRADES,
-    OPTION, CALL, PUT,
+    OPTION, CALL, PUT
-    OPEN_INTEREST,
 )
 from cryptofeed.symbols import Symbol
-from cryptofeed.types import (
+# types for managing the cb callbacks.
-    L1Book,
+# from cryptofeed.types import L1Book
-    Trade,
-    OpenInterest,
-)
-from piker.brokers import SymbolNotFound
 from .venues import (
     _ws_url,
     MarketType,
@@ -69,7 +64,9 @@ from .venues import (
     Pair,
     OptionPair,
     JSONRPCResult,
+    JSONRPCChannel,
     KLinesResult,
+    Trade,
     LastTradesResult,
 )
 from piker.accounting import (
@@ -80,7 +77,7 @@ from piker.accounting import (
 from piker.data import (
     def_iohlcv_fields,
     match_from_pairs,
-    # Struct,
+    Struct,
 )
 from piker.data._web_bs import (
     open_jsonrpc_session
@@ -99,21 +96,9 @@ _spawn_kwargs = {
 }
 
 
-def deribit_timestamp(when: datetime) -> int:
+# convert datetime obj timestamp to unixtime in milliseconds
-    '''
+def deribit_timestamp(when):
-    Convert conventional epoch timestamp, in secs, to unixtime in
+    return int((when.timestamp() * 1000) + (when.microsecond / 1000))
-    milliseconds.
-
-    '''
-    return int(
-        (when.timestamp() * 1000)
-        +
-        (when.microsecond / 1000)
-    )
-
-
-def get_timestamp_int(expiry_date: str) -> int:
-    return int(time.mktime(time.strptime(expiry_date, '%d%b%y')))
 
 
 def str_to_cb_sym(name: str) -> Symbol:
@@ -122,40 +107,32 @@ def str_to_cb_sym(name: str) -> Symbol:
     quote = base
 
     if option_type == 'put':
-        option_type = PUT
+        option_type = PUT 
-    elif option_type == 'call':
+    elif option_type  == 'call':
         option_type = CALL
     else:
         raise Exception("Couldn\'t parse option type")
 
-    new_expiry_date: int = get_timestamp_int(
+    new_expiry_date = get_values_from_cb_normalized_date(expiry_date)
-        get_values_from_cb_normalized_date(expiry_date)
+
-    )
     return Symbol(
         base=base,
         quote=quote,
         type=OPTION,
         strike_price=strike_price,
         option_type=option_type,
-        expiry_date=new_expiry_date
+        expiry_date=new_expiry_date)
-    )
 
 
 def piker_sym_to_cb_sym(name: str) -> Symbol:
-    (
+    base, expiry_date, strike_price, option_type = tuple(
-        base,
-        expiry_date,
-        strike_price,
-        option_type,
-    )= tuple(
         name.upper().split('-'))
 
-    new_expiry_date = get_timestamp_int(expiry_date)
+    quote = base
-    quote: str = base
 
-    if option_type == 'P' or option_type == 'PUT':
+    if option_type == 'P':
-        option_type = PUT
+        option_type = PUT 
-    elif option_type == 'C' or option_type == 'CALL':
+    elif option_type == 'C':
         option_type = CALL
     else:
         raise Exception("Couldn\'t parse option type")
@@ -166,32 +143,14 @@ def piker_sym_to_cb_sym(name: str) -> Symbol:
         type=OPTION,
         strike_price=strike_price,
         option_type=option_type,
-        expiry_date=new_expiry_date
+        expiry_date=expiry_date)
-    )
 
 
-# TODO, instead can't we just lookup the `MktPair` directly
+def cb_sym_to_deribit_inst(sym: Symbol):
-# and pass it upward to `stream_quotes()`??
-def cb_sym_to_deribit_inst(sym: Symbol) -> str:
-    '''
-    Generate our own internal `str`-repr for a `cryptofeed.Symbol`
-    uniquely from its fields.
-
-    This is the equiv of generating a `Pair.fmqe` from `cryptofeed`
-    for now i suppose..?
-
-    '''
     new_expiry_date = get_values_from_cb_normalized_date(sym.expiry_date)
-    otype = (
+    otype = 'C' if sym.option_type == CALL else 'P'
-        'C' if sym.option_type == CALL
+
-        else 'P'
+    return f'{sym.base}-{new_expiry_date}-{sym.strike_price}-{otype}'
-    )
-    return (
-        f'{sym.base}-'
-        f'{new_expiry_date}-'
-        f'{sym.strike_price}-'
-        f'{otype}'
-    )
 
 
 def get_values_from_cb_normalized_date(expiry_date: str) -> str:
@@ -220,39 +179,32 @@ def get_config() -> dict[str, Any]:
 
     conf: dict
     path: Path
+
     conf, path = config.load(
         conf_name='brokers',
         touch_if_dne=True,
     )
-    section: dict|None = conf.get('deribit')
+    section: dict = {}
+    section = conf.get('deribit')
     if section is None:
-        raise ValueError(
+        log.warning(f'No config section found for deribit in {path}')
-            f'No `[deribit]` section found in\n'
+        return {}
-            f'{path!r}\n\n'
-            f'See the template config from the core repo for samples..\n'
-            # f'<TODO put repo link here??>'
-        )
 
     conf_option = section.get('option', {})
-    conf_log = conf_option.get('log', {})
+    section.clear # clear the dict to reuse it
-    return {
+    section['deribit'] = {}
-        'deribit': {
+    section['deribit']['key_id'] = conf_option.get('api_key')
-            'key_id': conf_option['key_id'],
+    section['deribit']['key_secret'] = conf_option.get('api_secret')
-            'key_secret': conf_option['key_secret'],
+
-        },
+    section['log'] = {}
-        'log': {
+    section['log']['filename'] = 'feedhandler.log'
-            'filename': conf_log['filename'],
+    section['log']['level'] = 'DEBUG'
-            'level': conf_log['level'],
+
-            'disabled': conf_log['disabled'],
+    return section
-        }
-    }
 
 
 class Client:
-    '''
-    Hi-level interface for the jsron-RPC over websocket API.
 
-    '''
     def __init__(
         self,
 
@@ -271,12 +223,8 @@ class Client:
         self._auth_ts = None
         self._auth_renew_ts = 5 # seconds to renew auth
 
-    async def _json_rpc_auth_wrapper(
+    async def _json_rpc_auth_wrapper(self, *args, **kwargs) -> JSONRPCResult:
-        self,
+        
-        *args,
-        **kwargs,
-    ) -> JSONRPCResult:
-
         """Background task that adquires a first access token and then will
         refresh the access token.
 
@@ -302,6 +250,9 @@ class Client:
 
         return await self.json_rpc(*args, **kwargs)
 
+
+
+
     async def get_balances(
         self,
         kind: str = 'option'
@@ -321,44 +272,28 @@ class Client:
 
         return balances
 
-    async def get_currencies(
-        self,
-
-    ) -> list[dict]:
-        '''
-        Return the set of currencies for deribit.
-        '''
-        assets = {}
-        resp = await self._json_rpc_auth_wrapper(
-            'public/get_currencies',
-            params={}
-        )
-        return resp.result
-
     async def get_assets(
         self,
         venue: str | None = None,
 
     ) -> dict[str, Asset]:
-        '''
+        """Return the set of asset balances for this account
-        Return the set of asset balances for this account
+        by symbol.
-        by (deribit's) symbol.
+        """
-
-
-        '''
         assets = {}
-        currencies = await self.get_currencies()
+        resp = await self._json_rpc_auth_wrapper(
+            'public/get_currencies',
+            params={}
+        )
+        currencies = resp.result
         for currency in currencies:
-            name: str = currency['currency']
+            name = currency['currency']
-            tx_tick: Decimal = digits_to_dec(currency['fee_precision'])
+            tx_tick = digits_to_dec(currency['fee_precision']) 
-
+            atype='crypto_currency'
-            # TODO, handling of options, futures, perps etc. more
-            # specifically with diff `.atype`s?
             assets[name] = Asset(
                 name=name,
-                atype='crypto_currency',
+                atype=atype,
-                tx_tick=tx_tick,
+                tx_tick=tx_tick)
-            )
 
             instruments = await self.symbol_info(currency=name)
             for instrument in instruments:
@@ -366,10 +301,9 @@ class Client:
                 assets[pair.symbol] = Asset(
                     name=pair.symbol,
                     atype=pair.venue,
-                    tx_tick=pair.size_tick,
+                    tx_tick=pair.size_tick)
-                )
 
-        return assets
+        return assets 
 
     async def get_mkt_pairs(self) -> dict[str, Pair]:
         flat: dict[str, Pair] = {}
@@ -424,19 +358,6 @@ class Client:
             return cached_pair
 
         if sym:
-            opt: OptionPair|None = pair_table.get(sym)
-            if not opt:
-                closest_matches: dict[str, Pair] = match_from_pairs(
-                    pairs=pair_table,
-                    query=sym,
-                    score_cutoff=40,
-                )
-                closest_syms: list[str] = list(closest_matches.keys())
-                raise ValueError(
-                    f'No contract found for {sym!r}\n\n'
-                    f'Closest {len(closest_syms)} available contracts:\n\n'
-                    f'{pformat(closest_syms)}\n'
-                )
             return pair_table[sym]
         else:
             return self._pairs
@@ -460,7 +381,7 @@ class Client:
         params: dict[str, str] = {
             'currency': currency.upper(),
             'kind': kind,
-            'expired': expired,
+            'expired': str(expired).lower()
         }
 
         resp: JSONRPCResult = await self._json_rpc_auth_wrapper(
@@ -468,9 +389,9 @@ class Client:
             params,
         )
         # convert to symbol-keyed table
-        pair_type: Pair = PAIRTYPES[kind]
+        pair_type: Type = PAIRTYPES[kind]
         results: list[dict] | None = resp.result
-
+        
         instruments: dict[str, Pair] = {}
         for item in results:
             symbol=item['instrument_name'].lower()
@@ -506,15 +427,12 @@ class Client:
         mkt_pairs = await self.symbol_info()
 
         if not mkt_pairs:
-            raise SymbolNotFound(
+            raise SymbolNotFound(f'No market pairs found!?:\n{resp}')
-                f'No market pairs found!?:\n'
-                f'{mkt_pairs}'
-            )
 
         pairs_view_subtable: dict[str, Pair] = {}
 
         for instrument in mkt_pairs:
-            pair_type: Pair|OptionPair = PAIRTYPES[venue]
+            pair_type: Type = PAIRTYPES[venue]
 
             pair: Pair = pair_type(**mkt_pairs[instrument].to_dict())
 
@@ -562,14 +480,12 @@ class Client:
         if end_dt is None:
             end_dt = now('UTC')
 
-        _orig_start_dt = start_dt
         if start_dt is None:
             start_dt = end_dt.start_of(
-                'minute'
+                'minute').subtract(minutes=limit)
-            ).subtract(minutes=limit)
 
-        start_time: int = deribit_timestamp(start_dt)
+        start_time = deribit_timestamp(start_dt)
-        end_time: int = deribit_timestamp(end_dt)
+        end_time = deribit_timestamp(end_dt)
 
         # https://docs.deribit.com/#public-get_tradingview_chart_data
         resp = await self._json_rpc_auth_wrapper(
@@ -583,13 +499,9 @@ class Client:
 
         result = KLinesResult(**resp.result)
         new_bars: list[tuple] = []
-        # if _orig_start_dt is None:
-        # if not new_bars:
-        #     import tractor
-        #     await tractor.pause()
-
         for i in range(len(result.close)):
-            row = [
+
+            row = [ 
                 (start_time + (i * (60 * 1000))) / 1000.0,  # time
                 result.open[i],
                 result.high[i],
@@ -642,7 +554,7 @@ async def get_client(
 
 
 @acm
-async def open_feed_handler() -> FeedHandler:
+async def open_feed_handler():
     fh = FeedHandler(config=get_config())
     yield fh
     await to_asyncio.run_task(fh.stop_async)
@@ -663,37 +575,43 @@ async def aio_price_feed_relay(
     from_trio: asyncio.Queue,
     to_trio: trio.abc.SendChannel,
 ) -> None:
-    '''
+    async def _trade(data: dict, receipt_timestamp):
-    Relay price feed quotes from the `cryptofeed.FeedHandler` to
+        to_trio.send_nowait(('trade', {
-    the `piker`-side `trio.task` consumers for delivery to consumer
+            'symbol': cb_sym_to_deribit_inst(
-    sub-actors for various subsystems.
+                str_to_cb_sym(data.symbol)).lower(),
-
+            'last': data,
-    '''
+            'broker_ts': time.time(),
-    async def _trade(
+            'data': data.to_dict(),
-        trade: Trade,  # cryptofeed, NOT ours from `.venues`!
+            'receipt': receipt_timestamp
-        receipt_timestamp: int,
+        }))
-    ) -> None:
-        '''
-        Proxy-thru `cryptofeed.FeedHandler` "trades" to `piker`-side.
-
-        '''
-        to_trio.send_nowait(('trade', trade))
-
-    async def _l1(
-        book: L1Book,
-        receipt_timestamp: int,
-    ) -> None:
-        '''
-        Relay-thru "l1 book" updates.
-
-        '''
-
-        to_trio.send_nowait(('l1', book))
-
-        # TODO, make this work!
-        # -[ ] why isn't this working in `tractor.pause_from_sync()`??
-        # breakpoint()
 
+    async def _l1(data: dict, receipt_timestamp):
+        to_trio.send_nowait(('l1', {
+            'symbol': cb_sym_to_deribit_inst(
+                str_to_cb_sym(data.symbol)).lower(),
+            'ticks': [
+                {
+                    'type': 'bid',
+                    'price': float(data.bid_price),
+                    'size': float(data.bid_size)
+                },
+                {
+                    'type': 'bsize',
+                    'price': float(data.bid_price),
+                    'size': float(data.bid_size)
+                },
+                {
+                    'type': 'ask',
+                    'price': float(data.ask_price),
+                    'size': float(data.ask_size)
+                },
+                {
+                    'type': 'asize',
+                    'price': float(data.ask_price),
+                    'size': float(data.ask_size)
+                }
+            ]
+        }))
     sym: Symbol = piker_sym_to_cb_sym(instrument)
     fh.add_feed(
         DERIBIT,
@@ -707,35 +625,27 @@ async def aio_price_feed_relay(
     if not fh.running:
         fh.run(
             start_loop=False,
-            install_signal_handlers=False
+            install_signal_handlers=False)
-        )
 
     # sync with trio
     to_trio.send_nowait(None)
 
-    # run until cancelled
     await asyncio.sleep(float('inf'))
 
 
 @acm
 async def open_price_feed(
     instrument: str
-) -> to_asyncio.LinkedTaskChannel:
+) -> trio.abc.ReceiveStream:
-
+    async with maybe_open_feed_handler() as fh:
-    fh: FeedHandler
+        async with to_asyncio.open_channel_from(
-    first: None
-    chan: to_asyncio.LinkedTaskChannel
-    async with (
-        maybe_open_feed_handler() as fh,
-        to_asyncio.open_channel_from(
             partial(
                 aio_price_feed_relay,
                 fh,
                 instrument
             )
-        ) as (first, chan)
+        ) as (first, chan):
-    ):
+            yield chan
-        yield chan
 
 
 @acm
@@ -744,7 +654,6 @@ async def maybe_open_price_feed(
 ) -> trio.abc.ReceiveStream:
 
     # TODO: add a predicate to maybe_open_context
-    feed: to_asyncio.LinkedTaskChannel
     async with maybe_open_context(
         acm_func=open_price_feed,
         kwargs={
@@ -759,69 +668,68 @@ async def maybe_open_price_feed(
 
 
 
-# TODO, move all to `.broker` submod!
+async def aio_order_feed_relay(
-# async def aio_order_feed_relay(
+    fh: FeedHandler,
-#     fh: FeedHandler,
+    instrument: Symbol,
-#     instrument: Symbol,
+    from_trio: asyncio.Queue,
-#     from_trio: asyncio.Queue,
+    to_trio: trio.abc.SendChannel,
-#     to_trio: trio.abc.SendChannel,
+) -> None:
-# ) -> None:
+    async def _fill(data: dict, receipt_timestamp):
-#     async def _fill(data: dict, receipt_timestamp):
+        breakpoint()
-#         breakpoint()
 
-#     async def _order_info(data: dict, receipt_timestamp):
+    async def _order_info(data: dict, receipt_timestamp):
-#         breakpoint()
+        breakpoint()
 
-#     fh.add_feed(
+    fh.add_feed(
-#         DERIBIT,
+        DERIBIT,
-#         channels=[FILLS, ORDER_INFO],
+        channels=[FILLS, ORDER_INFO],
-#         symbols=[instrument.upper()],
+        symbols=[instrument.upper()],
-#         callbacks={
+        callbacks={
-#             FILLS: _fill,
+            FILLS: _fill,
-#             ORDER_INFO: _order_info,
+            ORDER_INFO: _order_info,
-#         })
+        })
 
-#     if not fh.running:
+    if not fh.running:
-#         fh.run(
+        fh.run(
-#             start_loop=False,
+            start_loop=False,
-#             install_signal_handlers=False)
+            install_signal_handlers=False)
 
-#     # sync with trio
+    # sync with trio
-#     to_trio.send_nowait(None)
+    to_trio.send_nowait(None)
 
-#     await asyncio.sleep(float('inf'))
+    await asyncio.sleep(float('inf'))
 
 
-# @acm
+@acm
-# async def open_order_feed(
+async def open_order_feed(
-#     instrument: list[str]
+    instrument: list[str]
-# ) -> trio.abc.ReceiveStream:
+) -> trio.abc.ReceiveStream:
-#     async with maybe_open_feed_handler() as fh:
+    async with maybe_open_feed_handler() as fh:
-#         async with to_asyncio.open_channel_from(
+        async with to_asyncio.open_channel_from(
-#             partial(
+            partial(
-#                 aio_order_feed_relay,
+                aio_order_feed_relay,
-#                 fh,
+                fh,
-#                 instrument
+                instrument
-#             )
+            )
-#         ) as (first, chan):
+        ) as (first, chan):
-#             yield chan
+            yield chan
 
 
-# @acm
+@acm
-# async def maybe_open_order_feed(
+async def maybe_open_order_feed(
-#     instrument: str
+    instrument: str
-# ) -> trio.abc.ReceiveStream:
+) -> trio.abc.ReceiveStream:
 
-#     # TODO: add a predicate to maybe_open_context
+    # TODO: add a predicate to maybe_open_context
-#     async with maybe_open_context(
+    async with maybe_open_context(
-#         acm_func=open_order_feed,
+        acm_func=open_order_feed,
-#         kwargs={
+        kwargs={
-#             'instrument': instrument.split('.')[0],
+            'instrument': instrument.split('.')[0],
-#             'fh': fh
+            'fh': fh
-#         },
+        },
-#         key=f'{instrument.split('.')[0]}-order',
+        key=f'{instrument.split('.')[0]}-order',
-#     ) as (cache_hit, feed):
+    ) as (cache_hit, feed):
-#         if cache_hit:
+        if cache_hit:
-#             yield broadcast_receiver(feed, 10)
+            yield broadcast_receiver(feed, 10)
-#         else:
+        else:
-#             yield feed
+            yield feed

piker/brokers/deribit/feed.py

@@ -18,58 +18,56 @@
 Deribit backend.
 
 '''
-from __future__ import annotations
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
-from typing import (
+from typing import Any, Optional, Callable
-    # Any,
+from pprint import pformat
-    # Optional,
-    Callable,
-)
-# from pprint import pformat
 import time
 
-import cryptofeed
 import trio
 from trio_typing import TaskStatus
 from pendulum import (
     from_timestamp,
+    now,
 )
+from rapidfuzz import process as fuzzy
 import numpy as np
 import tractor
 
 from piker.accounting import (
-    Asset,
     MktPair,
     unpack_fqme,
 )
 from piker.brokers import (
     open_cached_client,
     NoData,
-    DataUnavailable,
 )
 from piker._cacheables import (
     async_lifo_cache,
 )
-from piker.log import (
+from piker.log import get_logger, get_console_log
-    get_logger,
+from piker.data import ShmArray
-    mk_repr,
-)
 from piker.data.validate import FeedInit
+from piker.brokers._util import (
+    BrokerError,
+    DataUnavailable,
+)
 
+from cryptofeed import FeedHandler
+from cryptofeed.defines import (
+    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
+)
+from cryptofeed.symbols import Symbol
 
 from .api import (
-    Client,
+    Client, Trade,
-    # get_config,
+    get_config,
-    piker_sym_to_cb_sym,
+    piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
-    cb_sym_to_deribit_inst,
-    str_to_cb_sym,
     maybe_open_price_feed
 )
 from .venues import (
     Pair,
     OptionPair,
-    Trade,
 )
 
 _spawn_kwargs = {
@@ -88,10 +86,6 @@ async def open_history_client(
     # TODO implement history getter for the new storage layer.
     async with open_cached_client('deribit') as client:
 
-        pair: OptionPair = client._pairs[mkt.dst.name]
-        # XXX NOTE, the cuckers use ms !!!
-        creation_time_s: int = pair.creation_timestamp/1000
-
         async def get_ohlc(
             timeframe: float,
             end_dt: datetime | None = None,
@@ -111,31 +105,6 @@ async def open_history_client(
                 end_dt=end_dt,
             )
             if len(array) == 0:
-                if (
-                    end_dt is None
-                ):
-                    raise DataUnavailable(
-                        'No history seems to exist yet?\n\n'
-                        f'{mkt}'
-                    )
-                elif (
-                    end_dt
-                    and
-                    end_dt.timestamp() < creation_time_s
-                ):
-                    # the contract can't have history
-                    # before it was created.
-                    pair_type_str: str = type(pair).__name__
-                    create_dt: datetime = from_timestamp(creation_time_s)
-                    raise DataUnavailable(
-                        f'No history prior to\n'
-                        f'`{pair_type_str}.creation_timestamp: int = '
-                        f'{pair.creation_timestamp}\n\n'
-                        f'------ deribit sux ------\n'
-                        f'WHICH IN "NORMAL PEOPLE WHO USE EPOCH TIME" form is,\n'
-                        f'creation_time_s: {creation_time_s}\n'
-                        f'create_dt: {create_dt}\n'
-                    )
                 raise NoData(
                     f'No frame for {start_dt} -> {end_dt}\n'
                 )
@@ -157,20 +126,14 @@ async def open_history_client(
 
             return array, start_dt, end_dt
 
-        yield (
+        yield get_ohlc, {'erlangs': 3, 'rate': 3}
-            get_ohlc,
-            {  # backfill config
-                'erlangs': 3,
-                'rate': 3,
-            }
-        )
 
 
 @async_lifo_cache()
 async def get_mkt_info(
     fqme: str,
 
-) -> tuple[MktPair, Pair|OptionPair] | None:
+) -> tuple[MktPair, Pair] | None:
 
     # uppercase since kraken bs_mktid is always upper
     if 'deribit' not in fqme.lower():
@@ -186,7 +149,7 @@ async def get_mkt_info(
     # returns, always!
     expiry: str = expiry.upper()
     venue: str = venue.upper()
-    # venue_lower: str = venue.lower()
+    venue_lower: str = venue.lower()
 
     mkt_mode: str = 'option'
 
@@ -212,88 +175,64 @@ async def get_mkt_info(
             price_tick=pair.price_tick,
             size_tick=pair.size_tick,
             bs_mktid=pair.symbol,
+            expiry=pair.expiry,
             venue=mkt_mode,
             broker='deribit',
             _atype=mkt_mode,
             _fqme_without_src=True,
-
-            # expiry=pair.expiry,
-            # XXX TODO, currently we don't use it since it's
-            # already "described" in the `OptionPair.symbol: str`
-            # and if we slap in the ISO repr it's kinda hideous..
-            # -[ ] figure out the best either std
         )
         return mkt, pair
 
 
 async def stream_quotes(
+
     send_chan: trio.abc.SendChannel,
     symbols: list[str],
     feed_is_live: trio.Event,
+    loglevel: str = None,
 
     # startup sync
     task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
 
 ) -> None:
-    '''
+    # XXX: required to propagate ``tractor`` loglevel to piker logging
-    Open a live quote stream for the market set defined by `symbols`.
+    get_console_log(loglevel or tractor.current_actor().loglevel)
 
-    Internally this starts a `cryptofeed.FeedHandler` inside an `asyncio`-side
-    task and relays through L1 and `Trade` msgs here to our `trio.Task`.
-
-    '''
     sym = symbols[0].split('.')[0]
-    init_msgs: list[FeedInit] = []
 
-    # multiline nested `dict` formatter (since rn quote-msgs are
+    init_msgs: list[FeedInit] = []
-    # just that).
-    pfmt: Callable[[str], str] = mk_repr(
-        # so we can see `deribit`'s delightfully mega-long bs fields..
-        maxstring=100,
-    )
 
     async with (
         open_cached_client('deribit') as client,
         send_chan as send_chan
     ):
-        mkt: MktPair
+
-        pair: Pair
         mkt, pair = await get_mkt_info(sym)
 
         # build out init msgs according to latest spec
         init_msgs.append(
-            FeedInit(
+            FeedInit(mkt_info=mkt)
-                mkt_info=mkt,
-            )
         )
-        # build `cryptofeed` feed-handle
+        nsym = piker_sym_to_cb_sym(sym)
-        cf_sym: cryptofeed.Symbol = piker_sym_to_cb_sym(sym)
 
-        from_cf: tractor.to_asyncio.LinkedTaskChannel
+        async with maybe_open_price_feed(sym) as stream:
-        async with maybe_open_price_feed(sym) as from_cf:
 
-            # load the "last trades" summary
+            cache = client._pairs
-            last_trades_res: cryptofeed.LastTradesResult = await client.last_trades(
-                cb_sym_to_deribit_inst(cf_sym),
-                count=1,
-            )
-            last_trades: list[Trade] = last_trades_res.trades
 
-            # TODO, do we even need this or will the above always
+            last_trades = (await client.last_trades(
-            # work?
+                cb_sym_to_deribit_inst(nsym), count=1)).trades
-            # if not last_trades:
-            #     await tractor.pause()
-            #     async for typ, quote in from_cf:
-            #         if typ == 'trade':
-            #             last_trade = Trade(**(quote['data']))
-            #             break
 
-            # else:
+            if len(last_trades) == 0:
-            last_trade = Trade(
+                last_trade = None
-                **(last_trades[0])
+                async for typ, quote in stream:
-            )
+                    if typ == 'trade':
+                        last_trade = Trade(**(quote['data']))
+                        break
 
-            first_quote: dict = {
+            else:
+                last_trade = Trade(**(last_trades[0]))
+
+            first_quote = {
                 'symbol': sym,
                 'last': last_trade.price,
                 'brokerd_ts': last_trade.timestamp,
@@ -304,84 +243,13 @@ async def stream_quotes(
                     'broker_ts': last_trade.timestamp
                 }]
             }
-            task_status.started((
+            task_status.started((init_msgs,  first_quote))
-                init_msgs,
-                first_quote,
-            ))
 
             feed_is_live.set()
 
-            # NOTE XXX, static for now!
+            async for typ, quote in stream:
-            # => since this only handles ONE mkt feed at a time we
+                topic = quote['symbol']
-            # don't need a lookup table to map interleaved quotes
+                await send_chan.send({topic: quote})
-            # from multiple possible mkt-pairs
-            topic: str = mkt.bs_fqme
-
-            # deliver until cancelled
-            async for typ, ref in from_cf:
-                match typ:
-                    case 'trade':
-                        trade: cryptofeed.types.Trade = ref
-
-                        # TODO, re-impl this according to teh ideal
-                        # fqme for opts that we choose!!
-                        bs_fqme: str = cb_sym_to_deribit_inst(
-                            str_to_cb_sym(trade.symbol)
-                        ).lower()
-
-                        piker_quote: dict = {
-                            'symbol': bs_fqme,
-                            'last': trade.price,
-                            'broker_ts': time.time(),
-                            # ^TODO, name this `brokerd/datad_ts` and
-                            # use `time.time_ns()` ??
-                            'ticks': [{
-                                'type': 'trade',
-                                'price': float(trade.price),
-                                'size': float(trade.amount),
-                                'broker_ts': trade.timestamp,
-                            }],
-                        }
-                        log.info(
-                            f'deribit {typ!r} quote for {sym!r}\n\n'
-                            f'{trade}\n\n'
-                            f'{pfmt(piker_quote)}\n'
-                        )
-
-                    case 'l1':
-                        book: cryptofeed.types.L1Book = ref
-
-                        # TODO, so this is where we can possibly change things
-                        # and instead lever the `MktPair.bs_fqme: str` output?
-                        bs_fqme: str = cb_sym_to_deribit_inst(
-                            str_to_cb_sym(book.symbol)
-                        ).lower()
-
-                        piker_quote: dict = {
-                            'symbol': bs_fqme,
-                            'ticks': [
-
-                                {'type': 'bid',
-                                 'price': float(book.bid_price),
-                                 'size': float(book.bid_size)},
-
-                                {'type': 'bsize',
-                                 'price': float(book.bid_price),
-                                 'size': float(book.bid_size),},
-
-                                {'type': 'ask',
-                                 'price': float(book.ask_price),
-                                 'size': float(book.ask_size),},
-
-                                {'type': 'asize',
-                                 'price': float(book.ask_price),
-                                 'size': float(book.ask_size),}
-                            ]
-                        }
-
-                await send_chan.send({
-                    topic: piker_quote,
-                })
 
 
 @tractor.context
@@ -391,13 +259,13 @@ async def open_symbol_search(
     async with open_cached_client('deribit') as client:
 
         # load all symbols locally for fast search
-        # cache = client._pairs
+        cache = client._pairs
         await ctx.started()
 
         async with ctx.open_stream() as stream:
+
             pattern: str
             async for pattern in stream:
-
                 # NOTE: pattern fuzzy-matching is done within
                 # the methd impl.
                 pairs: dict[str, Pair] = await client.search_symbols(

piker/brokers/deribit/venues.py

@@ -22,10 +22,11 @@ from __future__ import annotations
 import pendulum
 from typing import (
     Literal,
-    Optional,
 )
 from decimal import Decimal
 
+from msgspec import field
+
 from piker.types import Struct
 
 
@@ -110,21 +111,18 @@ class OptionPair(Pair, frozen=True):
     block_trade_min_trade_amount: int # '25'
     block_trade_commission: float # '0.003'
 
+
     # NOTE: see `.data._symcache.SymbologyCache.load()` for why
     ns_path: str = 'piker.brokers.deribit:OptionPair'
 
-    # TODO, impl this without the MM:SS part of
-    # the `'THH:MM:SS..'` etc..
     @property
     def expiry(self) -> str:
-        iso_date = pendulum.from_timestamp(
+        iso_date = pendulum.from_timestamp(self.expiration_timestamp / 1000).isoformat()
-            self.expiration_timestamp / 1000
-        ).isoformat()
         return iso_date 
 
     @property
     def venue(self) -> str:
-        return f'{self.instrument_type}_option'
+        return 'option'
 
     @property
     def bs_fqme(self) -> str:
@@ -154,7 +152,6 @@ class JSONRPCResult(Struct):
     error: Optional[dict] = None
     result: Optional[list[dict]] = None
 
-
 class JSONRPCChannel(Struct):
     method: str
     params: dict
@@ -171,7 +168,6 @@ class KLinesResult(Struct):
     status: str
     volume: list[float]
 
-
 class Trade(Struct):
     iv: float
     price: float
@@ -190,7 +186,6 @@ class Trade(Struct):
     block_trade_id: Optional[str] = '',
     block_trade_leg_count: Optional[int] = 0,
 
-
 class LastTradesResult(Struct):
     trades: list[Trade]
     has_more: bool

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(