무용성 중단(Futility Stopping)과 계층적 검정으로 카나리를 통계적으로 자동 종료하는 법: Beta-Spending 설계 실전 가이드

# Python 3.9+
import numpy as np
from scipy import stats
 
def conditional_power(
    z_current: float,    # 현재 시점의 검정 통계량
    t: float,            # 정보 분율 (현재 수집된 정보 / 계획된 총 정보)
    delta: float,        # 사전 가정 효과 크기 (표준화된 드리프트 파라미터)
    alpha: float = 0.025 # 단측 유의 수준
) -> float:
    """
    고정 효과 크기 가정 하 조건부 검정력 계산.
 
    수식 해설: 정보 분율 t 시점에서 관측된 Z 통계량 z_current를 기반으로
    최종 시점(t=1)의 검정 통계량 기댓값을 계산한다.
    여기서 delta는 n_total=1로 표준화된 드리프트 파라미터다.
    실제 샘플 수 기반 계산이 필요하면 delta에 sqrt(n_total)을 곱한 값을 전달하라.
 
    CP < 0.2이면 무용성 판정 후보.
    """
    z_alpha = stats.norm.ppf(1 - alpha)
    # 현재까지 수집된 정보(t)와 남은 정보(1-t)의 가중 합산
    z_final_mean = z_current * np.sqrt(t) + delta * np.sqrt(1 - t)
    cp = 1 - stats.norm.cdf(z_alpha - z_final_mean)
    return cp
 
# 예: 50% 데이터 수집 시점, 효과가 절반 수준만 관측됨
cp = conditional_power(z_current=0.8, t=0.5, delta=2.0)
print(f"조건부 검정력: {cp:.3f}")  # → 0.183 → 무용성 경계 검토

[1차 메트릭: 구매 전환율] ← α 전체 배분
        │
        │ 유의 (p < α)
        ▼
[2차 메트릭 그룹: 세션 시간, 클릭률] ← 각각 α/2로 Bonferroni
        │
        │ 그룹 내 하나 이상 유의 (disjunctive)
        ▼
[3차 메트릭 그룹: 페이지 로드, API 에러율] ← 각각 α/2로 Bonferroni

Z-통계량
  4 ┤
    │  ━━┓   ← 상한(efficacy): O'Brien-Fleming은 초반에 매우 높고
  3 ┤    ┗━━┓                    분석이 거듭될수록 점차 낮아짐
    │        ┗━━━┓
  2 ┤              ┗━━━━━━━━━━━ (최종: ≈ 고정 표본 기준과 유사)
  1 ┤
  0 ┼────────────────────────────────► 정보 분율 t
 -1 ┤  ─ ─ ─ ─┐   ← 하한(futility): 초반엔 음수(거의 무조건 통과)
    │           └ ─ ─┐               분석이 거듭될수록 점차 높아짐
  1 ┤                 └ ─ ─ ─ ─ ─  (최종: 상한과 수렴)
    │
    t=0.10    t=0.25   t=0.50   t=1.00
 
  통계량 > 상한 → 효과 탐지 → 배포 진행 또는 즉시 롤백
  통계량 < 하한 → 무용성  → 카나리 종료
  두 경계 사이  → 계속 진행

# install.packages("gsDesign")
library(gsDesign)
 
# 4회 분석(중간 3회 + 최종 1회), 단측 α=0.025, β=0.20(검정력 80%)
# delta1 = 0.3: Cohen's d 기준 "소-중간 효과 크기"
#   구매 전환율 맥락에서 약 0.5%p 차이에 해당.
#   과거 A/B 테스트 히스토리에서 "탐지할 가치 있는 최소 효과"로 결정하라.
design <- gsDesign(
  k         = 4,      # 중간 분석 3회 + 최종 1회
  test.type = 2,      # 양측 검정 (상한/하한 모두)
  alpha     = 0.025,  # 단측 α
  beta      = 0.20,   # 타입 II 오류 (검정력 80%)
  sfu       = sfLDOF, # Alpha-Spending: O'Brien-Fleming 유사
  sfl       = sfLDOF, # Beta-Spending: 무용성 경계
  delta1    = 0.3,    # 탐지 목표 최소 효과 크기 (Cohen's d)
  n.fix     = 1000    # 고정 표본 기준 샘플 수
)
 
print(design)
# 출력 예시:
# Analysis  N     Z (upper)  Z (lower/futility)
#    1      310    3.47          -0.48
#    2      620    2.78           0.94
#    3      930    2.29           1.71
#    4     1051    2.02           2.02

# Python 3.9+
from dataclasses import dataclass
from typing import Literal
 
@dataclass
class SequentialBoundary:
    upper: float  # efficacy (효과 탐지) 경계
    lower: float  # futility (무용성) 경계
 
# 예시 1의 gsDesign 출력에서 가져온 경계값
BOUNDARIES: dict[float, SequentialBoundary] = {
    0.10: SequentialBoundary(upper=3.47, lower=-0.48),
    0.25: SequentialBoundary(upper=2.78, lower=0.94),
    0.50: SequentialBoundary(upper=2.29, lower=1.71),
    1.00: SequentialBoundary(upper=2.02, lower=2.02),
}
 
def evaluate_canary(
    z_stat: float,
    traffic_fraction: float
) -> Literal["continue", "deploy", "stop_futility"]:
    """
    현재 검정 통계량과 트래픽 비율로 카나리 배포 결정.
    가장 최근에 통과한 체크포인트 기준을 사용한다.
    """
    checkpoints = sorted(BOUNDARIES.keys())
    passed = [c for c in checkpoints if c <= traffic_fraction]
    if not passed:
        return "continue"  # 아직 첫 체크포인트 미도달
    checkpoint = passed[-1]  # 가장 최근에 통과한 시점
    boundary = BOUNDARIES[checkpoint]
 
    if z_stat >= boundary.upper:
        return "deploy"           # 효과 탐지 → 전체 배포
    elif z_stat <= boundary.lower:
        return "stop_futility"    # 무용성 → 카나리 종료
    else:
        return "continue"         # 다음 분석까지 유지
 
# 사용 예시: 25% 체크포인트에서 z=0.72
result = evaluate_canary(z_stat=0.72, traffic_fraction=0.25)
print(f"카나리 판정: {result}")  # → stop_futility (0.72 < 0.94 하한)

# Python 3.9+
import numpy as np
from scipy import stats as scipy_stats
 
def compute_proportion_pvalue(
    obs_count: int, obs_total: int,
    ctrl_count: int, ctrl_total: int
) -> float:
    """비율 차이에 대한 양측 z-검정 p-value."""
    p_obs = obs_count / obs_total
    p_ctrl = ctrl_count / ctrl_total
    p_pool = (obs_count + ctrl_count) / (obs_total + ctrl_total)
    se = np.sqrt(p_pool * (1 - p_pool) * (1 / obs_total + 1 / ctrl_total))
    if se == 0:
        return 1.0
    z = (p_obs - p_ctrl) / se
    return float(2 * (1 - scipy_stats.norm.cdf(abs(z))))
 
 
def hierarchical_gate(
    p_values: dict[str, float],
    hierarchy: list[list[str]],
    alpha: float = 0.05
) -> dict[str, dict]:
    """
    계층적 게이트키핑: 상위 계층이 유의해야 하위 계층 검정 진행.
 
    각 계층 내에서 Bonferroni 보정 후 "하나라도 유의"(disjunctive)하면
    다음 계층 게이트를 연다. FWER <= alpha 보장.
 
    더 보수적인 "모두 유의"(conjunctive) 방식으로 바꾸려면
    gate_open = all(sig_list) 로 변경하라.
    """
    results: dict[str, dict] = {}
    gate_open = True
 
    for metric_group in hierarchy:
        if not gate_open:
            for name in metric_group:
                results[name] = {"tested": False, "reason": "gate_closed"}
            continue
 
        alpha_adj = alpha / len(metric_group)  # 계층 내 Bonferroni
        level_any_significant = False
 
        for name in metric_group:
            p = p_values.get(name, 1.0)
            significant = p < alpha_adj
            if significant:
                level_any_significant = True
            results[name] = {
                "tested": True,
                "p_value": p,
                "alpha_used": alpha_adj,
                "significant": significant,
            }
 
        gate_open = level_any_significant  # disjunctive gatekeeping
 
    return results
 
 
# ─────────────────────────────────────────────────────────────
# 사용 예시: e-commerce 카나리, 25% 체크포인트 데이터 (각 500명)
# ─────────────────────────────────────────────────────────────
raw_p_values = {
    # 1차: 구매 전환율 (카나리 27/500 vs 베이스라인 25/500)
    "purchase_rate": compute_proportion_pvalue(27, 500, 25, 500),
    # 2차: 세션 시간, 클릭률 (실제로는 t-test 등 적합한 방법 사용)
    "session_time": 0.08,
    "click_rate":   0.03,
    # 3차: 페이지 로드, API 에러율
    "page_load_ms": 0.12,
    "error_rate":   0.04,
}
 
hierarchy = [
    ["purchase_rate"],               # 1차: 가장 중요
    ["session_time", "click_rate"],  # 2차: 1차 통과 후
    ["page_load_ms", "error_rate"],  # 3차: 2차 통과 후
]
 
gate_results = hierarchical_gate(raw_p_values, hierarchy, alpha=0.05)
for name, r in gate_results.items():
    if r["tested"]:
        sig = "★ 유의" if r["significant"] else "  비유의"
        print(f"{sig} | {name}: p={r['p_value']:.4f} (α_adj={r['alpha_used']:.4f})")
    else:
        print(f"  미검정 | {name}: {r['reason']}")

# argo-rollout.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: api-server-canary
spec:
  strategy:
    canary:
      steps:
        - setWeight: 10       # 1차 체크포인트: 10% 트래픽
        - pause: {duration: 5m}
        - analysis:
            templates:
              - templateName: sequential-boundary-check
            args:
              - name: checkpoint
                value: "0.10"
              - name: futility_lower   # 예시 1 참조: 10% 하한 = -0.48
                value: "-0.48"
        - setWeight: 25       # 2차 체크포인트: 25% 트래픽
        - pause: {duration: 10m}
        - analysis:
            templates:
              - templateName: sequential-boundary-check
            args:
              - name: checkpoint
                value: "0.25"
              - name: futility_lower   # 예시 1 참조: 25% 하한 = 0.94
                value: "0.94"
        - setWeight: 50
        - pause: {duration: 20m}
---
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: sequential-boundary-check
spec:
  args:
    - name: checkpoint
    - name: futility_lower
  metrics:
    - name: z-stat-purchase-rate
      provider:
        prometheus:
          address: http://prometheus:9090
          # Prometheus recording rule에서 카나리 vs 베이스라인 z-통계량을 미리 계산
          query: |
            canary_z_statistic{
              metric="purchase_rate",
              checkpoint="{{args.checkpoint}}"
            }
      # 무용성 하한(예시 1 gsDesign 출력값)을 초과해야 "계속 진행" 성공
      successCondition: "result[0] > {{args.futility_lower}}"
      failureCondition: "result[0] <= {{args.futility_lower}}"