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

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

A/B тестирование торговых моделей — это контролируемое сравнение двух стратегий на живом рынке. В отличие от backtesting, A/B тест учитывает реальные условия: slippage, latency, market impact, regime changes. Но это и сложнее: рынок меняется, поэтому нельзя просто запустить модели последовательно.

Принципы A/B тестирования торговых систем

Simultaneity: обе модели должны торговать одновременно, иначе сравнение нечестное — разные рыночные условия.

Capital allocation: разделяем торговый капитал между двумя моделями. Например, 50/50 или 70/30.

Symbol separation: для каждого символа используется одна модель (A или B), но символы распределены равномерно по характеристикам.

Statistical significance: нельзя принять решение после 10 сделок. Нужен достаточный объём данных для статистически значимых выводов.

Архитектура системы

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

Сбор метрик и статистический анализ

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 (stopping rules)

Классический 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)

Разрабатываем систему A/B тестирования с capital allocation routing, statistical significance testing (frequentist + Bayesian), sequential stopping rules, guardrail monitoring и decision dashboard.