Price Monitoring Competitor Intelligence Workflows

Detecting Competitor Price Changes

A scraper delivers a competitor price every time it runs, but almost every observation repeats the last one — the interesting event is the small fraction where the price actually moved, and moved for real rather than because a scrape glitched. This is a focused, runnable recipe for turning a raw stream of price observations into a clean series of change events, each carrying a magnitude and direction, with repeats collapsed and scrape errors filtered out. It sits under price positioning and competitive benchmarking: a detected change is precisely the trigger to recompute the competitive price index for the affected SKU, and a large confirmed move is what feeds price anomaly alerting and monitoring.

Prerequisites & Input Contract

Change detection is stateful: it compares each observation against the last known price for the same (product, competitor) pair, so it needs a place to hold that last-known state and it needs observations that are already matched and normalized. Matching is upstream per fuzzy matching algorithms for SKU alignment; currency and unit normalization per currency conversion and exchange-rate sync. Feeding un-normalized prices here produces “changes” that are really just a currency or tax-basis flip.

Library versions: Python 3.11+, pandas>=2.1. Each observation is one scrape of one competitor’s price for one product:

import pandas as pd
from dataclasses import dataclass
from datetime import datetime, timezone

@dataclass
class Observation:
    product_id: str
    competitor_id: str
    price: float
    in_stock: bool
    observed_at: datetime      # UTC-aware

# The last-known state we diff against, held per (product, competitor).
@dataclass
class PriceState:
    price: float
    observed_at: datetime
    pending_price: float | None = None   # a candidate change awaiting confirmation
    pending_since: datetime | None = None

Step-by-Step Implementation

Step 1 — Diff each observation against the last-known price

The first gate is trivial and eliminates the overwhelming majority of traffic: if the new price equals the stored one, nothing happened. Use a relative epsilon so floating-point noise and sub-cent rounding do not register as changes.

def is_material(old: float, new: float, min_pct: float = 0.005,
                min_abs: float = 0.01) -> bool:
    """A change must clear BOTH a relative and an absolute floor to count."""
    delta = abs(new - old)
    return delta >= min_abs and (delta / old) >= min_pct

# Repeats and sub-0.5% wobble are ignored outright.
print(is_material(199.00, 199.00))   # -> False (no move)
print(is_material(199.00, 199.50))   # -> False (0.25% < 0.5% floor)
print(is_material(199.00, 209.00))   # -> True  (5.0% move)

Step 2 — Reject scrape errors before trusting a move

A large apparent jump is as likely to be a parser regression — a dropped decimal, a grabbed strike-through price, a currency symbol misread — as a real reprice. Bound each observation against a plausible range derived from the recent history for that pair, and route anything outside it to quarantine rather than emitting a change. This is the same logic as statistical outlier detection for price data, applied per competitor series.

def is_plausible(new: float, recent: list[float], max_jump: float = 0.60) -> bool:
    """Reject values implausibly far from the recent median for this pair."""
    if not recent:
        return True                       # no history yet; accept and learn
    med = sorted(recent)[len(recent) // 2]
    if new <= 0:
        return False                      # non-positive is always an error
    return abs(new - med) / med <= max_jump

# A $129.99 item scraped as $12.99 (dropped digit) is rejected as implausible.
print(is_plausible(12.99, [129.99, 131.00, 129.50]))   # -> False
print(is_plausible(119.99, [129.99, 131.00, 129.50]))  # -> True (a real markdown)

Step 3 — Debounce: confirm a move before emitting it

A single observation of a new price can be a transient — a regional A/B test, a cart-level flicker, a momentary glitch that reverts on the next scrape. Debouncing requires the new price to persist across a confirmation window (or a second confirming scrape) before it is accepted as the new state. This trades a little latency for a large drop in false events.

def update(state: PriceState, obs: Observation,
           confirm_after_s: int = 900) -> tuple[PriceState, dict | None]:
    """Returns (new_state, change_event_or_None)."""
    # Same as current confirmed price: clear any pending candidate, no event.
    if not is_material(state.price, obs.price):
        state.pending_price = None
        state.pending_since = None
        return state, None

    # A material move. Is it the same candidate we saw before?
    if state.pending_price is not None and not is_material(state.pending_price, obs.price):
        held = (obs.observed_at - state.pending_since).total_seconds()
        if held >= confirm_after_s:             # persisted long enough -> confirm
            event = _emit(state, obs)
            state.price, state.observed_at = obs.price, obs.observed_at
            state.pending_price = state.pending_since = None
            return state, event
        return state, None                      # still within debounce window
    # A new candidate: start the confirmation clock.
    state.pending_price = obs.price
    state.pending_since = obs.observed_at
    return state, None

Step 4 — Emit a deduped change event with magnitude and direction

When a move confirms, emit exactly one event describing it: the from/to prices, signed magnitude, direction, and the timestamps that bound it. Emitting on confirmation — not on first sight — is what dedupes the stream, because the pending state absorbs the repeated observations of the same candidate price.

def _emit(state: PriceState, obs: Observation) -> dict:
    delta = obs.price - state.price
    return {
        "product_id": obs.product_id,
        "competitor_id": obs.competitor_id,
        "old_price": state.price,
        "new_price": obs.price,
        "delta": round(delta, 2),
        "pct_change": round(delta / state.price, 4),
        "direction": "increase" if delta > 0 else "decrease",
        "confirmed_at": obs.observed_at.isoformat(),
    }

Running a short stream through the machine shows repeats and a transient blip suppressed, and one confirmed drop emitted:

from datetime import timedelta
t0 = datetime(2026, 7, 15, 8, 0, tzinfo=timezone.utc)
state = PriceState(price=199.00, observed_at=t0)
recent = [199.00, 199.00, 198.50]
stream = [
    (199.00, 0),      # repeat -> no event
    (189.00, 5),      # candidate drop -> pending, no event yet
    (189.00, 25),     # persisted 20 min -> confirm, EMIT
    (189.00, 40),     # now the confirmed price -> no event
]
for price, mins in stream:
    o = Observation("p-100", "acme", price, True, t0 + timedelta(minutes=mins))
    if is_plausible(o.price, recent):
        state, ev = update(state, o)
        if ev:
            print(ev["direction"], ev["old_price"], "->", ev["new_price"], ev["pct_change"])
# -> decrease 199.0 -> 189.0 -0.0503

The state machine is easier to read as a diagram.

State machine for confirming a competitor price changeAn observation enters a materiality diff. If the price is unchanged it is dropped as a repeat. If it is a material move it passes a plausibility check that rejects scrape errors to quarantine, then enters a pending state. If the same candidate persists past the confirmation window a single change event is emitted; if it reverts first, the pending candidate is cleared as a transient.unchanged → drop repeatscrape error → quarantinereverts → clear as transientObservationMateriality diff≥ 0.5% and ≥ $0.01Plausibilitywithin recent rangePendingconfirm windowChangeevent emittedheld

Verification & Testing

Test the three behaviours that make or break the detector: repeats emit nothing, a transient blip is suppressed, and a persisted move emits exactly one event.

def test_detector():
    t0 = datetime(2026, 7, 15, 8, 0, tzinfo=timezone.utc)
    s = PriceState(price=100.0, observed_at=t0)
    # repeat -> no event
    s, ev = update(s, Observation("p", "c", 100.0, True, t0 + timedelta(minutes=5)))
    assert ev is None
    # candidate at t+5, confirmed at t+25 -> exactly one event
    s, ev = update(s, Observation("p", "c", 90.0, True, t0 + timedelta(minutes=5)))
    assert ev is None                      # pending, not yet confirmed
    s, ev = update(s, Observation("p", "c", 90.0, True, t0 + timedelta(minutes=25)))
    assert ev and ev["direction"] == "decrease" and ev["pct_change"] == -0.10
    # a further repeat of the new price emits nothing
    s, ev = update(s, Observation("p", "c", 90.0, True, t0 + timedelta(minutes=40)))
    assert ev is None
    print("detector verified")

test_detector()   # -> detector verified

For a live check, reconcile emitted events against raw observations for a sample of pairs: every confirmed event should correspond to a price that was seen at least twice across the confirmation window, and no event should exist for a value that reverted.

Edge Cases & Gotchas

  • Oscillating / flickering prices. Some sellers A/B test and bounce between two prices every few minutes. Naïve diffing emits a flood of events. The debounce window absorbs short flickers; for persistent oscillation, add a minimum dwell time between accepted events per pair so you record the envelope, not every bounce.
  • Stock-out masquerading as a change. A price vanishing because the item went out of stock is not a price cut. Gate on in_stock before the diff — an out-of-stock observation should update stock state, never emit a price-change event.
  • Cold start with no history. The plausibility check accepts the first observations for a new pair because there is nothing to compare against. Mark events from a pair with fewer than a handful of prior observations as low-confidence until the recent-window history fills.
  • Clock skew and out-of-order scrapes. Distributed scrapers can deliver observations out of timestamp order. Key the confirmation window on observed_at, not arrival time, and ignore any observation older than the current confirmed state to avoid resurrecting a stale price.

Performance Notes

Detection is O(1) per observation — a couple of comparisons and a state read/write — so throughput is bounded by the state store, not the logic. Hold PriceState in a keyed store such as Redis or an in-memory dict per worker, sharded by product_id so each pair’s state lives on one worker and updates need no locking. Keep the recent history for the plausibility check to a short bounded deque (the last 10–20 prices) rather than the full series; a fixed window keeps memory flat across millions of pairs and keeps the median cheap. Emit change events onto the same message broker that carries the rest of the pipeline so index recomputation and price anomaly alerting consume them asynchronously.

Frequently Asked Questions

How large should the materiality threshold be? Start at a 0.5% relative floor combined with a small absolute floor so sub-cent rounding and tiny wobble never register. Tune per category: high-ticket electronics can justify a smaller relative floor because a 1% move is real money, while low-ticket commodities need a larger one to avoid event spam.

Why confirm a change instead of emitting on first sight? Because a single observation of a new price is frequently a transient — a regional test, a cart flicker, or a glitch that reverts on the next scrape. Requiring the price to persist across a confirmation window trades a little latency for a large reduction in false change events that would otherwise trigger needless repricing.

How do I tell a real markdown from a scrape error? Bound each observation against the recent median for that pair and reject moves beyond a plausible jump, exactly as in statistical outlier detection. A dropped decimal or a misread strike-through price lands far outside the recent range and is quarantined, while a genuine markdown stays within a believable distance of recent prices.