Reconciling F&A indirect-cost recovery rates in Python

Problem statement #

You need a deterministic Python routine that takes an award’s direct-cost lines, strips out the categories 2 CFR 200.1 excludes from the modified total direct cost (MTDC) base, multiplies the remaining base by the negotiated F&A rate in force for the posting’s effective period, and compares that expected recovery against the indirect charge the ledger actually posted — flagging any variance beyond tolerance so that a re-run over the same award never double-counts and never lets a silent over- or under-recovery through.

This task sits under Indirect Cost Recovery Reconciliation, part of the broader Compliance Reporting & Budget Reconciliation practice. The routine is deliberately narrow: it recomputes the entitlement from first principles and reports the variance. It does not post correcting entries or move money — it hands a flagged record to a compliance officer, consistent with the flag-never-correct contract the parent guide establishes.

Prerequisites #

• Python 3.10+ for modern type hints and X | None unions.
• Libraries: SQLAlchemy>=2.0 with psycopg[binary] for the idempotent upsert. The arithmetic uses only the standard-library decimal and datetime modules — never float for money. Install with pip install "SQLAlchemy>=2.0" "psycopg[binary]".
• Environment variables (never hard-code credentials, per Security Boundary Configuration):
  • RECON_DB_URL — the SQLAlchemy connection string for the reconciliation ledger.
• Policy config: a version-controlled copy of the institution’s negotiated rate agreement (NICRA) — each rate with its effective start and end dates — kept alongside your University Policy Mapping Frameworks. The direct-cost detail itself is assumed already mapped and validated upstream by the Grant Lifecycle Architecture Design layer.

Step-by-step implementation #

The flow below is what the routine enforces: direct-cost lines are filtered against the 2 CFR 200.1 exclusion rules to form the MTDC base, the negotiated rate is resolved for the effective period, the expected indirect is computed and rounded once, and the result is compared to the posted charge and written idempotently on the (award_id, rate_effective_period) key.

Step 1 — Load the award’s direct-cost lines #

Read the direct-cost detail for the award, keeping each line’s object category intact — the category is what the exclusion filter keys on. Money is parsed straight into Decimal from the source string so no binary-float rounding ever enters the base.

from decimal import Decimal
from dataclasses import dataclass


@dataclass(frozen=True)
class DirectLine:
    """A single direct-cost line, pre-classified by object category."""
    category: str          # 'equipment' | 'subaward' | 'participant_support' | 'other'
    amount: Decimal        # parsed from the source string, never float


def load_lines(rows: list[dict]) -> list[DirectLine]:
    # rows arrive already validated upstream; we only re-type money as Decimal.
    return [DirectLine(category=r["category"], amount=Decimal(str(r["amount"])))
            for r in rows]

Step 2 — Apply the MTDC exclusions #

This is the compliance core. Each branch encodes one exclusion from 2 CFR 200.1, and the comment names the rule so an auditor can trace the dollar. The equipment test uses a strict greater-than against 5000.00, and the subaward branch subtracts only the remainder over the first 25000.00 of each subaward.

EQUIPMENT_THRESHOLD = Decimal("5000.00")   # capital equipment floor, 2 CFR 200.1
SUBAWARD_INCLUDED = Decimal("25000.00")    # first 25k of EACH subaward stays in base

FULLY_EXCLUDED = {
    "participant_support",   # participant support costs — excluded from MTDC
    "tuition_remission",
    "scholarship",
    "rent_space",            # rental of off-site space
    "capital_expenditure",
}


def compute_base(lines: list[DirectLine]) -> tuple[Decimal, Decimal]:
    """Return (mtdc_base, excluded_amount). Deterministic for a given line set."""
    direct_total = sum((l.amount for l in lines), Decimal("0.00"))
    excluded = Decimal("0.00")
    for line in lines:
        if line.category == "equipment" and line.amount > EQUIPMENT_THRESHOLD:
            excluded += line.amount                      # whole capital item out
        elif line.category == "subaward":
            # Per subaward: only the amount OVER the first 25k is excluded.
            excluded += max(Decimal("0.00"), line.amount - SUBAWARD_INCLUDED)
        elif line.category in FULLY_EXCLUDED:
            excluded += line.amount
    return direct_total - excluded, excluded

Step 3 — Resolve the negotiated rate for the period #

A multi-year award crosses NICRA boundaries, so the rate is a function of the posting date. An uncovered date returns None and must route to quarantine — never fall back to a default rate, which would fabricate an entitlement the agreement does not authorize.

from datetime import date


@dataclass(frozen=True)
class RatePeriod:
    negotiated_rate: Decimal    # e.g. Decimal("0.545")
    effective_start: date
    effective_end: date


def resolve_rate(posting_date: date, agreement: list[RatePeriod]) -> RatePeriod | None:
    for period in agreement:
        if period.effective_start <= posting_date <= period.effective_end:
            return period
    return None   # uncovered — caller quarantines rather than guessing a rate

Step 4 — Compute expected indirect and the variance #

Multiply the base by the rate and quantize once, at the end. Rounding before the multiplication, or rounding twice, is the classic source of a phantom one-cent variance. The variance is signed: positive is over-recovery, negative is under-recovery.

from decimal import ROUND_HALF_UP

TOLERANCE = Decimal("0.50")   # cents of drift accepted before a record is flagged


def expected_indirect(base: Decimal, rate: Decimal) -> Decimal:
    # Single rounding step at the end — 2 CFR 200.414 entitlement, to the cent.
    return (base * rate).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)


def classify(variance: Decimal) -> str:
    if abs(variance) <= TOLERANCE:
        return "reconciled"
    return "over_recovered" if variance > 0 else "under_recovered"

Step 5 — Compare, flag, and write idempotently #

The driver assembles the record and upserts it on the composite (award_id, rate_effective_period) key. A flagged record is both written to the ledger and handed to the quarantine callback; a re-run with unchanged inputs updates the same row to identical values and inserts nothing.

import logging
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

logger = logging.getLogger("fa_reconciler")


def reconcile(
    award_id: str,
    lines: list[DirectLine],
    posted_indirect: Decimal,
    posting_date: date,
    agreement: list[RatePeriod],
    session: Session,
    quarantine,           # callable: (record: dict) -> None
) -> dict:
    base, excluded = compute_base(lines)
    period = resolve_rate(posting_date, agreement)
    direct_total = sum((l.amount for l in lines), Decimal("0.00"))

    if period is None:
        record = _row(award_id, "UNCOVERED", direct_total, excluded, base,
                      Decimal("0"), Decimal("0.00"), posted_indirect,
                      posted_indirect, "uncovered_period")
        quarantine(record)
    else:
        rate_key = f"{period.effective_start.isoformat()}/{period.effective_end.isoformat()}"
        expected = expected_indirect(base, period.negotiated_rate)
        variance = (posted_indirect - expected).quantize(Decimal("0.01"))
        status = classify(variance)
        record = _row(award_id, rate_key, direct_total, excluded, base,
                      period.negotiated_rate, expected, posted_indirect,
                      variance, status)
        if status != "reconciled":
            logger.warning("F&A variance award=%s variance=%s status=%s",
                           award_id, variance, status)
            quarantine(record)

    stmt = pg_insert(ReconciliationRecord).values(**record)
    stmt = stmt.on_conflict_do_update(
        index_elements=["award_id", "rate_effective_period"],
        set_={k: stmt.excluded[k] for k in (
            "direct_cost", "excluded_amount", "mtdc_base", "negotiated_rate",
            "expected_indirect", "posted_indirect", "variance", "status")},
    )
    session.execute(stmt)
    session.commit()
    return record


def _row(award_id, period, direct, excluded, base, rate,
         expected, posted, variance, status) -> dict:
    return {
        "award_id": award_id, "rate_effective_period": period,
        "direct_cost": direct, "excluded_amount": excluded, "mtdc_base": base,
        "negotiated_rate": rate, "expected_indirect": expected,
        "posted_indirect": posted, "variance": variance, "status": status,
    }

The on_conflict_do_update on the composite key is the single idempotency guarantee: reconciling a period twice yields one row, a corrected posted charge re-derives the variance in place, and a new award-period inserts once. The same validated record can then flow into the audit deliverables built by Immutable Audit Report Generation.

Schema and field reference #

Verification #

1. Base check. For the persisted row, confirm direct_cost − excluded_amount == mtdc_base. A gap means the exclusion sum and the base were computed from different inputs.
2. Reproduce the expectation. Recompute expected_indirect(mtdc_base, negotiated_rate) and compare to the stored value — because both are Decimal, the match must be exact.
3. Idempotency dry-run. Call reconcile twice with the identical payload; exactly one row must exist for that (award_id, rate_effective_period), with byte-identical field values after the second pass.
4. Rate-period boundary. Reconcile a posting dated one day after a rate change and confirm it resolves to the new rate, and one day inside a renegotiation gap and confirm it yields uncovered_period.

Troubleshooting #

• Equipment USD 5,000 threshold off-by-one and capitalization policy. The rule excludes equipment costing more than USD 5,000, so a line at exactly 5000.00 stays in the base — a >= test wrongly excludes it. Just as important, only items your institution actually capitalizes count as equipment; a 4,900.00 instrument bought as a supply is not equipment even if it functions like one, and reclassifying it after the fact moves the base. Confirm the object-code classification at the source before reconciling.
• Subaward first-USD 25k rule applied per award instead of per subaward. The USD 25,000 that stays in the base is per subaward, not once across the award. Summing all subawards and subtracting a single USD 25,000, or applying the threshold per invoice line, both produce the wrong base. Group lines by subaward identity first, then apply max(0, amount − 25000) to each.
• Rate change proration across effective dates. When an award spans a rate change, do not reconcile the whole award against one rate. Split the direct costs by posting date, resolve the rate for each date, and produce one reconciliation row per rate_effective_period. A posting whose date falls in an uncovered gap must be quarantined, never reconciled against the nearest rate.

Frequently asked questions #

Related #

• Up to the parent topic: Indirect Cost Recovery Reconciliation
• Detecting Cost Transfer Anomalies in Grant Ledgers — catching the transfers that distort the base
• Compliance Reporting & Budget Reconciliation — the parent practice this reconciliation feeds
• University Policy Mapping Frameworks — where the negotiated rate agreement lives