広告運用レポートをClaude Codeで自動化する方法【2026】Google広告・Meta広告API×スクリプト×Skills（設計と実装）

ad-report/
├── CLAUDE.md                     # 指標定義と厳守ルール（毎回読み込まれる）
├── .claude/
│   ├── settings.json             # permissions と hooks
│   ├── skills/weekly-ad-report/SKILL.md   # 週次レポートの手順書
│   └── agents/ad-investigator.md # 掘り下げ調査用のサブエージェント
├── scripts/
│   ├── fetch_ads.py              # ① 取得＋② 正規化
│   ├── aggregate.py              # ③ 集計・前週比・異常値
│   ├── breakdown.py              # 掘り下げ用（読み取り専用）
│   ├── verify_report.py          # 数値の突き合わせ検証
│   ├── guard_bash.py             # 書き込み系コマンドのブロック
│   └── post_slack.py             # ⑥ Slack投稿
├── data/                         # 取得・集計結果（Git管理しない）
├── reports/                      # ④ Claude Codeが書く週次レポート
├── proposals/                    # 変更の「提案」置き場（実行は人）
├── run_weekly.sh                 # ヘッドレス実行の入口
├── google-ads.yaml / .env        # 認証情報（Git・Claudeの読み取り対象外）
└── .gitignore

# scripts/fetch_ads.py : 媒体APIから取得し「共通スキーマ」に揃えてCSV保存する
import json
import os

import pandas as pd
import requests
from google.ads.googleads.client import GoogleAdsClient

# 共通スキーマ（1行 = 媒体 × 日付 × キャンペーン）
COMMON_COLUMNS = [
    "date", "platform", "account_id", "campaign_id", "campaign_name",
    "impressions", "clicks", "cost", "conversions", "conversion_value",
]


def fetch_google_ads(customer_id: str, start: str, end: str) -> pd.DataFrame:
    """Google Ads API（GAQL）でキャンペーン別・日別の実績を取得する"""
    # 認証情報（開発者トークン・OAuthクライアント・リフレッシュトークン）は google-ads.yaml に置く
    client = GoogleAdsClient.load_from_storage("google-ads.yaml")
    service = client.get_service("GoogleAdsService")
    query = f"""
        SELECT segments.date, campaign.id, campaign.name,
               metrics.impressions, metrics.clicks, metrics.cost_micros,
               metrics.conversions, metrics.conversions_value
        FROM campaign
        WHERE segments.date BETWEEN '{start}' AND '{end}'
    """
    rows = []
    for batch in service.search_stream(customer_id=customer_id, query=query):
        for r in batch.results:
            rows.append({
                "date": r.segments.date,
                "platform": "google_ads",
                "account_id": customer_id,
                "campaign_id": str(r.campaign.id),
                "campaign_name": r.campaign.name,
                "impressions": r.metrics.impressions,
                "clicks": r.metrics.clicks,
                # Google広告の費用は micros（100万分の1単位）で返る
                "cost": r.metrics.cost_micros / 1_000_000,
                "conversions": r.metrics.conversions,
                "conversion_value": r.metrics.conversions_value,
            })
    return pd.DataFrame(rows, columns=COMMON_COLUMNS)


# Metaの「何をCVとみなすか」はアカウント設定次第。案件ごとに決めて設定ファイルで管理する
META_CONVERSION_ACTION = os.environ.get("META_CONVERSION_ACTION", "purchase")
META_API_VERSION = os.environ["META_API_VERSION"]  # 例: "vXX.0"（最新は公式で確認）


def _sum_action(actions: list | None, action_type: str) -> float:
    return sum(float(a["value"]) for a in (actions or []) if a["action_type"] == action_type)


def fetch_meta(ad_account_id: str, start: str, end: str) -> pd.DataFrame:
    """Meta Marketing API（Insights）でキャンペーン別・日別の実績を取得する"""
    url = f"https://graph.facebook.com/{META_API_VERSION}/act_{ad_account_id}/insights"
    params = {
        "level": "campaign",
        "fields": "campaign_id,campaign_name,impressions,clicks,spend,actions,action_values",
        "time_range": json.dumps({"since": start, "until": end}),
        "time_increment": 1,
        "access_token": os.environ["META_ACCESS_TOKEN"],
    }
    rows = []
    while url:
        res = requests.get(url, params=params, timeout=30)
        res.raise_for_status()
        body = res.json()
        for r in body.get("data", []):
            rows.append({
                "date": r["date_start"],
                "platform": "meta",
                "account_id": ad_account_id,
                "campaign_id": r["campaign_id"],
                "campaign_name": r["campaign_name"],
                # Metaは数値が文字列で返るので明示的に変換する
                "impressions": int(r.get("impressions", 0)),
                "clicks": int(r.get("clicks", 0)),
                "cost": float(r.get("spend", 0)),
                "conversions": _sum_action(r.get("actions"), META_CONVERSION_ACTION),
                "conversion_value": _sum_action(r.get("action_values"), META_CONVERSION_ACTION),
            })
        # 次ページのURLにはパラメータが含まれるので params は空にする
        url = body.get("paging", {}).get("next")
        params = {}
    return pd.DataFrame(rows, columns=COMMON_COLUMNS)


def load_all(start: str, end: str) -> pd.DataFrame:
    frames = [
        fetch_google_ads(os.environ["GOOGLE_ADS_CUSTOMER_ID"], start, end),
        fetch_meta(os.environ["META_AD_ACCOUNT_ID"], start, end),
    ]
    df = pd.concat(frames, ignore_index=True)
    df["date"] = pd.to_datetime(df["date"])
    return df


def fetch_ga4_channels(property_id: str, start: str, end: str) -> pd.DataFrame:
    """GA4 Data API：チャネル別のセッション・キーイベント（サイト側の答え合わせ用）"""
    from google.analytics.data_v1beta import BetaAnalyticsDataClient
    from google.analytics.data_v1beta.types import DateRange, Dimension, Metric, RunReportRequest

    client = BetaAnalyticsDataClient()  # サービスアカウント（GOOGLE_APPLICATION_CREDENTIALS）で認証
    res = client.run_report(RunReportRequest(
        property=f"properties/{property_id}",
        dimensions=[Dimension(name="date"), Dimension(name="sessionDefaultChannelGroup")],
        metrics=[Metric(name="sessions"), Metric(name="keyEvents")],
        date_ranges=[DateRange(start_date=start, end_date=end)],
    ))
    return pd.DataFrame([{
        "date": r.dimension_values[0].value,
        "channel": r.dimension_values[1].value,
        "sessions": int(r.metric_values[0].value),
        "key_events": float(r.metric_values[1].value),
    } for r in res.rows])


if __name__ == "__main__":
    import argparse

    p = argparse.ArgumentParser()
    p.add_argument("--start", required=True)
    p.add_argument("--end", required=True)
    a = p.parse_args()
    os.makedirs("data/raw", exist_ok=True)
    load_all(a.start, a.end).to_csv("data/raw/ads_daily.csv", index=False)
    print("saved: data/raw/ads_daily.csv")

# scripts/aggregate.py : 計算はすべてコードで行う（Claudeに四則演算させない）
import json

import numpy as np
import pandas as pd

BASE = ["impressions", "clicks", "cost", "conversions", "conversion_value"]
RATIOS = ["ctr", "cvr", "cpa", "roas"]


def _safe_div(a: pd.Series, b: pd.Series) -> pd.Series:
    # 0除算は NaN にする（「CV0件のCPA」を無限大や0として扱わない）
    return a / b.replace(0, np.nan)


def add_ratios(t: pd.DataFrame) -> pd.DataFrame:
    t = t.copy()
    t["ctr"] = (_safe_div(t["clicks"], t["impressions"]) * 100).round(2)   # %
    t["cvr"] = (_safe_div(t["conversions"], t["clicks"]) * 100).round(2)   # %
    t["cpa"] = _safe_div(t["cost"], t["conversions"]).round(0)             # 円
    t["roas"] = (_safe_div(t["conversion_value"], t["cost"]) * 100).round(1)  # %
    return t


def weekly_summary(df: pd.DataFrame, week_start: str) -> pd.DataFrame:
    """今週と前週を媒体別＋全体で集計し、前週比(%)を付ける"""
    start = pd.Timestamp(week_start)
    prev_start, end = start - pd.Timedelta(days=7), start + pd.Timedelta(days=7)

    # 媒体横断の合計行（ALL）を作る。比率は必ず「合計してから割る」
    both = pd.concat([df, df.assign(platform="ALL")], ignore_index=True)
    cur = both[(both["date"] >= start) & (both["date"] < end)]
    prev = both[(both["date"] >= prev_start) & (both["date"] < start)]

    cur = add_ratios(cur.groupby("platform")[BASE].sum())
    prev = add_ratios(prev.groupby("platform")[BASE].sum())
    out = cur.join(prev, rsuffix="_prev", how="outer")

    for m in ["cost", "conversions"] + RATIOS:
        out[f"{m}_wow_pct"] = (_safe_div(out[m] - out[f"{m}_prev"], out[f"{m}_prev"]) * 100).round(1)
    out["cost"] = out["cost"].round(0)
    out["cost_prev"] = out["cost_prev"].round(0)
    return out


def flag_cost_anomalies(df: pd.DataFrame, window: int = 28, z: float = 3.0) -> pd.DataFrame:
    """日別費用が過去window日の平均から z×標準偏差 以上ずれた日を検出する"""
    daily = df.groupby(["platform", "date"], as_index=False)[BASE].sum().sort_values("date")

    def _zscore(s: pd.Series) -> pd.Series:
        hist = s.shift(1).rolling(window, min_periods=14)  # 当日を含めない
        return (s - hist.mean()) / hist.std().replace(0, np.nan)

    daily["cost_z"] = daily.groupby("platform")["cost"].transform(_zscore).round(2)
    return daily[daily["cost_z"].abs() >= z]


def rule_flags(summary: pd.DataFrame, min_conv: float = 10, cpa_worse_pct: float = 20.0) -> list[dict]:
    """運用ルールに基づくフラグ。サンプルが少ない週は判定しない"""
    flags = []
    for platform, r in summary.iterrows():
        if r["conversions_prev"] >= min_conv and pd.notna(r["cpa_wow_pct"]) and r["cpa_wow_pct"] >= cpa_worse_pct:
            flags.append({"platform": platform, "type": "cpa_worse",
                          "cpa": float(r["cpa"]), "cpa_prev": float(r["cpa_prev"]),
                          "cpa_wow_pct": float(r["cpa_wow_pct"])})
        if r["conversions"] == 0 and r["cost"] > 0:
            flags.append({"platform": platform, "type": "zero_conversion", "cost": float(r["cost"])})
    return flags


def build_payload(df: pd.DataFrame, week_start: str) -> dict:
    """AIに渡す入力JSON。ここにある数値だけをAIに引用させる"""
    summary = weekly_summary(df, week_start)
    anomalies = flag_cost_anomalies(df)
    anomalies = anomalies[anomalies["date"] >= pd.Timestamp(week_start)]
    anomalies = anomalies.assign(date=anomalies["date"].dt.strftime("%Y-%m-%d"))
    return {
        "period": {"week_start": week_start,
                   "week_end": str((pd.Timestamp(week_start) + pd.Timedelta(days=6)).date())},
        "currency": "JPY",
        # to_json は NaN を null にしてくれる
        "summary": json.loads(summary.reset_index().to_json(orient="records", force_ascii=False)),
        "flags": rule_flags(summary),
        "cost_anomalies": json.loads(
            anomalies[["platform", "date", "cost", "cost_z"]].to_json(orient="records")),
    }


if __name__ == "__main__":
    import argparse
    import os

    p = argparse.ArgumentParser()
    p.add_argument("--week-start", required=True)
    a = p.parse_args()
    df = pd.read_csv("data/raw/ads_daily.csv", parse_dates=["date"])
    payload = build_payload(df, a.week_start)
    out_dir = f"data/weekly/{a.week_start}"
    os.makedirs(out_dir, exist_ok=True)
    with open(f"{out_dir}/summary.json", "w", encoding="utf-8") as f:
        json.dump(payload, f, ensure_ascii=False, indent=2)
    weekly_summary(df, a.week_start).to_csv(f"{out_dir}/summary.csv")
    print(f"saved: {out_dir}/summary.json, summary.csv")

{
  "period": {
    "week_start": "2026-09-14",
    "week_end": "2026-09-20"
  },
  "currency": "JPY",
  "summary": [
    {
      "platform": "ALL",
      "cost": 1545921.0,
      "conversions": 195.0,
      "cpa": 7928.0,
      "roas": 113.5,
      "cost_prev": 1477231.0,
      "conversions_prev": 314.0,
      "cpa_prev": 4705.0,
      "cost_wow_pct": 4.6,
      "cpa_wow_pct": 68.5
    }
  ],
  "flags": [
    {
      "platform": "ALL",
      "type": "cpa_worse",
      "cpa": 7928.0,
      "cpa_prev": 4705.0,
      "cpa_wow_pct": 68.5
    },
    {
      "platform": "google_ads",
      "type": "cpa_worse",
      "cpa": 8988.0,
      "cpa_prev": 4690.0,
      "cpa_wow_pct": 91.6
    }
  ],
  "cost_anomalies": [
    {
      "platform": "google_ads",
      "date": "2026-09-17",
      "cost": 256139.0,
      "cost_z": 15.24
    }
  ]
}

# 広告レポート自動化リポジトリ

## このリポジトリの役割
Google広告・Meta広告の実績を取得・集計し、週次レポート（Markdown）を作る。
Claudeの担当は「集計結果の解釈・文章化・掘り下げ分析・スクリプト保守」。数値計算はしない。

## ディレクトリ
- scripts/fetch_ads.py   … 媒体APIから取得 → data/raw/ads_daily.csv
- scripts/aggregate.py   … 集計・前週比・異常値 → data/weekly/<週初日>/summary.json
- scripts/breakdown.py   … 掘り下げ用の読み取り専用クエリ
- scripts/verify_report.py … レポート中の数値が集計結果に存在するか検証
- reports/<週初日>.md    … 週次レポートの出力先
- proposals/             … 入札・予算などの変更「提案」の出力先（実行は人）

## 指標定義（summary.json）
- cost=広告費（円・税抜）、conversions=CV数、conversion_value=CV価値（円）
- ctr=clicks/impressions×100（%）、cvr=conversions/clicks×100（%）
- cpa=cost/conversions（円、CV0件はnull）、roas=conversion_value/cost×100（%）
- *_prev=前週、*_wow_pct=前週比（%）、platform="ALL"=媒体合計

## 厳守ルール
- 数値を自分で計算・合算・換算しない。必要な数値は scripts/ を修正・実行して出す
- レポートには summary.json の数値を表記どおり引用する（「約」「万」への丸め禁止）
- summary.json にない事実・原因は「推測：」と明記し、確認方法を添える
- 成果を保証・断定する表現（「必ず改善」「確実に増える」等）を使わない
- 入札・予算・配信ON/OFFの変更は実行しない。proposals/ に提案を書くまで
- .env、google-ads.yaml、*.key は読まない・表示しない

---
name: weekly-ad-report
description: 週次の広告運用レポートを作成する。「週次レポート」「先週の広告の振り返り」などと頼まれたときに使う。引数は週初日（YYYY-MM-DD、月曜）。
---

# 週次広告レポートの作成手順

1. 週初日（月曜）を確認する。指定がなければ直近の完了週の月曜を使う
2. 前週比と異常値検知のため、週初日の35日前〜週末日の範囲でデータを取得する
   `python scripts/fetch_ads.py --start <週初日-35日> --end <週末日>`
3. 集計する: `python scripts/aggregate.py --week-start <週初日>`
4. `data/weekly/<週初日>/summary.json` を読み、以下の構成で `reports/<週初日>.md` を書く
   - `## サマリー`（3文以内）
   - `## 媒体別の数値`（summary.json の値をそのまま表にする）
   - `## ハイライト` / `## リスク・要確認`（flags と cost_anomalies は必ずどちらかで触れる）
   - `## 次のアクション（提案）`（優先度 high/medium/low、理由つき）
5. flags に cpa_worse がある媒体は、`python scripts/breakdown.py campaign ...` で
   キャンペーン別に分解し、悪化の主因候補を1〜3点に絞って「リスク」に書く
6. `python scripts/verify_report.py reports/<週初日>.md` を実行し、OK になるまで直す
7. Slack投稿はしない（投稿は run_weekly.sh が人のレビュー用チャンネルに行う）

## 書き方の注意
- 箇条書きは「-」を使う（番号付きリストは数値検証の邪魔になる）
- 数値には単位を付ける（円・%・件）

{
  "permissions": {
    "allow": [
      "Bash(python scripts/fetch_ads.py:*)",
      "Bash(python scripts/aggregate.py:*)",
      "Bash(python scripts/breakdown.py:*)",
      "Bash(python scripts/verify_report.py:*)",
      "Read(./data/**)",
      "Read(./reports/**)",
      "Write(./reports/**)",
      "Write(./proposals/**)"
    ],
    "ask": [
      "Edit(./scripts/**)"
    ],
    "deny": [
      "Read(./.env)",
      "Read(./google-ads.yaml)",
      "Read(./**/*.key)",
      "Bash(curl:*)",
      "Bash(python scripts/apply_changes.py:*)"
    ]
  },
  "hooks": {
    "PreToolUse": [
      {
        "matcher": "Bash",
        "hooks": [{ "type": "command", "command": "python scripts/guard_bash.py" }]
      }
    ],
    "Stop": [
      {
        "hooks": [{ "type": "command", "command": "python scripts/verify_report.py" }]
      }
    ]
  }
}

# scripts/guard_bash.py : PreToolUse フック。広告アカウントへの書き込み系コマンドを止める
import json
import re
import sys

BLOCK_PATTERNS = [
    r"apply_changes",      # 変更反映用スクリプトは人が手で実行する
    r"\bmutate",           # Google Ads API の更新系（Mutate）呼び出し
    r"requests\.post|curl\s.*-X\s*POST",  # 媒体APIへのPOST
    r"\.env\b|google-ads\.yaml",          # 認証情報ファイルの参照
]

event = json.load(sys.stdin)
command = event.get("tool_input", {}).get("command", "")

for pattern in BLOCK_PATTERNS:
    if re.search(pattern, command):
        print(f"ブロック: '{pattern}' に該当するコマンドは実行できません。"
              "入札・予算・配信の変更は proposals/ に提案として書き出し、人の承認を待ってください。",
              file=sys.stderr)
        sys.exit(2)  # exit 2 = ツール実行をブロックし、理由をClaudeに返す
sys.exit(0)

# scripts/verify_report.py : レポート中の数値が集計JSONに実在するかを突き合わせる
import glob
import json
import os
import re
import sys

NUM_RE = re.compile(r"\d[\d,]*(?:\.\d+)?")


def numbers_in(text: str) -> set[float]:
    return {float(m.replace(",", "")) for m in NUM_RE.findall(text)}


def collect_numbers(obj) -> set[float]:
    """集計JSON内の数値（文字列中の日付なども含む）をすべて集める"""
    if isinstance(obj, bool) or obj is None:
        return set()
    if isinstance(obj, (int, float)):
        return {abs(float(obj))}  # 「12.3%悪化」のように符号なしで書かれることがある
    if isinstance(obj, str):
        return numbers_in(obj)
    if isinstance(obj, dict):
        obj = list(obj.values())
    return set().union(*(collect_numbers(v) for v in obj)) if obj else set()


def unknown_numbers(report_md: str, payload: dict, tol: float = 0.05) -> list[float]:
    allowed = collect_numbers(payload)
    return [n for n in sorted(numbers_in(report_md))
            if n > 3  # 「3点」など小さな数は対象外（運用に合わせて調整）
            and not any(abs(n - a) <= tol for a in allowed)]


def latest_report() -> str | None:
    files = sorted(glob.glob("reports/*.md"))
    return files[-1] if files else None


def main() -> int:
    hook_input = {}
    if not sys.stdin.isatty():
        raw = sys.stdin.read()
        hook_input = json.loads(raw) if raw.strip() else {}

    path = sys.argv[1] if len(sys.argv) > 1 else latest_report()
    if not path:
        return 0
    week = os.path.basename(path).removesuffix(".md")
    summary_path = f"data/weekly/{week}/summary.json"
    if not os.path.exists(summary_path):
        return 0  # 週次レポート以外のMarkdownは対象外

    with open(path, encoding="utf-8") as f:
        report_md = f.read()
    with open(summary_path, encoding="utf-8") as f:
        payload = json.load(f)

    bad = unknown_numbers(report_md, payload)
    if not bad:
        print(f"OK: {path} の数値はすべて集計結果に存在します")
        return 0
    msg = (f"{path} に集計結果（{summary_path}）に存在しない数値があります: {bad}\n"
           "該当箇所を集計JSONの値そのままに直すか、数値を削除してください。")
    print(msg, file=sys.stderr)
    # Stopフックが2回目以降なら無限ループを避けて止める（最終確認は run_weekly.sh 側でも行う）
    if hook_input.get("stop_hook_active"):
        return 1
    return 2  # exit 2 = Claudeに理由を返して、修正させる


if __name__ == "__main__":
    sys.exit(main())

#!/usr/bin/env bash
# run_weekly.sh : 毎週月曜の朝に cron から起動する
set -euo pipefail
cd "$(dirname "$0")"
mkdir -p logs

# 直近の完了週（月〜日）の月曜日（GNU date 前提）
WEEK_START=$(date -d "-$(( $(date +%u) + 6 )) days" +%F)

claude -p "/weekly-ad-report ${WEEK_START}" \
  --allowedTools "Read,Write,Bash(python scripts/fetch_ads.py:*),Bash(python scripts/aggregate.py:*),Bash(python scripts/breakdown.py:*),Bash(python scripts/verify_report.py:*)" \
  --output-format json > "logs/claude-${WEEK_START}.json"

test -f "reports/${WEEK_START}.md"   # レポートが無ければここで失敗させる

# フックとは別に、最後にもう一度コードで検証してから共有する
if python scripts/verify_report.py "reports/${WEEK_START}.md" < /dev/null; then
  python scripts/post_slack.py "reports/${WEEK_START}.md"
else
  python scripts/post_slack.py "reports/${WEEK_START}.md" "数値検証NG。公開前に必ず人が確認してください"
fi

# 毎週月曜 8:45（サーバーのタイムゾーン）に実行。APIキー等は run_weekly.sh 側で読み込む
45 8 * * 1 /home/ops/ad-report/run_weekly.sh >> /home/ops/ad-report/logs/cron.log 2>&1

# .github/workflows/weekly-report.yml
name: weekly-ad-report
on:
  schedule:
    - cron: "0 0 * * 1"   # 毎週月曜 9:00 JST（UTCで指定）
  workflow_dispatch:

jobs:
  report:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.12"
      - run: pip install -r requirements.txt
      - run: npm install -g @anthropic-ai/claude-code
      - name: Write credentials
        run: echo "$GOOGLE_ADS_YAML" > google-ads.yaml
        env:
          GOOGLE_ADS_YAML: ${{ secrets.GOOGLE_ADS_YAML }}
      - name: Generate report
        run: ./run_weekly.sh
        env:
          ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
          META_ACCESS_TOKEN: ${{ secrets.META_ACCESS_TOKEN }}
          META_API_VERSION: ${{ vars.META_API_VERSION }}
          META_AD_ACCOUNT_ID: ${{ vars.META_AD_ACCOUNT_ID }}
          GOOGLE_ADS_CUSTOMER_ID: ${{ vars.GOOGLE_ADS_CUSTOMER_ID }}
          SLACK_REVIEW_WEBHOOK_URL: ${{ secrets.SLACK_REVIEW_WEBHOOK_URL }}
      - uses: actions/upload-artifact@v4
        with:
          name: weekly-report
          path: reports/

# scripts/post_slack.py : 保存済みのMarkdownレポートをSlackのレビュー用チャンネルへ投稿する
import os
import sys

import requests


def post(path: str, warning: str = "") -> None:
    with open(path, encoding="utf-8") as f:
        body = f.read()
    text = f"*【下書き・要レビュー】{os.path.basename(path)}*\n"
    if warning:
        text += f":warning: {warning}\n"
    text += "\n" + body
    # Incoming Webhook のURLは秘密情報。コードに直書きせず環境変数から読む
    res = requests.post(os.environ["SLACK_REVIEW_WEBHOOK_URL"], json={"text": text}, timeout=10)
    res.raise_for_status()


if __name__ == "__main__":
    post(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else "")

# scripts/breakdown.py : 掘り下げ分析用の読み取り専用クエリ（Claude Codeから呼ばせる）
import argparse

import pandas as pd

SOURCES = {
    "campaign": ("data/raw/ads_daily.csv", ["platform", "campaign_id", "campaign_name"]),
    "search_terms": ("data/raw/search_terms.csv", ["campaign_id", "search_term"]),
    "creative": ("data/raw/creative_daily.csv", ["platform", "campaign_id", "ad_id", "ad_name"]),
}


def main() -> None:
    p = argparse.ArgumentParser(description="期間を指定して内訳を集計する（読み取り専用）")
    p.add_argument("kind", choices=SOURCES.keys())
    p.add_argument("--start", required=True)
    p.add_argument("--end", required=True)
    p.add_argument("--campaign-id")
    p.add_argument("--top", type=int, default=30)
    a = p.parse_args()

    path, keys = SOURCES[a.kind]
    df = pd.read_csv(path, dtype={"campaign_id": str, "ad_id": str})
    df = df[(df["date"] >= a.start) & (df["date"] <= a.end)]
    if a.campaign_id:
        df = df[df["campaign_id"] == a.campaign_id]

    t = df.groupby(keys, as_index=False)[["clicks", "cost", "conversions"]].sum()
    t["cpa"] = (t["cost"] / t["conversions"].where(t["conversions"] > 0)).round(0)
    t = t.sort_values("cost", ascending=False).head(a.top)
    print(t.to_csv(index=False))  # Claudeが読みやすいようにCSVで標準出力へ


if __name__ == "__main__":
    main()

---
name: ad-investigator
description: CPAやCVの悪化原因を、キャンペーン→検索語句→クリエイティブの順に掘り下げる調査係。数値の変動理由を調べるときに使う。
tools: Read, Grep, Glob, Bash
---

あなたは広告運用の調査担当です。
- データは `python scripts/breakdown.py` の出力だけを根拠にする（campaign / search_terms / creative）
- 「前週と比べて何が変わったか」を、費用の大きい順に確認する
- 仮説ごとに「根拠となった出力」「確からしさ（高・中・低）」「人が追加で確認すべきこと」を書く
- 入札・予算・配信の変更は提案のみ。proposals/ に書き出し、実行しない

# 補足：自社プロダクトに組み込む場合（Claude API の structured outputs）
import json

import anthropic

client = anthropic.Anthropic()  # 環境変数 ANTHROPIC_API_KEY を読む

schema = {
    "type": "object",
    "properties": {
        "summary": {"type": "string"},
        "highlights": {"type": "array", "items": {"type": "string"}},
        "risks": {"type": "array", "items": {"type": "string"}},
        "next_actions": {"type": "array", "items": {"type": "string"}},
    },
    "required": ["summary", "highlights", "risks", "next_actions"],
    "additionalProperties": False,
}

with open("data/weekly/2026-09-14/summary.json", encoding="utf-8") as f:
    payload = json.load(f)

response = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=16000,
    system="集計JSONの数値だけを引用し、再計算しないこと。推測は「推測：」と明記すること。",
    messages=[{"role": "user", "content": json.dumps(payload, ensure_ascii=False)}],
    output_config={"format": {"type": "json_schema", "schema": schema}},
)
text = next(b.text for b in response.content if b.type == "text")
report = json.loads(text)
print(report["summary"])

# 自作の媒体データ取得MCPサーバー（stdio）を、プロジェクト共有で登録
claude mcp add --scope project ads-data -- python mcp_server/ads_server.py

# リモートのMCPサーバー（Streamable HTTP）を登録する場合
claude mcp add --transport http ads-remote https://mcp.example.com/mcp