収益化指標APIリファレンス

import requests

# 指定する必要がある値
client_id = "<クライアントID>"
client_secret = "<クライアントシークレット>"
package_name = "<アプリのパッケージ名>"

BASE_URL = "https://developer.amazon.com/api/appstore"

scope = "adx_reporting::appstore:marketer"
data = {
    "grant_type": "client_credentials",
    "client_id": client_id,
    "client_secret": client_secret,
    "scope": scope
}
auth_response = requests.post("https://api.amazon.com/auth/o2/token", data=data)
auth_token = auth_response.json()["access_token"]

headers = {
    "Authorization": f"Bearer {auth_token}",
    "Content-Type": "application/json"
}

freshness_url = f"{BASE_URL}/v2/applications/{package_name}/monetization/subsSignupMetricSet"
freshness_response = requests.get(freshness_url, headers=headers)
freshness = freshness_response.json()

for info in freshness["freshnessInfo"]:
    print(f"{info['aggregationPeriod']}: {info['latestDataTimestamp']}")

query_url = f"{BASE_URL}/v2/applications/{package_name}/monetization/subsSignupMetricSet:query"
body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": ["signups", "signupRevenue"]
}
response = requests.post(query_url, headers=headers, json=body)
result = response.json()

body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": ["signups", "signupRevenue"],
    "dimensions": ["subscriptionType", "marketplace"]
}
response = requests.post(query_url, headers=headers, json=body)
result = response.json()

body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": ["signups"],
    "filter": {"subscriptionType": ["FREE_TRIAL"], "marketplace": ["US"]}
}
response = requests.post(query_url, headers=headers, json=body)
result = response.json()

body = {
    "timelineSpec": {
        "aggregationPeriod": "HOURLY",
        "startTime": "2026-07-07T08:00:00Z",
        "endTime": "2026-07-07T11:00:00Z"
    },
    "metrics": ["signups"],
    "dimensions": ["subscriptionType"]
}

body = {
    "timelineSpec": {
        "aggregationPeriod": "HOURLY",
        "startTime": "2026-07-07T00:00:00Z",
        "endTime": "2026-07-07T23:00:00Z"
    },
    "metrics": ["signups"],
    "dimensions": ["subscriptionType"]
}
response = requests.post(query_url, headers=headers, json=body)
result = response.json()

fulfillment_url = f"{BASE_URL}/v2/applications/{package_name}/monetization/subsFulfillmentMetricSet:query"
body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": [
        "totalReceipts",
        "fulfilled",
        "unavailable",
        "pending",
        "p50AcknowledgementDelay",
        "p90AcknowledgementDelay",
        "p95AcknowledgementDelay"
    ],
    "filter": {"purchaseChannel": ["QS"]}
}
response = requests.post(fulfillment_url, headers=headers, json=body)
result = response.json()

health_url = f"{BASE_URL}/v2/applications/{package_name}/monetization/subsHealthMetricSet:query"
body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": ["autoRenewalOff", "autoRenewalOn", "gracePeriodEntries", "gracePeriodRecoveries"],
    "dimensions": ["subscriptionType"]
}
response = requests.post(health_url, headers=headers, json=body)
result = response.json()

survey_url = f"{BASE_URL}/v2/applications/{package_name}/monetization/subsCancellationSurveyMetricSet:query"
body = {
    "timelineSpec": {
        "aggregationPeriod": "DAILY",
        "startTime": "2026-07-01T00:00:00Z",
        "endTime": "2026-07-07T00:00:00Z"
    },
    "metrics": ["surveyedCancellations"],
    "dimensions": ["cancellationReason"]
}
response = requests.post(survey_url, headers=headers, json=body)
result = response.json()

all_rows = []
page_token = None
while True:
    body = {
        "timelineSpec": {
            "aggregationPeriod": "DAILY",
            "startTime": "2026-07-01T00:00:00Z",
            "endTime": "2026-07-07T00:00:00Z"
        },
        "metrics": ["signups"],
        "dimensions": ["subscriptionType", "marketplace"],
        "pageSize": 100
    }
    if page_token:
        body["pageToken"] = page_token
    response = requests.post(query_url, headers=headers, json=body)
    result = response.json()
    all_rows.extend(result["rows"])
    page_token = result.get("nextPageToken")
    if not page_token:
        break

import time

MAX_RETRIES = 3
RETRIABLE_CODES = {429, 500, 502, 503, 504}

def query_with_retry(url, headers、 体):
    for attempt in range(MAX_RETRIES):
        resp = requests.post(url, headers=headers, json=body)
        if resp.status_code == 200:
            return resp.json()
        if resp.status_code not in RETRIABLE_CODES:
            break  # 再試行不可のエラー
        wait = (2 ** attempt) + 0.5  # 1.5秒、2.5秒、4.5秒
        print(f"Retry {attempt + 1}/{MAX_RETRIES} after {wait}s (HTTP {resp.status_code})")
        time.sleep(wait)
    # 最後の試行は失敗しました
    error = resp.json() if resp.headers.get("content-type", "").startswith("application/json") else {}
    raise RuntimeError(f"Failed after {MAX_RETRIES} retries: {resp.status_code} {error.get('message', '')}")