Keepa Python Client

Keepa requests consume tokens. By default, the client will wait for tokens to become available before proceeding.

• Default behavior: The client waits for tokens.
• Manual management: Set wait=False only if your application manages token availability itself.

For large batches, it is recommended to keep wait=True and monitor your usage by inspecting api.tokens_left or api.status between calls.

The default response format is a dictionary. You can use the typed=True option to receive generated Pydantic Product models instead. Note that typed=True and raw=True cannot be used together, and the asynchronous client does not support raw=True.

# Using Pydantic models via typed=True
product = api.query("B0088PUEPK", history=False, typed=True)[0]
print(product.asin, product.title)

The liveOffersOrder field contains indices that point to the currently active offers within the offers list. Not all historical offers are active, and this field may be absent from the response. To safely access active offers, iterate through the indices in liveOffersOrder and check them against the length of the offers list.

active_offers = [
    offers[index]
    for index in product.get("liveOffersOrder", [])
    if index < len(offers)
]

for offer in active_offers:
    times, prices = keepa.convert_offer_history(offer["offerCSV"])

When using ProductParams, use the exact field names documented by Keepa.

• Range filters: Use suffixes like _gte (greater than or equal) and _lte (less than or equal).
• List-like filters: Accept either a single string or a list of strings (e.g., buyBoxSellerId, partNumber, categories_include).
• Sorting: Use the sort key with [field, direction] pairs.

request = keepa.ProductParams(
    buyBoxSellerId=["A2L77EE7U53NWQ", "ATVPDKIKX0DER"],
    partNumber=["MX-1000", "MX-1001"],
    current_SALES_lte=50000,
    sort=[["current_SALES", "desc"]],
)
asins = api.product_finder(request)

The keepa.Keepa class provides a synchronous interface for interacting with the Keepa API. When using this client, all API calls are blocking; the execution will wait until the API response is received or the configured timeout is reached before proceeding.

import keepa

# Initialize the synchronous client
client = keepa.Keepa(api_key='YOUR_API_KEY')

The keepa library provides two distinct client types depending on your application's architecture:

• sync_client: Use this for regular, synchronous Python applications.
• async_client: Use this when your requests must integrate with an asyncio event loop.

Both clients expose the same set of Keepa endpoints, ensuring feature parity between synchronous and asynchronous workflows.

When using typed=True, the library still performs its standard history and statistics parsing. These parsed values are attached as extra attributes to the model rather than being part of the raw backend schema:

• History: If history=True is passed to the query, parsed NumPy arrays are available via product.data.
• Statistics: If requested statistics are available, parsed values are available via product.stats_parsed.

Because these are client-generated conveniences, use getattr() or check for existence to safely access them.

products = api.query("B0088PUEPK", history=True, stats=90, typed=True)
product = products[0]

# Accessing parsed history
history = getattr(product, "data", {})
new_prices = history.get("NEW")

# Accessing parsed statistics
stats = getattr(product, "stats_parsed", {})
current_amazon_price = stats.get("current", {}).get("AMAZON")

Deal arrays like current, delta, deltaPercent, and deltaLast are indexed by Keepa's product CSV type order rather than names. To correctly access a specific price type (e.g., "NEW"), use keepa.csv_indices to find the corresponding integer index.

Important: Keepa uses sentinel values like -1 for unavailable prices. Always filter these values before performing arithmetic operations to avoid incorrect calculations.

deal = response.dr[0]
# Map the price type name to its array position
csv_index = next(index for index, name, _ in keepa.csv_indices if name == "NEW")

current_new = deal.current[csv_index]
change_from_last = deal.deltaLast[csv_index]

# Example calculation of a previous value
if current_new != -1 and change_from_last != -1:
    previous_new = current_new - change_from_last

To request current offer stock data, set stock=True in your api.query call alongside the offers parameter.

Important details:

• Keepa only collects stock for live offers.
• The backend caps reported offer stock at 10 units.
• Setting stock=True asks Keepa to refresh stock for current live offers and consumes additional tokens.
• While stockCSV history might exist in other queries, stock=True is required to trigger a refresh.

Stock data can be found in offer.get("stockCSV") (history) or offer.get("stock") (latest value).

product = api.query("1454857935", offers=20, stock=True)[0]

for index in product.get("liveOffersOrder", []):
    offer = product["offers"][index]
    stock_history = offer.get("stockCSV")
    latest_stock = offer.get("stock")

Install the keepa library using pip to access Keepa's Amazon product, price-history, offer, seller, category, best-seller, product-finder, and deals APIs.

pip install keepa

The timeout parameter in Keepa(accesskey, timeout=...) controls how long the client waits for the backend to send a single response; it is not a total runtime limit for a large batch.

Optimizing requests: Token cost is calculated by Keepa and can exceed one token per ASIN if certain options are enabled. If you experience timeouts or high token consumption, consider the following:

1. Reduce expensive options: Options like offers, stock, buybox, rating, stats, or forced updates increase token cost.
2. Adjust timeout: Increase the timeout value only if the backend legitimately needs more time to produce the requested data.
3. Chunking: The client automatically chunks product queries into backend-sized requests, but for debugging large batches, use smaller chunks.