Исходный код
В условиях высокой волатильности криптовалютного рынка классические стоп-лоссы могут не сработать из-за проскальзывания (slippage) или отсутствия ликвидности в стакане. Механизм Circuit Breaker (рыночный предохранитель) позволяет полностью остановить торговую логику бота, отменить все лимитные ордера и предотвратить открытие новых позиций во время панических распродаж.
Для раннего обнаружения аномалий полезно комбинировать ценовые триггеры с анализом потока ордеров. Например, резкий всплеск принудительных закрытий позиций часто предшествует обвалу. Подробнее о том, как это реализовать, читайте в статье Отслеживание ликвидаций на Binance Futures через WebSocket на Python.
Ниже представлен готовый к продакшену класс CircuitBreaker, использующий скользящее временное окно на базе двусторонней очереди deque для мгновенного обнаружения аномальных ценовых движений.
import asyncio
import time
from collections import deque
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
class CircuitBreaker:
def __init__(self, price_drop_threshold: float = 0.05, time_window: float = 10.0, cooldown_period: float = 60.0):
self.price_drop_threshold = price_drop_threshold
self.time_window = time_window
self.cooldown_period = cooldown_period
self.price_history = deque()
self.is_tripped = False
self.tripped_at = 0.0
def update_price(self, price: float) -> bool:
current_time = time.time()
self.price_history.append((current_time, price))
# Очищаем устаревшие данные за пределами временного окна
while self.price_history and self.price_history[0][0] < current_time - self.time_window:
self.price_history.popleft()
# Проверяем, находится ли предохранитель в режиме охлаждения
if self.is_tripped:
if current_time - self.tripped_at > self.cooldown_period:
logging.info("Период охлаждения завершен. Сброс Circuit Breaker.")
self.is_tripped = False
else:
return False
# Вычисляем максимальную просадку внутри окна
if len(self.price_history) > 1:
max_price = max(p[1] for p in self.price_history)
price_drop = (max_price - price) / max_price
if price_drop >= self.price_drop_threshold:
self._trip()
return False
return True
def _trip(self):
self.is_tripped = True
self.tripped_at = time.time()
logging.error(f"КРИТИЧЕСКАЯ СИТУАЦИЯ: Сработал Circuit Breaker! Падение цены >= {self.price_drop_threshold * 100}%. Торговля остановлена.")
self._cancel_all_orders()
def _cancel_all_orders(self):
# Симуляция экстренного вызова API биржи
logging.warning("Экстренная отмена всех активных ордеров на бирже...")
async def main():
# Инициализируем предохранитель: триггер на 3% падения за 5 секунд
cb = CircuitBreaker(price_drop_threshold=0.03, time_window=5.0, cooldown_period=10.0)
# Симуляция входящего потока цен (резкий пролив на 5-м шаге)
prices = [100.0, 99.5, 99.0, 98.5, 95.0, 92.0, 91.0, 91.5, 95.0, 98.0, 100.0]
for price in prices:
logging.info(f"Новый тик цены: {price}")
allowed = cb.update_price(price)
if not allowed:
logging.warning("Торговые операции ЗАБЛОКИРОВАНЫ.")
else:
logging.info("Торговля разрешена.")
await asyncio.sleep(1)
if __name__ == '__main__':
asyncio.run(main())Разбор параметров
price_drop_threshold: Процент падения цены (например,0.05для 5%), при превышении которого активируется защита.time_window: Временной интервал в секундах, внутри которого анализируется скорость изменения цены.cooldown_period: Время блокировки торговых операций в секундах после срабатывания триггера.price_history: Очередьdeque, хранящая временные метки и цены для вычисления скользящей просадки.
Для точной калибровки параметров price_drop_threshold и time_window необходимы качественные исторические данные высокой точности. О том, как их получить, мы писали в руководстве Как извлечь исторические данные стакана L2 с Tardis.dev в Python pandas.
Также не забывайте мониторить общие рыночные индикаторы кредитного плеча. Высокие ставки финансирования могут указывать на перегрев рынка, что увеличивает вероятность каскадных ликвидаций. Настроить их мониторинг поможет статья Асинхронный сбор Funding Rate с Binance, Bybit и OKX на Python.
Как запустить
Для запуска скрипта вам понадобится установленный интерпретатор Python версии 3.8 или выше. Скопируйте код в файл circuit_breaker.py и запустите его через терминал:
python circuit_breaker.pyВ консоли вы увидите симуляцию ценового потока. При резком падении цены с 98.5 до 95.0 сработает триггер безопасности, бот автоматически инициирует отмену всех активных ордеров и перейдет в режим ожидания (cooldown).
