InventoryAllocation

Building Batch Reconciliation Scripts for Daily Syncs

This page is the operational build guide for the runnable script that executes a rate-parity audit every night across a portfolio of properties: how to shape the work into idempotent units, extract both sides under a concurrency budget, run the diff per property, and schedule the whole thing to resume cleanly if a run dies mid-flight. It sits under Batch Reconciliation Workflows, which specifies the canonical schema and the vectorized diff engine this script drives — here we focus on the orchestration, sharding, and scheduling that turn that engine into a dependable daily job. It is the periodic backstop to the near-real-time async polling loop: where polling catches minute-to-minute deltas on the fly, this nightly job is what polling escalates a whole property to whenever accumulated drift exceeds tolerance, resyncing every open room-night atomically instead of in a risky burst of piecemeal corrections.

Prerequisites & environment

The script assumes you already have a working diff engine and a canonical record contract from the parent workflow; this page wires them into a scheduled, portfolio-wide runner. Pin these versions so the async and Polars behaviour below is reproducible:

Every room_type_code and rate_plan_code the script compares must already conform to your rate plan taxonomy and the shared standardized JSON payload shape. Running this job upstream of those contracts manufactures false discrepancies on every property.

Daily-sync reconciliation runner orchestration flow A per-timezone off-peak scheduler fires the nightly run at 02:30 local property time. A job planner turns the portfolio into idempotent work units keyed by a deterministic batch_id over an incremental booking-horizon date window, then fans out across N properties into a bounded-concurrency extractor. An asyncio semaphore caps in-flight extractions to the OTA rate-limit budget while each property shard pulls its PMS and channel-manager snapshots inside a single gather. Each shard is normalized and diffed independently in Polars, then concatenated into one aggregated portfolio report that is upserted to a checkpoint store with ON CONFLICT on batch_id. A violet dashed feedback path shows the next run reading the checkpoint store to skip any batch_id already persisted, so a crashed run resumes cleanly, and a pink annotation marks the concurrency cap on the fan-out. resume — skip batch_id already persisted concurrency cap Semaphore(N) ≤ OTA budget fan-out ×N properties sharded frames Off-peak scheduler (per timezone) CronTrigger 02:30 local property time Job planner idempotent batch_id work units incremental horizon date window Bounded-concurrency extractor property shard 1 property shard 2 property shard N PMS ⇄ channel 1 asyncio.gather PMS ⇄ channel 1 asyncio.gather PMS ⇄ channel 1 asyncio.gather Per-property Polars diff normalize → reconcile per shard flat peak memory as portfolio grows Portfolio report → checkpoint store pl.concat then persist ON CONFLICT (batch_id) upsert

Step-by-step implementation

The runner is four moving parts: a planner that turns the portfolio into idempotent work units, a bounded-concurrency extractor, a per-property diff pass, and a resumable scheduler. Wire them in the order below.

Step 1 — Plan idempotent per-property work units

Each nightly run reconciles a bounded date window (today plus the booking horizon you care about, typically 90–365 nights out) for every property. The unit of work is one property over one window, keyed by a deterministic batch_id so a re-run overwrites rather than duplicates.

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

@dataclass(frozen=True)
class ReconJob:
    property_id: str
    window_start: date
    window_end: date
    batch_id: str

def plan_jobs(property_ids: list[str], horizon_nights: int = 180) -> list[ReconJob]:
    run_day = datetime.now(timezone.utc).date()
    end = run_day + timedelta(days=horizon_nights)
    # batch_id is derived only from (property, run_day) — never from wall-clock time —
    # so a retry on the same night collides on the ON CONFLICT key and refreshes rows.
    return [
        ReconJob(
            property_id=pid,
            window_start=run_day,
            window_end=end,
            batch_id=f"recon_{pid}_{run_day:%Y%m%d}",
        )
        for pid in property_ids
    ]

Deriving batch_id from the run day rather than the run timestamp is the whole trick to idempotent retries: a job that crashed at 02:14 and reruns at 02:40 produces the identical key, so the upsert converges instead of writing a second, conflicting audit.

Why batch_id is derived from the run day, not the run timestamp Two lanes compare the same night's crash-and-retry sequence. In the top lane the batch_id is derived from the run day: a job that crashes at 02:14 and a retry at 02:40 both produce the identical key recon_…_20260702, so the ON CONFLICT upsert converges to one canonical audit. In the bottom lane the batch_id embeds the wall-clock time: the 02:14 attempt keys on …_021400 and the 02:40 retry keys on …_024000, two different keys that diverge into two conflicting audits and double-count every open correction. batch_id from run DAY — idempotent 02:14 crash recon_…_20260702 02:40 retry recon_…_20260702 identical key → converges 1 canonical audit ON CONFLICT (batch_id) upsert refreshes rows batch_id from run TIMESTAMP — duplicates 02:14 crash recon_…_021400 02:40 retry recon_…_024000 keys differ → diverge 2 conflicting audits corrections double-counted on every retry

Step 2 — Extract both sides under a concurrency budget

Fanning out extraction across a large portfolio will trip OTA API rate limits instantly if it is unbounded. Cap in-flight extractions with a semaphore sized to the channel’s published budget, and keep the retry decorator at module level so tenacity compiles the retry machinery once.

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

log = structlog.get_logger()

# Module-level decorator: defining @retry inside the caller would rebuild the retry
# state on every invocation and reset the attempt counter, defeating stop_after_attempt.
@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError)),
    reraise=True,
)
async def _fetch(client: httpx.AsyncClient, url: str, headers: dict) -> dict:
    resp = await client.get(url, headers=headers)
    resp.raise_for_status()
    return resp.json()

async def extract_job(client, sem: asyncio.Semaphore, job, endpoints, headers) -> dict:
    async with sem:  # global cap across the whole portfolio fan-out
        pms_raw, channel_raw = await asyncio.gather(
            _fetch(client, endpoints.pms(job), headers),
            _fetch(client, endpoints.channel(job), headers),
        )
    for source, payload in (("pms", pms_raw), ("channel", channel_raw)):
        digest = hashlib.sha256(repr(payload).encode()).hexdigest()
        # Stage the raw hash BEFORE parsing so the run is replayable at this exact byte state.
        log.info("snapshot_staged", batch_id=job.batch_id,
                 property_id=job.property_id, source=source, payload_sha256=digest)
    return {"job": job, "pms": pms_raw, "channel": channel_raw}

Hashing and logging each raw payload before it is parsed means a discrepancy raised weeks later can be replayed against the exact bytes that produced it — the SHA-256 digest is the point-in-time anchor that makes the audit defensible, not just the parsed frame.

Step 3 — Diff each property and aggregate the portfolio

Each staged pair is normalized and passed to the diff engine from the parent workflow. Running the diff per property shard rather than over one giant concatenated frame keeps peak memory flat as the portfolio grows and lets a single property’s failure be quarantined without losing the rest of the run.

python
import polars as pl

async def run_portfolio(jobs, endpoints, headers, max_concurrency: int = 8) -> pl.DataFrame:
    sem = asyncio.Semaphore(max_concurrency)
    reports: list[pl.DataFrame] = []
    async with httpx.AsyncClient(timeout=30.0, http2=True) as client:
        async with asyncio.TaskGroup() as tg:  # one failed extract raises, siblings finish
            tasks = [tg.create_task(extract_job(client, sem, j, endpoints, headers))
                     for j in jobs]
        for task in tasks:
            staged = task.result()
            pms = normalize(staged["pms"], "pms")           # from the parent workflow
            channel = normalize(staged["channel"], "channel")
            diff = reconcile(pms, channel).with_columns(
                pl.lit(staged["job"].batch_id).alias("batch_id"),
                pl.lit(staged["job"].property_id).alias("property_id"),
            )
            reports.append(diff)
    return pl.concat(reports) if reports else pl.DataFrame()

Sharding the diff per property caps working-set memory at a single property’s room-nights instead of the whole portfolio’s, which is what lets one runner reconcile hundreds of properties inside a fixed off-peak window without an out-of-memory kill.

Step 4 — Schedule off-peak and make the run resumable

Fire the job in each property’s local off-peak window (02:00–04:00 property time), not one global UTC time, and checkpoint completed batch_ids so a restart skips work already persisted. The ON CONFLICT upsert covered in the parent workflow makes re-running an unfinished job safe.

python
from apscheduler.schedulers.asyncio import AsyncIOScheduler
from apscheduler.triggers.cron import CronTrigger

def schedule_nightly(scheduler: AsyncIOScheduler, properties_by_tz: dict[str, list[str]]):
    for tz, property_ids in properties_by_tz.items():
        scheduler.add_job(
            nightly_run, CronTrigger(hour=2, minute=30, timezone=tz),
            args=[property_ids], id=f"recon_{tz}", replace_existing=True,
            misfire_grace_time=3600,  # a late-firing job (host asleep) still runs, not skipped
        )

async def nightly_run(property_ids: list[str]) -> None:
    jobs = [j for j in plan_jobs(property_ids) if not checkpoint_done(j.batch_id)]
    diffs = await run_portfolio(jobs, ENDPOINTS, auth_headers())
    for batch_id in diffs["batch_id"].unique():
        persist(conn, diffs.filter(pl.col("batch_id") == batch_id), batch_id)  # parent workflow
        checkpoint_mark(batch_id)
    log.info("nightly_complete", properties=len(jobs), discrepancies=diffs.height)

Scheduling per property timezone rather than at a single UTC hour is what keeps the read-only audit off the peak booking window everywhere — a 02:30 UTC run lands at 21:30 the previous evening in a US property, squarely inside its busiest reservation traffic.

Gotchas & production notes

Verification snippet

Confirm the runner’s two load-bearing properties — that the planner is idempotent and that the aggregated report carries the keys persistence relies on — before trusting a green scheduler log.

python
import polars as pl

def test_batch_id_is_stable_within_a_run_day() -> None:
    # Same property, same day → identical batch_id, so a retry upserts in place.
    a = plan_jobs(["prop_0a1b2c3d"])[0].batch_id
    b = plan_jobs(["prop_0a1b2c3d"])[0].batch_id
    assert a == b and a.startswith("recon_prop_0a1b2c3d_")

def test_portfolio_report_carries_property_and_batch_keys() -> None:
    diff = pl.DataFrame({"discrepancy": ["RATE_PARITY_VIOLATION"]}).with_columns(
        pl.lit("recon_prop_0a1b2c3d_20260702").alias("batch_id"),
        pl.lit("prop_0a1b2c3d").alias("property_id"),
    )
    # Persistence keys on (batch_id, property_id, ...); both must survive aggregation.
    assert {"batch_id", "property_id"}.issubset(diff.columns)

test_batch_id_is_stable_within_a_run_day()
test_portfolio_report_carries_property_and_batch_keys()

Asserting batch_id stability directly tests the idempotency guarantee the whole schedule depends on — if it ever varies within a run day, retries silently double every open correction ticket. In production, also assert that the count of persisted rows equals the count of discrepancy != OK rows per property (no silent write loss) and that checkpoint_done short-circuits an already-finished batch_id on restart.

← Back to Batch Reconciliation Workflows