Price Monitoring Competitor Intelligence Workflows

Building Price-Drop Alerts in Python

A competitor cuts a price by 20% overnight and you want to know in minutes, not at tomorrow’s stand-up — but you do not want a flood of duplicate pings, and you emphatically do not want to page anyone over a scraping glitch that reported a $0 price. This recipe builds a competitor price-drop alert that detects a meaningful drop against a baseline, debounces transient dips, formats a Slack-style webhook payload, and suppresses repeats with a cooldown. It sits under price anomaly alerting & monitoring, and the drop it detects is the same signal consumed by detecting competitor price changes for benchmarking.

Prerequisites & Input Contract

Each price observation arrives already matched to a canonical product and normalized to base currency. The recipe uses only the standard library, so it runs anywhere; Python 3.11+ is assumed.

# A single competitor price observation plus the trailing baseline
# history for that (product, seller) pair, oldest-to-newest.
observation = {
    "product_id": "p-44182",
    "seller_id": "mkt:acme-electronics",
    "price": 149.00,                 # latest observed price, base currency
    "currency": "USD",
    "observed_at": "2026-07-15T09:14:00Z",
    "history": [199.0, 201.0, 199.5, 200.0, 198.0, 199.0, 200.0, 199.0],
}

The history list is the trailing window your pipeline already maintains per pair — a rolling buffer of the last N observed prices. It is the reference the drop is measured against; without it, an absolute price carries no meaning.

Step-by-Step Implementation

Step 1 — Compute a robust baseline and the drop magnitude

Use the median of the trailing window, not the mean, so a single earlier spike or bad scrape does not distort the reference. Express the drop as a fraction of that baseline.

import statistics

def baseline_and_drop(price, history):
    """Return (baseline, drop_fraction). Positive drop = price fell."""
    if len(history) < 5:
        return None, 0.0             # not enough history to judge a drop
    baseline = statistics.median(history)
    if baseline <= 0:
        return None, 0.0             # guard against bad reference
    drop = (baseline - price) / baseline
    return baseline, round(drop, 4)

print(baseline_and_drop(149.0, observation["history"]))
# -> (199.0, 0.2513)   ~25% below the trailing median

Step 2 — Validate the price before trusting the drop

A “drop” to an absurd value is almost always a data error, not a price move. Reject non-positive prices and physically implausible collapses so a scrape glitch never pages anyone.

def is_plausible(price, baseline, max_drop=0.90):
    """Reject non-positive prices and drops too large to be real."""
    if price is None or price <= 0:
        return False                 # $0 / negative = parse error
    if baseline and (baseline - price) / baseline > max_drop:
        return False                 # >90% off is almost always bad data
    return True

Step 3 — Debounce with a confirmation count

A single low reading may be a transient render error or a flash promo that reverts in minutes. Require the drop to persist across a small number of consecutive observations before it counts, tracked in a per-pair state store.

def debounce(key, dropped, state, need=2):
    """Increment a streak while `dropped` holds; fire only at `need`.

    state: dict-like key -> consecutive-drop count.
    Returns True exactly when the streak first reaches `need`.
    """
    streak = state.get(key, 0)
    if not dropped:
        state[key] = 0               # reset on any non-drop observation
        return False
    streak += 1
    state[key] = streak
    return streak == need            # fire once, on the confirming reading

Step 4 — Format and dispatch the notification, with cooldown

Build a compact webhook payload and suppress a repeat inside a cooldown so a sustained low price does not re-alert every crawl.

import json, time
from datetime import datetime

def format_alert(obs, baseline, drop):
    pct = round(drop * 100, 1)
    return {                          # Slack-style webhook JSON
        "text": f":arrow_down: Competitor price drop: {obs['product_id']}",
        "blocks": [{
            "type": "section",
            "text": {"type": "mrkdwn", "text": (
                f"*{obs['seller_id']}* dropped *{pct}%* on "
                f"`{obs['product_id']}`\n"
                f"{obs['currency']} {baseline:.2f} → *{obs['price']:.2f}*")},
        }],
        "meta": {"observed_at": obs["observed_at"], "drop_pct": pct},
    }

def dispatch(obs, state, send, threshold=0.15, cooldown_s=7200, now=None):
    """Full pipeline for one observation. `send` posts the webhook."""
    now = now or time.time()
    baseline, drop = baseline_and_drop(obs["price"], obs["history"])
    if baseline is None or not is_plausible(obs["price"], baseline):
        return "rejected"
    key = f"{obs['product_id']}:{obs['seller_id']}"
    if drop < threshold:
        state.pop(f"streak:{key}", None)     # not a drop; clear streak
        return "no_drop"
    streak_state = {key: state.get(f"streak:{key}", 0)}
    fire = debounce(key, True, streak_state, need=2)
    state[f"streak:{key}"] = streak_state[key]
    if not fire:
        return "debouncing"
    last = state.get(f"cd:{key}", 0)
    if now - last < cooldown_s:
        return "suppressed_cooldown"
    state[f"cd:{key}"] = now
    send(format_alert(obs, baseline, drop))
    return "sent"

Driving two observations through it (the first arms the debounce, the second confirms and fires):

sent = []
state = {}
r1 = dispatch(observation, state, send=sent.append)   # arms debounce
r2 = dispatch(observation, state, send=sent.append)   # confirms -> sends
print(r1, r2, len(sent))
# -> debouncing sent 1
print(sent[0]["meta"])
# -> {'observed_at': '2026-07-15T09:14:00Z', 'drop_pct': 25.1}

The outcomes the dispatcher can return:

Return valueMeaningNotified?
rejectedImplausible price (≤0 or >90% off)No
no_dropBelow the drop thresholdNo
debouncingDrop seen but not yet confirmedNo
suppressed_cooldownConfirmed but within cooldown of last alertNo
sentConfirmed, plausible, outside cooldownYes

Verification & Testing

Cover each return path so a broken threshold, debounce, or cooldown fails a test rather than shipping.

def test_dispatch():
    obs = dict(observation)
    # implausible price is rejected
    assert dispatch(dict(obs, price=0.0), {}, send=lambda a: None) == "rejected"
    # small dip is not a drop
    assert dispatch(dict(obs, price=195.0), {}, send=lambda a: None) == "no_drop"
    # first real drop debounces, second fires
    st, out = {}, []
    assert dispatch(obs, st, send=out.append) == "debouncing"
    assert dispatch(obs, st, send=out.append) == "sent"
    # immediate repeat is suppressed by cooldown
    assert dispatch(obs, st, send=out.append) == "suppressed_cooldown"
    assert len(out) == 1
    print("all alert paths covered")

test_dispatch()   # -> all alert paths covered

For an integration check, point send at a webhook test endpoint (or a captured-request fixture) and assert the posted JSON has the expected drop_pct and product id. Sample real sent alerts weekly and confirm the competitor truly dropped at capture time; that confirmation rate is the live precision of the rule.

Edge Cases & Gotchas

  • Fake drops from data errors. A $0, negative, or 99%-off price is nearly always a parse or render failure, not a price move. Step 2’s plausibility gate rejects them; pair it with the fake-sale filtering in filtering fake sale prices using historical averages upstream so obvious garbage never reaches the alerter.
  • Flapping around the threshold. A price oscillating just under and over the drop threshold fires and clears repeatedly. The debounce streak and the cooldown together damp this; if a pair still flaps, it is a sign the threshold sits inside the price’s normal noise band and should be widened for that category.
  • Baseline poisoned by a promo. If the trailing window already contains a promotional dip, the median baseline sags and a later real drop looks smaller than it is. Use a window long enough to outvote a single promo, or exclude known-promo observations from the baseline buffer.
  • Cold start on new products. Fewer than five history points returns a neutral result and never fires — correct behavior, since you cannot define a drop without a reference. Rely on hard business thresholds until the window fills.

Performance Notes

Every step is O(1) amortized except the median, which is O(n log n) in the (small, fixed) window size — for a trailing window of a few dozen prices this is negligible, and a single worker clears tens of thousands of observations per second. The only shared state is the per-pair streak and cooldown store; keep it in Redis so horizontally-scaled workers share suppression state, and set a TTL on cooldown keys equal to the cooldown so the store self-prunes. Batch the webhook posts if a mass event (a market-wide sale) trips many pairs at once, to avoid rate-limiting your notification endpoint.

Frequently Asked Questions

What drop threshold should I start with? Start around 10–15% below the trailing median for most categories, then tune against a weekly false-positive review. Volatile categories need a wider threshold or a robust z-score instead of a fixed percentage, because their normal price noise would otherwise trip a fixed cut constantly.

Why debounce instead of alerting on the first low reading? Because a single low observation is often a transient — a flash promo that reverts, or a momentary render glitch. Requiring the drop to persist across two consecutive observations removes most of that noise at the cost of a few minutes’ latency, which is a good trade for an alert a human will act on.

How do I stop the same drop from alerting every crawl cycle? Apply a per-pair cooldown keyed on product and seller. Once an alert fires, suppress repeats of the same drop until the cooldown elapses, so a sustained low price yields one notification rather than one per cycle.