P7-2.3 비교 실험 연습¶

# 구독 이탈 stress test에서 baseline, raw 1-NN, 부분 스케일 조정, z-score 1-NN을 비교해 실패 원인을 진단하는 예제입니다.
import csv
import numpy as np
from pathlib import Path

train_path = Path("docs/assets/part-07/chapter-02/p7-2-churn-dataset.csv")
stress_path = Path("docs/assets/part-07/chapter-02/p7-2-stress-test.csv")

train_rows = list(csv.DictReader(train_path.open(encoding="utf-8")))
stress_rows = list(csv.DictReader(stress_path.open(encoding="utf-8")))

for row in train_rows:
    row["unresolved_tickets"] = int(row["unresolved_tickets"])
    row["days_since_login"] = int(row["days_since_login"])
    row["usage_minutes_30d"] = int(row["usage_minutes_30d"])
    row["label"] = int(row["label"])

for row in stress_rows:
    row["unresolved_tickets"] = int(row["unresolved_tickets"])
    row["days_since_login"] = int(row["days_since_login"])
    row["usage_minutes_30d"] = int(row["usage_minutes_30d"])
    row["label"] = int(row["label"])

train_only = [row for row in train_rows if row["split"] == "train"]

def to_matrix(rows):
    return np.array([
        [row["unresolved_tickets"], row["days_since_login"], row["usage_minutes_30d"]]
        for row in rows
    ], dtype=float)

X_train = to_matrix(train_only)
y_train = np.array([row["label"] for row in train_only])
X_test = to_matrix(stress_rows)
y_test = np.array([row["label"] for row in stress_rows])

def predict_1nn(train_x, train_y, test_x):
    predictions = []
    nearest_ids = []
    for x in test_x:
        distances = np.linalg.norm(train_x - x, axis=1)
        nearest_index = int(np.argmin(distances))
        predictions.append(int(train_y[nearest_index]))
        nearest_ids.append(train_only[nearest_index]["sample_id"])
    return np.array(predictions), nearest_ids

baseline_class = int(np.bincount(y_train).argmax())
baseline_pred = np.full(len(y_test), baseline_class, dtype=int)

raw_pred, raw_nearest = predict_1nn(X_train, y_train, X_test)

X_train_scaled = X_train.copy()
X_test_scaled = X_test.copy()
X_train_scaled[:, 2] = X_train_scaled[:, 2] / 60.0
X_test_scaled[:, 2] = X_test_scaled[:, 2] / 60.0
scaled_pred, scaled_nearest = predict_1nn(X_train_scaled, y_train, X_test_scaled)

train_mean = X_train.mean(axis=0)
train_std = X_train.std(axis=0)
if np.any(train_std == 0):
    raise ValueError("표준편차가 0인 특징이 있어 z-score 정규화를 할 수 없습니다.")
X_train_z = (X_train - train_mean) / train_std
X_test_z = (X_test - train_mean) / train_std
z_pred, z_nearest = predict_1nn(X_train_z, y_train, X_test_z)

comparison_rows = []
for i, row in enumerate(stress_rows):
    model_errors = {
        "baseline": int(baseline_pred[i] != y_test[i]),
        "raw_1nn": int(raw_pred[i] != y_test[i]),
        "scaled_1nn": int(scaled_pred[i] != y_test[i]),
        "zscore_1nn": int(z_pred[i] != y_test[i]),
    }

    if model_errors["raw_1nn"] and not model_errors["zscore_1nn"]:
        diagnosis = "전처리로 해결됨"
    elif model_errors["zscore_1nn"]:
        diagnosis = "정규화 후에도 남는 경계 사례"
    elif len({int(raw_pred[i]), int(scaled_pred[i]), int(z_pred[i])}) > 1:
        diagnosis = "설정에 따라 갈리는 경계 사례"
    else:
        diagnosis = "현재 비교 실험에서는 안정적"

    comparison_rows.append({
        "샘플": row["sample_id"],
        "focus": row["focus"],
        "정답": row["label"],
        "baseline": int(baseline_pred[i]),
        "raw_1nn": int(raw_pred[i]),
        "scaled_1nn": int(scaled_pred[i]),
        "zscore_1nn": int(z_pred[i]),
        "raw_nearest": raw_nearest[i],
        "z_nearest": z_nearest[i],
        "failure_diagnosis": diagnosis,
    })

summary = {
    "baseline 정확도": round(float((baseline_pred == y_test).mean()), 3),
    "raw 1-NN 정확도": round(float((raw_pred == y_test).mean()), 3),
    "부분 스케일 조정 1-NN 정확도": round(float((scaled_pred == y_test).mean()), 3),
    "z-score 1-NN 정확도": round(float((z_pred == y_test).mean()), 3),
    "전처리로 해결된 샘플": [
        row["샘플"] for row in comparison_rows if row["failure_diagnosis"] == "전처리로 해결됨"
    ],
    "정규화 후에도 남는 샘플": [
        row["샘플"] for row in comparison_rows if row["failure_diagnosis"] == "정규화 후에도 남는 경계 사례"
    ],
}

representative_ids = {"stress-01", "stress-02", "stress-03", "stress-04"}
representative_rows = [
    row for row in comparison_rows if row["샘플"] in representative_ids
]

print("비교 실험 요약 =", summary)
print("학습 파일 =", str(train_path))
print("스트레스 평가 파일 =", str(stress_path))
print("대표 샘플별 비교 =")
for row in representative_rows:
    print(row)

비교 실험 요약 = {'baseline 정확도': 0.361, 'raw 1-NN 정확도': 0.861, '부분 스케일 조정 1-NN 정확도': 0.722, 'z-score 1-NN 정확도': 0.833, '전처리로 해결된 샘플': ['stress-01', 'stress-05', 'stress-13', 'stress-21', 'stress-29'], '정규화 후에도 남는 샘플': ['stress-02', 'stress-03', 'stress-06', 'stress-14', 'stress-22', 'stress-30']}
학습 파일 = docs/assets/part-07/chapter-02/p7-2-churn-dataset.csv
스트레스 평가 파일 = docs/assets/part-07/chapter-02/p7-2-stress-test.csv
대표 샘플별 비교 =
{'샘플': 'stress-01', 'focus': 'raw 거리에서는 retained 고객과 가까워 보이지만 전처리 후에는 churn 신호가 살아나는 사례', '정답': 1, 'baseline': 0, 'raw_1nn': 0, 'scaled_1nn': 1, 'zscore_1nn': 1, 'raw_nearest': '학습-02', 'z_nearest': '학습-08', 'failure_diagnosis': '전처리로 해결됨'}
{'샘플': 'stress-02', 'focus': '현재 학습 데이터 경계가 비어 있어 전처리만으로는 해결되지 않는 경계 사례', '정답': 1, 'baseline': 0, 'raw_1nn': 1, 'scaled_1nn': 0, 'zscore_1nn': 0, 'raw_nearest': '학습-11', 'z_nearest': '학습-04', 'failure_diagnosis': '정규화 후에도 남는 경계 사례'}
{'샘플': 'stress-03', 'focus': '질문 수와 미접속 일수는 높지만 실제로는 유지 고객인 애매한 retained 사례', '정답': 0, 'baseline': 0, 'raw_1nn': 0, 'scaled_1nn': 1, 'zscore_1nn': 1, 'raw_nearest': '학습-05', 'z_nearest': '학습-11', 'failure_diagnosis': '정규화 후에도 남는 경계 사례'}
{'샘플': 'stress-04', 'focus': '대부분의 비교 실험에서 일관되게 churn으로 잡혀야 하는 명확한 사례', '정답': 1, 'baseline': 0, 'raw_1nn': 1, 'scaled_1nn': 1, 'zscore_1nn': 1, 'raw_nearest': '학습-09', 'z_nearest': '학습-09', 'failure_diagnosis': '현재 비교 실험에서는 안정적'}

# action event 요약에서 원시 tracking 오류, 구간 특징, 기준선 gap flag를 비교해 반복 drift 후보를 찾는 예제입니다.
import csv
from pathlib import Path

summary_path = Path("docs/assets/part-07/chapter-01/p7-action-unit-summary.csv")
rows = list(csv.DictReader(summary_path.open(encoding="utf-8")))

for row in rows:
    row["event_order"] = int(row["event_order"])
    row["mid_flow_mean"] = float(row["mid_flow_mean"])
    row["late_drop_rate"] = float(row["late_drop_rate"])
    row["tracking_error_mean"] = float(row["tracking_error_mean"])

baseline_rows = [row for row in rows if row["period"] == "baseline"]
recent_rows = [row for row in rows if row["period"] == "recent"]
baseline_mid = sum(row["mid_flow_mean"] for row in baseline_rows) / len(baseline_rows)
baseline_late = sum(row["late_drop_rate"] for row in baseline_rows) / len(baseline_rows)

comparison = []
for row in recent_rows:
    raw_flag = row["tracking_error_mean"] > 0.12
    segment_flag = row["mid_flow_mean"] < 2.10 or row["late_drop_rate"] > 0.15
    baseline_gap_flag = (
        row["mid_flow_mean"] - baseline_mid < -0.20
        or row["late_drop_rate"] - baseline_late > 0.05
    )

    if segment_flag and not raw_flag:
        diagnosis = "원시 오류만 보면 놓치지만 구간 특징으로 잡힘"
    elif raw_flag and not segment_flag:
        diagnosis = "단발 튐 가능성이 있어 재현 확인 필요"
    elif segment_flag and baseline_gap_flag:
        diagnosis = "반복 drift 후보"
    else:
        diagnosis = "현재 비교에서는 낮은 우선순위"

    comparison.append({
        "event_id": row["event_id"],
        "raw_tracking_flag": raw_flag,
        "segment_feature_flag": segment_flag,
        "baseline_gap_flag": baseline_gap_flag,
        "diagnosis": diagnosis,
    })

for row in comparison:
    print(row)

{'event_id': 'E009', 'raw_tracking_flag': False, 'segment_feature_flag': True, 'baseline_gap_flag': True, 'diagnosis': '원시 오류만 보면 놓치지만 구간 특징으로 잡힘'}
{'event_id': 'E010', 'raw_tracking_flag': True, 'segment_feature_flag': False, 'baseline_gap_flag': False, 'diagnosis': '단발 튐 가능성이 있어 재현 확인 필요'}
{'event_id': 'E011', 'raw_tracking_flag': False, 'segment_feature_flag': True, 'baseline_gap_flag': True, 'diagnosis': '원시 오류만 보면 놓치지만 구간 특징으로 잡힘'}
{'event_id': 'E012', 'raw_tracking_flag': False, 'segment_feature_flag': True, 'baseline_gap_flag': True, 'diagnosis': '원시 오류만 보면 놓치지만 구간 특징으로 잡힘'}