Автоматическое Дельта-Хеджирование на Deribit с Python: Гайд для Квантов

import requests
import time
import math
import logging
import sys

# --- Настройка логирования ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- Конфигурация --- 
API_KEY = "YOUR_API_KEY" # Ваш API ключ Deribit
API_SECRET = "YOUR_API_SECRET" # Ваш API секрет Deribit
TESTNET = True # Установите False для реальной торговли
INSTRUMENT_CURRENCY = "BTC" # Валюта портфеля (например, "BTC" или "ETH")
HEDGE_THRESHOLD_DELTA = 0.05 # Абсолютное значение дельты, при котором запускается хеджирование (например, 0.05 BTC)
HEDGE_INTERVAL_SECONDS = 30 # Как часто проверять и хеджировать дельту (в секундах)
HEDGE_FUTURES_INSTRUMENT = f"{INSTRUMENT_CURRENCY}-PERPETUAL" # Инструмент фьючерсов для хеджирования (например, BTC-PERPETUAL)
HEDGE_ORDER_TYPE = "market" # Тип ордера для хеджирования: "market" или "limit"
HEDGE_LIMIT_PRICE_OFFSET_BPS = 5 # Смещение цены для лимитных ордеров в базисных пунктах (например, 5 bps = 0.05%)
HEDGE_AMOUNT_PRECISION = 4 # Точность округления для объема хеджирования (например, 0.0001 BTC)

class DeribitClient:
    """
    Упрощенный клиент для взаимодействия с Deribit API (HTTP).
    Обрабатывает аутентификацию через OAuth2 (client_credentials) и отправку запросов.
    """
    def __init__(self, client_id, client_secret, testnet=True):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = "https://test.deribit.com/api/v2/private/" if testnet else "https://www.deribit.com/api/v2/private/"
        self.public_url = "https://test.deribit.com/api/v2/public/" if testnet else "https://www.deribit.com/api/v2/public/"
        self.access_token = None
        self.refresh_token = None
        self.expires_in = 0
        self.last_auth_time = 0

    def _send_request(self, method, endpoint, params=None, is_private=True):
        headers = {'Content-Type': 'application/json'}
        url_prefix = self.base_url if is_private else self.public_url
        full_url = url_prefix + endpoint

        if is_private:
            # Обновляем токен, если он отсутствует или истекает в течение 60 секунд
            if not self.access_token or time.time() > self.last_auth_time + self.expires_in - 60:
                self._get_access_token()
            if not self.access_token:
                logging.error("Не удалось получить токен доступа. Невозможно выполнить приватный API-вызов.")
                return None
            headers['Authorization'] = f'Bearer {self.access_token}'

        try:
            if method == 'GET':
                response = requests.get(full_url, headers=headers, params=params)
            elif method == 'POST':
                response = requests.post(full_url, headers=headers, json=params)
            else:
                raise ValueError("Неподдерживаемый HTTP-метод")
            
            response.raise_for_status() # Вызывает HTTPError для плохих ответов (4xx или 5xx)
            data = response.json()
            if data.get('error'):
                logging.error(f"Ошибка Deribit API для {endpoint}: {data['error']}")
                return None
            return data
        except requests.exceptions.RequestException as e:
            logging.error(f"Ошибка запроса API для {endpoint}: {e}")
            return None
        except Exception as e:
            logging.error(f"Неожиданная ошибка при запросе API для {endpoint}: {e}")
            return None

    def _get_access_token(self):
        """
        Получает или обновляет токен доступа OAuth2.
        """
        auth_url = self.public_url.replace('/private/', '/public/') + "auth"
        params = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }
        try:
            response = requests.post(auth_url, json=params)
            response.raise_for_status()
            data = response.json()
            if data and data.get('result'):
                self.access_token = data['result']['access_token']
                self.refresh_token = data['result']['refresh_token']
                self.expires_in = data['result']['expires_in']
                self.last_auth_time = time.time()
                logging.info("Успешная аутентификация с Deribit.")
            else:
                logging.error(f"Ошибка аутентификации: {data.get('error', 'Неизвестная ошибка')}")
                self.access_token = None
        except requests.exceptions.RequestException as e:
            logging.error(f"Ошибка запроса аутентификации: {e}")
            self.access_token = None
        except Exception as e:
            logging.error(f"Неожиданная ошибка при аутентификации: {e}")
            self.access_token = None

    def get_account_summary(self, currency):
        """ Получает сводку по счету. """
        return self._send_request('GET', 'get_account_summary', params={'currency': currency, 'extended': True})

    def get_positions(self, currency, kind='option'):
        """ Получает открытые позиции (опционы или фьючерсы). """
        return self._send_request('GET', 'get_positions', params={'currency': currency, 'kind': kind})

    def get_index_price(self, index_name):
        """ Получает цену индекса. """
        return self._send_request('GET', 'get_index_price', params={'index_name': index_name}, is_private=False)

    def buy(self, instrument_name, amount, type='market', price=None):
        """ Размещает ордер на покупку. """
        params = {
            "instrument_name": instrument_name,
            "amount": amount,
            "type": type
        }
        if price is not None:
            params['price'] = price
        return self._send_request('POST', 'buy', params)

    def sell(self, instrument_name, amount, type='market', price=None):
        """ Размещает ордер на продажу. """
        params = {
            "instrument_name": instrument_name,
            "amount": amount,
            "type": type
        }
        if price is not None:
            params['price'] = price
        return self._send_request('POST', 'sell', params)

    def get_order_book(self, instrument_name):
        """ Получает стакан цен для инструмента. """
        return self._send_request('GET', 'get_order_book', params={'instrument_name': instrument_name}, is_private=False)

    def get_instruments(self, currency, kind='future'):
        """ Получает список инструментов. """
        return self._send_request('GET', 'get_instruments', params={'currency': currency, 'kind': kind, 'expired': False}, is_private=False)

def calculate_portfolio_delta(client, currency):
    """
    Рассчитывает общую дельту портфеля опционов и фьючерсов.
    """
    total_delta = 0.0

    # 1. Получаем позиции по опционам
    options_data = client.get_positions(currency, kind='option')
    if options_data and options_data.get('result'):
        for pos in options_data['result']:
            # Дельта позиции Deribit уже выражена в базовом активе (например, BTC)
            total_delta += pos['delta']
            logging.info(f"Позиция опциона: {pos['instrument_name']}, Размер: {pos['size']}, Дельта: {pos['delta']}")
    else:
        logging.warning(f"Не удалось получить позиции опционов для {currency}.")

    # 2. Получаем позиции по фьючерсам (конкретно хеджирующий инструмент)
    futures_data = client.get_positions(currency, kind='future')
    if futures_data and futures_data.get('result'):
        for pos in futures_data['result']:
            if pos['instrument_name'] == HEDGE_FUTURES_INSTRUMENT:
                # Дельта фьючерса равна его размеру (1 фьючерс = 1 дельта базового актива)
                # Размер фьючерса Deribit выражен в базовом активе.
                total_delta += pos['size']
                logging.info(f"Позиция фьючерса: {pos['instrument_name']}, Размер: {pos['size']}, Дельта: {pos['size']}")
    else:
        logging.warning(f"Не удалось получить позиции фьючерсов для {currency}.")

    return total_delta

def get_hedge_price(client, instrument_name, order_type, offset_bps):
    """
    Получает соответствующую цену для хеджирующего ордера (рыночного или лимитного).
    """
    if order_type == "market":
        return None # Рыночные ордера не требуют указания цены
    
    order_book = client.get_order_book(instrument_name)
    if not order_book or not order_book.get('result'):
        logging.error(f"Не удалось получить стакан цен для {instrument_name}.")
        return None

    bids = order_book['result']['bids']
    asks = order_book['result']['asks']

    if not bids or not asks:
        logging.error(f"Стакан цен для {instrument_name} пуст.")
        return None

    best_bid = bids[0][0]
    best_ask = asks[0][0]
    
    return {'best_bid': best_bid, 'best_ask': best_ask}

def hedge_portfolio(client):
    """
    Основная функция для расчета дельты и размещения хеджирующих ордеров.
    """
    logging.info("--- Запуск цикла дельта-хеджирования ---")

    total_delta = calculate_portfolio_delta(client, INSTRUMENT_CURRENCY)
    logging.info(f"Текущая общая дельта портфеля для {INSTRUMENT_CURRENCY}: {total_delta:.{HEDGE_AMOUNT_PRECISION}f}")

    if abs(total_delta) < HEDGE_THRESHOLD_DELTA:
        logging.info(f"Дельта портфеля ({total_delta:.{HEDGE_AMOUNT_PRECISION}f}) находится в пределах порога ({HEDGE_THRESHOLD_DELTA}). Хеджирование не требуется.")
        return

    hedge_amount = -total_delta # Объем фьючерсов для торговли, чтобы нейтрализовать дельту
    hedge_amount = round(hedge_amount, HEDGE_AMOUNT_PRECISION) # Округляем до точности инструмента

    if hedge_amount == 0:
        logging.info("Рассчитанный объем хеджирования равен нулю после округления. Хеджирование не требуется.")
        return

    logging.info(f"Требуется хеджирование: {hedge_amount:.{HEDGE_AMOUNT_PRECISION}f} {HEDGE_FUTURES_INSTRUMENT}")

    order_price = None
    if HEDGE_ORDER_TYPE == "limit":
        hedge_price_info = get_hedge_price(client, HEDGE_FUTURES_INSTRUMENT, HEDGE_ORDER_TYPE, HEDGE_LIMIT_PRICE_OFFSET_BPS)
        if not hedge_price_info:
            logging.error("Не удалось получить цену для лимитного ордера. Пропускаем хеджирование.")
            return
        
        if hedge_amount > 0: # Нужно купить фьючерсы (для уменьшения отрицательной дельты)
            # Размещаем лимитный ордер на покупку немного ниже лучшего аска, чтобы получить лучшую цену
            order_price = hedge_price_info['best_ask'] * (1 - HEDGE_LIMIT_PRICE_OFFSET_BPS / 10000)
        else: # Нужно продать фьючерсы (для уменьшения положительной дельты)
            # Размещаем лимитный ордер на продажу немного выше лучшего бида, чтобы получить лучшую цену
            order_price = hedge_price_info['best_bid'] * (1 + HEDGE_LIMIT_PRICE_OFFSET_BPS / 10000)
        order_price = round(order_price, 2) # Цены Deribit обычно имеют 2 знака после запятой для фьючерсов

    order_result = None
    if hedge_amount > 0: # Покупаем фьючерсы
        logging.info(f"Размещение ордера BUY на {hedge_amount:.{HEDGE_AMOUNT_PRECISION}f} {HEDGE_FUTURES_INSTRUMENT} по {'рыночной цене' if HEDGE_ORDER_TYPE == 'market' else order_price}...")
        order_result = client.buy(HEDGE_FUTURES_INSTRUMENT, abs(hedge_amount), type=HEDGE_ORDER_TYPE, price=order_price)
    else: # Продаем фьючерсы
        logging.info(f"Размещение ордера SELL на {abs(hedge_amount):.{HEDGE_AMOUNT_PRECISION}f} {HEDGE_FUTURES_INSTRUMENT} по {'рыночной цене' if HEDGE_ORDER_TYPE == 'market' else order_price}...")
        order_result = client.sell(HEDGE_FUTURES_INSTRUMENT, abs(hedge_amount), type=HEDGE_ORDER_TYPE, price=order_price)

    if order_result and order_result.get('result'):
        logging.info(f"Ордер хеджирования успешно размещен: {order_result['result']['order']['order_id']}")
        logging.info(f"Детали ордера: {order_result['result']['order']}")
    else:
        logging.error(f"Не удалось разместить ордер хеджирования. Результат: {order_result}")

def main():
    if API_KEY == "YOUR_API_KEY" or API_SECRET == "YOUR_API_SECRET":
        logging.error("Пожалуйста, установите ваши API_KEY и API_SECRET Deribit в скрипте.")
        sys.exit(1)

    client = DeribitClient(API_KEY, API_SECRET, testnet=TESTNET)

    while True:
        try:
            hedge_portfolio(client)
        except Exception as e:
            logging.error(f"Произошла ошибка во время цикла хеджирования: {e}", exc_info=True)
        
        logging.info(f"Ожидание {HEDGE_INTERVAL_SECONDS} секунд перед следующим циклом хеджирования...")
        time.sleep(HEDGE_INTERVAL_SECONDS)

if __name__ == "__main__":
    # Убедитесь, что библиотека requests установлена
    try:
        import requests
    except ImportError:
        logging.error("Библиотека 'requests' не установлена. Пожалуйста, установите ее с помощью 'pip install requests'.")
        sys.exit(1)
    
    main()