A/B 测试与灰度发布

from dataclasses import dataclass
from typing import List
import hashlib

@dataclass
class ExperimentConfig:
    experiment_id: str
    variants: List[str]
    traffic_split: List[float]  # 例如 [0.5, 0.5]
    primary_metric: str
    guardrail_metrics: List[str]

    def validate(self) -> bool:
        if len(self.variants) != len(self.traffic_split):
            raise ValueError("Variants and split must match")
        if abs(sum(self.traffic_split) - 1.0) > 1e-6:
            raise ValueError("Traffic split must sum to 1.0")
        return True

def assign_variant(user_id: str, config: ExperimentConfig) -> str:
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
    bucket = hash_val % 10000  # 0-9999
    cumulative = 0.0
    for i, split in enumerate(config.traffic_split):
        cumulative += split * 10000
        if bucket < cumulative:
            return config.variants[i]
    return config.variants[-1]

# 离线评估与在线评估的对比
class ModelEvaluation:
    def __init__(self, model, train_data, test_data):
        self.model = model
        self.train_data = train_data
        self.test_data = test_data

    def offline_eval(self) -> dict:
        metrics = {}
        metrics["accuracy"] = self.model.score(self.test_data.X, self.test_data.y)
        metrics["auc"] = roc_auc_score(self.test_data.y, self.model.predict_proba(self.test_data.X)[:, 1])
        return metrics

    def online_eval(self, experiment_config: ExperimentConfig) -> dict:
        online_metrics = {
            "click_through_rate": 0.0,
            "conversion_rate": 0.0,
            "revenue_per_user": 0.0,
            "latency_p99_ms": 0
        }
        return online_metrics

eval_system = ModelEvaluation(model, train_df, test_df)
print(f"Offline: {eval_system.offline_eval()}")

import math
from scipy import stats

def calculate_sample_size(
    baseline_rate: float,
    minimum_detectable_effect: float,
    alpha: float = 0.05,
    power: float = 0.80
) -> int:
    beta = 1 - power
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    z_beta = stats.norm.ppf(power)
    p1 = baseline_rate
    p2 = baseline_rate + minimum_detectable_effect
    p_bar = (p1 + p2) / 2
    n = ((z_alpha * math.sqrt(2 * p_bar * (1 - p_bar)) +
          z_beta * math.sqrt(p1 * (1 - p1) + p2 * (1 - p2)))  2 /
         (p2 - p1)  2)
    return math.ceil(n)

n = calculate_sample_size(
    baseline_rate=0.05,
    minimum_detectable_effect=0.005,
    alpha=0.05,
    power=0.80
)
print(f"Required sample size per group: {n:,}")

from enum import Enum
import random

class RandomizationUnit(Enum):
    USER_ID = "user_id"
    SESSION_ID = "session_id"
    REQUEST_ID = "request_id"

class ExperimentDesigner:
    def __init__(self, alpha=0.05, power=0.80):
        self.alpha = alpha
        self.power = power

    def design_experiment(
        self,
        baseline_rate: float,
        mde: float,
        daily_traffic: int,
        randomization_unit: RandomizationUnit
    ) -> dict:
        n_per_group = calculate_sample_size(baseline_rate, mde, self.alpha, self.power)
        total_n = n_per_group * 2
        duration_days = math.ceil(total_n / daily_traffic)
        return {
            "sample_size_per_group": n_per_group,
            "total_sample_size": total_n,
            "estimated_duration_days": duration_days,
            "randomization_unit": randomization_unit.value,
            "significance_level": self.alpha,
            "statistical_power": self.power
        }

designer = ExperimentDesigner()
config = designer.design_experiment(
    baseline_rate=0.03, mde=0.003, daily_traffic=50000,
    randomization_unit=RandomizationUnit.USER_ID
)
for k, v in config.items():
    print(f"  {k}: {v}")

from scipy import stats
import numpy as np

def ab_test_analysis(control_conversions, control_total,
                     treatment_conversions, treatment_total):
    control_rate = control_conversions / control_total
    treatment_rate = treatment_conversions / treatment_total
    # Z 检验
    p_pool = (control_conversions + treatment_conversions) / (control_total + treatment_total)
    se = math.sqrt(p_pool * (1 - p_pool) * (1/control_total + 1/treatment_total))
    z_stat = (treatment_rate - control_rate) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))
    # 置信区间
    diff = treatment_rate - control_rate
    margin = 1.96 * se
    ci_lower = diff - margin
    ci_upper = diff + margin
    # 效应量 (Cohen's h)
    h = 2 * np.arcsin(np.sqrt(treatment_rate)) - 2 * np.arcsin(np.sqrt(control_rate))
    return {
        "control_rate": control_rate,
        "treatment_rate": treatment_rate,
        "lift": (treatment_rate - control_rate) / control_rate,
        "z_statistic": z_stat,
        "p_value": p_value,
        "ci_95": (ci_lower, ci_upper),
        "effect_size_cohens_h": h,
        "significant": p_value < 0.05
    }

result = ab_test_analysis(250, 5000, 290, 5000)
for k, v in result.items():
    print(f"  {k}: {v}")

def sequential_analysis(control_metrics, treatment_metrics, alpha=0.05):
    sequential_results = []
    for day in range(1, len(control_metrics) + 1):
        c_sum = sum(control_metrics[:day])
        c_total = day * 1000
        t_sum = sum(treatment_metrics[:day])
        t_total = day * 1000
        result = ab_test_analysis(c_sum, c_total, t_sum, t_total)
        result["day"] = day
        result["cumulative_samples"] = day * 2000
        sequential_results.append(result)
        if result["significant"] and day >= 7:
            print(f"Day {day}: Significant! p={result['p_value']:.4f}")
            break
    return sequential_results

def check_power_post_hoc(effect_size, n_per_group, alpha=0.05):
    z_alpha = stats.norm.ppf(1 - alpha / 2)
    z_beta = effect_size * math.sqrt(n_per_group / 2) - z_alpha
    achieved_power = stats.norm.cdf(z_beta)
    return achieved_power

power = check_power_post_hoc(0.02, 50000)
print(f"Achieved power: {power:.2%}")

import numpy as np
from scipy.stats import beta

class ThompsonSampling:
    def __init__(self, n_arms):
        self.n_arms = n_arms
        self.successes = np.ones(n_arms)  # Beta 先验 alpha
        self.failures = np.ones(n_arms)   # Beta 先验 beta

    def select_arm(self) -> int:
        samples = np.random.beta(self.successes, self.failures)
        return int(np.argmax(samples))

    def update(self, arm: int, reward: float):
        if reward > 0:
            self.successes[arm] += 1
        else:
            self.failures[arm] += 1

    def get_traffic_allocation(self) -> np.ndarray:
        means = self.successes / (self.successes + self.failures)
        return means / means.sum()

ts = ThompsonSampling(n_arms=3)
for round_idx in range(1000):
    arm = ts.select_arm()
    reward = np.random.random() < [0.03, 0.05, 0.04][arm]
    ts.update(arm, float(reward))
print(f"Traffic allocation: {ts.get_traffic_allocation()}")

class UCB:
    def __init__(self, n_arms, c=2.0):
        self.n_arms = n_arms
        self.c = c
        self.counts = np.zeros(n_arms)
        self.total_reward = np.zeros(n_arms)
        self.total_pulls = 0

    def select_arm(self) -> int:
        untried = np.where(self.counts == 0)[0]
        if len(untried) > 0:
            return int(untried[0])
        mean_rewards = self.total_reward / self.counts
        confidence = self.c * np.sqrt(
            np.log(self.total_pulls) / self.counts
        )
        ucb_values = mean_rewards + confidence
        return int(np.argmax(ucb_values))

    def update(self, arm: int, reward: float):
        self.counts[arm] += 1
        self.total_reward[arm] += reward
        self.total_pulls += 1

    def regret(self, optimal_rate: float) -> float:
        return optimal_rate * self.total_pulls - self.total_reward.sum()

ucb = UCB(n_arms=3)
for _ in range(1000):
    arm = ucb.select_arm()
    reward = np.random.random() < [0.03, 0.05, 0.04][arm]
    ucb.update(arm, float(reward))
print(f"Cumulative regret: {ucb.regret(0.05):.2f}")

from dataclasses import dataclass
from enum import Enum
import time

class CanaryStage(Enum):
    BASELINE = "1% 基线流量"
    EARLY = "5% 早期用户"
    GROWING = "25% 增长阶段"
    MAJORITY = "50% 大部分用户"
    FULL = "100% 全量发布"

@dataclass
class CanaryConfig:
    stages: list = None
    stage_duration_min: int = 30
    auto_promote: bool = True
    auto_rollback: bool = True
    max_error_rate: float = 0.01
    max_latency_p99_ms: int = 500

    def __post_init__(self):
        if self.stages is None:
            self.stages = [1, 5, 25, 50, 100]

class CanaryDeployer:
    def __init__(self, config: CanaryConfig):
        self.config = config
        self.current_stage = 0
        self.current_traffic_pct = 0

    def deploy_next_stage(self, health_metrics: dict) -> bool:
        if self.current_traffic_pct == 0:
            self.current_traffic_pct = self.config.stages[0]
            self.current_stage = 0
            return True

        if health_metrics["error_rate"] > self.config.max_error_rate:
            return False
        if health_metrics["latency_p99"] > self.config.max_latency_p99_ms:
            return False

        self.current_stage += 1
        if self.current_stage >= len(self.config.stages):
            self.current_traffic_pct = 100
            print("Full rollout complete!")
            return True

        self.current_traffic_pct = self.config.stages[self.current_stage]
        print(f"Promoting to {self.current_traffic_pct}% traffic")
        return True

# istio-virtual-service-canary.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: model-serving-canary
spec:
  hosts:
    - model-api.internal
  http:
    - route:
        - destination:
            host: model-serving
            subset: stable
          weight: 90
        - destination:
            host: model-serving
            subset: canary
          weight: 10
      timeout: 2s
      retries:
        attempts: 3
        perTryTimeout: 500ms
---
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: model-serving-versions
spec:
  host: model-serving
  subsets:
    - name: stable
      labels:
        version: v1.2.0
    - name: canary
      labels:
        version: v1.3.0-rc1

import time
import logging
from typing import Optional

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("model_rollback")

class ModelRollbackManager:
    def __init__(self):
        self.active_version: Optional[str] = None
        self.previous_version: Optional[str] = None
        self.version_history = []
        self.rollback_thresholds = {
            "error_rate": 0.05,
            "latency_p99_ms": 1000,
            "prediction_drift_psi": 0.25,
            "business_metric_drop_pct": 5.0
        }

    def check_and_rollback(self, metrics: dict) -> bool:
        violations = []
        if metrics.get("error_rate", 0) > self.rollback_thresholds["error_rate"]:
            violations.append(f"Error rate {metrics['error_rate']:.2%} exceeds threshold")
        if metrics.get("latency_p99_ms", 0) > self.rollback_thresholds["latency_p99_ms"]:
            violations.append(f"Latency {metrics['latency_p99_ms']}ms exceeds threshold")
        if metrics.get("prediction_drift_psi", 0) > self.rollback_thresholds["prediction_drift_psi"]:
            violations.append(f"PSI {metrics['prediction_drift_psi']:.3f} exceeds threshold")
        if violations:
            logger.warning(f"Rollback triggered: {violations}")
            return self.execute_rollback()
        return True

    def execute_rollback(self) -> bool:
        if not self.previous_version:
            logger.error("No previous version to rollback to")
            return False
        logger.info(f"Rolling back from {self.active_version} to {self.previous_version}")
        self.active_version, self.previous_version = self.previous_version, self.active_version
        logger.info(f"Rollback complete. Active version: {self.active_version}")
        return True

import numpy as np

def calculate_psi(expected: np.ndarray, actual: np.ndarray, bins=10) -> float:
    psi_values = []
    min_val = min(expected.min(), actual.min())
    max_val = max(expected.max(), actual.max())
    bin_edges = np.linspace(min_val, max_val, bins + 1)
    for i in range(bins):
        expected_pct = ((expected >= bin_edges[i]) & (expected < bin_edges[i+1])).mean()
        actual_pct = ((actual >= bin_edges[i]) & (actual < bin_edges[i+1])).mean()
        expected_pct = max(expected_pct, 1e-6)
        actual_pct = max(actual_pct, 1e-6)
        psi = (expected_pct - actual_pct) * np.log(expected_pct / actual_pct)
        psi_values.append(psi)
    return sum(psi_values)

baseline_preds = np.random.beta(2, 5, 10000)
current_preds = np.random.beta(2.5, 5.5, 10000)
psi = calculate_psi(baseline_preds, current_preds)
print(f"PSI: {psi:.4f}")
if psi > 0.25:
    print("Warning: Significant prediction distribution drift detected!")
elif psi > 0.1:
    print("Caution: Moderate drift detected")
else:
    print("OK: Prediction distribution is stable")

import mlflow
from mlflow.tracking import MlflowClient

class MLPipelineWithABTest:
    def __init__(self, tracking_uri, experiment_name):
        mlflow.set_tracking_uri(tracking_uri)
        mlflow.set_experiment(experiment_name)
        self.client = MlflowClient()

    def register_candidate(self, run_id, model_name):
        model_uri = f"runs:/{run_id}/model"
        result = mlflow.register_model(model_uri, model_name)
        self.client.transition_model_version_stage(
            name=model_name, version=result.version, stage="Staging"
        )
        print(f"Registered {model_name} v{result.version} as Staging")
        return result.version

    def get_production_model(self, model_name):
        versions = self.client.search_model_versions(
            f"name='{model_name}'"
        )
        for v in versions:
            if v.current_stage == "Production":
                return v.version
        return None

    def promote_after_ab_test(self, model_name, version):
        self.client.transition_model_version_stage(
            name=model_name, version=version, stage="Production"
        )
        print(f"Promoted {model_name} v{version} to Production")

pipeline = MLPipelineWithABTest(
    tracking_uri="http://localhost:5000",
    experiment_name="recommendation_model"
)
version = pipeline.register_candidate("abc123def456", "rec_model")

# 完整的 A/B 测试 + 灰度发布流水线
import json
from datetime import datetime

class FullReleasePipeline:
    def __init__(self, ml_pipeline: MLPipelineWithABTest):
        self.ml = ml_pipeline
        self.ab_results = {}
        self.canary_stage = 0

    def run_full_pipeline(self, model_name, candidate_version):
        print(f"=== Starting release pipeline for {model_name} v{candidate_version} ===")
        # Step 1: Shadow mode validation
        print("[1/4] Shadow mode validation...")
        shadow_metrics = self._run_shadow_test(model_name, candidate_version)
        if not self._check_shadow_metrics(shadow_metrics):
            print("Shadow test failed. Aborting.")
            return False
        # Step 2: A/B testing
        print("[2/4] Running A/B test...")
        ab_result = self._run_ab_test(model_name, candidate_version)
        self.ab_results = ab_result
        if not ab_result["significant"] or ab_result["lift"] <= 0:
            print(f"A/B test not significant. Lift: {ab_result['lift']:.2%}")
            return False
        # Step 3: Canary deployment
        print("[3/4] Canary deployment...")
        canary_ok = self._run_canary_deployment(model_name, candidate_version)
        if not canary_ok:
            print("Canary deployment failed. Rolling back.")
            return False
        # Step 4: Full rollout
        print("[4/4] Full rollout...")
        self.ml.promote_after_ab_test(model_name, candidate_version)
        print("Release pipeline completed successfully!")
        return True

    def _run_shadow_test(self, model_name, version):
        return {"latency_p99": 120, "error_rate": 0.001}

    def _check_shadow_metrics(self, metrics):
        return metrics["error_rate"] < 0.01

    def _run_ab_test(self, model_name, version):
        return {"significant": True, "lift": 0.08, "p_value": 0.003}

    def _run_canary_deployment(self, model_name, version):
        return True