D ビジネス/PM — 販促クーポンの「効果報告書」を増分利益で書き直す（利用者 vs 非利用者比較の自己選択バイアス × 眠っていた10%ホールドアウトでITT比較 × 売上ではなく粗利−値引き−配信費 × 前倒し購入を拾う60日窓 × 信頼区間とSRM検査 × 事前登録の判断ルールと次回3群実験の検出力設計）

📋 コピー"""誕生日クーポン施策の増分利益を、ホールドアウト群との比較（ITT）で推定する。

入力は BigQuery の集計結果（群ごとの会員数・1会員あたり売上の平均と分散）。
「クーポンを使った人」ではなく「配信対象に割り付けられた全員」を比べるので、
自己選択バイアス（もともと買う人ほど使う）が入らない。
"""

from __future__ import annotations

import math
from dataclasses import dataclass
from enum import StrEnum

# ③ 粗利率（値引き前売上に対する比率）。経理の月次粗利率実績を使う（本問では仮定値）
GROSS_MARGIN_RATE = 0.35
# クーポン額面。値引きは全額が当社負担（メーカー協賛なし）
COUPON_FACE_VALUE_YEN = 1_000
# 配信1通あたりの費用（メール配信 SaaS の従量単価。仮定値）
SEND_COST_PER_MEMBER_YEN = 1.0
# ⑥ 95% 信頼区間の係数（正規近似。各群 6 万人以上あるので妥当）
Z_95 = 1.96
# 配信基盤が member_id ハッシュ下1桁=0 を配信対象外にしている比率
EXPECTED_HOLDOUT_RATIO = 0.10
# ⑥ SRM（割付比の崩れ）検査の閾値。自由度1のカイ二乗で p=0.001 に相当
SRM_CHI2_THRESHOLD = 10.83


class Decision(StrEnum):
    """事前に決めた判断ルールの出力。"""

    CONTINUE = "継続（95%CI の下限が 0 円超）"
    STOP = "現行設計は停止（95%CI の上限が 0 円未満）"
    INCONCLUSIVE = "判断保留（0 円を含む。検出力を上げて再測定）"


@dataclass(frozen=True, slots=True)
class ArmStats:
    """1群分の集計。購入しなかった会員も売上 0 円として含む。

    Attributes:
        members: 群に割り付けられた会員数。
        mean_gmv_yen: 観測期間内の 1 会員あたり値引き前売上の平均。
        var_gmv: 同じく 1 会員あたり売上の標本分散（円^2）。
    """

    members: int
    mean_gmv_yen: float
    var_gmv: float


@dataclass(frozen=True, slots=True)
class Estimate:
    """点推定と 95% 信頼区間（円／配信対象1人）。"""

    point: float
    low: float
    high: float


def srm_chi2(treated: int, holdout: int, expected_holdout_ratio: float) -> float:
    """割付比が設計通りかを確かめるカイ二乗統計量を返す。

    これが閾値を超えたら、割付か集計のどこかが壊れている。効果の数字を読む前に止める。
    """
    total = treated + holdout
    exp_h = total * expected_holdout_ratio
    exp_t = total - exp_h
    return (holdout - exp_h) ** 2 / exp_h + (treated - exp_t) ** 2 / exp_t


def incremental_profit_per_member(
    treated: ArmStats, holdout: ArmStats, redeemed: int
) -> Estimate:
    """配信対象1人あたりの増分利益（粗利 − 値引き − 配信費）を推定する。

    値引きは「使った全員分」を差し引く。クーポンが無くても買った人への値引きも
    ここで費用として効く（Bad の報告ではこれが「効果」に数えられていた）。

    Args:
        treated: 配信群の集計。
        holdout: ホールドアウト群の集計。
        redeemed: 配信群でクーポンを使った会員数。

    Returns:
        増分利益の点推定と 95% 信頼区間。値引き額と配信費は確定値として扱う。
    """
    lift_gmv = treated.mean_gmv_yen - holdout.mean_gmv_yen  # ①② 割付単位の差
    se_gmv = math.sqrt(treated.var_gmv / treated.members + holdout.var_gmv / holdout.members)
    discount_per_member = redeemed * COUPON_FACE_VALUE_YEN / treated.members  # ④
    point = GROSS_MARGIN_RATE * lift_gmv - discount_per_member - SEND_COST_PER_MEMBER_YEN  # ③
    half = Z_95 * GROSS_MARGIN_RATE * se_gmv
    return Estimate(point=point, low=point - half, high=point + half)


def decide(est: Estimate) -> Decision:
    """事前登録した判断ルール。結果を見てからルールを変えない。"""
    # ⑦ 継続・停止・保留の条件を先に決めておく
    if est.low > 0:
        return Decision.CONTINUE
    if est.high < 0:
        return Decision.STOP
    return Decision.INCONCLUSIVE


if __name__ == "__main__":
    REDEEMED = 47_530
    chi2 = srm_chi2(594_120, 65_880, EXPECTED_HOLDOUT_RATIO)
    print(f"SRM chi2 = {chi2:.2f} (閾値 {SRM_CHI2_THRESHOLD}) -> {'OK' if chi2 < SRM_CHI2_THRESHOLD else 'STOP'}")
    windows = {  # ⑤ 30日と60日を並べ、符号が変わるかを見る
        "30日": (ArmStats(594_120, 1_372.0, 20_523_216), ArmStats(65_880, 1_140.0, 17_210_400)),
        "60日": (ArmStats(594_120, 2_422.0, 36_640_716), ArmStats(65_880, 2_248.8, 34_126_083)),
    }
    for label, (t, h) in windows.items():
        est = incremental_profit_per_member(t, h, REDEEMED)
        total = [round(x * t.members / 10_000) for x in (est.point, est.low, est.high)]
        print(
            f"[{label}] 売上リフト {t.mean_gmv_yen - h.mean_gmv_yen:+.1f} 円/人 | "
            f"増分利益 {est.point:+.1f} 円/人 (95%CI {est.low:+.1f} 〜 {est.high:+.1f}) | "
            f"総額 {total[0]:+,} 万円 ({total[1]:+,} 〜 {total[2]:+,}) | {decide(est)}"
        )