Price monitoring pipeline: from card matching to an alert worth acting on

I keep a price monitor running for an appliance and power tool retailer. Forty thousand of our own positions, nine competitor sites, one table that lands in the buyers' dashboard before they walk in. It has been running long enough that I know which link snaps first, and it is never the one people expect.

The pipeline has eight links, and each hands a defined object to the next. Discovery turns our position list into candidate competitor cards. Matching decides which of those cards describes the same physical item we stock. The schedule decides how often each matched card gets visited. Extraction lifts the number off the page. History stores every observation as a fact with a timestamp. Detection compares the newest observation against the stored series. Alerting decides which of those movements a person should see this morning. Acceptance decides whether the run gets published at all.

Everything below is that conveyor, with the arithmetic I size it by. Code is Python and Postgres, trimmed of logging and argument parsing. I have kept the sizing tables, because a monitor that cannot state its own request budget gets throttled into uselessness inside a week.

Matching our positions to somebody else's cards

Nothing downstream matters until this link holds. A monitor that lines our position up against the wrong competitor card produces confident, precise, wrong numbers, and the buyers act on them within the hour.

Three keys carry the matching, and I run them in that order. The manufacturer code comes first. It is printed on the box, it survives translation, and where both sides publish it the match is exact. On my nine sites it appears on 62 percent of cards, sometimes in a spec table, sometimes in the title tail, sometimes only inside the payload the grid fetches.

The barcode is second, and it is the strongest key that exists. Two cards carrying the same barcode are the same box from the same factory. Coverage is the problem: 31 percent across my set, concentrated on the two marketplaces that force sellers to supply it.

The normalised title is the fallback, and it is the one that needs actual work. Raw titles carry seller noise, pack counts, colour words, warranty phrases and the retailer's own house brand slapped on the front. I strip all of it down to a token set, then compare token sets.

That last row earns its place more often than I expected. Small retailers pull the manufacturer's press photo straight from the vendor portal, so a perceptual hash of the primary shot pairs cards that share nothing in their titles. I treat it as supporting evidence only, since a shared press photo also pairs two capacities of the same model.

import re, unicodedata

NOISE = {"new", "original", "genuine", "warranty", "official", "boxed",
         "pack", "pcs", "set", "kit", "bundle", "free", "delivery"}
UNITS = {"litre": "l", "liter": "l", "l": "l", "ml": "ml",
         "kilogram": "kg", "kg": "kg", "gram": "g", "g": "g",
         "watt": "w", "w": "w", "kilowatt": "kw", "kw": "kw"}

def norm_title(s: str) -> frozenset:
    s = unicodedata.normalize("NFKD", s).encode("ascii", "ignore").decode()
    s = s.lower().replace("-", " ").replace("/", " ")
    s = re.sub(r"[^a-z0-9 .]", " ", s)
    out = []
    for tok in s.split():
        if tok in NOISE:
            continue
        m = re.fullmatch(r"(\d+(?:\.\d+)?)([a-z]+)", tok)
        if m and m.group(2) in UNITS:
            out.append(m.group(1).rstrip("0").rstrip(".") + UNITS[m.group(2)])
        else:
            out.append(tok)
    return frozenset(out)

def mpn_key(s: str) -> str:
    return re.sub(r"[^A-Z0-9]", "", (s or "").upper())

Two details in that function came from damage. Stripping the trailing zeros off a numeric token pairs a card written with one decimal against a card written with two, which was silently splitting a third of my power tool matches. Folding unit spellings into a single form pairs the litre spelled out against the litre abbreviated, and those two spellings were treated as separate models for a whole season before I noticed.

Discovery feeds this link. For positions where I have no competitor card yet, I push the product name and the manufacturer code through search collection and harvest the result pages, which turns up shop pages no sitemap of ours would ever reach. That job is a query grinder, so I run query harvesting in Key Collector over the pool and let it work through the backlog overnight. Roughly one query in five yields a card I did not already hold.

Deciding what to do with a match that could go either way

Exact keys give me 71 percent of pairs. The remaining 29 percent land in a scoring queue, and how that queue behaves decides whether the monitor is trusted.

I score every candidate pair on a weighted sum: token overlap, numeric token agreement, brand agreement, image hash distance, and a category compatibility flag. The sum lands in a band, the band picks an action, and every action is written down so the decision can be replayed later.

The human queue sorts by turnover, and that ordering is the whole reason the queue works. Nobody clears a backlog of eleven thousand ambiguous pairs. Everybody clears the ninety pairs that cover a fifth of the revenue, and the rest wait in the middle band doing no harm, since a pair that never got confirmed never produces an alert.

Three ambiguity types recur, and each has its own handling. Pack counts are the first: a single unit on our side against a carton of six on theirs. I keep the pair, store the pack multiplier as a column on the match, and divide at comparison time. Variants are the second: colour or finish differences that the manufacturer treats as one model. I merge those into a variant group and compare the group against the group. Condition is the third: refurbished, open box, display unit. Those never merge with a sealed item, and a card whose title or badge carries a condition marker gets a flag that keeps it out of the comparison set entirely.

CREATE TABLE matches (
  match_id     BIGSERIAL PRIMARY KEY,
  position_id  INTEGER NOT NULL REFERENCES positions(position_id),
  card_id      BIGINT  NOT NULL REFERENCES competitor_cards(card_id),
  method       TEXT    NOT NULL CHECK (method IN ('gtin','mpn','title','image','manual')),
  score        NUMERIC(4,3) NOT NULL,
  pack_ratio   NUMERIC(8,3) NOT NULL DEFAULT 1,
  state        TEXT    NOT NULL DEFAULT 'candidate',
  decided_by   TEXT,
  decided_at   TIMESTAMPTZ,
  UNIQUE (position_id, card_id)
);

CREATE INDEX matches_live ON matches (position_id)
  WHERE state = 'confirmed';

The partial index on confirmed rows is small and it carries every query the comparison stage makes. The candidate rows sit in the same table and stay out of the way, which means a pair rejected today can be reviewed next quarter without any archaeology.

How often each shelf gets swept

A uniform sweep interval wastes requests on half the catalogue and misses movements on the other half. I set the interval per category from measured movement, and I remeasure it every quarter.

Measuring it takes one cheap experiment: sweep a random sample of two hundred cards per category every hour for a week, then count how many distinct values each card took. Categories separate immediately, and the separation is wider than any guess I would have made.

Seasonality shows up in that table as two rows for one category, and treating it as one row cost me a whole spring. Garden equipment sat on a weekly interval through the season opening, our competitors ran three promotions in eleven days, and the buyers heard about all three from a supplier while my dashboard sat quiet.

The interval also bends per card, on two triggers. A card that moved in the previous two sweeps drops to the next tighter interval for twenty four hours, because movements cluster. A card that has held the same value for six weeks steps out one interval, up to a weekly floor. That adaptive layer removed 23 percent of my daily requests with no measurable loss in detection latency, and the accounting is one integer column on the match row.

Lifting the number off a page that does not want to give it up

The extraction link is where a monitor quietly turns into a random number generator. The page shows one figure to a person and carries four or five in its markup, and picking the wrong one produces a series that looks plausible for months.

I resolve the value against a rung chain, the same way I handle every other field, and I store which rung answered. What matters more here is the trap catalogue, because each trap needs its own rule.

The last one is worth spelling out. On marketplaces the headline figure belongs to whichever seller currently holds the buy box, and that seller changes hourly. A series built from the buy box measures seller rotation. A series built from one pinned seller identity measures that seller's behaviour. I collect both, and the buyers look at the buy box series while the category managers look at the pinned one.

from decimal import Decimal, InvalidOperation
import re

MINOR = 100
SEP = re.compile(r"[   ']")

def to_minor(raw: str) -> int | None:
    """Text figure to an integer in minor units, no float anywhere."""
    if not raw:
        return None
    s = SEP.sub("", raw.strip())
    s = re.sub(r"[^0-9.,]", "", s)
    if s.count(",") and s.count("."):
        s = s.replace("," if s.rfind(",") < s.rfind(".") else ".", "")
    s = s.replace(",", ".")
    if s.count(".") > 1:
        s = s.replace(".", "", s.count(".") - 1)
    try:
        return int((Decimal(s) * MINOR).to_integral_value())
    except (InvalidOperation, ValueError):
        return None

def comparable(obs: dict, match: dict) -> int | None:
    v = obs.get("open_value") or obs.get("headline_value")
    if v is None or obs.get("is_instalment"):
        return None
    v = int(round(v / float(match["pack_ratio"])))
    if obs.get("per_unit_basis"):
        v = int(round(v * obs["package_size"] / obs["per_unit_basis"]))
    return v

Integers in minor units are the only representation I let past this function. A float crept into an early version of the monitor, two sites published figures with a decimal comma while seven used a decimal point, and the parser accepted both without complaint for eleven days. Every comparison in that window was arithmetic on numbers that were a hundred times apart.

Keeping the history so a movement can be proved

Two tables carry the whole history, and neither of them ever gets an update statement pointed at it. Observations are append only facts. Change events are derived rows that anybody can rebuild from the observations if they doubt them.

CREATE TABLE observations (
  obs_id       BIGSERIAL PRIMARY KEY,
  match_id     BIGINT      NOT NULL REFERENCES matches(match_id),
  run_id       BIGINT      NOT NULL,
  seen_at      TIMESTAMPTZ NOT NULL,
  value_minor  BIGINT,
  headline_minor BIGINT,
  in_stock     BOOLEAN,
  is_promo     BOOLEAN     NOT NULL DEFAULT FALSE,
  seller_key   TEXT,
  rung         SMALLINT    NOT NULL,
  status       TEXT        NOT NULL DEFAULT 'ok'
);

CREATE INDEX obs_series ON observations (match_id, seen_at DESC);
CREATE INDEX obs_run    ON observations (run_id) WHERE status <> 'ok';

CREATE TABLE change_events (
  event_id     BIGSERIAL PRIMARY KEY,
  match_id     BIGINT      NOT NULL,
  from_minor   BIGINT      NOT NULL,
  to_minor     BIGINT      NOT NULL,
  delta_pct    NUMERIC(6,2) NOT NULL,
  confirmed_at TIMESTAMPTZ NOT NULL,
  basis        TEXT        NOT NULL
);

A failed fetch writes a row too, with a null value and a status saying why. That sounds like bookkeeping fussiness until the first argument about coverage, at which point the distinction between never visited and visited and failed settles the argument in one query.

Detection runs as a window function over the confirmed series. I compare the newest observation against the last observation that differed, skipping over any row whose status is anything other than ok, which stops a single timeout from registering as two movements.

WITH ranked AS (
  SELECT match_id, seen_at, value_minor,
         LAG(value_minor) OVER w AS prev_minor,
         LAG(seen_at)     OVER w AS prev_at
  FROM observations
  WHERE status = 'ok' AND value_minor IS NOT NULL
  WINDOW w AS (PARTITION BY match_id ORDER BY seen_at)
)
SELECT match_id, prev_minor, value_minor, prev_at, seen_at,
       ROUND(100.0 * (value_minor - prev_minor) / prev_minor, 2) AS delta_pct
FROM ranked
WHERE prev_minor IS NOT NULL
  AND value_minor <> prev_minor
  AND seen_at > now() - INTERVAL '26 hours'
ORDER BY ABS(value_minor - prev_minor) DESC;

Twenty six hours of overlap on a daily job is deliberate. A run that starts late still sees its own previous window, and a movement never falls between two runs because the boundaries happened to shift by twenty minutes.

Turning movements into alerts people keep reading

A monitor that fires on every movement gets muted in four days, and a muted monitor is worth nothing at all. Five gates sit between a detected movement and a message, and each one exists because something got through.

The relative floor comes first. Movements under 1.5 percent go into the daily digest and never page anybody, since rounding adjustments and separator noise both live below that line. The absolute floor sits beside it for cheap consumables, where a fractional move is arithmetically large and commercially irrelevant.

Confirmation is the second gate. A movement has to survive one more sweep before it becomes an alert, which adds latency equal to one interval and removes 40 percent of my false positives outright. Marketplace tests are the reason: a site pushes a value for twenty minutes, watches the click through, and rolls it back.

Sanity banding is the third. Any movement above 60 percent, in either direction, is treated as a parsing failure until proven otherwise. It goes to a review queue with the raw markup snapshot attached, and about one in nine turns out to be real.

Suppression is the fourth. Known campaign windows, published by the competitors themselves in their own banners, get a suppression flag that routes movements into the digest for the duration. A promotion that moves four hundred cards at once is one event, and it should read as one line.

Cooldown is the fifth. One position produces at most one alert per twelve hours, no matter how many times it oscillates, and the digest carries the full oscillation for anyone who wants it.

from datetime import timedelta

FLOOR_PCT   = 1.5
SANITY_PCT  = 60.0
COOLDOWN    = timedelta(hours=12)

def gate(ev, series, state, campaign_open: bool):
    pct = abs(ev.delta_pct)
    if pct < FLOOR_PCT or abs(ev.to_minor - ev.from_minor) < state.floor_minor:
        return "digest"
    if not confirmed_twice(series, ev):
        return "hold"
    if pct > SANITY_PCT:
        return "review"
    if campaign_open:
        return "digest"
    if state.last_alert_at and ev.confirmed_at - state.last_alert_at < COOLDOWN:
        return "digest"
    return "alert"

def confirmed_twice(series, ev) -> bool:
    tail = [o for o in series if o.status == "ok"][-2:]
    return len(tail) == 2 and all(o.value_minor == ev.to_minor for o in tail)

The routing values matter as much as the thresholds. Nothing gets dropped. A movement that fails a gate goes to the digest, where it is one row among many, and the review queue keeps the suspicious ones visible with their evidence attached. My alert volume settled at eleven a day across forty thousand positions, and the buyers read all eleven.

What a day of sweeping actually costs

Sizing this before the first run is what keeps the pace honest, so here is the budget for one full day at the intervals in the schedule table.

Three readings come out of that table. Card visits are 96 percent of the request count, so every field I can lift from a grid payload removes tens of thousands of requests from the day. The render path is 0.14 percent of requests and 14 percent of the bytes, which is why it stays a separate worker pool with its own budget. And the daily figure sits above nine gigabytes, so I run the whole thing where volume stays unmetered on every tier and size the job by request count alone.

The adaptive interval layer is what keeps this budget flat while the catalogue grows. Without it the same schedule needs 847 000 requests, and the difference is entirely cards that have not moved since spring being visited as though they might.

Spreading the sweep across addresses and holding a pace

The aggregate number is one thing. What each origin observes is a different number, and that is the one that decides whether the run finishes on schedule or spends its afternoon in a retry loop.

The pool I work from holds around 12 000 active endpoints with rotation handled inside it automatically, so the collector never picks an exit and never tracks one. My side controls two dials: how many tunnels run at once, and how fast each one is allowed to go. At 220 concurrent tunnels each doing 14 requests a minute, any single exit an origin sees produces well under one request per second, which sits under the threshold on all nine sites I sweep. Running the sweep over endpoints intended for collection runs keeps the latency spread tight enough that the pacing arithmetic holds from one day to the next.

Thread arithmetic comes from the package ceiling. A regular tier allows 1 000 concurrent connections, the corporate tier goes to 3 000, packages do not stack, and binding a second source address halves the count. With two bound addresses on a regular tier I have 500 concurrent connections to allocate. The sweep takes 220, the render workers take 60, discovery takes 80, the health checker takes 30, and the remainder stays free for the ad hoc pulls the category managers ask for mid afternoon.

I hold the second row. The third finishes sooner on paper and arrives later in practice once the retry queue starts growing faster than it drains. Per host budgets sit on top of the aggregate, since one strict marketplace should never slow the eight sites that do not care.

import asyncio, random, time
from collections import defaultdict

class HostBudget:
    """Per origin spacing with a shared ceiling and a soft back off."""
    def __init__(self, spacing: dict[str, float], ceiling: int):
        self.spacing = defaultdict(lambda: 0.07, spacing)
        self.next_at = defaultdict(float)
        self.gate = asyncio.Semaphore(ceiling)
        self.locks = defaultdict(asyncio.Lock)

    async def slot(self, host: str):
        async with self.gate:
            async with self.locks[host]:
                now = time.monotonic()
                wait = max(0.0, self.next_at[host] - now)
                self.next_at[host] = max(now, self.next_at[host]) + self.spacing[host]
            if wait:
                await asyncio.sleep(wait)
            await asyncio.sleep(random.uniform(0.0, 0.28))

    def back_off(self, host: str, factor: float = 2.0, cap: float = 3.5):
        self.spacing[host] = min(cap, self.spacing[host] * factor)

    def recover(self, host: str, step: float = 0.85, floor: float = 0.05):
        self.spacing[host] = max(floor, self.spacing[host] * step)

Back off doubles the spacing for the offending host and recovery walks it down by 15 percent for every clear minute, which settles each origin at its own sustainable pace within about forty minutes of a run starting. The jitter at the end of the slot costs nothing measurable and removes the metronome pattern that a fixed sleep leaves in an access log. Endpoint hygiene matters as much as pacing here, so the whole conveyor sits on private server endpoints on owned hardware and the list arrives as IP:PORT:LOGIN:PASS from the panel at the start of each cycle. A monitor is a standing job with no end date, so I take a month of access to the pool and stop rebuilding credentials every few days. The free trial window is long enough to run one full sweep of a single competitor and read the real numbers off your own origins before committing to a term. Discovery has its own posture inside all of this, and the Key Collector proxy setup is where I keep the query side of the job configured.

Accepting a run and repairing holes in the series

A sweep is finished when it passes acceptance, and acceptance is a fixed set of queries with numbers attached. Failing acceptance means the run does not publish, the previous table stays live, and I get a message telling me which check failed.

Gaps in the series get their own handling, and the rule is short: I never fill one. A missing observation stays missing, marked with its status, and every downstream calculation skips it. Forward filling a gap invents a fact, and the invented fact reads exactly like a real one six weeks later when somebody asks why a competitor held a figure for three days.

SELECT m.match_id, c.host, MAX(o.seen_at) AS last_ok,
       COUNT(*) FILTER (WHERE o.status <> 'ok'
                        AND o.seen_at > now() - INTERVAL '3 days') AS recent_fails
FROM matches m
JOIN competitor_cards c USING (card_id)
LEFT JOIN observations o ON o.match_id = m.match_id AND o.status = 'ok'
WHERE m.state = 'confirmed'
GROUP BY m.match_id, c.host
HAVING MAX(o.seen_at) IS NULL
    OR MAX(o.seen_at) < now() - INTERVAL '48 hours'
ORDER BY recent_fails DESC, last_ok NULLS FIRST;

That query drives the repair job, and its output splits into three buckets by cause. Transport failures go straight back into the queue at a slower per host pace and usually clear on the first retry cycle. Structural failures, where the fetch succeeded and the extractor found nothing, go to a review queue with a markup snapshot, since those signal a template edit that will hit every card on that site within the day. Disappearances, where the card returns a 404 across three consecutive cycles, close the match with a reason code and drop the position back into discovery.

One habit sits outside all of the queries and I keep it anyway. Every morning I open the published table, sort by movement descending, and read the top fifteen rows with the source cards in a second window. Thirty seconds of looking has caught a pack multiplier applied twice, a seller pin that silently detached, and a category whose entire schedule had stopped firing because a cron entry lost its final line. No query I had written was looking for any of those.

The three companion pieces cover the collection jobs this one leans on. Bulk field extraction from catalogue cards, with the sharding and resume machinery, sits in scraping product pages at scale. Records with thin fields and heavy pagination behave differently and are worked through in collecting business directory records. Postings that expire while a sweep is still walking them need their own freshness handling, which is the subject of gathering job board listings. And before fixing the pace for a sweep of this size, the request pace planner turns a daily request count and a package ceiling into the spacing figure to hold per origin.