Excel Budget Template Sync for Grants

This guide is part of the Data Ingestion & Grant Parsing Workflows reference. Excel Budget Template Sync is the discrete ingestion gate that accepts funder-mandated .xlsx/.xls budget workbooks, proves their structure against a frozen schema, and emits either a validated canonical payload or a deterministic quarantine directive.

The scope is strictly confined to the ingestion, structural validation, and compliance verification of registered Excel budget templates. This stage does not perform narrative extraction, cross-grant financial aggregation, currency conversion, or downstream reconciliation. Binary narrative extraction belongs to PDF Grant Application Parsing; canonical field translation and aliasing belong to Field Mapping & Normalization; portal-sourced financial deltas arrive only after API Polling & Rate Limiting has materialized them locally. Every code path here terminates at exactly one of two outcomes: a validated schema emission, or a deterministic quarantine routing that preserves the original artifact for audit.

Nonprofit operations teams, grant managers, Python automation engineers, and compliance officers should treat this stage as a validation switch, not a repair engine. No heuristic cell-fixing, no silent coercion of out-of-range values, and no merging of malformed layouts happens at this boundary — repair, if any, occurs in an isolated consumer that reads from the quarantine queue.

Operational Boundaries & Stage Isolation #

The sync stage accepts exclusively pre-authorized Excel templates matching a registered funder schema. It explicitly rejects ad-hoc spreadsheets, merged-cell layouts, macro-enabled workbooks (.xlsm), and password-protected files. Boundary enforcement is absolute:

• Ingress boundary: processing begins on receipt of a version-tagged workbook from secure storage. The stage never reaches out to a portal or polls an endpoint — transport concerns are owned upstream by API Polling & Rate Limiting.
• Egress boundary: the stage concludes on successful schema validation or deterministic fallback. Validated payloads advance to normalization; nothing here writes to a ledger or a funder report.
• Failure routing: any deviation from the registered schema triggers immediate quarantine, preserving the original binary alongside a structured validation report. Quarantine directives are consumed by Error Categorization & Logging, which owns the failure taxonomy and the audit sink.

There is zero overlap into reconciliation, forecasting, or reporting. A workbook is either proven sound and handed off, or it is isolated for review — there is no third state.

Prerequisites #

Version pinning is itself a compliance control: an unpinned spreadsheet or validation library can silently change parsing or rejection behavior between deployments and break reproducibility of the audit trail.

• Python: 3.11 or newer (the examples use decimal.Decimal arithmetic and pydantic v2 validator semantics).
• Pinned packages (requirements.txt):
  • openpyxl==3.1.2 — low-level workbook inspection, merged-cell and protection detection.
  • pandas==2.2.2 — tabular structuring and vectorized column access.
  • pydantic==2.7.1 — frozen, declarative schema contracts at the row boundary.
  • pytest==8.2.0 — verification harness.
  • hypothesis==6.100.1 — property-based budget-row fuzzing.
• Environment variables:
  • GRANT_BUDGET_SCHEMA_VERSION — the active frozen schema tag (e.g. v2.1); pinned per funder, never inferred from the file.
  • GRANT_BUDGET_QUARANTINE_PATH — append-only directory where rejected originals and their JSON reports are written.
  • GRANT_INDIRECT_RATE_CAP — the funder’s negotiated indirect-cost rate ceiling, sourced from the agreement, not the workbook.
• Upstream dependency: a registered funder schema (the frozen header registry and category allow-list). This stage consumes that contract; it does not author it.

Validation Sequence #

Canonical Python tooling forms the foundation of the stage. openpyxl is mandated for structural inspection and cell-level metadata, pandas for tabular access, and pydantic for the row-level contract. The gates execute in a fixed, deterministic order — structural failures short-circuit before any data is interpreted, so a malformed layout never reaches numeric validation:

1. Structural integrity (openpyxl): scan for merged cells, protected sheets, hidden rows, embedded objects, pivot tables, and external data connections. Any of these rejects the workbook outright — they signal a non-tabular or tampered artifact.
2. Header verification: match column names against the frozen schema registry. Reject templates with missing, renamed, or reordered required headers. Case-insensitive matching is permitted only when explicitly declared in the funder registry.
3. Type & precision enforcement: validate numeric columns through decimal.Decimal to prevent floating-point drift, enforce ISO-8601 period dates, and validate categorical fields against an enumerated allow-list (PERSONNEL, TRAVEL, EQUIPMENT, INDIRECT, OTHER_DIRECT).
4. Cross-column reconciliation: verify that the declared total equals the arithmetic sum of line items, and that the indirect-cost rate does not exceed the funder’s negotiated cap.

Templates passing all four gates proceed to normalization. Templates failing any gate route to quarantine with a deterministic error code, original file intact.

Core Implementation #

The implementation isolates structural inspection from data validation to maintain strict separation of concerns. It uses type hints throughout, a standard-library structured audit logger (never print), and structured return values — failures are routed, never swallowed as bare exceptions.

import logging
import hashlib
import uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd
import openpyxl
from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator, model_validator

# Structured audit logger configuration (stdlib only; no print).
AUDIT_LOGGER = logging.getLogger("grant_budget_audit")
AUDIT_LOGGER.setLevel(logging.INFO)
# Python's stdlib has no JSONFormatter; emit JSON-shaped lines via a plain Formatter.
_formatter = logging.Formatter(
    '{"timestamp":"%(asctime)s","level":"%(levelname)s","event":"%(message)s"}'
)
_handler = logging.StreamHandler()
_handler.setFormatter(_formatter)
AUDIT_LOGGER.addHandler(_handler)

ALLOWED_CATEGORIES = {"PERSONNEL", "TRAVEL", "EQUIPMENT", "INDIRECT", "OTHER_DIRECT"}


class BudgetLineItem(BaseModel):
    """Frozen, versioned contract for a single validated budget row."""

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    category: str
    description: str
    amount: Decimal
    period_start: datetime
    period_end: datetime
    indirect_rate: Optional[Decimal] = Field(None, ge=Decimal("0"), le=Decimal("0.15"))

    @field_validator("amount", mode="before")
    @classmethod
    def coerce_decimal(cls, v: Any) -> Decimal:
        try:
            return Decimal(str(v).replace(",", "").replace("$", "").strip())
        except InvalidOperation as exc:
            raise ValueError(f"Non-numeric budget value detected: {v!r}") from exc

    @field_validator("category", mode="before")
    @classmethod
    def validate_category(cls, v: str) -> str:
        normalized = str(v).upper().strip()
        if normalized not in ALLOWED_CATEGORIES:
            raise ValueError(f"Invalid budget category: {v!r}")
        return normalized

    @model_validator(mode="after")
    def validate_period_alignment(self) -> "BudgetLineItem":
        if self.period_end <= self.period_start:
            raise ValueError("Period end must strictly follow period start")
        return self


class BudgetValidator:
    """Single-pass validation gate: structural inspection, schema contract, reconciliation."""

    def __init__(self, file_path: Path, schema_version: str = "v2.1") -> None:
        self.file_path = file_path
        self.schema_version = schema_version
        self.trace_id = str(uuid.uuid4())
        self.checksum = self._compute_checksum()

    def _compute_checksum(self) -> str:
        sha256 = hashlib.sha256()
        with open(self.file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
        return sha256.hexdigest()

    def _log(self, level: str, message: str, **kwargs: Any) -> None:
        AUDIT_LOGGER.log(
            getattr(logging, level.upper()),
            message,
            extra={"trace_id": self.trace_id, **kwargs},
        )

    def _quarantine(self, reason: str) -> Dict[str, Any]:
        return {"status": "QUARANTINE", "reason": reason, "trace_id": self.trace_id}

    def validate(self) -> Dict[str, Any]:
        self._log(
            "INFO",
            "Initiating budget template validation",
            file=str(self.file_path),
            schema=self.schema_version,
        )

        # Gate 1 — structural integrity via openpyxl (short-circuits before data parsing).
        try:
            wb = openpyxl.load_workbook(self.file_path, read_only=True, data_only=True)
            ws = wb.active
            if any(ws.merged_cells.ranges):
                raise ValueError("Merged cells violate structural integrity policy")
            if ws.protection.sheet:
                raise ValueError("Protected sheets block programmatic validation")
            wb.close()
        except Exception as exc:  # noqa: BLE001 - boundary gate routes, never swallows
            self._log("ERROR", f"Structural validation failed: {exc}")
            return self._quarantine("STRUCTURAL_FAILURE")

        # Gate 2-4 — header verification, typed contract, reconciliation.
        try:
            df = pd.read_excel(self.file_path, engine="openpyxl")
            df.columns = df.columns.str.lower().str.strip()
            required = {"category", "description", "amount", "period_start", "period_end"}
            missing = required - set(df.columns)
            if missing:
                raise ValueError(f"Missing required headers: {sorted(missing)}")

            df["period_start"] = pd.to_datetime(df["period_start"], errors="raise")
            df["period_end"] = pd.to_datetime(df["period_end"], errors="raise")

            validated: List[BudgetLineItem] = []
            for idx, row in df.iterrows():
                try:
                    validated.append(BudgetLineItem(**row.to_dict()))
                except ValidationError as ve:
                    self._log("ERROR", f"Row {idx} validation failed", errors=ve.errors())
                    return self._quarantine("SCHEMA_VIOLATION")

            declared_total = Decimal(str(df["amount"].sum()))
            computed_total = sum((item.amount for item in validated), Decimal("0"))
            if abs(computed_total - declared_total) > Decimal("0.01"):
                raise ValueError("Arithmetic mismatch between line items and declared total")

            self._log(
                "INFO",
                "Validation successful",
                row_count=len(validated),
                total_expenses=str(computed_total),
            )
            return {
                "status": "VALIDATED",
                "trace_id": self.trace_id,
                "checksum": self.checksum,
                "schema_version": self.schema_version,
                "validated_count": len(validated),
                "total_budget": str(computed_total),
                "audit_timestamp": datetime.now(timezone.utc).isoformat(),
            }
        except Exception as exc:  # noqa: BLE001 - boundary gate routes, never swallows
            self._log("ERROR", f"Data validation failed: {exc}")
            return self._quarantine("DATA_FAILURE")

The validated payload is the only thing that crosses the egress boundary — a row count, a SHA-256 checksum, the schema version, and the reconciled total. Field aliasing, currency conversion, and grant-level aggregation are deliberately absent; those responsibilities belong to Field Mapping & Normalization.

Field Mapping & Schema Contract #

The frozen schema registry maps each funder header to a canonical field, a type-coercion rule, and a rejection condition. The contract is versioned (GRANT_BUDGET_SCHEMA_VERSION) so that two ingestions of the same template version are bit-for-bit reproducible in their validation outcome.

Two coercion rules carry compliance weight. First, every monetary value is parsed through Decimal, never float — binary floating point cannot represent most decimal cent values exactly, and a reconciliation that sums float line items will drift, producing a spurious mismatch or, worse, hiding a real one. Second, category normalization is allow-list-bound, not free-text: an unrecognized category is a rejection, not a new category, because downstream cost-principle mapping depends on a closed enumeration.

Validation & Testing #

Verification asserts on two things: the structured return contract, and the audit log. A passing workbook must return status == "VALIDATED" with a reconciled total; a tampered workbook must route to quarantine with the correct reason code and must not mutate the input.

from decimal import Decimal
from pathlib import Path

import pytest

from budget_sync import BudgetValidator, BudgetLineItem


def test_valid_template_emits_canonical_payload(tmp_path: Path, valid_workbook: Path) -> None:
    result = BudgetValidator(valid_workbook, schema_version="v2.1").validate()
    assert result["status"] == "VALIDATED"
    assert result["validated_count"] == 3
    assert Decimal(result["total_budget"]) == Decimal("125000.00")
    assert len(result["checksum"]) == 64  # SHA-256 hex digest


def test_merged_cells_route_to_structural_quarantine(merged_cell_workbook: Path) -> None:
    result = BudgetValidator(merged_cell_workbook).validate()
    assert result["status"] == "QUARANTINE"
    assert result["reason"] == "STRUCTURAL_FAILURE"


def test_indirect_rate_over_cap_is_rejected() -> None:
    with pytest.raises(ValueError):
        BudgetLineItem(
            category="indirect",
            description="F&A",
            amount="10000",
            period_start="2026-01-01",
            period_end="2026-12-31",
            indirect_rate=Decimal("0.42"),  # exceeds the 0.15 contract ceiling
        )


def test_total_mismatch_routes_to_data_failure(mismatched_total_workbook: Path) -> None:
    result = BudgetValidator(mismatched_total_workbook).validate()
    assert result["status"] == "QUARANTINE"
    assert result["reason"] == "DATA_FAILURE"

Pair the example-based tests with a property-based check: generate random well-typed rows with Hypothesis and assert that any row whose period_end <= period_start raises, and that any in-range row round-trips through BudgetLineItem without loss of Decimal precision. The deterministic trace_id and checksum in the return value make every test assertion reconstructable from the audit log alone.

Performance & Scale Considerations #

Grant budget templates are small — tens to low hundreds of rows — so the dominant cost is workbook open and parse, not row validation. Tune for predictable memory and reproducibility rather than raw throughput:

• Read-only, data-only loads: load_workbook(..., read_only=True, data_only=True) streams rows and discards formula ASTs, keeping a single workbook’s footprint flat regardless of styling complexity. Never hold more than one workbook in memory per worker.
• Batch sizing: process workbooks in small batches (50–200 files) rather than loading a directory at once. Each file is independent and synchronous by design, which guarantees deterministic memory allocation and a clean one-file-per-trace audit record.
• Concurrency: parallelize across files with a bounded process pool, but keep each file’s validation single-threaded — pydantic validation is CPU-light, and concurrency inside one workbook buys nothing while complicating the trace. Concurrent dispatch across many submissions belongs to Async Batch Processing Pipelines.
• Memory ceiling: target well under 256 MB per worker; this stage fits comfortably on the small instances typical of nonprofit infrastructure.
• Checksum cost: the SHA-256 hash reads the file once in 8 KB chunks — negligible for budget-sized workbooks and worth it for non-repudiation.

Failure Modes & Troubleshooting #

Every quarantine outcome is observable: a sustained shift in one reason code is the primary operational signal. A surge of SCHEMA_VIOLATION on headers almost always means a funder published a new template version — fix the contract upstream rather than loosening validation here. Bounded retries and backoff for transient storage faults are not this stage’s concern; they belong to Pipeline Fallback & Retry Logic.

Compliance Alignment #

This gate satisfies specific federal cost-principle and internal-control obligations rather than generic “compliance,” and every validated payload and quarantine report is auditable evidence.

Compliance officers can query the structured audit logs by trace_id to reconstruct the exact validation state at ingestion time. Quarantined files retain their original binary state alongside a JSON validation report, ensuring non-repudiation during federal audits or internal financial reviews. Under the 2 CFR §200.334 three-year minimum, the quarantine artifacts at GRANT_BUDGET_QUARANTINE_PATH are retention records and must be preserved (longer where a funder or state statute extends it). Audit artifacts emitted here conform to the conventions defined in Compliance Metadata Standards, and indirect-cost ceilings reconcile against the structures in IRS 990 Data Schema Mapping. The gate guarantees that only structurally sound, compliance-verified budget data enters the financial reconciliation layer.

Related #

• Parent: Data Ingestion & Grant Parsing Workflows
• Automating Excel to CSV conversion for budget tracking — the downstream serialization step that consumes validated payloads
• Field Mapping & Normalization — owns the canonical schema and alias resolution this stage hands off to
• Error Categorization & Logging — the sink that classifies and routes every quarantine directive
• Async Batch Processing Pipelines — concurrent dispatch across many submissions
• Compliance Metadata Standards — cross-domain audit-artifact field conventions