Keeper APIで一括ASIN分析——効率的な仕入れリスト作成

Keepaとは——Amazon価格履歴の宝庫

Keepa（キーパ）はAmazonの価格・ランキング・在庫状況を長期間記録している価格追跡サービスです。個人・法人向けにAPIを提供しており、大量のASINについて詳細な価格履歴・ランキング変動・セール情報を取得できます。

仕入れビジネス・EC分析において、Keepa APIは単なる「今の価格」ではなく「価格の歴史」を取得できる点で非常に強力です。

Keepa APIで取得できるデータ

APIの基本実装（Python）

import requests
import time
from datetime import datetime, timedelta
import os

KEEPA_API_KEY = os.environ['KEEPA_API_KEY']
KEEPA_BASE_URL = 'https://api.keepa.com'

def keepa_request(endpoint: str, params: dict) -> dict:
    """Keepa APIにリクエストを送る"""
    params['key'] = KEEPA_API_KEY

    response = requests.get(
        f"{KEEPA_BASE_URL}/{endpoint}",
        params=params,
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()

    # レート制限の確認
    tokens_left = data.get('tokensLeft', 0)
    if tokens_left < 100:
        print(f"⚠️ トークン残量: {tokens_left}")

    return data

ASINの詳細情報を取得

def get_product_data(asins: list[str], domain: int = 5) -> list[dict]:
    """
    複数ASINの商品データを取得
    domain: 1=com, 5=co.jp
    """
    # Keepaは1リクエストで最大100 ASINまで
    results = []

    for i in range(0, len(asins), 100):
        batch = asins[i:i + 100]
        data = keepa_request('product', {
            'asin':   ','.join(batch),
            'domain': domain,
            'stats':  1,     # 統計情報を含む
            'offers': 20,    # マーケットプレイス出品を20件取得
            'history': 1,    # 価格履歴を含む
        })

        results.extend(data.get('products', []))
        time.sleep(1)  # レート制限対策

    return results

Keepaのタイムスタンプ変換

Keepaは時刻を独自形式（Keepa Minutes）で返します。

def keepa_minutes_to_datetime(keepa_minutes: int) -> datetime:
    """Keepa Minutesを datetimeに変換"""
    # Keepa Minutesは2011年1月1日00:00 UTC からの分数
    epoch = datetime(2011, 1, 1, 0, 0, 0)
    return epoch + timedelta(minutes=keepa_minutes)

def parse_price_history(csv_data: list, price_type: int = 0) -> list[dict]:
    """
    価格履歴CSVをパース
    price_type: 0=Amazon, 1=Marketplace新品, 2=Marketplace中古
    """
    if not csv_data or len(csv_data) <= price_type:
        return []

    raw = csv_data[price_type]
    if not raw:
        return []

    history = []
    for i in range(0, len(raw) - 1, 2):
        timestamp = raw[i]
        price     = raw[i + 1]

        if price == -1:
            continue  # 在庫なし

        history.append({
            'datetime': keepa_minutes_to_datetime(timestamp),
            'price':    price,  # 単位: セント（日本円の場合は100で割る）
        })

    return history

価格分析レポートの生成

import pandas as pd
import numpy as np

def analyze_asin(asin: str) -> dict:
    """ASINの価格分析を行う"""
    products = get_product_data([asin])
    if not products:
        return {}

    product = products[0]
    csv = product.get('csv', [])

    # Amazon価格の履歴
    amazon_history = parse_price_history(csv, 0)
    df = pd.DataFrame(amazon_history)

    if df.empty:
        return {'asin': asin, 'error': '価格履歴なし'}

    df['price_jpy'] = df['price'] / 100  # センントを円に変換（JPの場合はそのまま）
    df = df.set_index('datetime').sort_index()

    # 直近30日のデータ
    last_30 = df.last('30D')

    # 統計
    stats = product.get('stats', {})
    current_price = stats.get('current', [None, None, None])[0]

    return {
        'asin':           asin,
        'title':          product.get('title', ''),
        'current_price':  current_price / 100 if current_price and current_price != -1 else None,
        'avg_30d':        round(last_30['price_jpy'].mean(), 0),
        'min_30d':        int(last_30['price_jpy'].min()),
        'max_30d':        int(last_30['price_jpy'].max()),
        'price_stability': round(last_30['price_jpy'].std() / last_30['price_jpy'].mean() * 100, 1),
        'sales_rank':     product.get('salesRanks', {}).get('91', [])[-1:],
    }

仕入れリストの一括分析

def analyze_purchase_list(asins: list[str], purchase_costs: dict) -> pd.DataFrame:
    """仕入れ候補リストを一括分析"""
    products = get_product_data(asins)
    records  = []

    for product in products:
        asin     = product.get('asin', '')
        analysis = analyze_asin(asin)

        cost = purchase_costs.get(asin, 0)
        if cost and analysis.get('avg_30d'):
            margin = (analysis['avg_30d'] - cost) / analysis['avg_30d'] * 100
        else:
            margin = None

        records.append({
            'ASIN':    asin,
            '商品名':  (product.get('title', '') or '')[:40],
            '仕入れ値': cost,
            '平均価格': analysis.get('avg_30d'),
            '最安値':  analysis.get('min_30d'),
            '最高値':  analysis.get('max_30d'),
            '価格安定性': analysis.get('price_stability'),
            '推定利益率': round(margin, 1) if margin else None,
        })

    df = pd.DataFrame(records)
    df = df.sort_values('推定利益率', ascending=False)
    return df

まとめ

Keepa APIはAmazonのビジネスに必要な「価格の歴史」を提供する強力なツールです。仕入れ判断・価格設定・在庫管理に活用することで、データドリブンなEC運営が実現できます。弊社ではKeepa APIを組み込んだ仕入れ判断支援ツールの開発実績があります。

EC自動化・データ分析ツールの開発についてはお気軽にご相談ください。