From 58d49b1a7277b8437179635ff402012db9d5a6bd Mon Sep 17 00:00:00 2001 From: Nelson Torres Date: Thu, 30 Jan 2025 01:38:37 -0300 Subject: [PATCH] Deribit api key changes introduce: - `get_timestamp_int`: added this is the hack, so we can aboid use the custom deribit date format. - `get_currencies`: added so we could get all deribit's available currencies. - `get_instruments`: for a especific expiration date, it return a list of criptofeed.Symbol. - `get_expiration_dates`: expirations dates available for btc's option contracts . - `get_strikes_dict`: all the strike prices for an especific expiration date. - `aio_open_interest_feed_relay` `open_oi_feed` `maybe_open_oi_feed`: this three handles all the portal stuff and the cryptofeed callbacks for the open interest and trades, for some reason it need both to work, i need to check that out at some point. - Also a couple of format fixes. --- piker/brokers/deribit/api.py | 250 ++++++++++++++++++++++++++++++++--- 1 file changed, 228 insertions(+), 22 deletions(-) diff --git a/piker/brokers/deribit/api.py b/piker/brokers/deribit/api.py index f846a5c0..0bd50fbb 100644 --- a/piker/brokers/deribit/api.py +++ b/piker/brokers/deribit/api.py @@ -52,12 +52,14 @@ 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 ( L1Book, Trade, + OpenInterest, ) from piker.brokers import SymbolNotFound from .venues import ( @@ -110,6 +112,10 @@ def deribit_timestamp(when: datetime) -> int: ) +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: base, strike_price, expiry_date, option_type = name.split('-') @@ -117,13 +123,14 @@ def str_to_cb_sym(name: str) -> Symbol: if 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 = get_values_from_cb_normalized_date(expiry_date) - + new_expiry_date: int = get_timestamp_int( + get_values_from_cb_normalized_date(expiry_date) + ) return Symbol( base=base, quote=quote, @@ -143,11 +150,12 @@ def piker_sym_to_cb_sym(name: str) -> Symbol: )= tuple( name.upper().split('-')) + new_expiry_date = get_timestamp_int(expiry_date) quote: str = base - if option_type == 'P': + if option_type == 'P' or option_type == 'PUT': option_type = PUT - elif option_type == 'C': + elif option_type == 'C' or option_type == 'CALL': option_type = CALL else: raise Exception("Couldn\'t parse option type") @@ -158,7 +166,7 @@ def piker_sym_to_cb_sym(name: str) -> Symbol: type=OPTION, strike_price=strike_price, option_type=option_type, - expiry_date=expiry_date + expiry_date=new_expiry_date ) @@ -226,16 +234,18 @@ def get_config() -> dict[str, Any]: ) conf_option = section.get('option', {}) - 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 + conf_log = conf_option.get('log', {}) + return { + 'deribit': { + 'key_id': conf_option['key_id'], + 'key_secret': conf_option['key_secret'], + }, + 'log': { + 'filename': conf_log['filename'], + 'level': conf_log['level'], + 'disabled': conf_log['disabled'], + } + } class Client: @@ -311,6 +321,20 @@ 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, @@ -323,11 +347,7 @@ class Client: ''' assets = {} - resp = await self._json_rpc_auth_wrapper( - 'public/get_currencies', - params={} - ) - currencies: list[dict] = resp.result + currencies = await self.get_currencies() for currency in currencies: name: str = currency['currency'] tx_tick: Decimal = digits_to_dec(currency['fee_precision']) @@ -359,6 +379,82 @@ class Client: return flat + async def get_instruments( + self, + currency: str = 'btc', + kind: str = 'option', + expired: bool = False, + expiry_date: str = None, + + ) -> list[Symbol]: + """ + Get instruments for cryptoFeed.FeedHandler. + """ + params: dict[str, str] = { + 'currency': currency.upper(), + 'kind': kind, + 'expired': expired, + } + + r: JSONRPCResult = await self._json_rpc_auth_wrapper( + 'public/get_instruments', + params, + ) + resp = r.result + response_list = [] + + for i in range(len(resp)): + element = resp[i] + name = f'{element["instrument_name"].split("-")[1]}' + if not expiry_date or name == expiry_date.upper(): + response_list.append(piker_sym_to_cb_sym(element['instrument_name'])) + + return response_list + + async def get_expiration_dates( + self, + currency: str = 'btc', + kind: str = 'option', + + ) -> list[str]: + """ + Get a dict with all expiration dates listed as value and currency as key. + """ + + params: dict[str, str] = { + 'currency': currency.upper(), + 'kind': kind, + } + + r: JSONRPCResult = await self._json_rpc_auth_wrapper( + 'public/get_expirations', + params, + ) + resp = r.result + + return resp[currency][kind] + + def get_strikes_dict( + self, + instruments: list[Symbol], + + ) -> dict[str, dict[str, Decimal | None]]: + """ + Get a dict with strike prices as keys. + """ + + response: dict[str, dict[str, Decimal | None]] = {} + + for i in range(len(instruments)): + element = instruments[i] + strike = f'{str(element).split('-')[1]}' + response[f'{strike}'] = { + 'C': None, + 'P': None, + } + + return response + async def submit_limit( self, symbol: str, @@ -738,6 +834,116 @@ async def maybe_open_price_feed( yield feed +async def aio_open_interest_feed_relay( + fh: FeedHandler, + instruments: list[Symbol], + from_trio: asyncio.Queue, + to_trio: trio.abc.SendChannel, +) -> None: + async def _trade( + trade: Trade, # cryptofeed, NOT ours from `.venues`! + receipt_timestamp: int, + ) -> None: + ''' + Proxy-thru `cryptofeed.FeedHandler` "trades" to `piker`-side. + + ''' + to_trio.send_nowait(('trade', trade)) + + # trade and oi are user defined functions that + # will be called when trade and open interest updates are received + # data type is not dict, is an object: cryptofeed.types.OpenINterest + async def _oi( + oi: OpenInterest, + receipt_timestamp: int, + ) -> None: + ''' + Proxy-thru `cryptofeed.FeedHandler` "oi" to `piker`-side. + + ''' + symbol: Symbol = str_to_cb_sym(oi.symbol) + piker_sym: str = cb_sym_to_deribit_inst(symbol) + ( + base, + expiry_date, + strike_price, + option_type + ) = tuple( + piker_sym.split('-') + ) + msg = { + 'timestamp': oi.timestamp, + 'strike_price': strike_price, + 'option_type': option_type, + 'open_interest': Decimal(oi.open_interest), + } + to_trio.send_nowait(('oi', msg)) + + + channels = [TRADES, OPEN_INTEREST] + callbacks={TRADES: _trade, OPEN_INTEREST: _oi} + + fh.add_feed( + DERIBIT, + channels=channels, + symbols=instruments, + callbacks=callbacks + ) + + if not fh.running: + fh.run( + start_loop=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_oi_feed( + instruments: list[Symbol], +) -> to_asyncio.LinkedTaskChannel: + + fh: FeedHandler + first: None + chan: to_asyncio.LinkedTaskChannel + async with ( + maybe_open_feed_handler() as fh, + to_asyncio.open_channel_from( + partial( + aio_open_interest_feed_relay, + fh, + instruments, + ) + ) as (first, chan) + ): + yield chan + + +@acm +async def maybe_open_oi_feed( + instruments: list[Symbol], +) -> trio.abc.ReceiveStream: + + # TODO: add a predicate to maybe_open_context + feed: to_asyncio.LinkedTaskChannel + async with maybe_open_context( + acm_func=open_oi_feed, + kwargs={ + 'instruments': instruments + }, + key=f'{instruments[0].base}', + + ) as (cache_hit, feed): + if cache_hit: + yield broadcast_receiver(feed, 10) + else: + yield feed + # TODO, move all to `.broker` submod! # async def aio_order_feed_relay(