ある月曜日の朝、AdMobのダッシュボードを開いたら前日比で収益が62%下落していました。原因を特定し始めたのはその時点からで、対処が完了したのは4時間後のことです。あの4時間の損失は、1時間以内に検知できていれば半分以下に抑えられていたはずだと今でも思います。
2014年から個人でモバイルアプリを開発し、壁紙系アプリを中心に累計5,000万ダウンロードを達成してきました。AdMobの月収はピーク時で150万円を超えていたこともあります。そうなると、収益の異常をどれだけ早く気づけるかは、事業継続の問題と直結してきます。
Gemini APIのFunction Callingを使って、AdMob Reporting APIから収益データを自動取得し、前日比・前週比で異常を検知したら即座にSlack通知を送る仕組みを作りました。今回はその実装を紹介します。
なぜGemini APIのFunction Callingを選んだか
最初は単純なシェルスクリプトでAdMob APIを叩き、前日比が一定閾値を下回ったらアラートを送る実装を試みました。ただ、固定閾値では誤検知が多く使いものになりませんでした。月曜日は週明け特有の低水準があり、祝日前後は変動が大きく、キャンペーン期間中は逆に急増するからです。
Gemini APIを間に挟むことで、単純な数値比較ではなく「この変動は通常の曜日変動の範囲内か、それとも真の異常か」という判断をGeminiに委ねることができます。コンテキストを渡してGeminiに診断させる、という使い方です。
Functionの定義 → AdMob APIの実行 → Geminiによる結果解析 → 異常時にSlack通知、という流れです。
AdMob Reporting APIの準備
まずGoogle Cloud ConsoleでAdMob APIを有効化し、サービスアカウントを作成します。サービスアカウントのJSONキーをダウンロードして環境変数に設定します。
export GOOGLE_APPLICATION_CREDENTIALS="/path/to/service-account-key.json"
export ADMOB_PUBLISHER_ID="pub-XXXXXXXXXXXXXXXX" # あなたのパブリッシャーID
export GEMINI_API_KEY="YOUR_GEMINI_API_KEY"AdMob Reporting APIでは「ネットワークレポート」と「メディエーションレポート」の2種類が取得できます。今回は収益の総額を確認するためにネットワークレポートを使用します。
from google.oauth2 import service_account
from googleapiclient.discovery import build
from datetime import datetime, timedelta
import os
def get_admob_credentials():
"""AdMob APIの認証情報を取得する"""
credentials = service_account.Credentials.from_service_account_file(
os.environ["GOOGLE_APPLICATION_CREDENTIALS"],
scopes=["https://www.googleapis.com/auth/admob.readonly"]
)
return credentials
def fetch_admob_revenue(days_ago: int = 1) -> dict:
"""
AdMob Reporting APIから指定日の収益データを取得する
Args:
days_ago: 何日前のデータを取得するか（1=昨日）
Returns:
{"date": "YYYY-MM-DD", "revenue": float, "impressions": int, "ecpm": float}
"""
credentials = get_admob_credentials()
service = build("admob", "v1", credentials=credentials)
target_date = datetime.now() - timedelta(days=days_ago)
date_str = target_date.strftime("%Y%m%d")
publisher_id = os.environ["ADMOB_PUBLISHER_ID"]
report_request = {
"reportSpec": {
"dateRange": {
"startDate": {
"year": target_date.year,
"month": target_date.month,
"day": target_date.day
},
"endDate": {
"year": target_date.year,
"month": target_date.month,
"day": target_date.day
}
},
"metrics": ["ESTIMATED_EARNINGS", "IMPRESSIONS", "IMPRESSION_RPM"],
"dimensions": [], # 集計（アプリ・広告ユニット別に分解しない）
}
}
response = service.accounts().networkReport().generate(
parent=f"accounts/{publisher_id}",
body=report_request
).execute()
# レスポンスのパース
revenue = 0.0
impressions = 0
ecpm = 0.0
for row in response:
if "row" in row:
metrics = row["row"].get("metricValues", {})
revenue += float(metrics.get("ESTIMATED_EARNINGS", {}).get("microsValue", 0)) / 1_000_000
impressions += int(metrics.get("IMPRESSIONS", {}).get("integerValue", 0))
ecpm = float(metrics.get("IMPRESSION_RPM", {}).get("doubleValue", 0))
return {
"date": target_date.strftime("%Y-%m-%d"),
"revenue": round(revenue, 2),
"impressions": impressions,
"ecpm": round(ecpm, 4)
}Gemini APIのFunction Callingで収益データを取得・分析する
Function Callingを使って、Geminiが必要なデータを自ら取得しながら分析できるようにします。
import google.generativeai as genai
import json
import os
genai.configure(api_key=os.environ["GEMINI_API_KEY"])
# Geminiが呼び出せるFunctionを定義
tools = [
{
"function_declarations": [
{
"name": "get_revenue_data",
"description": "AdMob収益データを取得する。days_agoで何日前のデータかを指定する",
"parameters": {
"type": "object",
"properties": {
"days_ago": {
"type": "integer",
"description": "何日前のデータを取得するか。1=昨日、2=一昨日、7=1週間前"
}
},
"required": ["days_ago"]
}
}
]
}
]
def run_revenue_analysis():
"""
Gemini APIのFunction Callingを使って収益異常を分析する
"""
model = genai.GenerativeModel(
model_name="gemini-2.5-pro-preview-05-06",
tools=tools
)
today_weekday = datetime.now().strftime("%A") # 曜日情報をコンテキストで渡す
# Geminiへの分析依頼
initial_prompt = f"""
今日は{today_weekday}です。
get_revenue_data関数を使って以下を取得し、収益の異常がないか分析してください:
1. 昨日（1日前）の収益データ
2. 2日前の収益データ
3. 3日前の収益データ
4. 1週間前（7日前）の収益データ
5. 2週間前（14日前）の収益データ
分析の観点:
- 前日比での変化率
- 前週同曜日との比較
- eCPM（1000インプレッション単価）の変化
- 異常と判断する場合はその根拠と考えられる原因
最終的に以下のJSON形式で結果を返してください:
{{
"is_anomaly": true/false,
"severity": "low/medium/high",
"yesterday_revenue": 昨日の収益（ドル）,
"change_from_prev_day": 前日比変化率（%）,
"change_from_last_week": 前週同曜日比変化率（%）,
"summary": "状況の要約（日本語で1〜2文）",
"possible_causes": ["考えられる原因のリスト"]
}}
"""
chat = model.start_chat()
response = chat.send_message(initial_prompt)
# Function Callのループ処理
while response.candidates[0].content.parts[0].HasField("function_call"):
fc = response.candidates[0].content.parts[0].function_call
if fc.name == "get_revenue_data":
days_ago = fc.args["days_ago"]
result = fetch_admob_revenue(days_ago=days_ago)
print(f" 📊 {result['date']}: ${result['revenue']:.2f} ({result['impressions']:,} impressions, eCPM ${result['ecpm']:.2f})")
# Function Callの結果をGeminiに返す
response = chat.send_message(
genai.protos.Content(
parts=[genai.protos.Part(
function_response=genai.protos.FunctionResponse(
name="get_revenue_data",
response={"result": result}
)
)]
)
)
# 最終分析結果を取得
final_text = response.text
# JSONブロックを抽出
import re
json_match = re.search(r'\{.*\}', final_text, re.DOTALL)
if json_match:
return json.loads(json_match.group())
return None
# 実行例
print("🔍 AdMob収益分析を開始します...")
analysis = run_revenue_analysis()
if analysis:
if analysis["is_anomaly"]:
severity_emoji = {"low": "🟡", "medium": "🟠", "high": "🔴"}.get(analysis["severity"], "⚠️")
print(f"\n{severity_emoji} 異常検知: {analysis['summary']}")
print(f" 昨日の収益: ${analysis['yesterday_revenue']:.2f}")
print(f" 前日比: {analysis['change_from_prev_day']:+.1f}%")
print(f" 前週同曜日比: {analysis['change_from_last_week']:+.1f}%")
print(f" 考えられる原因: {', '.join(analysis['possible_causes'])}")
else:
print(f"\n✅ 収益は正常範囲内です。昨日の収益: ${analysis['yesterday_revenue']:.2f}")期待される実行例（異常検知時）:
🔍 AdMob収益分析を開始します...
📊 2026-05-12: $42.30 (128,500 impressions, eCPM $0.33)
📊 2026-05-11: $98.20 (182,300 impressions, eCPM $0.54)
📊 2026-05-10: $91.80 (179,100 impressions, eCPM $0.51)
📊 2026-05-06: $95.40 (180,200 impressions, eCPM $0.53)
📊 2026-04-29: $93.10 (177,800 impressions, eCPM $0.52)
🔴 異常検知: 昨日の収益が前日比57%減少。eCPMも0.54→0.33と急落しています。
昨日の収益: $42.30
前日比: -56.9%
前週同曜日比: -55.7%
考えられる原因: インプレッション数の減少, eCPMの急落, 広告埋め込みの問題または広告主予算の消化
実際に動かして気づいたこと
Function Callingの実装で最初に詰まったのは、Function Callのループ処理でした。Geminiは1回のリクエストで複数のFunction Callを行うことがあります。「昨日と1週間前のデータを取得してください」と頼むと、2回分のFunction Callをまとめて返してくることがあり、そのハンドリングが必要です。
また、AdMob Reporting APIのレスポンスで収益がマイクロ通貨（microsValue）で返ってくる点も注意が必要です。$1.00 = 1,000,000 microsValue という変換を忘れると、数値が100万倍になります。私は最初これに気づかず、「収益が突然100万ドルになった」とシステムが検知する事態になりました。
もう一つ気づいたのは、GeminiはFunction Callingの結果に基づいて追加のデータを要求することがあるという点です。最初は3日分だけ取得していましたが、Geminiが「前週のデータも確認したい」と判断して追加のFunction Callを要求してくることがあります。上記コードのようにループで処理するのが重要です。
詳細な実装パターンについてはGemini APIのFunction Calling完全ガイドも参照してみてください。
Slack通知まで組み合わせる
分析結果を受け取ったら、requestsライブラリでSlackのIncoming Webhookに送信するだけです。
import requests
def send_slack_alert(analysis: dict, webhook_url: str):
"""異常検知時にSlackに通知する"""
if not analysis["is_anomaly"]:
return
severity_color = {"low": "#FFC107", "medium": "#FF5722", "high": "#F44336"}.get(
analysis["severity"], "#9E9E9E"
)
payload = {
"attachments": [{
"color": severity_color,
"title": f"AdMob収益アラート [{analysis['severity'].upper()}]",
"text": analysis["summary"],
"fields": [
{"title": "昨日の収益", "value": f"${analysis['yesterday_revenue']:.2f}", "short": True},
{"title": "前日比", "value": f"{analysis['change_from_prev_day']:+.1f}%", "short": True},
{"title": "前週同曜日比", "value": f"{analysis['change_from_last_week']:+.1f}%", "short": True},
{"title": "考えられる原因", "value": "\n".join(f"• {c}" for c in analysis["possible_causes"]), "short": False}
],
"footer": "AdMob Anomaly Detector powered by Gemini API",
"ts": int(datetime.now().timestamp())
}]
}
requests.post(webhook_url, json=payload)
print("✅ Slack通知を送信しました")このコードをCloud RunかHerokuに乗せて、毎朝8時にCronで実行するように設定しています。朝の最初のコーヒーを飲む前に、収益が正常かどうかの通知が届く仕組みです。
この仕組みを導入してからは、収益の急落を最大4時間以内に検知できるようになりました。以前と比べて、対応の初動が格段に早くなっています。同じように個人でアプリを運営している方の参考になれば嬉しいです。
Pythonの非同期処理との組み合わせで複数アプリを並行監視する方法についても、別途まとめています。