企业官网建设流程全解析

一、为什么需要智能实验

传统 A/B 的问题 ──────────────────────────────────────────────────── 场景：你要优化一个推荐系统的点击率 传统做法： - 方案A：协同过滤 - 方案B：内容推荐 - 方案C：混合推荐 你把流量三等分，跑了 2 周，发现方案C赢了。 但在这 2 周里，70% 的用户一直在看较差的推荐。 如果方案C确实更好，为什么不让更多人早点享受到？ 这就是 Bandit 解决的问题：边探索边利用，减少"试错成本"。

二、多臂老虎机（Multi-Armed Bandit）

# bandit/core.py from typing import List, Dict, Optional import numpy as np import random from dataclasses import dataclass, field from abc import ABC, abstractmethod @dataclass class Arm: """老虎机的摇臂（一个变体）""" id: str name: str trials: int = 0 # 被选择的次数 rewards: float = 0.0 # 累计奖励 mean: float = 0.0 # 当前平均奖励 class BanditStrategy(ABC): """Bandit 策略基类""" def __init__(self, arms: List[Arm]): self.arms = arms self.total_trials = 0 @abstractmethod def select_arm(self) -> Arm: """选择一个摇臂""" pass def update(self, arm: Arm, reward: float): """更新摇臂的统计信息""" arm.trials += 1 arm.rewards += reward arm.mean = arm.rewards / arm.trials self.total_trials += 1 # ─── ε-Greedy ─── class EpsilonGreedy(BanditStrategy): """ ε-Greedy 策略 以 ε 的概率随机探索，以 1-ε 的概率选择当前最优 """ def __init__(self, arms: List[Arm], epsilon: float = 0.1): super().__init__(arms) self.epsilon = epsilon def select_arm(self) -> Arm: if random.random() < self.epsilon: # 探索：随机选一个 return random.choice(self.arms) else: # 利用：选当前均值最高的 return max(self.arms, key=lambda a: a.mean) # ─── UCB (Upper Confidence Bound) ─── class UCB1(BanditStrategy): """ UCB1 策略 选择上置信界最高的摇臂：mean + sqrt(2 * ln(N) / n) 天然平衡探索与利用 """ def select_arm(self) -> Arm: total = self.total_trials # 每个臂至少选一次 for arm in self.arms: if arm.trials == 0: return arm # 计算 UCB 值 log_total = np.log(total) ucb_values = [] for arm in self.arms: exploitation = arm.mean exploration = np.sqrt(2 * log_total / arm.trials) ucb = exploitation + exploration ucb_values.append(ucb) best_idx = np.argmax(ucb_values) return self.arms[best_idx] # ─── Thompson Sampling ─── class ThompsonSampling(BanditStrategy): """ Thompson Sampling（汤普森采样） 对每个臂维护一个 Beta 分布，从分布中采样，选最大的 适合二项分布奖励（点击/不点击） """ def __init__(self, arms: List[Arm]): super().__init__(arms) # Beta 分布参数：alpha = 成功次数+1, beta = 失败次数+1 self.alphas = {arm.id: 1.0 for arm in arms} self.betas = {arm.id: 1.0 for arm in arms} def select_arm(self) -> Arm: samples = [] for arm in self.arms: # 从 Beta 分布采样 sample = np.random.beta(self.alphas[arm.id], self.betas[arm.id]) samples.append(sample) best_idx = np.argmax(samples) return self.arms[best_idx] def update(self, arm: Arm, reward: float): super().update(arm, reward) # 更新 Beta 分布参数 if reward >= 0.5: # 视为成功 self.alphas[arm.id] += 1.0 else: self.betas[arm.id] += 1.0 # ─── 模拟对比 ─── def simulate_bandit(strategy_class, arm_means: List[float], total_trials: int = 10000, **kwargs) -> Dict: """ 模拟 Bandit 策略的表现 Args: strategy_class: Bandit 策略类 arm_means: 每个臂的真实平均奖励 total_trials: 总试验次数 """ arms = [Arm(id=f"arm{i}", name=f"变体{i}") for i in range(len(arm_means))] strategy = strategy_class(arms, **kwargs) cumulative_reward = 0.0 optimal_reward = max(arm_means) # 理想情况下每次都选最优 regret = [] # 累积遗憾 for t in range(total_trials): selected_arm = strategy.select_arm() idx = int(selected_arm.id.replace("arm", "")) # 模拟奖励（伯努利分布） reward = 1.0 if random.random() < arm_means[idx] else 0.0 strategy.update(selected_arm, reward) cumulative_reward += reward # 计算遗憾 instant_regret = optimal_reward - arm_means[idx] regret.append(instant_regret) total_regret = sum(regret) return { "strategy": strategy_class.__name__, "total_trials": total_trials, "cumulative_reward": cumulative_reward, "optimal_possible": optimal_reward * total_trials, "total_regret": total_regret, "arm_stats": [ { "name": arm.name, "trials": arm.trials, "mean": arm.mean, "true_mean": arm_means[i], } for i, arm in enumerate(arms) ], } # ─── 运行对比 ─── if __name__ == "__main__": # 三个变体，真实点击率分别为 5%, 8%, 12% arm_means = [0.05, 0.08, 0.12] strategies = [ ("ε-Greedy (ε=0.1)", EpsilonGreedy, {"epsilon": 0.1}), ("UCB1", UCB1, {}), ("Thompson Sampling", ThompsonSampling, {}), ] for name, cls, kwargs in strategies: result = simulate_bandit(cls, arm_means, total_trials=5000, **kwargs) print(f"\n{'='*40}") print(f"策略：{name}") print(f"累计奖励：{result['cumulative_reward']:.0f}") print(f"理论最优：{result['optimal_possible']:.0f}") print(f"总遗憾：{result['total_regret']:.2f}") print("各臂情况：") for arm_stat in result["arm_stats"]: print(f" {arm_stat['name']}: 被选 {arm_stat['trials']} 次, " f"平均奖励 {arm_stat['mean']:.4f}, " f"真实值 {arm_stat['true_mean']:.4f}")

三、什么时候用 Bandit，什么时候用 A/B

# bandit/decider.py from enum import Enum from typing import Dict class ExperimentType(Enum): FIXED_AB = "fixed_ab" # 固定比例 A/B 测试 BANDIT = "bandit" # 多臂老虎机 HYBRID = "hybrid" # 混合：先 A/B 后 Bandit class ExperimentDecider: """实验类型决策器""" def recommend(self, scenario: Dict) -> Dict: """ 根据场景推荐实验类型 考量因素： - 风险等级：高风险必须用固定 A/B - 评估周期：需要长期评估的用固定 A/B - 样本量：样本充足的可以用 Bandit - 指标类型：二项分布适合 Thompson Sampling """ risk_level = scenario.get("risk_level", "low") eval_period = scenario.get("eval_period", "short") sample_size = scenario.get("sample_size", "large") metric_type = scenario.get("metric_type", "binomial") recommendations = [] # 风险判断 if risk_level == "high": recommendations.append({ "type": ExperimentType.FIXED_AB, "reason": "高风险变更（如支付流程），必须用固定 A/B 确保统计有效性", "confidence": 0.95, }) if risk_level == "low" and sample_size == "large": recommendations.append({ "type": ExperimentType.BANDIT, "reason": "低风险 + 大流量，Bandit 可以减少试错成本", "confidence": 0.90, }) if eval_period == "long": recommendations.append({ "type": ExperimentType.FIXED_AB, "reason": "需要评估长期效应（如留存），固定 A/B 更可靠", "confidence": 0.88, }) if metric_type == "continuous" and sample_size == "small": recommendations.append({ "type": ExperimentType.FIXED_AB, "reason": "连续指标 + 小样本，Bandit 的探索成本太高", "confidence": 0.82, }) if risk_level == "medium" and sample_size == "very_large": recommendations.append({ "type": ExperimentType.HYBRID, "reason": "中等风险 + 超大流量，可以先 A/B 确定方向再用 Bandit 优化", "confidence": 0.78, }) # 综合推荐 types = [r["type"] for r in recommendations] most_common = max(set(types), key=types.count) if types else ExperimentType.FIXED_AB return { "recommended": most_common.value, "details": recommendations, "scenario_analysis": { "risk_level": risk_level, "eval_period": eval_period, "sample_size": sample_size, "metric_type": metric_type, } } # ─── 使用示例 ─── def decide_experiment(): decider = ExperimentDecider() scenarios = [ { "name": "首页 Banner 颜色测试", "risk_level": "low", "eval_period": "short", "sample_size": "large", "metric_type": "binomial", }, { "name": "支付流程改版", "risk_level": "high", "eval_period": "medium", "sample_size": "large", "metric_type": "continuous", }, { "name": "推荐算法升级", "risk_level": "medium", "eval_period": "long", "sample_size": "very_large", "metric_type": "continuous", }, ] for scenario in scenarios: result = decider.recommend(scenario) print(f"\n场景：{scenario['name']}") print(f"推荐方案：{result['recommended']}") for detail in result['details']: print(f" - {detail['reason']}")

四、贝叶斯 A/B 测试

# bayesian/ab_test.py from typing import List, Tuple, Dict import numpy as np from scipy import stats from dataclasses import dataclass @dataclass class BayesianResult: """贝叶斯 A/B 测试结果""" variant_a_success: int variant_a_total: int variant_b_success: int variant_b_total: int prob_a_better: float # A 优于 B 的概率 prob_better_a: float # B 优于 A 的概率 expected_loss_a: float # 选 A 的期望损失 expected_loss_b: float # 选 B 的期望损失 credible_interval_a: Tuple[float, float] credible_interval_b: Tuple[float, float] class BayesianABTest: """ 贝叶斯 A/B 测试 优势： - 直接告诉你"A 比 B 好的概率" - 支持受控早停（不用等到固定样本量） - 结果解释更直观 """ def __init__(self, alpha_prior: float = 1.0, beta_prior: float = 1.0): """ Args: alpha_prior, beta_prior: Beta 先验分布参数 默认 Uniform(1,1) 无信息先验 """ self.alpha_prior = alpha_prior self.beta_prior = beta_prior def analyze(self, a_success: int, a_total: int, b_success: int, b_total: int, simulations: int = 100000) -> BayesianResult: """ 执行贝叶斯分析 Args: a_success, a_total: A 组的成功数和总样本 b_success, b_total: B 组的成功数和总样本 simulations: 蒙特卡洛模拟次数 """ # 后验分布参数 alpha_a = self.alpha_prior + a_success beta_a = self.beta_prior + a_total - a_success alpha_b = self.alpha_prior + b_success beta_b = self.beta_prior + b_total - b_success # 从后验分布采样 samples_a = np.random.beta(alpha_a, beta_a, simulations) samples_b = np.random.beta(alpha_b, beta_b, simulations) # 计算 A > B 的概率 prob_a_better = np.mean(samples_a > samples_b) # 计算期望损失 loss_a = np.mean(np.maximum(0, samples_b - samples_a)) loss_b = np.mean(np.maximum(0, samples_a - samples_b)) # 可信区间 ci_a = tuple(np.percentile(samples_a, [2.5, 97.5])) ci_b = tuple(np.percentile(samples_b, [2.5, 97.5])) return BayesianResult( variant_a_success=a_success, variant_a_total=a_total, variant_b_success=b_success, variant_b_total=b_total, prob_a_better=prob_a_better, prob_better_a=1 - prob_a_better, expected_loss_a=loss_a, expected_loss_b=loss_b, credible_interval_a=ci_a, credible_interval_b=ci_b, ) def should_stop_early(self, result: BayesianResult, prob_threshold: float = 0.95, loss_threshold: float = 0.005) -> Dict: """ 判断是否可以提前停止 Args: result: 贝叶斯分析结果 prob_threshold: 胜出概率阈值 loss_threshold: 期望损失阈值 """ stop_reasons = [] if result.prob_a_better >= prob_threshold: stop_reasons.append(f"A 胜出概率 {result.prob_a_better:.4f} ≥ {prob_threshold}") if result.prob_better_a >= prob_threshold: stop_reasons.append(f"B 胜出概率 {result.prob_better_a:.4f} ≥ {prob_threshold}") if result.expected_loss_a <= loss_threshold: stop_reasons.append(f"选 A 的期望损失 {result.expected_loss_a:.4f} ≤ {loss_threshold}") if result.expected_loss_b <= loss_threshold: stop_reasons.append(f"选 B 的期望损失 {result.expected_loss_b:.4f} ≤ {loss_threshold}") return { "should_stop": len(stop_reasons) > 0, "reasons": stop_reasons, "winner": "A" if result.prob_a_better >= prob_threshold else \ "B" if result.prob_better_a >= prob_threshold else \ "inconclusive", } # ─── 与频率学派对比 ─── def compare_frequentist_vs_bayesian(): """对比频率学派和贝叶斯方法""" # 同样的数据 a_success, a_total = 120, 1000 # 12% b_success, b_total = 150, 1000 # 15% print("="*50) print("频率学派（Z 检验）：") from statsmodels.stats.proportion import proportions_ztest z_stat, p_value = proportions_zcount = [a_success, b_success] z_stat, p_value = proportions_ztest( [a_success, b_success], [a_total, b_total] ) print(f" Z 统计量: {z_stat:.4f}") print(f" P 值: {p_value:.4f}") print(f" 结论: {'显著' if p_value < 0.05 else '不显著'}") print("\n贝叶斯方法：") bayes = BayesianABTest() result = bayes.analyze(a_success, a_total, b_success, b_total) print(f" A 优于 B 的概率: {result.prob_a_better:.4f}") print(f" B 优于 A 的概率: {result.prob_better_a:.4f}") print(f" A 的可信区间: ({result.credible_interval_a[0]:.4f}, {result.credible_interval_a[1]:.4f})") print(f" B 的可信区间: ({result.credible_interval_b[0]:.4f}, {result.credible_interval_b[1]:.4f})") print(f" 选 A 的期望损失: {result.expected_loss_a:.6f}") print(f" 选 B 的期望损失: {result.expected_loss_b:.6f}") stop_decision = bayes.should_stop_early(result) print(f"\n早停判断: {'可以停止' if stop_decision['should_stop'] else '继续收集数据'}") if stop_decision['should_stop']: print(f" 胜出者: {stop_decision['winner']}") for reason in stop_decision['reasons']: print(f" - {reason}")

五、CUPED 方差缩减

# variance_reduction/cuped.py from typing import List, Dict, Optional import numpy as np from scipy import stats from dataclasses import dataclass @dataclass class CUPEDResult: """CUPED 方差缩减结果""" original_mean: float original_variance: float adjusted_mean: float adjusted_variance: float variance_reduction: float # 方差缩减比例 theta: float # 最优系数 sample_size: int class CUPED: """ CUPED（Controlled-experiment Using Pre-Experiment Data） 核心思想：利用实验前的数据来降低实验指标的方差 效果：同等样本量下，统计功效更高；或同等功效下，所需样本更少 """ def __init__(self): self.theta: Optional[float] = None def fit(self, pre_experiment: List[float], experiment: List[float]) -> CUPEDResult: """ 拟合 CUPED 模型 Args: pre_experiment: 实验前的指标值（协变量） experiment: 实验期间的指标值 """ X = np.array(pre_experiment) Y = np.array(experiment) n = len(X) # 计算最优 theta cov_xy = np.cov(X, Y)[0, 1] var_x = np.var(X, ddof=1) theta = cov_xy / var_x if var_x > 0 else 0.0 # 计算调整后的 Y Y_adjusted = Y - theta * (X - np.mean(X)) # 计算方差 original_var = np.var(Y, ddof=1) adjusted_var = np.var(Y_adjusted, ddof=1) variance_reduction = 1 - adjusted_var / original_var if original_var > 0 else 0.0 self.theta = theta return CUPEDResult( original_mean=np.mean(Y), original_variance=original_var, adjusted_mean=np.mean(Y_adjusted), adjusted_variance=adjusted_var, variance_reduction=variance_reduction, theta=theta, sample_size=n, ) def required_sample_size(self, effect_size: float, power: float = 0.8, alpha: float = 0.05, variance_reduction: float = 0.0) -> Dict: """ 计算所需样本量（考虑方差缩减） Args: effect_size: 最小可检测效应量 power: 统计功效 alpha: 显著性水平 variance_reduction: CUPED 带来的方差缩减比例 """ # 标准样本量公式（双样本 t 检验） z_alpha = stats.norm.ppf(1 - alpha / 2) z_beta = stats.norm.ppf(power) # 不考虑方差缩减的标准样本量 standard_n = int(np.ceil( 2 * ((z_alpha + z_beta) / effect_size) ** 2 )) # 考虑方差缩减后的样本量 reduced_n = int(np.ceil(standard_n * (1 - variance_reduction))) return { "standard_sample_size": standard_n, "cuped_sample_size": reduced_n, "savings": standard_n - reduced_n, "savings_percentage": round((1 - reduced_n/standard_n) * 100, 1), } # ─── 使用示例 ─── def demonstrate_cuped(): """演示 CUPED 的效果""" np.random.seed(42) # 生成模拟数据 n = 1000 # 实验前数据（协变量） pre = np.random.normal(100, 15, n) # 实验数据：处理效应 + 噪声 treatment_effect = 2.0 post = pre + treatment_effect + np.random.normal(0, 10, n) # 对照组（无处理效应） control = pre + np.random.normal(0, 10, n) print("="*50) print("CUPED 方差缩减演示") print("="*50) cuped = CUPED() # 对照组 result_control = cuped.fit(pre, control) print(f"\n对照组:") print(f" 原始方差: {result_control.original_variance:.4f}") print(f" 调整后方差: {result_control.adjusted_variance:.4f}") print(f" 方差缩减: {result_control.variance_reduction*100:.1f}%") print(f" 最优 theta: {result_control.theta:.4f}") # 实验组 result_treatment = cuped.fit(pre, post) print(f"\n实验组:") print(f" 原始方差: {result_treatment.original_variance:.4f}") print(f" 调整后方差: {result_treatment.adjusted_variance:.4f}") print(f" 方差缩减: {result_treatment.variance_reduction*100:.1f}%") print(f" 最优 theta: {result_treatment.theta:.4f}") # 样本量节省 sample_req = cuped.required_sample_size( effect_size=0.2, variance_reduction=result_treatment.variance_reduction ) print(f"\n样本量需求:") print(f" 标准方法: {sample_req['standard_sample_size']} 人/组") print(f" CUPED 后: {sample_req['cuped_sample_size']} 人/组") print(f" 节省: {sample_req['savings']} 人 ({sample_req['savings_percentage']}%)")

六、序贯检验（Sequential Testing）

# sequential/sequential_test.py from typing import List, Optional, Dict import numpy as np from dataclasses import dataclass @dataclass class SequentialBoundary: """序贯检验边界""" look_index: int # 第几次期中分析 information_time: float # 信息时间（样本量占比） z_boundary: float # Z 值边界 class SequentialTesting: """ 序贯检验 解决"天天偷看 p 值"的问题 允许在实验过程中多次分析，但使用校正后的边界 """ def __init__(self, looks: int = 5, alpha: float = 0.05): """ Args: looks: 计划进行的期中分析次数 alpha: 总的显著性水平 """ self.looks = looks self.alpha = alpha self.boundaries = self._compute_boundaries() def _compute_boundaries(self) -> List[SequentialBoundary]: """ 计算 Pocock 边界（等耗散函数） 每次分析的边界相同 """ boundaries = [] for i in range(1, self.looks + 1): info_time = i / self.looks # Pocock 边界近似 z_boundary = 2.073 # 5 次分析时的 Pocock 边界 boundaries.append(SequentialBoundary( look_index=i, information_time=info_time, z_boundary=z_boundary, )) return boundaries def analyze(self, group_a: List[float], group_b: List[float], look_index: int) -> Dict: """ 执行一次期中分析 Args: group_a, group_b: 两组数据 look_index: 第几次分析（从 1 开始） """ n_a, n_b = len(group_a), len(group_b) # 计算 Z 统计量 mean_a, mean_b = np.mean(group_a), np.mean(group_b) var_a, var_b = np.var(group_a, ddof=1), np.var(group_b, ddof=1) se = np.sqrt(var_a/n_a + var_b/n_b) z_stat = (mean_b - mean_a) / se if se > 0 else 0.0 # 获取当前边界 boundary = self.boundaries[look_index - 1] # 判断是否显著 is_significant = abs(z_stat) > boundary.z_boundary return { "look_index": look_index, "information_time": boundary.information_time, "z_statistic": round(z_stat, 4), "z_boundary": boundary.z_boundary, "is_significant": is_significant, "mean_diff": round(mean_b - mean_a, 4), "sample_size_a": n_a, "sample_size_b": n_b, } def get_final_decision(self, analyses: List[Dict]) -> Dict: """综合多次分析结果做出最终决策""" for analysis in analyses: if analysis["is_significant"]: return { "decision": "significant", "stopped_at_look": analysis["look_index"], "final_z": analysis["z_statistic"], "mean_diff": analysis["mean_diff"], } return { "decision": "not_significant", "stopped_at_look": None, "final_z": analyses[-1]["z_statistic"], "mean_diff": analyses[-1]["mean_diff"], } # ─── 演示：偷看 p 值的危害 ─── def demonstrate_peeking_problem(): """演示频繁偷看 p 值会导致假阳性飙升""" np.random.seed(123) # 模拟两个完全相同的组（真实效应 = 0） n_per_group = 500 print("="*50) print("频繁偷看 p 值的危害") print("两组真实效应 = 0，不应该得出显著结论") print("="*50) # 模拟每天偷看一次 daily_results = [] for day in range(1, 31): # 每天新增一些数据 new_a = np.random.normal(0, 1, 50) new_b = np.random.normal(0, 1, 50) if day == 1: group_a, group_b = new_a, new_b else: group_a = np.concatenate([group_a, new_a]) group_b = np.concatenate([group_b, new_b]) # 做 t 检验 t_stat, p_value = stats.ttest_ind(group_a, group_b) daily_results.append({ "day": day, "sample_size": len(group_a), "p_value": p_value, "significant": p_value < 0.05, }) significant_days = sum(1 for r in daily_results if r["significant"]) print(f"\n30 天内发现有显著结果的次数: {significant_days}") print(f"假阳性率: {significant_days/30 * 100:.1f}%") print("（理论上应该只有 5% 的天数显著，但实际上远高于此）") print("\n序贯检验可以解决这个问题：") seq_test = SequentialTesting(looks=5) # 只在计划的 5 个时间点分析 for look in range(1, 6): split_point = look * 100 result = seq_test.analyze( group_a[:split_point], group_b[:split_point], look ) print(f" 第 {look} 次分析: Z={result['z_statistic']:.3f}, " f"边界={result['z_boundary']:.3f}, " f"{'显著' if result['is_significant'] else '不显著'}")

七、本讲坑位

八、本讲小结

┌─────────────────────────────────────────────────────────────┐ │ 番外3 核心收获 │ │ │ │ ✅ 理解三种 Bandit 策略：ε-Greedy / UCB / Thompson Sampling │ │ ✅ 掌握 Bandit 和固定 A/B 的选择依据 │ │ ✅ 实现贝叶斯 A/B 测试（后验概率、期望损失、早停判断） │ │ ✅ 实现 CUPED 方差缩减（节省 30-50% 样本量） │ │ ✅ 理解序贯检验与偷看 p 值的危害 │ │ │ │ 核心理念：智能实验的本质是用更少的样本、更快的时间、 │ │ 做出同样可靠的决策 │ └─────────────────────────────────────────────────────────────┘

配套工具推荐

🧰 开发之余的小工具推荐

在做 Bandit 模拟和贝叶斯计算时，经常需要处理大量数值运算、概率计算和结果可视化：

• 百分比计算：zz365.top 的百分比工具可以快速计算不同变体的提升幅度和置信区间
• 小数与分数互转：处理效应量（Cohen's d）、Beta 分布参数时很方便
• JSON 格式化：将模拟结果的嵌套 JSON 展开，直观对比各策略的累计奖励和遗憾值
• UUID 生成器：为每次实验、每个变体分配唯一标识