Revenue Management Pricing & forecasting pipelines
5 min read Ingestion

Data Quality & Schema Contracts

Every wrong price eventually traces back to a bad record that nobody stopped at the door. Data quality and schema contracts are the discipline of refusing malformed OTA and competitor payloads at the ingestion boundary, before they can poison forecasts and pricing. This component sits inside the Data Ingestion & OTA API Integration Workflows pillar as the gatekeeper between raw external feeds and the internal model, working alongside the transport concerns of Async Polling & Pagination Handling and the resilience of Rate Limiting & Retry Strategies. Without it, a renamed field or a currency-format change propagates silently until a revenue manager notices rates are wrong.

Contract gate at ingestion Raw payloads pass a schema contract; valid records continue to the store while invalid ones divert to a quarantine queue. Contract gate at ingestion Raw payload OTA / scrape Contract validate Store valid record Quarantine invalid record pass fail

Taxonomy and schema governance

A schema contract is a versioned, machine-checkable definition of what a valid record looks like: field names, types, ranges, and cross-field invariants. For a rate record that means a positive base rate, an ISO-4217 currency, an ISO-8601 date, a check-out strictly after check-in, and a rate plan that maps to the internal taxonomy from Rate Plan Structuring & Mapping. Contracts are versioned so a provider adding a field is a compatible change while a provider renaming one is a breaking change that must be handled explicitly. The contract is the single authority on validity; downstream consumers trust that anything past the gate already conforms, which is what lets the pricing and forecasting layers skip defensive re-validation on every read.

Ingestion and normalization

Validation and normalization are one step, not two: a record is coerced into canonical form and rejected if coercion is impossible. Currencies are upper-cased and checked against the ISO-4217 set, dates are parsed to ISO-8601 in a known timezone, and monetary amounts are converted to integer minor units. Upserts are idempotent, keyed on a natural identity (property, room type, stay date, source), so re-ingesting the same payload updates in place rather than duplicating. Schema drift — a field appearing, disappearing, or changing type relative to the contract version — is detected here and routed for handling rather than silently absorbed. The staging layer keeps the raw payload alongside the normalized record so a quarantined item can be re-examined and replayed after a contract fix.

Core implementation

A Pydantic model expresses the contract; a validation function coerces and routes. Valid records return normalized; invalid ones are logged with the specific violation and handed to the quarantine sink.

python 
from __future__ import annotations

import logging
from datetime import date
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, field_validator

logger = logging.getLogger("ingestion.contract")

_ISO_4217 = {"USD", "EUR", "GBP", "JPY", "AUD", "CAD"}   # extend from a reference table


class RateRecord(BaseModel):
    property_id: str
    room_type: str
    check_in: date
    check_out: date
    base_rate_cents: int = Field(gt=0)
    currency: str

    @field_validator("currency")
    @classmethod
    def known_currency(cls, v: str) -> str:
        up = v.upper()
        if up not in _ISO_4217:
            raise ValueError(f"unknown currency {v!r}")
        return up

    @field_validator("check_out")
    @classmethod
    def after_check_in(cls, v: date, info) -> date:
        ci = info.data.get("check_in")
        if ci is not None and v <= ci:
            raise ValueError("check_out must be after check_in")
        return v


def validate_record(payload: dict) -> Optional[RateRecord]:
    try:
        return RateRecord(**payload)
    except ValidationError as exc:
        logger.warning("quarantined record %s: %s",
                       payload.get("property_id"), exc.errors())
        return None

Returning None rather than raising lets the caller batch-process a feed, quarantining the few bad records while committing the good ones — a single malformed row never blocks an entire ingestion cycle.

Cross-system dependencies

Upstream, contracts guard the raw feeds arriving from OTA integrations and from Competitor Rate Scraping Pipelines. Downstream, every consumer benefits: Occupancy Forecasting & Demand Analytics trains only on validated history, and Dynamic Pricing Rule Engines & Optimization prices only from conforming records. The contract’s currency and rate-plan rules are inherited directly from the core architecture pillar, so a change to the canonical taxonomy propagates into the contract as a version bump rather than a scattered set of ad-hoc checks.

Operational governance

Contracts are code: versioned, reviewed, and tested against a corpus of real historical payloads in CI so a tightening never silently quarantines a legitimate stream. Drift detection runs continuously, comparing incoming field sets against the active contract and alerting when an unrecognized field appears or an expected one vanishes. The quarantine sink is monitored like any other queue — depth, age of oldest item, and rejection reason distribution — because a spike in a single reason is usually a provider-side change, not random noise. Audit logs retain the raw payload and the violation for every rejection so a contract fix can be validated by replaying the quarantined backlog.

Troubleshooting and failure modes

Symptom Likely cause Remediation
Quarantine depth spikes Provider renamed or retyped a field Bump contract version; add a compatibility shim
Valid records rejected Contract tightened without a corpus test Test against historical payloads before release
Duplicate rows downstream Non-idempotent upsert key Key on natural identity + source
Silent bad prices Field absent from contract entirely Add the field and a range check; replay history

Conclusion

Production readiness here is measured by what never reaches the model: a versioned contract, coercion-as-validation, an idempotent upsert, and a monitored quarantine make bad data a visible, replayable event rather than a silent corruption. That gate is what lets every downstream pillar trust the records it reads.