直接替换 vs 灰度发布：

  直接替换:
  v1.0 ──────────────→ v2.0 (100% 切换)
       │                │
       │                └─→ 如果 v2.0 有 bug → 全量故障
       │
  灰度发布:
  v1.0 ────────── 90% ──────────── 80% ──── ... ──── 0%
  v2.0 ────────── 10% ──────────── 20% ──── ... ──── 100%
       │          │                │
       │          └── 监控指标      └── 无异常 → 继续增加
       │              异常 → 自动回滚

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class ReleaseStage(Enum):
    CANARY = "canary"            # 金丝雀 1-5%
    GRADUAL = "gradual"          # 灰度 5-50%
    MAJORITY = "majority"        # 多数 50-95%
    FULL = "full"                # 全量 100%
    ROLLBACK = "rollback"        # 回滚


@dataclass
class FunctionVersion:
    """函数版本"""
    function_id: str
    version: str                  # 语义化版本 e.g., "2.1.0"
    stage: ReleaseStage = ReleaseStage.CANARY
    traffic_weight: float = 0.0   # 0.0 - 1.0
    is_default: bool = False
    created_at: datetime = field(default_factory=datetime.utcnow)
    promoted_at: datetime | None = None
    metrics: VersionMetrics | None = None
    changelog: str = ""


@dataclass
class VersionMetrics:
    """版本指标"""
    total_invocations: int = 0
    success_count: int = 0
    error_count: int = 0
    timeout_count: int = 0
    avg_duration_ms: float = 0.0
    p99_duration_ms: float = 0.0
    avg_confidence: float = 0.0
    business_metric_a: float = 0.0   # 自定义业务指标
    business_metric_b: float = 0.0


@dataclass
class TrafficRule:
    """流量路由规则"""
    function_id: str
    rules: list[VersionWeight]
    sticky_key: str | None = None    # 用户粘性键（确保同一用户始终命中同一版本）


@dataclass
class VersionWeight:
    """版本权重"""
    version: str
    weight: float                     # 0.0 - 1.0
    conditions: dict[str, Any] = field(default_factory=dict)  # 条件路由

import hashlib
import random


class TrafficRouter:
    """流量路由器"""

    def __init__(self):
        self._rules: dict[str, TrafficRule] = {}

    def set_rule(self, rule: TrafficRule) -> None:
        self._rules[rule.function_id] = rule

    def route(self, function_id: str,
              context: dict[str, Any] | None = None) -> str:
        """根据权重路由到具体版本"""
        rule = self._rules.get(function_id)
        if rule is None:
            raise ValueError(f"No traffic rule for {function_id}")

        # 条件路由
        for vw in rule.rules:
            if vw.conditions and self._match_conditions(vw.conditions, context):
                return vw.version

        # 粘性路由
        if rule.sticky_key and context:
            sticky_value = context.get(rule.sticky_key, "")
            if sticky_value:
                return self._sticky_route(rule, sticky_value)

        # 权重路由
        return self._weighted_route(rule)

    def _weighted_route(self, rule: TrafficRule) -> str:
        """基于权重的随机路由"""
        rand = random.random()
        cumulative = 0.0
        for vw in rule.rules:
            cumulative += vw.weight
            if rand < cumulative:
                return vw.version
        return rule.rules[-1].version

    def _sticky_route(self, rule: TrafficRule,
                       sticky_value: str) -> str:
        """基于哈希的粘性路由"""
        hash_val = int(hashlib.md5(
            sticky_value.encode()
        ).hexdigest(), 16) % 10000

        cumulative = 0
        for vw in rule.rules:
            cumulative += int(vw.weight * 10000)
            if hash_val < cumulative:
                return vw.version
        return rule.rules[-1].version

    def _match_conditions(self, conditions: dict,
                           context: dict | None) -> bool:
        if context is None:
            return False
        for key, value in conditions.items():
            if context.get(key) != value:
                return False
        return True

@dataclass
class ABExperiment:
    """A/B 实验"""
    experiment_id: str
    function_id: str
    control_version: str              # A 组（对照）
    treatment_version: str            # B 组（实验）
    traffic_split: float = 0.1        # B 组流量比例
    primary_metric: str = "success_rate"
    secondary_metrics: list[str] = field(default_factory=list)
    min_sample_size: int = 1000
    confidence_level: float = 0.95
    max_duration_hours: int = 168     # 最长 7 天
    status: str = "running"           # running, concluded, stopped
    started_at: datetime = field(default_factory=datetime.utcnow)
    concluded_at: datetime | None = None
    result: ExperimentResult | None = None


@dataclass
class ExperimentResult:
    """实验结果"""
    winner: str                       # control / treatment / inconclusive
    primary_metric_control: float
    primary_metric_treatment: float
    relative_improvement: float
    p_value: float
    is_significant: bool
    confidence_interval: tuple[float, float]
    sample_size_control: int
    sample_size_treatment: int

import math


class StatisticalTester:
    """统计显著性检验"""

    @staticmethod
    def two_proportion_z_test(
        successes_a: int, total_a: int,
        successes_b: int, total_b: int,
        confidence_level: float = 0.95,
    ) -> ExperimentResult:
        """双比例 Z 检验"""
        p_a = successes_a / total_a if total_a > 0 else 0
        p_b = successes_b / total_b if total_b > 0 else 0

        # 合并比例
        p_pool = (successes_a + successes_b) / (total_a + total_b)

        # Z 统计量
        se = math.sqrt(p_pool * (1 - p_pool) * (1/total_a + 1/total_b))
        if se == 0:
            z_stat = 0
        else:
            z_stat = (p_b - p_a) / se

        # p-value（双尾检验）
        p_value = 2 * (1 - StatisticalTester._normal_cdf(abs(z_stat)))

        # 置信区间
        alpha = 1 - confidence_level
        z_critical = StatisticalTester._z_critical(alpha / 2)
        se_diff = math.sqrt(p_a * (1-p_a) / total_a + p_b * (1-p_b) / total_b)
        ci_lower = (p_b - p_a) - z_critical * se_diff
        ci_upper = (p_b - p_a) + z_critical * se_diff

        relative_improvement = (p_b - p_a) / p_a if p_a > 0 else 0
        is_significant = p_value < (1 - confidence_level)

        winner = "inconclusive"
        if is_significant:
            winner = "treatment" if p_b > p_a else "control"

        return ExperimentResult(
            winner=winner,
            primary_metric_control=p_a,
            primary_metric_treatment=p_b,
            relative_improvement=relative_improvement,
            p_value=p_value,
            is_significant=is_significant,
            confidence_interval=(ci_lower, ci_upper),
            sample_size_control=total_a,
            sample_size_treatment=total_b,
        )

    @staticmethod
    def _normal_cdf(x: float) -> float:
        return 0.5 * (1 + math.erf(x / math.sqrt(2)))

    @staticmethod
    def _z_critical(alpha: float) -> float:
        # 常用值近似
        if alpha <= 0.005:
            return 2.576
        elif alpha <= 0.01:
            return 2.326
        elif alpha <= 0.025:
            return 1.960
        elif alpha <= 0.05:
            return 1.645
        else:
            return 1.282

class GradualReleaseManager:
    """渐进式发布管理"""

    STAGE_WEIGHTS = {
        ReleaseStage.CANARY: 0.05,
        ReleaseStage.GRADUAL: 0.25,
        ReleaseStage.MAJORITY: 0.75,
        ReleaseStage.FULL: 1.0,
    }

    def __init__(self, router: TrafficRouter,
                 metrics_collector,
                 rollback_threshold: dict[str, float] | None = None):
        self._router = router
        self._metrics = metrics_collector
        self._thresholds = rollback_threshold or {
            "error_rate": 0.05,
            "p99_latency_ms": 500,
            "timeout_rate": 0.02,
        }

    async def promote(self, function_id: str,
                       new_version: str,
                       current_version: str) -> dict:
        """推进到下一发布阶段"""
        current_stage = self._get_stage(function_id, new_version)

        # 检查当前阶段指标
        metrics = await self._metrics.get_version_metrics(
            function_id, new_version
        )
        health = self._check_health(metrics)

        if not health["healthy"]:
            await self._rollback(function_id, new_version, current_version)
            return {
                "action": "rollback",
                "reason": health["violations"],
            }

        # 推进阶段
        next_stage = self._next_stage(current_stage)
        if next_stage is None:
            return {"action": "completed", "stage": "full"}

        weight = self.STAGE_WEIGHTS[next_stage]
        self._router.set_rule(TrafficRule(
            function_id=function_id,
            rules=[
                VersionWeight(new_version, weight),
                VersionWeight(current_version, 1.0 - weight),
            ],
        ))

        return {
            "action": "promoted",
            "stage": next_stage.value,
            "new_weight": weight,
        }

    def _check_health(self, metrics: VersionMetrics) -> dict:
        violations = []

        if metrics.total_invocations > 0:
            error_rate = metrics.error_count / metrics.total_invocations
            if error_rate > self._thresholds["error_rate"]:
                violations.append(
                    f"error_rate {error_rate:.2%} > {self._thresholds['error_rate']:.2%}"
                )

            timeout_rate = metrics.timeout_count / metrics.total_invocations
            if timeout_rate > self._thresholds["timeout_rate"]:
                violations.append(
                    f"timeout_rate {timeout_rate:.2%} > {self._thresholds['timeout_rate']:.2%}"
                )

        if metrics.p99_duration_ms > self._thresholds["p99_latency_ms"]:
            violations.append(
                f"p99_latency {metrics.p99_duration_ms}ms > {self._thresholds['p99_latency_ms']}ms"
            )

        return {"healthy": len(violations) == 0, "violations": violations}

    def _next_stage(self, current: ReleaseStage) -> ReleaseStage | None:
        order = [ReleaseStage.CANARY, ReleaseStage.GRADUAL,
                 ReleaseStage.MAJORITY, ReleaseStage.FULL]
        try:
            idx = order.index(current)
            return order[idx + 1] if idx + 1 < len(order) else None
        except ValueError:
            return ReleaseStage.CANARY

    async def _rollback(self, function_id: str,
                          bad_version: str,
                          good_version: str) -> None:
        self._router.set_rule(TrafficRule(
            function_id=function_id,
            rules=[VersionWeight(good_version, 1.0)],
        ))

    def _get_stage(self, function_id: str, version: str) -> ReleaseStage:
        rule = self._router._rules.get(function_id)
        if rule is None:
            return ReleaseStage.CANARY
        for vw in rule.rules:
            if vw.version == version:
                weight = vw.weight
                for stage, w in sorted(self.STAGE_WEIGHTS.items(),
                                       key=lambda x: x[1]):
                    if weight <= w:
                        return stage
        return ReleaseStage.CANARY

函数发布报告：
═══════════════════════════════════

  函数: fn-credit-custom-001
  版本: v1.2.0 -> v2.0.0
  发布阶段: GRADUAL (25%)
  持续时间: 4h 32m

  指标对比:
  ─────────────────────────────────
  指标          | v1.2.0  | v2.0.0   | 变化
  成功率        | 98.3%   | 98.7%    | +0.4%
  平均延迟      | 12.5ms  | 8.2ms    | -34.4%
  P99 延迟      | 45ms    | 28ms     | -37.8%
  超时率        | 0.3%    | 0.1%     | -66.7%
  平均置信度    | 0.72    | 0.75     | +4.2%

  A/B 检验结果:
  ─────────────────────────────────
  主指标(成功率): p-value = 0.032 (显著)
  相对提升: +0.41%
  95% 置信区间: [+0.05%, +0.77%]
  样本量: A=8,432  B=2,118
  结论: treatment 胜出

  建议: 指标全面优于旧版本，建议推进至 MAJORITY 阶段

syntax = "proto3";
package onto.function.version.v1;

service FunctionVersionService {
    rpc CreateVersion(CreateVersionRequest) returns (CreateVersionResponse);
    rpc Promote(PromoteRequest) returns (PromoteResponse);
    rpc Rollback(RollbackRequest) returns (RollbackResponse);
    rpc GetVersionMetrics(MetricsRequest) returns (MetricsResponse);
    rpc StartExperiment(ExperimentRequest) returns (ExperimentResponse);
    rpc GetExperimentResult(GetResultRequest) returns (ExperimentResultResponse);
    rpc SetTrafficRule(TrafficRuleRequest) returns (TrafficRuleResponse);
}

# 场景：信贷评分函数 v1 -> v2 的灰度发布

# 1. 注册新版本
new_spec = FunctionSpec(
    function_id="fn-credit-001",
    name="credit_scorer",
    language=FunctionLanguage.PYTHON,
    version="2.0.0",
    source_code="...",
    entry_point="score",
)
await registry.register_function(new_spec)

# 2. 设置金丝雀路由 (5%)
router.set_rule(TrafficRule(
    function_id="fn-credit-001",
    rules=[
        VersionWeight("2.0.0", 0.05),
        VersionWeight("1.2.0", 0.95),
    ],
    sticky_key="user_id",
))

# 3. 启动 A/B 实验
experiment = ABExperiment(
    experiment_id="exp-credit-v2",
    function_id="fn-credit-001",
    control_version="1.2.0",
    treatment_version="2.0.0",
    traffic_split=0.05,
    primary_metric="success_rate",
    min_sample_size=1000,
)

# 4. 等待数据积累后检查结果
result = tester.two_proportion_z_test(
    successes_a=8302, total_a=8432,
    successes_b=2091, total_b=2118,
)
# ExperimentResult(winner="treatment", p_value=0.032, is_significant=True)

# 5. 推进到下一阶段
release_result = await release_manager.promote(
    "fn-credit-001", "2.0.0", "1.2.0"
)
# {"action": "promoted", "stage": "gradual", "new_weight": 0.25}