InventoryAllocation

Competitor Rate Shopping Ingestion for Rate Parity Automation

A dynamic pricing engine is only as trustworthy as the competitor benchmark that feeds it, and that benchmark almost always arrives as a mess: a scraped Booking.com search result quoting a tax-inclusive twin-room rate in THB, an Expedia meta feed quoting a net room-only rate in USD, and a rate-shopping vendor’s CSV that silently dropped a currency column overnight. Feed that directly into a margin rule and the engine will happily undercut a phantom competitor, dump inventory on a high-demand date, or trip a parity breaker on a rate that was never actually out of line. The revenue manager sees a rate that “looks wrong,” the operations lead fields the fallout, and the engineer who owns the ingestion job gets paged to explain why a €210 suite went out at €148. This guide specifies the ingestion layer that sits between raw rate-shopping feeds and the pricing engine — the collection, normalization, and guarding that turn noisy market signals into a benchmark the engine can defend. It is part of the broader rate parity monitoring and revenue optimization programme, and it exists so that the competitor number the engine consumes is always comparable, current, and sane.

The contract this layer enforces is narrow and strict. Every raw observation is collected on a polite schedule that never trips a source’s rate limiter, normalized to one canonical competitor-rate shape that resolves meal-plan, occupancy, and currency differences before comparison, and gated on staleness and outlier bounds so a single bad scrape can never move a live price. Anything that fails those guards is quarantined with full context rather than blended silently into the benchmark. The sections below build each piece end to end for the Python engineer who owns the shopper and the revenue manager who has to trust its output.

Architecture & prerequisites

The pipeline is a directed flow with four stages and two side-channels. A scheduler emits shop requests for a matrix of (competitor_property_id, check_in, los, occupancy) tuples; an async collector fetches each politely and captures the raw response for audit; a normalizer resolves every observation onto the canonical shape; and a guard layer applies staleness and outlier checks before the surviving benchmark is written to the store the pricing engine reads. Rejected observations divert to a quarantine sink, and every raw response is tapped to an append-only audit blob so any published benchmark can be traced back to the exact bytes it came from.

Competitor rate-shopping ingestion and normalization data flow A left-to-right flow. A polite scheduler emits shop requests for a matrix of competitor property, check-in date, length of stay and occupancy tuples. An async collector fetches each source — Booking.com search, an Expedia meta feed and a rate-shopping vendor — and captures the raw response. Raw observations pass into a normalizer that resolves meal plan, occupancy and currency onto one canonical competitor-rate shape. Normalized rates then pass through a guard layer that applies staleness and outlier checks. Rates that pass are written to the benchmark store the pricing engine reads; rates that fail divert to a quarantine sink. A dashed tap from the collector writes every raw response to an append-only audit blob. Competitor rate-shopping ingestion pipeline Shopped politely, normalized to one shape, guarded on staleness and outliers before it can move a price. Sources shopped Polite scheduler shop matrix + jitter Async collector httpx + semaphore captures raw bytes Normalizer meal · occupancy · currency → canonical Guard layer staleness · outlier bounds Benchmark store read by pricing engine Quarantine sink stale / outlier rejects Append-only audit blob raw response per shop pass reject tap
Figure 1: the ingestion path — the scheduler paces a shop matrix, the collector fetches politely and captures raw bytes, the normalizer resolves every observation onto one canonical shape, and the guard layer admits only current, in-band rates to the benchmark store. Rejects quarantine; every raw response is tapped to an immutable audit blob.

Inputs, outputs, and environment for the reference implementation:

The one non-negotiable prerequisite is that collection is polite by construction. A rate shopper that fans out at full speed is indistinguishable from an attack and will get the property’s IP blocked, so pacing, jitter, and per-host concurrency limits are part of the design, not an afterthought — the mechanics are built in full on scheduling competitor rate shops with async httpx.

Implementation

Step 1 — Define the shop matrix and emit paced requests

Start from the demand the revenue team actually prices against: a comp set of competitor properties, the date horizon, and the occupancy/length-of-stay combinations that matter. Expand that into an explicit request matrix so collection is deterministic and auditable rather than an opaque crawl.

python
from dataclasses import dataclass
from datetime import date, timedelta
import structlog

log = structlog.get_logger()

@dataclass(frozen=True)
class ShopRequest:
    competitor_property_id: str   # e.g. "BKG-CMP-4471"
    check_in: date
    los: int                      # length of stay in nights
    occupancy_basis: int          # guests the quote is priced for

def build_shop_matrix(comp_set: list[str], horizon_days: int,
                      los_options: tuple[int, ...] = (1, 2),
                      occupancies: tuple[int, ...] = (2,)) -> list[ShopRequest]:
    today = date.today()
    matrix = [
        ShopRequest(cid, today + timedelta(days=d), los, occ)
        for cid in comp_set
        for d in range(horizon_days)
        for los in los_options
        for occ in occupancies
    ]
    log.info("shop_matrix_built", requests=len(matrix), comp_set=len(comp_set))
    return matrix

Materializing the full matrix up front — rather than looping and fetching inline — lets the scheduler count, pace, and jitter the whole batch as one unit, so the request rate against any single host is a property of the plan instead of an emergent surprise.

Step 2 — Collect each observation and capture the raw response

Every fetch returns both the extracted rate fields and the untouched raw payload. Capturing the raw bytes is what makes a published benchmark defensible later: when a revenue manager disputes a number, you replay the exact response the shopper saw rather than re-scraping a source that has since changed.

python
import hashlib, json
from datetime import datetime, timezone

@dataclass(frozen=True)
class RawObservation:
    request: ShopRequest
    source: str                   # "booking_com" | "expedia" | "vendor_x"
    captured_at: datetime
    raw_amount: str               # as-quoted, still a string to avoid early float coercion
    raw_currency: str
    tax_inclusive: bool
    meal_plan: str                # source's own meal-plan token
    quoted_occupancy: int
    raw_blob: dict                # full extracted response, for audit

def audit_key(obs: RawObservation) -> str:
    ident = f"{obs.source}|{obs.request.competitor_property_id}|" \
            f"{obs.request.check_in}|{obs.request.los}|{obs.captured_at.isoformat()}"
    return hashlib.sha256(ident.encode()).hexdigest()[:16]

def persist_audit(obs: RawObservation, sink: dict) -> str:
    key = audit_key(obs)
    sink[key] = json.dumps(obs.raw_blob, default=str)   # append-only; never overwritten
    log.info("raw_observation_captured", source=obs.source,
             competitor=obs.request.competitor_property_id, audit_key=key)
    return key

Keeping raw_amount as a string at capture time is deliberate: coercing to float here would bake in a rounding error before the normalizer has decided the correct precision and currency, so the raw value is preserved verbatim and only becomes a Decimal inside the normalization step.

Step 3 — Normalize meal plan, occupancy, and currency onto the canonical shape

A raw observation is not comparable to your own rate until it is on the same basis. The normalizer strips or adds tax, converts currency against a pinned snapshot, and adjusts for meal-plan and occupancy differences so every benchmark cell speaks one language. The full tax and gross-versus-net reconciliation is specified on normalizing competitor rates: gross vs net; here is the shape of the transform.

python
from decimal import Decimal, ROUND_HALF_EVEN

# Value of the standard meal plan per person, in the property's base currency,
# used to strip a source's board back to a room-only basis for comparison.
MEAL_PLAN_CREDIT = {"ro": Decimal("0"), "bb": Decimal("18"), "hb": Decimal("42")}

def normalize(obs: RawObservation, fx_to_base: Decimal,
              base_currency: str, target_occupancy: int) -> Decimal:
    amount = Decimal(obs.raw_amount)
    # 1. currency: convert against the pinned snapshot for this run, never a live lookup
    if obs.raw_currency != base_currency:
        amount = amount * fx_to_base
    # 2. board: subtract the meal-plan credit to reach a room-only basis
    credit = MEAL_PLAN_CREDIT.get(obs.meal_plan, Decimal("0")) * obs.quoted_occupancy
    amount = amount - credit
    # 3. occupancy: only compare like-for-like; a mismatch is flagged, not fudged
    if obs.quoted_occupancy != target_occupancy:
        log.warning("occupancy_mismatch", source=obs.source,
                    quoted=obs.quoted_occupancy, target=target_occupancy)
    return amount.quantize(Decimal("0.01"), rounding=ROUND_HALF_EVEN)

Using Decimal with explicit ROUND_HALF_EVEN at the single point where the number is finalized means two observations that should be equal compare exactly, which matters because the pricing engine will later diff this benchmark against your own rate to the cent.

Step 4 — Write only vetted benchmarks to the store the engine reads

The guard layer (built out in the next section) decides whether a normalized observation is fresh and in-band. Only survivors are written to the benchmark store, and each carries its captured_at timestamp so the engine can apply its own staleness policy on read.

python
def write_benchmark(store: dict, room_type_code: str, req: ShopRequest,
                    normalized: Decimal, obs: RawObservation) -> None:
    cell = (room_type_code, req.check_in.isoformat(), req.los, req.occupancy_basis)
    store[cell] = {
        "competitor_rate": str(normalized),   # Decimal serialized as string, never float
        "source": obs.source,
        "captured_at": obs.captured_at.isoformat(),
        "occupancy_basis": req.occupancy_basis,
    }
    log.info("benchmark_written", cell=cell, rate=str(normalized), source=obs.source)

Keying the store on the full (room_type_code, check_in, los, occupancy_basis) tuple rather than just room type prevents the classic collision where a one-night and a two-night quote for the same room overwrite each other, which would leave the engine comparing your flexible nightly rate against a competitor’s discounted multi-night average.

Schema & data contracts

Before any observation reaches the benchmark store it passes through a Pydantic v2 model that rejects malformed or nonsensical rates at the boundary. This is the canonical competitor-rate contract every source adapter must produce, and it encodes the invariants that make a benchmark safe to price against — a positive amount, a valid ISO currency, a known meal-plan token, and a capture timestamp that is actually in the past.

python
from datetime import datetime, timezone, date
from decimal import Decimal
from pydantic import BaseModel, Field, field_validator, model_validator

class CompetitorRate(BaseModel):
    competitor_property_id: str = Field(pattern=r"^[A-Z]{3}-[A-Z0-9]{3,}-\d{2,}$")
    room_type_code: str = Field(pattern=r"^[A-Z0-9_]+$")
    source: str
    check_in: date
    los: int = Field(ge=1, le=30)
    occupancy_basis: int = Field(ge=1, le=6)
    amount: Decimal = Field(gt=0, decimal_places=2)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    meal_plan: str
    tax_inclusive: bool
    captured_at: datetime

    @field_validator("source")
    @classmethod
    def known_source(cls, v: str) -> str:
        allowed = {"booking_com", "expedia", "agoda", "vendor_x"}
        if v not in allowed:
            raise ValueError(f"unknown rate-shopping source: {v}")
        return v

    @field_validator("meal_plan")
    @classmethod
    def known_meal_plan(cls, v: str) -> str:
        if v not in {"ro", "bb", "hb", "fb", "ai"}:
            raise ValueError(f"unknown meal_plan token: {v}")
        return v

    @model_validator(mode="after")
    def capture_not_in_future(self) -> "CompetitorRate":
        if self.captured_at > datetime.now(timezone.utc):
            raise ValueError("captured_at is in the future — clock skew or bad parse")
        return self

Rejecting a future captured_at at the contract boundary catches a whole class of silent bugs — a source that returns a local timestamp misread as UTC, or a parser that defaulted a missing field to now() — before a stale-looking-but-actually-broken rate can masquerade as the freshest observation in the store. This contract is the local specialization of the property-wide rules in data schema standardization.

Error handling & retry strategy

Rate-shopping fetches fail in ways that are distinct from a normal API sync, because you are often collecting from a source that would rather you did not. The disposition of each outcome has to distinguish “try again gently” from “this source is telling me to stop” from “this data is unusable” — and the last of those must never reach the benchmark store.

The staleness guard is the one that protects revenue most directly, because a benchmark the engine believes is current but is actually hours old will price against a market that has already moved:

python
from datetime import datetime, timezone, timedelta

def is_fresh(captured_at: datetime, max_age: timedelta = timedelta(hours=6)) -> bool:
    age = datetime.now(timezone.utc) - captured_at
    fresh = age <= max_age
    if not fresh:
        log.warning("benchmark_stale", captured_at=captured_at.isoformat(),
                    age_minutes=round(age.total_seconds() / 60))
    return fresh

Making staleness an explicit, logged decision rather than an implicit read-time assumption means a source that quietly stops responding degrades loudly — the quarantine fills and the metric fires — instead of freezing an old benchmark in place and letting the engine price against yesterday’s market. When a cell has no fresh observation, the engine falls back to the last known-good rate and flags the decision, exactly as the dynamic pricing rule engines that consume this feed require.

Verification & testing

Prove the ingestion path is comparable-in, comparable-out before it feeds a live pricing run. Assert on normalization arithmetic, contract rejection, and staleness disposition rather than eyeballing a scraped number.

python
import pytest
from decimal import Decimal
from datetime import datetime, timezone, timedelta
from pydantic import ValidationError

def test_currency_and_board_normalization():
    obs = RawObservation(
        request=ShopRequest("BKG-CMP-4471", date(2026, 8, 1), 1, 2),
        source="booking_com", captured_at=datetime.now(timezone.utc),
        raw_amount="7200.00", raw_currency="THB", tax_inclusive=True,
        meal_plan="bb", quoted_occupancy=2, raw_blob={},
    )
    # 7200 THB * 0.026 = 187.20, minus 2 * 18 breakfast credit = 151.20
    result = normalize(obs, fx_to_base=Decimal("0.026"),
                       base_currency="EUR", target_occupancy=2)
    assert result == Decimal("151.20")

def test_zero_amount_is_rejected_at_contract():
    with pytest.raises(ValidationError):
        CompetitorRate(
            competitor_property_id="BKG-CMP-4471", room_type_code="DLX_KING",
            source="booking_com", check_in=date(2026, 8, 1), los=1, occupancy_basis=2,
            amount=Decimal("0.00"), currency="EUR", meal_plan="ro",
            tax_inclusive=False, captured_at=datetime.now(timezone.utc),
        )

def test_stale_observation_is_flagged():
    old = datetime.now(timezone.utc) - timedelta(hours=8)
    assert is_fresh(old) is False

The three assertions map to the three ways a competitor benchmark silently corrupts a price: a normalization that compares different currencies or board bases, a nonsensical amount that slips past the boundary, and a stale value that the engine trusts as current. A green run proves each door is closed. In production, also assert the quarantine sink receives exactly one record per rejected observation so no failed shop is dropped on the floor, and reconcile the audit blob count against the shop-matrix size to confirm every planned shop actually ran.

Troubleshooting

Symptom Root cause Fix
Benchmark suddenly undercuts the whole comp set A source switched to net (tax-exclusive) quoting and the normalizer still treats it as gross Read tax_inclusive per observation and reconcile on a single basis; see the gross-vs-net normalization guide
Rates jump every morning then settle FX was looked up live per observation instead of pinned per run, so early and late shops used different rates Pin one FX snapshot at run start and pass it through normalization; never call a live FX API mid-batch
Comp set silently shrinks over days A source began returning 403/CAPTCHA and its cells fell to quarantine unnoticed Alert on per-source quarantine rate and freshness coverage, not just on total row count
One-night and multi-night quotes overwrite each other Benchmark store keyed on room type only Key the store on (room_type_code, check_in, los, occupancy_basis)
Engine prices against a market that already moved Stale benchmark read as current because staleness was an implicit assumption Stamp captured_at, enforce a max-age guard on write and read, fall back to last known-good on miss

FAQ

How often should competitor rates be shopped?

Match the shop cadence to how fast the pricing engine can act and how tolerant the sources are. A six-hour refresh on the near horizon with a daily pass further out is a common starting point. The staleness guard, not the schedule alone, is what protects revenue: if a shop is missed the benchmark expires and the engine falls back to last known-good rather than pricing against an old number.

Should a competitor rate ever move a price on its own?

No. The ingestion layer only produces a vetted benchmark; the decision to move a price belongs to the dynamic pricing rule engine, which clamps the benchmark to a sane band, checks inventory, and respects floors before acting. A single observation, however fresh, is an input to a rule, never a command.

Why capture the raw response instead of just the extracted rate?

Because sources change and disputes happen. When a revenue manager challenges a published benchmark, replaying the exact captured bytes settles it definitively, whereas re-scraping hits a source that may have changed its layout or price since. The raw blob also lets you re-run a fixed parser over historical shops without re-collecting.

How do you keep rate shopping from getting the property IP blocked?

Pace by construction: bounded per-host concurrency with a semaphore, jittered scheduling so requests never arrive in a detectable burst, honouring Retry-After on any 429, and widening the interval for the rest of a run the moment a throttle response appears. A 403 or CAPTCHA is treated as terminal for that host, not something to retry through. The full scheduler is built on scheduling competitor rate shops with async httpx.

← Back to Rate Parity Monitoring & Revenue Optimization