Competitor price monitoring with Python

Disclosure: This article was prepared for Portproof, which sells shared residential and mobile proxy traffic by the GB. The worked example uses fictional offers served locally; no retailer or proxy service was tested. AI assisted the drafting and technical review. Recorded fixture results are dated below; they do not represent a live retailer test.

A price monitor can fetch a page successfully and still record the wrong price. It may have received a different pack size, another seller's offer, an unavailable item or a cookie-preference page. HTTP 200 describes successful HTTP processing; your collector must decide whether the response contains a comparable offer.

Start with that decision before choosing a browser, proxy route or collection schedule.

Define what counts as the same offer

For each monitored product, specify the SKU, variant, pack quantity, seller, currency, delivery destination and tax treatment. Decide how promotions and membership conditions affect eligibility. A headline price for a smaller pack or a members-only discount should not silently replace an ordinary offer.

In this example, an accepted offer must describe fictional SKU EXAMPLE-500, orange variant, a 500 ml pack, seller fictional-test-shop, EUR currency, destination TEST-POSTCODE, tax included and InStock availability. Both item price and shipping must be present. The comparable value is their sum.

This is a deliberately small normalised JSON format. It does not extract prices from arbitrary retail HTML. A real implementation still needs permission to use its source, a parser for that source and explicit rules for its offer conditions.

Test plausible failures locally

Copy the complete local fixture at the end of this article into local_price_observation_fixture.py. It runs with Python 3.9 or later and only the standard library:

python3 local_price_observation_fixture.py

It opens a temporary HTTP listener on 127.0.0.1, makes one request per case and closes the listener. Inherited proxy settings are explicitly disabled. It uses no browser, external shop, account or credential, and performs no retries.

The recorded run on 5 October 2026 produced:

Response group	Responses	Accepted offers
Correct fictional offers	2	2
Wrong SKU, variant or currency	3	0
Missing item price, shipping or stock	3	0
Cookie-preference HTML, with HTTP 200	1	0
HTTP 429 and HTTP 403	2	0
Total	11	2

Nine responses return HTTP 200. Only two satisfy the offer requirements. These cases were selected to exercise failures; their proportion is not a real-shop success rate.

Treat missing amounts as missing

The validator checks status, content type, document shape, offer identity and stock before calculating a total. For this EUR example, amounts must be non-negative strings with exactly two fractional digits and at most ten whole-number digits. Other currencies require their own rules.

This excerpt is the fixture's amount parser, not the complete validator:

import re
from decimal import Decimal

def amount(value):
    if not isinstance(value, str) or not re.fullmatch(
        r"(?:0|[1-9][0-9]{0,9})\.[0-9]{2}", value
    ):
        raise ValueError("missing-or-invalid-amount")
    result = Decimal(value)
    if not result.is_finite() or result < 0:
        raise ValueError("missing-or-invalid-amount")
    return result

An explicit shipping value of "0.00" is accepted. Missing shipping is rejected rather than treated as free delivery. Boolean values, negative amounts and non-finite or exponent-formatted values are rejected. Python's decimal arithmetic preserves decimal values without binary floating-point rounding in this calculation.

That distinction changes the result:

Observation	Item	Shipping	Delivered total
First accepted offer	€24.00	€4.00	€28.00
Next accepted offer	€25.00	€0.00	€25.00

The item price rose by €1, while the comparable delivered total fell by €3. Keep the components as well as the total so an alert can explain why it changed.

Use a browser when the source requires it

An authorised feed or HTTP response may already contain the necessary fields. If a permitted source renders them through JavaScript, use a browser and validate its extracted data with the same acceptance rules. The Puppeteer proxy setup guide covers routing and a bounded connection check separately.

domcontentloaded does not establish that a delayed offer is ready. Wait for the required source-specific state with a time limit, then check identity and values. Preserve cookies and market settings deliberately. A browser context, a provider's sticky session and an HTTP connection are different things; changing a route does not reset storefront cookies or guarantee a new connection for every request.

Keep proxy credentials restricted to recognised proxy authentication challenges. Do not let an arbitrary shop's authentication challenge receive them. The Python measurement above exercises none of this browser behaviour.

Budget for accepted observations

The fixture reads 1,718 response-body bytes across all eleven cases. Dividing by two accepted offers gives 859 body bytes per accepted offer. Counting only the two valid responses would discard the cost of rejected attempts.

Those local body bytes exclude headers, TLS, browser subresources and provider metering. They are not a usable real-shop bandwidth forecast. Run a separate permitted sample using the intended collection method, count failed and retried traffic too, and reconcile the counter with the provider's usage meter. Put that sample into the price-monitoring traffic calculator, keeping different markets or workflows separate when their costs differ.

Bound simultaneous tasks, response sizes and request durations. Record HTTP 429 instructions; the fixture's Retry-After: 60 is recorded without retrying. The 429 specification describes rate limiting and possible retry instructions. Stop refused requests and investigate permissions rather than changing routes to repeat them.

Choose a route after establishing source permission, market requirements and session lifetime. The price-monitoring use case explains HTTP versus browser workflows and shared mobile versus residential pools. A route choice cannot certify the resulting currency, delivery context or offer.

Store the source URL, collection time, extraction version and comparison context with every accepted observation. Keep rejected attempts with a reason and no invented price. Then alerts and cost estimates can use the same definition of a valid observation.

Complete local fixture

Save this entire block as local_price_observation_fixture.py, then run the command above. The script needs permission to open a temporary listener on 127.0.0.1. It makes no external shop requests, requires no proxy credentials, and disables inherited proxy settings for its local requests.

#!/usr/bin/env python3
"""Synthetic local observations: no live proxy, shop, credentials or retry."""
import copy
import json
import re
import threading
from decimal import Decimal, InvalidOperation
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.error import HTTPError
from urllib.request import ProxyHandler, build_opener

BASE = {"sku": "EXAMPLE-500", "variant": "orange", "pack_ml": 500,
        "currency": "EUR", "seller": "fictional-test-shop",
        "destination": "TEST-POSTCODE", "tax": "included",
        "availability": "InStock", "price": "24.00", "shipping": "4.00"}
CASES = []
def case(name, changes, status=200):
    offer = copy.deepcopy(BASE)
    offer.update(changes)
    CASES.append((name, status, json.dumps(offer, separators=(",", ":")).encode()))
case("valid-first", {})
case("valid-next", {"price": "25.00", "shipping": "0.00"})
case("wrong-sku", {"sku": "EXAMPLE-999"})
case("wrong-variant", {"variant": "apple"})
case("wrong-currency", {"currency": "USD"})
case("missing-price", {"price": None})
case("missing-shipping", {"shipping": None})
case("missing-stock", {"availability": None})
CASES.append(("consent-page", 200, b"<html>Choose your cookie preferences</html>"))
CASES.append(("rate-limit", 429, b"slow down"))
CASES.append(("refusal", 403, b"not permitted"))

class Fixture(BaseHTTPRequestHandler):
    def do_GET(self):
        name = self.path.removeprefix("/")
        chosen = next((row for row in CASES if row[0] == name), None)
        if chosen is None:
            self.send_error(404)
            return
        _, status, body = chosen
        self.send_response(status)
        self.send_header("Content-Type", "text/html" if name == "consent-page" else "application/json")
        self.send_header("Content-Length", str(len(body)))
        if status == 429:
            self.send_header("Retry-After", "60")
        self.end_headers()
        self.wfile.write(body)
    def log_message(self, *args):
        pass

def amount(value):
    if not isinstance(value, str) or not re.fullmatch(r"(?:0|[1-9][0-9]{0,9})\.[0-9]{2}", value):
        raise ValueError("missing-or-invalid-amount")
    result = Decimal(value)
    if not result.is_finite() or result < 0:
        raise ValueError("missing-or-invalid-amount")
    return result

def validate(status, content_type, body):
    if status != 200:
        return None, "http-" + str(status)
    if content_type != "application/json":
        return None, "unexpected-content-type"
    try:
        row = json.loads(body)
        if not isinstance(row, dict):
            return None, "unexpected-document"
        for key in ("sku", "variant", "pack_ml", "currency", "seller", "destination", "tax"):
            if row.get(key) != BASE[key]:
                return None, "mismatch-" + key
        if row.get("availability") != "InStock":
            return None, "missing-or-unavailable-stock"
        total = amount(row.get("price")) + amount(row.get("shipping"))
        return str(total), None
    except (ValueError, InvalidOperation, TypeError):
        return None, "missing-or-invalid-amount"

def main():
    rows = []
    # ProxyHandler({}) avoids inherited proxy settings for this loopback-only fixture.
    opener = build_opener(ProxyHandler({}))
    with HTTPServer(("127.0.0.1", 0), Fixture) as server:
        worker = threading.Thread(target=server.serve_forever, daemon=True)
        worker.start()
        try:
            for name, _, _ in CASES:
                target = "http://127.0.0.1:" + str(server.server_port) + "/" + name
                try:
                    response = opener.open(target, timeout=2)
                except HTTPError as error:
                    response = error
                with response:
                    body = response.read(4097)
                    if len(body) > 4096:
                        raise RuntimeError("fixture body exceeded bound")
                    status = response.status
                    total, reason = validate(status, response.headers.get_content_type(), body)
                    rows.append({"case": name, "status": status, "body_bytes": len(body),
                                 "valid": reason is None, "delivered_total": total, "reason": reason,
                                 "retry_after": response.headers.get("Retry-After")})
        finally:
            server.shutdown()
            worker.join(timeout=3)
    valid = sum(row["valid"] for row in rows)
    body_bytes = sum(row["body_bytes"] for row in rows)
    summary = {"scope": "synthetic loopback fixture; response body bytes only, not billed traffic",
               "requests": len(rows), "http_200": sum(row["status"] == 200 for row in rows),
               "valid_observations": valid, "response_body_bytes": body_bytes,
               "body_bytes_per_valid_observation": body_bytes / valid,
               "rows": rows}
    assert (summary["requests"], summary["http_200"], valid) == (11, 9, 2)
    assert [row["delivered_total"] for row in rows[:2]] == ["28.00", "25.00"]
    assert rows[9]["retry_after"] == "60"
    print(json.dumps(summary, indent=2))

if __name__ == "__main__":
    main()