IRS 990 Data Schema Mapping

This module is part of the Core Architecture & Compliance Mapping reference, where it functions as the discrete transformation layer that turns raw IRS Form 990 filings into a canonical, type-safe payload. Its mandate is strictly bounded: ingest a 990 submission, normalize heterogeneous field representations into one canonical JSON structure, enforce deterministic validation, and emit an audit-ready object. It terminates at the validation gate.

Explicitly out of scope. This subsystem does not disburse grants, synchronize donor CRM records, or render regulatory filings. Jurisdictional registration checks belong to State Charity Registration Compliance; funder eligibility logic belongs to Grantor-Specific Rule Taxonomies; access control and encryption belong to Data Security & Access Boundaries. Document-level extraction mechanics — XML coordinate handling, OCR confidence thresholds, streaming chunk boundaries — are owned by the Data Ingestion & Grant Parsing Workflows reference. By enforcing this separation of concerns, the mapping layer guarantees structural integrity, type fidelity, and immutable lineage for every 990 payload entering the platform.

The four stakeholders served are nonprofit operations and grant managers (predictable standardized financial disclosures), Python automation developers (deterministic parsing contracts and structured error payloads), and compliance officers (auditable validation trails and explicit reconciliation proofs).

Prerequisites #

This module pins its runtime and dependencies so that normalization is byte-for-byte reproducible across CI and production workers.

• Python: 3.11 or newer (relies on datetime.UTC semantics and the faster decimal C backend).
• Pinned packages:

# requirements.txt (mapping layer only)
pydantic==2.7.1          # canonical validation models (v2 model_validator API)
lxml==5.2.1              # native IRS eFile XML parsing
orjson==3.10.3          # deterministic, fast JSON serialization
python-dateutil==2.9.0  # tolerant fiscal-year date parsing
pytest==8.2.0           # validation & reconciliation tests
hypothesis==6.100.1     # property-based numeric edge-case generation

• Environment variables: IRS990_SCHEMA_VERSION (pins the active canonical schema, e.g. irs990_v2.1), IRS990_RECON_TOLERANCE (reconciliation delta ceiling in dollars, default 0.00), and AUDIT_SINK_URL (write-once log target).
• Upstream dependency: this stage consumes the parsed-but-untyped dictionary produced by Field Mapping & Normalization in the ingestion reference. It assumes transport-level extraction has already succeeded; it does not re-open source files.

Core Implementation: Ingestion & Canonical Normalization #

The ingestion boundary accepts 990 submissions across three transport formats — native XML (IRS eFiling), structured CSV (third-party aggregators), and OCR-extracted text (legacy archives). Each format routes through a dedicated parser adapter that emits a flat or lightly nested dictionary. Those dictionaries converge into a single normalize_payload function responsible for alias resolution, type coercion, and structural alignment.

Field aliases (EIN, TaxID, EmployerIdentificationNumber, 990_EIN) collapse into a single canonical ein key via a deterministic lookup table. Numeric strings carrying currency symbols, thousands separators, or trailing whitespace are sanitized and cast to decimal.Decimal to eliminate floating-point drift. Boolean indicators (X, Yes, 1, true, checked) normalize to strict Python bool. Complex compensation arrays in Part VII follow the structural conventions documented in How to map IRS 990 Part VII to JSON schema; deviations trigger immediate normalization rejection.

import logging
import decimal
from datetime import datetime, timezone
from typing import Any, Dict, List

logger = logging.getLogger("irs990.normalizer")

# Canonical alias mapping (subset for demonstration).
FIELD_ALIASES: Dict[str, set] = {
    "ein": {"ein", "tax_id", "employer_identification_number", "990_ein"},
    "organization_name": {"name", "org_name", "legal_name"},
    "total_revenue": {"part_i_line_12", "total_revenue", "gross_receipts"},
    "is_501c3": {"is_501c3", "exempt_status", "section_501c3", "is_charitable"},
}

BOOLEAN_TRUTHY = {"x", "yes", "1", "true", "checked", "y"}


def _sanitize_numeric(value: Any) -> decimal.Decimal:
    """Strip non-numeric characters and return Decimal. Raises decimal.InvalidOperation on garbage."""
    if isinstance(value, (int, float, decimal.Decimal)):
        return decimal.Decimal(str(value))
    cleaned = str(value).replace(",", "").replace("$", "").strip()
    if not cleaned or cleaned.lower() in ("n/a", "none", "null"):
        return decimal.Decimal("0")
    return decimal.Decimal(cleaned)


def normalize_payload(raw: Dict[str, Any], trace_id: str) -> Dict[str, Any]:
    """Normalize a raw parsed 990 dictionary into the canonical structure.

    Returns the canonical payload with immutable audit metadata attached.
    Missing aliases are recorded as structured warnings, never silently dropped.
    """
    canonical: Dict[str, Any] = {}
    audit_log: List[str] = []

    for canonical_key, aliases in FIELD_ALIASES.items():
        matched = False
        for alias in aliases:
            if alias in raw:
                val = raw[alias]
                if canonical_key == "ein":
                    canonical["ein"] = str(val).replace("-", "").strip()
                elif canonical_key == "total_revenue":
                    canonical["total_revenue"] = _sanitize_numeric(val)
                elif canonical_key == "is_501c3":
                    canonical["is_501c3"] = str(val).strip().lower() in BOOLEAN_TRUTHY
                else:
                    canonical[canonical_key] = str(val).strip()
                matched = True
                break
        if not matched:
            audit_log.append(f"MISSING_ALIAS:{canonical_key}")

    canonical["_meta"] = {
        "trace_id": trace_id,
        "normalized_at": datetime.now(timezone.utc).isoformat(),
        "normalization_warnings": audit_log,
        "schema_version": "irs990_v2.1",
    }

    logger.info(
        "Normalization complete",
        extra={"trace_id": trace_id, "warnings": len(audit_log)},
    )
    return canonical

Normalization terminates at the validation gate. No reconciliation, rule evaluation, or compliance tagging happens inside this boundary.

Field Mapping & Schema Contract #

The canonical contract is the single source of truth that every ingestion format must satisfy. The alias resolution table below is the authoritative lookup; the type coercion column defines exactly how each raw value is normalized before it reaches validation.

Required keys (ein, total_revenue, authorized_officer) trigger immediate rejection when absent or null. Optional keys default to null or decimal.Decimal("0") per the canonical schema defaults. Downstream consumers — including the Compliance Metadata Standards layer — read only these canonical keys, never the original source aliases, which is what keeps the three transport formats interchangeable.

Validation & Reconciliation #

Normalized payloads enter the validation layer, which executes explicit structural and semantic checks using pydantic v2 strict models. Required fields trigger immediate rejection if absent. Type mismatches generate structured error payloads carrying the JSON path, expected type, and offending value.

Cross-record reconciliation enforces mathematical consistency across 990 sections. Part I, Line 12 (Total Revenue) must equal the sum of constituent lines (1h and 2–11). Validation computes the delta and flags any discrepancy exceeding the configured tolerance (IRS990_RECON_TOLERANCE, default $0.00). Reconciliation failures do not crash the pipeline; they emit a structured exception payload for downstream triage.

from pydantic import BaseModel, Field, ValidationError, model_validator
from typing import Any, Dict
import decimal


class IRS990PartI(BaseModel):
    line_1h: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_2: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_3: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_4: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_5: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_6: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_7: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_8: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_9: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_10: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_11: decimal.Decimal = Field(default=decimal.Decimal("0"))
    line_12_total_revenue: decimal.Decimal

    @model_validator(mode="after")
    def validate_revenue_sum(self) -> "IRS990PartI":
        """Cross-field reconciliation: Line 12 must equal the sum of Lines 1h-11.

        In Pydantic v2, cross-field validation uses model_validator(mode="after");
        every field is accessible as self.<field_name>.
        """
        constituent_sum = (
            self.line_1h + self.line_2 + self.line_3 + self.line_4
            + self.line_5 + self.line_6 + self.line_7 + self.line_8
            + self.line_9 + self.line_10 + self.line_11
        )
        delta = abs(self.line_12_total_revenue - constituent_sum)
        if delta > decimal.Decimal("0.00"):
            raise ValueError(
                f"Revenue reconciliation failed: Line 12 ({self.line_12_total_revenue}) "
                f"!= Sum of Lines ({constituent_sum}). Delta: {delta}"
            )
        return self


def validate_and_reconcile(normalized: Dict[str, Any], trace_id: str) -> Dict[str, Any]:
    """Run Pydantic validation and reconciliation.

    Returns the validated payload on success, or a structured error payload on
    failure. Never raises into the caller; errors are routed as data.
    """
    try:
        part_i = IRS990PartI(**normalized.get("part_i", {}))
        normalized["part_i"] = part_i.model_dump(mode="json")
        normalized["_meta"]["validation_status"] = "PASSED"
        logger.info("Validation passed", extra={"trace_id": trace_id})
        return normalized
    except ValidationError as e:
        error_payload = {
            "trace_id": trace_id,
            "validation_status": "FAILED",
            "errors": [
                {
                    "field": ".".join(str(loc) for loc in err["loc"]),
                    "expected_type": "decimal.Decimal",
                    "actual_value": str(err.get("input", "")),
                    "message": err["msg"],
                }
                for err in e.errors()
            ],
            "timestamp": datetime.now(timezone.utc).isoformat(),
        }
        logger.warning(
            "Validation failed",
            extra={"trace_id": trace_id, "error_count": len(error_payload["errors"])},
        )
        return error_payload

Testing the contract #

Reconciliation is the kind of arithmetic invariant that property-based testing exercises far better than hand-picked fixtures. The pytest plus hypothesis suite below asserts both the happy path and the structured-failure path, and confirms the audit log carries the right status.

import decimal
import hypothesis.strategies as st
from hypothesis import given

money = st.decimals(min_value=0, max_value=10_000_000, places=2)


def test_balanced_filing_passes() -> None:
    normalized = {
        "part_i": {"line_1h": "100.00", "line_2": "50.00", "line_12_total_revenue": "150.00"},
        "_meta": {},
    }
    result = validate_and_reconcile(normalized, trace_id="t-pass")
    assert result["_meta"]["validation_status"] == "PASSED"


def test_unbalanced_filing_emits_structured_error() -> None:
    normalized = {
        "part_i": {"line_1h": "100.00", "line_2": "50.00", "line_12_total_revenue": "999.00"},
        "_meta": {},
    }
    result = validate_and_reconcile(normalized, trace_id="t-fail")
    assert result["validation_status"] == "FAILED"
    assert "reconciliation failed" in result["errors"][0]["message"].lower()


@given(a=money, b=money)
def test_sum_invariant_always_reconciles(a: decimal.Decimal, b: decimal.Decimal) -> None:
    normalized = {
        "part_i": {"line_1h": str(a), "line_2": str(b), "line_12_total_revenue": str(a + b)},
        "_meta": {},
    }
    result = validate_and_reconcile(normalized, trace_id="t-prop")
    assert result["_meta"]["validation_status"] == "PASSED"

A passing payload returns _meta.validation_status == "PASSED"; a failing one returns a top-level validation_status == "FAILED" with one error object per offending field. Business-rule evaluation, jurisdictional checks, and funder constraints are explicitly deferred to downstream consumers.

Performance & Scale Considerations #

Nonprofit-scale 990 corpora are large but not web-scale: a state-level grantmaker might reprocess 50,000–250,000 historical filings during a schema migration. The mapping layer is CPU-bound on decimal arithmetic and XML parsing, not I/O-bound, so the tuning levers are batch size and worker count rather than connection pools.

• Batch sizing: normalize in batches of 500–1,000 filings. Smaller batches inflate per-batch logging overhead; larger batches delay the first audit checkpoint and grow the memory high-water mark.
• Concurrency: parsing and normalization are pure functions, so a ProcessPoolExecutor sized to os.cpu_count() scales linearly. Keep the canonical schema and alias table as module-level constants so they are copied once per worker, not per call.
• Memory ceilings: stream XML with lxml.etree.iterparse and clear processed elements; a single Part VII array can exceed 500 contractor records. Cap supplemental narratives at the 100 KB schema limit so a malformed Schedule O cannot balloon a worker.
• Determinism over speed: never replace decimal.Decimal with float to save cycles. A single floating-point reconciliation error costs more in audit remediation than the entire batch saved in CPU time. Concurrent worker fan-out and retry semantics are governed by Pipeline Fallback & Retry Logic.

Failure Modes & Troubleshooting #

Audit logs are written to a write-once sink (AUDIT_SINK_URL); no payload mutation occurs post-validation. This module does not implement backoff or circuit breakers — transient ingestion failures are handled by the retry layer above it.

Compliance Alignment #

This subsystem satisfies specific, named obligations rather than generic “compliance”:

• 2 CFR §200.302 (financial management). The Uniform Guidance requires records that adequately identify the source and application of funds and that permit reconciliation. The immutable trace_id, schema_version, and validation_status lineage attached to every payload provides exactly that reconciliation evidence to a federal pass-through entity.
• 2 CFR §200.334 (record retention). The write-once audit sink supports the three-year retention floor for financial records; the schema_version stamp lets auditors replay a filing against the schema that was active when it was processed.
• IRS Form 990, Part I, Line 12. The reconciliation validator enforces the Total Revenue identity (Line 12 = sum of Lines 1h, 2–11) at a $0.00 tolerance, catching transcription errors before they propagate into grant-eligibility decisions.
• IRS Form 990, Part VII, Section A. Officer and key-employee compensation is normalized into a sorted, type-checked array, preserving the governance-transparency disclosure required for audit.

The canonical mapping follows the official Instructions for Form 990 and adheres to Python’s decimal arithmetic standard for financial precision. Jurisdiction-specific charitable-solicitation thresholds (for example, California Attorney General registration tied to gross revenue) are evaluated downstream in State Charity Registration Compliance; this layer only guarantees the numbers those checks depend on are present, typed, and reconciled.

Related #

• Core Architecture & Compliance Mapping — parent architecture reference and stage contracts.
• State Charity Registration Compliance — consumes the validated 990 payload for jurisdictional filing.
• Grantor-Specific Rule Taxonomies — evaluates funder eligibility on the canonical fields.
• Compliance Metadata Standards — the audit-metadata contract this module emits against.
• How to map IRS 990 Part VII to JSON schema — step-by-step compensation-array mapping.
• Field Mapping & Normalization — the upstream ingestion stage that feeds this layer.