בדיקות A/B לאלגוריתמי מסחר — אינטגרציה סוהר

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

איסוף מדדים וניתוח סטטיסטי הוא שלב מפתח. אנו משתמשים בגישות פרוונטיסטיות (frequentist) ובגישות בייסיאניות. מובהקות סטטיסטית נקבעת על ידי 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 קלאסית דורשת גודל מדגם קבוע מראש. בדיקות רציפות מאפשרות להחליט מוקדם יותר. הן מהירות עד פי 2 מבדיקות בגודל מדגם קבוע.

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 metrics)

בדיקת A/B לא אמורה לגרום נזק. מדדי הגנה הם דרישות מינימום לשתי הגרסאות:

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

אם מדדי ההגנה מופרים, הגרסה המתאימה מושבתת מיד.

לוח מחוונים וקבלת החלטות

לוח המחוונים בזמן אמת מציג:

• רווח והפסד מצטבר (P&L) של כל גרסה (עקומות הון)
• P-value ורווח בר-סמך
• הסתברות בייסיאנית B > A
• טבלת מדדים: Sharpe, Win Rate, Max DD, סה"כ עסקאות

מסגרת קבלת החלטות:

• P-value < 0.05 ומספר עסקאות N > min_trades → ניתן לקבל החלטה
• הסתברות בייסיאנית P(B > A) > 95% → ביטחון גבוה ש-B מנצח
• גודל אפקט Cohen's d < 0.1 → אין הבדל מעשי; בחירה לפי קריטריונים אחרים (מורכבות, זמן השהיה)

השוואה בין גישות פרוונטיסטיות ובייסיאניות

כדי להשיג עוצמה של 80% ב-α=0.05 וגודל אפקט Cohen's d=0.5, נדרשים כ-n=64 לכל קבוצה. עם 20 עסקאות ביום, זה מתאים ל-3.2 ימי בדיקה.

מה כלול בעבודה?

1. ארכיטקטורה ועיצוב — תכנית ניתוב, מודל נתונים, בחירת כלים.
2. מימוש — כתיבת קוד לנתב, מנתח, ולוח מחוונים.
3. אינטגרציה עם פלטפורמת המסחר שלך — חיבור ל-API של ברוקרים, מסדי נתונים.
4. בדיקות — סימולציות על נתונים היסטוריים ומסחר מדומה.
5. תיעוד — תיאור עיצוב הניסוי, API, הוראות למפעילים.
6. תמיכה — ליווי ניסויים חיים ראשונים, ייעוץ.

מהניסיון: הפחתת החלקת מחיר ב-40%

בפרויקט אחד, יישמנו A/B testing כדי להשוות מודל חדש לביצוע הזמנות עם המודל הנוכחי. במשך שבועיים, צברנו 500 עסקאות על כל גרסה. תוצאה: המודל החדש הפחית את החלקת המחיר ב-40% (p-value < 0.01, Cohen's d = 0.6). החיסכון מהחלקת מחיר הסתכם ב-$12,000 לחודש, והחיסכון השנתי עלה על $100,000. בדיקת ה-split test החזירה את עצמה תוך פחות מ-3 חודשים. ללא הניסוי, לא היינו יכולים להפריד בין השפעת המודל לבין תנודות השוק. כפי שהנתונים מראים, A/B testing יעיל פי 3 מבדיקות backtest פשוטות לזיהוי ביצועים אמיתיים.

מדוע להפקיד בידינו את הפיתוח?

הניסיון שלנו: למעלה מ-10 שנים בסביבות ייצור, כולל מערכות בעומס גבוה, מסחר ופיתוח בלוקצ'יין, עם יותר מ-30 פרויקטים מוצלחים. המהנדסים שלנו בקיאים ב-Solidity, Rust, Python, ויש להם ניסיון עם מערכות בתדירות גבוהה. אנו מבטיחים איכות קוד באמצעות ביקורת ו-audit. צרו קשר כדי להעריך את הפרויקט שלכם—נדון בפרטים ובלוחות זמנים. העריכו את האפקטיביות של האסטרטגיות שלכם—הזמינו יישום של מערכת A/B testing. פיתוח מותאם אישית מתחיל מ-$50,000 עם ROI טיפוסי העולה על 400% בשנה הראשונה. עם ניסיון של 10+ שנים ו-30+ פרויקטים שהושלמו, הצוות שלנו מספק פתרונות חזקים.