18.3 A/B测试与金丝雀发布

在互联网行业，A/B测试是产品优化的标准流程；在量化交易领域，A/B测试同样适用于策略迭代、参数调优和新策略引入的场景。与互联网产品不同的是，量化策略的 A/B 测试涉及真实的资金和风险，因此需要更谨慎的设计和更严格的统计标准。

策略A/B测试的设计原则

什么是策略A/B测试

策略 A/B 测试是指在相同时间段内，将资金（或交易机会）随机分配给两个不同的策略版本，通过对实际表现的统计比较来判断哪个版本更优。A 组是对照组（现有策略或基准），B 组是实验组（新策略或改进版本）。

A/B测试的适用场景

策略改进验证：对因子进行微调、加入新的过滤条件、更换执行算法等。典型问题：新的止损规则是否提升了夏普比率？
参数优化确认：回测中最优参数是否在样本外（实盘中）仍然最优？A/B 测试可以提供真实的样本外证据。
新策略引入：将一个全新策略以少量资金引入实盘，评估其与现有策略的相关性和互补性。
执行算法比较：比较 TWAP、VWAP、IS（Implementation Shortfall）等不同执行算法的实际效果。

核心设计原则

随机化（Randomization）：资金或交易信号的分配必须是随机的，以确保两个组在统计上可比。如果分配过程存在系统性偏差（例如 B 组总是拿到流动性更好的订单），测试结果将无效。

同时性（Contemporaneous）：A/B 两组必须在完全相同的时间段内运行。前后对比（先运行 A、再运行 B）会受到市场环境变化的干扰，不是严格意义上的 A/B 测试。

独立性（Independence）：A/B 两组的交易不应该互相干扰。如果两个组交易相同的股票，B 组的大单可能影响 A 组的成交价格，造成交叉污染。

足够的样本量：金融时间序列的信噪比极低，需要大量样本才能获得统计显著性。对于日频策略，建议至少收集 3-6 个月的数据；对于高频策略，可以使用 bootstrap 方法评估显著性。

统计显著性判断

A/B 测试最核心的问题是：观测到的差异是真实的 Alpha 改善还是随机噪声？这需要统计检验来回答。

两样本 t 检验

对于日度收益序列，最常用的是两样本 t 检验：

$$t = \frac{\bar{r}_B - \bar{r}_A}{\sqrt{\frac{s_A^2}{n_A} + \frac{s_B^2}{n_B}}}$$

其中 ${\bar{r}}_{A}$ 和 ${\bar{r}}_{B}$ 分别是 A/B 组的日均收益， $s_{A}^{2}$ 和 $s_{B}^{2}$ 是样本方差， $n_{A}$ 和 $n_{B}$ 是样本量。

但 t 检验假设收益服从正态分布且独立同分布，这在金融数据中往往不成立（存在尖峰厚尾、波动率聚集、自相关等特征）。

替代检验方法

Bootstrap 检验：通过对收益序列有放回抽样，构造收益差异的经验分布，计算 p 值。Bootstrap 不依赖正态性假设。

Newey-West 修正：当收益存在自相关和异方差时，使用 Newey-West 标准误来调整 t 统计量。

排列检验（Permutation Test）：将 A/B 标签随机打乱多次，计算每次的"伪差异"，看真实差异在打乱分布中的位置。

PYTHON202 行 · 7.2 KB

📄此处有展示代码202 行 · 7.2 KB展开 ▼

python

import numpy as np
import pandas as pd
from scipy import stats
from typing import Tuple


class ABTestAnalyzer:
    """量化策略 A/B 测试分析器"""
    
    def __init__(self, returns_a: pd.Series, returns_b: pd.Series, 
                 benchmark_returns: pd.Series = None):
        """
        Parameters
        ----------
        returns_a : pd.Series
            对照组（A）的日度收益序列
        returns_b : pd.Series
            实验组（B）的日度收益序列
        benchmark_returns : pd.Series, optional
            基准收益序列，用于计算 Alpha
        """
        self.returns_a = returns_a
        self.returns_b = returns_b
        self.benchmark = benchmark_returns
        
    def basic_stats(self) -> pd.DataFrame:
        """计算两组的基本统计量"""
        stats_dict = {}
        for name, rets in [('A', self.returns_a), ('B', self.returns_b)]:
            rets = rets.dropna()
            stats_dict[name] = {
                'mean_daily': rets.mean(),
                'std_daily': rets.std(),
                'skewness': rets.skew(),
                'kurtosis': rets.kurtosis(),
                'annual_return': rets.mean() * 252,
                'annual_vol': rets.std() * np.sqrt(252),
                'sharpe': (rets.mean() / rets.std()) * np.sqrt(252) if rets.std() > 0 else 0,
                'max_drawdown': (rets.cumsum() - rets.cumsum().cummax()).min(),
                'win_rate': (rets > 0).mean(),
                'n_days': len(rets)
            }
        return pd.DataFrame(stats_dict)
    
    def two_sample_ttest(self) -> dict:
        """两样本 t 检验（Welch's t-test，不假设方差相等）"""
        rets_a = self.returns_a.dropna()
        rets_b = self.returns_b.dropna()
        
        t_stat, p_value = stats.ttest_ind(rets_b, rets_a, equal_var=False)
        
        return {
            'method': "Welch's t-test",
            't_statistic': t_stat,
            'p_value': p_value,
            'significant_5pct': p_value < 0.05,
            'significant_1pct': p_value < 0.01
        }
    
    def bootstrap_test(self, n_bootstrap: int = 10000, 
                       statistic: str = 'sharpe') -> dict:
        """
        Bootstrap 检验：对夏普比率差异进行 Bootstrap
        
        Parameters
        ----------
        n_bootstrap : int
            Bootstrap 抽样次数
        statistic : str
            比较的统计量，'sharpe' 或 'mean'
        """
        rets_a = self.returns_a.dropna().values
        rets_b = self.returns_b.dropna().values
        
        def compute_stat(r):
            ann_r = r.mean() * 252
            if statistic == 'sharpe':
                ann_vol = r.std() * np.sqrt(252)
                return ann_r / ann_vol if ann_vol > 0 else 0
            return ann_r
        
        obs_diff = compute_stat(rets_b) - compute_stat(rets_a)
        
        # Bootstrap
        np.random.seed(42)
        boot_diffs = []
        for _ in range(n_bootstrap):
            a_sample = np.random.choice(rets_a, size=len(rets_a), replace=True)
            b_sample = np.random.choice(rets_b, size=len(rets_b), replace=True)
            boot_diffs.append(compute_stat(b_sample) - compute_stat(a_sample))
        
        boot_diffs = np.array(boot_diffs)
        
        # 双尾 p 值
        p_value = np.mean(np.abs(boot_diffs) >= np.abs(obs_diff))
        
        return {
            'method': f'Bootstrap ({statistic})',
            'observed_difference': obs_diff,
            'bootstrap_mean': boot_diffs.mean(),
            'bootstrap_std': boot_diffs.std(),
            'ci_95_lower': np.percentile(boot_diffs, 2.5),
            'ci_95_upper': np.percentile(boot_diffs, 97.5),
            'p_value': p_value,
            'significant_5pct': p_value < 0.05
        }
    
    def permutation_test(self, n_permutations: int = 10000) -> dict:
        """排列检验：无假设的 A/B 差异检验"""
        rets_a = self.returns_a.dropna().values
        rets_b = self.returns_b.dropna().values
        
        combined = np.concatenate([rets_a, rets_b])
        n_a = len(rets_a)
        
        obs_diff = rets_b.mean() - rets_a.mean()
        
        np.random.seed(42)
        perm_diffs = []
        for _ in range(n_permutations):
            np.random.shuffle(combined)
            perm_diffs.append(combined[:n_a].mean() - combined[n_a:].mean())
        
        perm_diffs = np.array(perm_diffs)
        p_value = np.mean(np.abs(perm_diffs) >= np.abs(obs_diff))
        
        return {
            'method': 'Permutation Test',
            'observed_difference': obs_diff,
            'p_value': p_value,
            'significant_5pct': p_value < 0.05
        }
    
    def calculate_required_sample_size(self, effect_size: float = 0.2, 
                                        power: float = 0.8) -> int:
        """
        计算所需样本量
        
        Parameters
        ----------
        effect_size : float
            期望检测到的最小效应量（以年化夏普差异表示）
        power : float
            统计功效（1 - Type II Error）
        """
        daily_effect = effect_size / np.sqrt(252)
        daily_std = self.returns_a.std()
        
        # Cohen's d
        d = daily_effect / daily_std
        
        # 近似公式计算所需样本量（每组）
        from scipy.stats import norm
        z_alpha = norm.ppf(0.975)  # 双尾 5% 显著性
        z_beta = norm.ppf(power)
        
        n = 2 * ((z_alpha + z_beta) / d) ** 2
        return int(np.ceil(n))
    
    def generate_report(self) -> str:
        """生成 A/B 测试综合报告"""
        basic = self.basic_stats()
        ttest = self.two_sample_ttest()
        boot = self.bootstrap_test()
        perm = self.permutation_test()
        
        report_lines = [
            "=" * 60,
            "          量化策略 A/B 测试报告",
            "=" * 60,
            "",
            "【基本统计】",
            basic.to_string(),
            "",
            "【统计检验结果】",
            "",
            f"1. {ttest['method']}:",
            f"   t = {ttest['t_statistic']:.4f}, p = {ttest['p_value']:.4f}",
            f"   5%显著性: {'是' if ttest['significant_5pct'] else '否'}",
            "",
            f"2. {boot['method']}:",
            f"   观测差异 = {boot['observed_difference']:.4f}",
            f"   95% CI = [{boot['ci_95_lower']:.4f}, {boot['ci_95_upper']:.4f}]",
            f"   p = {boot['p_value']:.4f}",
            f"   5%显著性: {'是' if boot['significant_5pct'] else '否'}",
            "",
            f"3. {perm['method']}:",
            f"   观测差异 = {perm['observed_difference']:.4f}",
            f"   p = {perm['p_value']:.4f}",
            "",
            "【结论】",
        ]
        
        # 综合判断
        if ttest['p_value'] < 0.05 and boot['p_value'] < 0.05:
            report_lines.append("B组显著优于A组，建议推进到下一阶段。")
        elif ttest['p_value'] < 0.1:
            report_lines.append("B组有一定优势但不够显著，建议延长测试时间。")
        else:
            report_lines.append("A/B两组无显著差异，B组的改进可能无效。")
        
        return '\n'.join(report_lines)

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：类 `ABTestAnalyzer`（量化策略 A/B 测试分析器）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

样本量计算

这是 A/B 测试中最容易被忽视的问题。金融收益的波动率远大于均值（日度夏普比率 0.1-0.3 很常见），这意味着需要极大的样本量才能检测到真实的改善。

假设日度收益的标准差为 1.5%，希望检测到年化夏普比率 0.3 的改善：

$$n \approx 2 \times \left(\frac{z_{\alpha/2} + z_\beta}{d}\right)^2 = 2 \times \left(\frac{1.96 + 0.84}{0.3/\sqrt{252} / 0.015}\right)^2 \approx 246$$

即每组需要约 246 个交易日（约一年）的数据，才能在 80% 的统计功效下检测到 0.3 的夏普改善。如果希望检测到更小的改善（如夏普提升 0.1），所需样本量会是上面的 9 倍。

金丝雀发布流程

金丝雀发布（Canary Release）的名称来源于煤矿工人用金丝雀检测有毒气体——先让少量"金丝雀"暴露在新环境中，如果出现问题，损失可控且可以快速回滚。

渐进式放量

阶段0:     0%（准备阶段）
阶段1:     1%（1-3天）    首次放量，密切监控
阶段2:     5%（3-7天）    确认无异常
阶段3:    20%（1-2周）    收集足够统计数据
阶段4:    50%（1-2周）    大规模验证
阶段5:   100%（全量）     正式上线

每个阶段之间的放量决策基于自动化的监控指标：

PYTHON81 行 · 3.0 KB

📄此处有展示代码81 行 · 3.0 KB展开 ▼

python

class CanaryReleaseManager:
    """金丝雀发布管理器"""
    
    def __init__(self, stages: list = None):
        if stages is None:
            self.stages = [
                {'pct': 0.01, 'min_days': 2, 'description': '1% canary'},
                {'pct': 0.05, 'min_days': 5, 'description': '5% expansion'},
                {'pct': 0.20, 'min_days': 7, 'description': '20% rollout'},
                {'pct': 0.50, 'min_days': 10, 'description': '50% rollout'},
                {'pct': 1.00, 'min_days': 14, 'description': '100% full'},
            ]
        else:
            self.stages = stages
        
        self.current_stage = 0
        self.days_in_stage = 0
        self.stage_metrics = []
        self.rollback_history = []
        
    def evaluate_promotion(self, daily_metrics: dict) -> dict:
        """
        评估是否可以推进到下一阶段
        
        Parameters
        ----------
        daily_metrics : dict
            包含 sharpe, max_drawdown, tracking_error, pnl 等指标
        """
        if self.current_stage >= len(self.stages) - 1:
            return {'action': 'COMPLETE', 'message': '已处于最终阶段'}
        
        current = self.stages[self.current_stage]
        self.days_in_stage += 1
        
        # 检查是否满足最低天数
        if self.days_in_stage < current['min_days']:
            return {
                'action': 'HOLD',
                'message': f"当前阶段需至少{current['min_days']}天，已运行{self.days_in_stage}天"
            }
        
        # 检查回撤是否在允许范围内
        if daily_metrics.get('drawdown_pct', 0) > 0.03:
            return {
                'action': 'ROLLBACK',
                'message': f"日内回撤{daily_metrics['drawdown_pct']:.2%}超过3%阈值，触发回滚"
            }
        
        # 检查累计 PnL
        if daily_metrics.get('cumulative_pnl', 0) < -0.02 * daily_metrics.get('allocated_capital', 1):
            return {
                'action': 'ROLLBACK',
                'message': '累计亏损超过分配资金的2%，触发回滚'
            }
        
        # 通过检查，推进到下一阶段
        self.current_stage += 1
        self.days_in_stage = 0
        
        next_stage = self.stages[self.current_stage]
        return {
            'action': 'PROMOTE',
            'message': f"推进到阶段{self.current_stage+1}: {next_stage['description']}",
            'new_allocation': next_stage['pct'],
            'current_stage': self.current_stage + 1
        }
    
    def trigger_rollback(self, reason: str) -> dict:
        """触发回滚"""
        rollback_info = {
            'from_stage': self.current_stage + 1,
            'to_stage': max(1, self.current_stage),  # 回到前一阶段，最低回到1%
            'reason': reason,
            'timestamp': pd.Timestamp.now()
        }
        
        self.rollback_history.append(rollback_info)
        self.current_stage = max(0, self.current_stage - 1)
        
        return rollback_info

▶ 点击展开可浏览运行结果

📘 本段代码定义了 1 个函数/类：类 `CanaryReleaseManager`（金丝雀发布管理器）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。

回滚决策标准

回滚（Rollback）是金丝雀发布安全机制的核心。何时决定回滚需要明确的定量标准，而不是靠直觉判断。

硬性回滚条件（自动触发）

以下任一条件触发，系统自动回滚，无需人工干预：

累计亏损阈值：策略累计亏损超过当前阶段分配资金的 5%。
连续回撤：连续 3 个交易日出现亏损，且累计亏损超过 2%。
技术异常：信号缺失超过 30 分钟，或订单失败率超过 10%。
偏离预期：实盘与模拟盘的日内跟踪差异超过 10 倍标准差。

软性回滚条件（人工决策）

夏普比率下降：滚动 20 日夏普比率低于回测夏普的 50%。
相关性异常：新策略与现有策略池的相关性显著高于预期（>0.7），可能带来集中风险。
成交量异常：交易量或换手率与预期偏差超过 3 倍标准差。
市场环境变化：发生了回测中未覆盖的极端市场事件（如熔断、政策突变）。

回滚执行流程

1. 立即暂停新策略的所有新开仓订单
2. 对于已有持仓，执行平仓（使用 TWAP 算法减少市场冲击）
3. 通知所有相关方（交易团队、风控团队、技术团队）
4. 生成回滚事件报告
5. 排查根本原因，确定是策略问题还是执行问题
6. 修复后在更低的资金比例下重新开始测试

A/B测试和金丝雀发布不是一次性的流程，而是持续的策略治理机制。一个成熟的量化团队应当在任何时候都有 2-3 个策略处于金丝雀阶段，保持策略储备的持续更新。

18.3 A/B测试与金丝雀发布 ​

策略A/B测试的设计原则 ​

什么是策略A/B测试 ​

A/B测试的适用场景 ​

核心设计原则 ​

统计显著性判断 ​

两样本 t 检验 ​

替代检验方法 ​

样本量计算 ​

金丝雀发布流程 ​

渐进式放量 ​

回滚决策标准 ​

硬性回滚条件（自动触发） ​

软性回滚条件（人工决策） ​

回滚执行流程 ​