piker/brokers/__init__.py

@@ -51,7 +51,6 @@ __brokers__: list[str] = [
     'ib',
     'kraken',
     'kucoin',
-    'deribit',
 
     # broken but used to work
     # 'questrade',
@@ -62,6 +61,7 @@ __brokers__: list[str] = [
     # wstrade
     # iex
 
+    # deribit
     # bitso
 ]
 

piker/brokers/deribit/__init__.py

@@ -25,7 +25,6 @@ from .api import (
     get_client,
 )
 from .feed import (
-    get_mkt_info,
     open_history_client,
     open_symbol_search,
     stream_quotes,
@@ -35,20 +34,15 @@ from .feed import (
     # open_trade_dialog,
     # norm_trade_records,
 # )
-from .venues import (
-    OptionPair,
-)
 
 log = get_logger(__name__)
 
 __all__ = [
     'get_client',
 #    'trades_dialogue',
-    'get_mkt_info',
     'open_history_client',
     'open_symbol_search',
     'stream_quotes',
-    'OptionPair',
 #    'norm_trade_records',
 ]
 

piker/brokers/deribit/api.py

@@ -19,14 +19,10 @@ Deribit backend.
 
 '''
 import asyncio
-from collections import ChainMap
 from contextlib import (
     asynccontextmanager as acm,
 )
 from datetime import datetime
-from decimal import (
-    Decimal,
-)
 from functools import partial
 import time
 from typing import (
@@ -35,7 +31,7 @@ from typing import (
     Callable,
 )
 
-from pendulum import now
+import pendulum
 import trio
 from trio_typing import TaskStatus
 from rapidfuzz import process as fuzzy
@@ -55,25 +51,7 @@ from cryptofeed.defines import (
     OPTION, CALL, PUT
 )
 from cryptofeed.symbols import Symbol
-# types for managing the cb callbacks.
+
-# from cryptofeed.types import L1Book
-from .venues import (
-    _ws_url,
-    MarketType,
-    PAIRTYPES,
-    Pair,
-    OptionPair,
-    JSONRPCResult,
-    JSONRPCChannel,
-    KLinesResult,
-    Trade,
-    LastTradesResult,
-)
-from piker.accounting import (
-    Asset,
-    digits_to_dec,
-    MktPair,
-)
 from piker.data import (
     def_iohlcv_fields,
     match_from_pairs,
@@ -96,6 +74,57 @@ _spawn_kwargs = {
 }
 
 
+_url = 'https://www.deribit.com'
+_ws_url = 'wss://www.deribit.com/ws/api/v2'
+_testnet_ws_url = 'wss://test.deribit.com/ws/api/v2'
+
+
+class JSONRPCResult(Struct):
+    jsonrpc: str = '2.0'
+    id: int
+    result: Optional[list[dict]] = None
+    error: Optional[dict] = None
+    usIn: int
+    usOut: int
+    usDiff: int
+    testnet: bool
+
+class JSONRPCChannel(Struct):
+    jsonrpc: str = '2.0'
+    method: str
+    params: dict
+
+
+class KLinesResult(Struct):
+    close: list[float]
+    cost: list[float]
+    high: list[float]
+    low: list[float]
+    open: list[float]
+    status: str
+    ticks: list[int]
+    volume: list[float]
+
+class Trade(Struct):
+    trade_seq: int
+    trade_id: str
+    timestamp: int
+    tick_direction: int
+    price: float
+    mark_price: float
+    iv: float
+    instrument_name: str
+    index_price: float
+    direction: str
+    combo_trade_id: Optional[int] = 0,
+    combo_id: Optional[str] = '',
+    amount: float
+
+class LastTradesResult(Struct):
+    trades: list[Trade]
+    has_more: bool
+
+
 # convert datetime obj timestamp to unixtime in milliseconds
 def deribit_timestamp(when):
     return int((when.timestamp() * 1000) + (when.microsecond / 1000))
@@ -113,15 +142,13 @@ def str_to_cb_sym(name: str) -> Symbol:
     else:
         raise Exception("Couldn\'t parse option type")
 
-    new_expiry_date = get_values_from_cb_normalized_date(expiry_date)
-
     return Symbol(
-        base=base,
+        base, quote,
-        quote=quote,
         type=OPTION,
         strike_price=strike_price,
         option_type=option_type,
-        expiry_date=new_expiry_date)
+        expiry_date=expiry_date,
+        expiry_normalize=False)
 
 
 def piker_sym_to_cb_sym(name: str) -> Symbol:
@@ -138,132 +165,77 @@ def piker_sym_to_cb_sym(name: str) -> Symbol:
         raise Exception("Couldn\'t parse option type")
 
     return Symbol(
-        base=base,
+        base, quote,
-        quote=quote,
         type=OPTION,
         strike_price=strike_price,
         option_type=option_type,
-        expiry_date=expiry_date)
+        expiry_date=expiry_date.upper())
 
 
 def cb_sym_to_deribit_inst(sym: Symbol):
-    new_expiry_date = get_values_from_cb_normalized_date(sym.expiry_date)
+    # cryptofeed normalized
-    otype = 'C' if sym.option_type == CALL else 'P'
+    cb_norm = ['F', 'G', 'H', 'J', 'K', 'M', 'N', 'Q', 'U', 'V', 'X', 'Z']
 
-    return f'{sym.base}-{new_expiry_date}-{sym.strike_price}-{otype}'
-
-
-def get_values_from_cb_normalized_date(expiry_date: str) -> str:
     # deribit specific 
-    cb_norm = [
+    months = ['JAN', 'FEB', 'MAR', 'APR', 'MAY', 'JUN', 'JUL', 'AUG', 'SEP', 'OCT', 'NOV', 'DEC']
-        'F', 'G', 'H', 'J',
+
-        'K', 'M', 'N', 'Q',
+    exp = sym.expiry_date
-        'U', 'V', 'X', 'Z'
+
-    ]
-    months = [
-        'JAN', 'FEB', 'MAR', 'APR',
-        'MAY', 'JUN', 'JUL', 'AUG',
-        'SEP', 'OCT', 'NOV', 'DEC'
-    ]
     # YYMDD
     # 01234
-    day, month, year = (
+    year, month, day = (
-        expiry_date[3:],
+        exp[:2], months[cb_norm.index(exp[2:3])], exp[3:])
-        months[cb_norm.index(expiry_date[2:3])],
+
-        expiry_date[:2]
+    otype = 'C' if sym.option_type == CALL else 'P'
-    )
+
-    return f'{day}{month}{year}'
+    return f'{sym.base}-{day}{month}{year}-{sym.strike_price}-{otype}'
 
 
 def get_config() -> dict[str, Any]:
 
-    conf: dict
+    conf, path = config.load()
-    path: Path
 
-    conf, path = config.load(
-        conf_name='brokers',
-        touch_if_dne=True,
-    )
-    section: dict = {}
     section = conf.get('deribit')
+
+    # TODO: document why we send this, basically because logging params for cryptofeed
+    conf['log'] = {}
+    conf['log']['disabled'] = True
+
     if section is None:
         log.warning(f'No config section found for deribit in {path}')
-        return {}
 
-    conf_option = section.get('option', {})
+    return conf 
-    section.clear # clear the dict to reuse it
-    section['deribit'] = {}
-    section['deribit']['key_id'] = conf_option.get('api_key')
-    section['deribit']['key_secret'] = conf_option.get('api_secret')
-
-    section['log'] = {}
-    section['log']['filename'] = 'feedhandler.log'
-    section['log']['level'] = 'DEBUG'
-
-    return section
 
 
 class Client:
 
-    def __init__(
+    def __init__(self, json_rpc: Callable) -> None:
-        self,
+        self._pairs: dict[str, Any] = None
-
-        json_rpc: Callable
-
-    ) -> None:
-        self._pairs: ChainMap[str, Pair] = ChainMap()
 
         config = get_config().get('deribit', {})
 
-        self._key_id = config.get('key_id')
+        if ('key_id' in config) and ('key_secret' in config):
-        self._key_secret = config.get('key_secret')
+            self._key_id = config['key_id']
+            self._key_secret = config['key_secret']
+
+        else:
+            self._key_id = None
+            self._key_secret = None
 
         self.json_rpc = json_rpc
 
-        self._auth_ts = None
+    @property
-        self._auth_renew_ts = 5 # seconds to renew auth
+    def currencies(self):
+        return ['btc', 'eth', 'sol', 'usd']
 
-    async def _json_rpc_auth_wrapper(self, *args, **kwargs) -> JSONRPCResult:
+    async def get_balances(self, kind: str = 'option') -> dict[str, float]:
-        
-        """Background task that adquires a first access token and then will
-        refresh the access token.
-
-        https://docs.deribit.com/?python#authentication-2
-        """
-        access_scope = 'trade:read_write'
-        current_ts = time.time()
-
-        if not self._auth_ts or current_ts - self._auth_ts < self._auth_renew_ts:
-            # if we are close to token expiry time
-
-            params = {
-                'grant_type': 'client_credentials',
-                'client_id': self._key_id,
-                'client_secret': self._key_secret,
-                'scope': access_scope
-            }
-
-            resp = await self.json_rpc('public/auth', params)
-            result = resp.result
-
-            self._auth_ts = time.time() + result['expires_in']
-
-        return await self.json_rpc(*args, **kwargs)
-
-
-
-
-    async def get_balances(
-        self,
-        kind: str = 'option'
-    ) -> dict[str, float]:
         """Return the set of positions for this account
         by symbol.
         """
         balances = {}
 
         for currency in self.currencies:
-            resp = await self._json_rpc_auth_wrapper(
+            resp = await self.json_rpc(
                 'private/get_positions', params={
                     'currency': currency.upper(),
                     'kind': kind})
@@ -272,46 +244,20 @@ class Client:
 
         return balances
 
-    async def get_assets(
+    async def get_assets(self) -> dict[str, float]:
-        self,
-        venue: str | None = None,
-
-    ) -> dict[str, Asset]:
         """Return the set of asset balances for this account
         by symbol.
         """
-        assets = {}
+        balances = {}
-        resp = await self._json_rpc_auth_wrapper(
-            'public/get_currencies',
-            params={}
-        )
-        currencies = resp.result
-        for currency in currencies:
-            name = currency['currency']
-            tx_tick = digits_to_dec(currency['fee_precision']) 
-            atype='crypto_currency'
-            assets[name] = Asset(
-                name=name,
-                atype=atype,
-                tx_tick=tx_tick)
 
-            instruments = await self.symbol_info(currency=name)
+        for currency in self.currencies:
-            for instrument in instruments:
+            resp = await self.json_rpc(
-                pair = instruments[instrument]
+                'private/get_account_summary', params={
-                assets[pair.symbol] = Asset(
+                    'currency': currency.upper()})
-                    name=pair.symbol,
-                    atype=pair.venue,
-                    tx_tick=pair.size_tick)
 
-        return assets 
+            balances[currency] = resp.result['balance']
 
-    async def get_mkt_pairs(self) -> dict[str, Pair]:
+        return balances
-        flat: dict[str, Pair] = {}
-        for key in self._pairs:
-            item = self._pairs.get(key)
-            flat[item.bs_fqme] = item
-
-        return flat
 
     async def submit_limit(
         self,
@@ -328,7 +274,7 @@ class Client:
             'type': 'limit',
             'price': price,
         }
-        resp = await self._json_rpc_auth_wrapper(
+        resp = await self.json_rpc(
             f'private/{action}', params)
 
         return resp.result
@@ -336,32 +282,10 @@ class Client:
     async def submit_cancel(self, oid: str):
         """Send cancel request for order id
         """
-        resp = await self._json_rpc_auth_wrapper(
+        resp = await self.json_rpc(
             'private/cancel', {'order_id': oid})
         return resp.result
 
-    async def exch_info(
-        self,
-        sym: str | None = None,
-
-        venue: MarketType = 'option',
-        expiry: str | None = None,
-
-    ) -> dict[str, Pair] | Pair:
-
-        pair_table: dict[str, Pair] = self._pairs
-
-        if (
-            sym
-            and (cached_pair := pair_table.get(sym))
-        ):
-            return cached_pair
-
-        if sym:
-            return pair_table[sym]
-        else:
-            return self._pairs
-
     async def symbol_info(
         self,
         instrument: Optional[str] = None,
@@ -369,7 +293,7 @@ class Client:
         kind: str = 'option',
         expired: bool = False
 
-    ) -> dict[str, Pair] | Pair:
+    ) -> dict[str, dict]:
         '''
         Get symbol infos.
 
@@ -384,65 +308,28 @@ class Client:
             'expired': str(expired).lower()
         }
 
-        resp: JSONRPCResult = await self._json_rpc_auth_wrapper(
+        resp: JSONRPCResult = await self.json_rpc(
             'public/get_instruments',
             params,
         )
         # convert to symbol-keyed table
-        pair_type: Type = PAIRTYPES[kind]
         results: list[dict] | None = resp.result
-        
+        instruments: dict[str, dict] = {
-        instruments: dict[str, Pair] = {}
+            item['instrument_name'].lower(): item
-        for item in results:
+            for item in results
-            symbol=item['instrument_name'].lower()
+        }
-            try:
-                pair: Pair = pair_type(
-                    symbol=symbol,
-                    **item
-                )
-            except Exception as e:
-                e.add_note(
-                    "\nDon't panic, prolly stupid deribit changed their symbology schema again..\n"
-                    'Check out their API docs here:\n\n'
-                    'https://docs.deribit.com/?python#deribit-api-v2-1-1'
-                )
-                raise
-
-            instruments[symbol] = pair
 
         if instrument is not None:
-            return instruments[instrument.lower()]
+            return instruments[instrument]
         else:
             return instruments
 
     async def cache_symbols(
         self,
-        venue: MarketType = 'option',
+    ) -> dict:
 
-    ) -> None:
+        if not self._pairs:
-        # lookup internal mkt-specific pair table to update
+            self._pairs = await self.symbol_info()
-        pair_table: dict[str, Pair] = self._pairs
-
-        # make API request(s)
-        mkt_pairs = await self.symbol_info()
-
-        if not mkt_pairs:
-            raise SymbolNotFound(f'No market pairs found!?:\n{resp}')
-
-        pairs_view_subtable: dict[str, Pair] = {}
-
-        for instrument in mkt_pairs:
-            pair_type: Type = PAIRTYPES[venue]
-
-            pair: Pair = pair_type(**mkt_pairs[instrument].to_dict())
-
-            pair_table[pair.symbol.upper()] = pair
-
-            # update an additional top-level-cross-venue-table
-            # `._pairs: ChainMap` for search B0
-            pairs_view_subtable[pair.bs_fqme] = pair
-
-        self._pairs.maps.append(pairs_view_subtable)
 
         return self._pairs
 
@@ -450,35 +337,37 @@ class Client:
         self,
         pattern: str,
         limit: int = 30,
-    ) -> dict[str, Pair]:
+    ) -> dict[str, Any]:
         '''
         Fuzzy search symbology set for pairs matching `pattern`.
 
         '''
-        pairs: dict[str, Pair] = await self.exch_info()
+        pairs: dict[str, Any] = await self.symbol_info()
-
+        matches: dict[str, Pair] = match_from_pairs(
-        return match_from_pairs(
             pairs=pairs,
             query=pattern.upper(),
             score_cutoff=35,
             limit=limit
         )
 
+       # repack in name-keyed table
+        return {
+            pair['instrument_name'].lower(): pair
+            for pair in matches.values()
+        }
+
     async def bars(
         self,
-        mkt: MktPair,
+        symbol: str,
-
         start_dt: Optional[datetime] = None,
         end_dt: Optional[datetime] = None,
-
         limit: int = 1000,
         as_np: bool = True,
-
+    ) -> dict:
-    ) -> list[tuple] | np.ndarray:
+        instrument = symbol
-        instrument: str = mkt.bs_fqme.split('.')[0]
 
         if end_dt is None:
-            end_dt = now('UTC')
+            end_dt = pendulum.now('UTC')
 
         if start_dt is None:
             start_dt = end_dt.start_of(
@@ -488,7 +377,7 @@ class Client:
         end_time = deribit_timestamp(end_dt)
 
         # https://docs.deribit.com/#public-get_tradingview_chart_data
-        resp = await self._json_rpc_auth_wrapper(
+        resp = await self.json_rpc(
             'public/get_tradingview_chart_data',
             params={
                 'instrument_name': instrument.upper(),
@@ -498,34 +387,36 @@ class Client:
             })
 
         result = KLinesResult(**resp.result)
-        new_bars: list[tuple] = []
+        new_bars = []
         for i in range(len(result.close)):
 
+            _open = result.open[i]
+            high = result.high[i]
+            low = result.low[i]
+            close = result.close[i]
+            volume = result.volume[i]
+
             row = [
                 (start_time + (i * (60 * 1000))) / 1000.0,  # time
                 result.open[i],
                 result.high[i],
                 result.low[i],
                 result.close[i],
-                result.volume[i]
+                result.volume[i],
+                0
             ]
 
             new_bars.append((i,) + tuple(row))
 
-        if not as_np:
+        array = np.array(new_bars, dtype=def_iohlcv_fields) if as_np else klines
-            return result
+        return array
-
-        return np.array(
-            new_bars,
-            dtype=def_iohlcv_fields
-        )
 
     async def last_trades(
         self,
         instrument: str,
         count: int = 10
     ):
-        resp = await self._json_rpc_auth_wrapper(
+        resp = await self.json_rpc(
             'public/get_last_trades_by_instrument',
             params={
                 'instrument_name': instrument,
@@ -537,17 +428,78 @@ class Client:
 
 @acm
 async def get_client(
-    is_brokercheck: bool = False,
+    is_brokercheck: bool = False
-    venue: MarketType = 'option',
 ) -> Client:
 
     async with (
         trio.open_nursery() as n,
         open_jsonrpc_session(
-            _ws_url, response_type=JSONRPCResult
+            _testnet_ws_url, dtype=JSONRPCResult) as json_rpc
-        ) as json_rpc
     ):
         client = Client(json_rpc)
+
+        _refresh_token: Optional[str] = None
+        _access_token: Optional[str] = None
+
+        async def _auth_loop(
+            task_status: TaskStatus = trio.TASK_STATUS_IGNORED
+        ):
+            """Background task that adquires a first access token and then will
+            refresh the access token while the nursery isn't cancelled.
+
+            https://docs.deribit.com/?python#authentication-2
+            """
+            renew_time = 10
+            access_scope = 'trade:read_write'
+            _expiry_time = time.time()
+            got_access = False
+            nonlocal _refresh_token
+            nonlocal _access_token
+
+            while True:
+                if time.time() - _expiry_time < renew_time:
+                    # if we are close to token expiry time
+
+                    if _refresh_token != None:
+                        # if we have a refresh token already dont need to send
+                        # secret
+                        params = {
+                            'grant_type': 'refresh_token',
+                            'refresh_token': _refresh_token,
+                            'scope': access_scope
+                        }
+
+                    else:
+                        # we don't have refresh token, send secret to initialize
+                        params = {
+                            'grant_type': 'client_credentials',
+                            'client_id': client._key_id,
+                            'client_secret': client._key_secret,
+                            'scope': access_scope
+                        }
+
+                    resp = await json_rpc('public/auth', params)
+                    result = resp.result
+
+                    _expiry_time = time.time() + result['expires_in']
+                    _refresh_token = result['refresh_token']
+
+                    if 'access_token' in result:
+                        _access_token = result['access_token']
+
+                    if not got_access:
+                        # first time this loop runs we must indicate task is
+                        # started, we have auth
+                        got_access = True
+                        task_status.started()
+
+                else:
+                    await trio.sleep(renew_time / 2)
+
+        # if we have client creds launch auth loop
+        if client._key_id is not None:
+            await n.start(_auth_loop)
+
         await client.cache_symbols()
         yield client
         n.cancel_scope.cancel()
@@ -571,7 +523,7 @@ async def maybe_open_feed_handler() -> trio.abc.ReceiveStream:
 
 async def aio_price_feed_relay(
     fh: FeedHandler,
-    instrument: str,
+    instrument: Symbol,
     from_trio: asyncio.Queue,
     to_trio: trio.abc.SendChannel,
 ) -> None:
@@ -590,33 +542,21 @@ async def aio_price_feed_relay(
             'symbol': cb_sym_to_deribit_inst(
                 str_to_cb_sym(data.symbol)).lower(),
             'ticks': [
-                {
+                {'type': 'bid',
-                    'type': 'bid',
+                    'price': float(data.bid_price), 'size': float(data.bid_size)},
-                    'price': float(data.bid_price),
+                {'type': 'bsize',
-                    'size': float(data.bid_size)
+                    'price': float(data.bid_price), 'size': float(data.bid_size)},
-                },
+                {'type': 'ask',
-                {
+                    'price': float(data.ask_price), 'size': float(data.ask_size)},
-                    'type': 'bsize',
+                {'type': 'asize',
-                    'price': float(data.bid_price),
+                    'price': float(data.ask_price), 'size': float(data.ask_size)}
-                    '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,
         channels=[TRADES, L1_BOOK],
-        symbols=[sym],
+        symbols=[piker_sym_to_cb_sym(instrument)],
         callbacks={
             TRADES: _trade,
             L1_BOOK: _l1
@@ -657,9 +597,9 @@ async def maybe_open_price_feed(
     async with maybe_open_context(
         acm_func=open_price_feed,
         kwargs={
-            'instrument': instrument.split('.')[0]
+            'instrument': instrument
         },
-        key=f'{instrument.split('.')[0]}-price',
+        key=f'{instrument}-price',
     ) as (cache_hit, feed):
         if cache_hit:
             yield broadcast_receiver(feed, 10)
@@ -724,10 +664,10 @@ async def maybe_open_order_feed(
     async with maybe_open_context(
         acm_func=open_order_feed,
         kwargs={
-            'instrument': instrument.split('.')[0],
+            'instrument': instrument,
             'fh': fh
         },
-        key=f'{instrument.split('.')[0]}-order',
+        key=f'{instrument}-order',
     ) as (cache_hit, feed):
         if cache_hit:
             yield broadcast_receiver(feed, 10)

piker/brokers/deribit/feed.py

@@ -21,33 +21,18 @@ Deribit backend.
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
 from typing import Any, Optional, Callable
-from pprint import pformat
 import time
 
 import trio
 from trio_typing import TaskStatus
-from pendulum import (
+import pendulum
-    from_timestamp,
-    now,
-)
 from rapidfuzz import process as fuzzy
 import numpy as np
 import tractor
 
-from piker.accounting import (
+from piker.brokers import open_cached_client
-    MktPair,
-    unpack_fqme,
-)
-from piker.brokers import (
-    open_cached_client,
-    NoData,
-)
-from piker._cacheables import (
-    async_lifo_cache,
-)
 from piker.log import get_logger, get_console_log
 from piker.data import ShmArray
-from piker.data.validate import FeedInit
 from piker.brokers._util import (
     BrokerError,
     DataUnavailable,
@@ -62,13 +47,9 @@ from cryptofeed.symbols import Symbol
 from .api import (
     Client, Trade,
     get_config,
-    piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
+    str_to_cb_sym, piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
     maybe_open_price_feed
 )
-from .venues import (
-    Pair,
-    OptionPair,
-)
 
 _spawn_kwargs = {
     'infect_asyncio': True,
@@ -83,107 +64,36 @@ async def open_history_client(
     mkt: MktPair,
 ) -> tuple[Callable, int]:
 
+    fnstrument: str = mkt.bs_fqme
     # TODO implement history getter for the new storage layer.
     async with open_cached_client('deribit') as client:
 
         async def get_ohlc(
-            timeframe: float,
+            end_dt: Optional[datetime] = None,
-            end_dt: datetime | None = None,
+            start_dt: Optional[datetime] = None,
-            start_dt: datetime | None = None,
 
         ) -> tuple[
             np.ndarray,
             datetime,  # start
             datetime,  # end
         ]:
-            if timeframe != 60:
-                raise DataUnavailable('Only 1m bars are supported')
 
-            array: np.ndarray = await client.bars(
+            array = await client.bars(
-                mkt,
+                instrument,
                 start_dt=start_dt,
                 end_dt=end_dt,
             )
             if len(array) == 0:
-                raise NoData(
+                raise DataUnavailable
-                    f'No frame for {start_dt} -> {end_dt}\n'
-                )
 
-            start_dt = from_timestamp(array[0]['time'])
+            start_dt = pendulum.from_timestamp(array[0]['time'])
-            end_dt = from_timestamp(array[-1]['time'])
+            end_dt = pendulum.from_timestamp(array[-1]['time'])
-
-            times = array['time']
-            if not times.any():
-                raise ValueError(
-                    'Bad frame with null-times?\n\n'
-                    f'{times}'
-                )
-
-            if end_dt is None:
-                inow: int = round(time.time())
-                if (inow - times[-1]) > 60:
-                    await tractor.pause()
 
             return array, start_dt, end_dt
 
         yield get_ohlc, {'erlangs': 3, 'rate': 3}
 
 
-@async_lifo_cache()
-async def get_mkt_info(
-    fqme: str,
-
-) -> tuple[MktPair, Pair] | None:
-
-    # uppercase since kraken bs_mktid is always upper
-    if 'deribit' not in fqme.lower():
-        fqme += '.deribit'
-
-    mkt_mode: str = ''
-    broker, mkt_ep, venue, expiry = unpack_fqme(fqme)
-
-    # NOTE: we always upper case all tokens to be consistent with
-    # binance's symbology style for pairs, like `BTCUSDT`, but in
-    # theory we could also just keep things lower case; as long as
-    # we're consistent and the symcache matches whatever this func
-    # returns, always!
-    expiry: str = expiry.upper()
-    venue: str = venue.upper()
-    venue_lower: str = venue.lower()
-
-    mkt_mode: str = 'option'
-
-    async with open_cached_client(
-        'deribit',
-    ) as client:
-
-        assets: dict[str, Asset] = await client.get_assets()
-        pair_str: str = mkt_ep.lower()
-
-        pair: Pair = await client.exch_info(
-            sym=pair_str,
-        )
-        mkt_mode = pair.venue
-        client.mkt_mode = mkt_mode
-
-        dst: Asset | None = assets.get(pair.bs_dst_asset)
-        src: Asset | None = assets.get(pair.bs_src_asset)
-
-        mkt = MktPair(
-            dst=dst,
-            src=src,
-            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,
-        )
-        return mkt, pair
-
-
 async def stream_quotes(
 
     send_chan: trio.abc.SendChannel,
@@ -198,26 +108,31 @@ async def stream_quotes(
     # XXX: required to propagate ``tractor`` loglevel to piker logging
     get_console_log(loglevel or tractor.current_actor().loglevel)
 
-    sym = symbols[0].split('.')[0]
+    sym = symbols[0]
-
-    init_msgs: list[FeedInit] = []
 
     async with (
         open_cached_client('deribit') as client,
         send_chan as send_chan
     ):
 
-        mkt, pair = await get_mkt_info(sym)
+        init_msgs = {
+            # pass back token, and bool, signalling if we're the writer
+            # and that history has been written
+            sym: {
+                'symbol_info': {
+                    'asset_type': 'option',
+                    'price_tick_size': 0.0005
+                },
+                'shm_write_opts': {'sum_tick_vml': False},
+                'fqsn': sym,
+            },
+        }
 
-        # build out init msgs according to latest spec
-        init_msgs.append(
-            FeedInit(mkt_info=mkt)
-        )
         nsym = piker_sym_to_cb_sym(sym)
 
         async with maybe_open_price_feed(sym) as stream:
 
-            cache = client._pairs
+            cache = await client.cache_symbols()
 
             last_trades = (await client.last_trades(
                 cb_sym_to_deribit_inst(nsym), count=1)).trades
@@ -259,21 +174,12 @@ async def open_symbol_search(
     async with open_cached_client('deribit') as client:
 
         # load all symbols locally for fast search
-        cache = client._pairs
+        cache = await client.cache_symbols()
         await ctx.started()
 
         async with ctx.open_stream() as stream:
 
-            pattern: str
             async for pattern in stream:
-                # NOTE: pattern fuzzy-matching is done within
+                # repack in dict form
-                # the methd impl.
+                await stream.send(
-                pairs: dict[str, Pair] = await client.search_symbols(
+                    await client.search_symbols(pattern))
-                    pattern,
-                )
-                # repack in fqme-keyed table
-                byfqme: dict[str, Pair] = {}
-                for pair in pairs.values():
-                    byfqme[pair.bs_fqme] = pair
-
-                await stream.send(byfqme)

piker/brokers/deribit/venues.py

@@ -1,191 +0,0 @@
-# piker: trading gear for hackers
-# Copyright (C) Tyler Goodlet (in stewardship for pikers)
-
-# This program is free software: you can redistribute it and/or modify
-# it under the terms of the GNU Affero General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-
-# This program is distributed in the hope that it will be useful,
-# but WITHOUT ANY WARRANTY; without even the implied warranty of
-# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
-# GNU Affero General Public License for more details.
-
-# You should have received a copy of the GNU Affero General Public License
-# along with this program.  If not, see <https://www.gnu.org/licenses/>.
-
-"""
-Per market data-type definitions and schemas types.
-
-"""
-from __future__ import annotations
-import pendulum
-from typing import (
-    Literal,
-)
-from decimal import Decimal
-
-from msgspec import field
-
-from piker.types import Struct
-
-
-# API endpoint paths by venue / sub-API
-_domain: str = 'deribit.com'
-_url = f'https://www.{_domain}'
-
-# WEBsocketz
-_ws_url: str = f'wss://www.{_domain}/ws/api/v2'
-
-# test nets
-_testnet_ws_url: str = f'wss://test.{_domain}/ws/api/v2'
-
-MarketType = Literal[
-    'option'
-]
-
-
-def get_api_eps(venue: MarketType) -> tuple[str, str]:
-    '''
-    Return API ep root paths per venue.
-
-    '''
-    return {
-        'option': (
-            _ws_url,
-        ),
-    }[venue]
-
-
-class Pair(Struct, frozen=True, kw_only=True):
-
-    symbol: str
-
-    # src
-    quote_currency: str # 'BTC'
-
-    # dst
-    base_currency: str # "BTC",
-
-    tick_size: float # 0.0001 # [{'above_price': 0.005, 'tick_size': 0.0005}]
-    tick_size_steps: list[dict[str, float]] 
-
-    @property
-    def price_tick(self) -> Decimal:
-        return Decimal(str(self.tick_size_steps[0]['above_price']))
-
-    @property
-    def size_tick(self) -> Decimal:
-        return Decimal(str(self.tick_size))
-
-    @property
-    def bs_fqme(self) -> str:
-        return f'{self.symbol}'
-
-    @property
-    def bs_mktid(self) -> str:
-        return f'{self.symbol}.{self.venue}'
-
-
-class OptionPair(Pair, frozen=True):
-
-    taker_commission: float # 0.0003
-    strike: float # 5000.0
-    settlement_period: str # 'day'
-    settlement_currency: str # "BTC",
-    rfq: bool # false
-    price_index: str # 'btc_usd'
-    option_type: str # 'call'
-    min_trade_amount: float # 0.1
-    maker_commission: float # 0.0003
-    kind: str # 'option'
-    is_active: bool # true
-    instrument_type: str # 'reversed'
-    instrument_name: str # 'BTC-1SEP24-55000-C'
-    instrument_id: int # 364671
-    expiration_timestamp: int # 1725177600000
-    creation_timestamp: int # 1724918461000
-    counter_currency: str # 'USD' 
-    contract_size: float # '1.0'
-    block_trade_tick_size: float # '0.0001'
-    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'
-
-    @property
-    def expiry(self) -> str:
-        iso_date = pendulum.from_timestamp(self.expiration_timestamp / 1000).isoformat()
-        return iso_date 
-
-    @property
-    def venue(self) -> str:
-        return 'option'
-
-    @property
-    def bs_fqme(self) -> str:
-        return f'{self.symbol}'
-
-    @property
-    def bs_src_asset(self) -> str:
-        return f'{self.quote_currency}'
-
-    @property
-    def bs_dst_asset(self) -> str:
-        return f'{self.symbol}'
-
-
-PAIRTYPES: dict[MarketType, Pair] = {
-    'option': OptionPair,
-}
-
-
-class JSONRPCResult(Struct):
-    id: int
-    usIn: int
-    usOut: int
-    usDiff: int
-    testnet: bool
-    jsonrpc: str = '2.0'
-    error: Optional[dict] = None
-    result: Optional[list[dict]] = None
-
-class JSONRPCChannel(Struct):
-    method: str
-    params: dict
-    jsonrpc: str = '2.0'
-
-
-class KLinesResult(Struct):
-    low: list[float]
-    cost: list[float]
-    high: list[float]
-    open: list[float]
-    close: list[float]
-    ticks: list[int]
-    status: str
-    volume: list[float]
-
-class Trade(Struct):
-    iv: float
-    price: float
-    amount: float
-    trade_id: str
-    contracts: float
-    direction: str
-    trade_seq: int
-    timestamp: int
-    mark_price: float
-    index_price: float
-    tick_direction: int
-    instrument_name: str
-    combo_id: Optional[str] = '',
-    combo_trade_id: Optional[int] = 0,
-    block_trade_id: Optional[str] = '',
-    block_trade_leg_count: Optional[int] = 0,
-
-class LastTradesResult(Struct):
-    trades: list[Trade]
-    has_more: bool