Исходный код
Ниже представлен готовый к использованию класс BybitAuthClient, реализующий генерацию подписи и отправку GET и POST запросов к приватным эндпоинтам Bybit V5. При построении высоконагруженных торговых систем часто требуется интеграция с брокерами сообщений, например, интеграция Kafka для обработки потоковых тиковых данных на Python, что позволяет эффективно обрабатывать рыночные события параллельно с отправкой ордеров.
import time
import hmac
import hashlib
import requests
import json
class BybitAuthClient:
def __init__(self, api_key: str, api_secret: str, base_url: str = 'https://api.bybit.com'):
self.api_key = api_key
self.api_secret = api_secret
self.base_url = base_url
def _generate_signature(self, timestamp: str, recv_window: str, payload: str) -> str:
param_str = timestamp + self.api_key + recv_window + payload
return hmac.new(
self.api_secret.encode('utf-8'),
param_str.encode('utf-8'),
hashlib.sha256
).hexdigest()
def _get_headers(self, timestamp: str, recv_window: str, signature: str) -> dict:
return {
'X-BAPI-API-KEY': self.api_key,
'X-BAPI-SIGN': signature,
'X-BAPI-SIGN-TYPE': '2',
'X-BAPI-TIMESTAMP': timestamp,
'X-BAPI-RECV-WINDOW': recv_window,
'Content-Type': 'application/json'
}
def send_request(self, method: str, endpoint: str, params: dict = None) -> dict:
timestamp = str(int(time.time() * 1000))
recv_window = '5000'
method = method.upper()
url = f'{self.base_url}{endpoint}'
if method == 'GET':
query_string = ''
if params:
query_string = '&'.join([f'{k}={v}' for k, v in sorted(params.items())])
signature = self._generate_signature(timestamp, recv_window, query_string)
headers = self._get_headers(timestamp, recv_window, signature)
response = requests.get(url, params=params, headers=headers)
elif method == 'POST':
payload_str = json.dumps(params) if params else ''
signature = self._generate_signature(timestamp, recv_window, payload_str)
headers = self._get_headers(timestamp, recv_window, signature)
response = requests.post(url, data=payload_str, headers=headers)
else:
raise ValueError(f'Unsupported HTTP method: {method}')
return response.json()
if __name__ == '__main__':
API_KEY = 'your_api_key'
API_SECRET = 'your_api_secret'
client = BybitAuthClient(api_key=API_KEY, api_secret=API_SECRET)
wallet_balance = client.send_request('GET', '/v5/account/wallet-balance', {'accountType': 'UNIFIED'})
print('Wallet Balance:', wallet_balance) Разбор параметров
X-BAPI-API-KEY: Публичный API-ключ, полученный в личном кабинете биржи Bybit.X-BAPI-TIMESTAMP: Текущее время в миллисекундах (Unix timestamp), используемое для предотвращения атак повторного воспроизведения (replay attacks).X-BAPI-RECV-WINDOW: Окно приема запроса в миллисекундах. Определяет, в течение какого времени после генерации таймстампа запрос считается валидным. По умолчанию рекомендуется использовать5000.X-BAPI-SIGN: Результирующая подпись HMAC SHA256, закодированная в hex-формате.payload: Строка параметров. Для GET-запросов это отсортированная query-строка (например,accountType=UNIFIED), для POST-запросов — сериализованный JSON-объект.
Как запустить
Для запуска скрипта вам понадобится установленная библиотека requests. Установите её с помощью команды pip install requests.
Замените значения переменных API_KEY и API_SECRET на ваши реальные ключи от тестовой сети (Testnet) или основной сети Bybit. Обратите внимание, что при работе с реальными торговыми сессиями и частыми запросами к API бирж могут возникать ограничения по IP. Чтобы избежать подобных проблем, ознакомьтесь с материалом о том, как обойти блокировки по IP при парсинге стакана Raydium на Python, принципы которого применимы и для Bybit.
После успешной настройки авторизации вы сможете автоматизировать свои торговые стратегии. Например, вы можете реализовать скрипт автоматического ребалансирования криптовалютного портфеля на Python через CCXT или напрямую через REST-клиент Bybit.




