From 005d9c34b6187e2fd0459c1a5c97cc555a6ada17 Mon Sep 17 00:00:00 2001
From: Nelson Torres <nelson.torres.alvarado1@gmail.com>
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(