InventoryAllocation

Building Batch Reconciliation Scripts for Daily Syncs

Daily rate parity reconciliation between Property Management Systems (PMS) and Channel Managers requires deterministic batch processing to eliminate sync drift before it impacts OTA visibility and booking conversion. Revenue managers and Python automation engineers must implement idempotent ingestion pipelines that normalize disparate payload schemas, enforce strict matching keys, and generate auditable compliance logs. The foundation of this architecture relies on structured API Sync & Data Ingestion Workflows that pull snapshot data at off-peak intervals, typically between 02:00 and 04:00 local property time, to avoid contention with live booking transactions and minimize API throttling during peak reservation windows.

Deterministic Ingestion & Payload Staging

Reconciliation drift originates from non-deterministic data pulls. Mid-batch OTA price updates, channel manager cache invalidation, and timezone misalignment between systems can corrupt parity baselines if ingestion is not strictly versioned. The ingestion layer must query both PMS and Channel Manager endpoints using paginated requests with exponential backoff and jitter. Every raw response is immediately hashed using SHA-256 and persisted to a staging table alongside a batch_id, source_system, and ingestion_timestamp. This guarantees payload immutability and enables point-in-time replay when downstream validation fails.

python
import hashlib
import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

async def fetch_rate_snapshot(client: httpx.AsyncClient, endpoint: str, params: dict) -> dict:
    @retry(
        stop=stop_after_attempt(4),
        wait=wait_exponential(multiplier=1, min=2, max=10),
        retry=retry_if_exception_type(httpx.HTTPStatusError),
        reraise=True
    )
    async def _paginated_pull() -> httpx.Response:
        response = await client.get(endpoint, params=params)
        response.raise_for_status()
        return response

    response = await _paginated_pull()
    payload = response.json()
    checksum = hashlib.sha256(response.content).hexdigest()
    return {"payload": payload, "checksum": checksum, "fetched_at": response.headers.get("Date")}

Storing raw payloads with cryptographic checksums prevents silent data corruption during transformation. When combined with idempotent batch_id generation (typically YYYYMMDD_HHMMSS_SOURCE), the pipeline can safely resume interrupted runs without duplicating records or skipping rate updates.

Schema Normalization & Modifier Stripping

Hospitality APIs rarely share identical rate structures. The PMS may return gross rates with embedded VAT, while the Channel Manager returns net rates with separate tax objects. Normalization must isolate the baseline parity contract by stripping dynamic pricing modifiers, promotional overlays, and member-only discounts. A strict Pydantic v2 model enforces schema validation at ingestion time:

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

class NormalizedRate(BaseModel):
    room_type_id: str
    date: date
    rate_plan_code: str
    currency: str
    base_rate: Decimal = Field(ge=0)
    tax_inclusive_flag: bool
    availability_status: str  # "open", "closed_to_arrival", "stop_sell"

    @field_validator("rate_plan_code")
    @classmethod
    def normalize_code(cls, v: str) -> str:
        return v.strip().upper().replace(" ", "_")

During normalization, currency conversion is deferred until the matching phase to preserve source-of-truth precision. The tax_inclusive_flag is explicitly parsed from vendor-specific fields (e.g., pricing.tax_mode vs rate_details.incl_taxes) to ensure accurate delta calculations later in the pipeline.

Core Matching Algorithm & Tolerance Thresholds

The reconciliation engine performs a left-join against the PMS baseline dataset using a composite key: (room_type_id, rate_plan_code, date). Absolute rate deltas exceeding a configurable tolerance threshold—typically 0.5%—trigger parity flags. This margin accounts for currency rounding, VAT recalculation variances, and OTA commission structures that differ by distribution channel.

Edge cases must be handled explicitly. Closed-to-arrival (CTA) restrictions, stop-sell overrides, and length-of-stay (LOS) minimums are parsed as boolean constraints rather than numeric rate values. When a Channel Manager returns a null rate for a restricted date, the engine classifies it as a compliance pass rather than a parity violation. Legacy rate plan codes are resolved via a preloaded mapping dictionary to prevent false-positive drift alerts caused by deprecated naming conventions or property-level rebranding.

python
from typing import Dict, Tuple, List
from decimal import Decimal, ROUND_HALF_UP

def evaluate_parity(
    pms_record: NormalizedRate,
    cm_record: NormalizedRate | None,
    tolerance_pct: Decimal = Decimal("0.005"),
    legacy_map: Dict[str, str] = None
) -> Dict:
    if cm_record is None:
        return {"status": "missing_in_cm", "action": "flag_for_audit"}

    if cm_record.rate_plan_code in (legacy_map or {}):
        cm_record.rate_plan_code = legacy_map[cm_record.rate_plan_code]

    if pms_record.availability_status in ("closed_to_arrival", "stop_sell"):
        if cm_record.base_rate == 0 or cm_record.availability_status == pms_record.availability_status:
            return {"status": "compliant_restriction", "action": "pass"}

    delta = abs(pms_record.base_rate - cm_record.base_rate)
    threshold = pms_record.base_rate * tolerance_pct

    if delta <= threshold:
        return {"status": "compliant", "delta": delta, "action": "pass"}

    return {"status": "parity_violation", "delta": delta, "threshold": threshold, "action": "escalate"}

Domain-Specific Error Handling & Circuit Breakers

Silent failures in hospitality integrations directly impact revenue. Error handling must be categorized by failure domain to prevent cascading data corruption. Network timeouts and 5xx server errors trigger immediate retry queues with jittered delays, while 401 Unauthorized responses invoke OAuth2 token refresh routines before resuming the batch. Rate limit exhaustion requires a circuit breaker that pauses ingestion, logs the remaining quota window, and schedules a deferred continuation rather than forcing a hard failure.

Payload validation errors are strictly isolated from parity mismatches. Malformed JSON, missing required fields, or invalid date formats route to a quarantine queue with structured diagnostic metadata. This separation ensures that a single malformed rate object does not halt the entire reconciliation run. Implementing a token-aware retry strategy aligns with modern Batch Reconciliation Workflows that prioritize graceful degradation over aggressive polling.

python
import structlog
from time import time

logger = structlog.get_logger()

class RateLimitCircuitBreaker:
    def __init__(self, threshold: int = 429, cooldown: int = 300):
        self.threshold = threshold
        self.cooldown = cooldown
        self.tripped_at: float | None = None

    def evaluate(self, status_code: int) -> bool:
        if status_code == self.threshold:
            self.tripped_at = time()
            logger.warning("rate_limit_exhausted", cooldown_seconds=self.cooldown)
            return True
        if self.tripped_at and (time() - self.tripped_at) < self.cooldown:
            return True
        self.tripped_at = None
        return False

Structured Logging & Compliance Reporting

Auditable compliance logs are non-negotiable in rate parity automation. Every ingestion step, validation result, and parity flag must emit structured JSON logs containing trace_id, batch_id, source_system, composite_key, and action_taken. Using structlog ensures logs are machine-readable for downstream SIEM integration and human-readable for revenue manager dashboards.

The final output generates a compliance report containing:

This telemetry enables revenue managers to distinguish between systemic sync drift and isolated OTA pricing anomalies. By anchoring the pipeline to deterministic staging, strict schema enforcement, and domain-aware error routing, hospitality tech teams can maintain daily rate parity at scale without manual intervention.