Building a Price Monitoring Pipeline for Non-Amazon Retailers: The General Architecture

Amazon scraping gets all the attention — the tutorials, the tools, the war stories. But most price-monitoring projects I've built have nothing to do with Amazon. They're about the long tail of retailers: electronics vendors, specialty shops, regional e-commerce sites, B2B part suppliers. And that long tail is a fundamentally different engineering problem.

On Amazon, one scraper architecture works for millions of products because the site is one site. In the general case, you're monitoring 50–500 different retailers, each with its own markup, its own anti-bot posture, its own currency, its own notion of what "in stock" means. The thesis of this post: in general price monitoring, the scraper is the easy 20% — the pipeline is the 80% that determines whether the product is useful. Getting a price is trivial. Getting a comparable, deduplicated, change-detected, alerted stream of prices across hundreds of heterogeneous stores is the actual job.

Here's the architecture that has worked for me, layer by layer.

Layer 0: The Source Registry — Your Single Source of Truth

Before writing any scraper, model the domain. Every monitored store and product goes into a registry:

from dataclasses import dataclass, field

@dataclass
class Store:
    store_id: str
    domain: str
    currency: str
    country: str                  # for geo-pinned fetching
    extractor: str                # which parser module handles it
    politeness: dict = field(default_factory=lambda: {
        "min_interval_s": 8,
        "burst": 3,
    })

@dataclass
class MonitoredProduct:
    sku: str                      # YOUR canonical product id
    canonical_name: str
    urls: dict                    # store_id -> product URL on that store

# registry entry example
REGISTRY = [
    MonitoredProduct(
        sku="LOGI-MX3-WIRELESS",
        canonical_name="Logitech MX Master 3 Wireless Mouse",
        urls={
            "store_a": "https://store-a.example.com/mx3",
            "store_b": "https://www.store-b.co.uk/logitech-mx-master-3",
        },
    ),
]

Two things in this schema matter enormously later. First, the canonical SKU is yours, not any store's — because store product IDs change, listings get delisted, and you need a stable join key. Second, politeness is per-store. A 500-store monitor is 500 independent conversations, each with its own pacing.

Layer 1: Fetching — One Job, Many Personalities

Each store gets its fetch profile from the registry: HTTP client vs. browser, geo, session policy. A dispatcher keeps it clean:

import time, random, requests

class Fetcher:
    def __init__(self, store: Store):
        self.store = store
        self.session = requests.Session()
        self.session.proxies = {
            "http": "http://USER-cc-{}:[e-mail]:9000".format(
                store.country),
            "https": "http://USER-cc-{}:[e-mail]:9000".format(
                store.country),
        }
        self._last_hit = 0.0

    def get(self, url):
        # per-store politeness: the store's server sets the rhythm
        wait = self.store.politeness["min_interval_s"]
        elapsed = time.time() - self._last_hit
        if elapsed < wait:
            time.sleep(wait - elapsed + random.uniform(0, 2))
        self._last_hit = time.time()
        r = self.session.get(url, timeout=30)
        r.raise_for_status()
        return r

Note the country-pinned proxy per store: a UK retailer should be fetched from a UK exit IP, both for correctness (geo-pricing!) and for believability. Also note what's absent: any CAPTCHA-solving. If a store serves challenges, that's a signal your pacing or fingerprint is wrong — fix the cause, don't pay to treat the symptom.

The HTTP-vs-browser decision is per-store. My rule of thumb: if the price is in the initial HTML, use requests; if it requires JavaScript, that store's extractor gets escalated to a Playwright worker. A mature pipeline is always a mix.

Layer 2: Extraction — Per-Store Selectors, Common Output Contract

Every store gets its own extractor module, but they all emit the same record:

@dataclass
class PriceObservation:
    sku: str
    store_id: str
    observed_at: str          # ISO timestamp
    price: float | None       # None = listed but no price visible
    currency: str
    in_stock: bool
    raw_text: str             # the exact scraped string, for auditing

def extract_store_a(html: str, sku: str) -> PriceObservation:
    from parsel import Selector
    sel = Selector(html)
    raw = sel.css("span.price__current::text").get()
    price = float(raw.replace("$", "").replace(",", "").strip()) if raw else None
    stock = bool(sel.css("button.add-to-cart:not([disabled])"))
    return PriceObservation(
        sku=sku, store_id="store_a",
        observed_at=_now_iso(),
        price=price, currency="USD", in_stock=stock,
        raw_text=(raw or "").strip(),
    )

Two hard-won details here. Keep raw_text: when a selector breaks (and selectors always break — stores redesign quarterly), the raw string is what lets you diagnose whether the site changed or your extractor rotted. Make price=None explicit: "we visited the page and found no price" is different information from "we didn't visit" and from "out of stock." Conflating them corrupts your history.

Selector maintenance is the real ongoing cost of this business. Budget for it: I run a nightly job that flags any store whose extraction success rate dropped below 95%, because a silent selector break produces confidently wrong data — the worst kind.

Layer 3: Normalization — Where Comparisons Are Actually Won

Now the layer most tutorials skip: making observations comparable.

FX = {"GBP": 1.27, "EUR": 1.08, "USD": 1.0}   # refresh from an API in prod

def normalize(obs: PriceObservation, stores: dict) -> dict:
    store = stores[obs.store_id]
    usd = round(obs.price * FX[obs.currency], 2) if obs.price else None
    return {
        "sku": obs.sku,
        "store": obs.store_id,
        "price_usd": usd,
        "in_stock": obs.in_stock,
        "t": obs.observed_at,
        # flags that change the comparison semantics:
        "effective_price": usd,
        "gift_card_only": "giftcard" in obs.raw_text.lower(),
    }

The normalization questions that actually bite: currency conversion (obvious), tax-inclusive vs. tax-exclusive pricing (EU sites include VAT, US sites don't — a 20% distortion if you miss it), per-store shipping thresholds (a "$5 higher" price with free shipping often wins), and weird payment restrictions ("gift card only" listings are not real market prices). You don't have to solve all of these on day one — but your schema needs the slots for them, or you'll be rewriting history later.

Layer 4: Change Detection — Don't Alert on Noise

Here's the mistake I made in my first price monitor: storing every observation and alerting on every delta. Prices on retail sites are noisy — A/B tests, regional experiments, temporary cart discounts, and plain flapping mean the same product can oscillate between two values all day. Alert on every change and your pipeline gets muted within a week.

The fix is to treat observations as a signal and detect sustained changes:

from statistics import median

class ChangeDetector:
    """Alerts only when the median of recent windows truly shifts."""
    def __init__(self, window=5, threshold_pct=2.0):
        self.history = {}                     # (sku, store) -> [prices]
        self.window = window
        self.threshold = threshold_pct

    def observe(self, key, price) -> dict | None:
        if price is None:
            return None
        h = self.history.setdefault(key, [])
        h.append(price)
        h[:] = h[-self.window * 2:]
        if len(h) < self.window:
            return None
        old = median(h[:-self.window]) if len(h) >= self.window * 2 else None
        new = median(h[-self.window:])
        if old and abs(new - old) / old * 100 >= self.threshold:
            return {"key": key, "old": old, "new": new}
        return None

Median-based windows filter single-fetch flapping; the percentage threshold filters penny-changes. Tune window to your polling cadence — I poll volatile stores hourly and stable ones daily, so the window represents roughly the same wall-clock duration everywhere.

Layer 5: The Best-Price View

With clean, normalized, change-detected data, the product's payoff is one query:

def best_offer(observations, stores):
    live = [o for o in observations if o["in_stock"] and o["price_usd"]]
    return min(live, key=lambda o: o["effective_price"]) if live else None

Which "best price" means for your use case — cheapest overall, cheapest from a reputable store, cheapest with shipping included — is a product decision, not a technical one. The architecture's job is to make any of those queries cheap once the data is disciplined.

Scheduling: Freshness Per Store, Not Global Cadence

One last piece of real engineering: don't run the monitor on one global cron. Group stores by volatility. Flash-sale electronics retailers need hourly checks; industrial B2B suppliers change prices monthly — hitting them hourly is waste (and risk). A per-store poll_class in the registry, with a scheduler that fans out accordingly, cuts your fetch volume by 60–80% in practice and keeps the data just as fresh where it matters.

Wrapping Up

The pipeline that survives is boring in the right places: a registry that owns the domain model, per-store politeness and geo, extractors with a common output contract, normalization that makes prices comparable, change detection that respects noise, and per-store freshness. The scraper is one small module in that machine.

Disclosure: I use Thordata's residential proxies for the geo-pinned fetching in this pipeline — country-level targeting per store is what keeps geo-prices accurate. Check them out at thordata.com, and use code **thor020* for 10% off.*

Comments

[Luis Cruz]

Your approach to modeling the source registry is a solid foundation for managing the complexities of different retailers. I particularly appreciate how you've emphasized the importance of a canonical SKU and store-specific politeness, which is crucial for scraping reliability and ethical considerations. It might be worth exploring how a graph database could enhance your registry's flexibility, especially as the number of monitored retailers grows. If you’re looking for help refining this pipeline or implementing additional features, I’d be glad to discuss a paid collaboration. What challenges have you encountered with scaling the fetching layer as the number of stores increases?