Automating Excel to CSV Conversion for Budget Tracking

Convert funder Excel budget templates to audit-grade CSV in Python: read-only openpyxl streaming, canonical header resolution, decimal.Decimal precision, SHA-256 hashing, and 2 CFR §200.302/§200.403 quarantine routing.

This guide is part of the Excel Budget Template Sync section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: how do you turn a funder-supplied .xlsx budget template into a flat CSV that downstream stages can trust, without silently corrupting a single monetary cell or losing the audit trail a federal grant requires?

The naive answer — pandas.read_excel(...).to_csv(...) — fails an audit on the first row. It loads the whole workbook into RAM, coerces 120000.10 into a binary float that serializes as 120000.09999999999, and leaves no record of what entered, what left, or why a malformed file was accepted. This guide builds a deterministic converter instead: read-only streaming, string-preserving precision, cryptographic hashing on both sides of the transform, and explicit quarantine routing for anything that drifts from the registered schema.

When to Use This Approach

Reach for this converter when all three conditions hold:

  • The input is an already-validated, version-tagged template. Header drift, merged cells, and macro-enabled workbooks must be resolved before this stage by the parent Excel Budget Template Sync gate, which maintains the canonical header dictionary and version hash. This converter consumes a clean template and produces a clean CSV — nothing more.
  • The output feeds a regulated artifact. Because the resulting CSV ultimately reconciles against 2 CFR §200.302 financial-management records and the cost-principle rules in 2 CFR §200.403, a single rounded or dropped value is a compliance event, not a cosmetic glitch. Precision and traceability are non-negotiable.
  • Volume is bounded but bursty. A handful of grants each carry a few thousand budget lines, arriving in clusters around reporting deadlines. The streaming design holds memory flat regardless of workbook size, so the same code path serves a 40-line modular budget and a 200,000-line consolidated one.

Narrative attachments, OCR, and document extraction are explicitly out of scope — those belong to PDF Grant Application Parsing. Canonical field translation and currency normalization belong to Field Mapping & Normalization. File polling, retries, and rate limiting belong to API Polling & Rate Limiting and Async Batch Processing Pipelines. This stage executes synchronously, one file at a time, so memory allocation and audit ordering stay deterministic.

The converter boundary: validated .xlsx to audit-grade CSV with quarantine routing A left-to-right data-flow diagram. A validated .xlsx file enters and is hashed with SHA-256 for the input audit record. A read-only openpyxl row generator streams rows into two validation gates in series: a header-resolution gate and a per-cell decimal-coercion gate, both drawn in accent. The header gate branches down on schema drift and the decimal gate branches down on a precision failure; both branches feed a single quarantine box that emits MANIFEST_*.json into Error Categorization and Logging. On success, rows flow through csv.DictWriter, the output is hashed with SHA-256, and an audit_manifest.json record ties the output hash to the input hash. A bottom rail notes that a correlation_id is annotated through every hop. Deterministic converter — terminates at a hash-linked audit manifest Validated .xlsx SHA-256 input hash read-only openpyxl iter_rows generator Header resolution canonical + alias map Decimal coercion ROUND_HALF_UP, 2dp csv.DictWriter → SHA-256 output hash audit_manifest.json input_hash ≠ output_hash PASS Quarantine — MANIFEST_*.json handoff → Error Categorization & Logging SchemaDrift Precision correlation_id annotated through every hop · one JSON audit line per file · no cell silently rounded or dropped

Step-by-Step Implementation

The reference implementation targets Python 3.10+ and uses only openpyxl plus the standard library — deliberately no pandas, to keep float coercion out of the precision path. Install the one pinned dependency first:

bash
pip install "openpyxl==3.1.2"

Refer to the Python decimal module documentation for the precision guarantees this stage depends on.

Step 1: Configure structured audit logging and a typed error hierarchy

Every conversion must be replayable from its logs, and every failure must carry a name a triage queue can route on. Configure a file-backed logger (never print) and a small exception tree so the caller can distinguish a schema problem from a precision problem.

python
import csv
import hashlib
import json
import logging
from decimal import Decimal, ROUND_HALF_UP, InvalidOperation
from pathlib import Path
from typing import Dict, Generator, List, Optional
from datetime import datetime, timezone

import openpyxl
from openpyxl.worksheet.worksheet import Worksheet

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.FileHandler("grant_budget_audit.log", encoding="utf-8")],
)
logger = logging.getLogger("grant_budget_converter")


class BudgetConversionError(Exception):
    """Base exception for converter-stage failures."""


class SchemaDriftError(BudgetConversionError):
    """Raised when canonical headers cannot be resolved."""


class PrecisionValidationError(BudgetConversionError):
    """Raised when decimal coercion fails or exceeds audit thresholds."""


class MemoryBoundExceededError(BudgetConversionError):
    """Raised when row iteration exceeds the configured memory ceiling."""

The three subclasses map one-to-one onto the validation gates below, so a downstream consumer can catch SchemaDriftError and PrecisionValidationError separately without string-matching log lines.

Step 2: Hash the input for the audit trail

2 CFR §200.302 requires that financial records be traceable. Compute a SHA-256 digest of the file before you touch its contents, in fixed 8192-byte chunks so the hash itself never loads the whole file into memory.

python
def compute_sha256(file_path: Path) -> str:
    """Deterministic file hash for the 2 CFR §200.302 audit trail."""
    sha = hashlib.sha256()
    with open(file_path, "rb") as handle:
        for chunk in iter(lambda: handle.read(8192), b""):
            sha.update(chunk)
    return sha.hexdigest()

Recomputing the same digest on the emitted CSV (Step 5) lets an auditor prove the output corresponds to exactly this input and was not edited in place afterward.

Step 3: Resolve raw headers against the canonical schema

Funders rename columns between cycles — FY Budget one year, Budget Amount the next. Normalize each header, match it against the canonical names or a declared alias map, and fail fast if any required column is missing. This is the first gate; an unresolvable header never reaches the data rows.

python
CANONICAL_SCHEMA: List[str] = [
    "line_item_id", "category", "description",
    "fiscal_year", "budget_amount", "actuals", "variance_pct",
]

ALIAS_MAP: Dict[str, str] = {
    "personnel": "category",
    "line_item": "line_item_id",
    "fy_budget": "budget_amount",
    "actual_spend": "actuals",
    "variance_%": "variance_pct",
    "description/notes": "description",
    "fiscal_year": "fiscal_year",
}

MONETARY_FIELDS = frozenset({"budget_amount", "actuals", "variance_pct"})


def resolve_headers(raw_headers: List[str]) -> Dict[int, str]:
    """Map raw Excel headers to canonical names; raise on missing columns."""
    mapping: Dict[int, str] = {}
    for idx, raw in enumerate(raw_headers):
        normalized = str(raw).strip().lower().replace(" ", "_")
        if normalized in CANONICAL_SCHEMA:
            mapping[idx] = normalized
        elif normalized in ALIAS_MAP:
            mapping[idx] = ALIAS_MAP[normalized]
        else:
            logger.warning("Unmapped header ignored: %s", raw)

    missing = set(CANONICAL_SCHEMA) - set(mapping.values())
    if missing:
        raise SchemaDriftError(f"Canonical columns missing after alias resolution: {missing}")
    return mapping

The alias table is the single place to absorb a funder’s naming change — see Field Mapping & Normalization for the broader alias-resolution pattern this mirrors. Anything still unmapped is logged and dropped, never guessed.

Step 4: Coerce money as strings, never floats

This is the gate auditors care about most. Keep every monetary cell as a str through ingestion, then cast with decimal.Decimal and quantize with explicit ROUND_HALF_UP to two places. A value that will not parse raises PrecisionValidationError rather than slipping through as NaN.

python
def coerce_decimal(value: Optional[str], field: str) -> str:
    """Enforce 2 CFR §200.403 precision: parse via Decimal, return a CSV-safe string."""
    if value is None or str(value).strip() == "":
        return "0.00"
    try:
        parsed = Decimal(str(value).replace(",", "").strip())
        return str(parsed.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP))
    except (InvalidOperation, ValueError) as exc:
        raise PrecisionValidationError(f"Invalid monetary value for {field}: {value!r}") from exc

Stripping the thousands separator before parsing handles 1,250.00 cleanly; routing through Decimal rather than float is what stops 120000.10 from serializing as 120000.09999999999.

Step 5: Stream rows and serialize with both hashes

Open the workbook in read_only=True, data_only=True mode so openpyxl yields rows from a generator instead of materializing the sheet. A hard row ceiling converts a runaway file into a typed error instead of an out-of-memory kill. Write each record through csv.DictWriter, then hash the output and emit one structured audit record.

python
ROW_CEILING = 500_000


def stream_rows(ws: Worksheet, header_map: Dict[int, str]) -> Generator[Dict[str, str], None, None]:
    """Memory-safe row iteration with per-cell type preservation."""
    for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        if row_idx > ROW_CEILING:
            raise MemoryBoundExceededError("Row ceiling exceeded; chunk this file upstream.")
        record: Dict[str, str] = {}
        for col_idx, canonical in header_map.items():
            raw = row[col_idx] if col_idx < len(row) else None
            if canonical in MONETARY_FIELDS:
                record[canonical] = coerce_decimal(raw, canonical)
            else:
                record[canonical] = str(raw).strip() if raw is not None else ""
        yield record


def convert_excel_to_csv(
    input_path: Path,
    output_path: Path,
    sheet_name: Optional[str] = None,
) -> Dict[str, object]:
    """Deterministic Excel-to-CSV conversion with audit logging and quarantine routing."""
    input_hash = compute_sha256(input_path)
    logger.info("Starting conversion | input=%s | hash=%s", input_path.name, input_hash)
    try:
        wb = openpyxl.load_workbook(str(input_path), read_only=True, data_only=True)
        ws = wb[sheet_name] if sheet_name else wb.active
        header_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True))
        header_map = resolve_headers(list(header_row))

        with open(output_path, "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=CANONICAL_SCHEMA)
            writer.writeheader()
            row_count = 0
            for record in stream_rows(ws, header_map):
                writer.writerow(record)
                row_count += 1
        wb.close()

        audit_record: Dict[str, object] = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "input_file": input_path.name,
            "input_hash": input_hash,
            "output_file": output_path.name,
            "output_hash": compute_sha256(output_path),
            "rows_processed": row_count,
            "compliance_status": "PASS",
            "standard": "2 CFR §200.302/§200.403",
        }
        logger.info(json.dumps(audit_record))
        return audit_record

    except (SchemaDriftError, PrecisionValidationError, MemoryBoundExceededError) as exc:
        logger.error("Conversion failed | input=%s | error=%s", input_path.name, exc)
        route_to_quarantine(input_path, exc)
        raise
    except Exception as exc:  # noqa: BLE001 — re-wrapped, never swallowed
        logger.critical("Unhandled failure | input=%s | error=%s", input_path.name, exc)
        raise BudgetConversionError(f"Critical pipeline failure: {exc}") from exc

Step 6: Quarantine instead of dropping

A non-compliant file must never vanish. On any typed failure, move the original aside and write a manifest with the error class, message, and a manual-review flag. These manifests are the handoff contract into Error Categorization & Logging.

python
QUARANTINE_DIR = Path("./quarantine")
QUARANTINE_DIR.mkdir(exist_ok=True)


def route_to_quarantine(file_path: Path, error: Exception) -> None:
    """Preserve the original artifact and emit a triage manifest — never silently drop."""
    quarantine_path = QUARANTINE_DIR / f"QUARANTINED_{file_path.name}"
    file_path.rename(quarantine_path)
    manifest = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "original_file": file_path.name,
        "error_type": error.__class__.__name__,
        "error_message": str(error),
        "routing_action": "QUARANTINE",
        "requires_manual_review": True,
    }
    manifest_path = QUARANTINE_DIR / f"MANIFEST_{file_path.stem}.json"
    with open(manifest_path, "w", encoding="utf-8") as handle:
        json.dump(manifest, handle, indent=2)
    logger.warning("Quarantined | path=%s | manifest=%s", quarantine_path, manifest_path)

Verification

Confirm the converter behaves deterministically with four checks:

  1. Precision survives the round trip. Feed a cell containing 120000.10 and assert the CSV output column reads exactly 120000.10 — not 120000.09999999999. This is the proof that Decimal, not float, is in the path.
  2. A renamed required column is rejected, not guessed. Hand resolve_headers a row missing budget_amount (and without a known alias) and assert it raises SchemaDriftError; confirm a MANIFEST_*.json lands in ./quarantine and the original file is gone from its inbox.
  3. The audit record ties output to input. After a clean run, assert audit_record["input_hash"] and ["output_hash"] are present, distinct, and that re-hashing the files reproduces both digests byte-for-byte.
  4. Memory stays flat. Convert a 200,000-row workbook and assert process RSS does not climb with row count — proof the read-only generator, not a full load, is in effect. A workbook past ROW_CEILING must raise MemoryBoundExceededError.

A compliant run leaves exactly one JSON audit line per file; a failed run leaves an ERROR line plus a quarantine manifest. Ship grant_budget_audit.log to a write-once tier so the trail satisfies the three-year retention period under 2 CFR §200.334.

python
record = convert_excel_to_csv(Path("FY25_budget.xlsx"), Path("FY25_budget.csv"))
assert record["compliance_status"] == "PASS"
assert record["input_hash"] != record["output_hash"]
assert coerce_decimal("120000.10", "budget_amount") == "120000.10"

Common Errors & Fixes

Error Cause Fix
Money serializes as 120000.0999… pandas.read_excel or float() coerced the cell into binary floating point Drop pandas from this stage; keep values as str and cast through coerce_decimal with ROUND_HALF_UP.
SchemaDriftError on a valid-looking file Funder renamed or reordered a required column between cycles Add the new spelling to ALIAS_MAP; if drift is structural, resolve it upstream in Excel Budget Template Sync before this stage runs.
MemoryError / OOM kill on a large workbook Workbook opened without read_only=True, materializing every cell Load with read_only=True, data_only=True and iterate iter_rows; cap at ROW_CEILING and chunk via Async Batch Processing Pipelines.
PrecisionValidationError on 1,250.00 Thousands separator or stray currency symbol left in the cell Strip , (and any $) before Decimal(...); reject anything still unparseable rather than defaulting to zero.
Audit log shows mojibake Default platform encoding used when writing CSV or the log Pass encoding="utf-8" to every open(...) and the FileHandler.
Quarantined files pile up unreviewed Manifests written but never consumed Wire MANIFEST_*.json into Error Categorization & Logging for triage and reprocessing.