Build a Blinkit price tracker in Python

A useful Blinkit price tracker is not a script that searches for a product every morning and saves whatever appears first.

If you collect data for an FMCG ecommerce team, a market-research firm or a product that shows price and availability, you need to answer a stricter question:

What happened to this exact product, at this exact delivery location, on this date?

That distinction matters because Blinkit data depends on location. As the Upscrape Blinkit API page explains, a pincode resolves to a representative delivery point and its serving merchants, not to one price or stock state for every address in the pincode. Search visibility is a separate signal again: a product can be missing from search results while an exact product check shows it in stock.

This guide builds a daily tracker around those constraints. The Python script records selling price, MRP, discount, stock state, inventory and merchant for a fixed list of Blinkit product IDs at the pincodes you choose. It saves every run to SQLite, writes the current run to CSV, and flags what changed since the previous day.

The examples track two global-brand protein products, Optimum Nutrition Gold Standard 100% Whey (907 g) and a Nature Valley protein bar pack, at HSR Layout (560102) and Whitefield (560066) in Bengaluru. Swap in your own products once the tracker works.

The tracker uses three Upscrape capabilities: blinkit.location to resolve the pincode, blinkit.product to read an exact product, and blinkit.search only when you first need to find product IDs. The Blinkit API reference lists each at 1 credit per request.

The row matters more than the scraper

Before writing the loop, decide what one observation means.

For price history, product_id and pincode are not enough on their own. You also want the coordinates the pincode resolved to, the merchants involved, and the time you collected the observation. Otherwise a merchant change can look like a price change with no explanation.

A practical row looks like this:

This is also why zero is a bad substitute for missing data. The blinkit.product reference describes its raw output as open-ended, with examples that are illustrative rather than a fixed schema. Its current sold-out example has is_sold_out: true and no price or MRP fields. A missing price is not ₹0.

Resolve the pincode before you read the product

You can send a pincode directly to any Blinkit capability, but for a tracker it helps to make location resolution explicit.

Call blinkit.location first. For a serviceable pincode it returns the representative latitude and longitude, the main merchant ID, the merchants Blinkit reports for that point, and address details. The blinkit.location reference warns that a pincode represents one point, not whole-area coverage. If Blinkit does not serve the point, or the pincode does not resolve, the request fails with the error code not_found instead of returning a result.

Then use the returned coordinates for every exact product check in that run. Each daily batch follows the same path:

pincode -> resolved delivery point -> exact product ID -> observation

It also gives you something to audit, because merchants change. These are the merchants Upscrape observed at the two pincodes in this guide:

At HSR Layout the main merchant changed between the two checks, at the same coordinates, and both products came from a third merchant, 42814. In Whitefield everything came from the point's main merchant. The tracker keeps both IDs, so a price change that coincides with a merchant change is visible as such.

For a business workflow, that is more useful than pretending "the price in 560102" is a single permanent attribute.

Discover product IDs once, then stop searching for identity

Search is useful when you set up the tracker. It should not decide which product you track every day.

Suppose you want to track Optimum Nutrition Gold Standard 100% Whey, Double Rich Chocolate, 907 g. Make a one-time blinkit.search request:

curl --request POST \
  --url https://data.upscrape.com/execute \
  --header "Authorization: Bearer YOUR_API_KEY" \
  --header "Content-Type: application/json" \
  --header "Prefer: wait=30" \
  --data '{
    "capability": "blinkit.search",
    "input": {
      "pincode": "560102",
      "query": "optimum nutrition",
      "limit": 12
    }
  }'

Each search result includes the product ID, name, pack size, price, MRP, inventory and merchant ID. When Upscrape ran this search, the first result was not the product we wanted:

Inspect the name and pack size, confirm the product you want, and put its ID into your tracked-product configuration. Here that is 512941, at ₹5,079 against an MRP of ₹6,319. The second product in this guide, Nature Valley Crunchy Oats & Honey Protein Bar Pack (5 x 42 g), was found the same way: product 475798.

From then on, use blinkit.product.

Do not write a daily tracker that searches for "optimum nutrition" and takes result number one. In the search above, that would have tracked a tub more than twice the size. Ranking changes, other pack sizes and flavours appear, and the product you want can drop out of the result set. Your time series should describe one product, not whichever result ranked highest that morning.

Build the daily Blinkit price tracker

Save the following as blinkit_tracker.py. It needs Python 3.9 or later and the requests package:

pip install requests
export UPSCRAPE_API_KEY="your_api_key"

Replace PINCODES and PRODUCTS with your own locations and confirmed product IDs.

The script handles both inline results and queued jobs. As the jobs reference describes, completed work returns HTTP 200, while HTTP 202 means the job is still running and should be polled. A job that fails while you wait returns a non-2xx status with an error code, and a failed job you poll returns HTTP 200 with state: "failed". The script checks the state and the error code in both cases (see errors and retries).

import csv
import os
import sqlite3
import time
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo

import requests

API = "https://data.upscrape.com"
KEY = os.environ["UPSCRAPE_API_KEY"]
DB = "blinkit_tracker.sqlite3"
CSV = "blinkit_latest.csv"
IST = ZoneInfo("Asia/Kolkata")

PINCODES = ["560102", "560066"]
PRODUCTS = [
    {
        "product_id": 512941,
        "label": "Optimum Nutrition Gold Standard Whey Double Rich Chocolate 907 g",
    },
    {
        "product_id": 475798,
        "label": "Nature Valley Crunchy Oats & Honey Protein Bar Pack 5 x 42 g",
    },
]

FIELDS = [
    "captured_at", "run_date", "pincode", "latitude", "longitude",
    "location_merchant_id", "product_id", "label", "name", "variant",
    "price", "mrp", "discount_pct", "stock_state", "inventory",
    "product_merchant_id", "run_status", "error", "change_flag", "changes",
]


class UpscrapeError(RuntimeError):
    def __init__(self, code, message):
        super().__init__(f"{code}: {message}")
        self.code = code


def raise_job_error(r):
    try:
        body = r.json()
    except ValueError:
        body = {}

    err = body.get("error") or {}
    raise UpscrapeError(
        err.get("code") or f"http_{r.status_code}",
        err.get("message") or r.text[:300],
    )


def execute(capability, input_data, deadline=90):
    headers = {
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
        "Prefer": "wait=30",
    }

    r = requests.post(
        f"{API}/execute",
        headers=headers,
        json={"capability": capability, "input": input_data},
        timeout=40,
    )

    started = time.monotonic()
    delays = [1, 2, 4, 8, 10]
    attempt = 0

    while True:
        if r.status_code == 200:
            body = r.json()
            if body.get("state") == "completed" and body.get("success") is True:
                results = body.get("results") or []
                if not results:
                    raise UpscrapeError("empty_result", "completed job had no results")
                return results[0].get("data") or {}
            # A polled job that failed arrives as HTTP 200 with state "failed".
            raise_job_error(r)

        if r.status_code != 202:
            # Rejected requests and jobs that failed during the wait.
            raise_job_error(r)

        job_id = r.json().get("job_id")
        if not job_id:
            raise UpscrapeError("missing_job_id", "pending response had no job_id")

        if time.monotonic() - started >= deadline:
            raise UpscrapeError(
                "local_timeout",
                f"job {job_id} still pending after {deadline}s; "
                "poll it later instead of resubmitting",
            )

        time.sleep(delays[min(attempt, len(delays) - 1)])
        attempt += 1

        r = requests.get(
            f"{API}/jobs/{job_id}",
            headers={"Authorization": f"Bearer {KEY}"},
            timeout=20,
        )


def init_db(conn):
    conn.execute("""
        CREATE TABLE IF NOT EXISTS observations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            captured_at TEXT NOT NULL,
            run_date TEXT NOT NULL,
            pincode TEXT NOT NULL,
            latitude REAL,
            longitude REAL,
            location_merchant_id TEXT,
            product_id INTEGER NOT NULL,
            label TEXT NOT NULL,
            name TEXT,
            variant TEXT,
            price INTEGER,
            mrp INTEGER,
            discount_pct REAL,
            stock_state TEXT NOT NULL,
            inventory INTEGER,
            product_merchant_id TEXT,
            run_status TEXT NOT NULL,
            error TEXT,
            change_flag INTEGER NOT NULL,
            changes TEXT
        )
    """)
    conn.commit()


def row_template(now, pincode, product, location=None):
    location = location or {}

    return {
        "captured_at": now.isoformat(timespec="seconds"),
        "run_date": now.date().isoformat(),
        "pincode": pincode,
        "latitude": location.get("latitude"),
        "longitude": location.get("longitude"),
        "location_merchant_id": location.get("merchant_id"),
        "product_id": product["product_id"],
        "label": product["label"],
        "name": None,
        "variant": None,
        "price": None,
        "mrp": None,
        "discount_pct": None,
        "stock_state": "unknown",
        "inventory": None,
        "product_merchant_id": None,
        "run_status": "error",
        "error": None,
        "change_flag": 0,
        "changes": None,
    }


def apply_product(row, product):
    price = product.get("price")
    mrp = product.get("mrp")
    inventory = product.get("inventory")

    # The response leaves out false and zero values, so only an explicit
    # sold-out flag or positive inventory counts as evidence.
    if product.get("is_sold_out") is True:
        stock = "sold_out"
    elif isinstance(inventory, int) and inventory > 0:
        stock = "in_stock"
    else:
        stock = "unknown"

    if (
        isinstance(price, (int, float))
        and isinstance(mrp, (int, float))
        and mrp > 0
    ):
        discount = round((mrp - price) / mrp * 100, 2)
    else:
        discount = None

    row.update({
        "name": product.get("name"),
        "variant": product.get("variant"),
        "price": price,
        "mrp": mrp,
        "discount_pct": discount,
        "stock_state": stock,
        "inventory": inventory,
        "product_merchant_id": product.get("merchant_id"),
        "run_status": "ok",
    })


COMPARED = [
    "run_status",
    "price",
    "mrp",
    "discount_pct",
    "stock_state",
    "location_merchant_id",
    "product_merchant_id",
]


def previous_day(conn, row):
    yesterday = (
        datetime.fromisoformat(row["run_date"]).date()
        - timedelta(days=1)
    ).isoformat()

    found = conn.execute(f"""
        SELECT {', '.join(COMPARED)}
        FROM observations
        WHERE pincode=? AND product_id=? AND run_date=?
        ORDER BY captured_at DESC
        LIMIT 1
    """, (
        row["pincode"],
        row["product_id"],
        yesterday,
    )).fetchone()

    if not found:
        return None

    return dict(zip(COMPARED, found))


def compare(previous, current):
    if previous is None:
        return 0, "no_baseline"

    changes = []

    if previous["run_status"] != current["run_status"]:
        changes.append(
            f"status {previous['run_status']} -> {current['run_status']}"
        )

    if previous["run_status"] == current["run_status"] == "ok":
        for field in ("price", "mrp", "discount_pct", "stock_state"):
            if previous.get(field) != current.get(field):
                changes.append(
                    f"{field} "
                    f"{previous.get(field)} -> {current.get(field)}"
                )

    for field in ("location_merchant_id", "product_merchant_id"):
        before = previous.get(field)
        after = current.get(field)

        if before and after and before != after:
            changes.append(f"{field} {before} -> {after}")

    return (
        int(bool(changes)),
        "; ".join(changes) if changes else "no_change",
    )


def collect():
    now = datetime.now(IST)
    rows = []

    for pincode in PINCODES:
        try:
            location = execute(
                "blinkit.location",
                {"pincode": pincode},
            )
        except Exception as exc:
            # not_found: Blinkit does not serve this point,
            # or the pincode did not resolve.
            if getattr(exc, "code", None) == "not_found":
                status = "location_not_found"
            else:
                status = "error"

            for product in PRODUCTS:
                row = row_template(now, pincode, product)
                row.update(run_status=status, error=f"location: {exc}")
                rows.append(row)
            continue

        lat = location.get("latitude")
        lon = location.get("longitude")

        if lat is None or lon is None:
            for product in PRODUCTS:
                row = row_template(
                    now,
                    pincode,
                    product,
                    location,
                )
                row["error"] = "location returned no coordinates"
                rows.append(row)
            continue

        for product in PRODUCTS:
            row = row_template(
                now,
                pincode,
                product,
                location,
            )

            try:
                data = execute(
                    "blinkit.product",
                    {
                        "product_id": product["product_id"],
                        "latitude": lat,
                        "longitude": lon,
                    },
                )

                apply_product(
                    row,
                    data.get("product") or {},
                )

            except Exception as exc:
                row["error"] = f"product: {exc}"

            rows.append(row)

    return rows


def main():
    rows = collect()

    with sqlite3.connect(DB) as conn:
        init_db(conn)

        for row in rows:
            row["change_flag"], row["changes"] = compare(
                previous_day(conn, row),
                row,
            )

            conn.execute(
                f"INSERT INTO observations "
                f"({', '.join(FIELDS)}) "
                f"VALUES ({', '.join('?' for _ in FIELDS)})",
                [row[field] for field in FIELDS],
            )

        conn.commit()

    with open(CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(
            f,
            fieldnames=FIELDS,
        )
        writer.writeheader()
        writer.writerows(rows)

    for row in rows:
        if row["changes"] != "no_change":
            print(
                row["pincode"],
                row["label"],
                row["run_status"],
                row["changes"],
            )


if __name__ == "__main__":
    main()

What the first run records

Given the Blinkit responses collected on 23 September 2026, the script writes these rows. The table shows a selection of the columns; discount_pct is 19.62 for the whey and 12.0 for the bar pack.

The prices match across the two pincodes, but the merchants and inventory do not. A tracker that stored only price would miss that the HSR Layout rows came from a different merchant than the point's main one.

What the script refuses to infer

The important part of the script is not the HTTP request. It is what the script refuses to infer.

A product becomes sold_out only when the exact response says so, and in_stock only when it reports positive inventory. The response leaves out false and zero values, so if neither signal is present, stock stays unknown.

Likewise, an API failure stays run_status=error. A pincode that Blinkit does not serve, or that does not resolve, becomes location_not_found. Neither is rewritten as "sold out", and failed requests are not charged.

If a job is still running when the script's 90-second deadline passes, the script records an error but does not cancel the job, which may still complete and be charged. Do not resubmit it in the same run; tomorrow's run is a new observation.

That separation keeps operational failures out of your availability metrics.

A missing search result is not a stockout

There is a concrete reason to keep search visibility apart from exact-product availability. At HSR Layout, Upscrape searched for protein bar with a limit of 50 and then checked the Nature Valley pack directly:

If a daily tracker had read "not in the protein bar results" as "out of stock", it would have written the wrong fact into its history.

The reverse matters too. A price field is not proof that the product can be bought right now. Price and availability belong in separate columns and should be read separately.

Search still has value. It tells you what a shopper sees for a query and where a product ranks. That is a useful dataset, but a different one from exact product availability.

Run it once a day

On a Linux host whose scheduler runs in India time, a cron entry runs the tracker every morning:

15 7 * * * cd /path/to/tracker && /path/to/.venv/bin/python blinkit_tracker.py >> tracker.log 2>&1

If the machine uses another timezone, set the scheduler's timezone explicitly instead of assuming server time is IST.

Each run creates or updates two files:

• blinkit_tracker.sqlite3 is the append-only history used for day-over-day comparison.

• blinkit_latest.csv holds only the current batch, ready to hand to an analyst or load into another system.

On the first day every row gets changes=no_baseline. On later days the script prints flags such as price 330 -> 315, stock_state in_stock -> sold_out or location_merchant_id 40589 -> 36475, the real change at HSR Layout described above.

The script compares with the previous calendar day, not the most recent successful response. If yesterday failed, today's row should say so. Otherwise a three-day gap could pass for an ordinary one-day comparison.

What does a daily Blinkit price tracker cost?

Each blinkit.location request and each blinkit.product request is published at 1 credit. If you track N products at M pincodes once a day, a normal run costs:

That is about 180 credits a month for this guide's basket.

The location calls are worth keeping because they record the delivery point and merchant you are comparing.

Failed requests, including a pincode Blinkit does not serve, do not consume credits under the current public pricing rules. Credit pack prices and terms can change, so check the Upscrape pricing page rather than hard-coding a rupee or dollar cost into the tracker.

If you also run blinkit.search every day for search-rank analysis, budget those calls separately. The fixed-product history built here does not need them.

Turn the snapshots into Blinkit price history

Upscrape's Blinkit endpoints return current observations. They do not return a ready-made historical price series. Blinkit price history comes from scheduling repeated observations and keeping them yourself.

After seven daily runs, inspect the history directly:

SELECT
    run_date,
    pincode,
    product_id,
    label,
    price,
    mrp,
    discount_pct,
    stock_state,
    run_status,
    changes
FROM observations
WHERE run_date >= date('now', '+330 minutes', '-6 days')
ORDER BY product_id, pincode, run_date;

date('now') in SQLite is UTC, while run_date is an India date, so the query shifts by 330 minutes (five and a half hours) before counting back six days.

For analysis, group by product and pincode together, not by product alone. A useful seven-day price chart draws one series per product and location. Keep stock events visible beside the price series instead of turning missing prices into zero: a sold-out product with no returned price did not suddenly cost ₹0, and neither did a failed request.

For an ecommerce account team, the history then answers questions such as whether the MRP changed or only the selling price, whether a discount appeared only at some delivery points, whether a stockout was limited to one location, and whether a price move coincided with a different merchant.

SQLite is enough for a small fixed basket. At higher volume the same row structure moves into Postgres, a warehouse or your existing pipeline without changing the identity rules.

Three rules keep the history usable

1. Resolve location before collecting the product. The pincode is the business geography you asked about; the returned coordinates and merchants describe the observation you actually made. Keep both.

2. Track exact product IDs. Use search to find IDs and confirm names and pack sizes, then fix the IDs in configuration. Search order is not product identity.

3. Preserve uncertainty. sold_out, unknown, error and location_not_found mean different things. Merging them makes a dashboard simpler and the history less trustworthy.

These rules matter more as the dataset grows. Once hundreds of daily snapshots feed reports or customer-facing features, fixing identity or state rules after the fact is much harder than getting them right in the first run.

A Blinkit price tracker is less about fetching a number repeatedly and more about asking the same well-defined question every day: what did Blinkit return for this exact product, at this resolved delivery point, at this time?