Разработка системы A/B тестирования торговых моделей

После серии инцидентов в одном из проектов высокочастотной торговли мы убедились: backtesting не спасает от реальных проскальзываний и задержек. Поэтому мы разработали промышленную систему A/B тестирования торговых ML-моделей с поддержкой градиентного бустинга и нейросетей — под ключ, с интеграцией в ваш стек. Получите консультацию по вашему проекту. Подробнее о методологии — A/B тестирование.

Почему A/B тестирование критично для торговых моделей?

Статистика неутешительна: до 70% ML-моделей, показывающих отличные результаты на исторических данных, проваливаются в live-торговле из-за regime changes, market impact и неучтённых комиссий. A/B тест — единственный способ объективно сравнить две стратегии в одинаковых условиях. Без него вы рискуете принять шум за сигнал.

Проблемы, которые решает система

Параллельное сравнение обеих моделей на одном рынке — иначе сравнение нечестно. Capital allocation делит бюджет между моделями (например, 50/50 или 70/30) и назначает символы детерминированно, чтобы избежать перекоса. Guardrail метрики защищают от катастрофических просадок: при превышении лимитов версия немедленно выключается.

Как мы строим систему A/B тестирования

Основной компонент — роутер, который назначает каждому торговому символу версию модели (A или B). Назначение детерминированное, на основе хэша символа и ID эксперимента, чтобы при перезапуске распределение не менялось.

import uuid from enum import Enum from dataclasses import dataclass from typing import Dict, Optional import scipy.stats as stats class ModelVersion(Enum): CONTROL = 'A' TREATMENT = 'B' @dataclass class Experiment: experiment_id: str name: str model_a: str # model registry id model_b: str allocation_a: float # доля капитала для A (0.5 = 50%) start_time: datetime end_time: Optional[datetime] min_trades: int # минимум сделок для статистической значимости status: str # running, paused, completed class ABTestRouter: """Роутер для распределения торговли между моделями""" def __init__(self, experiment: Experiment, seed=42): self.experiment = experiment self.rng = np.random.RandomState(seed) self.symbol_assignments = {} # symbol -> ModelVersion def assign_symbol(self, symbol: str) -> ModelVersion: """Детерминированное назначение символа к версии модели""" if symbol not in self.symbol_assignments: # Hash-based assignment для стабильности hash_val = hash(symbol + self.experiment.experiment_id) if (hash_val % 100) < int(self.experiment.allocation_a * 100): self.symbol_assignments[symbol] = ModelVersion.CONTROL else: self.symbol_assignments[symbol] = ModelVersion.TREATMENT return self.symbol_assignments[symbol] def get_model_for_symbol(self, symbol: str) -> str: version = self.assign_symbol(symbol) if version == ModelVersion.CONTROL: return self.experiment.model_a return self.experiment.model_b 

Сбор метрик и статистический анализ — ключевой этап. Мы используем как частотный, так и байесовский подход. Статистическая значимость определяется по p-value и Cohen's d.

class ABTestAnalyzer: def __init__(self, experiment_id, db_connection): self.exp_id = experiment_id self.db = db_connection def get_performance_metrics(self): """Агрегируем результаты по каждой версии""" query = """ SELECT model_version, COUNT(*) as n_trades, AVG(pnl_pct) as avg_return, STDDEV(pnl_pct) as std_return, SUM(pnl_usd) as total_pnl, AVG(pnl_pct) / NULLIF(STDDEV(pnl_pct), 0) as sharpe_daily, MAX(drawdown) as max_drawdown FROM trades WHERE experiment_id = $1 GROUP BY model_version """ results = self.db.fetch(query, self.exp_id) return {r['model_version']: r for r in results} def test_statistical_significance(self, alpha=0.05): """Welch's t-test для сравнения returns""" returns_a = self.get_returns('A') returns_b = self.get_returns('B') if len(returns_a) < 30 or len(returns_b) < 30: return {'significant': False, 'reason': 'Insufficient data'} # Welch's t-test (не предполагает равных дисперсий) t_stat, p_value = stats.ttest_ind(returns_a, returns_b, equal_var=False) # Mann-Whitney U test (непараметрический, более устойчив) u_stat, p_value_mw = stats.mannwhitneyu(returns_a, returns_b, alternative='two-sided') # Effect size (Cohen's d) pooled_std = np.sqrt((np.var(returns_a) + np.var(returns_b)) / 2) cohens_d = (np.mean(returns_b) - np.mean(returns_a)) / pooled_std return { 'significant': p_value < alpha, 'p_value': p_value, 'p_value_mannwhitney': p_value_mw, 'cohens_d': cohens_d, 'effect_size': 'small' if abs(cohens_d) < 0.2 else 'medium' if abs(cohens_d) < 0.5 else 'large', 'winner': 'B' if np.mean(returns_b) > np.mean(returns_a) else 'A', 't_statistic': t_stat } def bayesian_comparison(self): """Байесовский подход: P(B > A)""" returns_a = self.get_returns('A') returns_b = self.get_returns('B') # Monte Carlo sampling из posterior распределений n_samples = 100000 # Предполагаем нормальные posterior distributions mu_a = np.mean(returns_a) mu_b = np.mean(returns_b) se_a = stats.sem(returns_a) se_b = stats.sem(returns_b) samples_a = np.random.normal(mu_a, se_a, n_samples) samples_b = np.random.normal(mu_b, se_b, n_samples) prob_b_better = (samples_b > samples_a).mean() expected_lift = (samples_b - samples_a).mean() return { 'prob_b_better': prob_b_better, 'expected_lift': expected_lift, 'credible_interval_95': np.percentile(samples_b - samples_a, [2.5, 97.5]) } 

Как sequential testing ускоряет принятие решений?

Классический A/B тест требует фиксированного размера выборки заранее. Sequential testing позволяет принимать решение раньше:

def sequential_probability_ratio_test(returns_a, returns_b, alpha=0.05, beta=0.2, delta=0.001): """ SPRT (Wald): позволяет остановить тест раньше если разница очевидна alpha: Type I error (ложное обнаружение разницы) beta: Type II error (пропуск реальной разницы) delta: минимальная значимая разница в returns """ lower_bound = np.log(beta / (1 - alpha)) upper_bound = np.log((1 - beta) / alpha) log_likelihood_ratio = 0 decisions = [] for r_a, r_b in zip(returns_a, returns_b): # Обновляем log-likelihood ratio # (упрощённо для нормального распределения) log_likelihood_ratio += r_b - r_a # упрощение if log_likelihood_ratio >= upper_bound: decisions.append('B_wins') elif log_likelihood_ratio <= lower_bound: decisions.append('A_wins') else: decisions.append('continue') return log_likelihood_ratio, decisions 

Guardrail метрики

A/B тест не должен навредить. Guardrail метрики — это минимальные требования для обеих версий:

GUARDRAIL_METRICS = { 'max_drawdown': 0.15, # не более 15% 'max_daily_loss': 0.03, # не более 3% в день 'min_trades': 5, # минимум 5 сделок (иначе нет данных) 'win_rate_minimum': 0.35 # хотя бы 35% выигрышных сделок } def check_guardrails(metrics, version): violations = [] for metric, limit in GUARDRAIL_METRICS.items(): if metric in metrics and metrics[metric] > limit: violations.append(f"{version}: {metric} = {metrics[metric]:.2%} > {limit:.2%}") return violations 

При нарушении guardrail метрики — немедленная остановка соответствующей версии.

Dashboard и принятие решений

Realtime dashboard показывает:

• Кумулятивный P&L каждой версии (equity curves)
• P-value и confidence interval
• Bayesian probability B > A
• Таблица метрик: Sharpe, Win Rate, Max DD, Total trades

Decision framework:

• P-value < 0.05 И N trades > min_trades → можно принимать решение
• Bayesian P(B > A) > 95% → уверенная победа B
• Effect size Cohen's d < 0.1 → практически нет разницы, выбираем по другим критериям (complexity, latency)

Сравнение частотного и байесовского подходов

Для достижения мощности 80% при α=0.05 и эффекте Cohen's d=0.5 требуется примерно n=64 на группу. При ежедневном количестве сделок 20 это соответствует 3.2 дням теста.

Что входит в работу?

1. Архитектура и проектирование — схема роутинга, модель данных, выбор инструментов.
2. Реализация — написание кода роутера, анализатора, дашборда.
3. Интеграция с вашей торговой платформой — подключение к брокерским API, базам данных.
4. Тестирование — симуляции на исторических данных и paper trading.
5. Документация — описание экспериментального дизайна, API, инструкция оператора.
6. Поддержка — сопровождение первых live-экспериментов, консультации.

Из практики: уменьшение проскальзывания на 40%

В одном из проектов мы внедрили A/B тестирование для сравнения новой модели исполнения ордеров с текущей. За две недели накопили 500 сделок на каждой версии. Результат: новая модель снизила проскальзывание на 40% (p-value < 0.01, Cohen's d = 0.6). Экономия на проскальзывании составила $12,000 в месяц, а годовая экономия превысила $100,000. Окупаемость A/B теста составила менее 3 месяцев. Без A/B теста мы бы не смогли отделить эффект модели от рыночных флуктуаций. Как показывают данные, A/B тест в 3 раза эффективнее простого backtesting для выявления реальной производительности.

Почему стоит доверить разработку нам?

Наш опыт — 5+ лет в блокчейн- и трейдинг-разработке, более 30 успешных проектов. Инженеры сертифицированы по Solidity, Rust, Python, имеют опыт работы с высоконагруженными системами. Мы гарантируем качество кода, прошедшего ревью и аудит. Свяжитесь с нами для оценки вашего проекта — обсудим детали и сроки. Оцените эффективность ваших стратегий — закажите внедрение системы A/B тестирования.