Разработка системы перехода из бэктеста в live-торговлю

Разработка системы перехода из бэктеста в live-торговлю

Переход из бэктеста в live trading — критический момент, где большинство алготрейдеров теряют деньги. Причина не в плохих стратегиях, а в недооценке разрыва между идеальной симуляцией и реальным миром. Систематический подход к этому переходу снижает риски.

Checklist перед запуском

Технические проверки:

• Стратегия прошла walk-forward validation
• OOS результаты приемлемы (Sharpe > 0.5)
• Paper trading 2–4 недели без критических инцидентов
• Unit и integration тесты покрывают > 80% кода
• Все сценарии ошибок обработаны (API timeout, partial fill, rejected order)
• Monitoring и алерты настроены
• Emergency stop механизм реализован и протестирован

Риск-менеджмент:

• Определены максимальные daily loss limit и total drawdown limit
• Установлен начальный размер позиции (1–5% от капитала)
• Настроены автоматические kill-switch триггеры

Staged Deployment

class LiveTradingDeployment:
    """Поэтапный запуск стратегии в production"""

    STAGES = [
        {'name': 'Paper Trading', 'capital_pct': 0, 'duration_days': 14, 'max_drawdown': None},
        {'name': 'Micro Live', 'capital_pct': 0.05, 'duration_days': 30, 'max_drawdown': -0.05},
        {'name': 'Small Live', 'capital_pct': 0.20, 'duration_days': 60, 'max_drawdown': -0.10},
        {'name': 'Medium Live', 'capital_pct': 0.50, 'duration_days': 90, 'max_drawdown': -0.15},
        {'name': 'Full Scale', 'capital_pct': 1.00, 'duration_days': None, 'max_drawdown': -0.20},
    ]

    async def advance_stage(self, current_stage: int, performance: dict) -> bool:
        """Решаем можно ли перейти к следующему этапу"""
        stage = self.STAGES[current_stage]
        next_stage = self.STAGES[current_stage + 1] if current_stage + 1 < len(self.STAGES) else None

        if not next_stage:
            return False

        # Критерии для продвижения
        checks = {
            'min_duration': performance['days_running'] >= stage['duration_days'],
            'positive_sharpe': performance['sharpe_ratio'] > 0,
            'drawdown_ok': performance['max_drawdown'] > (stage['max_drawdown'] or -1.0),
            'no_critical_errors': performance['critical_errors'] == 0,
        }

        if all(checks.values()):
            return True

        # Логируем что не прошло
        failed = [k for k, v in checks.items() if not v]
        print(f"Stage {stage['name']} advancement blocked: {failed}")
        return False

Kill Switch Implementation

class KillSwitch:
    """Аварийная остановка торговли"""

    def __init__(
        self,
        daily_loss_limit_pct: float = 0.03,  # 3% дневного капитала
        total_drawdown_limit_pct: float = 0.10,  # 10% от исходного капитала
    ):
        self.daily_loss_limit = daily_loss_limit_pct
        self.drawdown_limit = total_drawdown_limit_pct
        self.triggered = False
        self.trigger_reason = None

    async def check(self, portfolio: Portfolio):
        if self.triggered:
            return

        # Дневные потери
        daily_loss = portfolio.get_daily_pnl_pct()
        if daily_loss < -self.daily_loss_limit:
            await self.trigger(f"Daily loss limit: {daily_loss:.2%}")
            return

        # Общая просадка
        total_drawdown = portfolio.get_drawdown_from_peak()
        if total_drawdown < -self.drawdown_limit:
            await self.trigger(f"Total drawdown limit: {total_drawdown:.2%}")
            return

    async def trigger(self, reason: str):
        self.triggered = True
        self.trigger_reason = reason

        # 1. Останавливаем генерацию новых сигналов
        await self.signal_engine.stop()

        # 2. Отменяем все pending ордера
        await self.broker.cancel_all_orders()

        # 3. Опционально: закрываем все позиции
        # await self.broker.close_all_positions()  # зависит от стратегии

        # 4. Алерт команде
        await self.alerter.send_critical(
            f"KILL SWITCH TRIGGERED: {reason}\n"
            f"All orders cancelled. Manual intervention required."
        )

Monitoring переходного периода

class TransitionMonitor:
    def compare_with_expectations(
        self,
        live_result: LiveResult,
        backtest_result: BacktestResult,
        period_days: int,
    ) -> dict:
        """Сравниваем live результаты с ожиданиями из бэктеста"""
        live_daily_returns = live_result.equity.pct_change().dropna()
        backtest_avg_daily = backtest_result.metrics.annual_return_pct / 252 / 100

        return {
            'live_avg_daily_return': live_daily_returns.mean(),
            'expected_avg_daily_return': backtest_avg_daily,
            'performance_ratio': live_daily_returns.mean() / backtest_avg_daily if backtest_avg_daily > 0 else 0,
            'execution_quality': self.assess_execution_quality(live_result),
            'slippage_vs_backtest': live_result.avg_slippage - backtest_result.assumed_slippage,
            'recommendation': 'CONTINUE' if live_daily_returns.mean() > backtest_avg_daily * 0.3 else 'REVIEW',
        }

Ключевое правило перехода: если live performance систематически (2+ недели) составляет менее 30% от ожидаемого из бэктеста — нужен анализ причин до масштабирования капитала. Это может быть market regime change, implementation bug или фундаментальный overfit.