piker/piker/brokers/deribit/venues.py

# 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,
    Optional,
)
from decimal import Decimal

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'

    # TODO, impl this without the MM:SS part of
    # the `'THH:MM:SS..'` etc..
    @property
    def expiry(self) -> str:
        iso_date = pendulum.from_timestamp(
            self.expiration_timestamp / 1000
        ).isoformat()
        return iso_date 

    @property
    def venue(self) -> str:
        return f'{self.instrument_type}_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