Revenue Management Pricing & forecasting pipelines
5 min read Ingestion

Detecting schema drift in OTA payloads

OTA providers change their payloads without asking. A field gets renamed, a nested object gains a level, a string becomes a number — and if your pipeline notices only when prices go wrong, the drift has already cost you. This guide builds a schema-drift detector that flags contract-breaking changes at ingestion time, implementing the drift-detection concern of Data Quality & Schema Contracts within the Data Ingestion & OTA API Integration Workflows pillar.

Prerequisites

  • Python 3.11+
  • Standard library logging, dataclasses, hashlib
  • A registered schema contract (field names and types) per feed source
  • A sample of recent valid payloads to seed the expected field set
  • A sink for drift alerts (your observability stack)

Step 1 — Capture the expected field signature

Drift is a diff against an expected signature. Represent the contract as a set of field_path -> type_name pairs, flattening nested objects so a change at any depth is visible. Seeding it from the contract (not from live traffic) keeps the baseline authoritative.

python 
from __future__ import annotations

import logging
from typing import Any

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


def flatten_types(obj: Any, prefix: str = "") -> dict[str, str]:
    """Map each leaf path to its Python type name."""
    out: dict[str, str] = {}
    if isinstance(obj, dict):
        for key, value in obj.items():
            out.update(flatten_types(value, f"{prefix}{key}."))
    elif isinstance(obj, list):
        if obj:                                   # inspect first element as representative
            out.update(flatten_types(obj[0], f"{prefix}[]."))
    else:
        out[prefix.rstrip(".")] = type(obj).__name__
    return out

Step 2 — Diff a live payload against the expected signature

Compare each incoming payload’s signature to the expected one, classifying differences as added fields, removed fields, or type changes. Type changes and removals of required fields are breaking; additions are usually compatible but still worth reporting.

python 
from dataclasses import dataclass, field


@dataclass
class Drift:
    added: dict[str, str] = field(default_factory=dict)
    removed: dict[str, str] = field(default_factory=dict)
    retyped: dict[str, tuple[str, str]] = field(default_factory=dict)

    def is_breaking(self, required: set[str]) -> bool:
        return bool(self.retyped) or bool(set(self.removed) & required)


def detect_drift(expected: dict[str, str], payload: dict) -> Drift:
    actual = flatten_types(payload)
    drift = Drift()
    for path, typ in actual.items():
        if path not in expected:
            drift.added[path] = typ
        elif expected[path] != typ:
            drift.retyped[path] = (expected[path], typ)
    for path, typ in expected.items():
        if path not in actual:
            drift.removed[path] = typ
    return drift

Step 3 — Alert on breaking drift, log the rest

Route breaking drift to an alert immediately — a retyped price field is a production incident — while recording compatible additions for review so the contract can be extended deliberately.

python 
def check_payload(expected: dict[str, str], required: set[str], payload: dict) -> bool:
    drift = detect_drift(expected, payload)
    if drift.is_breaking(required):
        logger.error("breaking schema drift: retyped=%s removed=%s",
                     drift.retyped, {k: drift.removed[k] for k in set(drift.removed) & required})
        return False
    if drift.added or drift.removed:
        logger.info("compatible drift: added=%s removed=%s", drift.added, drift.removed)
    return True

Separating breaking from compatible drift is the judgment the pipeline needs: a new optional field should not page anyone at 3 a.m., but a base_rate that arrived as a string instead of a number absolutely should.

Verification and testing

python 
def test_retyped_price_is_breaking() -> None:
    expected = {"property_id": "str", "base_rate": "int", "currency": "str"}
    required = {"property_id", "base_rate"}
    good = {"property_id": "h1", "base_rate": 18000, "currency": "USD"}
    bad = {"property_id": "h1", "base_rate": "18000", "currency": "USD"}   # string!
    assert check_payload(expected, required, good) is True
    assert check_payload(expected, required, bad) is False

The failing case is the one that matters: a silently stringified rate is caught at the gate instead of surfacing as mispriced inventory hours later.

Common pitfalls and edge cases

  • Empty lists hide element drift. flatten_types cannot infer a list element’s type when the list is empty; treat missing element types as unknown, not as removed.
  • Nullable fields. A field that is sometimes null flips type to NoneType; allow a declared set of types per path rather than a single one.
  • Sampling one element. Inspecting only the first list element misses heterogeneous arrays; sample several in high-risk feeds.
  • Alert fatigue. Treating every added field as breaking trains people to ignore alerts; reserve paging for retypes and required removals.
  • Baseline from live traffic. Seeding the expected signature from live data bakes in existing drift; seed from the reviewed contract.