Після серії інцидентів в одному з проєктів високочастотної торгівлі ми переконалися: 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 тестування.