P7-1.3 기준선 재설계 연습¶

# 웹 트래픽 기준선을 전체 합계, 채널-일자, 최근 집중 기간으로 바꿔 우선 검토 항목과 다음 질문을 재설계하는 예제입니다.
import csv
from collections import defaultdict
from datetime import datetime
from pathlib import Path

data_path = Path("docs/assets/part-07/chapter-01/p7-1-traffic-log.csv")
rows = list(csv.DictReader(data_path.open(encoding="utf-8")))

for row in rows:
    row["date"] = datetime.strptime(row["date"], "%Y-%m-%d").date()
    row["visitors"] = int(row["visitors"])
    row["signups"] = int(row["signups"])
    row["errors"] = int(row["errors"])

def summarize(group_rows):
    visitors = sum(row["visitors"] for row in group_rows)
    signups = sum(row["signups"] for row in group_rows)
    errors = sum(row["errors"] for row in group_rows)
    if visitors == 0:
        raise ValueError("비교 구간에 visitors가 없습니다. 기준선 경계일을 확인하세요.")
    return {
        "visitors": visitors,
        "signups": signups,
        "conversion_rate": round(signups / visitors, 4),
        "error_rate": round(errors / visitors, 4),
    }

def aggregate_by_day(group_rows):
    grouped = defaultdict(lambda: {"visitors": 0, "signups": 0, "errors": 0})
    for row in group_rows:
        grouped[row["date"]]["visitors"] += row["visitors"]
        grouped[row["date"]]["signups"] += row["signups"]
        grouped[row["date"]]["errors"] += row["errors"]
    return [
        {"date": date, **values}
        for date, values in sorted(grouped.items())
    ]

experiments = [
    # 조작 변수: cutoff와 unit을 바꾸면 회고 앞줄에 올라오는 항목이 달라집니다.
    {"name": "전체 합계 / 7일 기준선", "cutoff": "2026-06-08", "unit": "date-total"},
    {"name": "채널-일자 / 7일 기준선", "cutoff": "2026-06-08", "unit": "channel-day"},
    {"name": "채널-일자 / 최근 4일 집중", "cutoff": "2026-06-11", "unit": "channel-day"},
]

results = []
for experiment in experiments:
    cutoff = datetime.strptime(experiment["cutoff"], "%Y-%m-%d").date()
    baseline_rows = [row for row in rows if row["date"] < cutoff]
    recent_rows = [row for row in rows if row["date"] >= cutoff]

    if experiment["unit"] == "date-total":
        baseline = summarize(aggregate_by_day(baseline_rows))
        recent = summarize(aggregate_by_day(recent_rows))
        priority = "전체 하락 여부 재확인"
        detail = {
            "conversion_delta": round(recent["conversion_rate"] - baseline["conversion_rate"], 4),
            "error_delta": round(recent["error_rate"] - baseline["error_rate"], 4),
        }
    else:
        by_channel = defaultdict(lambda: {"baseline": [], "recent": []})
        for row in rows:
            period = "recent" if row["date"] >= cutoff else "baseline"
            by_channel[row["channel"]][period].append(row)

        channel_deltas = []
        for channel, grouped in by_channel.items():
            baseline = summarize(grouped["baseline"])
            recent = summarize(grouped["recent"])
            channel_deltas.append({
                "channel": channel,
                "conversion_delta": round(recent["conversion_rate"] - baseline["conversion_rate"], 4),
                "error_delta": round(recent["error_rate"] - baseline["error_rate"], 4),
            })

        channel_deltas.sort(key=lambda row: row["conversion_delta"])
        detail = channel_deltas[0]
        priority = f"{detail['channel']} 채널 우선 검토"

    results.append({
        "실험": experiment["name"],
        "기준선 경계일": experiment["cutoff"],
        "단위": experiment["unit"],
        "우선 검토 항목": priority,
        "핵심 변화": detail,
    })

retrospective = []
for result in results:
    if result["단위"] == "date-total":
        retrospective.append({
            "실험": result["실험"],
            "사실": f"전체 합계 기준으로 보면 전환율 변화는 {result['핵심 변화']['conversion_delta']}이고 오류율 변화는 {result['핵심 변화']['error_delta']}이다.",
            "해석": "서비스 전체 흐름이 약하게 흔들렸는지는 볼 수 있지만, 어떤 채널이 원인인지는 아직 분해되지 않는다.",
            "다음 질문": "채널-일자 단위로 다시 쪼개면 어느 채널이 먼저 튀는가?",
        })
    else:
        retrospective.append({
            "실험": result["실험"],
            "사실": f"{result['핵심 변화']['channel']} 채널의 전환율 변화가 {result['핵심 변화']['conversion_delta']}로 가장 크게 내려갔다.",
            "해석": "현재 질문이 원인 후보 좁히기라면 채널 단위 기준선이 더 적합하다.",
            "다음 질문": "ads 안에서도 campaign, browser, release_version 같은 더 세분화된 축이 필요한가?",
        })

print("기준선 재설계 결과 =")
for row in results:
    print(row)
print("[회고 메모]")
for row in retrospective:
    print(row)

기준선 재설계 결과 =
{'실험': '전체 합계 / 7일 기준선', '기준선 경계일': '2026-06-08', '단위': 'date-total', '우선 검토 항목': '전체 하락 여부 재확인', '핵심 변화': {'conversion_delta': -0.0108, 'error_delta': 0.0032}}
{'실험': '채널-일자 / 7일 기준선', '기준선 경계일': '2026-06-08', '단위': 'channel-day', '우선 검토 항목': 'ads 채널 우선 검토', '핵심 변화': {'channel': 'ads', 'conversion_delta': -0.0361, 'error_delta': 0.0114}}
{'실험': '채널-일자 / 최근 4일 집중', '기준선 경계일': '2026-06-11', '단위': 'channel-day', '우선 검토 항목': 'ads 채널 우선 검토', '핵심 변화': {'channel': 'ads', 'conversion_delta': -0.0279, 'error_delta': 0.0091}}
[회고 메모]
{'실험': '전체 합계 / 7일 기준선', '사실': '전체 합계 기준으로 보면 전환율 변화는 -0.0108이고 오류율 변화는 0.0032이다.', '해석': '서비스 전체 흐름이 약하게 흔들렸는지는 볼 수 있지만, 어떤 채널이 원인인지는 아직 분해되지 않는다.', '다음 질문': '채널-일자 단위로 다시 쪼개면 어느 채널이 먼저 튀는가?'}
{'실험': '채널-일자 / 7일 기준선', '사실': 'ads 채널의 전환율 변화가 -0.0361로 가장 크게 내려갔다.', '해석': '현재 질문이 원인 후보 좁히기라면 채널 단위 기준선이 더 적합하다.', '다음 질문': 'ads 안에서도 campaign, browser, release_version 같은 더 세분화된 축이 필요한가?'}
{'실험': '채널-일자 / 최근 4일 집중', '사실': 'ads 채널의 전환율 변화가 -0.0279로 가장 크게 내려갔다.', '해석': '현재 질문이 원인 후보 좁히기라면 채널 단위 기준선이 더 적합하다.', '다음 질문': 'ads 안에서도 campaign, browser, release_version 같은 더 세분화된 축이 필요한가?'}

# action event 단위 요약에서 기준선 경계 순서를 바꿔 mid-flow와 late-drop gap이 어떻게 달라지는지 비교하는 예제입니다.
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"])

def summarize(group_rows):
    if not group_rows:
        raise ValueError("비교 구간이 비어 있습니다. cutoff_order를 확인하세요.")
    return {
        "count": len(group_rows),
        "mid_flow_mean": round(sum(row["mid_flow_mean"] for row in group_rows) / len(group_rows), 3),
        "late_drop_rate": round(sum(row["late_drop_rate"] for row in group_rows) / len(group_rows), 3),
        "review_ratio": round(
            sum(row["review_needed"] == "yes" for row in group_rows) / len(group_rows),
            3,
        ),
    }

comparison_rows = [row for row in rows if row["period"] in {"baseline", "recent"}]

# 조작 변수: 최근 구간을 어디서 시작할지 바꿔 봅니다.
for cutoff_order in [9, 10]:
    baseline = [row for row in comparison_rows if row["event_order"] < cutoff_order]
    recent = [row for row in comparison_rows if row["event_order"] >= cutoff_order]
    baseline_summary = summarize(baseline)
    recent_summary = summarize(recent)
    result = {
        "recent_start_order": cutoff_order,
        "baseline_count": baseline_summary["count"],
        "recent_count": recent_summary["count"],
        "mid_flow_gap": round(recent_summary["mid_flow_mean"] - baseline_summary["mid_flow_mean"], 3),
        "late_drop_gap": round(recent_summary["late_drop_rate"] - baseline_summary["late_drop_rate"], 3),
        "recent_review_ratio": recent_summary["review_ratio"],
    }
    print(result)

{'recent_start_order': 9, 'baseline_count': 8, 'recent_count': 4, 'mid_flow_gap': -0.219, 'late_drop_gap': 0.064, 'recent_review_ratio': 0.75}
{'recent_start_order': 10, 'baseline_count': 9, 'recent_count': 3, 'mid_flow_gap': -0.172, 'late_drop_gap': 0.051, 'recent_review_ratio': 0.667}