Gemini API × Google Analytics 4 完全連携ガイド — AI駆動型ビジネスインテリジェンス・異常検知・予測分析パイプラインの本番実装

# 必要なPythonパッケージをインストール
pip install google-analytics-data google-generativeai python-dotenv pandas matplotlib slack-sdk

GOOGLE_APPLICATION_CREDENTIALS=path/to/service-account-key.json
GA4_PROPERTY_ID=12345678
GEMINI_API_KEY=YOUR_GEMINI_API_KEY
SLACK_BOT_TOKEN=YOUR_SLACK_BOT_TOKEN
SLACK_CHANNEL_ID=YOUR_CHANNEL_ID

import os
from dotenv import load_dotenv
from google.analytics.data_v1beta import BetaAnalyticsDataClient
from google.analytics.data_v1beta.types import (
    DateRange, Dimension, Metric, RunReportRequest
)
import pandas as pd
 
load_dotenv()
 
PROPERTY_ID = os.environ["GA4_PROPERTY_ID"]
 
def get_ga4_data(start_date: str, end_date: str, dimensions: list, metrics: list) -> pd.DataFrame:
    """GA4 Data APIでデータを取得してDataFrameとして返す"""
    client = BetaAnalyticsDataClient()
 
    request = RunReportRequest(
        property=f"properties/{PROPERTY_ID}",
        dimensions=[Dimension(name=d) for d in dimensions],
        metrics=[Metric(name=m) for m in metrics],
        date_ranges=[DateRange(start_date=start_date, end_date=end_date)],
    )
 
    response = client.run_report(request)
 
    # DataFrameに変換
    rows = []
    for row in response.rows:
        row_dict = {}
        for i, dim in enumerate(dimensions):
            row_dict[dim] = row.dimension_values[i].value
        for i, metric in enumerate(metrics):
            row_dict[metric] = float(row.metric_values[i].value)
        rows.append(row_dict)
 
    return pd.DataFrame(rows)
 
# テスト実行
df = get_ga4_data(
    start_date="7daysAgo",
    end_date="today",
    dimensions=["date", "sessionSourceMedium"],
    metrics=["sessions", "engagedSessions", "conversions", "totalRevenue"]
)
print(df.head())
# 出力例:
#         date sessionSourceMedium  sessions  engagedSessions  conversions  totalRevenue
# 0  20260401          google / cpc    1243.0          867.0        42.0       125400.0
# 1  20260401       organic / search    3421.0         2156.0        89.0       267000.0

import google.generativeai as genai
import json
from datetime import datetime, timedelta
 
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
 
# GA4クエリ生成のためのFunction定義
ga4_query_tool = genai.protos.Tool(
    function_declarations=[
        genai.protos.FunctionDeclaration(
            name="get_analytics_data",
            description="GA4から指定条件でアクセス解析データを取得する",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "start_date": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="開始日（例: '7daysAgo', '30daysAgo', '2026-04-01'）"
                    ),
                    "end_date": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="終了日（例: 'today', 'yesterday', '2026-04-06'）"
                    ),
                    "dimensions": genai.protos.Schema(
                        type=genai.protos.Type.ARRAY,
                        items=genai.protos.Schema(type=genai.protos.Type.STRING),
                        description="ディメンション名リスト（date, sessionSourceMedium, deviceCategory, country, landingPage等）"
                    ),
                    "metrics": genai.protos.Schema(
                        type=genai.protos.Type.ARRAY,
                        items=genai.protos.Schema(type=genai.protos.Type.STRING),
                        description="メトリクス名リスト（sessions, engagedSessions, conversions, totalRevenue, bounceRate等）"
                    ),
                    "analysis_question": genai.protos.Schema(
                        type=genai.protos.Type.STRING,
                        description="このデータで答えたい質問"
                    ),
                },
                required=["start_date", "end_date", "dimensions", "metrics"]
            )
        )
    ]
)
 
SYSTEM_PROMPT = """あなたはGoogle Analytics 4の専門アナリストです。
ユーザーの質問を分析し、必要なGA4データを取得して洞察を提供してください。
データ取得後は、数値の変化率・トレンド・原因・改善提案を含む詳細な分析を行ってください。
回答は日本語で、マーケターが理解しやすい形式で提供してください。"""
 
def natural_language_analytics(question: str) -> str:
    """自然言語でGA4データを問い合わせる"""
    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro",
        system_instruction=SYSTEM_PROMPT,
        tools=[ga4_query_tool]
    )
 
    chat = model.start_chat()
    response = chat.send_message(question)
 
    # Function Callingが発生した場合はデータを取得して再送信
    while response.candidates[0].content.parts[0].function_call.name:
        fc = response.candidates[0].content.parts[0].function_call
        func_args = dict(fc.args)
 
        # GA4からデータ取得
        df = get_ga4_data(
            start_date=func_args["start_date"],
            end_date=func_args["end_date"],
            dimensions=func_args["dimensions"],
            metrics=func_args["metrics"]
        )
 
        # DataFrameをJSON文字列に変換してGeminiに渡す
        data_json = df.to_json(orient="records", force_ascii=False, indent=2)
 
        response = chat.send_message(
            genai.protos.Part(
                function_response=genai.protos.FunctionResponse(
                    name=fc.name,
                    response={"result": data_json, "rows": len(df)}
                )
            )
        )
 
        # 次のfunctionCallがない場合はループを抜ける
        if not response.candidates[0].content.parts[0].HasField("function_call"):
            break
 
    return response.text
 
# 使用例
question = "先週と先々週のコンバージョン数を比較して、大きな変化があった流入元を教えてください"
result = natural_language_analytics(question)
print(result)
# 出力例:
# 先週（3/30-4/5）と先々週（3/23-3/29）の比較分析
#
# 【主要な変化】
# - organic / search: 89件 → 112件（+25.8%増）
# - cpc / google: 42件 → 28件（-33.3%減）
#
# 【考えられる原因】
# CPC広告のコンバージョン急減は3/31〜4/2のキャンペーン休止期間と重なっています...

import numpy as np
from scipy import stats
 
def detect_anomalies(df: pd.DataFrame, metric: str, z_threshold: float = 2.5) -> pd.DataFrame:
    """Z-scoreを用いた統計的異常検知"""
    values = df[metric].values
    z_scores = np.abs(stats.zscore(values))
 
    df_result = df.copy()
    df_result["z_score"] = z_scores
    df_result["is_anomaly"] = z_scores > z_threshold
    df_result["deviation_pct"] = ((df[metric] - df[metric].mean()) / df[metric].mean() * 100).round(1)
 
    return df_result
 
def analyze_anomalies_with_gemini(anomaly_df: pd.DataFrame, metric: str) -> str:
    """Geminiを使って異常の原因分析と対策を提案する"""
    # 異常日のデータを抽出
    anomalies = anomaly_df[anomaly_df["is_anomaly"]].copy()
    if anomalies.empty:
        return "分析期間中に統計的な異常は検出されませんでした。"
 
    # サマリーデータを準備
    summary = {
        "metric": metric,
        "period_avg": round(anomaly_df[metric].mean(), 1),
        "period_std": round(anomaly_df[metric].std(), 1),
        "anomalies": anomalies[["date", metric, "z_score", "deviation_pct"]].to_dict(orient="records")
    }
 
    model = genai.GenerativeModel("gemini-2.5-pro")
    prompt = f"""
以下のGA4データ異常検知結果を分析してください。
 
メトリクス: {metric}
分析期間の平均値: {summary['period_avg']}
標準偏差: {summary['period_std']}
検出された異常:
{json.dumps(summary['anomalies'], ensure_ascii=False, indent=2)}
 
以下の観点で分析してください:
1. 各異常日の数値変化とその重大性
2. 考えられる原因（季節性・キャンペーン・外部要因・技術的問題など）
3. 追加調査が必要なポイント
4. 具体的な改善アクション（優先度付き）
 
マーケティング担当者向けに、専門用語を避けてわかりやすく説明してください。
"""
    response = model.generate_content(prompt)
    return response.text
 
# 実行例: 過去30日のセッション数の異常検知
df_30days = get_ga4_data(
    start_date="30daysAgo",
    end_date="today",
    dimensions=["date"],
    metrics=["sessions", "conversions", "engagedSessions"]
)
df_30days = df_30days.sort_values("date")
 
anomaly_result = detect_anomalies(df_30days, "sessions")
analysis = analyze_anomalies_with_gemini(anomaly_result, "sessions")
print(analysis)

def compute_derived_metrics(df: pd.DataFrame) -> pd.DataFrame:
    """CVR・エンゲージメント率などの複合指標を計算"""
    df = df.copy()
    df["cvr"] = (df["conversions"] / df["sessions"] * 100).round(3)
    df["engagement_rate"] = (df["engagedSessions"] / df["sessions"] * 100).round(1)
    return df
 
def run_full_anomaly_scan() -> dict:
    """全主要指標の異常スキャンを実行"""
    df = get_ga4_data("30daysAgo", "today", ["date"],
                       ["sessions", "conversions", "engagedSessions", "totalRevenue"])
    df = df.sort_values("date")
    df = compute_derived_metrics(df)
 
    metrics_to_check = ["sessions", "cvr", "engagement_rate", "totalRevenue"]
    results = {}
 
    for metric in metrics_to_check:
        anomaly_df = detect_anomalies(df, metric)
        anomalies = anomaly_df[anomaly_df["is_anomaly"]]
 
        if not anomalies.empty:
            results[metric] = {
                "anomaly_count": len(anomalies),
                "worst_day": anomalies.loc[anomalies["z_score"].idxmax(), "date"],
                "worst_deviation": anomalies["deviation_pct"].abs().max(),
                "analysis": analyze_anomalies_with_gemini(anomaly_df, metric)
            }
 
    return results
 
# 実行
scan_results = run_full_anomaly_scan()
for metric, data in scan_results.items():
    print(f"\n{'='*50}")
    print(f"⚠️  {metric} — {data['anomaly_count']}件の異常を検出")
    print(f"最大偏差日: {data['worst_day']} ({data['worst_deviation']:.1f}%)")
    print(data['analysis'][:500])

from datetime import datetime, timedelta
 
def generate_weekly_report() -> str:
    """Geminiを使った週次BIレポートを自動生成する"""
 
    # 今週・先週のデータを取得
    today = datetime.now()
    last_monday = today - timedelta(days=today.weekday() + 7)
    last_sunday = last_monday + timedelta(days=6)
    prev_monday = last_monday - timedelta(days=7)
    prev_sunday = last_monday - timedelta(days=1)
 
    def fmt(d): return d.strftime("%Y-%m-%d")
 
    # チャネル別データ
    channel_data_this = get_ga4_data(
        fmt(last_monday), fmt(last_sunday),
        ["sessionSourceMedium"],
        ["sessions", "engagedSessions", "conversions", "totalRevenue"]
    )
    channel_data_prev = get_ga4_data(
        fmt(prev_monday), fmt(prev_sunday),
        ["sessionSourceMedium"],
        ["sessions", "engagedSessions", "conversions", "totalRevenue"]
    )
 
    # デバイス別データ
    device_data = get_ga4_data(
        fmt(last_monday), fmt(last_sunday),
        ["deviceCategory"],
        ["sessions", "engagedSessions", "conversions"]
    )
 
    # ランディングページTOP10
    page_data = get_ga4_data(
        fmt(last_monday), fmt(last_sunday),
        ["landingPage"],
        ["sessions", "engagedSessions", "conversions"]
    )
    page_data = page_data.sort_values("sessions", ascending=False).head(10)
 
    # Geminiでレポート生成
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
以下のGA4データから、経営者・マーケター向けの週次レポートを作成してください。
 
【今週: {fmt(last_monday)} 〜 {fmt(last_sunday)}】
チャネル別パフォーマンス:
{channel_data_this.to_string(index=False)}
 
【先週: {fmt(prev_monday)} 〜 {fmt(prev_sunday)}】
チャネル別パフォーマンス:
{channel_data_prev.to_string(index=False)}
 
デバイス別（今週）:
{device_data.to_string(index=False)}
 
ランディングページTOP10（今週）:
{page_data.to_string(index=False)}
 
レポート要件:
1. エグゼクティブサマリー（3行以内）
2. 主要KPI前週比較（セッション・CVR・売上）
3. チャネル別パフォーマンス分析と注目点
4. デバイス別トレンドとモバイル最適化状況
5. トップランディングページの成果と改善余地
6. 来週の推奨アクション（優先度順に3点）
 
Markdown形式で、読みやすく整理してください。
"""
    response = model.generate_content(prompt)
    return response.text
 
weekly_report = generate_weekly_report()
print(weekly_report)

from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError
 
def send_report_to_slack(report_text: str, anomaly_results: dict):
    """週次レポートとアラートをSlackに送信する"""
    client = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    channel = os.environ["SLACK_CHANNEL_ID"]
 
    try:
        # アノマリーアラートがある場合はまず送信
        if anomaly_results:
            alert_lines = ["*⚠️ 今週のアノマリーアラート*"]
            for metric, data in anomaly_results.items():
                alert_lines.append(
                    f"• `{metric}`: {data['anomaly_count']}件の異常 "
                    f"（最大偏差: {data['worst_deviation']:.1f}%、{data['worst_day']}）"
                )
            client.chat_postMessage(channel=channel, text="\n".join(alert_lines))
 
        # 週次レポート送信（Slackの文字数制限対応）
        max_len = 3000
        chunks = [report_text[i:i+max_len] for i in range(0, len(report_text), max_len)]
 
        for i, chunk in enumerate(chunks):
            prefix = "*📊 週次GA4レポート*\n\n" if i == 0 else ""
            client.chat_postMessage(channel=channel, text=prefix + chunk)
 
        print("✅ Slackへの送信が完了しました")
 
    except SlackApiError as e:
        print(f"❌ Slack送信エラー: {e.response['error']}")
 
# 実行例
anomalies = run_full_anomaly_scan()
report = generate_weekly_report()
send_report_to_slack(report, anomalies)

from google.cloud import bigquery
 
def natural_language_to_bigquery(nl_question: str, project_id: str, dataset: str) -> pd.DataFrame:
    """自然言語をBigQueryクエリに変換してGA4イベントデータを取得"""
 
    # BigQueryスキーマの説明をGeminiに提供
    schema_description = f"""
GA4 BigQueryエクスポートのテーブル: `{project_id}.{dataset}.events_*`
主要カラム:
- event_date (STRING): イベント日付（YYYYMMDD形式）
- event_timestamp (INTEGER): マイクロ秒タイムスタンプ
- event_name (STRING): イベント名（page_view, purchase, scroll等）
- user_pseudo_id (STRING): ユーザー識別子（匿名）
- geo.country (STRING): 国
- device.category (STRING): デバイス種別（mobile/desktop/tablet）
- traffic_source.medium (STRING): 流入メディア
- ecommerce.purchase_revenue (FLOAT): 購入金額
- event_params: ネスト配列（key-value形式のイベントパラメータ）
"""
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
{schema_description}
 
以下の質問に答えるBigQueryクエリを生成してください。
質問: {nl_question}
 
要件:
- 過去30日分のデータを対象とする（_TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)) AND FORMAT_DATE('%Y%m%d', CURRENT_DATE())）
- 実行可能な標準SQLのみ
- SQLのみを返す（説明文不要）
"""
 
    response = model.generate_content(prompt)
    sql_query = response.text.strip().replace("```sql", "").replace("```", "").strip()
 
    # BigQueryでクエリを実行
    bq_client = bigquery.Client(project=project_id)
    df = bq_client.query(sql_query).to_dataframe()
 
    return df, sql_query
 
# 使用例
df_result, generated_sql = natural_language_to_bigquery(
    "モバイルユーザーの購入完了率が高い上位5ページを教えて",
    project_id="my-analytics-project",
    dataset="analytics_12345678"
)
print("生成されたSQL:")
print(generated_sql[:300])
print("\n結果:")
print(df_result)

# main.py (Cloud Functions / Cloud Run で実行)
import functions_framework
from flask import Request
 
@functions_framework.http
def weekly_analytics_report(request: Request):
    """週次レポート生成エンドポイント（Cloud Functions用）"""
    try:
        # 異常検知
        anomaly_results = run_full_anomaly_scan()
 
        # 週次レポート生成
        report = generate_weekly_report()
 
        # Slack通知
        send_report_to_slack(report, anomaly_results)
 
        # ログ記録
        print(f"週次レポート完了: {datetime.now().isoformat()}")
        print(f"検出アノマリー: {len(anomaly_results)}件")
 
        return {"status": "success", "anomaly_count": len(anomaly_results)}, 200
 
    except Exception as e:
        print(f"エラー: {str(e)}")
        return {"status": "error", "message": str(e)}, 500

gcloud scheduler jobs create http weekly-analytics-report \
  --schedule="0 0 * * MON" \
  --uri="https://YOUR_CLOUD_FUNCTION_URL" \
  --time-zone="Asia/Tokyo" \
  --message-body="{}" \
  --headers="Content-Type=application/json"

# crontab -e に追加（毎週月曜 9:00）
0 9 * * 1 cd /path/to/project && python run_weekly_report.py >> /var/log/analytics_report.log 2>&1

def analyze_user_segments() -> str:
    """ユーザーセグメント別の行動分析"""
 
    # 新規・リピート別データ
    segment_data = get_ga4_data(
        "30daysAgo", "today",
        ["newVsReturning", "deviceCategory", "country"],
        ["sessions", "engagedSessions", "conversions", "totalRevenue", "userEngagementDuration"]
    )
 
    # エンゲージメント時間を分単位に変換
    segment_data["avg_engagement_min"] = (
        segment_data["userEngagementDuration"] / segment_data["sessions"] / 60
    ).round(1)
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    prompt = f"""
以下のGA4ユーザーセグメントデータを深掘り分析してください。
 
{segment_data.to_string(index=False)}
 
分析してほしい点:
1. 新規・リピーターそれぞれの特徴と価値
2. 高コンバージョンセグメントの共通点
3. エンゲージメント時間とCVRの相関
4. 国別・デバイス別の特徴と最適化余地
5. 最も注力すべきセグメントとアクション提案
 
日本語でデータドリブンな分析をお願いします。
"""
    return model.generate_content(prompt).text
 
segment_analysis = analyze_user_segments()
print(segment_analysis)