Разработка AI-системы контроля drawdown и автоматической остановки торговли
Drawdown control — критически важная система управления рисками для AI-трейдинг-ботов. Даже прибыльная долгосрочная стратегия может пройти через экстремальные просадки, которые психологически и финансово неприемлемы. Автоматическая остановка торговли при достижении пороговых значений drawdown — обязательный элемент риск-менеджмента.
Типы drawdown
Maximum Drawdown (MDD) — максимальное снижение от пика до дна за весь период: MDD = max((peak - trough) / peak).
Current Drawdown — текущая просадка от последнего пика: отслеживается в реальном времени.
Daily Drawdown — просадка за текущий торговый день (важно для внутридневных стратегий).
Consecutive Losses — серия убыточных сделок подряд (психологически важный показатель).
Архитектура системы контроля
from dataclasses import dataclass, field
from enum import Enum
import threading
class TradingStatus(Enum):
ACTIVE = "active"
PAUSED = "paused"
STOPPED = "stopped"
@dataclass
class DrawdownControlConfig:
max_daily_drawdown_pct: float = 0.03 # -3% в день
max_total_drawdown_pct: float = 0.10 # -10% от начала
max_consecutive_losses: int = 5 # 5 убытков подряд
pause_on_loss_streak: int = 3 # Пауза после 3 убытков
recovery_time_minutes: int = 30 # Пауза перед возобновлением
class DrawdownController:
def __init__(self, config: DrawdownControlConfig, initial_equity: float):
self.config = config
self.initial_equity = initial_equity
self.peak_equity = initial_equity
self.day_start_equity = initial_equity
self.current_equity = initial_equity
self.consecutive_losses = 0
self.status = TradingStatus.ACTIVE
self._lock = threading.Lock()
self._pause_until = None
def update_equity(self, new_equity: float) -> TradingStatus:
with self._lock:
self.current_equity = new_equity
self.peak_equity = max(self.peak_equity, new_equity)
current_drawdown = (self.peak_equity - new_equity) / self.peak_equity
daily_drawdown = (self.day_start_equity - new_equity) / self.day_start_equity
# Проверка лимитов
if current_drawdown >= self.config.max_total_drawdown_pct:
self._stop_trading(f"Max total drawdown {current_drawdown:.2%} exceeded")
elif daily_drawdown >= self.config.max_daily_drawdown_pct:
self._stop_trading_for_day(f"Max daily drawdown {daily_drawdown:.2%} exceeded")
return self.status
def on_trade_result(self, pnl: float) -> TradingStatus:
with self._lock:
if pnl < 0:
self.consecutive_losses += 1
if self.consecutive_losses >= self.config.max_consecutive_losses:
self._stop_trading(f"{self.consecutive_losses} consecutive losses")
elif self.consecutive_losses >= self.config.pause_on_loss_streak:
self._pause_trading(self.config.recovery_time_minutes)
else:
self.consecutive_losses = 0 # Сброс при прибыльной сделке
return self.status
def _stop_trading(self, reason: str):
self.status = TradingStatus.STOPPED
self._notify_team(f"TRADING STOPPED: {reason}", urgent=True)
self._close_all_positions()
def _pause_trading(self, minutes: int):
self.status = TradingStatus.PAUSED
self._pause_until = datetime.utcnow() + timedelta(minutes=minutes)
self._notify_team(f"Trading paused for {minutes}min: consecutive losses")
Динамические лимиты
Статические пороговые значения не всегда оптимальны. Динамический подход:
class DynamicDrawdownLimits:
def __init__(self, volatility_window=20):
self.window = volatility_window
def compute_dynamic_limit(self, returns_history: list) -> float:
"""Лимит drawdown как функция от волатильности рынка"""
if len(returns_history) < self.window:
return 0.05 # Базовый лимит 5%
recent_vol = np.std(returns_history[-self.window:]) * np.sqrt(252)
# При высокой волатильности — более строгий лимит
if recent_vol > 0.3: # VIX-эквивалент > 30%
return 0.03 # 3%
elif recent_vol > 0.2:
return 0.05 # 5%
else:
return 0.08 # 8% при низкой волатильности
Интеграция с риск-менеджментом
Система контроля drawdown должна быть синхронной с исполнительной системой: при TradingStatus.STOPPED не должны проходить никакие новые ордера. Рекомендуется добавить hardware-уровень защиты (broker-side stop) независимо от программного контроля — некоторые брокеры поддерживают Risk Limits API.
Ключевое правило: возобновление торговли после принудительной остановки требует явного ручного подтверждения от риск-менеджера, а не автоматического.