Building a real-time price monitoring pipeline that does not poison your data

A common failure mode in pricing systems is not that the scraper stops. It is that it keeps running and starts returning believable garbage: a price from the wrong variant, a sale price without the coupon condition, or a placeholder value from a partially rendered page.

If that bad data flows straight into a pricing engine, you do not have real-time intelligence. You have an automated way to make bad decisions faster.

Real-time data is mostly a freshness and trust problem

Teams usually talk about real-time web data as if the hard part is speed. Speed matters, but freshness without validation is dangerous.

For a competitor price monitor, you usually need to know four things:

• What product did we observe?
• What price did we observe?
• When did we observe it?
• How confident are we that the observation is correct?

That last field is the one many systems skip.

A simple pipeline might look like this:

1. Fetch competitor product pages or APIs
2. Extract price, availability, and metadata
3. Normalize currencies, units, and variants
4. Validate the observation against rules and history
5. Store the raw response and normalized record
6. Publish accepted changes to pricing or analytics systems

The important part is that extraction and decision-making stay separate. A scraper should not decide that your company should match a price. It should produce an observation with enough context for another service to evaluate it.

If your bottleneck is reliable extraction across changing product pages, Wire can sit in the extraction layer where teams need current competitor prices, availability signals, and failure visibility rather than another pricing rule engine.

Store observations, not just current state

A lot of teams start with a table like this:

CREATE TABLE competitor_prices (
  competitor TEXT NOT NULL,
  sku TEXT NOT NULL,
  price_cents INTEGER NOT NULL,
  currency TEXT NOT NULL,
  updated_at TIMESTAMP NOT NULL,
  PRIMARY KEY (competitor, sku)
);

That works for a dashboard showing current prices, but it destroys evidence. When a price changes from $99 to $19, you need to know whether that was a real discount, a parsing bug, or a product mismatch.

A better model keeps append-only observations:

CREATE TABLE price_observations (
  id BIGSERIAL PRIMARY KEY,
  competitor TEXT NOT NULL,
  sku TEXT NOT NULL,
  source_url TEXT NOT NULL,
  observed_price_cents INTEGER,
  currency TEXT,
  availability TEXT,
  confidence NUMERIC NOT NULL,
  raw_hash TEXT NOT NULL,
  observed_at TIMESTAMP NOT NULL,
  accepted BOOLEAN NOT NULL DEFAULT false,
  rejection_reason TEXT
);

CREATE INDEX idx_price_observations_lookup
ON price_observations (competitor, sku, observed_at DESC);

Then your “current price” becomes a query or materialized view over accepted observations, not the only copy of the truth.

Validate before publishing

Here is a simplified Python example. It rejects missing prices, unexpected currency changes, and large price movements unless the same value appears twice in a short window.

from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class Observation:
    competitor: str
    sku: str
    price_cents: int | None
    currency: str | None
    observed_at: datetime
    source_url: str

@dataclass
class ValidationResult:
    accepted: bool
    reason: str | None = None


def validate_observation(obs: Observation, history: list[Observation]) -> ValidationResult:
    if obs.price_cents is None:
        return ValidationResult(False, "missing_price")

    if obs.price_cents <= 0:
        return ValidationResult(False, "non_positive_price")

    previous = history[0] if history else None
    if not previous:
        return ValidationResult(True)

    if previous.currency and obs.currency != previous.currency:
        return ValidationResult(False, "currency_changed")

    change = abs(obs.price_cents - previous.price_cents) / previous.price_cents

    if change > 0.40:
        recent_matches = [
            h for h in history
            if h.price_cents == obs.price_cents
            and obs.observed_at - h.observed_at <= timedelta(minutes=30)
        ]

        if len(recent_matches) < 1:
            return ValidationResult(False, "large_price_change_needs_confirmation")

    return ValidationResult(True)

This is not enough for every business, but it shows the shape of the guardrail. You can add checks for pack size, variant name, shipping inclusion, coupon text, marketplace seller, and stock status.

The point is not to block every unusual price. The point is to stop one bad scrape from becoming an automated pricing action.

Treat freshness as a product requirement

“Real time” should have a number attached to it. A grocery delivery app, an airline fare monitor, and a B2B catalog do not need the same refresh interval.

For each data source, define:

• Maximum acceptable age, for example 5 minutes or 6 hours
• Retry policy when extraction fails
• Whether stale data can still be shown
• Whether stale data can drive automated decisions

A useful pattern is to publish data with an explicit freshness field:

{
  "competitor": "example-shop",
  "sku": "ABC-123",
  "price_cents": 12999,
  "currency": "USD",
  "observed_at": "2026-08-18T10:15:00Z",
  "fresh_until": "2026-08-18T10:45:00Z",
  "confidence": 0.94
}

Downstream systems can then make their own decision. A dashboard may show stale data with a warning. An automated repricing service may refuse to act on it.

Failure modes worth logging explicitly

Do not collapse every problem into scrape_failed. You want enough detail to know whether the issue is temporary, structural, or business-related.

Useful failure categories include:

• http_403: blocked or unauthorized
• http_429: rate limited
• selector_missing: page structure changed
• render_timeout: client-side page did not finish loading
• product_unavailable: valid page, no purchasable item
• variant_ambiguous: multiple possible matches
• validation_rejected: extracted data failed quality checks

These categories make alerting less noisy. A few http_429 responses might trigger backoff. A sudden spike in selector_missing probably means someone needs to update the extractor.

Keep humans in the loop for high-impact changes

Fully automated pricing can work, but only when the risk is bounded. If a competitor drops a price by 3%, automation may be fine. If your system sees a 70% drop on a top-selling SKU, route it for review or require confirmation from another source.

You can encode that as policy:

def action_for_change(percent_change: float, revenue_rank: int) -> str:
    if percent_change < 0.05:
        return "auto_publish"

    if percent_change < 0.20 and revenue_rank > 100:
        return "auto_publish"

    if percent_change < 0.40:
        return "notify_category_manager"

    return "require_manual_review"

This keeps the system fast for routine changes while reducing the blast radius of bad or surprising data.

A practical next step

Pick one competitor, one product category, and one downstream decision. Build the pipeline with append-only observations, validation reasons, freshness metadata, and explicit failure categories before you connect it to pricing automation. You will learn more from a narrow system with good evidence than from a broad scraper that nobody trusts.