После серии инцидентов в одном из проектов высокочастотной торговли мы убедились: backtesting не спасает от реальных проскальзываний и задержек. Поэтому мы разработали промышленную систему A/B тестирования торговых ML-моделей с поддержкой градиентного бустинга и нейросетей — под ключ, с интеграцией в ваш стек. Получите консультацию по вашему проекту. Подробнее о методологии — A/B тестирование.
Почему A/B тестирование критично для торговых моделей?
Статистика неутешительна: до 70% ML-моделей, показывающих отличные результаты на исторических данных, проваливаются в live-торговле из-за regime changes, market impact и неучтённых комиссий. A/B тест — единственный способ объективно сравнить две стратегии в одинаковых условиях. Без него вы рискуете принять шум за сигнал.
Проблемы, которые решает система
Параллельное сравнение обеих моделей на одном рынке — иначе сравнение нечестно. Capital allocation делит бюджет между моделями (например, 50/50 или 70/30) и назначает символы детерминированно, чтобы избежать перекоса. Guardrail метрики защищают от катастрофических просадок: при превышении лимитов версия немедленно выключается.
| Метод | Условия | Достоверность | Скорость получения результата |
|---|---|---|---|
| Backtest | Исторические данные | Низкая (overfitting, look-ahead bias) | Быстро |
| Paper trade | Симулированное исполнение | Средняя (нет impact, slippage) | Медленно |
| A/B тест | Живой рынок, разделение капитала | Высокая (статистический вывод) | Средняя |
Как мы строим систему 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)
Сравнение частотного и байесовского подходов
| Критерий | Частотный (Welch t-test) | Байесовский |
|---|---|---|
| Интерпретация | p-value (вероятность данных при H0) | P(B > A) (вероятность превосходства) |
| Ранняя остановка | SPRT | Sequential Bayesian |
| Чувствительность к размеру выборки | Требует большую выборку | Работает и с малыми выборками |
| Устойчивость к выбросам | Mann-Whitney U | Использует robust likelihood |
Для достижения мощности 80% при α=0.05 и эффекте Cohen's d=0.5 требуется примерно n=64 на группу. При ежедневном количестве сделок 20 это соответствует 3.2 дням теста.
Что входит в работу?
- Архитектура и проектирование — схема роутинга, модель данных, выбор инструментов.
- Реализация — написание кода роутера, анализатора, дашборда.
- Интеграция с вашей торговой платформой — подключение к брокерским API, базам данных.
- Тестирование — симуляции на исторических данных и paper trading.
- Документация — описание экспериментального дизайна, API, инструкция оператора.
- Поддержка — сопровождение первых 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 тестирования.