A コーディング — itertools.groupby × functools.cache × Generator × TypeAlias (PEP 695) × StrEnum × Decimal × frozen dataclass（キャンペーンイベント集計 Bad→Good）

📋 コピー"""event_aggregator.py
Ch2:  StrEnum / TypeAlias PEP 695 / Decimal
Ch4:  frozen dataclass CampaignSummary slots=True
Ch6:  match 文（イベント種別集計）
Ch7:  名前付き定数 ZERO / PERCENT / MONEY_PLACES
Ch10: ValueError（未知 EventType を即拒否）
Ch11: Generator / sorted+groupby / O(N log N)
"""
from __future__ import annotations
import functools, itertools
from dataclasses import dataclass
from decimal import ROUND_HALF_UP, Decimal
from enum import StrEnum
from typing import Final, Iterator

# ── 名前付き定数（Ch7）───────────────────────────
ZERO: Final[Decimal]         = Decimal("0")
PERCENT: Final[Decimal]      = Decimal("100")
MONEY_PLACES: Final[Decimal] = Decimal("1")   # 円単位
CVR_PLACES: Final[Decimal]   = Decimal("0.01")

# ── PEP 695 TypeAlias（Python 3.12+）────────────
type EventRow = dict[str, str | int | float]

# ── StrEnum（Ch2）───────────────────────────────
class EventType(StrEnum):
    CLICK   = "click"
    OPEN    = "open"
    CONVERT = "convert"

# ── 値オブジェクト（Ch4）───────────────────────
@dataclass(frozen=True, slots=True)    # ← slots で軽量化
class CampaignSummary:
    """キャンペーン別集計（イミュータブル値オブジェクト）"""
    campaign_id:   str
    click_count:   int
    open_count:    int
    convert_count: int
    total_revenue: Decimal  # 割引後・円単位 Decimal

    @property
    def cvr(self) -> Decimal:
        """CVR（%）= convert / click × 100"""
        if self.click_count == 0:
            return ZERO              # ゼロ除算ガード（Ch10）
        return (
            Decimal(self.convert_count)
            / Decimal(self.click_count) * PERCENT
        ).quantize(CVR_PLACES, rounding=ROUND_HALF_UP)

# ── メモ化（functools.cache）────────────────────
@functools.cache                       # str はhashable → cache可
def _calc_discount_rate(cid: str) -> Decimal:
    """campaign_id → 割引率（バッチ内で1回だけ計算）"""
    seed = int(cid[-2:]) if cid[-2:].isdigit() else 5
    return Decimal(seed) / PERCENT

# ── ジェネレータ（Ch11）────────────────────────
def _iter_events(
    rows: list[EventRow],
) -> Iterator[tuple[str, EventType, Decimal]]:
    """(campaign_id, EventType, revenue) を逐次 yield。

    中間リストを作らないためメモリ O(1) を維持する。
    """
    for row in rows:
        cid   = str(row["campaign_id"])
        etype = EventType(str(row["event_type"])) # 未知→ValueError(Ch10)
        rev   = Decimal(str(row.get("revenue","0"))).quantize(
            MONEY_PLACES, rounding=ROUND_HALF_UP
        )
        yield cid, etype, rev

# ── 集計メイン ──────────────────────────────────
def aggregate_events(
    rows: list[EventRow],
) -> list[CampaignSummary]:
    """注文イベントをキャンペーン別に集計する。

    sorted + groupby で O(N log N)、groupby は
    ソート後の連続キーをグループ化する。
    """
    if not rows:                      # 早期リターン（空ガード）
        return []

    # ① ジェネレータ → sorted（ソート済みをgroupbyへ）
    sorted_events = sorted(
        _iter_events(rows), key=lambda t: t[0]   # campaign_id 昇順
    )

    summaries: list[CampaignSummary] = []

    # ② groupby でキャンペーン別グループ化
    for cid, group in itertools.groupby(
        sorted_events, key=lambda t: t[0]
    ):
        clicks = opens = converts = 0
        total  = ZERO

        for _, event_type, revenue in group:
            match event_type:        # match 文（Ch6）
                case EventType.CLICK:
                    clicks   += 1
                case EventType.OPEN:
                    opens    += 1
                case EventType.CONVERT:
                    converts += 1
                    total    += revenue
                case _:
                    raise ValueError(f"未知 EventType: {event_type}")

        # ③ functools.cache でメモ化した割引率を適用
        rate    = _calc_discount_rate(cid)
        net_rev = (total * (Decimal("1") - rate)).quantize(
            MONEY_PLACES, rounding=ROUND_HALF_UP
        )

        summaries.append(
            CampaignSummary(
                campaign_id   = cid,
                click_count   = clicks,
                open_count    = opens,
                convert_count = converts,
                total_revenue = net_rev,   # 割引後売上（円）
            )
        )

    return summaries   # campaign_id 昇順（sorted 順序を維持）


# ── 実行例 ──────────────────────────────────────
if __name__ == "__main__":
    sample: list[EventRow] = [
        {"campaign_id":"CP05","event_type":"click",  "revenue":0},
        {"campaign_id":"CP05","event_type":"open",   "revenue":0},
        {"campaign_id":"CP05","event_type":"convert","revenue":9800},
        {"campaign_id":"CP05","event_type":"click",  "revenue":0},
        {"campaign_id":"CP10","event_type":"click",  "revenue":0},
        {"campaign_id":"CP10","event_type":"convert","revenue":5500},
    ]
    for s in aggregate_events(sample):
        print(
            f"{s.campaign_id}: "
            f"click={s.click_count}, open={s.open_count}, "
            f"convert={s.convert_count}, "
            f"revenue={s.total_revenue}円, cvr={s.cvr}%"
        )
    # CP05: click=2, open=1, convert=1, revenue=9310円, cvr=50.00%
    # CP10: click=1, open=0, convert=1, revenue=4950円, cvr=100.00%