config/brokers.toml

@@ -50,8 +50,3 @@ prefer_data_account = [
 paper = "XX0000000"
 margin = "X0000000"
 ira = "X0000000"
-
-
-[deribit]
-key_id = 'XXXXXXXX'
-key_secret = 'Xx_XxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXx'

piker/brokers/deribit/README.rst

@@ -1,70 +0,0 @@
-``deribit`` backend
-------------------
-pretty good liquidity crypto derivatives, uses custom json rpc over ws for
-client methods, then `cryptofeed` for data streams.
-
-status
-******
-- supports option charts
-- no order support yet 
-
-
-config
-******
-In order to get order mode support your ``brokers.toml``
-needs to have something like the following:
-
-.. code:: toml
-
-    [deribit]
-    key_id = 'XXXXXXXX'
-    key_secret = 'Xx_XxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXxXx'
-
-To obtain an api id and secret you need to create an account, which can be a
-real market account over at:
-
-    - deribit.com  (requires KYC for deposit address)
-
-Or a testnet account over at:
-
-    - test.deribit.com
-
-For testnet once the account is created here is how you deposit fake crypto to
-try it out:
-
-1) Go to Wallet:
-
-.. figure:: assets/0_wallet.png
-    :align: center
-    :target: assets/0_wallet.png
-    :alt: wallet page
-
-2) Then click on the elipsis menu and select deposit
-
-.. figure:: assets/1_wallet_select_deposit.png
-    :align: center
-    :target: assets/1_wallet_select_deposit.png
-    :alt: wallet deposit page
-
-3) This will take you to the deposit address page
-
-.. figure:: assets/2_gen_deposit_addr.png
-    :align: center
-    :target: assets/2_gen_deposit_addr.png
-    :alt: generate deposit address page
-
-4) After clicking generate you should see the address, copy it and go to the 
-[coin faucet](https://test.deribit.com/dericoin/BTC/deposit) and send fake
-coins to that address.
-
-.. figure:: assets/3_deposit_address.png
-    :align: center
-    :target: assets/3_deposit_address.png
-    :alt: generated address
-
-5) Back in the deposit address page you should see the deposit in your history
-
-.. figure:: assets/4_wallet_deposit_history.png
-    :align: center
-    :target: assets/4_wallet_deposit_history.png
-    :alt: wallet deposit history

piker/brokers/deribit/__init__.py

@@ -32,8 +32,8 @@ from .feed import (
     stream_quotes,
 )
 # from .broker import (
-    # trades_dialogue,
+#     trades_dialogue,
-    # norm_trade_records,
+#     norm_trade_records,
 # )
 
 __all__ = [
@@ -50,7 +50,7 @@ __all__ = [
 __enable_modules__: list[str] = [
     'api',
     'feed',
-    'broker',
+#    'broker',
 ]
 
 # passed to ``tractor.ActorNursery.start_actor()``

piker/brokers/deribit/api.py

@@ -1,6 +1,3 @@
-# piker: trading gear for hackers
-# Copyright (C) Guillermo Rodriguez (in stewardship for piker0)
-
 # 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
@@ -18,52 +15,29 @@
 Deribit backend.
 
 '''
-import json
-import time
-import asyncio
 
-from contextlib import asynccontextmanager as acm, AsyncExitStack
+from contextlib import asynccontextmanager as acm
-from itertools import count
-from functools import partial
 from datetime import datetime
-from typing import Any, List, Dict, Optional, Iterable, Callable
+from typing import Any, Optional, List
 
 import pendulum
 import asks
-import trio
-from trio_typing import Nursery, TaskStatus
 from fuzzywuzzy import process as fuzzy
 import numpy as np
 
 from piker.data.types import Struct
-from piker.data._web_bs import NoBsWs, open_autorecon_ws
 
 from .._util import resproc
 
 from piker import config
 from piker.log import get_logger
 
-from tractor.trionics import broadcast_receiver, BroadcastReceiver, maybe_open_context
-from tractor import to_asyncio
-
-from cryptofeed import FeedHandler
-
-from cryptofeed.defines import (
-    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
-)
 from cryptofeed.symbols import Symbol
 
 log = get_logger(__name__)
 
 
-_spawn_kwargs = {
-    'infect_asyncio': True,
-}
-
-
 _url = 'https://www.deribit.com'
-_ws_url = 'wss://www.deribit.com/ws/api/v2'
-_testnet_ws_url = 'wss://test.deribit.com/ws/api/v2'
 
 
 # Broker specific ohlc schema (rest)
@@ -81,9 +55,7 @@ _ohlc_dtype = [
 
 class JSONRPCResult(Struct):
     jsonrpc: str = '2.0'
-    id: int
+    result: dict 
-    result: Optional[dict] = None
-    error: Optional[dict] = None
     usIn: int 
     usOut: int 
     usDiff: int 
@@ -111,8 +83,6 @@ class Trade(Struct):
     instrument_name: str
     index_price: float
     direction: str
-    combo_trade_id: Optional[int] = 0,
-    combo_id: Optional[str] = '',
     amount: float
 
 class LastTradesResult(Struct):
@@ -125,161 +95,24 @@ def deribit_timestamp(when):
     return int((when.timestamp() * 1000) + (when.microsecond / 1000))
 
 
-def str_to_cb_sym(name: str) -> Symbol:
-    base, strike_price, expiry_date, option_type = name.split('-')
-
-    quote = base
-
-    if option_type == 'put':
-        option_type = PUT 
-    elif option_type  == 'call':
-        option_type = CALL
-    else:
-        raise Exception("Couldn\'t parse option type")
-
-    return Symbol(
-        base, quote,
-        type=OPTION,
-        strike_price=strike_price,
-        option_type=option_type,
-        expiry_date=expiry_date,
-        expiry_normalize=False)
-
-
-def piker_sym_to_cb_sym(name: str) -> Symbol:
-    base, expiry_date, strike_price, option_type = tuple(
-        name.upper().split('-'))
-
-    quote = base
-
-    if option_type == 'P':
-        option_type = PUT 
-    elif option_type  == 'C':
-        option_type = CALL
-    else:
-        raise Exception("Couldn\'t parse option type")
-
-    return Symbol(
-        base, quote,
-        type=OPTION,
-        strike_price=strike_price,
-        option_type=option_type,
-        expiry_date=expiry_date.upper())
-
-
-def cb_sym_to_deribit_inst(sym: Symbol):
-    # cryptofeed normalized
-    cb_norm = ['F', 'G', 'H', 'J', 'K', 'M', 'N', 'Q', 'U', 'V', 'X', 'Z']
-
-    # deribit specific 
-    months = ['JAN', 'FEB', 'MAR', 'APR', 'MAY', 'JUN', 'JUL', 'AUG', 'SEP', 'OCT', 'NOV', 'DEC']
-
-    exp = sym.expiry_date
-
-    # YYMDD
-    # 01234
-    year, month, day = (
-        exp[:2], months[cb_norm.index(exp[2:3])], exp[3:])
-
-    otype = 'C' if sym.option_type == CALL else 'P'
-
-    return f'{sym.base}-{day}{month}{year}-{sym.strike_price}-{otype}'
-
-
-def get_config() -> dict[str, Any]:
-
-    conf, path = config.load()
-
-    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 
-
-
 class Client:
 
-    def __init__(self, json_rpc: Callable) -> None:
+    def __init__(self) -> None:
-        self._pairs: dict[str, Any] = None
+        self._sesh = asks.Session(connections=4)
+        self._sesh.base_location = _url
+        self._pairs: dict[str, Any] = {}
 
-        config = get_config().get('deribit', {})
+    async def _api(
-
-        if ('key_id' in config) and ('key_secret' in config):
-            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
-
-    @property
-    def currencies(self):
-        return ['btc', 'eth', 'sol', 'usd']
-
-    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(
-                'private/get_positions', params={
-                    'currency': currency.upper(),
-                    'kind': kind})
-
-            balances[currency] = resp.result
-
-        return balances
-
-    async def get_assets(self) -> dict[str, float]:
-        """Return the set of asset balances for this account
-        by symbol.
-        """
-        balances = {}
-
-        for currency in self.currencies:
-            resp = await self.json_rpc(
-                'private/get_account_summary', params={
-                    'currency': currency.upper()})
-
-            balances[currency] = resp.result['balance']
-
-        return balances
-
-    async def submit_limit(
         self,
-        symbol: str,
+        method: str,
-        price: float,
+        params: dict,
-        action: str,
+    ) -> dict[str, Any]:
-        size: float
+        resp = await self._sesh.get(
-    ) -> dict:
+            path=f'/api/v2/public/{method}',
-        """Place an order
+            params=params,
-        """
+            timeout=float('inf')
-        params = {
+        )
-            'instrument_name': symbol.upper(),
+        return resproc(resp, log)
-            'amount': size,
-            'type': 'limit',
-            'price': price,
-        }
-        resp = await self.json_rpc(
-            f'private/{action}', params)
-
-        return resp.result
-
-    async def submit_cancel(self, oid: str):
-        """Send cancel request for order id
-        """
-        resp = await self.json_rpc(
-            'private/cancel', {'order_id': oid})
-        return resp.result
 
     async def symbol_info(
         self,
@@ -288,11 +121,11 @@ class Client:
         kind: str = 'option',
         expired: bool = False
     ) -> dict[str, Any]:
-        """Get symbol info for the exchange.
+        '''Get symbol info for the exchange.
 
-        """
+        '''
-        if self._pairs:
+        # TODO: we can load from our self._pairs cache
-            return self._pairs
+        # on repeat calls...
 
         # will retrieve all symbols by default
         params = {
@@ -301,13 +134,13 @@ class Client:
             'expired': str(expired).lower()
         }
 
-        resp = await self.json_rpc('public/get_instruments', params)
+        resp = await self._api(
-        results = resp.result
+            'get_instruments', params=params)
+
+        results = resp['result']
 
         instruments = {
-            item['instrument_name'].lower(): item
+            item['instrument_name']: item for item in results}
-            for item in results
-        }
 
         if instrument is not None:
             return instruments[instrument]
@@ -325,18 +158,20 @@ class Client:
     async def search_symbols(
         self,
         pattern: str,
-        limit: int = 30,
+        limit: int = None,
     ) -> dict[str, Any]:
-        data = await self.symbol_info()
+        if self._pairs is not None:
+            data = self._pairs
+        else:
+            data = await self.symbol_info()
 
         matches = fuzzy.extractBests(
             pattern,
             data,
-            score_cutoff=35,
+            score_cutoff=50,
-            limit=limit
         )
         # repack in dict form
-        return {item[0]['instrument_name'].lower(): item[0]
+        return {item[0]['instrument_name']: item[0]
                 for item in matches}
 
     async def bars(
@@ -360,16 +195,19 @@ class Client:
         end_time = deribit_timestamp(end_dt)
 
         # https://docs.deribit.com/#public-get_tradingview_chart_data
-        resp = await self.json_rpc(
+        response = await self._api(
-            'public/get_tradingview_chart_data',
+            'get_tradingview_chart_data',
             params={
                 'instrument_name': instrument.upper(),
                 'start_timestamp': start_time,
                 'end_timestamp': end_time,
                 'resolution': '1'
-            })
+            }
+        )
 
-        result = KLinesResult(**resp.result)
+        klines = JSONRPCResult(**response)
+    
+        result = KLinesResult(**klines.result)
         new_bars = []
         for i in range(len(result.close)):
 
@@ -399,308 +237,19 @@ class Client:
         instrument: str,
         count: int = 10
     ):
-        resp = await self.json_rpc(
+        response = await self._api(
-            'public/get_last_trades_by_instrument',
+            'get_last_trades_by_instrument',
             params={
                 'instrument_name': instrument,
                 'count': count
-            })
+            }
+        )
 
-        return LastTradesResult(**resp.result)
+        return LastTradesResult(**(JSONRPCResult(**response).result))
 
 
 @acm
 async def get_client() -> Client:
-    async with (
+    client = Client()
-        trio.open_nursery() as n,
+    await client.cache_symbols()
-        open_autorecon_ws(_testnet_ws_url) as ws
+    yield client
-    ):
-
-        _rpc_id: Iterable = count(0)
-        _rpc_results: Dict[int, Dict] = {}
-
-        _expiry_time: int = float('inf')
-        _access_token: Optional[str] = None
-        _refresh_token: Optional[str] = None
-
-        def _next_json_body(method: str, params: Dict):
-            """get the typical json rpc 2.0 msg body and increment the req id
-            """
-            return {
-                'jsonrpc': '2.0',
-                'id': next(_rpc_id),
-                'method': method,
-                'params': params
-            }
-
-        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
-            """
-            msg = _next_json_body(method, params)
-            _id = msg['id']
-
-            _rpc_results[_id] = {
-                'result': None,
-                'event': trio.Event()
-            }
-
-            await ws.send_msg(msg)
-
-            await _rpc_results[_id]['event'].wait()
-
-            ret = _rpc_results[_id]['result']
-
-            del _rpc_results[_id]
-
-            if ret.error is not None:
-                raise Exception(json.dumps(ret.error, indent=4))
-
-            return ret
-
-        async def _recv_task():
-            """receives every ws message and stores it in its corresponding result
-            field, then sets the event to wakeup original sender tasks.
-            """
-            async for msg in ws:
-                msg = JSONRPCResult(**msg)
-
-                if msg.id not in _rpc_results:
-                    # in case this message wasn't beign accounted for store it 
-                    _rpc_results[msg.id] = {
-                        'result': None,
-                        'event': trio.Event()
-                    }
-
-                _rpc_results[msg.id]['result'] = msg
-                _rpc_results[msg.id]['event'].set()
-
-        client = Client(json_rpc)
-
-        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)
-
-        n.start_soon(_recv_task)
-        # 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
-
-
-@acm
-async def open_feed_handler():
-    fh = FeedHandler(config=get_config())
-    yield fh
-    await to_asyncio.run_task(fh.stop_async)
-
-
-@acm
-async def maybe_open_feed_handler() -> trio.abc.ReceiveStream:
-    async with maybe_open_context(
-        acm_func=open_feed_handler,
-        key='feedhandler',
-    ) as (cache_hit, fh):
-        yield fh
-
-
-@acm
-async def open_price_feed(
-    instrument: str
-) -> trio.abc.ReceiveStream:
-
-    # XXX: hangs when going into this ctx mngr
-    async with maybe_open_feed_handler() as fh:
-
-        async def relay(
-            from_trio: asyncio.Queue,
-            to_trio: trio.abc.SendChannel,
-        ) -> None:
-            async def _trade(data: dict, receipt_timestamp):
-                to_trio.send_nowait(('trade', {
-                    'symbol': cb_sym_to_deribit_inst(
-                        str_to_cb_sym(data.symbol)).lower(),
-                    'last': data,
-                    'broker_ts': time.time(),
-                    'data': data.to_dict(),
-                    'receipt': receipt_timestamp
-                }))
-
-            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)}
-                    ]
-                }))
-
-            fh.add_feed(
-                DERIBIT,
-                channels=[TRADES, L1_BOOK],
-                symbols=[instrument],
-                callbacks={
-                    TRADES: _trade,
-                    L1_BOOK: _l1
-                })
-
-            if not fh.running:
-                fh.run(
-                    start_loop=False,
-                    install_signal_handlers=False)
-
-            # sync with trio
-            to_trio.send_nowait(None)
-
-            try:
-                await asyncio.sleep(float('inf'))
-
-            except asyncio.exceptions.CancelledError:
-                ...
-
-        async with to_asyncio.open_channel_from(
-            relay
-        ) as (first, chan):
-            yield chan
-
-
-@acm
-async def maybe_open_price_feed(
-    instrument: str
-) -> trio.abc.ReceiveStream:
-
-    # TODO: add a predicate to maybe_open_context
-    async with maybe_open_context(
-        acm_func=open_price_feed,
-        kwargs={
-            'instrument': instrument
-        },
-        key=f'{instrument}-price',
-    ) as (cache_hit, feed):
-        if cache_hit:
-            yield broadcast_receiver(feed, 10)
-        else:
-            yield feed
-
-@acm
-async def open_order_feed(
-    instrument: List[str]
-) -> trio.abc.ReceiveStream:
-
-    async with maybe_open_feed_handler() as fh:
-
-        async def relay(
-            from_trio: asyncio.Queue,
-            to_trio: trio.abc.SendChannel,
-        ) -> None:
-            async def _fill(data: dict, receipt_timestamp):
-                breakpoint()
-
-            async def _order_info(data: dict, receipt_timestamp):
-                breakpoint()
-
-            fh.add_feed(
-                DERIBIT,
-                channels=[FILLS, ORDER_INFO],
-                symbols=[instrument],
-                callbacks={
-                    FILLS: _fill,
-                    ORDER_INFO: _order_info,
-                })
-
-            if not fh.running:
-                fh.run(
-                    start_loop=False,
-                    install_signal_handlers=False)
-
-            # sync with trio
-            to_trio.send_nowait(None)
-
-            try:
-                await asyncio.sleep(float('inf'))
-
-            except asyncio.exceptions.CancelledError:
-                ...
-
-        async with to_asyncio.open_channel_from(
-            relay
-        ) as (first, chan):
-            yield chan
-
-@acm
-async def maybe_open_order_feed(
-    instrument: str
-) -> trio.abc.ReceiveStream:
-
-    # TODO: add a predicate to maybe_open_context
-    async with maybe_open_context(
-        acm_func=open_order_feed,
-        kwargs={
-            'instrument': instrument
-        },
-        key=f'{instrument}-order',
-    ) as (cache_hit, feed):
-        if cache_hit:
-            yield broadcast_receiver(feed, 10)
-        else:
-            yield feed

piker/brokers/deribit/broker.py

@@ -1,44 +0,0 @@
-# piker: trading gear for hackers
-# Copyright (C) Guillermo Rodriguez (in stewardship for piker0)
-
-# 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/>.
-'''
-Order api and machinery
-
-'''
-
-from typing import Any, AsyncIterator
-
-import tractor
-
-
-@tractor.context
-async def trades_dialogue(
-    ctx: tractor.Context,
-    loglevel: str = None,
-
-) -> AsyncIterator[dict[str, Any]]:
-
-    # XXX: required to propagate ``tractor`` loglevel to piker logging
-    get_console_log(loglevel or tractor.current_actor().loglevel)
-
-    async with open_cached_client('deribit') as client:
-        if not client._key_id:
-            raise RuntimeError('Missing Deribit API key in `brokers.toml`!?!?')
-
-        acc_name = f'deribit.{client._key_id}'
-
-        await client.cache_symbols()
-
-        breakpoint()

piker/brokers/deribit/feed.py

@@ -1,6 +1,3 @@
-# piker: trading gear for hackers
-# Copyright (C) Guillermo Rodriguez (in stewardship for piker0)
-
 # 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
@@ -18,6 +15,9 @@
 Deribit backend.
 
 '''
+
+import asyncio
+from async_generator import aclosing
 from contextlib import asynccontextmanager as acm
 from datetime import datetime
 from typing import Any, Optional, List, Callable
@@ -29,7 +29,9 @@ import pendulum
 from fuzzywuzzy import process as fuzzy
 import numpy as np
 import tractor
+from tractor import to_asyncio
 
+from piker import config
 from piker._cacheables import open_cached_client
 from piker.log import get_logger, get_console_log
 from piker.data import ShmArray
@@ -45,21 +47,170 @@ from cryptofeed.defines import (
 )
 from cryptofeed.symbols import Symbol
 
-from .api import (
+from .api import Client, Trade
-    Client, Trade,
-    get_config,
-    str_to_cb_sym, piker_sym_to_cb_sym, cb_sym_to_deribit_inst,
-    maybe_open_price_feed
-)
 
 _spawn_kwargs = {
     'infect_asyncio': True,
 }
 
 
+def get_config() -> dict[str, Any]:
+
+    conf, path = config.load()
+
+    section = conf.get('deribit')
+
+    if section is None:
+        log.warning(f'No config section found for deribit in {path}')
+        return {}
+
+    conf['log'] = {}
+    conf['log']['disabled'] = True
+
+#    conf['log']['filename'] = '/tmp/feedhandler.log'
+#    conf['log']['level'] = 'WARNING'
+
+    return conf 
+
+
 log = get_logger(__name__)
 
 
+_url = 'https://www.deribit.com'
+
+
+def str_to_cb_sym(name: str) -> Symbol:
+    base, strike_price, expiry_date, option_type = name.split('-')
+
+    quote = base
+
+    if option_type == 'put':
+        option_type = PUT 
+    elif option_type  == 'call':
+        option_type = CALL
+    else:
+        raise BaseException("Couldn\'t parse option type")
+
+    return Symbol(
+        base, quote,
+        type=OPTION,
+        strike_price=strike_price,
+        option_type=option_type,
+        expiry_date=expiry_date,
+        expiry_normalize=False)
+
+
+
+def piker_sym_to_cb_sym(name: str) -> Symbol:
+    base, expiry_date, strike_price, option_type = tuple(
+        name.upper().split('-'))
+
+    quote = base
+
+    if option_type == 'P':
+        option_type = PUT 
+    elif option_type  == 'C':
+        option_type = CALL
+    else:
+        raise BaseException("Couldn\'t parse option type")
+
+    return Symbol(
+        base, quote,
+        type=OPTION,
+        strike_price=strike_price,
+        option_type=option_type,
+        expiry_date=expiry_date.upper())
+
+
+def cb_sym_to_deribit_inst(sym: Symbol):
+    # cryptofeed normalized
+    cb_norm = ['F', 'G', 'H', 'J', 'K', 'M', 'N', 'Q', 'U', 'V', 'X', 'Z']
+
+    # deribit specific 
+    months = ['JAN', 'FEB', 'MAR', 'APR', 'MAY', 'JUN', 'JUL', 'AUG', 'SEP', 'OCT', 'NOV', 'DEC']
+    
+    exp = sym.expiry_date
+
+    # YYMDD
+    # 01234
+    year, month, day = (
+        exp[:2], months[cb_norm.index(exp[2:3])], exp[3:])
+
+    otype = 'C' if sym.option_type == CALL else 'P'
+
+    return f'{sym.base}-{day}{month}{year}-{sym.strike_price}-{otype}'
+
+
+# inside here we are in an asyncio context
+async def open_aio_cryptofeed_relay(
+    from_trio: asyncio.Queue,
+    to_trio: trio.abc.SendChannel,
+    instruments: List[str] = []
+) -> None:
+
+    instruments = [piker_sym_to_cb_sym(i) for i in instruments]
+
+    async def trade_cb(data: dict, receipt_timestamp):
+        to_trio.send_nowait(('trade', {
+            'symbol': cb_sym_to_deribit_inst(
+                str_to_cb_sym(data.symbol)).lower(),
+            'last': data,
+            'broker_ts': time.time(),
+            'data': data.to_dict(),
+            'receipt': receipt_timestamp
+        }))
+
+    async def l1_book_cb(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)}
+            ]
+        }))
+
+    fh = FeedHandler(config=get_config())
+    fh.run(start_loop=False)
+
+    fh.add_feed(
+        DERIBIT,
+        channels=[L1_BOOK],
+        symbols=instruments,
+        callbacks={L1_BOOK: l1_book_cb})
+
+    fh.add_feed(
+        DERIBIT,
+        channels=[TRADES],
+        symbols=instruments,
+        callbacks={TRADES: trade_cb})
+
+    # sync with trio
+    to_trio.send_nowait(None)
+
+    await asyncio.sleep(float('inf'))
+
+
+@acm
+async def open_cryptofeeds(
+
+    instruments: List[str]
+
+) -> trio.abc.ReceiveStream:
+
+    async with to_asyncio.open_channel_from(
+        open_aio_cryptofeed_relay,
+        instruments=instruments,
+    ) as (first, chan):
+        yield chan
+
+
 @acm
 async def open_history_client(
     instrument: str,
@@ -88,7 +239,7 @@ async def open_history_client(
 
             start_dt = pendulum.from_timestamp(array[0]['time'])
             end_dt = pendulum.from_timestamp(array[-1]['time'])
-
+            
             return array, start_dt, end_dt
 
         yield get_ohlc, {'erlangs': 3, 'rate': 3}
@@ -127,7 +278,9 @@ async def stream_quotes(
 
     async with (
         open_cached_client('deribit') as client,
-        send_chan as send_chan
+        send_chan as send_chan,
+        trio.open_nursery() as n,
+        open_cryptofeeds(symbols) as stream 
     ):
 
         init_msgs = {
@@ -145,36 +298,26 @@ async def stream_quotes(
 
         nsym = piker_sym_to_cb_sym(sym)
 
-        async with maybe_open_price_feed(sym) as stream:
+        # keep client cached for real-time section
+        cache = await client.cache_symbols()
 
-            cache = await client.cache_symbols()
+        last_trade = Trade(**(await client.last_trades(
+            cb_sym_to_deribit_inst(nsym), count=1)).trades[0])
 
-            last_trades = (await client.last_trades(
+        first_quote = {
-                cb_sym_to_deribit_inst(nsym), count=1)).trades
+            'symbol': sym,
-
+            'last': last_trade.price,
-            if len(last_trades) == 0:
+            'brokerd_ts': last_trade.timestamp,
-                last_trade = None
+            'ticks': [{
-                async for typ, quote in stream:
+                'type': 'trade',
-                    if typ == 'trade':
+                'price': last_trade.price,
-                        last_trade = Trade(**(quote['data']))
+                'size': last_trade.amount,
-                        break
+                'broker_ts': last_trade.timestamp
-
+            }]
-            else:
+        }
-                last_trade = Trade(**(last_trades[0]))
+        task_status.started((init_msgs,  first_quote))
-
-            first_quote = {
-                'symbol': sym,
-                'last': last_trade.price,
-                'brokerd_ts': last_trade.timestamp,
-                'ticks': [{
-                    'type': 'trade',
-                    'price': last_trade.price,
-                    'size': last_trade.amount,
-                    'broker_ts': last_trade.timestamp
-                }]
-            }
-            task_status.started((init_msgs,  first_quote))
 
+        async with aclosing(stream):
             feed_is_live.set()
 
             async for typ, quote in stream:
@@ -195,6 +338,15 @@ async def open_symbol_search(
         async with ctx.open_stream() as stream:
 
             async for pattern in stream:
+                # results = await client.symbol_info(sym=pattern.upper())
+
+                matches = fuzzy.extractBests(
+                    pattern,
+                    cache,
+                    score_cutoff=30,
+                )
                 # repack in dict form
                 await stream.send(
-                    await client.search_symbols(pattern))
+                    {item[0]['instrument_name']: item[0]
+                     for item in matches}
+                )

piker/data/_web_bs.py

@@ -20,7 +20,7 @@ ToOlS fOr CoPInG wITh "tHE wEB" protocols.
 """
 from contextlib import asynccontextmanager, AsyncExitStack
 from types import ModuleType
-from typing import Any, Optional, Callable, AsyncGenerator
+from typing import Any, Callable, AsyncGenerator
 import json
 
 import trio
@@ -54,8 +54,8 @@ class NoBsWs:
         self,
         url: str,
         stack: AsyncExitStack,
-        fixture: Optional[Callable] = None,
+        fixture: Callable,
-        serializer: ModuleType = json
+        serializer: ModuleType = json,
     ):
         self.url = url
         self.fixture = fixture
@@ -80,14 +80,12 @@ class NoBsWs:
                 self._ws = await self._stack.enter_async_context(
                     trio_websocket.open_websocket_url(self.url)
                 )
+                # rerun user code fixture
+                ret = await self._stack.enter_async_context(
+                    self.fixture(self)
+                )
 
-                if self.fixture is not None:
+                assert ret is None
-                    # rerun user code fixture
-                    ret = await self._stack.enter_async_context(
-                        self.fixture(self)
-                    )
-
-                    assert ret is None
 
                 log.info(f'Connection success: {self.url}')
                 return self._ws
@@ -123,19 +121,13 @@ class NoBsWs:
             except self.recon_errors:
                 await self._connect()
 
-    def __aiter__(self):
-        return self
-
-    async def __anext__(self):
-        return await self.recv_msg()
-
 
 @asynccontextmanager
 async def open_autorecon_ws(
     url: str,
 
     # TODO: proper type annot smh
-    fixture: Optional[Callable] = None,
+    fixture: Callable,
 
 ) -> AsyncGenerator[tuple[...],  NoBsWs]:
     """Apparently we can QoS for all sorts of reasons..so catch em.

requirements.txt

@@ -20,4 +20,4 @@
 -e git+https://github.com/pikers/asyncvnc.git@main#egg=asyncvnc
 
 # ``cryptofeed`` for connecting to various crypto exchanges + custom fixes
--e git+https://github.com/pikers/cryptofeed.git@date_parsing#egg=cryptofeed
+-e git+https://github.com/guilledk/cryptofeed.git@date_parsing#egg=cryptofeed