Only pass `is_brokercheck: bool` to endpoints that declare it

piker/brokers/cli.py

@@ -21,11 +21,15 @@ import os
 from functools import partial
 from operator import attrgetter
 from operator import itemgetter
+from typing import (
+    Any,
+)
 
 import click
 import trio
 import tractor
 
+# from .._daemon import maybe_open_runtime
 from ..cli import cli
 from .. import watchlists as wl
 from ..log import get_console_log, colorize_json, get_logger
@@ -39,6 +43,142 @@ _config_dir = click.get_app_dir('piker')
 _watchlists_data_path = os.path.join(_config_dir, 'watchlists.json')
 
 
+@cli.command()
+@click.argument('broker', nargs=1, required=True)
+@click.pass_obj
+def brokercheck(config, broker):
+    '''
+    Test broker apis for completeness.
+
+    '''
+    OK = '\033[92m'
+    WARNING = '\033[93m'
+    FAIL = '\033[91m'
+    ENDC = '\033[0m'
+
+    def print_ok(s: str, **kwargs):
+        print(OK + s + ENDC, **kwargs)
+
+    def print_error(s: str, **kwargs):
+        print(FAIL + s + ENDC, **kwargs)
+
+    async def run_method(
+        client,
+        meth_name: str,
+        **kwargs,
+
+    ) -> Any:
+        print(f'checking client for method \'{meth_name}\'...', end='', flush=True)
+        method = getattr(client, meth_name, None)
+        assert method, f'.{meth_name} does not exist for {client}!'
+        print_ok('found!, running...', end='', flush=True)
+        result = await method(**kwargs)
+        print_ok(f'done! result: {type(result)}')
+        return result
+
+    async def run_test(broker_name: str):
+
+        brokermod = get_brokermod(broker_name)
+        total = 0
+        passed = 0
+        failed = 0
+
+        print('getting client...', end='', flush=True)
+        if not hasattr(brokermod, 'get_client'):
+            print_error('fail! no \'get_client\' context manager found.')
+            return
+
+        # extra_tractor_kwargs = getattr(
+        #     brokermod,
+        #     '_spawn_kwargs',
+        #     {},
+        # )
+
+        # TODO: eventually avoid this hack for `ib` XD
+        import inspect
+        get_client = brokermod.get_client
+        if 'is_brokercheck' in inspect.signature(get_client).parameters:
+            kwargs = {'is_brokercheck': True}
+        else:
+            kwargs = {}
+
+        async with (
+            # TODO: in theory we can actually spawn a local `brokerd`
+            # and then try to make some basic feed queries?
+            # maybe_open_runtime(**extra_tractor_kwargs),
+            brokermod.get_client(**kwargs) as client,
+        ):
+            print_ok('done! inside client context.')
+
+            # check for methods present on brokermod
+            method_list = [
+                # not required eps i'm pretty sure?
+                # 'backfill_bars',
+                # 'stream_messages',
+
+                'open_history_client',
+                'stream_quotes',
+                'open_symbol_search',
+                'trades_dialogue',
+            ]
+
+            for method in method_list:
+                print(
+                    f"checking brokermod for method '{method}'...",
+                    end='',
+                    flush=True,
+                )
+                if not hasattr(brokermod, method):
+                    print_error(f"fail! method '{method}' not found.")
+                    failed += 1
+                else:
+                    print_ok('done!')
+                    passed += 1
+
+                total += 1
+
+            # check for methods present con brokermod.Client and attempt
+            # to use them and gather output results.
+
+            symbol_info = getattr(client, 'symbol_info', None)
+            if symbol_info:
+                syms = await run_method(
+                    client,
+                    'symbol_info',
+                )
+
+                total += 1
+
+                if len(syms) == 0:
+                    raise BaseException('Empty Symbol list?')
+
+                passed += 1
+
+                first_sym = tuple(syms.keys())[0]
+
+                method_list = [
+                    ('cache_symbols', {}),
+                    ('search_symbols', {'pattern': first_sym[:-1]}),
+                    ('bars', {'symbol': first_sym})
+                ]
+
+                for method_name, method_kwargs in method_list:
+                    try:
+                        await run_method(client, method_name, **method_kwargs)
+                        passed += 1
+
+                    except AssertionError:
+                        print_error(f'fail! method \'{method_name}\' not found.')
+                        failed += 1
+
+                    total += 1
+
+            print(f'total: {total}, passed: {passed}, failed: {failed}')
+
+    trio.run(run_test, broker)
+
+
+
 @cli.command()
 @click.option('--keys', '-k', multiple=True,
               help='Return results only for these keys')
@@ -193,6 +333,8 @@ def contracts(ctx, loglevel, broker, symbol, ids):
     brokermod = get_brokermod(broker)
     get_console_log(loglevel)
 
+
+
     contracts = trio.run(partial(core.contracts, brokermod, symbol))
     if not ids:
         # just print out expiry dates which can be used with

piker/brokers/deribit.py

@@ -0,0 +1,483 @@
+# 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/>.
+
+"""
+Deribit backend
+
+"""
+import asyncio
+from contextlib import asynccontextmanager as acm
+from datetime import datetime
+from typing import (
+    Any, Union, Optional, List,
+    AsyncGenerator, Callable,
+)
+import time
+
+import trio
+from trio_typing import TaskStatus
+import pendulum
+import asks
+from fuzzywuzzy import process as fuzzy
+import numpy as np
+import tractor
+from tractor import to_asyncio
+from pydantic.dataclasses import dataclass
+from pydantic import BaseModel
+import wsproto
+
+from .. import config
+from .._cacheables import open_cached_client
+from ._util import resproc, SymbolNotFound
+from ..log import get_logger, get_console_log
+from ..data import ShmArray
+from ..data._web_bs import open_autorecon_ws, NoBsWs
+
+
+from cryptofeed import FeedHandler
+
+from cryptofeed.callback import (
+    L1BookCallback,
+    TradeCallback
+)
+from cryptofeed.defines import (
+    DERIBIT, L1_BOOK, TRADES, OPTION, CALL, PUT
+)
+from cryptofeed.symbols import Symbol
+
+_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']['filename'] = 'feedhandler.log'
+    conf['log']['level'] = 'WARNING'
+
+    return conf 
+
+
+log = get_logger(__name__)
+
+
+_url = 'https://www.deribit.com'
+
+
+# Broker specific ohlc schema (rest)
+_ohlc_dtype = [
+    ('index', int),
+    ('time', int),
+    ('open', float),
+    ('high', float),
+    ('low', float),
+    ('close', float),
+    ('volume', float),
+    ('bar_wap', float),  # will be zeroed by sampler if not filled
+]
+
+class KLinesResult(BaseModel):
+    close: List[float]
+    cost: List[float]
+    high: List[float]
+    low: List[float]
+    open: List[float]
+    status: str
+    ticks: List[int]
+    volume: List[float]
+
+class KLines(BaseModel):
+    jsonrpc: str = '2.0'
+    result: KLinesResult
+    usIn: int 
+    usOut: int 
+    usDiff: int 
+    testnet: bool
+
+
+# convert datetime obj timestamp to unixtime in milliseconds
+def deribit_timestamp(when):
+    return int((when.timestamp() * 1000) + (when.microsecond / 1000))
+
+
+class Client:
+
+    def __init__(self) -> None:
+        self._sesh = asks.Session(connections=4)
+        self._sesh.base_location = _url
+        self._pairs: dict[str, Any] = {}
+
+    async def _api(
+        self,
+        method: str,
+        params: dict,
+    ) -> dict[str, Any]:
+        resp = await self._sesh.get(
+            path=f'/api/v2/public/{method}',
+            params=params,
+            timeout=float('inf')
+        )
+        return resproc(resp, log)
+
+    async def symbol_info(
+        self,
+        instrument: Optional[str] = None,
+        currency: str = 'btc',  # BTC, ETH, SOL, USDC
+        kind: str = 'option',
+        expired: bool = False
+    ) -> dict[str, Any]:
+        '''Get symbol info for the exchange.
+
+        '''
+        # TODO: we can load from our self._pairs cache
+        # on repeat calls...
+
+        # will retrieve all symbols by default
+        params = {
+            'currency': currency.upper(),
+            'kind': kind,
+            'expired': str(expired).lower()
+        }
+
+        resp = await self._api(
+            'get_instruments', params=params)
+
+        results = resp['result']
+
+        instruments = {
+            item['instrument_name']: item for item in results}
+
+        if instrument is not None:
+            return instruments[instrument]
+        else:
+            return instruments
+
+    async def cache_symbols(
+        self,
+    ) -> dict:
+        if not self._pairs:
+            self._pairs = await self.symbol_info()
+
+        return self._pairs
+
+    async def search_symbols(
+        self,
+        pattern: str,
+        limit: int = None,
+    ) -> dict[str, Any]:
+        if self._pairs is not None:
+            data = self._pairs
+        else:
+            data = await self.symbol_info()
+
+        matches = fuzzy.extractBests(
+            pattern,
+            data,
+            score_cutoff=50,
+        )
+        # repack in dict form
+        return {item[0]['instrument_name']: item[0]
+                for item in matches}
+
+    async def bars(
+        self,
+        symbol: str,
+        start_dt: Optional[datetime] = None,
+        end_dt: Optional[datetime] = None,
+        limit: int = 1000,
+        as_np: bool = True,
+    ) -> dict:
+        instrument = symbol
+
+        if end_dt is None:
+            end_dt = pendulum.now('UTC')
+
+        if start_dt is None:
+            start_dt = end_dt.start_of(
+                'minute').subtract(minutes=limit)
+
+        start_time = deribit_timestamp(start_dt)
+        end_time = deribit_timestamp(end_dt)
+
+        # https://docs.deribit.com/#public-get_tradingview_chart_data
+        response = await self._api(
+            'get_tradingview_chart_data',
+            params={
+                'instrument_name': instrument.upper(),
+                'start_timestamp': start_time,
+                'end_timestamp': end_time,
+                'resolution': '1'
+            }
+        )
+
+        klines = KLines(**response)
+    
+        result = klines.result
+        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]
+            ]
+
+            new_bars.append((i,) + tuple(row))
+
+        array = np.array(
+            [i, ], dtype=_ohlc_dtype) if as_np else klines
+        return array
+
+
+@acm
+async def get_client() -> Client:
+    client = Client()
+    await client.cache_symbols()
+    yield client
+
+
+# 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:
+
+    conf = get_config()
+    
+    def format_sym(name: str) -> str:
+        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("Instrument name must end in 'c' for calls or 'p' for puts")
+
+        return Symbol(
+            base, quote,
+            type=OPTION,
+            strike_price=strike_price,
+            option_type=option_type,
+            expiry_date=expiry_date.upper()).normalized 
+
+    instruments = [format_sym(i) for i in instruments]
+
+    async def trade_cb(data: dict, receipt_timestamp):
+        breakpoint()
+        # to_trio.send_nowait(('trade', {
+        #     'symbol': 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': 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=conf)
+    fh.run(start_loop=False)
+
+    fh.add_feed(
+        DERIBIT,
+        channels=[L1_BOOK, TRADES],
+        symbols=instruments,
+        callbacks={
+            L1_BOOK: L1BookCallback(l1_book_cb),
+            TRADES: TradeCallback(trade_cb)
+        })
+
+    # sync with trio
+    to_trio.send_nowait(None)
+
+    await from_trio.get()
+
+
+async def open_cryptofeeds(
+    instruments: List[str],
+    to_chart: trio.abc.SendChannel,
+
+    # startup sync
+    task_status: TaskStatus[tuple[dict, dict]] = trio.TASK_STATUS_IGNORED,
+):
+    async with to_asyncio.open_channel_from(
+        open_aio_cryptofeed_relay,
+        instruments=instruments,
+    ) as (first, chan):
+        assert first is None
+
+        await chan.send(None)
+
+        async with chan.subscribe() as msg_stream:
+            task_status.started()
+            async for msg in msg_stream:
+                await to_chart.send(msg) 
+
+
+@acm
+async def open_history_client(
+    instrument: str,
+) -> tuple[Callable, int]:
+
+    # TODO implement history getter for the new storage layer.
+    async with open_cached_client('deribit') as client:
+
+        async def get_ohlc(
+            end_dt: Optional[datetime] = None,
+            start_dt: Optional[datetime] = None,
+
+        ) -> tuple[
+            np.ndarray,
+            datetime,  # start
+            datetime,  # end
+        ]:
+
+            array = await client.bars(
+                instrument,
+                start_dt=start_dt,
+                end_dt=end_dt,
+            )
+            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}
+
+
+async def backfill_bars(
+    symbol: str,
+    shm: ShmArray,  # type: ignore # noqa
+    task_status: TaskStatus[trio.CancelScope] = trio.TASK_STATUS_IGNORED,
+) -> None:
+    """Fill historical bars into shared mem / storage afap.
+    """
+    instrument = symbol
+    with trio.CancelScope() as cs:
+        async with open_cached_client('deribit') as client:
+            bars = await client.bars(instrument)
+            shm.push(bars)
+            task_status.started(cs)
+
+
+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
+    get_console_log(loglevel or tractor.current_actor().loglevel)
+
+    sym = symbols[0]
+    to_chart, from_feed = trio.open_memory_channel(1)
+
+    async with (
+        open_cached_client('deribit') as client,
+        send_chan as send_chan,
+        trio.open_nursery() as n
+    ):
+        await n.start(
+            open_cryptofeeds, symbols, to_chart)
+
+        init_msgs = {
+            # pass back token, and bool, signalling if we're the writer
+            # and that history has been written
+            sym: {
+                'symbol_info': {},
+                'shm_write_opts': {'sum_tick_vml': False},
+                'fqsn': sym,
+            },
+        }
+
+        # keep client cached for real-time section
+        cache = await client.cache_symbols()
+
+        async with from_feed:
+            typ, quote = await anext(from_feed)
+
+            while typ != 'trade':
+                typ, quote = await anext(from_feed)
+
+            task_status.started((init_msgs,  quote))
+
+            async for typ, msg in from_feed:
+                topic = msg['symbol']
+                await send_chan.send({topic: msg}) 
+
+
+@tractor.context
+async def open_symbol_search(
+    ctx: tractor.Context,
+) -> Client:
+    async with open_cached_client('deribit') as client:
+
+        # load all symbols locally for fast search
+        cache = await client.cache_symbols()
+        await ctx.started()
+
+        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=50,
+                )
+                # repack in dict form
+                await stream.send(
+                    {item[0]['instrument_name']: item[0]
+                     for item in matches}
+                )

piker/brokers/ib.py

@@ -1353,6 +1353,7 @@ async def open_client_proxy(
 
 @acm
 async def get_client(
+    is_brokercheck: bool = False,
     **kwargs,
 
 ) -> Client:
@@ -1361,6 +1362,10 @@ async def get_client(
     a method proxy to it.
 
     '''
+    if is_brokercheck:
+        yield Client
+        return
+
     # TODO: the IPC via portal relay layer for when this current
     # actor isn't in aio mode.
     async with open_data_client() as proxy:

setup.py

@@ -61,6 +61,7 @@ setup(
         # brokers
         'asks==2.4.8',
         'ib_insync',
+        'cryptofeed',
 
         # numerics
         'pendulum', # easier datetimes