Исходный код
Пинг-понг (Ping-Pong) — это классическая маркет-мейкинговая стратегия, основанная на выставлении лимитных ордеров по обе стороны спреда или на фиксированных уровнях поддержки и сопротивления. Главная задача квант-разработчика при реализации такой стратегии — минимизировать транзакционные издержки (комиссии биржи). На бирже Bybit V5 для этого используется тип выполнения PostOnly.
Опция PostOnly гарантирует, что ваш ордер будет добавлен в стакан как мейкер (Maker) и не исполнится мгновенно по рынку как тейкер (Taker). Если цена изменилась и ордер должен исполниться сразу, биржа автоматически отменит его, защищая вас от уплаты повышенной комиссии тейкера.
При проектировании высокочастотных систем, таких как наш пинг-понг бот, критически важно понимать структуру транзакций и работу с мемпулом на L1/L2 уровнях. Рекомендуем ознакомиться со статьями о том, как устроено декодирование транзакций Uniswap V3 Mint/Burn из мемпула на Python и как реализовать мониторинг мемпула Ethereum на Python: Скрипт Whale Alerts через Web3.py.
Ниже представлен готовый класс на Python для запуска Ping-Pong стратегии на линейных деривативах (USDT-M) Bybit V5 с использованием официальной библиотеки pybit.
import time
import logging
from pybit.unified_trading import HTTP
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class BybitPingPong:
def __init__(self, api_key: str, api_secret: str, symbol: str, buy_price: float, sell_price: float, qty: float):
self.session = HTTP(
testnet=True,
api_key=api_key,
api_secret=api_secret
)
self.symbol = symbol
self.buy_price = str(buy_price)
self.sell_price = str(sell_price)
self.qty = str(qty)
self.active_order_id = None
self.state = 'PLACING_PING' # PLACING_PING, WAITING_PING, PLACING_PONG, WAITING_PONG
def place_post_only_order(self, side: str, price: str) -> str:
try:
response = self.session.place_order(
category='linear',
symbol=self.symbol,
side=side,
orderType='Limit',
qty=self.qty,
price=price,
timeInForce='PostOnly',
orderLinkId=f'pingpong_{int(time.time())}'
)
if response.get('retCode') == 0:
order_id = response['result']['orderId']
logging.info(f'Успешно размещен Post-Only ордер {side} на {price}. ID: {order_id}')
return order_id
else:
logging.error(f'Ошибка размещения ордера: {response.get("retMsg")}')
except Exception as e:
logging.error(f'Исключение при размещении ордера: {e}')
return None
def check_order_status(self, order_id: str) -> str:
try:
response = self.session.get_open_orders(
category='linear',
symbol=self.symbol,
orderId=order_id
)
if response.get('retCode') == 0 and response['result']['list']:
order = response['result']['list'][0]
return order['orderStatus']
else:
history = self.session.get_order_history(
category='linear',
symbol=self.symbol,
orderId=order_id
)
if history.get('retCode') == 0 and history['result']['list']:
order = history['result']['list'][0]
return order['orderStatus']
except Exception as e:
logging.error(f'Ошибка проверки статуса ордера {order_id}: {e}')
return 'UNKNOWN'
def run(self):
logging.info('Запуск Ping-Pong стратегии...')
while True:
if self.state == 'PLACING_PING':
logging.info('Размещение Ping (Buy) ордера...')
order_id = self.place_post_only_order('Buy', self.buy_price)
if order_id:
self.active_order_id = order_id
self.state = 'WAITING_PING'
else:
logging.info('Повтор размещения через 5 секунд...')
time.sleep(5)
elif self.state == 'WAITING_PING':
status = self.check_order_status(self.active_order_id)
logging.info(f'Статус Ping ордера {self.active_order_id}: {status}')
if status == 'Filled':
logging.info('Ping ордер исполнен! Переходим к размещению Pong (Sell)...')
self.state = 'PLACING_PONG'
elif status in ['Cancelled', 'Rejected']:
logging.warning('Ping ордер был отменен (возможно, отклонен Post-Only). Пересоздаем...')
self.state = 'PLACING_PING'
time.sleep(3)
elif self.state == 'PLACING_PONG':
logging.info('Размещение Pong (Sell) ордера...')
order_id = self.place_post_only_order('Sell', self.sell_price)
if order_id:
self.active_order_id = order_id
self.state = 'WAITING_PONG'
else:
logging.info('Повтор размещения через 5 секунд...')
time.sleep(5)
elif self.state == 'WAITING_PONG':
status = self.check_order_status(self.active_order_id)
logging.info(f'Статус Pong ордера {self.active_order_id}: {status}')
if status == 'Filled':
logging.info('Pong ордер исполнен! Цикл завершен. Начинаем заново...')
self.state = 'PLACING_PING'
elif status in ['Cancelled', 'Rejected']:
logging.warning('Pong ордер был отменен (возможно, отклонен Post-Only). Пересоздаем...')
self.state = 'PLACING_PONG'
time.sleep(3)Разбор параметров
category: Определяет тип торгового инструмента. Для бессрочных фьючерсов на Bybit V5 используется значениеlinear.timeInForce='PostOnly': Инструктирует биржу исполнять ордер исключительно как Maker. Если ордер пытается исполниться немедленно по рыночной цене, он автоматически отменяется биржей без взимания комиссий.orderLinkId: Уникальный идентификатор ордера, генерируемый на стороне клиента. Помогает избежать дублирования транзакций при сетевых сбоях.orderStatus: Текущий статус ордера. Мы отслеживаем переход в статусFilledдля переключения состояний конечного автомата бота.
Как запустить
Для запуска скрипта вам понадобятся установленная библиотека pybit и API-ключи от тестовой или реальной сети Bybit. Установите зависимость с помощью команды pip install pybit.
Замените значения api_key, api_secret, а также целевой инструмент symbol и ценовые уровни на ваши собственные. Запустите скрипт в терминале. Бот начнет циклическое выставление ордеров: сначала покупка (Ping), затем продажа (Pong).
Для обеспечения безопасности API-ключей на децентрализованных площадках полезно изучить подпись запросов Ed25519 для Jupiter DEX на Python: Гайд.
