9장: 합성 데이터 평가와 벤치마킹

import numpy as np
from sklearn.model_selection import StratifiedKFold
from sklearn.metrics import (
    accuracy_score,
    f1_score,
    precision_score,
    recall_score,
    roc_auc_score,
)
from dataclasses import dataclass
 
 
@dataclass
class EvaluationResult:
    experiment: str  # TRTR, TSTR, TSTS, TRTS
    model_name: str
    accuracy: float
    f1_macro: float
    precision_macro: float
    recall_macro: float
    auc_roc: float | None = None
 
 
def run_tstr_suite(
    real_train_X: np.ndarray,
    real_train_y: np.ndarray,
    real_test_X: np.ndarray,
    real_test_y: np.ndarray,
    synthetic_X: np.ndarray,
    synthetic_y: np.ndarray,
    models: dict,
) -> list[EvaluationResult]:
    """TSTR 평가 스위트를 실행합니다."""
    results = []
 
    experiments = {
        "TRTR": (real_train_X, real_train_y, real_test_X, real_test_y),
        "TSTR": (synthetic_X, synthetic_y, real_test_X, real_test_y),
        "TSTS": (synthetic_X, synthetic_y, synthetic_X, synthetic_y),
    }
 
    for exp_name, (train_X, train_y, test_X, test_y) in experiments.items():
        for model_name, model_class in models.items():
            model = model_class()
            model.fit(train_X, train_y)
            predictions = model.predict(test_X)
 
            result = EvaluationResult(
                experiment=exp_name,
                model_name=model_name,
                accuracy=accuracy_score(test_y, predictions),
                f1_macro=f1_score(
                    test_y, predictions, average="macro"
                ),
                precision_macro=precision_score(
                    test_y, predictions, average="macro"
                ),
                recall_macro=recall_score(
                    test_y, predictions, average="macro"
                ),
            )
            results.append(result)
 
    return results
 
 
def compute_tstr_ratio(results: list[EvaluationResult]) -> dict:
    """TSTR/TRTR 비율을 계산합니다."""
    ratios = {}
 
    trtr_results = {
        r.model_name: r for r in results if r.experiment == "TRTR"
    }
    tstr_results = {
        r.model_name: r for r in results if r.experiment == "TSTR"
    }
 
    for model_name in trtr_results:
        if model_name in tstr_results:
            trtr = trtr_results[model_name]
            tstr = tstr_results[model_name]
            ratios[model_name] = {
                "accuracy_ratio": (
                    tstr.accuracy / trtr.accuracy
                    if trtr.accuracy > 0 else 0
                ),
                "f1_ratio": (
                    tstr.f1_macro / trtr.f1_macro
                    if trtr.f1_macro > 0 else 0
                ),
            }
 
    return ratios

from collections import Counter
import numpy as np
 
 
def type_token_ratio(texts: list[str]) -> float:
    """TTR(Type-Token Ratio)을 계산합니다."""
    all_tokens = []
    for text in texts:
        all_tokens.extend(text.split())
 
    if not all_tokens:
        return 0.0
 
    unique_tokens = set(all_tokens)
    return len(unique_tokens) / len(all_tokens)
 
 
def vocabulary_size(texts: list[str]) -> int:
    """고유 어휘 수를 계산합니다."""
    all_tokens = set()
    for text in texts:
        all_tokens.update(text.split())
    return len(all_tokens)
 
 
def hapax_legomena_ratio(texts: list[str]) -> float:
    """한 번만 등장하는 단어의 비율을 계산합니다."""
    all_tokens = []
    for text in texts:
        all_tokens.extend(text.split())
 
    counter = Counter(all_tokens)
    hapax = sum(1 for count in counter.values() if count == 1)
 
    return hapax / len(counter) if counter else 0.0

import numpy as np
from sklearn.metrics.pairwise import cosine_similarity
 
 
def semantic_diversity_score(embeddings: np.ndarray) -> dict:
    """임베딩 기반 의미적 다양성을 측정합니다."""
    # 쌍별 코사인 유사도 행렬
    sim_matrix = cosine_similarity(embeddings)
 
    # 대각선(자기 자신) 제외
    n = len(embeddings)
    mask = ~np.eye(n, dtype=bool)
    pairwise_sims = sim_matrix[mask]
 
    return {
        "mean_pairwise_similarity": pairwise_sims.mean(),
        "diversity_score": 1 - pairwise_sims.mean(),
        "min_similarity": pairwise_sims.min(),
        "max_similarity": pairwise_sims.max(),
        "std_similarity": pairwise_sims.std(),
    }
 
 
def coverage_score(
    synthetic_embeddings: np.ndarray,
    real_embeddings: np.ndarray,
    threshold: float = 0.8,
) -> float:
    """합성 데이터가 실제 데이터 분포를 얼마나 커버하는지 측정합니다."""
    sim_matrix = cosine_similarity(
        real_embeddings, synthetic_embeddings
    )
 
    # 각 실제 데이터 포인트에 대해 가장 유사한 합성 데이터와의 유사도
    max_sims = sim_matrix.max(axis=1)
 
    # 임계값 이상의 유사도를 가진 실제 데이터의 비율
    covered = (max_sims >= threshold).mean()
 
    return covered

import numpy as np
from collections import Counter
 
 
def structural_diversity_metrics(texts: list[str]) -> dict:
    """구조적 다양성 메트릭을 계산합니다."""
    lengths = [len(t) for t in texts]
    sentence_counts = [t.count(".") + t.count("!") + t.count("?") for t in texts]
 
    # 길이 분포 엔트로피
    length_bins = np.histogram(lengths, bins=20)[0]
    length_probs = length_bins / length_bins.sum()
    length_probs = length_probs[length_probs > 0]
    length_entropy = -np.sum(length_probs * np.log2(length_probs))
 
    # 코드 블록 포함 비율
    code_ratio = sum(1 for t in texts if "```" in t) / len(texts)
 
    # 목록 포함 비율
    list_ratio = sum(
        1 for t in texts if "\n-" in t or "\n1." in t
    ) / len(texts)
 
    return {
        "length_mean": np.mean(lengths),
        "length_std": np.std(lengths),
        "length_entropy": length_entropy,
        "sentence_count_mean": np.mean(sentence_counts),
        "code_block_ratio": code_ratio,
        "list_ratio": list_ratio,
    }

import numpy as np
from scipy import stats
from scipy.spatial.distance import jensenshannon
 
 
def comprehensive_distribution_comparison(
    real_data: np.ndarray,
    synthetic_data: np.ndarray,
) -> dict:
    """포괄적인 분포 비교를 수행합니다."""
    results = {}
 
    # 1. KS 검정 (Kolmogorov-Smirnov)
    ks_stat, ks_p = stats.ks_2samp(
        real_data.flatten(), synthetic_data.flatten()
    )
    results["ks_test"] = {"statistic": ks_stat, "p_value": ks_p}
 
    # 2. Jensen-Shannon 발산
    bins = np.linspace(
        min(real_data.min(), synthetic_data.min()),
        max(real_data.max(), synthetic_data.max()),
        100,
    )
    real_hist, _ = np.histogram(real_data, bins=bins, density=True)
    synth_hist, _ = np.histogram(
        synthetic_data, bins=bins, density=True
    )
    real_hist += 1e-10
    synth_hist += 1e-10
 
    js = jensenshannon(real_hist, synth_hist)
    results["js_divergence"] = js
 
    # 3. Wasserstein 거리
    w_dist = stats.wasserstein_distance(
        real_data.flatten(), synthetic_data.flatten()
    )
    results["wasserstein_distance"] = w_dist
 
    # 4. 모멘트 비교
    for i, name in enumerate(["평균", "분산", "왜도", "첨도"], 1):
        real_moment = stats.moment(real_data.flatten(), moment=i)
        synth_moment = stats.moment(
            synthetic_data.flatten(), moment=i
        )
        results[f"moment_{i}_{name}_diff"] = abs(
            real_moment - synth_moment
        )
 
    return results

import matplotlib.pyplot as plt
import numpy as np
 
 
def plot_distribution_comparison(
    real: np.ndarray,
    synthetic: np.ndarray,
    feature_name: str,
    output_path: str,
) -> None:
    """실제/합성 데이터의 분포를 시각적으로 비교합니다."""
    fig, axes = plt.subplots(1, 3, figsize=(15, 5))
 
    # 히스토그램 비교
    axes[0].hist(real, bins=50, alpha=0.5, label="Real", density=True)
    axes[0].hist(
        synthetic, bins=50, alpha=0.5, label="Synthetic", density=True
    )
    axes[0].set_title(f"{feature_name} - Histogram")
    axes[0].legend()
 
    # CDF 비교
    real_sorted = np.sort(real)
    synth_sorted = np.sort(synthetic)
    axes[1].plot(
        real_sorted,
        np.linspace(0, 1, len(real_sorted)),
        label="Real",
    )
    axes[1].plot(
        synth_sorted,
        np.linspace(0, 1, len(synth_sorted)),
        label="Synthetic",
    )
    axes[1].set_title(f"{feature_name} - CDF")
    axes[1].legend()
 
    # Q-Q Plot
    quantiles = np.linspace(0.01, 0.99, 100)
    real_q = np.quantile(real, quantiles)
    synth_q = np.quantile(synthetic, quantiles)
    axes[2].scatter(real_q, synth_q, alpha=0.5, s=10)
    lim = [
        min(real_q.min(), synth_q.min()),
        max(real_q.max(), synth_q.max()),
    ]
    axes[2].plot(lim, lim, "r--", label="y=x")
    axes[2].set_title(f"{feature_name} - Q-Q Plot")
    axes[2].set_xlabel("Real")
    axes[2].set_ylabel("Synthetic")
    axes[2].legend()
 
    plt.tight_layout()
    plt.savefig(output_path, dpi=150, bbox_inches="tight")
    plt.close()

import numpy as np
from math import comb
 
 
def pass_at_k(
    n: int, c: int, k: int
) -> float:
    """pass@k 메트릭을 계산합니다.
 
    Args:
        n: 총 생성 수
        c: 테스트를 통과한 수
        k: 시도 횟수
    """
    if n - c < k:
        return 1.0
    return 1.0 - comb(n - c, k) / comb(n, k)
 
 
def evaluate_code_generation(
    problems: list[dict],
    generation_fn,
    n_samples: int = 20,
    k_values: list[int] = [1, 5, 10],
) -> dict:
    """코드 생성 모델을 pass@k로 평가합니다."""
    results = {f"pass@{k}": [] for k in k_values}
 
    for problem in problems:
        # n개의 코드 샘플 생성
        samples = [
            generation_fn(problem["prompt"])
            for _ in range(n_samples)
        ]
 
        # 각 샘플 테스트
        passed = sum(
            1 for s in samples
            if run_test(s, problem["test_cases"])
        )
 
        # pass@k 계산
        for k in k_values:
            score = pass_at_k(n_samples, passed, k)
            results[f"pass@{k}"].append(score)
 
    # 평균
    return {
        metric: np.mean(scores)
        for metric, scores in results.items()
    }
 
 
def run_test(code: str, test_cases: list[str]) -> bool:
    """코드를 실행하여 테스트를 검증합니다."""
    import subprocess
    full_code = code + "\n" + "\n".join(test_cases)
    try:
        result = subprocess.run(
            ["python", "-c", full_code],
            capture_output=True,
            timeout=10,
        )
        return result.returncode == 0
    except subprocess.TimeoutExpired:
        return False

from dataclasses import dataclass
 
 
@dataclass
class ExperimentConfig:
    name: str
    description: str
    real_data_sizes: list[int]  # [100, 500, 1000, 5000]
    synthetic_ratios: list[float]  # [0, 0.25, 0.5, 0.75, 1.0]
    models: list[str]
    metrics: list[str]
    num_seeds: int = 5  # 랜덤 시드 수
 
 
EXPERIMENT_MATRIX = [
    # 실험 1: 합성 데이터만으로 충분한가?
    {
        "name": "synthetic_only",
        "conditions": [
            ("100% 실제", 1.0, 0.0),
            ("100% 합성", 0.0, 1.0),
        ],
    },
    # 실험 2: 최적 혼합 비율은?
    {
        "name": "optimal_mix",
        "conditions": [
            ("100% 실제", 1.0, 0.0),
            ("75% 실제 + 25% 합성", 0.75, 0.25),
            ("50% 실제 + 50% 합성", 0.50, 0.50),
            ("25% 실제 + 75% 합성", 0.25, 0.75),
            ("100% 합성", 0.0, 1.0),
        ],
    },
    # 실험 3: 실제 데이터 양에 따른 합성 데이터의 한계 효용
    {
        "name": "marginal_utility",
        "real_sizes": [50, 100, 500, 1000, 5000],
        "synthetic_added": [0, 1000, 5000, 10000],
    },
]

import numpy as np
from scipy import stats
 
 
def analyze_experiment_results(
    results: list[dict],
) -> dict:
    """실험 결과를 분석합니다."""
    analysis = {}
 
    # 1. 합성 데이터만의 성능 (TSTR/TRTR 비율)
    # 2. 최적 혼합 비율 결정
    # 3. 한계 효용 분석
    # 4. 통계적 유의성 검정
 
    # 예: 혼합 비율별 성능 비교에서 최적점 찾기
    mix_results = [
        r for r in results if r["experiment"] == "optimal_mix"
    ]
 
    if mix_results:
        best = max(mix_results, key=lambda r: r["f1_score"])
        analysis["optimal_mix"] = {
            "real_ratio": best["real_ratio"],
            "synthetic_ratio": best["synthetic_ratio"],
            "f1_score": best["f1_score"],
        }
 
    return analysis

from dataclasses import dataclass
 
 
@dataclass
class SyntheticDataBenchmark:
    name: str
    version: str
 
    # 평가 데이터셋
    real_test_sets: list[dict]  # 실제 테스트 데이터
 
    # 평가 차원
    fidelity_metrics: list[str]    # 충실도
    utility_metrics: list[str]     # 유용성
    diversity_metrics: list[str]   # 다양성
    privacy_metrics: list[str]     # 프라이버시
 
    # 다운스트림 태스크
    downstream_tasks: list[dict]
 
    # 기준선
    baselines: dict
 
 
BENCHMARK_SPEC = SyntheticDataBenchmark(
    name="SynthEval-Ko",
    version="1.0",
    real_test_sets=[
        {"name": "KLUE-TC", "task": "분류", "size": 5000},
        {"name": "KLUE-NER", "task": "NER", "size": 3000},
        {"name": "KorSTS", "task": "유사도", "size": 1500},
    ],
    fidelity_metrics=[
        "js_divergence",
        "wasserstein_distance",
        "correlation_preservation",
    ],
    utility_metrics=[
        "tstr_accuracy",
        "tstr_f1",
        "tstr_trtr_ratio",
    ],
    diversity_metrics=[
        "type_token_ratio",
        "semantic_diversity",
        "length_entropy",
    ],
    privacy_metrics=[
        "mia_auc",
        "dcr_distance",
        "pii_leak_rate",
    ],
    downstream_tasks=[
        {"name": "텍스트 분류", "metric": "F1"},
        {"name": "개체명 인식", "metric": "Entity F1"},
        {"name": "텍스트 유사도", "metric": "Spearman"},
    ],
    baselines={
        "real_data_100pct": "실제 데이터 100%로 학습",
        "real_data_10pct": "실제 데이터 10%로 학습",
        "random_baseline": "랜덤 분류기",
    },
)